86 lines
3.8 KiB
Python
86 lines
3.8 KiB
Python
import struct
|
|
import time
|
|
from server.telemetry.data import TelemetryData
|
|
|
|
|
|
def get_parser():
|
|
return F124Parser()
|
|
|
|
|
|
class F124Parser:
|
|
"""F1 24 UDP telemetry parser based on official EA SPORTS specification."""
|
|
|
|
# Packet header (29 bytes)
|
|
HEADER_SIZE = 29
|
|
CAR_TELEMETRY_SIZE = 43 # sizeof(CarTelemetryData) with padding
|
|
|
|
def game_id(self) -> str:
|
|
return "f1_24"
|
|
|
|
def parse(self, data: bytes, addr: tuple) -> TelemetryData:
|
|
td = TelemetryData(game_id="f1_24", timestamp=time.time())
|
|
td.raw = {"raw_hex": data.hex(), "length": len(data), "addr": f"{addr[0]}:{addr[1]}"}
|
|
|
|
if len(data) < 50:
|
|
return td
|
|
|
|
try:
|
|
# Parse header
|
|
packet_id = struct.unpack_from("<B", data, 6)[0]
|
|
player_idx = struct.unpack_from("<B", data, 27)[0]
|
|
td.raw["packet_id"] = packet_id
|
|
td.raw["player_index"] = player_idx
|
|
|
|
if packet_id == 6: # Car Telemetry Packet (ID=6)
|
|
offset = self.HEADER_SIZE + player_idx * self.CAR_TELEMETRY_SIZE
|
|
|
|
td.speed_kmh = struct.unpack_from("<H", data, offset)[0]
|
|
td.speed_mph = td.speed_kmh * 0.621371
|
|
|
|
td.throttle = struct.unpack_from("<f", data, offset + 4)[0]
|
|
td.steering = struct.unpack_from("<f", data, offset + 8)[0]
|
|
td.brake = struct.unpack_from("<f", data, offset + 12)[0]
|
|
td.clutch = struct.unpack_from("<B", data, offset + 16)[0] / 100.0
|
|
td.gear = struct.unpack_from("<b", data, offset + 17)[0]
|
|
td.rpm = struct.unpack_from("<H", data, offset + 18)[0]
|
|
td.raw["drs"] = struct.unpack_from("<B", data, offset + 20)[0]
|
|
td.raw["rev_lights_percent"] = struct.unpack_from("<B", data, offset + 21)[0]
|
|
td.raw["rev_lights_bit"] = struct.unpack_from("<H", data, offset + 22)[0]
|
|
|
|
for i in range(4):
|
|
td.raw[f"tyre_surface_temp_{i}"] = struct.unpack_from("<B", data, offset + 30 + i)[0]
|
|
td.raw[f"tyre_inner_temp_{i}"] = struct.unpack_from("<B", data, offset + 34 + i)[0]
|
|
|
|
td.raw["engine_temp"] = struct.unpack_from("<H", data, offset + 38)[0]
|
|
|
|
elif packet_id == 0: # Motion Packet (ID=0)
|
|
offset = self.HEADER_SIZE + player_idx * 64 # sizeof(CarMotionData)
|
|
|
|
td.acceleration_x = struct.unpack_from("<f", data, offset + 28)[0]
|
|
td.acceleration_y = struct.unpack_from("<f", data, offset + 32)[0]
|
|
td.acceleration_z = struct.unpack_from("<f", data, offset + 36)[0]
|
|
|
|
elif packet_id == 2: # Lap Data Packet (ID=2)
|
|
offset = self.HEADER_SIZE + player_idx * 24 # sizeof(LapData)
|
|
|
|
td.last_lap = struct.unpack_from("<f", data, offset)[0]
|
|
td.lap_time = struct.unpack_from("<f", data, offset + 4)[0]
|
|
td.best_lap = struct.unpack_from("<f", data, offset + 8)[0]
|
|
|
|
elif packet_id == 7: # Car Status Packet (ID=7)
|
|
offset = self.HEADER_SIZE + player_idx * 41 # sizeof(CarStatusData)
|
|
|
|
td.raw["traction_control"] = struct.unpack_from("<B", data, offset)[0]
|
|
td.raw["anti_lock_brakes"] = struct.unpack_from("<B", data, offset + 1)[0]
|
|
td.fuel = struct.unpack_from("<f", data, offset + 2)[0] / 110.0 # normalized to 0-1
|
|
if td.fuel > 1.0:
|
|
td.fuel = min(1.0, td.fuel / 100.0)
|
|
td.max_rpm = struct.unpack_from("<H", data, offset + 26)[0]
|
|
td.raw["ers_deploy_mode"] = struct.unpack_from("<B", data, offset + 34)[0]
|
|
td.raw["ers_store_energy"] = struct.unpack_from("<f", data, offset + 36)[0]
|
|
|
|
except struct.error:
|
|
pass
|
|
|
|
return td
|