Eskimue/ExoMy_Cuno
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Kaera umschaltbar

Browse Source
This commit is contained in:
Eskimue
2026-05-25 15:30:56 +02:00
parent 1855a2446b
commit 8c8bfd9e5b
7 changed files with 452 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ExoMy_Software-master/scripts/exomy_admin_api.py
+134
View File
@@ -21,7 +21,38 @@ DRIVE_ESTIMATOR_CALIBRATION_FILE = os.path.join(
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()
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'
class ReusableThreadingTCPServer(socketserver.ThreadingTCPServer):
@@ -285,6 +316,77 @@ def set_speed_limit(percent):
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 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
selected_profile_id = str(payload.get('profile', '')).strip()
if selected_profile_id in CAMERA_PROFILE_MAP:
return dict(CAMERA_PROFILE_MAP[selected_profile_id])
try:
width = int(payload.get('width', default_profile['width']))
height = int(payload.get('height', default_profile['height']))
fps = int(payload.get('fps', default_profile['fps']))
except (TypeError, ValueError):
return default_profile
for profile in CAMERA_PROFILES:
if profile['width'] == width and profile['height'] == height and profile['fps'] == fps:
return dict(profile)
return default_profile
def write_camera_settings(profile):
profile = dict(profile)
os.makedirs(os.path.dirname(CAMERA_SETTINGS_FILE), exist_ok=True)
payload = {
'profile': profile['id'],
'width': profile['width'],
'height': profile['height'],
'fps': profile['fps'],
}
with open(CAMERA_SETTINGS_FILE, 'w', encoding='utf-8') as handle:
json.dump(payload, handle, indent=2, sort_keys=True)
handle.write('\n')
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'],
'profiles': [dict(item) for item in CAMERA_PROFILES],
}
}
def set_camera_profile(profile_id):
profile = normalize_camera_profile(profile_id)
write_camera_settings(profile)
restart_camera_service()
return profile
def rumble_controller(duration_ms=400):
# Linux FF constants
EV_FF = 0x15
@@ -856,6 +958,9 @@ class Handler(http.server.BaseHTTPRequestHandler):
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
@@ -932,6 +1037,35 @@ class Handler(http.server.BaseHTTPRequestHandler):
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_profile(payload.get('profile'))
self.write_json(200, {
'status': 'Kameraauflösung gespeichert, Kameradienst startet neu',
'camera': {
'selected_profile': profile['id'],
'width': profile['width'],
'height': profile['height'],
'fps': profile['fps'],
'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'))
ExoMy_Software-master/scripts/libcamera_mjpeg_server.py
+45 -7
View File
@@ -1,5 +1,8 @@
#!/usr/bin/env python3
import http.server
import json
import os
import urllib.parse
import socketserver
import subprocess
import threading
@@ -9,10 +12,40 @@ BOUNDARY = b'--frame'
LATEST_JPEG = None
LATEST_LOCK = threading.Lock()
WIDTH = 1280
HEIGHT = 720
FPS = 10
PORT = 8081
CONFIG_FILE = os.path.join(os.path.dirname(__file__), '..', 'config', 'camera_settings.json')
DEFAULT_CAMERA_SETTINGS = {
'width': 1280,
'height': 720,
'fps': 10,
}
def load_camera_settings():
settings = dict(DEFAULT_CAMERA_SETTINGS)
try:
with open(CONFIG_FILE, 'r', encoding='utf-8') as handle:
raw_settings = json.load(handle)
except (FileNotFoundError, json.JSONDecodeError, OSError, ValueError, TypeError):
return settings
try:
width = int(raw_settings.get('width', settings['width']))
height = int(raw_settings.get('height', settings['height']))
fps = int(raw_settings.get('fps', settings['fps']))
except (TypeError, ValueError):
return settings
settings['width'] = max(320, min(2592, width))
settings['height'] = max(240, min(1944, height))
settings['fps'] = max(1, min(30, fps))
return settings
CAMERA_SETTINGS = load_camera_settings()
WIDTH = CAMERA_SETTINGS['width']
HEIGHT = CAMERA_SETTINGS['height']
FPS = CAMERA_SETTINGS['fps']
def camera_reader():
@@ -65,14 +98,18 @@ class ReusableThreadingTCPServer(socketserver.ThreadingTCPServer):
class Handler(http.server.BaseHTTPRequestHandler):
def normalized_path(self):
return urllib.parse.urlsplit(self.path).path
def do_HEAD(self):
if self.path in ('/', '/index.html'):
path = self.normalized_path()
if path in ('/', '/index.html'):
self.send_response(200)
self.send_header('Content-Type', 'text/html; charset=utf-8')
self.end_headers()
return
if self.path == '/stream.mjpg':
if path == '/stream.mjpg':
self.send_response(200)
self.send_header('Age', '0')
self.send_header('Cache-Control', 'no-cache, private')
@@ -84,7 +121,8 @@ class Handler(http.server.BaseHTTPRequestHandler):
self.send_error(404)
def do_GET(self):
if self.path in ('/', '/index.html'):
path = self.normalized_path()
if path in ('/', '/index.html'):
body = b'<html><body><img src="/stream.mjpg" /></body></html>'
self.send_response(200)
self.send_header('Content-Type', 'text/html; charset=utf-8')
@@ -93,7 +131,7 @@ class Handler(http.server.BaseHTTPRequestHandler):
self.wfile.write(body)
return
if self.path != '/stream.mjpg':
if path != '/stream.mjpg':
self.send_error(404)
return