2023-04-27 08:46:42 +00:00
import asyncio
import logging
2023-04-27 08:53:08 +00:00
2023-04-27 08:46:42 +00:00
from telegram import Update
from telegram . ext import ApplicationBuilder , ContextTypes , CommandHandler , MessageHandler , filters
import os
import json
2023-04-28 12:04:41 +00:00
from cowmesh_router_iperf_test import CowmeshRouterIperfTester
from cowmesh_pi_iperf_test import CowmeshPiIperfTester
2023-04-27 08:46:42 +00:00
PROJECT_PATH = os . path . abspath ( os . path . dirname ( __file__ ) )
SECRETS_PATH = os . path . join ( PROJECT_PATH , " secrets.json " )
with open ( SECRETS_PATH , ' r ' ) as f :
SECRETS = json . loads ( f . read ( ) )
logging . basicConfig (
format = ' %(asctime)s - %(name)s - %(levelname)s - %(message)s ' ,
level = logging . INFO
)
async def start ( update : Update , context : ContextTypes . DEFAULT_TYPE ) :
await context . bot . send_message ( chat_id = update . effective_chat . id , text = " This is the moonlight analytics bot for mesh network diagnostics. \n \n Type /help to see available commands. " , message_thread_id = update . message . message_thread_id )
async def caps ( update : Update , context : ContextTypes . DEFAULT_TYPE ) :
"""
this is just a function to test that the bot is working as expected
"""
text_caps = ' ' . join ( context . args ) . upper ( )
print ( " chat_id: {} " . format ( update . effective_chat . id ) )
print ( " message_thread_id: {} " . format ( update . message . message_thread_id ) )
await context . bot . send_message ( chat_id = update . effective_chat . id , text = text_caps , message_thread_id = update . message . message_thread_id )
2023-04-28 12:04:41 +00:00
async def router_iperf ( update : Update , context : ContextTypes . DEFAULT_TYPE ) :
time = context . args [ 0 ] if context . args else None
if not time :
time = 10
2023-04-27 08:46:42 +00:00
async def log ( msg ) :
await context . bot . send_message ( chat_id = update . effective_chat . id , text = msg , message_thread_id = update . message . message_thread_id )
2023-04-28 12:04:41 +00:00
if update . effective_chat . id != int ( SECRETS [ " TELEGRAM_LOG_CHAT_ID " ] ) :
await log ( " ++ can only start iperf test from Moonlight Bot group " )
return
await log ( " ++ starting router iperf test with {} seconds per test " . format ( time ) )
tester = CowmeshRouterIperfTester ( log = log , seconds = time )
await tester . run_test ( )
await tester . output_results ( )
async def pi_iperf ( update : Update , context : ContextTypes . DEFAULT_TYPE ) :
time = context . args [ 0 ] if context . args else None
if not time :
time = 10
async def log ( msg ) :
await context . bot . send_message ( chat_id = update . effective_chat . id , text = msg , message_thread_id = update . message . message_thread_id )
if update . effective_chat . id != int ( SECRETS [ " TELEGRAM_LOG_CHAT_ID " ] ) :
await log ( " ++ can only start iperf test from Moonlight Bot group " )
return
await log ( " ++ starting pi iperf test with {} seconds per test " . format ( time ) )
tester = CowmeshPiIperfTester ( log = log , time = time )
2023-04-27 08:46:42 +00:00
await tester . run_test ( )
await tester . output_results ( )
async def unknown ( update : Update , context : ContextTypes . DEFAULT_TYPE ) :
await context . bot . send_message ( chat_id = update . effective_chat . id , text = " Sorry, I didn ' t understand that, please run /help for a list of available commands. " , message_thread_id = update . message . message_thread_id )
def about_message ( ) :
msg = " This is a bot designed to help measure the performance of a mesh network through active testing. " \
" Every night the bot runs a network test using iperf to measure the connectivity between all the nodes in the mesh, " \
" and logs the result to this channel. " \
" Members of this channel can also initiate a new network iperf test at any time by sending a message to this channel, " \
" with the command /iperf . " \
" Telegram users outside of this channel cannot initiate a test, to help keep the network secure from being " \
" overrun by malicious users. " \
" The network test runs at night because that is when the fewest people are using the network, and so is more likely to " \
" give consistent results with less random variability. " \
" Please be mindful of initiating too many iperf tests during the day, as it uses a lot of network resources " \
" to run the test and could interfere with the internet connections of people using the network. "
return msg
def help_message ( ) :
msg = " This bot runs an iperf test every night and logs the results here. You can also initiate a new test using the command /iperf " \
" or read a longer message explaining how this bot works by typing the command /readme. "
return msg
async def about ( update : Update , context : ContextTypes . DEFAULT_TYPE ) :
text = about_message ( )
await context . bot . send_message ( chat_id = update . effective_chat . id , text = text , message_thread_id = update . message . message_thread_id )
async def help_fun ( update : Update , context : ContextTypes . DEFAULT_TYPE ) :
text = help_message ( )
await context . bot . send_message ( chat_id = update . effective_chat . id , text = text , message_thread_id = update . message . message_thread_id )
2023-04-28 12:04:41 +00:00
async def nightly_router_iperf ( time = 10 ) :
token = SECRETS [ " TELEGRAM_TOKEN " ]
application = ApplicationBuilder ( ) . token ( token ) . build ( )
chat_id = SECRETS [ " TELEGRAM_LOG_CHAT_ID " ]
message_thread_id = SECRETS . get ( " TELEGRAM_LOG_MESSAGE_THREAD_ID " )
bot = application . bot
async def log ( msg ) :
await bot . send_message ( chat_id = chat_id , text = msg , message_thread_id = message_thread_id )
await log ( " ☾☾ starting nightly router-to-router iperf test with {} seconds per test " )
tester = CowmeshRouterIperfTester ( log = log , time = time )
await tester . run_test ( )
await tester . output_results ( )
async def nightly_pi_iperf ( time = 10 ) :
2023-04-27 08:46:42 +00:00
token = SECRETS [ " TELEGRAM_TOKEN " ]
application = ApplicationBuilder ( ) . token ( token ) . build ( )
chat_id = SECRETS [ " TELEGRAM_LOG_CHAT_ID " ]
message_thread_id = SECRETS . get ( " TELEGRAM_LOG_MESSAGE_THREAD_ID " )
bot = application . bot
async def log ( msg ) :
await bot . send_message ( chat_id = chat_id , text = msg , message_thread_id = message_thread_id )
2023-04-28 12:04:41 +00:00
await log ( " ☾☾ starting nightly computer-to-computer iperf test with {} seconds per test " . format ( time ) )
tester = CowmeshPiIperfTester ( log = log , time = time )
2023-04-27 08:46:42 +00:00
await tester . run_test ( )
await tester . output_results ( )
def init_bot_listener ( ) :
token = SECRETS [ " TELEGRAM_TOKEN " ]
application = ApplicationBuilder ( ) . token ( token ) . build ( )
start_handler = CommandHandler ( ' start ' , start )
application . add_handler ( start_handler )
caps_handler = CommandHandler ( ' caps ' , caps )
application . add_handler ( caps_handler )
about_handler = CommandHandler ( ' readme ' , about )
application . add_handler ( about_handler )
help_handler = CommandHandler ( ' help ' , help_fun )
application . add_handler ( help_handler )
2023-04-28 12:04:41 +00:00
iperf_handler = CommandHandler ( ' iperf ' , router_iperf )
2023-04-27 08:46:42 +00:00
application . add_handler ( iperf_handler )
2023-04-28 12:04:41 +00:00
pi_iperf_handler = CommandHandler ( ' pi_iperf ' , pi_iperf )
application . add_handler ( pi_iperf_handler )
2023-04-27 08:46:42 +00:00
unknown_handler = MessageHandler ( filters . COMMAND , unknown )
application . add_handler ( unknown_handler )
application . run_polling ( )
if __name__ == ' __main__ ' :
init_bot_listener ( )