feat: centralize application entry point, add active test cooldown, and improve graceful shutdown.

This commit is contained in:
eddieoz
2025-11-27 12:22:14 +02:00
parent b75b3adf8d
commit 1cb71b2ea4
5 changed files with 71 additions and 5 deletions
+3 -3
View File
@@ -61,19 +61,19 @@ If `priority_nodes` is empty in `config.yaml`, the monitor will automatically se
### Basic Run (USB/Serial)
Connect your Meshtastic device via USB and run:
```bash
python3 -m mesh_monitor.monitor
python3 main.py
```
### Network Connection (TCP)
If your node is on the network (e.g., WiFi):
```bash
python3 -m mesh_monitor.monitor --tcp 192.168.1.10
python3 main.py --tcp 192.168.1.10
```
### Options
* `--ignore-no-position`: Suppress warnings about routers without a position (useful for portable routers or privacy).
```bash
python3 -m mesh_monitor.monitor --ignore-no-position
python3 main.py --ignore-no-position
```
## Configuration (Priority Testing)
Executable
+8
View File
@@ -0,0 +1,8 @@
#!/usr/bin/env python3
"""
Entry point for the Meshtastic Network Monitor.
"""
from mesh_monitor.monitor import main
if __name__ == "__main__":
main()
+3
View File
@@ -334,8 +334,10 @@ class ActiveTester:
'timestamp': time.time()
})
self._check_cycle_completion(node_id)
self._check_cycle_completion(node_id)
if self.pending_traceroute == node_id:
self.pending_traceroute = None # Clear pending if this was the node we were waiting for
self.last_test_time = time.time() # Start cooldown
def record_timeout(self, node_id):
"""
@@ -350,6 +352,7 @@ class ActiveTester:
self._check_cycle_completion(node_id)
if self.pending_traceroute == node_id:
self.pending_traceroute = None # Clear pending if this was the node we were waiting for
self.last_test_time = time.time() # Start cooldown
def _check_cycle_completion(self, node_id):
"""
+8 -2
View File
@@ -354,7 +354,7 @@ class MeshMonitor:
logger.error(f"Error in main loop: {e}")
time.sleep(10)
if __name__ == "__main__":
def main():
# Simple CLI for testing
import argparse
parser = argparse.ArgumentParser(description='Meshtastic Network Monitor')
@@ -367,4 +367,10 @@ if __name__ == "__main__":
else:
monitor = MeshMonitor(interface_type='serial', ignore_no_position=args.ignore_no_position)
monitor.start()
try:
monitor.start()
except KeyboardInterrupt:
monitor.stop()
if __name__ == "__main__":
main()
+49
View File
@@ -168,6 +168,55 @@ class TestNetworkMonitor(unittest.TestCase):
self.assertEqual(selected, expected)
print("Stratified Discovery Test Passed!")
def test_cooldown_logic(self):
print("\nRunning Cooldown Logic Test...")
mock_interface = MagicMock()
# Setup
tester = ActiveTester(mock_interface, test_interval=30, traceroute_timeout=60)
tester.priority_nodes = ["!n1", "!n2"]
# 1. Start Test 1
tester.run_next_test()
self.assertEqual(tester.pending_traceroute, "!n1")
start_time = tester.last_test_time
# 2. Simulate Timeout (at T+61)
# We need to mock time.time() to control flow
with patch('time.time') as mock_time:
# Initial call was at T0.
# Advance to T+61
mock_time.return_value = start_time + 61
# Run next test -> Should trigger timeout recording
tester.run_next_test()
# Verify timeout recorded
self.assertIsNone(tester.pending_traceroute)
self.assertEqual(tester.test_results[-1]['status'], 'timeout')
# Verify last_test_time updated to T+61 (Cooldown start)
self.assertEqual(tester.last_test_time, start_time + 61)
# 3. Try to run next test immediately (at T+62)
mock_time.return_value = start_time + 62
mock_interface.reset_mock()
tester.run_next_test()
# Should NOT send because 62 - 61 = 1 < 30
mock_interface.sendTraceRoute.assert_not_called()
print(" [Pass] Cooldown enforced after timeout")
# 4. Advance past cooldown (at T+92)
mock_time.return_value = start_time + 92
tester.run_next_test()
# Should send now
mock_interface.sendTraceRoute.assert_called_with("!n2", hopLimit=7)
print(" [Pass] Next test sent after cooldown")
print("Cooldown Logic Test Passed!")
print("\nRunning Test Interval Config Test...")
mock_interface = MagicMock()