1462 lines
49 KiB
Python
1462 lines
49 KiB
Python
#!/usr/bin/env python3
|
||
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
|
||
|
||
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_SETTINGS_FILE = os.path.join(
|
||
os.path.dirname(__file__), '..', 'config', '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':
|
||
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_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)
|
||
os.makedirs(os.path.dirname(CAMERA_SETTINGS_FILE), exist_ok=True)
|
||
payload = {
|
||
'profile': profile['id'],
|
||
'width': profile['width'],
|
||
'height': profile['height'],
|
||
'fps': fps,
|
||
'updated_at': int(time.time()),
|
||
}
|
||
with open(CAMERA_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_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('<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 restart_gps_node():
|
||
subprocess.Popen([
|
||
'docker', 'exec', EXOMY_CONTAINER, 'bash', '-lc',
|
||
"pkill -f '/root/exomy_ws/src/exomy/src/gps_node.py' || true; "
|
||
"sleep 1; "
|
||
"source /opt/ros/melodic/setup.bash && source /root/exomy_ws/devel/setup.bash && "
|
||
"nohup python /root/exomy_ws/src/exomy/src/gps_node.py > /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 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/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()
|