#!/usr/bin/env python3 import datetime import errno import fcntl import glob import http.server import json import os import shutil import socketserver import struct import subprocess import termios import threading import time from zoneinfo import ZoneInfo PORT = 8082 EXOMY_CONTAINER = 'exomy_autostart' DELAY_FILE = '/tmp/exomy_delay.txt' SPEED_LIMIT_FILE = '/tmp/exomy_speed_limit.txt' DRIVE_ESTIMATOR_CALIBRATION_FILE = os.path.join( os.path.dirname(__file__), '..', 'config', 'drive_estimator_calibration.json' ) ROUTE_LIBRARY_FILE = os.path.join( os.path.dirname(__file__), '..', 'config', 'route_library.json' ) CAMERA_RUNTIME_SETTINGS_FILE = '/tmp/exomy_camera_settings.json' MOTOR_TEST_LOCK = threading.Lock() GPS_PORT_PATTERNS = [ '/dev/serial/by-id/*', '/dev/serial/by-path/*', '/dev/ttyUSB*', '/dev/ttyACM*', ] GPS_BAUD_RATE = 4800 GPS_COLD_START_COMMAND = '$PAIR006*3C\r\n' CAMERA_PROFILES = [ { 'id': '640x480', 'label': '640 x 480', 'description': '4:3, klein und sparsam', 'width': 640, 'height': 480, 'fps': 10, }, { 'id': '1024x768', 'label': '1024 x 768', 'description': '4:3, großes Sichtfeld mit mehr Details', 'width': 1024, 'height': 768, 'fps': 10, }, { 'id': '1296x972', 'label': '1296 x 972', 'description': '4:3, großes Sichtfeld mit noch mehr Details', 'width': 1296, 'height': 972, 'fps': 10, }, ] CAMERA_PROFILE_MAP = {profile['id']: profile for profile in CAMERA_PROFILES} DEFAULT_CAMERA_PROFILE_ID = '1296x972' CAMERA_FPS_OPTIONS = [1, 2, 3, 4, 5, 7, 10, 20] DEFAULT_CAMERA_FPS = 10 class ReusableThreadingTCPServer(socketserver.ThreadingTCPServer): allow_reuse_address = True def run_command(command): result = subprocess.run(command, capture_output=True, text=True, check=False) return result.stdout.strip(), result.stderr.strip(), result.returncode def read_cpu_temperature(): try: with open('/sys/class/thermal/thermal_zone0/temp', 'r', encoding='utf-8') as handle: raw_value = handle.read().strip() return f'{int(raw_value) / 1000:.1f} °C' except (FileNotFoundError, ValueError): return 'unbekannt' def read_cpu_usage(): try: with open('/proc/stat', 'r', encoding='utf-8') as handle: first = handle.readline().split()[1:] first = [int(value) for value in first] idle_first = first[3] + first[4] total_first = sum(first) time.sleep(0.12) with open('/proc/stat', 'r', encoding='utf-8') as handle: second = handle.readline().split()[1:] second = [int(value) for value in second] idle_second = second[3] + second[4] total_second = sum(second) total_delta = total_second - total_first idle_delta = idle_second - idle_first if total_delta <= 0: return 'unbekannt' usage = (1 - (idle_delta / total_delta)) * 100 return f'{usage:.0f}%' except (FileNotFoundError, ValueError, IndexError): return 'unbekannt' def read_memory_usage(): try: values = {} with open('/proc/meminfo', 'r', encoding='utf-8') as handle: for line in handle: key, raw_value = line.split(':', 1) values[key] = int(raw_value.strip().split()[0]) total = values.get('MemTotal') available = values.get('MemAvailable') if not total or available is None: return 'unbekannt' used = total - available percent = (used / total) * 100 return f'{percent:.0f}%' except (FileNotFoundError, ValueError): return 'unbekannt' def read_undervoltage_status(): stdout, _, returncode = run_command(['vcgencmd', 'get_throttled']) if returncode != 0 or not stdout or '=' not in stdout: return { 'text': 'unbekannt', 'state': 'unknown', } try: throttled_value = int(stdout.split('=', 1)[1].strip(), 16) except ValueError: return { 'text': 'unbekannt', 'state': 'unknown', } undervoltage_now = bool(throttled_value & 0x1) undervoltage_occurred = bool(throttled_value & 0x10000) if undervoltage_now: return { 'text': 'Ja, aktuell', 'state': 'active', } if undervoltage_occurred: return { 'text': 'Früher erkannt', 'state': 'past', } return { 'text': 'Nein', 'state': 'clear', } def format_uptime(): try: with open('/proc/uptime', 'r', encoding='utf-8') as handle: uptime_seconds = int(float(handle.read().split()[0])) except (FileNotFoundError, ValueError, IndexError): return 'unbekannt' days, remainder = divmod(uptime_seconds, 86400) hours, remainder = divmod(remainder, 3600) minutes, _ = divmod(remainder, 60) parts = [] if days: parts.append(f'{days}d') if days or hours: parts.append(f'{hours}h') parts.append(f'{minutes}m') return ' '.join(parts) def format_disk_free(): usage = shutil.disk_usage('/') free_gb = usage.free / (1024 ** 3) total_gb = usage.total / (1024 ** 3) return f'{free_gb:.1f} GB frei / {total_gb:.1f} GB' def get_ip_addresses(): stdout, _, returncode = run_command(['ip', '-4', '-o', 'addr', 'show', 'dev', 'wlan0', 'scope', 'global']) if returncode == 0 and stdout: for line in stdout.splitlines(): parts = line.split() if 'inet' in parts: inet_index = parts.index('inet') if inet_index + 1 < len(parts): return parts[inet_index + 1].split('/')[0] stdout, _, returncode = run_command(['hostname', '-I']) if returncode != 0 or not stdout: return 'unbekannt' ipv4_addresses = [] for item in stdout.split(): if item.count('.') == 3 and not item.startswith('172.17.'): ipv4_addresses.append(item) return ipv4_addresses[0] if ipv4_addresses else 'unbekannt' def get_wifi_status(): stdout, _, returncode = run_command(['nmcli', '-t', '-f', 'DEVICE,STATE,CONNECTION', 'device']) if returncode != 0 or not stdout: return 'unbekannt' for line in stdout.splitlines(): parts = line.split(':', 2) if len(parts) != 3: continue device, state, connection = parts if device == 'wlan0': if state == 'connected': ssid_stdout, _, ssid_returncode = run_command(['nmcli', '-t', '-f', 'ACTIVE,SSID', 'device', 'wifi', 'list', 'ifname', 'wlan0']) if ssid_returncode == 0 and ssid_stdout: for ssid_line in ssid_stdout.splitlines(): if ssid_line.startswith('yes:'): active_ssid = ssid_line.split(':', 1)[1].strip() if active_ssid: return f'verbunden: {active_ssid}' return f'verbunden: {connection}' return state return 'kein wlan0' def get_wifi_signal(): stdout, _, returncode = run_command(['nmcli', '-t', '-f', 'GENERAL.STATE,AP.SIGNAL', 'device', 'show', 'wlan0']) if returncode != 0 or not stdout: return 'unbekannt' state = '' signal = '' for line in stdout.splitlines(): if line.startswith('GENERAL.STATE:'): state = line.split(':', 1)[1].strip() elif line.startswith('AP[1].SIGNAL:'): signal = line.split(':', 1)[1].strip() if 'connected' not in state.lower(): return 'nicht verbunden' if not signal: return 'unbekannt' try: percent = max(0, min(100, int(float(signal)))) except ValueError: return 'unbekannt' return f'{percent} %' def get_systemd_state(service_name): stdout, _, returncode = run_command(['systemctl', 'is-active', service_name]) if returncode == 0 and stdout: return stdout return stdout or 'unknown' def get_container_state(container_name): stdout, _, returncode = run_command(['docker', 'inspect', '-f', '{{.State.Status}}', container_name]) if returncode == 0 and stdout: return stdout 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(): undervoltage = read_undervoltage_status() return { 'status': 'Bereit', 'system': { 'wifi': get_wifi_status(), 'wifi_signal': get_wifi_signal(), 'ips': get_ip_addresses(), 'undervoltage': undervoltage['text'], 'undervoltage_state': undervoltage['state'], 'cpu_temperature': read_cpu_temperature(), 'cpu_usage': read_cpu_usage(), 'memory_usage': read_memory_usage(), 'uptime': format_uptime(), 'disk_free': format_disk_free(), }, 'services': { 'camera': get_systemd_state('exomy-camera-stream.service'), 'camera_settings_version': read_camera_settings_version(), 'admin_api': get_systemd_state('exomy-admin-api.service'), 'exomy': get_container_state(EXOMY_CONTAINER), } } def get_delay(): try: with open(DELAY_FILE) as f: return max(0.0, min(10.0, float(f.read().strip()))) except Exception: return 0.0 def set_delay(seconds): seconds = max(0.0, min(10.0, float(seconds))) data = str(seconds).encode() try: fd = os.open(DELAY_FILE, os.O_WRONLY | os.O_TRUNC) except FileNotFoundError: fd = os.open(DELAY_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 /delay_seconds {seconds}'], capture_output=True, check=False ) 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 get_default_camera_profile(): return dict(CAMERA_PROFILE_MAP[DEFAULT_CAMERA_PROFILE_ID]) def normalize_camera_profile(profile_id): profile = CAMERA_PROFILE_MAP.get(str(profile_id or '').strip()) if profile is None: raise ValueError('Unbekanntes Kamera-Profil') return dict(profile) def normalize_camera_fps(value): try: fps = int(value) except (TypeError, ValueError): raise ValueError('Unbekannte Bildrate') if fps not in CAMERA_FPS_OPTIONS: raise ValueError('Unbekannte Bildrate') return fps def read_camera_settings(): default_profile = get_default_camera_profile() try: with open(CAMERA_RUNTIME_SETTINGS_FILE, 'r', encoding='utf-8') as handle: payload = json.load(handle) except (FileNotFoundError, json.JSONDecodeError, OSError, ValueError, TypeError): return default_profile try: width = int(payload.get('width', default_profile['width'])) height = int(payload.get('height', default_profile['height'])) fps = normalize_camera_fps(payload.get('fps', DEFAULT_CAMERA_FPS)) except (TypeError, ValueError): fps = DEFAULT_CAMERA_FPS selected_profile_id = str(payload.get('profile', '')).strip() if selected_profile_id in CAMERA_PROFILE_MAP: profile = dict(CAMERA_PROFILE_MAP[selected_profile_id]) else: profile = None for item in CAMERA_PROFILES: if item['width'] == width and item['height'] == height: profile = dict(item) break if profile is None: profile = default_profile profile['fps'] = fps return profile def write_camera_settings(profile, fps): profile = dict(profile) fps = normalize_camera_fps(fps) payload = { 'profile': profile['id'], 'width': profile['width'], 'height': profile['height'], 'fps': fps, 'updated_at': int(time.time()), } with open(CAMERA_RUNTIME_SETTINGS_FILE, 'w', encoding='utf-8') as handle: json.dump(payload, handle, indent=2, sort_keys=True) handle.write('\n') def read_camera_settings_version(): try: with open(CAMERA_RUNTIME_SETTINGS_FILE, 'r', encoding='utf-8') as handle: payload = json.load(handle) except (FileNotFoundError, json.JSONDecodeError, OSError, ValueError, TypeError): return 0 try: return int(payload.get('updated_at', 0)) except (TypeError, ValueError): return 0 def get_camera_config(): profile = read_camera_settings() return { 'status': 'Bereit', 'camera': { 'selected_profile': profile['id'], 'width': profile['width'], 'height': profile['height'], 'fps': profile['fps'], 'fps_options': list(CAMERA_FPS_OPTIONS), 'profiles': [dict(item) for item in CAMERA_PROFILES], } } def set_camera_config(profile_id=None, fps=None): current = read_camera_settings() if profile_id is None: profile = normalize_camera_profile(current['id']) else: profile = normalize_camera_profile(profile_id) if fps is None: fps = current.get('fps', DEFAULT_CAMERA_FPS) fps = normalize_camera_fps(fps) profile['fps'] = fps write_camera_settings(profile, fps) restart_camera_service() return profile def rumble_controller(duration_ms=400): # Linux FF constants EV_FF = 0x15 FF_RUMBLE = 0x50 # EVIOCSFF = _IOW('E', 0x80, struct ff_effect) — ff_effect is 48 bytes on ARM64 EVIOCSFF = (0x40000000 | (48 << 16) | (ord('E') << 8) | 0x80) device_path = None for path in sorted(glob.glob('/dev/input/event*')): try: sys_name = f'/sys/class/input/{os.path.basename(path)}/device/name' with open(sys_name, encoding='utf-8') as f: name = f.read().strip() if any(k in name for k in ('Gamepad', 'F710', 'GAMEPAD')): device_path = path break except OSError: pass if not device_path: raise RuntimeError('Kein Gamepad gefunden') fd = os.open(device_path, os.O_RDWR) try: # struct ff_effect layout on ARM64 (48 bytes): # offset 0: type (u16) # offset 2: id (s16) — set to -1, kernel assigns # offset 4: direction (u16) # offset 6: trigger.button (u16) # offset 8: trigger.interval (u16) # offset 10: replay.length (u16, ms) # offset 12: replay.delay (u16) # offset 14: padding (2 bytes) # offset 16: union.rumble.strong_magnitude (u16) # offset 18: union.rumble.weak_magnitude (u16) # offset 20–47: zeros buf = bytearray(48) struct.pack_into(' /tmp/gps_node.log 2>&1 &" ]) def resolve_gps_device(): candidates = [] seen = set() for pattern in GPS_PORT_PATTERNS: for path in sorted(glob.glob(pattern)): if path in seen: continue seen.add(path) candidates.append(path) return candidates[0] if candidates else None def configure_gps_serial(fd, baud_rate): baud_map = { 4800: termios.B4800, 9600: termios.B9600, 19200: termios.B19200, 38400: termios.B38400, 57600: termios.B57600, 115200: termios.B115200, } baud = baud_map.get(int(baud_rate), termios.B4800) attrs = termios.tcgetattr(fd) attrs[0] = 0 attrs[1] = 0 attrs[2] = termios.CS8 | termios.CREAD | termios.CLOCAL attrs[3] = 0 attrs[4] = baud attrs[5] = baud attrs[6][termios.VMIN] = 0 attrs[6][termios.VTIME] = 0 termios.tcsetattr(fd, termios.TCSANOW, attrs) termios.tcflush(fd, termios.TCIOFLUSH) def send_gps_serial_command(command_text, baud_rate=GPS_BAUD_RATE): device_path = resolve_gps_device() if not device_path: raise RuntimeError('Kein GPS-Empfänger gefunden') payload = command_text.encode('ascii') fd = None try: fd = os.open(device_path, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK) configure_gps_serial(fd, baud_rate) os.write(fd, payload) termios.tcdrain(fd) except OSError as exc: if exc.errno == errno.ENOENT: raise RuntimeError('GPS-Gerät nicht erreichbar') raise RuntimeError(f'GPS-Befehl fehlgeschlagen: {exc}') except termios.error as exc: raise RuntimeError(f'GPS-Port konnte nicht konfiguriert werden: {exc}') finally: if fd is not None: try: os.close(fd) except OSError: pass return device_path def cold_start_gps_receiver(): device_path = send_gps_serial_command(GPS_COLD_START_COMMAND) time.sleep(1.5) restart_gps_node() return device_path def read_gps_diagnostics_from_ros(): command = [ 'docker', 'exec', EXOMY_CONTAINER, 'bash', '-lc', "source /opt/ros/melodic/setup.bash && " "source /root/exomy_ws/devel/setup.bash && " "timeout 8s rostopic echo -n 1 /gps/diagnostics_json" ] stdout, stderr, returncode = run_command(command) if returncode != 0 or not stdout: raise RuntimeError('Keine GPS-Diagnosedaten verfügbar') raw_text = stdout.strip() lines = [line.strip() for line in raw_text.splitlines() if line.strip()] if not lines: raise RuntimeError('GPS-Diagnosedaten sind leer') if lines[0].startswith('data:'): payload_text = lines[0][5:].strip() if payload_text.startswith("'") and payload_text.endswith("'"): payload_text = payload_text[1:-1] payload_text = payload_text.encode('utf-8').decode('unicode_escape') else: payload_text = lines[0] try: return json.loads(payload_text) except json.JSONDecodeError as exc: raise RuntimeError(f'GPS-Diagnosedaten konnten nicht gelesen werden: {exc}') def parse_gps_datetime(): diagnostics = read_gps_diagnostics_from_ros() utc_date = str(diagnostics.get('utc_date') or '').strip() utc_time = str(diagnostics.get('utc_time') or '').strip() if not utc_date or not utc_time: raise RuntimeError('GPS liefert aktuell keine UTC-Zeit') combined = f'{utc_date} {utc_time}' for pattern in ('%Y-%m-%d %H:%M:%S.%f', '%Y-%m-%d %H:%M:%S'): try: return datetime.datetime.strptime(combined, pattern) except ValueError: continue raise RuntimeError(f'GPS-Zeitformat unbekannt: {combined}') def get_local_timezone(): try: with open('/etc/timezone', 'r', encoding='utf-8') as handle: timezone_name = handle.read().strip() if timezone_name: return ZoneInfo(timezone_name) except (FileNotFoundError, OSError, ValueError): pass return datetime.datetime.now().astimezone().tzinfo or datetime.timezone.utc def set_system_time_from_gps(): gps_dt = parse_gps_datetime() gps_utc = gps_dt.replace(tzinfo=datetime.timezone.utc) local_time = gps_utc.astimezone(get_local_timezone()) time_text = local_time.strftime('%Y-%m-%d %H:%M:%S') stdout, stderr, returncode = run_command(['timedatectl', 'set-time', time_text]) if returncode != 0: raise RuntimeError(stderr or stdout or 'Systemzeit konnte nicht gesetzt werden') return time_text 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']) 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-Neutralwerte geladen', 'drive_neutral': result, 'container_status': container_status, } except FileNotFoundError: return 500, {'status': 'Motor-Konfiguration fehlt'} except Exception: return 500, {'status': 'Fahr-Neutralwerte konnten nicht geladen werden'} def get_drive_estimator_config(): from admin_motor_test import WHEELS, _load_config try: config_dir = os.path.join(os.path.dirname(__file__), '..', 'config') config_candidates = [ os.path.join(config_dir, 'exomy.yaml'), os.path.join(config_dir, 'exomy.remote.yaml'), os.path.join(config_dir, 'exomy.yaml.template'), ] config_path = next((path for path in config_candidates if os.path.exists(path)), None) if config_path is None: raise FileNotFoundError('Konfigurationsdatei nicht gefunden') config = _load_config(config_path) neutral_values = [ int(config[f'drive_pwm_neutral_{wheel}']) for wheel in WHEELS ] average_neutral = sum(neutral_values) / len(neutral_values) return 200, { 'status': 'Bereit', 'estimator': { 'config_path': os.path.abspath(config_path), 'drive_pwm_neutral_average': average_neutral, 'drive_pwm_neutral_values': neutral_values, 'drive_pwm_range': int(config['drive_pwm_range']), 'pwm_frequency_hz': 50, 'servo_max_rpm': 50.0, 'servo_full_scale_pulse_delta_ms': 0.2, 'wheel_diameter_m_default': 0.1, }, } except FileNotFoundError: return 500, {'status': 'Motor-Konfiguration fehlt'} except Exception: return 500, {'status': 'Estimator-Konfiguration konnte nicht geladen werden'} def get_drive_estimator_calibration(): default_points = [0, 20, 30, 40, 50, 60, 80, 100] data = { 'points': [ {'percent': point, 'speed_mps': 0.0} for point in default_points ], 'updated_at': None, 'source': 'default', } try: with open(DRIVE_ESTIMATOR_CALIBRATION_FILE, 'r', encoding='utf-8') as handle: payload = json.load(handle) points = payload.get('points') if isinstance(points, list): cleaned_points = [] for item in points: if not isinstance(item, dict): continue percent = int(item.get('percent', 0)) speed_mps = float(item.get('speed_mps', 0.0)) cleaned_points.append({ 'percent': max(0, min(100, percent)), 'speed_mps': max(0.0, speed_mps), }) if cleaned_points: cleaned_points.sort(key=lambda item: item['percent']) data = { 'points': cleaned_points, 'updated_at': payload.get('updated_at'), 'source': 'saved', } except FileNotFoundError: pass except (OSError, ValueError, TypeError, json.JSONDecodeError): return 500, {'status': 'Kalibrierung konnte nicht geladen werden'} return 200, { 'status': 'Bereit', 'calibration': data, } def save_drive_estimator_calibration(points): if not isinstance(points, list) or not points: return 400, {'status': 'Kalibrierpunkte fehlen'} cleaned_points = [] seen_percents = set() for item in points: if not isinstance(item, dict): return 400, {'status': 'Kalibrierpunkte sind ungültig'} try: percent = int(item.get('percent', 0)) speed_mps = float(item.get('speed_mps', 0.0)) except (TypeError, ValueError): return 400, {'status': 'Kalibrierwerte sind ungültig'} percent = max(0, min(100, percent)) speed_mps = max(0.0, speed_mps) if percent in seen_percents: return 400, {'status': 'Jede Stufe darf nur einmal vorkommen'} seen_percents.add(percent) cleaned_points.append({ 'percent': percent, 'speed_mps': speed_mps, }) cleaned_points.sort(key=lambda item: item['percent']) os.makedirs(os.path.dirname(DRIVE_ESTIMATOR_CALIBRATION_FILE), exist_ok=True) payload = { 'updated_at': time.strftime('%Y-%m-%d %H:%M:%S'), 'points': cleaned_points, } try: with open(DRIVE_ESTIMATOR_CALIBRATION_FILE, 'w', encoding='utf-8') as handle: json.dump(payload, handle, ensure_ascii=False, indent=2) handle.write('\n') except OSError: return 500, {'status': 'Kalibrierung konnte nicht gespeichert werden'} return 200, { 'status': 'Kalibrierung gespeichert', 'calibration': { 'points': cleaned_points, 'updated_at': payload['updated_at'], 'source': 'saved', }, } def get_route_library(): data = { 'routes': [], 'updated_at': None, } try: with open(ROUTE_LIBRARY_FILE, 'r', encoding='utf-8') as handle: payload = json.load(handle) routes = payload.get('routes') if isinstance(routes, list): cleaned_routes = [] for route in routes: if not isinstance(route, dict): continue name = str(route.get('name', '')).strip() waypoints = route.get('waypoints', []) if not name or not isinstance(waypoints, list): continue cleaned_waypoints = [] for point in waypoints: if not isinstance(point, dict): continue try: lat = float(point.get('lat')) lon = float(point.get('lon')) except (TypeError, ValueError): continue cleaned_waypoints.append({ 'lat': lat, 'lon': lon, }) cleaned_routes.append({ 'name': name, 'waypoints': cleaned_waypoints, 'updated_at': route.get('updated_at'), }) data = { 'routes': cleaned_routes, 'updated_at': payload.get('updated_at'), } except FileNotFoundError: pass except (OSError, ValueError, TypeError, json.JSONDecodeError): return 500, {'status': 'Routen konnten nicht geladen werden'} return 200, { 'status': 'Bereit', 'library': data, } def save_named_route(name, waypoints): route_name = str(name or '').strip() if not route_name: return 400, {'status': 'Routenname fehlt'} if not isinstance(waypoints, list) or not waypoints: return 400, {'status': 'Route hat keine Wegpunkte'} cleaned_waypoints = [] for point in waypoints: if not isinstance(point, dict): return 400, {'status': 'Wegpunkte sind ungültig'} try: lat = float(point.get('lat')) lon = float(point.get('lon')) except (TypeError, ValueError): return 400, {'status': 'Wegpunkte sind ungültig'} cleaned_waypoints.append({ 'lat': lat, 'lon': lon, }) existing_status, existing_response = get_route_library() if existing_status != 200: return existing_status, existing_response routes = existing_response['library']['routes'] now_text = time.strftime('%Y-%m-%d %H:%M:%S') replaced = False for route in routes: if route['name'].lower() == route_name.lower(): route['name'] = route_name route['waypoints'] = cleaned_waypoints route['updated_at'] = now_text replaced = True break if not replaced: routes.append({ 'name': route_name, 'waypoints': cleaned_waypoints, 'updated_at': now_text, }) routes.sort(key=lambda item: item['name'].lower()) os.makedirs(os.path.dirname(ROUTE_LIBRARY_FILE), exist_ok=True) payload = { 'updated_at': now_text, 'routes': routes, } try: with open(ROUTE_LIBRARY_FILE, 'w', encoding='utf-8') as handle: json.dump(payload, handle, ensure_ascii=False, indent=2) handle.write('\n') except OSError: return 500, {'status': 'Route konnte nicht gespeichert werden'} return 200, { 'status': 'Route gespeichert', 'route': { 'name': route_name, 'waypoints': cleaned_waypoints, 'updated_at': now_text, }, 'library': payload, } def delete_named_route(name): route_name = str(name or '').strip() if not route_name: return 400, {'status': 'Routenname fehlt'} existing_status, existing_response = get_route_library() if existing_status != 200: return existing_status, existing_response routes = existing_response['library']['routes'] filtered_routes = [route for route in routes if route['name'].lower() != route_name.lower()] if len(filtered_routes) == len(routes): return 404, {'status': 'Route nicht gefunden'} now_text = time.strftime('%Y-%m-%d %H:%M:%S') payload = { 'updated_at': now_text, 'routes': filtered_routes, } try: with open(ROUTE_LIBRARY_FILE, 'w', encoding='utf-8') as handle: json.dump(payload, handle, ensure_ascii=False, indent=2) handle.write('\n') except OSError: return 500, {'status': 'Route konnte nicht gelöscht werden'} return 200, { 'status': 'Route gelöscht', 'library': payload, } def preview_drive_neutral_action(values): 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(values) return 200, { 'status': 'Fahr-Neutralwerte angefahren', '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(values): 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(values) return 200, { 'status': 'Fahr-Neutralwerte 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-Neutralwerte konnten nicht gespeichert werden'} class Handler(http.server.BaseHTTPRequestHandler): def end_headers(self): self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type') super().end_headers() def write_json(self, status_code, payload): body = json.dumps(payload).encode('utf-8') self.send_response(status_code) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Content-Length', str(len(body))) self.end_headers() self.wfile.write(body) def do_OPTIONS(self): self.send_response(204) self.end_headers() def do_GET(self): 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/camera-config': self.write_json(200, get_camera_config()) return 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 if self.path == '/api/drive-estimator/config': status_code, response = get_drive_estimator_config() self.write_json(status_code, response) return if self.path == '/api/drive-estimator/calibration': status_code, response = get_drive_estimator_calibration() self.write_json(status_code, response) return if self.path == '/api/routes': status_code, response = get_route_library() self.write_json(status_code, response) return self.write_json(404, {'status': 'Unbekannt'}) def do_POST(self): raw_body = b'{}' if self.path in ('/api/routes/save', '/api/routes/delete'): 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'{}' if self.path == '/api/delay': 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: seconds = set_delay(payload.get('delay_seconds', 0)) self.write_json(200, {'status': 'Delay gesetzt', 'delay_seconds': seconds}) except Exception as exc: 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/camera-config': 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: profile = set_camera_config(payload.get('profile'), payload.get('fps')) self.write_json(200, { 'status': 'Kameraprofil gespeichert, Kameradienst startet neu', 'camera': { 'selected_profile': profile['id'], 'width': profile['width'], 'height': profile['height'], 'fps': profile['fps'], 'fps_options': list(CAMERA_FPS_OPTIONS), 'profiles': [dict(item) for item in CAMERA_PROFILES], } }) except ValueError as exc: self.write_json(400, {'status': str(exc)}) except Exception as exc: self.write_json(500, {'status': str(exc)}) return if self.path == '/api/drive-estimator/calibration': 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_estimator_calibration(payload.get('points')) self.write_json(status_code, response) return if self.path == '/api/routes/save': 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_named_route( payload.get('name'), payload.get('waypoints'), ) self.write_json(status_code, response) return if self.path == '/api/routes/delete': 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 = delete_named_route(payload.get('name')) self.write_json(status_code, response) return 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 values = payload.get('values') if not isinstance(values, dict): self.write_json(400, {'status': 'Fahr-Neutralwerte fehlen'}) return status_code, response = preview_drive_neutral_action(values) 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 values = payload.get('values') if not isinstance(values, dict): self.write_json(400, {'status': 'Fahr-Neutralwerte fehlen'}) return status_code, response = save_drive_neutral_action(values) 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 if self.path == '/api/rumble': try: rumble_controller(400) self.write_json(200, {'status': 'Vibration ausgelöst'}) except RuntimeError as exc: self.write_json(500, {'status': str(exc)}) return actions = { '/api/cold-start-gps': { 'handler': cold_start_gps_receiver, 'status': 'GPS-Kaltstart wird ausgelöst' }, '/api/restart-gps': { 'handler': restart_gps_node, 'status': 'GPS-Node wird neu gestartet' }, '/api/set-time-from-gps': { 'handler': set_system_time_from_gps, 'status': 'Pi-Zeit wird aus GPS gesetzt' }, '/api/restart-camera': { 'handler': restart_camera_service, 'status': 'Kameradienst wird neu gestartet' }, '/api/restart-container': { 'handler': restart_exomy_container, 'status': 'ExoMy-Container wird neu gestartet' }, '/api/reboot': { 'handler': reboot_raspberry_pi, 'status': 'Raspberry Pi wird neu gestartet' }, '/api/shutdown': { 'handler': shutdown_raspberry_pi, 'status': 'Raspberry Pi wird heruntergefahren' } } action = actions.get(self.path) if action is None: self.write_json(404, {'status': 'Unbekannt'}) return self.write_json(202, {'status': action['status']}) self.wfile.flush() time.sleep(0.1) action['handler']() def log_message(self, fmt, *args): return if __name__ == '__main__': with ReusableThreadingTCPServer(('0.0.0.0', PORT), Handler) as server: server.daemon_threads = True server.serve_forever()