minisky_tangram¶

def convert_snapshot(snapshot: Snapshot) -> TangramPayload:
    """Convert a MiniSky SI-unit snapshot into the tangram wire payload.

    Aircraft come out with jet1090-style field names and aviation units
    (altitude in ft, speeds in kt, vertical rate in fpm). Pure function so
    it can be unit-tested without a running simulator.
    """
    siminfo = snapshot["siminfo"]
    acdata = snapshot["acdata"]

    state = siminfo["state"]
    simutc = siminfo["simutc"]
    try:
        utc = datetime.fromisoformat(simutc)
    except ValueError:
        utc = None
    if utc is not None and utc.tzinfo is None:
        # sim.utc is UTC by definition; never let a naive string be read as local time.
        utc = utc.replace(tzinfo=UTC)
    timestamp = utc.timestamp() if utc is not None else None

    out_siminfo: TangramSimInfo = {
        "simt": siminfo["simt"],
        "simdt": siminfo["simdt"],
        "simutc": simutc,
        "speed": siminfo["speed"],
        "ntraf": siminfo["ntraf"],
        "state": state,
        "state_name": _state_name(state),
        "scenname": siminfo["scenname"],
        "nconf_cur": acdata["nconf_cur"],
        "nlos_cur": acdata["nlos_cur"],
    }

    aircraft: list[TangramAircraft] = []
    for i, callsign in enumerate(acdata["callsign"]):
        altitude_ft: q.PressureAltitudeFt[float] = q.m_to_ft(acdata["alt"][i])
        groundspeed_kt: q.GroundSpeedKt[float] = q.mps_to_kt(acdata["gs"][i])
        tas_kt: q.TrueAirspeedKt[float] = q.mps_to_kt(acdata["tas"][i])
        ias_kt: q.CalibratedAirspeedKt[float] = q.mps_to_kt(acdata["cas"][i])
        vertical_rate_fpm: q.VerticalRateFpm[float] = q.mps_to_fpm(acdata["vs"][i])
        aircraft.append(
            {
                "id": callsign,
                "callsign": callsign,
                "typecode": acdata["typecode"][i],
                "latitude": acdata["lat"][i],
                "longitude": acdata["lon"][i],
                "altitude": round(altitude_ft),
                "groundspeed": round(groundspeed_kt, 1),
                "tas": round(tas_kt, 1),
                "ias": round(ias_kt, 1),
                "vertical_rate": round(vertical_rate_fpm),
                "track": acdata["trk"][i],
                "inconf": acdata["inconf"][i],
                "timestamp": timestamp,
            }
        )

    return {"aircraft": aircraft, "count": len(aircraft), "siminfo": out_siminfo}

def extract_command(payload: str | bytes) -> str | None:
    """Extract the stack command from a `from:<channel>:command` payload.

    Accepts the JSON envelope pushed by the tangram frontend
    (`{"command": "..."}`) and, for convenience when testing with
    redis-cli, a bare string.
    """
    if isinstance(payload, bytes):
        payload = payload.decode("utf-8", errors="replace")
    text = payload.strip()
    if not text:
        return None
    try:
        data = json.loads(text)
    except ValueError:
        return text
    if isinstance(data, dict):
        cmd = data.get("command")
        return str(cmd).strip() or None if cmd is not None else None
    if isinstance(data, str):
        return data.strip() or None
    return None

def build(context: PluginContext[TangramConfig]) -> PluginSpec:
    bridge = context.mount(
        TangramBridge(
            redis_url=context.config.redis_url,
            channel=context.config.channel,
            max_hz=context.config.max_hz,
        )
    )

    @asynccontextmanager
    async def lifespan(runtime: PluginRuntime) -> AsyncGenerator[None]:
        runtime.subscribe_console(bridge.capture_console)
        match bridge.start(runtime):
            case Ok(message):
                runtime.echo(message)
            case Err(message):
                runtime.echo(message)
                raise RuntimeError(message)
        try:
            yield
        finally:
            bridge.stop()

    return context.finish(lifespan=lifespan)