2 Commits

Author SHA1 Message Date
Cynopolis
185182b92f Added a callback manager (#2)
Co-authored-by: Cynopolis <megaveganzombie@gmail.com>
2025-04-02 13:44:42 -04:00
Cynopolis
976d2453a2 Added an installer script
* Added an installer script

* Finished deployment script

---------

Co-authored-by: Cynopolis <megaveganzombie@gmail.com>
2025-04-02 11:54:23 -04:00
2 changed files with 68 additions and 45 deletions

18
CommandHandler.py Normal file
View File

@@ -0,0 +1,18 @@
class CommandHandler:
def __init__(self, print_method):
self.commands: dict = {}
self.print = print_method
def register_callback(self, key: str, callback):
self.commands[key.lower()] = callback
def parse_command(self, command : str):
args :list[str] = command.split(" ")
if args[0].lower() in self.commands:
return_message = self.commands[args[0].lower()](args[1:])
if not (return_message is None):
self.print(return_message)
else:
self.print("Command not recognized. Type 'help' for a list of commands.")

93
app.py
View File

@@ -1,8 +1,12 @@
from flask import Flask, render_template, request, jsonify
import meshtastic
from MeshtasticLogger import MeshtasticLogger
from CommandHandler import CommandHandler
import os
# -------------------------------
# Initialize data and objects
# -------------------------------
messages = []
app = Flask(__name__)
interface = meshtastic.serial_interface.SerialInterface()
@@ -23,53 +27,56 @@ def web_print(message: str):
messages.append(message)
get_messages()
def callback(message: str):
web_print(message)
command_handler: CommandHandler = CommandHandler(web_print)
logger = MeshtasticLogger(interface, "logs/mesh_logs.log", channel=1, message_received_callback=web_print)
def help_response():
help_message: str = "Commands:\nhelp - show this message\clear - clear the text log\nsend <text> - sends text to all channels\nchannel <number> - sets the current channel (0-7)\n"
web_print(help_message)
logger = MeshtasticLogger(interface, "logs/mesh_logs.log", channel=1, message_received_callback=callback)
# -------------------------
# Command Callback Handlers
# -------------------------
def help_response(args: list[str]):
return "Commands:\nhelp - show this message\clear - clear the text log\nsend <text> - sends text to all channels\nchannel <number> - sets the current channel (0-7)\n"
def channel_response(args: list[str]):
if(len(args) < 2):
web_print(f"Current channel number: {logger.channel}")
return
channel: int = int(args[1])
if channel > 7 or channel < 0:
web_print("Channel must be between 0 and 7")
else:
try:
logger.channel = channel
except Exception as e:
web_print(e)
else:
web_print(f"Channel set to {channel}")
def send_response(args: list[str], command: str):
if len(args) < 2:
web_print("Please provide text to send")
else:
logger.send(command[5:])
web_print(f"Sent: {command[5:]}")
def parse_command(command):
args = command.split(" ")
if len(args) < 1:
return True
return f"Current channel number: {logger.channel}"
channel: int = int(args[0])
if channel > 7 or channel < 0:
return "Channel must be between 0 and 7"
if args[0] == "clear":
messages.clear()
elif args[0] == "help":
help_response()
elif args[0] == "channel":
channel_response(args)
elif args[0] == "send":
send_response(args, command)
else:
web_print("Command not recognized. Type \'help\' for a list of commands.")
try:
logger.channel = channel
except Exception as e:
return e
return f"Channel set to {channel}"
def send_response(args: list[str]):
if len(args) == 0:
return "Please provide text to send"
command = ""
for arg in args:
command += arg + " "
command = command[:-1] # remove the trailing space
logger.send(command)
return_message = f"Sent: {command}"
return return_message
def clear_response(args):
messages.clear()
return None
# register callbacks
command_handler.register_callback("clear", clear_response)
command_handler.register_callback("help", help_response)
command_handler.register_callback("channel", channel_response)
command_handler.register_callback("send", send_response)
# -----------------------
# Flask Callback Handlers
# -----------------------
@app.route('/')
def index():
return render_template('index.html')
@@ -78,12 +85,10 @@ def index():
def send():
data = request.json
message = data.get("message", "")
parse_command(message)
command_handler.parse_command(message)
return jsonify({"messages": messages})
@app.route('/messages', methods=['GET'])
def get_messages():
return jsonify({"messages": messages})
if __name__ == '__main__':
app.run(host="0.0.0.0", port=80, debug=True)