766 lines
26 KiB
Python
766 lines
26 KiB
Python
#!/usr/bin/env python3
|
||
import fcntl
|
||
import glob
|
||
import http.server
|
||
import json
|
||
import os
|
||
import shutil
|
||
import socketserver
|
||
import struct
|
||
import subprocess
|
||
import threading
|
||
import time
|
||
|
||
PORT = 8082
|
||
EXOMY_CONTAINER = 'exomy_autostart'
|
||
DELAY_FILE = '/tmp/exomy_delay.txt'
|
||
MOTOR_TEST_LOCK = threading.Lock()
|
||
|
||
|
||
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':
|
||
return f'verbunden: {connection}'
|
||
return state
|
||
|
||
return 'kein wlan0'
|
||
|
||
|
||
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(),
|
||
'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'),
|
||
'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 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('<H', buf, 0, FF_RUMBLE)
|
||
struct.pack_into('<h', buf, 2, -1)
|
||
struct.pack_into('<H', buf, 10, duration_ms)
|
||
struct.pack_into('<H', buf, 16, 0xFFFF)
|
||
struct.pack_into('<H', buf, 18, 0x8000)
|
||
|
||
fcntl.ioctl(fd, EVIOCSFF, buf)
|
||
effect_id = struct.unpack_from('<h', buf, 2)[0]
|
||
|
||
# input_event on ARM64: sec(u64) + usec(u64) + type(u16) + code(u16) + value(s32) = 24 bytes
|
||
os.write(fd, struct.pack('<QQHHi', 0, 0, EV_FF, effect_id, 1))
|
||
time.sleep(duration_ms / 1000.0 + 0.1)
|
||
os.write(fd, struct.pack('<QQHHi', 0, 0, EV_FF, effect_id, 0))
|
||
except OSError as exc:
|
||
raise RuntimeError(f'FF ioctl fehlgeschlagen: {exc}') from exc
|
||
finally:
|
||
os.close(fd)
|
||
|
||
|
||
def restart_camera_service():
|
||
subprocess.Popen(['systemctl', 'restart', 'exomy-camera-stream.service'])
|
||
|
||
|
||
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'])
|
||
|
||
|
||
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 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/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/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/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/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()
|