AskaEth/TurboSu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: extract test data generator to server/test_mode.py

Browse Source 
- TestDataGenerator class: builds 324-byte Forza packets, runs async loop at 30fps
- Simulated data passes through _handle_packet (parser -> WS -> forwarder)
- Listener delegates to test_mode.get_generator()
- Cleaner separation: listener.py no longer contains test logic
This commit is contained in:
AskaEth
2026-07-26 18:07:19 +08:00
parent 5690d792a5
commit 69b73e8ce0
2 changed files with 123 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/telemetry/listener.py
+7 -85
View File
@@ -1,9 +1,7 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
import random
import socket import socket
import struct
import time import time
from typing import Any, Callable from typing import Any, Callable
@@ -60,7 +58,6 @@ class TelemetryListener:
self._last_packet_time: float = 0.0 self._last_packet_time: float = 0.0
self._packet_count: int = 0 self._packet_count: int = 0
self.forwarder = DataForwarder() self.forwarder = DataForwarder()
self._test_task: asyncio.Task | None = None
@property @property
def is_running(self) -> bool: def is_running(self) -> bool:
@@ -126,92 +123,17 @@ class TelemetryListener:
@property @property
def is_testing(self) -> bool: def is_testing(self) -> bool:
return self._test_task is not None and not self._test_task.done() from server.test_mode import _test_generator
return _test_generator.running
def start_test_mode(self): def start_test_mode(self):
if self._test_task and not self._test_task.done(): from server.test_mode import _test_generator
return _test_generator.listener = self
self._test_task = asyncio.ensure_future(self._test_loop()) _test_generator.start()
logger.info("UI test mode started")
def stop_test_mode(self): def stop_test_mode(self):
if self._test_task and not self._test_task.done(): from server.test_mode import _test_generator
self._test_task.cancel() _test_generator.stop()
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 = bytearray(324)
struct.pack_into("<i", buf, 0, 1)
struct.pack_into("<I", buf, 4, int(time.time() * 1000))
struct.pack_into("<f", buf, 8, max_rpm)
struct.pack_into("<f", buf, 12, 800.0)
struct.pack_into("<f", buf, 16, rpm)
struct.pack_into("<f", buf, 256, speed_kmh / 3.6)
struct.pack_into("<f", buf, 260, rpm / max_rpm * 400000)
struct.pack_into("<f", buf, 264, torque)
struct.pack_into("<f", buf, 284, boost)
struct.pack_into("<f", buf, 288, fuel)
struct.pack_into("<f", buf, 296, best_lap)
struct.pack_into("<f", buf, 300, last_lap)
struct.pack_into("<f", buf, 304, lap_time)
struct.pack_into("<H", buf, 312, int(phase) % 15)
struct.pack_into("<B", buf, 314, 1)
struct.pack_into("<B", buf, 315, int(throttle * 255))
struct.pack_into("<B", buf, 316, int(brake * 255))
struct.pack_into("<B", buf, 319, gear)
struct.pack_into("<b", buf, 320, int(steering * 127))
self.forwarder.forward(bytes(buf))
for cb in self._callbacks:
try: cb(td)
except Exception: pass
def _get_parser(self, key: str) -> BaseParser: def _get_parser(self, key: str) -> BaseParser:
if key not in self._parser_cache: if key not in self._parser_cache:
server/test_mode.py
+116
View File
@@ -0,0 +1,116 @@
from __future__ import annotations
import asyncio
import math
import random
import struct
import time
from utils.logger import get_logger
logger = get_logger(__name__)
class TestDataGenerator:
def __init__(self, listener):
self._listener = listener
self._task: asyncio.Task | None = None
@property
def running(self) -> bool:
return self._task is not None and not self._task.done()
def start(self):
if self.running:
return
loop = asyncio.get_event_loop()
self._task = loop.create_task(self._loop())
logger.info("Test data generator started")
def stop(self):
if self._task and not self._task.done():
self._task.cancel()
self._task = None
logger.info("Test data generator stopped")
async def _loop(self):
phase = 0.0
gear = 1
gear_timer = 0
while True:
try:
await asyncio.sleep(1.0 / 30.0)
phase += 0.06
gear_timer += 1
if gear_timer > 55:
gear_timer = 0
gear = (gear % 6) + 1
max_rpm = 8000.0
rpm = max_rpm * (0.12 + 0.58 * abs(math.sin(phase)))
speed_mps = rpm / max_rpm * 89.0 + random.uniform(-1, 1)
throttle = max(0.0, min(1.0, 0.1 + 0.7 * abs(math.sin(phase + 0.5))))
brake = max(0.0, min(1.0, random.uniform(0, 0.35) if random.random() > 0.78 else 0))
steering = math.sin(phase * 0.25) * 0.45
fuel = max(0.0, 1.0 - phase * 0.007)
best_lap = 92.0 + random.uniform(-0.8, 0.8)
last_lap = best_lap + random.uniform(0.3, 2.0)
lap_time = last_lap * random.uniform(0.15, 0.80)
boost = random.uniform(0.2, 1.5)
hp = rpm / max_rpm * 420.0
torque = rpm / max_rpm * 520.0
buf = self._build_packet(max_rpm, rpm, speed_mps, hp, torque,
boost, fuel, best_lap, last_lap,
lap_time, int(phase) % 15,
throttle, brake, gear, steering)
self._listener._handle_packet(buf, ("127.0.0.1", 0))
except asyncio.CancelledError:
break
except Exception as e:
logger.error("Test data error: %s", e)
await asyncio.sleep(1.0)
@staticmethod
def _build_packet(max_rpm, rpm, speed_mps, hp, torque, boost, fuel,
best_lap, last_lap, lap_time, lap_num,
throttle, brake, gear, steering):
buf = bytearray(324)
struct.pack_into("<i", buf, 0, 1)
struct.pack_into("<I", buf, 4, int(time.time() * 1000))
struct.pack_into("<f", buf, 8, max_rpm)
struct.pack_into("<f", buf, 12, 800.0)
struct.pack_into("<f", buf, 16, rpm)
struct.pack_into("<f", buf, 244, random.uniform(-5000, 5000))
struct.pack_into("<f", buf, 248, random.uniform(-50, 50))
struct.pack_into("<f", buf, 252, random.uniform(-5000, 5000))
struct.pack_into("<f", buf, 256, speed_mps)
struct.pack_into("<f", buf, 260, hp * 745.7)
struct.pack_into("<f", buf, 264, torque)
struct.pack_into("<f", buf, 284, boost)
struct.pack_into("<f", buf, 288, fuel)
struct.pack_into("<f", buf, 296, best_lap)
struct.pack_into("<f", buf, 300, last_lap)
struct.pack_into("<f", buf, 304, lap_time)
struct.pack_into("<H", buf, 312, lap_num)
struct.pack_into("<B", buf, 314, 1)
struct.pack_into("<B", buf, 315, max(0, min(255, int(throttle * 255))))
struct.pack_into("<B", buf, 316, max(0, min(255, int(brake * 255))))
struct.pack_into("<B", buf, 319, gear)
struct.pack_into("<b", buf, 320, max(-127, min(127, int(steering * 127))))
return bytes(buf)
_test_generator: TestDataGenerator | None = None
def get_generator(listener=None) -> TestDataGenerator:
global _test_generator
if _test_generator is None:
_test_generator = TestDataGenerator(listener)
elif listener is not None:
_test_generator._listener = listener
return _test_generator