Race MCP Server

import asyncio
import json
import math
import random
import time
from dataclasses import dataclass, asdict
from typing import Dict, Any


@dataclass
class MockTelemetry:
    """Simplified telemetry snapshot matching iRacing fields."""

    SessionTime: float
    Lap: int
    LapCurrentLapTime: float
    LapDist: float
    Speed: float
    RPM: float
    Gear: int
    Throttle: float
    Brake: float
    SteeringWheelAngle: float
    TrackTemp: float
    AirTemp: float
    FuelLevel: float
    IsOnTrack: bool
    SessionState: str
    SessionFlags: str


class MockIRacingDataGenerator:
    """Generate somewhat realistic iRacing-like telemetry data."""

    def __init__(self, track_length_m: float = 5000.0, tick_rate: float = 10.0):
        self.track_length = track_length_m
        self.tick_rate = tick_rate
        self.start_time = time.time()
        self.lap_start_time = self.start_time
        self.lap = 1
        self.lap_dist = 0.0
        self.speed = 0.0
        self.rpm = 0.0
        self.gear = 1
        self.throttle = 0.0
        self.brake = 0.0
        self.fuel_level = 18.0
        self.track_temp = 85.0
        self.air_temp = 72.0
        self.steering = 0.0

    def _update_state(self, dt: float) -> None:
        # Randomly adjust throttle and brake for variability
        if random.random() < 0.1:
            self.throttle = max(0.0, min(1.0, self.throttle + random.uniform(-0.3, 0.3)))
        if random.random() < 0.05:
            self.brake = max(0.0, min(1.0, random.uniform(0.0, 1.0)))
        else:
            self.brake = max(0.0, self.brake - 0.1)

        # Update speed based on throttle and brake
        accel = (self.throttle - self.brake) * 5.0  # m/s^2
        self.speed = max(0.0, self.speed + accel * dt)
        self.speed = min(self.speed, 80.0)  # cap around 180 mph (~80 m/s)

        # Convert speed to mph for telemetry
        speed_mph = self.speed * 2.23694

        # Simple gear estimation based on speed
        self.gear = min(6, max(1, int(speed_mph // 30) + 1))

        # RPM tied to speed and gear with some noise
        gear_ratio = [0, 3.2, 2.1, 1.5, 1.2, 1.0, 0.85][self.gear]
        self.rpm = min(9000.0, speed_mph * gear_ratio * 50 + random.uniform(-50, 50))

        # Steering oscillates slightly with noise
        self.steering = math.sin(time.time() * 0.5) * 0.2 + random.uniform(-0.02, 0.02)

        # Update lap distance
        self.lap_dist += self.speed * dt
        if self.lap_dist >= self.track_length:
            self.lap += 1
            self.lap_dist -= self.track_length
            self.lap_start_time = time.time()

        # Fuel consumption
        self.fuel_level = max(0.0, self.fuel_level - self.speed * dt * 0.0005)

    def generate(self) -> Dict[str, Any]:
        now = time.time()
        dt = now - self.start_time - self.session_time if hasattr(self, 'session_time') else 1.0 / self.tick_rate
        self.session_time = now - self.start_time
        self._update_state(dt)

        # Simulate occasional flag changes and off-track events for AI testing
        elapsed = self.session_time
        
        # Simulate flag changes every 30-60 seconds
        flag_cycle = elapsed % 60
        if flag_cycle < 5:
            session_flags = "Yellow"  # Caution
        elif flag_cycle < 8:
            session_flags = "Red"     # Red flag
        elif flag_cycle < 55:
            session_flags = "Green"   # Normal racing
        else:
            session_flags = "Checkered"  # End of session
            
        # Simulate going off track occasionally (every 45 seconds for 3 seconds)
        off_track_cycle = elapsed % 45
        is_on_track = not (42 <= off_track_cycle <= 45)

        lap_time = now - self.lap_start_time
        telemetry = MockTelemetry(
            SessionTime=self.session_time,
            Lap=self.lap,
            LapCurrentLapTime=lap_time,
            LapDist=self.lap_dist,
            Speed=self.speed * 2.23694,  # mph
            RPM=self.rpm,
            Gear=self.gear,
            Throttle=self.throttle,
            Brake=self.brake,
            SteeringWheelAngle=self.steering,
            TrackTemp=self.track_temp,
            AirTemp=self.air_temp,
            FuelLevel=self.fuel_level,
            IsOnTrack=is_on_track,
            SessionState="Racing",
            SessionFlags=session_flags,
        )
        return asdict(telemetry)


class MockIRacingStreamServer:
    """TCP server emitting JSON telemetry frames to connected clients."""

    def __init__(self, host: str = "127.0.0.1", port: int = 9000, tick_rate: float = 10.0):
        self.generator = MockIRacingDataGenerator(tick_rate=tick_rate)
        self.host = host
        self.port = port
        self.tick_rate = tick_rate

    async def _handle_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
        try:
            while True:
                data = json.dumps(self.generator.generate()).encode() + b"\n"
                writer.write(data)
                await writer.drain()
                await asyncio.sleep(1.0 / self.tick_rate)
        except asyncio.CancelledError:  # pragma: no cover - server shutdown
            pass
        finally:
            writer.close()
            await writer.wait_closed()

    async def start(self) -> None:
        server = await asyncio.start_server(self._handle_client, self.host, self.port)
        async with server:
            await server.serve_forever()


def main() -> None:
    """Entry point for running the mock telemetry stream."""
    import argparse

    parser = argparse.ArgumentParser(description="Mock iRacing telemetry stream")
    parser.add_argument("--host", default="127.0.0.1", help="Host to bind server")
    parser.add_argument("--port", type=int, default=9000, help="Port to bind server")
    parser.add_argument("--rate", type=float, default=10.0, help="Telemetry frames per second")
    args = parser.parse_args()

    server = MockIRacingStreamServer(host=args.host, port=args.port, tick_rate=args.rate)
    try:
        asyncio.run(server.start())
    except KeyboardInterrupt:  # pragma: no cover - manual stop
        pass


if __name__ == "__main__":
    main()