Speed Max einstellbar

ExoMy_Software-master/scripts/exomy_admin_api.py

@@ -14,6 +14,7 @@ import time
 PORT = 8082
 EXOMY_CONTAINER = 'exomy_autostart'
 DELAY_FILE = '/tmp/exomy_delay.txt'
+SPEED_LIMIT_FILE = '/tmp/exomy_speed_limit.txt'
 MOTOR_TEST_LOCK = threading.Lock()
 
 
@@ -251,6 +252,33 @@ def set_delay(seconds):
     return seconds
 
 
+def get_speed_limit():
+    try:
+        with open(SPEED_LIMIT_FILE) as f:
+            return max(10, min(100, int(f.read().strip())))
+    except Exception:
+        return 100
+
+
+def set_speed_limit(percent):
+    percent = max(10, min(100, int(percent)))
+    data = str(percent).encode()
+    try:
+        fd = os.open(SPEED_LIMIT_FILE, os.O_WRONLY | os.O_TRUNC)
+    except FileNotFoundError:
+        fd = os.open(SPEED_LIMIT_FILE, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o666)
+    try:
+        os.write(fd, data)
+    finally:
+        os.close(fd)
+    subprocess.run(
+        ['docker', 'exec', EXOMY_CONTAINER, 'bash', '-c',
+         f'source /opt/ros/melodic/setup.bash && rosparam set /speed_limit_percent {percent}'],
+        capture_output=True, check=False
+    )
+    return percent
+
+
 def rumble_controller(duration_ms=400):
     # Linux FF constants
     EV_FF = 0x15
@@ -539,6 +567,9 @@ class Handler(http.server.BaseHTTPRequestHandler):
         if self.path == '/api/delay':
             self.write_json(200, {'delay_seconds': get_delay()})
             return
+        if self.path == '/api/speed-limit':
+            self.write_json(200, {'speed_limit_percent': get_speed_limit()})
+            return
         if self.path == '/api/status':
             self.write_json(200, collect_status())
             return
@@ -577,6 +608,24 @@ class Handler(http.server.BaseHTTPRequestHandler):
                 self.write_json(500, {'status': str(exc)})
             return
 
+        if self.path == '/api/speed-limit':
+            try:
+                content_length = int(self.headers.get('Content-Length', '0'))
+            except ValueError:
+                content_length = 0
+            raw_body = self.rfile.read(content_length) if content_length > 0 else b'{}'
+            try:
+                payload = json.loads(raw_body.decode('utf-8'))
+            except (UnicodeDecodeError, json.JSONDecodeError):
+                self.write_json(400, {'status': 'Ungültige Anfrage'})
+                return
+            try:
+                percent = set_speed_limit(payload.get('speed_limit_percent', 100))
+                self.write_json(200, {'status': 'Geschwindigkeitslimit gesetzt', 'speed_limit_percent': percent})
+            except Exception as exc:
+                self.write_json(500, {'status': str(exc)})
+            return
+
         if self.path == '/api/motor-test/drive-neutral/preview':
             try:
                 content_length = int(self.headers.get('Content-Length', '0'))