117 lines
4.3 KiB
Python
117 lines
4.3 KiB
Python
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) & 0xFFFFFFFF)
|
|
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
|