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

berechnung Weg und geschwindigkeit

Browse Source
This commit is contained in:
Eskimue
2026-05-23 09:27:19 +02:00
parent 27de6aec10
commit bbf7829099
5 changed files with 904 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ExoMy_Software-master/scripts/exomy_admin_api.py
+156
View File
@@ -15,6 +15,9 @@ 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'
)
MOTOR_TEST_LOCK = threading.Lock()
@@ -492,6 +495,136 @@ def get_drive_neutral_status():
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 preview_drive_neutral_action(values):
from admin_motor_test import preview_drive_neutral_value
@@ -587,6 +720,14 @@ class Handler(http.server.BaseHTTPRequestHandler):
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
self.write_json(404, {'status': 'Unbekannt'})
def do_POST(self):
@@ -626,6 +767,21 @@ class Handler(http.server.BaseHTTPRequestHandler):
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/motor-test/drive-neutral/preview':
try:
content_length = int(self.headers.get('Content-Length', '0'))