Add RS485 transport simulation layer

2026-06-03 00:52:33 -06:00
parent 131589fe6e
commit eff60ed9d2
4 changed files with 133 additions and 24 deletions
@@ -0,0 +1,43 @@
+import random
+import time
+
+
+class RS485Transport:
+    def __init__(self, controller):
+        self.controller = controller
+        self.connected = True
+        self.latency_ms = 0
+        self.packet_loss_percent = 0
+
+    def send(self, message):
+        if not self.connected:
+            return {
+                "type": "error",
+                "success": False,
+                "error": "RS-485 disconnected"
+            }
+
+        if self.latency_ms > 0:
+            time.sleep(self.latency_ms / 1000)
+
+        if self.packet_loss_percent > 0:
+            if random.randint(1, 100) <= self.packet_loss_percent:
+                return {
+                    "type": "error",
+                    "success": False,
+                    "error": "RS-485 packet lost"
+                }
+
+        return self.controller.handle_message(message)
+
+    def disconnect(self):
+        self.connected = False
+
+    def restore(self):
+        self.connected = True
+
+    def set_latency(self, latency_ms):
+        self.latency_ms = max(0, int(latency_ms))
+
+    def set_packet_loss(self, percent):
+        self.packet_loss_percent = max(0, min(100, int(percent)))