# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "duckdb>=1.5.5",
#     "marimo",
#     "pandas",
#     "scrydon",
#     "sqlglot>=30.15.0",
# ]
# ///
"""MIK zone exposure pipeline: which registry vessels last reported inside a governed area.

An authoring notebook, run by a workspace member — never published as an app.
It reads one managed table with SQL and one ontology type, joins them in
DataFrames, and writes the result to the managed table the
``MaritimeZoneExposure`` binding reads. The runtime ships ``scrydon`` without
its SQL engine, so the block above declares duckdb and sqlglot. The
North Sea app then shows those objects through its declared ``exposure`` query,
under the viewer's authority.
"""

import marimo

__generated_with = "0.23.14"
app = marimo.App(width="medium")

with app.setup:
    import json

    import pandas as pd

    ONTOLOGY = "scrydon-pack-mik-north-sea-v2"
    REGISTRY_TABLE = "mik_vessel_registry"
    EXPOSURE_TABLE = "mik_vessel_zone_exposure"
    # The MaritimeZoneExposure binding identifies objects by this column, and a
    # governed read serves the table only when it is the table's primary key.
    EXPOSURE_KEY = ["exposure_id"]
    PROVENANCE = "Fictional demo data · derived in exposure.py"
    # The columns MaritimeZoneExposure binds, in the order the table is created.
    EXPOSURE_COLUMNS = {
        "exposure_id": "object",
        "vessel_id": "object",
        "vessel_name": "object",
        "area_id": "object",
        "zone_name": "object",
        "zone_category": "object",
        "risk_score": "float64",
        "risk_level": "object",
        "latitude": "float64",
        "longitude": "float64",
        "computed_at": "datetime64[ns, UTC]",
        "provenance": "object",
    }


@app.function
def ring_contains(ring: list, longitude: float, latitude: float) -> bool:
    """Even-odd ray cast against one closed GeoJSON ring of [longitude, latitude] pairs."""
    inside = False
    for (x1, y1), (x2, y2) in zip(ring, ring[1:]):
        if (y1 > latitude) != (y2 > latitude):
            crossing = x1 + (latitude - y1) * (x2 - x1) / (y2 - y1)
            if longitude < crossing:
                inside = not inside
    return inside


@app.function
def boundary_contains(boundary: str, longitude: float, latitude: float) -> bool:
    """True when the point lies in a GeoJSON Polygon or MultiPolygon, outside its holes."""
    geometry = json.loads(boundary)
    if geometry["type"] == "Polygon":
        polygons = [geometry["coordinates"]]
    elif geometry["type"] == "MultiPolygon":
        polygons = geometry["coordinates"]
    else:
        raise ValueError(f"Unsupported boundary type: {geometry['type']}")
    return any(
        ring_contains(outer, longitude, latitude)
        and not any(ring_contains(hole, longitude, latitude) for hole in holes)
        for outer, *holes in polygons
    )


@app.function
def zone_exposure(vessels: pd.DataFrame, zones: pd.DataFrame, computed_at: pd.Timestamp) -> pd.DataFrame:
    """One row per vessel whose last position lies inside a zone's boundary."""
    rows = [
        {
            "exposure_id": f"{vessel.vessel_id}:{zone.areaId}",
            "vessel_id": vessel.vessel_id,
            "vessel_name": vessel.name,
            "area_id": zone.areaId,
            "zone_name": zone.name,
            "zone_category": zone.category,
            "risk_score": vessel.risk_score,
            "risk_level": vessel.risk_level,
            "latitude": vessel.last_latitude,
            "longitude": vessel.last_longitude,
            "computed_at": computed_at,
            "provenance": PROVENANCE,
        }
        for vessel in vessels.itertuples(index=False)
        for zone in zones.itertuples(index=False)
        if boundary_contains(zone.boundary, vessel.last_longitude, vessel.last_latitude)
    ]
    # Typed even when empty, so a first run on no matches still creates the
    # table with numeric and timestamp columns.
    return pd.DataFrame(rows, columns=list(EXPOSURE_COLUMNS)).astype(EXPOSURE_COLUMNS)


@app.cell
def _():
    import marimo as mo
    from scrydon import ontology, sql, tables

    return mo, ontology, sql, tables


@app.cell
def _(sql):
    # A managed table, through governed SQL: rows arrive already masked.
    vessels = sql.query(
        f"""
        SELECT vessel_id, name, risk_score, risk_level, last_latitude, last_longitude
        FROM {REGISTRY_TABLE}
        WHERE last_latitude IS NOT NULL AND last_longitude IS NOT NULL
        """
    )
    vessels
    return (vessels,)


@app.cell
def _(ontology):
    # Ontology objects, through the same governed edge, as a DataFrame.
    zones = ontology.read("MaritimeZone", ontology_slug=ONTOLOGY, limit=100)[
        ["areaId", "name", "category", "boundary"]
    ]
    zones
    return (zones,)


@app.cell
def _(vessels, zones):
    exposure = zone_exposure(vessels, zones, pd.Timestamp.now(tz="UTC"))
    exposure
    return (exposure,)


@app.cell
def _(exposure, mo, tables):
    # The first run creates the table; every later run replaces its rows, so a
    # vessel that left an area stops being reported in it.
    created = tables.create(
        EXPOSURE_TABLE, exposure, exist_ok=True, primary_key=EXPOSURE_KEY,
        description="MIK vessels inside a governed area (derived by exposure.py)",
    )
    if not created["created"]:
        tables.write(EXPOSURE_TABLE, exposure, mode="replace")
    mo.md(f"Wrote **{len(exposure)}** exposure rows to `{EXPOSURE_TABLE}`.")
    return


if __name__ == "__main__":
    app.run()