mirror of
https://github.com/SpudGunMan/meshing-around.git
synced 2026-08-07 09:22:44 +02:00
refactor scheduler
This commit is contained in:
SpudGunMan
+7
-1
@@ -1616,7 +1616,7 @@ def handle_boot(mesh=True):
|
||||
logger.debug("System: Scheduler Enabled")
|
||||
|
||||
|
||||
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"System: Error during boot: {e}")
|
||||
|
||||
@@ -2009,6 +2009,12 @@ async def main():
|
||||
|
||||
if my_settings.voxDetectionEnabled:
|
||||
tasks.append(asyncio.create_task(voxMonitor(), name="vox_detection"))
|
||||
|
||||
if my_settings.scheduler_enabled:
|
||||
from modules.scheduler import run_scheduler_loop, setup_scheduler
|
||||
setup_scheduler(schedulerMotd, MOTD, schedulerMessage, schedulerChannel, schedulerInterface,
|
||||
schedulerValue, schedulerTime, schedulerInterval)
|
||||
tasks.append(asyncio.create_task(run_scheduler_loop(), name="scheduler"))
|
||||
|
||||
logger.debug(f"System: Starting {len(tasks)} async tasks")
|
||||
|
||||
|
||||
+39
-23
@@ -6,12 +6,30 @@ from modules.log import logger
|
||||
from modules.settings import MOTD
|
||||
from modules.system import send_message
|
||||
|
||||
async def setup_scheduler(
|
||||
schedulerMotd, MOTD, schedulerMessage, schedulerChannel, schedulerInterface,
|
||||
schedulerValue, schedulerTime, schedulerInterval, logger, BroadcastScheduler):
|
||||
|
||||
# methods available for custom scheduler messages
|
||||
async def run_scheduler_loop(interval=1):
|
||||
logger.debug("Scheduler loop started")
|
||||
try:
|
||||
while True:
|
||||
try:
|
||||
schedule.run_pending()
|
||||
except Exception as e:
|
||||
logger.error(f"Scheduler loop exception: {e}")
|
||||
await asyncio.sleep(interval)
|
||||
except asyncio.CancelledError:
|
||||
logger.debug("Scheduler loop cancelled, shutting down.")
|
||||
|
||||
def safe_int(val, default=0, type=""):
|
||||
try:
|
||||
return int(val)
|
||||
except (ValueError, TypeError):
|
||||
logger.debug(f"System Scheduler: config {type} error '{val}' to int, using default {default}")
|
||||
return default
|
||||
|
||||
def setup_scheduler(
|
||||
schedulerMotd, MOTD, schedulerMessage, schedulerChannel, schedulerInterface,
|
||||
schedulerValue, schedulerTime, schedulerInterval):
|
||||
try:
|
||||
# Methods imported from mesh_bot for scheduling tasks
|
||||
from mesh_bot import (
|
||||
tell_joke,
|
||||
welcome_message,
|
||||
@@ -24,13 +42,15 @@ async def setup_scheduler(
|
||||
)
|
||||
except ImportError as e:
|
||||
logger.warning(f"Some mesh_bot schedule features are unavailable by option disable in config.ini: {e} comment out the use of these methods in your custom_scheduler.py")
|
||||
|
||||
|
||||
# Setup the scheduler based on configuration
|
||||
schedulerValue = schedulerValue.lower().strip()
|
||||
schedulerTime = schedulerTime.strip()
|
||||
schedulerInterval = schedulerInterval.strip()
|
||||
schedulerChannel = int(schedulerChannel)
|
||||
schedulerInterface = int(schedulerInterface)
|
||||
# Setup the scheduler based on configuration
|
||||
schedulerChannel = safe_int(schedulerChannel, 0, type="channel")
|
||||
schedulerInterface = safe_int(schedulerInterface, 1, type="interface")
|
||||
schedulerIntervalInt = safe_int(schedulerInterval, 5, type="interval")
|
||||
|
||||
try:
|
||||
if schedulerMotd:
|
||||
scheduler_message = MOTD
|
||||
@@ -45,7 +65,7 @@ async def setup_scheduler(
|
||||
if schedulerTime != '':
|
||||
schedule.every().day.at(schedulerTime).do(lambda: send_message(scheduler_message, schedulerChannel, 0, schedulerInterface))
|
||||
else:
|
||||
schedule.every(int(schedulerInterval)).days.do(lambda: send_message(scheduler_message, schedulerChannel, 0, schedulerInterface))
|
||||
schedule.every(schedulerIntervalInt).days.do(lambda: send_message(scheduler_message, schedulerChannel, 0, schedulerInterface))
|
||||
elif 'mon' in schedulerValue.lower() and schedulerTime != '':
|
||||
schedule.every().monday.at(schedulerTime).do(lambda: send_message(scheduler_message, schedulerChannel, 0, schedulerInterface))
|
||||
elif 'tue' in schedulerValue.lower() and schedulerTime != '':
|
||||
@@ -61,22 +81,22 @@ async def setup_scheduler(
|
||||
elif 'sun' in schedulerValue.lower() and schedulerTime != '':
|
||||
schedule.every().sunday.at(schedulerTime).do(lambda: send_message(scheduler_message, schedulerChannel, 0, schedulerInterface))
|
||||
elif 'hour' in schedulerValue.lower():
|
||||
schedule.every(int(schedulerInterval)).hours.do(lambda: send_message(scheduler_message, schedulerChannel, 0, schedulerInterface))
|
||||
schedule.every(schedulerIntervalInt).hours.do(lambda: send_message(scheduler_message, schedulerChannel, 0, schedulerInterface))
|
||||
elif 'min' in schedulerValue.lower():
|
||||
schedule.every(int(schedulerInterval)).minutes.do(lambda: send_message(scheduler_message, schedulerChannel, 0, schedulerInterface))
|
||||
logger.debug(f"System: Starting the basic scheduler to send '{scheduler_message}' on schedule '{schedulerValue}' every {schedulerInterval} interval at time '{schedulerTime}' on Device:{schedulerInterface} Channel:{schedulerChannel}")
|
||||
schedule.every(schedulerIntervalInt).minutes.do(lambda: send_message(scheduler_message, schedulerChannel, 0, schedulerInterface))
|
||||
logger.debug(f"System: Starting the basic scheduler to send '{scheduler_message}' on schedule '{schedulerValue}' every {schedulerIntervalInt} interval at time '{schedulerTime}' on Device:{schedulerInterface} Channel:{schedulerChannel}")
|
||||
elif 'joke' in schedulerValue.lower():
|
||||
# Schedule to send a joke every specified interval
|
||||
schedule.every(int(schedulerInterval)).minutes.do(lambda: send_message(tell_joke(), schedulerChannel, 0, schedulerInterface))
|
||||
logger.debug(f"System: Starting the joke scheduler to send a joke every {schedulerInterval} minutes on Device:{schedulerInterface} Channel:{schedulerChannel}")
|
||||
schedule.every(schedulerIntervalInt).minutes.do(lambda: send_message(tell_joke(), schedulerChannel, 0, schedulerInterface))
|
||||
logger.debug(f"System: Starting the joke scheduler to send a joke every {schedulerIntervalInt} minutes on Device:{schedulerInterface} Channel:{schedulerChannel}")
|
||||
elif 'link' in schedulerValue.lower():
|
||||
# Schedule to send a link message every specified interval
|
||||
schedule.every(int(schedulerInterval)).hours.do(lambda: send_message(handle_satpass(schedulerInterface, 'link'), schedulerChannel, 0, schedulerInterface))
|
||||
logger.debug(f"System: Starting the link scheduler to send link messages every {schedulerInterval} hours on Device:{schedulerInterface} Channel:{schedulerChannel}")
|
||||
schedule.every(schedulerIntervalInt).hours.do(lambda: send_message(handle_satpass(schedulerInterface, 'link'), schedulerChannel, 0, schedulerInterface))
|
||||
logger.debug(f"System: Starting the link scheduler to send link messages every {schedulerIntervalInt} hours on Device:{schedulerInterface} Channel:{schedulerChannel}")
|
||||
elif 'weather' in schedulerValue.lower():
|
||||
# Schedule to send weather updates every specified interval
|
||||
schedule.every(int(schedulerInterval)).hours.do(lambda: send_message(handle_wxc(0, schedulerInterface, 'wx'), schedulerChannel, 0, schedulerInterface))
|
||||
logger.debug(f"System: Starting the weather scheduler to send weather updates every {schedulerInterval} hours on Device:{schedulerInterface} Channel:{schedulerChannel}")
|
||||
schedule.every(schedulerIntervalInt).hours.do(lambda: send_message(handle_wxc(0, schedulerInterface, 'wx'), schedulerChannel, 0, schedulerInterface))
|
||||
logger.debug(f"System: Starting the weather scheduler to send weather updates every {schedulerIntervalInt} hours on Device:{schedulerInterface} Channel:{schedulerChannel}")
|
||||
elif 'custom' in schedulerValue.lower():
|
||||
# Import and setup custom schedules from custom_scheduler.py
|
||||
try:
|
||||
@@ -89,9 +109,5 @@ async def setup_scheduler(
|
||||
except Exception as e:
|
||||
logger.debug(f"System: Failed to import custom scheduler. {e}")
|
||||
logger.warning("Custom scheduler file not found or failed to import. cp etc/custom_scheduler.py modules/custom_scheduler.py")
|
||||
|
||||
# Start the Broadcast Scheduler
|
||||
await BroadcastScheduler()
|
||||
except Exception as e:
|
||||
logger.error(f"System: Scheduler Error {e}")
|
||||
|
||||
|
||||
@@ -255,10 +255,6 @@ if games_enabled is True:
|
||||
else:
|
||||
gamesCmdList = ""
|
||||
|
||||
# Scheduled Broadcast Configuration
|
||||
if scheduler_enabled:
|
||||
import schedule # pip install schedule
|
||||
|
||||
# Sentry Configuration
|
||||
if sentry_enabled:
|
||||
from math import sqrt
|
||||
@@ -1919,12 +1915,6 @@ def get_sysinfo(nodeID=0, deviceID=1):
|
||||
sysinfo += f"📊{stats}"
|
||||
return sysinfo
|
||||
|
||||
async def BroadcastScheduler():
|
||||
# handle schedule checks for the broadcast of messages
|
||||
while True:
|
||||
schedule.run_pending()
|
||||
await asyncio.sleep(1)
|
||||
|
||||
async def handleSignalWatcher():
|
||||
global lastHamLibAlert
|
||||
# monitor rigctld for signal strength and frequency
|
||||
|
||||
+158
-42
@@ -469,69 +469,185 @@ def onReceive(packet, interface):
|
||||
logger.debug(f"System: Error Packet = {packet}")
|
||||
|
||||
async def start_rx():
|
||||
print (CustomFormatter.bold_white + "\nMeshtastic Autoresponder Bot CTL+C to exit\n" + CustomFormatter.reset)
|
||||
# Start the receive subscriber using pubsub via meshtastic library
|
||||
pub.subscribe(onReceive, 'meshtastic.receive')
|
||||
pub.subscribe(onDisconnect, 'meshtastic.connection.lost')
|
||||
for i in range(1, 10):
|
||||
if globals().get(f'interface{i}_enabled', False):
|
||||
myNodeNum = globals().get(f'myNodeNum{i}', 0)
|
||||
logger.info(f"System: Autoresponder Started for Device{i} {get_name_from_number(myNodeNum, 'long', i)},"
|
||||
f"{get_name_from_number(myNodeNum, 'short', i)}. NodeID: {myNodeNum}, {decimal_to_hex(myNodeNum)}")
|
||||
|
||||
if my_settings.useDMForResponse:
|
||||
logger.debug(f"System: Respond by DM only")
|
||||
if my_settings.log_messages_to_file:
|
||||
logger.debug("System: Logging Messages to disk")
|
||||
if my_settings.syslog_to_file:
|
||||
logger.debug("System: Logging System Logs to disk")
|
||||
if my_settings.motd_enabled:
|
||||
logger.debug(f"System: MOTD Enabled using {my_settings.MOTD}")
|
||||
if my_settings.enableEcho:
|
||||
logger.debug(f"System: Echo command Enabled")
|
||||
if my_settings.sentry_enabled:
|
||||
logger.debug(f"System: Sentry Mode Enabled {my_settings.sentry_radius}m radius reporting to channel:{my_settings.secure_channel}")
|
||||
if my_settings.highfly_enabled:
|
||||
logger.debug(f"System: HighFly Enabled using {highfly_altitude}m limit reporting to channel:{my_settings.highfly_channel}")
|
||||
if my_settings.repeater_enabled and multiple_interface:
|
||||
logger.debug(f"System: Repeater Enabled for Channels: {my_settings.repeater_channels}")
|
||||
if my_settings.bbs_enabled:
|
||||
logger.debug(f"System: BBS Enabled, {bbsdb} has {len(bbs_messages)} messages. Direct Mail Messages waiting: {(len(bbs_dm) - 1)}")
|
||||
if my_settings.bbs_link_enabled:
|
||||
if len(my_settings.bbs_link_whitelist) > 0:
|
||||
logger.debug(f"System: BBS Link Enabled with {len(my_settings.bbs_link_whitelist)} peers")
|
||||
else:
|
||||
logger.debug("System: BBS Link Enabled allowing all")
|
||||
if my_settings.scheduler_enabled:
|
||||
# Examples of using the scheduler, Times here are in 24hr format
|
||||
# https://schedule.readthedocs.io/en/stable/
|
||||
|
||||
# Reminder Scheduler is enabled every Monday at noon send a log message
|
||||
schedule.every().monday.at("12:00").do(lambda: logger.info("System: Scheduled Broadcast Reminder"))
|
||||
logger.debug("System: Starting the broadcast scheduler")
|
||||
await BroadcastScheduler()
|
||||
|
||||
logger.debug("System: RX Subscriber started")
|
||||
# here we go loopty loo
|
||||
while True:
|
||||
await asyncio.sleep(0.5)
|
||||
pass
|
||||
|
||||
# Load the leaderboard data
|
||||
loadLeaderboard()
|
||||
def handle_boot(mesh=True):
|
||||
try:
|
||||
print (CustomFormatter.bold_white + f"\nMeshtastic Autoresponder Bot CTL+C to exit\n" + CustomFormatter.reset)
|
||||
if mesh:
|
||||
|
||||
for i in range(1, 10):
|
||||
if globals().get(f'interface{i}_enabled', False):
|
||||
myNodeNum = globals().get(f'myNodeNum{i}', 0)
|
||||
logger.info(f"System: Autoresponder Started for Device{i} {get_name_from_number(myNodeNum, 'long', i)},"
|
||||
f"{get_name_from_number(myNodeNum, 'short', i)}. NodeID: {myNodeNum}, {decimal_to_hex(myNodeNum)}")
|
||||
|
||||
if llm_enabled:
|
||||
logger.debug(f"System: Ollama LLM Enabled, loading model {my_settings.llmModel} please wait")
|
||||
llmLoad = llm_query(" ")
|
||||
if "trouble" not in llmLoad:
|
||||
logger.debug(f"System: LLM Model {my_settings.llmModel} loaded")
|
||||
|
||||
if my_settings.bbs_enabled:
|
||||
logger.debug(f"System: BBS Enabled, {bbsdb} has {len(bbs_messages)} messages. Direct Mail Messages waiting: {(len(bbs_dm) - 1)}")
|
||||
if my_settings.bbs_link_enabled:
|
||||
if len(bbs_link_whitelist) > 0:
|
||||
logger.debug(f"System: BBS Link Enabled with {len(bbs_link_whitelist)} peers")
|
||||
else:
|
||||
logger.debug(f"System: BBS Link Enabled allowing all")
|
||||
|
||||
if my_settings.solar_conditions_enabled:
|
||||
logger.debug("System: Celestial Telemetry Enabled")
|
||||
|
||||
if my_settings.location_enabled:
|
||||
if my_settings.use_meteo_wxApi:
|
||||
logger.debug("System: Location Telemetry Enabled using Open-Meteo API")
|
||||
else:
|
||||
logger.debug("System: Location Telemetry Enabled using NOAA API")
|
||||
print("debug my_settings.scheduler_enabled:", my_settings.scheduler_enabled)
|
||||
if my_settings.dad_jokes_enabled:
|
||||
logger.debug("System: Dad Jokes Enabled!")
|
||||
|
||||
if my_settings.coastalEnabled:
|
||||
logger.debug("System: Coastal Forecast and Tide Enabled!")
|
||||
|
||||
if games_enabled:
|
||||
logger.debug("System: Games Enabled!")
|
||||
|
||||
if my_settings.wikipedia_enabled:
|
||||
if my_settings.use_kiwix_server:
|
||||
logger.debug(f"System: Wikipedia search Enabled using Kiwix server at {kiwix_url}")
|
||||
else:
|
||||
logger.debug("System: Wikipedia search Enabled")
|
||||
|
||||
if my_settings.rssEnable:
|
||||
logger.debug(f"System: RSS Feed Reader Enabled for feeds: {rssFeedNames}")
|
||||
|
||||
if my_settings.radio_detection_enabled:
|
||||
logger.debug(f"System: Radio Detection Enabled using rigctld at {my_settings.rigControlServerAddress} broadcasting to channels: {my_settings.sigWatchBroadcastCh} for {get_freq_common_name(get_hamlib('f'))}")
|
||||
|
||||
if my_settings.file_monitor_enabled:
|
||||
logger.warning(f"System: File Monitor Enabled for {my_settings.file_monitor_file_path}, broadcasting to channels: {my_settings.file_monitor_broadcastCh}")
|
||||
if my_settings.enable_runShellCmd:
|
||||
logger.debug("System: Shell Command monitor enabled")
|
||||
if my_settings.allowXcmd:
|
||||
logger.warning("System: File Monitor shell XCMD Enabled")
|
||||
if my_settings.read_news_enabled:
|
||||
logger.debug(f"System: File Monitor News Reader Enabled for {my_settings.news_file_path}")
|
||||
if my_settings.bee_enabled:
|
||||
logger.debug("System: File Monitor Bee Monitor Enabled for bee.txt")
|
||||
|
||||
if my_settings.wxAlertBroadcastEnabled:
|
||||
logger.debug(f"System: Weather Alert Broadcast Enabled on channels {my_settings.wxAlertBroadcastChannel}")
|
||||
|
||||
if my_settings.emergencyAlertBrodcastEnabled:
|
||||
logger.debug(f"System: Emergency Alert Broadcast Enabled on channels {my_settings.emergencyAlertBroadcastCh} for FIPS codes {my_settings.myStateFIPSList}")
|
||||
if my_settings.myStateFIPSList == ['']:
|
||||
logger.warning("System: No FIPS codes set for iPAWS Alerts")
|
||||
|
||||
if my_settings.emergency_responder_enabled:
|
||||
logger.debug(f"System: Emergency Responder Enabled on channels {my_settings.emergency_responder_alert_channel} for interface {my_settings.emergency_responder_alert_interface}")
|
||||
|
||||
if my_settings.volcanoAlertBroadcastEnabled:
|
||||
logger.debug(f"System: Volcano Alert Broadcast Enabled on channels {my_settings.volcanoAlertBroadcastChannel}")
|
||||
|
||||
if my_settings.qrz_hello_enabled:
|
||||
if my_settings.train_qrz:
|
||||
logger.debug("System: QRZ Welcome/Hello Enabled with training mode")
|
||||
else:
|
||||
logger.debug("System: QRZ Welcome/Hello Enabled")
|
||||
|
||||
if my_settings.enableSMTP:
|
||||
if my_settings.enableImap:
|
||||
logger.debug("System: SMTP Email Alerting Enabled using IMAP")
|
||||
else:
|
||||
logger.warning("System: SMTP Email Alerting Enabled")
|
||||
|
||||
# Default Options
|
||||
if my_settings.useDMForResponse:
|
||||
logger.debug("System: Respond by DM only")
|
||||
|
||||
if my_settings.log_messages_to_file:
|
||||
logger.debug("System: Logging Messages to disk")
|
||||
if my_settings.syslog_to_file:
|
||||
logger.debug("System: Logging System Logs to disk")
|
||||
|
||||
if my_settings.motd_enabled:
|
||||
logger.debug(f"System: MOTD Enabled using {my_settings.MOTD} scheduler:{my_settings.schedulerMotd}")
|
||||
|
||||
if my_settings.sentry_enabled:
|
||||
logger.debug(f"System: Sentry Mode Enabled {my_settings.sentry_radius}m radius reporting to channel:{my_settings.secure_channel} requestLOC:{reqLocationEnabled}")
|
||||
if my_settings.sentryIgnoreList:
|
||||
logger.debug(f"System: Sentry BlockList Enabled for nodes: {my_settings.sentryIgnoreList}")
|
||||
if my_settings.sentryWatchList:
|
||||
logger.debug(f"System: Sentry WatchList Enabled for nodes: {my_settings.sentryWatchList}")
|
||||
|
||||
if my_settings.highfly_enabled:
|
||||
logger.debug(f"System: HighFly Enabled using {my_settings.highfly_altitude}m limit reporting to channel:{my_settings.highfly_channel}")
|
||||
|
||||
if my_settings.store_forward_enabled:
|
||||
logger.debug(f"System: S&F(messages command) Enabled using limit: {storeFlimit} and reverse queue:{my_settings.reverseSF}")
|
||||
|
||||
if my_settings.enableEcho:
|
||||
logger.debug("System: Echo command Enabled")
|
||||
|
||||
if my_settings.repeater_enabled and multiple_interface:
|
||||
logger.debug(f"System: Repeater Enabled for Channels: {my_settings.repeater_channels}")
|
||||
|
||||
if my_settings.checklist_enabled:
|
||||
logger.debug("System: CheckList Module Enabled")
|
||||
|
||||
if my_settings.ignoreChannels:
|
||||
logger.debug(f"System: Ignoring Channels: {my_settings.ignoreChannels}")
|
||||
|
||||
if my_settings.noisyNodeLogging:
|
||||
logger.debug("System: Noisy Node Logging Enabled")
|
||||
|
||||
if my_settings.logMetaStats:
|
||||
logger.debug("System: Logging Metadata Stats Enabled, leaderboard")
|
||||
|
||||
if my_settings.scheduler_enabled:
|
||||
logger.debug("System: Scheduler Enabled")
|
||||
|
||||
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"System: Error during boot: {e}")
|
||||
|
||||
|
||||
# Hello World
|
||||
async def main():
|
||||
tasks = []
|
||||
|
||||
try:
|
||||
handle_boot(mesh=False) # pong bot
|
||||
# Create core tasks
|
||||
tasks.append(asyncio.create_task(start_rx(), name="pong_rx"))
|
||||
tasks.append(asyncio.create_task(start_rx(), name="mesh_rx"))
|
||||
tasks.append(asyncio.create_task(watchdog(), name="watchdog"))
|
||||
|
||||
# Add optional tasks
|
||||
if my_settings.file_monitor_enabled:
|
||||
tasks.append(asyncio.create_task(handleFileWatcher(), name="file_monitor"))
|
||||
|
||||
if my_settings.radio_detection_enabled:
|
||||
tasks.append(asyncio.create_task(handleSignalWatcher(), name="hamlib"))
|
||||
|
||||
if my_settings.voxDetectionEnabled:
|
||||
tasks.append(asyncio.create_task(voxMonitor(), name="vox_detection"))
|
||||
|
||||
if my_settings.scheduler_enabled:
|
||||
from modules.scheduler import run_scheduler_loop, setup_scheduler
|
||||
setup_scheduler(schedulerMotd, MOTD, schedulerMessage, schedulerChannel, schedulerInterface,
|
||||
schedulerValue, schedulerTime, schedulerInterval)
|
||||
tasks.append(asyncio.create_task(run_scheduler_loop(), name="scheduler"))
|
||||
|
||||
logger.debug(f"System: Starting {len(tasks)} async tasks")
|
||||
|
||||
# Wait for all tasks with proper exception handling
|
||||
|
||||
Reference in New Issue
Block a user