"""
Live read-through adapter  —  §6 seam #1: data → engine.

Production replacement for the demo's batch exporter (demo/export/export_team_race.py). Instead of
pulling a fixed window of tracks from Firestore and scoring in one shot, this reads V2's live position
feed for an in-progress start group and produces the SAME feed dict the engine consumes —
`{meta, course, teams, frames}` — incrementally as positions arrive, so `core.score_race.enrich`
(via StreamingScorer) is reused verbatim. The batch exporter stays for replays / off-season.

Input (V2): the running backend exposes the live cache over HTTP —
    GET /api/viz/{instanceId}/positions  ->  {deviceId: {lat, lon, speed, course, timestampUtcMs, ...}}
(see _legacy/app_viewregatta_com/routers/visualization.py + services/live_state.py). This adapter polls that
endpoint and maps each device's latest fix into the engine's frame contract.

Output contract (identical to the demo exporter, so the engine is unchanged):
    meta   : {title, user, instanceId, groupId, openEnded, raceType,
              trackStartUtcMs, gunUtcMs, trackEndUtcMs, frameStepMs, windFromDeg, windKts}
    course : {name, center, startLine{port,starboard}, finishLine{port,starboard}, waypoints[]}
    teams  : [{teamId, name, boats:[{deviceId, sailId, visualId, color, sternM}]}]
    frames : {"<sec>": {deviceId: {lat, lon, speed, course}}}   sec = (timestampUtcMs - trackStartUtcMs)//1000

What is implemented vs. wired:
  * `positions_to_frames` — the pure GPS→frame transform (mirrors the exporter's `pull_frames` record
    shape: lat/lon 7 dp, speed 2 dp, course 1 dp). Fully unit-tested.
  * `poll_frames` — incremental streaming with de-dup, over an **injectable** position source (default:
    HTTP GET to the V2 endpoint via `requests`/`urllib`). Testable with a fake source.
  * `resolve_meta_course_teams` — needs V2 Firestore (raceInstances / startGroups / courses /
    deviceMounts, resolving the **time-scoped device→boat** assignment in effect at the gun). That
    requires V2 credentials, so it runs through an **injectable** `fetch_static` callable; the default
    raises with a pointer to the exporter logic to port (device_for_boat + course doc in
    demo/export/export_team_race.py). Wire this when running against a live V2 instance.

Wind is manual / weather-station only (never motion-derived) — passed in via `wind`.
Sub-phase: B-A (live scoring / live input).
"""
from __future__ import annotations

import sys
import time
from collections.abc import Callable, Iterator
from pathlib import Path

# the scoring engine is the shared core (kept on the path for callers that import enrich alongside)
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "scoring"))


def positions_to_frames(positions: dict, track_start_ms: int) -> dict:
    """Map a V2 live-positions snapshot to engine frames `{ "<sec>": {dev: {lat,lon,speed,course}} }`.
    Pure: skips devices with no fix or a pre-start (negative-sec) timestamp; rounds like the exporter."""
    out: dict[str, dict] = {}
    for dev, p in (positions or {}).items():
        ts = p.get("timestampUtcMs")
        if ts is None or p.get("lat") is None or p.get("lon") is None:
            continue
        sec = int((ts - track_start_ms) // 1000)
        if sec < 0:
            continue
        out.setdefault(str(sec), {})[dev] = {
            "lat": round(p["lat"], 7), "lon": round(p["lon"], 7),
            "speed": round(p.get("speed", 0) or 0, 2), "course": round(p.get("course", 0) or 0, 1),
        }
    return out


class LiveReadThroughAdapter:
    """Streams the broadcast feed for one live start group out of V2's live position cache."""

    def __init__(self, instance_id: str, group_id: str, wind: dict | None = None, *,
                 base_url: str | None = None, track_start_ms: int | None = None,
                 fetch_positions: Callable[[], dict] | None = None,
                 fetch_static: Callable[[], tuple[dict, dict, list]] | None = None):
        self.instance_id = instance_id
        self.group_id = group_id
        self.wind = wind or {}                       # {"fromDeg": ..., "kts": ...} — manual / weather station
        self.base_url = base_url.rstrip("/") if base_url else None
        self.track_start_ms = track_start_ms         # set here or by resolve_meta_course_teams
        self._fetch_positions = fetch_positions or self._http_fetch_positions
        self._fetch_static = fetch_static

    def resolve_meta_course_teams(self) -> tuple[dict, dict, list]:
        """Build the static feed parts at gun time. Delegates to the injected `fetch_static` (production
        wires this to the V2 Firestore reads — raceInstances/startGroups/courses/deviceMounts with the
        time-scoped device→boat assignment, as in the batch exporter). Caches trackStartUtcMs for
        frame timing."""
        if self._fetch_static is None:
            raise NotImplementedError(
                "Provide fetch_static (meta, course, teams) — port device_for_boat + course doc + the "
                "time-scoped deviceMounts from demo/export/export_team_race.py against the live V2 instance."
            )
        meta, course, teams = self._fetch_static()
        if meta.get("trackStartUtcMs") is not None:
            self.track_start_ms = meta["trackStartUtcMs"]
        return meta, course, teams

    def poll_once(self, seen: set[tuple[str, str]]) -> dict:
        """ONE poll: the frames that are new since `seen` (which this call updates). Split out of
        `poll_frames` so a producer that owns its own loop — one that can hand over to the NEXT race
        between ticks without restarting (`serve_v3_live.broadcast_following`) — shares this exact
        de-duplication rather than reimplementing it."""
        if self.track_start_ms is None:
            raise RuntimeError("track_start_ms unknown — call resolve_meta_course_teams() or pass it in.")
        batch = positions_to_frames(self._fetch_positions(), self.track_start_ms)
        new: dict[str, dict] = {}
        for sec, devs in batch.items():
            for dev, rec in devs.items():
                if (dev, sec) in seen:
                    continue
                seen.add((dev, sec))
                new.setdefault(sec, {})[dev] = rec
        return new

    def poll_frames(self, *, interval_s: float = 1.0, max_polls: int | None = None,
                    sleep: Callable[[float], None] = time.sleep) -> Iterator[dict]:
        """Yield incremental frame batches `{ "<sec>": {dev: {lat,lon,speed,course}} }` as new fixes land,
        de-duped by (device, sec) so each second is emitted once. `max_polls`/`sleep` are injectable for
        tests; in production leave them default and stop by breaking out of the generator."""
        seen: set[tuple[str, str]] = set()
        polls = 0
        while max_polls is None or polls < max_polls:
            polls += 1
            new = self.poll_once(seen)
            if new:
                yield new
            if max_polls is None or polls < max_polls:
                sleep(interval_s)

    # --- default HTTP source (isolated so the module imports without `requests`) ---
    def _http_fetch_positions(self) -> dict:
        if not self.base_url:
            raise RuntimeError("No base_url set and no fetch_positions injected — cannot reach V2 live_state.")
        url = f"{self.base_url}/api/viz/{self.instance_id}/positions"
        try:
            import requests
            return requests.get(url, timeout=5).json()
        except ImportError:
            import json
            import urllib.request
            with urllib.request.urlopen(url, timeout=5) as r:  # noqa: S310 (trusted internal V2 URL)
                return json.loads(r.read().decode("utf-8"))