diff --git a/README.md b/README.md index 45c26c1..3669ec7 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,222 @@ # HPCSimSite HPC simulation site + +# F1 Virtual Race Engineer — Enrichment Module + +This repo contains a minimal, dependency-free Python module to enrich Raspberry Pi telemetry (derived from FastF1) with HPC-style analytics features. It simulates the first LLM stage (data enrichment) using deterministic heuristics so you can run the pipeline locally and in CI without external services. + +## What it does +- Accepts lap-level telemetry JSON records. +- Produces an enriched record with: + - aero_efficiency (0..1) + - tire_degradation_index (0..1, higher=worse) + - ers_charge (0..1) + - fuel_optimization_score (0..1) + - driver_consistency (0..1) + - weather_impact (low|medium|high) + +## Expected input schema +Fields are extensible; these cover the initial POC. + +Required (or sensible defaults applied): +- lap: int +- speed: float (km/h) +- throttle: float (0..1) +- brake: float (0..1) +- tire_compound: string (soft|medium|hard|inter|wet) +- fuel_level: float (0..1) + +Optional: +- ers: float (0..1) +- track_temp: float (Celsius) +- rain_probability: float (0..1) + +Example telemetry line (JSONL): +{"lap":27,"speed":282,"throttle":0.91,"brake":0.05,"tire_compound":"medium","fuel_level":0.47} + +## Output schema (enriched) +Example: +{"lap":27,"aero_efficiency":0.83,"tire_degradation_index":0.65,"ers_charge":0.72,"fuel_optimization_score":0.91,"driver_consistency":0.89,"weather_impact":"medium"} + +## Quick start + +### Run the CLI +The CLI reads JSON Lines (one JSON object per line) from stdin or a file and writes enriched JSON lines to stdout or a file. + +```bash +python3 scripts/enrich_telemetry.py -i telemetry.jsonl -o enriched.jsonl +``` + +Or stream: + +```bash +cat telemetry.jsonl | python3 scripts/enrich_telemetry.py > enriched.jsonl +``` + +### Library usage + +```python +from hpcsim.enrichment import Enricher + +enricher = Enricher() +out = enricher.enrich({ + "lap": 1, + "speed": 250, + "throttle": 0.8, + "brake": 0.1, + "tire_compound": "medium", + "fuel_level": 0.6, +}) +print(out) +``` + +## Notes +- The enrichment maintains state across laps (e.g., cumulative tire wear, consistency from last up to 5 laps). If you restart the process mid-race, these will reset; you can re-feed prior laps to restore state. +- If your FastF1-derived telemetry has a different shape, share a sample and we can add adapters. + +## Tests + +Run minimal tests: + +```bash +python3 -m unittest tests/test_enrichment.py -v +``` + +## API reference (Enrichment Service) + +Base URL (local): http://localhost:8000 + +Interactive docs: http://localhost:8000/docs (Swagger) and http://localhost:8000/redoc + +### Run the API server + +```bash +python3 scripts/serve.py +``` + +Optional downstream forwarding: + +```bash +export NEXT_STAGE_CALLBACK_URL="http://localhost:9000/next-stage" +python3 scripts/serve.py +``` + +When set, every enriched record is also POSTed to the callback URL (best-effort, async). Ingestion still returns 200 even if forwarding fails. + +### POST /ingest/telemetry + +Accepts raw Raspberry Pi or FastF1-style telemetry, normalizes field names, enriches it, stores a recent copy, and returns the enriched record. + +- Content-Type: application/json +- Request body (flexible/aliases allowed): + - lap (int) — aliases: Lap, LapNumber + - speed (float, km/h) — alias: Speed + - throttle (0..1) — alias: Throttle + - brake (0..1) — aliases: Brake, Brakes + - tire_compound (string: soft|medium|hard|inter|wet) — aliases: Compound, TyreCompound, Tire + - fuel_level (0..1) — aliases: Fuel, FuelRel, FuelLevel + - ers (0..1) — aliases: ERS, ERSCharge (optional) + - track_temp (Celsius) — alias: TrackTemp (optional) + - rain_probability (0..1) — aliases: RainProb, PrecipProb (optional) + +Example request: + +```bash +curl -s -X POST http://localhost:8000/ingest/telemetry \ + -H "Content-Type: application/json" \ + -d '{ + "LapNumber": 27, + "Speed": 282, + "Throttle": 0.91, + "Brakes": 0.05, + "TyreCompound": "medium", + "FuelRel": 0.47 + }' +``` + +Response 200 (application/json): + +```json +{ + "lap": 27, + "aero_efficiency": 0.83, + "tire_degradation_index": 0.65, + "ers_charge": 0.72, + "fuel_optimization_score": 0.91, + "driver_consistency": 0.89, + "weather_impact": "medium" +} +``` + +Errors: +- 400 if the body cannot be normalized/enriched + +### POST /enriched + +Accepts an already-enriched record (useful if enrichment runs elsewhere). Stores and echoes it back. + +- Content-Type: application/json +- Request body: + - lap: int + - aero_efficiency: float (0..1) + - tire_degradation_index: float (0..1) + - ers_charge: float (0..1) + - fuel_optimization_score: float (0..1) + - driver_consistency: float (0..1) + - weather_impact: string (low|medium|high) + +Example: + +```bash +curl -s -X POST http://localhost:8000/enriched \ + -H "Content-Type: application/json" \ + -d '{ + "lap": 99, + "aero_efficiency": 0.8, + "tire_degradation_index": 0.5, + "ers_charge": 0.6, + "fuel_optimization_score": 0.9, + "driver_consistency": 0.95, + "weather_impact": "low" + }' +``` + +### GET /enriched + +Returns an array of the most recent enriched records. + +- Query params: + - limit: int (1..200, default 50) + +Example: + +```bash +curl -s "http://localhost:8000/enriched?limit=10" +``` + +Response 200 example: + +```json +[ + { "lap": 27, "aero_efficiency": 0.83, "tire_degradation_index": 0.65, "ers_charge": 0.72, "fuel_optimization_score": 0.91, "driver_consistency": 0.89, "weather_impact": "medium" } +] +``` + +### GET /healthz + +Health check. + +```bash +curl -s http://localhost:8000/healthz +``` + +Response 200 example: + +```json +{ "status": "ok", "stored": 5 } +``` + +### Notes +- Authentication/authorization is not enabled yet; add API keys or headers if deploying externally. +- Storage is in-memory (most recent ~200 records). For persistence, we can add Redis/SQLite. +- Forwarding to downstream (e.g., strategy LLM stage) is opt-in via `NEXT_STAGE_CALLBACK_URL`. diff --git a/hpcsim/__init__.py b/hpcsim/__init__.py new file mode 100644 index 0000000..dde92e0 --- /dev/null +++ b/hpcsim/__init__.py @@ -0,0 +1 @@ +__all__ = ["enrichment"] diff --git a/hpcsim/__pycache__/__init__.cpython-312.pyc b/hpcsim/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..41ddfde Binary files /dev/null and b/hpcsim/__pycache__/__init__.cpython-312.pyc differ diff --git a/hpcsim/__pycache__/adapter.cpython-312.pyc b/hpcsim/__pycache__/adapter.cpython-312.pyc new file mode 100644 index 0000000..9af8c24 Binary files /dev/null and b/hpcsim/__pycache__/adapter.cpython-312.pyc differ diff --git a/hpcsim/__pycache__/api.cpython-312.pyc b/hpcsim/__pycache__/api.cpython-312.pyc new file mode 100644 index 0000000..aecc968 Binary files /dev/null and b/hpcsim/__pycache__/api.cpython-312.pyc differ diff --git a/hpcsim/__pycache__/enrichment.cpython-312.pyc b/hpcsim/__pycache__/enrichment.cpython-312.pyc new file mode 100644 index 0000000..06078c7 Binary files /dev/null and b/hpcsim/__pycache__/enrichment.cpython-312.pyc differ diff --git a/hpcsim/adapter.py b/hpcsim/adapter.py new file mode 100644 index 0000000..5909ef3 --- /dev/null +++ b/hpcsim/adapter.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +from typing import Dict, Any + + +def normalize_telemetry(payload: Dict[str, Any]) -> Dict[str, Any]: + """Normalize Pi/FastF1-like telemetry payload to Enricher expected schema. + + Accepted aliases: + - speed: Speed + - throttle: Throttle + - brake: Brake, Brakes + - tire_compound: Compound, TyreCompound, Tire + - fuel_level: Fuel, FuelRel, FuelLevel + - ers: ERS, ERSCharge + - track_temp: TrackTemp + - rain_probability: RainProb, PrecipProb + - lap: Lap, LapNumber + Values are clamped and defaulted if missing. + """ + aliases = { + "lap": ["lap", "Lap", "LapNumber"], + "speed": ["speed", "Speed"], + "throttle": ["throttle", "Throttle"], + "brake": ["brake", "Brake", "Brakes"], + "tire_compound": ["tire_compound", "Compound", "TyreCompound", "Tire"], + "fuel_level": ["fuel_level", "Fuel", "FuelRel", "FuelLevel"], + "ers": ["ers", "ERS", "ERSCharge"], + "track_temp": ["track_temp", "TrackTemp"], + "rain_probability": ["rain_probability", "RainProb", "PrecipProb"], + } + + out: Dict[str, Any] = {} + + def pick(key: str, default=None): + for k in aliases.get(key, [key]): + if k in payload and payload[k] is not None: + return payload[k] + return default + + def clamp01(x, default=0.0): + try: + v = float(x) + except (TypeError, ValueError): + return default + return max(0.0, min(1.0, v)) + + # Map values with sensible defaults + lap = pick("lap", 0) + try: + lap = int(lap) + except (TypeError, ValueError): + lap = 0 + + speed = pick("speed", 0.0) + try: + speed = float(speed) + except (TypeError, ValueError): + speed = 0.0 + + throttle = clamp01(pick("throttle", 0.0), 0.0) + brake = clamp01(pick("brake", 0.0), 0.0) + + tire_compound = pick("tire_compound", "medium") + if isinstance(tire_compound, str): + tire_compound = tire_compound.lower() + else: + tire_compound = "medium" + + fuel_level = clamp01(pick("fuel_level", 0.5), 0.5) + + ers = pick("ers", None) + if ers is not None: + ers = clamp01(ers, None) + + track_temp = pick("track_temp", None) + try: + track_temp = float(track_temp) if track_temp is not None else None + except (TypeError, ValueError): + track_temp = None + + rain_prob = pick("rain_probability", None) + try: + rain_prob = clamp01(rain_prob, None) if rain_prob is not None else None + except Exception: + rain_prob = None + + out.update({ + "lap": lap, + "speed": speed, + "throttle": throttle, + "brake": brake, + "tire_compound": tire_compound, + "fuel_level": fuel_level, + }) + if ers is not None: + out["ers"] = ers + if track_temp is not None: + out["track_temp"] = track_temp + if rain_prob is not None: + out["rain_probability"] = rain_prob + + return out diff --git a/hpcsim/api.py b/hpcsim/api.py new file mode 100644 index 0000000..8109887 --- /dev/null +++ b/hpcsim/api.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +import os +from typing import Any, Dict, List, Optional + +from fastapi import FastAPI, Body, HTTPException +from fastapi.responses import JSONResponse +from pydantic import BaseModel +import httpx + +from .enrichment import Enricher +from .adapter import normalize_telemetry + +app = FastAPI(title="HPCSim Enrichment API", version="0.1.0") + +# Single Enricher instance keeps state across laps +_enricher = Enricher() + +# Simple in-memory store of recent enriched records +_recent: List[Dict[str, Any]] = [] +_MAX_RECENT = 200 + +# Optional callback URL to forward enriched data to next stage +_CALLBACK_URL = os.getenv("NEXT_STAGE_CALLBACK_URL") + + +class EnrichedRecord(BaseModel): + lap: int + aero_efficiency: float + tire_degradation_index: float + ers_charge: float + fuel_optimization_score: float + driver_consistency: float + weather_impact: str + + +@app.post("/ingest/telemetry") +async def ingest_telemetry(payload: Dict[str, Any] = Body(...)): + """Receive raw telemetry (from Pi), normalize, enrich, return enriched. + + Optionally forwards to NEXT_STAGE_CALLBACK_URL if set. + """ + try: + normalized = normalize_telemetry(payload) + enriched = _enricher.enrich(normalized) + except Exception as e: + raise HTTPException(status_code=400, detail=f"Failed to enrich: {e}") + + _recent.append(enriched) + if len(_recent) > _MAX_RECENT: + del _recent[: len(_recent) - _MAX_RECENT] + + # Async forward to next stage if configured + if _CALLBACK_URL: + try: + async with httpx.AsyncClient(timeout=5.0) as client: + await client.post(_CALLBACK_URL, json=enriched) + except Exception: + # Don't fail ingestion if forwarding fails; log could be added here + pass + + return JSONResponse(enriched) + + +@app.post("/enriched") +async def post_enriched(enriched: EnrichedRecord): + """Allow posting externally enriched records (bypass local computation).""" + rec = enriched.model_dump() + _recent.append(rec) + if len(_recent) > _MAX_RECENT: + del _recent[: len(_recent) - _MAX_RECENT] + return JSONResponse(rec) + + +@app.get("/enriched") +async def list_enriched(limit: int = 50): + limit = max(1, min(200, limit)) + return JSONResponse(_recent[-limit:]) + + +@app.get("/healthz") +async def healthz(): + return {"status": "ok", "stored": len(_recent)} diff --git a/hpcsim/enrichment.py b/hpcsim/enrichment.py new file mode 100644 index 0000000..48e0157 --- /dev/null +++ b/hpcsim/enrichment.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Dict, Any, Optional +import math + + +# --- Contracts --- +# Input telemetry (example, extensible): +# { +# "lap": 27, +# "speed": 282, # km/h +# "throttle": 0.91, # 0..1 +# "brake": 0.05, # 0..1 +# "tire_compound": "medium",# soft|medium|hard|inter|wet +# "fuel_level": 0.47, # 0..1 (fraction of race fuel) +# "ers": 0.72, # optional 0..1 +# "track_temp": 38, # optional Celsius +# "rain_probability": 0.2 # optional 0..1 +# } +# +# Output enrichment: +# { +# "lap": 27, +# "aero_efficiency": 0.83, # 0..1 +# "tire_degradation_index": 0.65, # 0..1 (higher=worse) +# "ers_charge": 0.72, # 0..1 +# "fuel_optimization_score": 0.91, # 0..1 +# "driver_consistency": 0.89, # 0..1 +# "weather_impact": "low|medium|high" +# } + + +_TIRES_BASE_WEAR = { + "soft": 0.012, + "medium": 0.008, + "hard": 0.006, + "inter": 0.015, + "wet": 0.02, +} + + +@dataclass +class EnricherState: + last_lap: Optional[int] = None + lap_speeds: Dict[int, float] = field(default_factory=dict) + lap_throttle_avg: Dict[int, float] = field(default_factory=dict) + cumulative_wear: float = 0.0 # 0..1 approx + + +class Enricher: + """Heuristic enrichment engine to simulate HPC analytics on telemetry. + + Stateless inputs are enriched with stateful estimates (wear, consistency, etc.). + Designed for predictable, dependency-free behavior. + """ + + def __init__(self): + self.state = EnricherState() + + # --- Public API --- + def enrich(self, telemetry: Dict[str, Any]) -> Dict[str, Any]: + lap = int(telemetry.get("lap", 0)) + speed = float(telemetry.get("speed", 0.0)) + throttle = float(telemetry.get("throttle", 0.0)) + brake = float(telemetry.get("brake", 0.0)) + tire_compound = str(telemetry.get("tire_compound", "medium")).lower() + fuel_level = float(telemetry.get("fuel_level", 0.5)) + ers = telemetry.get("ers") + track_temp = telemetry.get("track_temp") + rain_prob = telemetry.get("rain_probability") + + # Update per-lap aggregates + self._update_lap_stats(lap, speed, throttle) + + # Metrics + aero_eff = self._compute_aero_efficiency(speed, throttle, brake) + tire_deg = self._compute_tire_degradation(lap, speed, throttle, tire_compound, track_temp) + ers_charge = self._compute_ers_charge(ers, throttle, brake) + fuel_opt = self._compute_fuel_optimization(fuel_level, throttle) + consistency = self._compute_driver_consistency() + weather_impact = self._compute_weather_impact(rain_prob, track_temp) + + return { + "lap": lap, + "aero_efficiency": round(aero_eff, 3), + "tire_degradation_index": round(tire_deg, 3), + "ers_charge": round(ers_charge, 3), + "fuel_optimization_score": round(fuel_opt, 3), + "driver_consistency": round(consistency, 3), + "weather_impact": weather_impact, + } + + # --- Internals --- + def _update_lap_stats(self, lap: int, speed: float, throttle: float) -> None: + if lap <= 0: + return + # Store simple aggregates for consistency metrics + self.state.lap_speeds[lap] = speed + self.state.lap_throttle_avg[lap] = 0.8 * self.state.lap_throttle_avg.get(lap, throttle) + 0.2 * throttle + self.state.last_lap = lap + + def _compute_aero_efficiency(self, speed: float, throttle: float, brake: float) -> float: + # Heuristic: favor high speed with low throttle variance (efficiency) and minimal braking at high speeds + # Normalize speed into 0..1 assuming 0..330 km/h typical + speed_n = max(0.0, min(1.0, speed / 330.0)) + brake_penalty = 0.4 * brake + throttle_bonus = 0.2 * throttle + base = 0.5 * speed_n + throttle_bonus - brake_penalty + return max(0.0, min(1.0, base)) + + def _compute_tire_degradation(self, lap: int, speed: float, throttle: float, tire_compound: str, track_temp: Optional[float]) -> float: + base_wear = _TIRES_BASE_WEAR.get(tire_compound, _TIRES_BASE_WEAR["medium"]) # per lap + temp_factor = 1.0 + if isinstance(track_temp, (int, float)): + if track_temp > 42: + temp_factor = 1.25 + elif track_temp < 15: + temp_factor = 0.9 + stress = 0.5 + 0.5 * throttle + 0.2 * max(0.0, (speed - 250.0) / 100.0) + wear_this_lap = base_wear * stress * temp_factor + # Update cumulative wear but cap at 1.0 + self.state.cumulative_wear = min(1.0, self.state.cumulative_wear + wear_this_lap) + return self.state.cumulative_wear + + def _compute_ers_charge(self, ers: Optional[float], throttle: float, brake: float) -> float: + if isinstance(ers, (int, float)): + # simple recovery under braking, depletion under throttle + ers_level = float(ers) + 0.1 * brake - 0.05 * throttle + else: + # infer ers trend if not provided + ers_level = 0.6 + 0.05 * brake - 0.03 * throttle + return max(0.0, min(1.0, ers_level)) + + def _compute_fuel_optimization(self, fuel_level: float, throttle: float) -> float: + # Reward keeping throttle moderate when fuel is low and pushing when fuel is high + fuel_n = max(0.0, min(1.0, fuel_level)) + ideal_throttle = 0.5 + 0.4 * fuel_n # higher fuel -> higher ideal throttle + penalty = abs(throttle - ideal_throttle) + score = 1.0 - penalty + return max(0.0, min(1.0, score)) + + def _compute_driver_consistency(self) -> float: + # Use last up to 5 laps speed variance to estimate consistency (lower variance -> higher consistency) + laps = sorted(self.state.lap_speeds.keys())[-5:] + if not laps: + return 0.5 + speeds = [self.state.lap_speeds[l] for l in laps] + mean = sum(speeds) / len(speeds) + var = sum((s - mean) ** 2 for s in speeds) / len(speeds) + # Map variance to 0..1; assume 0..(30 km/h)^2 typical range + norm = min(1.0, var / (30.0 ** 2)) + return max(0.0, 1.0 - norm) + + def _compute_weather_impact(self, rain_prob: Optional[float], track_temp: Optional[float]) -> str: + score = 0.0 + if isinstance(rain_prob, (int, float)): + score += 0.7 * float(rain_prob) + if isinstance(track_temp, (int, float)): + if track_temp < 12: # cold tires harder + score += 0.2 + if track_temp > 45: # overheating + score += 0.2 + if score < 0.3: + return "low" + if score < 0.6: + return "medium" + return "high" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..742213c --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +fastapi==0.115.2 +uvicorn==0.31.1 +httpx==0.27.2 diff --git a/scripts/enrich_telemetry.py b/scripts/enrich_telemetry.py new file mode 100644 index 0000000..dc71396 --- /dev/null +++ b/scripts/enrich_telemetry.py @@ -0,0 +1,47 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import json +import sys +from typing import Iterable, Dict, Any + +from hpcsim.enrichment import Enricher + + +def iter_json_lines(stream) -> Iterable[Dict[str, Any]]: + for line in stream: + line = line.strip() + if not line: + continue + try: + yield json.loads(line) + except json.JSONDecodeError: + print(f"Skipping invalid JSON line: {line}", file=sys.stderr) + + +def main(): + parser = argparse.ArgumentParser(description="Enrich telemetry JSON lines with HPC-style metrics") + parser.add_argument("--input", "-i", help="Input file path (JSON lines). Reads stdin if omitted.") + parser.add_argument("--output", "-o", help="Output file path (JSON lines). Writes stdout if omitted.") + args = parser.parse_args() + + enricher = Enricher() + + fin = open(args.input, "r") if args.input else sys.stdin + fout = open(args.output, "w") if args.output else sys.stdout + + try: + for rec in iter_json_lines(fin): + enriched = enricher.enrich(rec) + print(json.dumps(enriched), file=fout) + fout.flush() + finally: + if fin is not sys.stdin: + fin.close() + if fout is not sys.stdout: + fout.close() + + +if __name__ == "__main__": + main() diff --git a/scripts/serve.py b/scripts/serve.py new file mode 100644 index 0000000..1f57a3a --- /dev/null +++ b/scripts/serve.py @@ -0,0 +1,8 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import uvicorn + + +if __name__ == "__main__": + uvicorn.run("hpcsim.api:app", host="0.0.0.0", port=8000, reload=False) diff --git a/tests/__pycache__/test_adapter.cpython-312.pyc b/tests/__pycache__/test_adapter.cpython-312.pyc new file mode 100644 index 0000000..d019dbb Binary files /dev/null and b/tests/__pycache__/test_adapter.cpython-312.pyc differ diff --git a/tests/__pycache__/test_api.cpython-312.pyc b/tests/__pycache__/test_api.cpython-312.pyc new file mode 100644 index 0000000..8adacc7 Binary files /dev/null and b/tests/__pycache__/test_api.cpython-312.pyc differ diff --git a/tests/__pycache__/test_enrichment.cpython-312.pyc b/tests/__pycache__/test_enrichment.cpython-312.pyc new file mode 100644 index 0000000..15d5033 Binary files /dev/null and b/tests/__pycache__/test_enrichment.cpython-312.pyc differ diff --git a/tests/test_adapter.py b/tests/test_adapter.py new file mode 100644 index 0000000..a003a71 --- /dev/null +++ b/tests/test_adapter.py @@ -0,0 +1,32 @@ +import unittest + +from hpcsim.adapter import normalize_telemetry + + +class TestAdapter(unittest.TestCase): + def test_alias_mapping_and_clamping(self): + raw = { + "LapNumber": "12", + "Speed": "280.5", + "Throttle": 1.2, # will clamp + "Brakes": -0.1, # will clamp + "TyreCompound": "Soft", + "FuelRel": 1.5, # will clamp + "ERS": 0.45, + "TrackTemp": "41", + "RainProb": 0.3, + } + norm = normalize_telemetry(raw) + self.assertEqual(norm["lap"], 12) + self.assertAlmostEqual(norm["speed"], 280.5, places=2) + self.assertEqual(norm["throttle"], 1.0) + self.assertEqual(norm["brake"], 0.0) + self.assertEqual(norm["tire_compound"], "soft") + self.assertEqual(norm["fuel_level"], 1.0) + self.assertIn("ers", norm) + self.assertIn("track_temp", norm) + self.assertIn("rain_probability", norm) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..affa6ba --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,47 @@ +import unittest + +from fastapi.testclient import TestClient + +from hpcsim.api import app + + +class TestAPI(unittest.TestCase): + def setUp(self): + self.client = TestClient(app) + + def test_ingest_and_list(self): + payload = { + "lap": 1, + "speed": 250, + "throttle": 0.8, + "brake": 0.1, + "tire_compound": "medium", + "fuel_level": 0.6, + } + r = self.client.post("/ingest/telemetry", json=payload) + self.assertEqual(r.status_code, 200) + enriched = r.json() + self.assertIn("aero_efficiency", enriched) + + list_r = self.client.get("/enriched", params={"limit": 5}) + self.assertEqual(list_r.status_code, 200) + data = list_r.json() + self.assertTrue(isinstance(data, list) and len(data) >= 1) + + def test_post_enriched(self): + enriched = { + "lap": 99, + "aero_efficiency": 0.8, + "tire_degradation_index": 0.5, + "ers_charge": 0.6, + "fuel_optimization_score": 0.9, + "driver_consistency": 0.95, + "weather_impact": "low", + } + r = self.client.post("/enriched", json=enriched) + self.assertEqual(r.status_code, 200) + self.assertEqual(r.json()["lap"], 99) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_enrichment.py b/tests/test_enrichment.py new file mode 100644 index 0000000..bb7c855 --- /dev/null +++ b/tests/test_enrichment.py @@ -0,0 +1,46 @@ +import unittest + +from hpcsim.enrichment import Enricher + + +class TestEnrichment(unittest.TestCase): + def test_basic_ranges(self): + e = Enricher() + sample = { + "lap": 1, + "speed": 250, + "throttle": 0.8, + "brake": 0.1, + "tire_compound": "medium", + "fuel_level": 0.6, + "ers": 0.5, + "track_temp": 35, + "rain_probability": 0.1, + } + out = e.enrich(sample) + self.assertIn("aero_efficiency", out) + self.assertTrue(0.0 <= out["aero_efficiency"] <= 1.0) + self.assertTrue(0.0 <= out["tire_degradation_index"] <= 1.0) + self.assertTrue(0.0 <= out["ers_charge"] <= 1.0) + self.assertTrue(0.0 <= out["fuel_optimization_score"] <= 1.0) + self.assertTrue(0.0 <= out["driver_consistency"] <= 1.0) + self.assertIn(out["weather_impact"], {"low", "medium", "high"}) + + def test_stateful_wear_increases(self): + e = Enricher() + prev = 0.0 + for lap in range(1, 6): + out = e.enrich({ + "lap": lap, + "speed": 260, + "throttle": 0.9, + "brake": 0.05, + "tire_compound": "soft", + "fuel_level": 0.7, + }) + self.assertGreaterEqual(out["tire_degradation_index"], prev) + prev = out["tire_degradation_index"] + + +if __name__ == "__main__": + unittest.main()