Crabmodus OK

2026-05-20 08:47:10 +02:00
parent e0cbed612a
commit ced9da92a5
8 changed files with 2314 additions and 23 deletions
@@ -4,10 +4,12 @@ import json
 import shutil
 import socketserver
 import subprocess
+import threading
 import time

 PORT = 8082
 EXOMY_CONTAINER = 'exomy_autostart'
+MOTOR_TEST_LOCK = threading.Lock()


 class ReusableThreadingTCPServer(socketserver.ThreadingTCPServer):
@@ -95,6 +97,15 @@ def get_container_state(container_name):
    return 'unknown'


+def get_motor_test_container_status():
+    state = get_container_state(EXOMY_CONTAINER)
+    return {
+        'container': EXOMY_CONTAINER,
+        'state': state,
+        'is_safe_for_motor_test': state in ('exited', 'stopped', 'created'),
+    }
+
+
 def collect_status():
    return {
        'status': 'Bereit',
@@ -121,6 +132,16 @@ def restart_exomy_container():
    subprocess.Popen(['docker', 'restart', EXOMY_CONTAINER])


+def stop_exomy_container():
+    stdout, _, returncode = run_command(['docker', 'stop', '-t', '2', EXOMY_CONTAINER])
+    return returncode == 0, stdout
+
+
+def start_exomy_container():
+    stdout, _, returncode = run_command(['docker', 'start', EXOMY_CONTAINER])
+    return returncode == 0, stdout
+
+
 def reboot_raspberry_pi():
    subprocess.Popen(['systemctl', 'reboot'])

@@ -129,6 +150,189 @@ def shutdown_raspberry_pi():
    subprocess.Popen(['systemctl', 'poweroff'])


+def run_motor_test_action(wheel, action):
+    from admin_motor_test import run_motor_test
+
+    acquired = MOTOR_TEST_LOCK.acquire(blocking=False)
+    if not acquired:
+        return 409, {'status': 'Motortest läuft bereits'}
+
+    try:
+        container_status = get_motor_test_container_status()
+        if not container_status['is_safe_for_motor_test']:
+            return 409, {
+                'status': 'ExoMy-Container läuft noch. Bitte zuerst auf der Motortest-Seite stoppen.',
+                'container_status': container_status,
+            }
+
+        result = run_motor_test(wheel, action)
+        return 200, {
+            'status': f"{result['wheel_label']}: {result['action_label']}",
+            'result': result,
+            'container_status': get_motor_test_container_status(),
+        }
+    except FileNotFoundError:
+        return 500, {'status': 'Motor-Konfiguration fehlt'}
+    except RuntimeError as exc:
+        return 500, {'status': str(exc)}
+    except ValueError as exc:
+        return 400, {'status': str(exc)}
+    except Exception:
+        return 500, {'status': 'Motortest fehlgeschlagen'}
+    finally:
+        MOTOR_TEST_LOCK.release()
+
+
+def get_steering_neutral_status():
+    from admin_motor_test import get_steering_neutral_values
+
+    container_status = get_motor_test_container_status()
+    if not container_status['is_safe_for_motor_test']:
+        return 409, {
+            'status': 'ExoMy-Container läuft noch. Bitte zuerst auf der Motortest-Seite stoppen.',
+            'container_status': container_status,
+        }
+
+    try:
+        result = get_steering_neutral_values()
+        return 200, {
+            'status': 'Servo-Mitten geladen',
+            'neutral_values': result,
+            'container_status': container_status,
+        }
+    except FileNotFoundError:
+        return 500, {'status': 'Motor-Konfiguration fehlt'}
+    except Exception:
+        return 500, {'status': 'Servo-Mitten konnten nicht geladen werden'}
+
+
+def preview_steering_neutral_action(wheel, value):
+    from admin_motor_test import preview_steering_neutral_value
+
+    container_status = get_motor_test_container_status()
+    if not container_status['is_safe_for_motor_test']:
+        return 409, {
+            'status': 'ExoMy-Container läuft noch. Bitte zuerst auf der Motortest-Seite stoppen.',
+            'container_status': container_status,
+        }
+
+    try:
+        result = preview_steering_neutral_value(str(wheel).lower(), int(value))
+        return 200, {
+            'status': f"{result['wheel_label']}: PWM {result['value']}",
+            'preview': result,
+            'container_status': container_status,
+        }
+    except ValueError as exc:
+        return 400, {'status': str(exc)}
+    except RuntimeError as exc:
+        return 500, {'status': str(exc)}
+    except Exception:
+        return 500, {'status': 'Servo-Vorschau fehlgeschlagen'}
+
+
+def save_steering_neutral_action(values):
+    from admin_motor_test import save_steering_neutral_values
+
+    container_status = get_motor_test_container_status()
+    if not container_status['is_safe_for_motor_test']:
+        return 409, {
+            'status': 'ExoMy-Container läuft noch. Bitte zuerst auf der Motortest-Seite stoppen.',
+            'container_status': container_status,
+        }
+
+    try:
+        result = save_steering_neutral_values(values)
+        return 200, {
+            'status': 'Servo-Mitten gespeichert',
+            'neutral_values': result,
+            'container_status': container_status,
+        }
+    except ValueError as exc:
+        return 400, {'status': str(exc)}
+    except RuntimeError as exc:
+        return 500, {'status': str(exc)}
+    except FileNotFoundError:
+        return 500, {'status': 'Motor-Konfiguration fehlt'}
+    except Exception:
+        return 500, {'status': 'Servo-Mitten konnten nicht gespeichert werden'}
+
+
+def get_drive_neutral_status():
+    from admin_motor_test import get_drive_neutral_value
+
+    container_status = get_motor_test_container_status()
+    if not container_status['is_safe_for_motor_test']:
+        return 409, {
+            'status': 'ExoMy-Container läuft noch. Bitte zuerst auf der Motortest-Seite stoppen.',
+            'container_status': container_status,
+        }
+
+    try:
+        result = get_drive_neutral_value()
+        return 200, {
+            'status': 'Fahr-Neutralwert geladen',
+            'drive_neutral': result,
+            'container_status': container_status,
+        }
+    except FileNotFoundError:
+        return 500, {'status': 'Motor-Konfiguration fehlt'}
+    except Exception:
+        return 500, {'status': 'Fahr-Neutralwert konnte nicht geladen werden'}
+
+
+def preview_drive_neutral_action(value):
+    from admin_motor_test import preview_drive_neutral_value
+
+    container_status = get_motor_test_container_status()
+    if not container_status['is_safe_for_motor_test']:
+        return 409, {
+            'status': 'ExoMy-Container läuft noch. Bitte zuerst auf der Motortest-Seite stoppen.',
+            'container_status': container_status,
+        }
+
+    try:
+        result = preview_drive_neutral_value(int(value))
+        return 200, {
+            'status': f"Fahr-Neutralwert: PWM {result['value']}",
+            'drive_neutral': result,
+            'container_status': container_status,
+        }
+    except ValueError as exc:
+        return 400, {'status': str(exc)}
+    except RuntimeError as exc:
+        return 500, {'status': str(exc)}
+    except Exception:
+        return 500, {'status': 'Fahr-Vorschau fehlgeschlagen'}
+
+
+def save_drive_neutral_action(value):
+    from admin_motor_test import save_drive_neutral_value
+
+    container_status = get_motor_test_container_status()
+    if not container_status['is_safe_for_motor_test']:
+        return 409, {
+            'status': 'ExoMy-Container läuft noch. Bitte zuerst auf der Motortest-Seite stoppen.',
+            'container_status': container_status,
+        }
+
+    try:
+        result = save_drive_neutral_value(int(value))
+        return 200, {
+            'status': 'Fahr-Neutralwert gespeichert',
+            'drive_neutral': result,
+            'container_status': container_status,
+        }
+    except ValueError as exc:
+        return 400, {'status': str(exc)}
+    except RuntimeError as exc:
+        return 500, {'status': str(exc)}
+    except FileNotFoundError:
+        return 500, {'status': 'Motor-Konfiguration fehlt'}
+    except Exception:
+        return 500, {'status': 'Fahr-Neutralwert konnte nicht gespeichert werden'}
+
+
 class Handler(http.server.BaseHTTPRequestHandler):
    def end_headers(self):
        self.send_header('Access-Control-Allow-Origin', '*')
@@ -152,9 +356,154 @@ class Handler(http.server.BaseHTTPRequestHandler):
        if self.path == '/api/status':
            self.write_json(200, collect_status())
            return
+        if self.path == '/api/motor-test/status':
+            self.write_json(200, {
+                'status': 'Bereit',
+                'container_status': get_motor_test_container_status(),
+            })
+            return
+        if self.path == '/api/motor-test/steering-neutral':
+            status_code, response = get_steering_neutral_status()
+            self.write_json(status_code, response)
+            return
+        if self.path == '/api/motor-test/drive-neutral':
+            status_code, response = get_drive_neutral_status()
+            self.write_json(status_code, response)
+            return
        self.write_json(404, {'status': 'Unbekannt'})

    def do_POST(self):
+        if self.path == '/api/motor-test/drive-neutral/preview':
+            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
+
+            status_code, response = preview_drive_neutral_action(payload.get('value', 0))
+            self.write_json(status_code, response)
+            return
+
+        if self.path == '/api/motor-test/drive-neutral/save':
+            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
+
+            status_code, response = save_drive_neutral_action(payload.get('value', 0))
+            self.write_json(status_code, response)
+            return
+
+        if self.path == '/api/motor-test/steering-neutral/preview':
+            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
+
+            status_code, response = preview_steering_neutral_action(
+                payload.get('wheel', ''),
+                payload.get('value', 0)
+            )
+            self.write_json(status_code, response)
+            return
+
+        if self.path == '/api/motor-test/steering-neutral/save':
+            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
+
+            values = payload.get('values')
+            if not isinstance(values, dict):
+                self.write_json(400, {'status': 'Neutralwerte fehlen'})
+                return
+
+            status_code, response = save_steering_neutral_action(values)
+            self.write_json(status_code, response)
+            return
+
+        if self.path == '/api/motor-test/stop-container':
+            stopped, _ = stop_exomy_container()
+            container_status = get_motor_test_container_status()
+            if not stopped and not container_status['is_safe_for_motor_test']:
+                self.write_json(500, {
+                    'status': 'ExoMy-Container konnte nicht gestoppt werden',
+                    'container_status': container_status,
+                })
+                return
+            self.write_json(200, {
+                'status': 'ExoMy-Container ist gestoppt',
+                'container_status': container_status,
+            })
+            return
+
+        if self.path == '/api/motor-test/start-container':
+            started, _ = start_exomy_container()
+            container_status = get_motor_test_container_status()
+            if not started and container_status['state'] != 'running':
+                self.write_json(500, {
+                    'status': 'ExoMy-Container konnte nicht gestartet werden',
+                    'container_status': container_status,
+                })
+                return
+            self.write_json(200, {
+                'status': 'ExoMy-Container läuft wieder',
+                'container_status': container_status,
+            })
+            return
+
+        if self.path == '/api/motor-test':
+            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
+
+            status_code, response = run_motor_test_action(
+                str(payload.get('wheel', '')).lower(),
+                str(payload.get('action', ''))
+            )
+            self.write_json(status_code, response)
+            return
+
        actions = {
            '/api/restart-camera': {
                'handler': restart_camera_service,