699 lines
23 KiB
Python
699 lines
23 KiB
Python
#!/usr/bin/env python3
|
|
import http.server
|
|
import json
|
|
import os
|
|
import shutil
|
|
import socketserver
|
|
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 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
|
|
|
|
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()
|