tcsenpai/emesh
1
0
Fork 0
mirror of https://github.com/tcsenpai/emesh.git synced 2025-06-07 02:35:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

properly packed

Browse Source
This commit is contained in:
tcsenpai 2024-03-31 14:58:43 +02:00
parent 27ea41d299
commit 545af545bc
5 changed files with 110 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

123
gui.py
View File

@ -1,20 +1,17 @@
import json
import os import os
import time import time
from textual.app import App, ComposeResult from textual.app import App, ComposeResult
from textual.widgets import Header, Footer from textual.widgets import Header, Footer
from textual.widgets import Input, Label, Pretty from textual.widgets import Input, Label
from textual.widgets import Button, Static, RichLog, Sparkline, Checkbox from textual.widgets import Button, RichLog, Sparkline, Checkbox
from textual.containers import Horizontal, VerticalScroll from textual.containers import Horizontal, VerticalScroll
from textual.validation import Function, Number, ValidationResult, Validator from textual import events
from textual import events, on
import threading import threading
import term import term
from dotenv import load_dotenv
class MeshTerm(App): class MeshTerm(App):
CSS_PATH = "meshterm.tcss" CSS_PATH = "tcss/meshterm.tcss"
stopWatchdog = False stopWatchdog = False
messageToShow = None messageToShow = None
@ -26,36 +23,72 @@ class MeshTerm(App):
yield Footer() yield Footer()
# Inputs # Inputs
yield Horizontal(VerticalScroll( yield Horizontal(
Label("Enter the serial port to connect to: "), VerticalScroll(
Input(placeholder="/dev/ttyUSB0", id="port"), Label("Enter the serial port to connect to: "),
Button("Connect to radio", id="connect"), Input(placeholder="/dev/ttyUSB0", id="port"),
Checkbox("Enable beaconing:", True, id="beaconingBox"), Button("Connect to radio", id="connect"),
Checkbox("Enable beaconing:", True, id="beaconingBox"),
), ),
VerticalScroll( VerticalScroll(
Label("Unknown Radio Name", id="radio_name"), Label("Unknown Radio Name", id="radio_name"),
Label(""), Label(""),
Input(placeholder="Send something...", id="msg"), Input(placeholder="Send something...", id="msg"),
Button("Send", id="send", disabled=True) Button("Send", id="send", disabled=True),
)) ),
)
yield Horizontal(VerticalScroll( yield Horizontal(
Button("Exit", id="exit"),
Label("CONNECTED RADIO INFO"),
VerticalScroll( VerticalScroll(
Label("No radio connected", id="radio_namebox"), Button("Exit", id="exit"),
Label("", id="radio_id"), Label("CONNECTED RADIO INFO"),
Label("", id="radio_user"), VerticalScroll(
) Label("No radio connected", id="radio_namebox"),
), Label("", id="radio_id"),
Label("", id="radio_user"),
),
),
VerticalScroll( VerticalScroll(
Sparkline([1, 2, 3, 3, 3, 3, 3], summary_function=min,), Sparkline(
Label("Received messages:"), [1, 2, 3, 3, 3, 3, 3],
RichLog(id="received_messages", auto_scroll=True) summary_function=min,
)) ),
Label("Received messages:"),
RichLog(id="received_messages", auto_scroll=True),
),
)
yield Label("", id="message_to_show") yield Label("", id="message_to_show")
yield Sparkline([1, 2, 3, 3, 3, 3, 3, 3, 4, 4, 5, 5, 6, 5, 5, 4, 4, 3, 3, 3, 3, 3, 3, 3, 2, 1], summary_function=min,) yield Sparkline(
[
1,
2,
3,
3,
3,
3,
3,
3,
4,
4,
5,
5,
6,
5,
5,
4,
4,
3,
3,
3,
3,
3,
3,
3,
2,
1,
],
summary_function=min,
)
# Main log # Main log
yield RichLog(id="main_log", auto_scroll=True) yield RichLog(id="main_log", auto_scroll=True)
@ -71,14 +104,14 @@ class MeshTerm(App):
def on_button_pressed(self, event: Button.Pressed) -> None: def on_button_pressed(self, event: Button.Pressed) -> None:
"""Handle button events.""" """Handle button events."""
text_log = self.query_one("#main_log") # text_log = self.query_one("#main_log")
action = str(event.button.id).lower() action = str(event.button.id).lower()
if action == "exit": if action == "exit":
try: try:
term.forceQuit = True term.forceQuit = True
self.stopWatchdog = True self.stopWatchdog = True
except: except Exception as e:
print("[SYSTEM] Failed to stop thread") print("[SYSTEM] Failed to stop thread: %s" % e)
exit(1) exit(1)
elif action == "connect": elif action == "connect":
self.connect() self.connect()
@ -111,6 +144,7 @@ class MeshTerm(App):
def change_value(self, id, replacement): def change_value(self, id, replacement):
self.query_one(id).update(replacement) self.query_one(id).update(replacement)
# !SECTION Actions # !SECTION Actions
def loadEnv(self): def loadEnv(self):
@ -131,7 +165,6 @@ class MeshTerm(App):
f.flush() f.flush()
return self.env return self.env
def watcher(self): def watcher(self):
while not self.stopWatchdog: while not self.stopWatchdog:
time.sleep(1) time.sleep(1)
@ -140,17 +173,23 @@ class MeshTerm(App):
term.emesh.beaconingPrioritySettings = False term.emesh.beaconingPrioritySettings = False
term.emesh.beaconOn = self.query_one("#beaconingBox").value term.emesh.beaconOn = self.query_one("#beaconingBox").value
print("[WATCHDOG] Refreshing environment variables...") print("[WATCHDOG] Refreshing environment variables...")
os.environ['BEACONING'] = str(term.emesh.beaconOn) os.environ["BEACONING"] = str(term.emesh.beaconOn)
print("[WATCHDOG] Environment variables refreshed: " + str(os.environ['BEACONING'])) print(
"[WATCHDOG] Environment variables refreshed: "
+ str(os.environ["BEACONING"])
)
except Exception as e: except Exception as e:
print("[WARNING] beaconingBox element is not reachable - this may be temporary.") print(
"[WARNING] beaconingBox element is not reachable - this may be temporary."
)
print("[WARNING] Error: " + str(e))
# Loading messages into the gui # Loading messages into the gui
try: try:
if (term.outputs != term.last_output): if term.outputs != term.last_output:
term.last_output = term.outputs term.last_output = term.outputs
self.query_one("#main_log").write(term.outputs) self.query_one("#main_log").write(term.outputs)
# Priority to us here # Priority to us here
if (self.messageToShow): if self.messageToShow:
messageToShow = self.messageToShow messageToShow = self.messageToShow
self.messageToShow = None self.messageToShow = None
else: else:
@ -174,7 +213,13 @@ class MeshTerm(App):
# Populating the received messages # Populating the received messages
for receivd in term.emesh.msg_received: for receivd in term.emesh.msg_received:
if receivd["portnum"] == "TEXT_MESSAGE_APP": if receivd["portnum"] == "TEXT_MESSAGE_APP":
headerMessage = "[" + str(receivd["from"]) + " -> " + str(receivd["to"]) + "] > " headerMessage = (
"["
+ str(receivd["from"])
+ " -> "
+ str(receivd["to"])
+ "] > "
)
textToShow = headerMessage + receivd["text"] textToShow = headerMessage + receivd["text"]
self.query_one("#received_messages").write(textToShow) self.query_one("#received_messages").write(textToShow)
term.emesh.msg_received = [] term.emesh.msg_received = []

