"""Mock 车辆设备实现。 用于在没有真实硬件接入时,生成可用于联调的车辆信号数据。 """ from __future__ import annotations import random from dataclasses import dataclass from backend.device.base import DeviceAdapter @dataclass(frozen=True) class VehicleSignalPayload: """一帧车辆信号载荷(用于广播/落库)。""" steering_wheel_angle_deg: float brake_pedal_travel_mm: float throttle_pedal_travel_mm: float gear: str handbrake: int vehicle_speed_kmh: float wheel_speed_rpm: dict lights: dict soc_percent: float voltage_v: float current_a: float temperature_c: float def to_dict(self) -> dict: """转为可 JSON 序列化的 dict。""" return { "steering_wheel_angle_deg": self.steering_wheel_angle_deg, "brake_pedal_travel_mm": self.brake_pedal_travel_mm, "throttle_pedal_travel_mm": self.throttle_pedal_travel_mm, "gear": self.gear, "handbrake": self.handbrake, "vehicle_speed_kmh": self.vehicle_speed_kmh, "wheel_speed_rpm": self.wheel_speed_rpm, "lights": self.lights, "soc_percent": self.soc_percent, "voltage_v": self.voltage_v, "current_a": self.current_a, "temperature_c": self.temperature_c, } class MockVehicleDevice(DeviceAdapter): """模拟车辆设备:提供 connect/disconnect/is_connected 与 sample()。""" def __init__(self, device_id: str = "controlbox_01") -> None: self.device_id = device_id self.device_type = "mock_vehicle" self._connected = False async def connect(self) -> None: """模拟建立连接。""" self._connected = True async def disconnect(self) -> None: """模拟断开连接。""" self._connected = False async def is_connected(self) -> bool: """返回当前连接状态。""" return self._connected def sample(self) -> VehicleSignalPayload: """采样生成一帧模拟信号。""" steering = random.uniform(-180.0, 180.0) brake = max(0.0, random.gauss(2.0, 1.0)) throttle = max(0.0, random.gauss(15.0, 5.0)) gear = random.choice(["P", "N", "D", "S"]) handbrake = 1 if gear == "P" else 0 speed = max(0.0, random.gauss(40.0, 10.0)) if gear in {"D", "S"} else 0.0 rpm = int(speed * 9 + random.uniform(-5, 5)) wheel = {"FL": rpm, "FR": rpm, "RL": rpm, "RR": rpm} lights = { "left_turn": int(random.random() < 0.05), "right_turn": int(random.random() < 0.05), "hazard": int(random.random() < 0.02), "brake": int(brake > 5.0), } soc = max(0.0, min(100.0, 80.0 - random.random() * 0.1)) voltage = 360.0 + random.uniform(-0.5, 0.5) current = max(0.0, random.gauss(15.0, 2.0)) temp = 28.0 + random.uniform(-0.2, 0.2) return VehicleSignalPayload( steering_wheel_angle_deg=steering, brake_pedal_travel_mm=brake, throttle_pedal_travel_mm=throttle, gear=gear, handbrake=handbrake, vehicle_speed_kmh=speed, wheel_speed_rpm=wheel, lights=lights, soc_percent=soc, voltage_v=voltage, current_a=current, temperature_c=temp, )