feat: UI test mode + per-game port configuration
- Settings: UI test mode toggle generates simulated telemetry data (speed/RPM/gear/laps...) - Simulated data broadcasts via WebSocket and forwards to downstream devices - game_ports config: per-game independent port overrides (shown when unified port off) - Settings UI: game port list with save button, hidden when unified port enabled - Sidebar: switch game reads game_ports[gameId] fallback to default_port
This commit is contained in:
@@ -1,7 +1,9 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import random
|
||||
import socket
|
||||
import struct
|
||||
import time
|
||||
from typing import Any, Callable
|
||||
|
||||
@@ -58,6 +60,7 @@ class TelemetryListener:
|
||||
self._last_packet_time: float = 0.0
|
||||
self._packet_count: int = 0
|
||||
self.forwarder = DataForwarder()
|
||||
self._test_task: asyncio.Task | None = None
|
||||
|
||||
@property
|
||||
def is_running(self) -> bool:
|
||||
@@ -112,6 +115,7 @@ class TelemetryListener:
|
||||
|
||||
def stop(self):
|
||||
self._running = False
|
||||
self.stop_test_mode()
|
||||
if self._transport:
|
||||
self._transport.close()
|
||||
self._transport = None
|
||||
@@ -120,6 +124,76 @@ class TelemetryListener:
|
||||
self.forwarder._sock = None
|
||||
logger.info("Telemetry listener stopped")
|
||||
|
||||
@property
|
||||
def is_testing(self) -> bool:
|
||||
return self._test_task is not None and not self._test_task.done()
|
||||
|
||||
def start_test_mode(self):
|
||||
if self._test_task and not self._test_task.done():
|
||||
return
|
||||
self._test_task = asyncio.ensure_future(self._test_loop())
|
||||
logger.info("UI test mode started")
|
||||
|
||||
def stop_test_mode(self):
|
||||
if self._test_task and not self._test_task.done():
|
||||
self._test_task.cancel()
|
||||
self._test_task = None
|
||||
logger.info("UI test mode stopped")
|
||||
|
||||
async def _test_loop(self):
|
||||
import math
|
||||
phase = 0.0
|
||||
gear = 1
|
||||
gear_timer = 0
|
||||
while True:
|
||||
await asyncio.sleep(1.0 / 30.0)
|
||||
phase += 0.06
|
||||
gear_timer += 1
|
||||
if gear_timer > 50:
|
||||
gear_timer = 0
|
||||
gear = (gear % 6) + 1
|
||||
|
||||
max_rpm = 8000.0
|
||||
rpm = max_rpm * (0.15 + 0.55 * abs(math.sin(phase)))
|
||||
speed_kmh = rpm / max_rpm * 320.0 + random.uniform(-5, 5)
|
||||
throttle = max(0.0, min(1.0, 0.15 + 0.65 * abs(math.sin(phase + 0.5))))
|
||||
brake = max(0.0, min(1.0, random.uniform(0, 0.4) if random.random() > 0.75 else 0))
|
||||
steering = math.sin(phase * 0.25) * 0.5
|
||||
fuel = max(0.0, 1.0 - phase * 0.008)
|
||||
best_lap = 92.0 + random.uniform(-1, 1)
|
||||
last_lap = best_lap + random.uniform(0.3, 2.5)
|
||||
lap_time = last_lap * random.uniform(0.2, 0.85)
|
||||
|
||||
td = TelemetryData(
|
||||
game_id=get_config().get("selected_game_id", "test"),
|
||||
timestamp=time.time(),
|
||||
speed_kmh=speed_kmh, speed_mph=speed_kmh * 0.621371,
|
||||
rpm=rpm, max_rpm=max_rpm, gear=gear,
|
||||
throttle=throttle, brake=brake, clutch=0.0, steering=steering,
|
||||
lap_time=lap_time, best_lap=best_lap, last_lap=last_lap,
|
||||
lap_number=int(phase) % 15, fuel=fuel,
|
||||
boost=random.uniform(0, 1.5),
|
||||
horsepower=rpm / max_rpm * 400, torque=rpm / max_rpm * 500,
|
||||
engine_temp=80 + random.uniform(-10, 20),
|
||||
oil_temp=90 + random.uniform(-5, 15),
|
||||
raw={"test_mode": True},
|
||||
)
|
||||
|
||||
self._last_packet_time = time.time()
|
||||
self._packet_count += 1
|
||||
self._latest_data = td
|
||||
|
||||
from server.websocket import ws_manager
|
||||
ws_manager.push_telemetry(td)
|
||||
|
||||
if self.forwarder._sock and self.forwarder._targets:
|
||||
buf = struct.pack("<32s", b"\x01" + b"\x00" * 323)
|
||||
self.forwarder.forward(buf)
|
||||
|
||||
for cb in self._callbacks:
|
||||
try: cb(td)
|
||||
except Exception: pass
|
||||
|
||||
def _get_parser(self, key: str) -> BaseParser:
|
||||
if key not in self._parser_cache:
|
||||
cls = PARSER_MAP.get(key)
|
||||
|
||||
Reference in New Issue
Block a user