3
emesh.py → libs/emesh.py
View File

@ -35,8 +35,9 @@ def onReceive(packet, interface):
decoded = packet["decoded"] decoded = packet["decoded"]
decoded["from"] = packet["from"] decoded["from"] = packet["from"]
decoded["to"] = packet["to"] decoded["to"] = packet["to"]
except: except Exception as e:
print("[ERROR] Could not decode packet: discarding it") print("[ERROR] Could not decode packet: discarding it")
print("[ERROR] " + str(e))
return return
# ANCHOR We have received a packet and we decoded it # ANCHOR We have received a packet and we decoded it
print(decoded) print(decoded)

0
keys.py → libs/keys.py
View File

0
meshterm.tcss → tcss/meshterm.tcss
View File

31
term.py
View File

@ -1,4 +1,5 @@
import emesh import libs.emesh as emesh
import builtins as __builtin__
import time import time
import os import os
from dotenv import load_dotenv from dotenv import load_dotenv
@ -15,30 +16,33 @@ forceQuit = False
beaconCooldown = 0 beaconCooldown = 0
import builtins as __builtin__
# Overriding print for the GUI # Overriding print for the GUI
def print(*args, **kwargs): def print(*args, **kwargs):
global outputs global outputs
outputs = "".join(map(str, args)) outputs = "".join(map(str, args))
__builtin__.print(*args, **kwargs) __builtin__.print(*args, **kwargs)
# INFO Initializing the emesh structure # INFO Initializing the emesh structure
def init(): def init():
print("[SYSTEM] Starting EMesh...") print("[SYSTEM] Starting EMesh...")
vars = preparse() vars = preparse()
emesh.connect(vars['port']) emesh.connect(vars["port"])
print("[LOADER] Initialized") print("[LOADER] Initialized")
# INFO Parsing our environment variables # INFO Parsing our environment variables
def preparse(): def preparse():
load_dotenv() load_dotenv()
vars = {} vars = {}
# Parsing the port # Parsing the port
if not os.getenv('PORT') == "default": if not os.getenv("PORT") == "default":
vars['port'] = os.getenv('PORT') vars["port"] = os.getenv("PORT")
print(os.getenv('PORT')) print(os.getenv("PORT"))
return vars return vars
def main(): def main():
global beaconCooldown global beaconCooldown
global messageToShow global messageToShow
@ -49,10 +53,10 @@ def main():
print("[MAIN CYCLE] Starting watchdog...") print("[MAIN CYCLE] Starting watchdog...")
was_connected = False was_connected = False
cooldownHeader = False cooldownHeader = False
while not ((os.getenv('FORCE_QUIT')=="True") or forceQuit): while not ((os.getenv("FORCE_QUIT") == "True") or forceQuit):
# This is just a way to check if we need to notify the gui # This is just a way to check if we need to notify the gui
are_connected = emesh.connected are_connected = emesh.connected
if (are_connected!= was_connected): if are_connected != was_connected:
print("[GUI] Changed connection status") print("[GUI] Changed connection status")
messageToShow = "CONNECTION ESTABLISHED" messageToShow = "CONNECTION ESTABLISHED"
was_connected = are_connected was_connected = are_connected
@ -61,7 +65,7 @@ def main():
# NOTE Overriding is always possible, otherwise we have to rely on gui.py # NOTE Overriding is always possible, otherwise we have to rely on gui.py
if emesh.beaconingPrioritySettings: if emesh.beaconingPrioritySettings:
print("[MAIN CYCLE] Terminal mode: getting beaconing from .env...") print("[MAIN CYCLE] Terminal mode: getting beaconing from .env...")
emesh.beaconOn = (os.getenv('BEACONING')=="True") emesh.beaconOn = os.getenv("BEACONING") == "True"
else: else:
print("[MAIN CYCLE] GUI mode: getting beaconing from GUI...") print("[MAIN CYCLE] GUI mode: getting beaconing from GUI...")
print(f"[MAIN CYCLE] Beaconing: {emesh.beaconOn}") print(f"[MAIN CYCLE] Beaconing: {emesh.beaconOn}")
@ -71,27 +75,28 @@ def main():
if emesh.beaconOn: if emesh.beaconOn:
print("[MAIN CYCLE] Checking for beacon cooldown...") print("[MAIN CYCLE] Checking for beacon cooldown...")
# The following keeps the code running while we cooldown beaconing too # The following keeps the code running while we cooldown beaconing too
if (beaconCooldown > 0): if beaconCooldown > 0:
if not cooldownHeader: if not cooldownHeader:
print("+++ COOLDOWN ACTIVE +++") print("+++ COOLDOWN ACTIVE +++")
cooldownHeader = True cooldownHeader = True
isMultipleOfTen = (beaconCooldown % 10 == 0) isMultipleOfTen = beaconCooldown % 10 == 0
if isMultipleOfTen: if isMultipleOfTen:
print(f"[MAIN CYCLE] Beacon cooldown: {str(beaconCooldown)}") print(f"[MAIN CYCLE] Beacon cooldown: {str(beaconCooldown)}")
beaconCooldown -= 1 beaconCooldown -= 1
else: else:
print("*** COOLDOWN COMPLETE ***") print("*** COOLDOWN COMPLETE ***")
print("[MAIN CYCLE] Beaconing is activated, proceeding...") print("[MAIN CYCLE] Beaconing is activated, proceeding...")
beaconCooldown = int(os.getenv('BEACONING_INTERVAL')) beaconCooldown = int(os.getenv("BEACONING_INTERVAL"))
emesh.beacon() emesh.beacon()
print("[MAIN CYCLE] Beacon emitted. Proceeding to the next cycle...") print("[MAIN CYCLE] Beacon emitted. Proceeding to the next cycle...")
else: else:
print("[MAIN CYCLE] Beaconing is not activated, proceeding...") print("[MAIN CYCLE] Beaconing is not activated, proceeding...")
# Sleep for N seconds # Sleep for N seconds
# print("[MAIN CYCLE] Sleeping for " + os.getenv('SLEEP_INTERVAL') + " seconds") # print("[MAIN CYCLE] Sleeping for " + os.getenv('SLEEP_INTERVAL') + " seconds")
time.sleep(int(os.getenv('SLEEP_INTERVAL'))) time.sleep(int(os.getenv("SLEEP_INTERVAL")))
# print("[MAIN CYCLE] Sleeping complete. Proceeding to the next cycle...") # print("[MAIN CYCLE] Sleeping complete. Proceeding to the next cycle...")
print("[SYSTEM] Ready to start.") print("[SYSTEM] Ready to start.")