Main ECU Simulator class that manages virtual CAN bus communication and simulates realistic automotive data.
| 302 | |
| 303 | |
| 304 | class ECUSimulator: |
| 305 | """ |
| 306 | Main ECU Simulator class that manages virtual CAN bus communication |
| 307 | and simulates realistic automotive data. |
| 308 | """ |
| 309 | |
| 310 | def __init__(self, debug: bool = False): |
| 311 | """Initialize the ECU simulator with physics state only.""" |
| 312 | self.can_sender: Optional[Callable[[int, bytes], None]] = None |
| 313 | self.debug = debug |
| 314 | |
| 315 | # Initialize ECU states |
| 316 | self.engine = EngineState() |
| 317 | self.transmission = TransmissionState() |
| 318 | self.battery = BatteryState() |
| 319 | self.chassis = ChassisState() |
| 320 | self.body = BodyControlState() |
| 321 | self.cluster = ClusterState() |
| 322 | |
| 323 | # Simulation control |
| 324 | self.running = False |
| 325 | self.threads: List[threading.Thread] = [] |
| 326 | self.simulation_time = 0.0 |
| 327 | |
| 328 | # Driving simulation parameters |
| 329 | self.target_speed = 0.0 |
| 330 | self.acceleration = 0.0 |
| 331 | |
| 332 | # Noise generators (Ornstein-Uhlenbeck process for realistic sensor noise) |
| 333 | self.rpm_noise = 0.0 |
| 334 | self.coolant_temp_noise = 0.0 |
| 335 | self.intake_temp_noise = 0.0 |
| 336 | self.fuel_pressure_noise = 0.0 |
| 337 | self.wheel_speed_noise = [0.0, 0.0, 0.0, 0.0] |
| 338 | self.brake_pressure_noise = 0.0 |
| 339 | self.steering_noise = 0.0 |
| 340 | self.battery_voltage_noise = 0.0 |
| 341 | self.battery_current_noise = 0.0 |
| 342 | |
| 343 | # Smooth brake pressure state |
| 344 | self.brake_pressure_actual = 0.0 |
| 345 | self.brake_pressure_target = 0.0 |
| 346 | |
| 347 | def set_can_sender(self, sender: Callable[[int, bytes], None]): |
| 348 | """ |
| 349 | Set the CAN frame sender callback. |
| 350 | |
| 351 | Args: |
| 352 | sender: Function that takes (can_id: int, data: bytes) and sends the frame |
| 353 | """ |
| 354 | self.can_sender = sender |
| 355 | |
| 356 | def send_frame(self, arbitration_id: int, data: bytes): |
| 357 | """Send a CAN frame using the configured sender.""" |
| 358 | if self.debug and arbitration_id == 0x100: |
| 359 | print(f"\n[DEBUG] CAN ID 0x{arbitration_id:03X}: {data.hex().upper()}") |
| 360 | print( |
| 361 | f" Byte 0-1 (RPM raw): 0x{data[0]:02X}{data[1]:02X} = {(data[0]<<8)|data[1]} → {((data[0]<<8)|data[1])*0.25:.1f} RPM" |