""".. title:: Service for Storm Based Warnings GeoJSON

Return to `API Services </api/#json>`_

Documentation for /geojson/sbw.geojson
--------------------------------------

This service supports a number of legacy systems and implements a number of
different calling methods.  The `API <https://mesonet.agron.iastate.edu
/api/1/docs#/vtec/service_vtec_sbw_interval__fmt__get>`_ service is recommended
for new applications.

Changelog
---------

- 2026-04-10: Corrected bug with service not providing SBWs issued at the
  requested timestamp when using the `ts` parameter.
- 2026-02-24: For IEM consistency, the `generation_time` root attribute was
  renamed `generated_at`.
- 2025-05-29: Implemented a `states` parameter to limit results to a list of
  given two-letter state identifiers.
- 2024-12-31: Five additional metadata fields are added with lifetime max
  values for windtag, hailtag, is_pds, is_emergency, and floodtag_damage. These
  fields are prefixed with `max_`.
- 2024-12-30: Parameters `sts` and `ets` can now be provided as ISO8601
  formatted strings.  The legacy format of %Y%m%d%H%M is still supported, but
  discouraged.
- 2024-11-19: Trimmed caching time from 60s to 15s. Added `ps` attribute that
  holds the VTEC Phenomena and Significance string.

Examples
--------

Provide all SBW polygons active at a given timestamp:

https://mesonet.agron.iastate.edu/geojson/sbw.geojson?ts=2024-05-26T20:00:00Z

Provide all SBW polygons on 26 May 2024 CDT.

https://mesonet.agron.iastate.edu/geojson/sbw.geojson\
?sts=2024-05-26T05:00:00Z&ets=2024-05-27T05:00:00Z

Provide all SBW polygons on 26 May 2025 CDF for Iowa.

https://mesonet.agron.iastate.edu/geojson/sbw.geojson\
?sts=2024-05-26T05:00:00Z&ets=2024-05-27T05:00:00Z&states=IA

"""

import json
from datetime import datetime, timedelta, timezone
from typing import Annotated

from pydantic import AwareDatetime, Field, field_validator, model_validator
from pyiem.database import get_sqlalchemy_conn, sql_helper
from pyiem.nws.vtec import get_ps_string
from pyiem.reference import ISO8601
from pyiem.util import utc
from pyiem.webutil import CGIModel, ListOrCSVType, iemapp

from iemweb.fields import CALLBACK_FIELD, STATE_LIST_FIELD_OPTIONAL
from iemweb.mlib import rectify_wfo, unrectify_wfo
from iemweb.util import get_ct, json_response_dict


class Schema(CGIModel):
    """See how we are called."""

    callback: CALLBACK_FIELD = None
    ets: Annotated[
        AwareDatetime | None,
        Field(
            description=("UTC end timestamp parameter"),
            ge=utc(1986, 1, 1),
        ),
    ] = None
    states: STATE_LIST_FIELD_OPTIONAL = None
    sts: Annotated[
        AwareDatetime | None,
        Field(
            description=("UTC start timestamp parameter"),
            ge=utc(1986, 1, 1),
        ),
    ] = None
    ts: AwareDatetime = Field(
        default=None,
        description=(
            "If sts and ets are not provided, this is the timestamp to "
            "request data for."
        ),
        ge=utc(1986, 1, 1),
    )
    wfo: str = Field(
        default=None,
        description="Optional WFO identifier",
        max_length=3,
        min_length=3,
    )
    wfos: ListOrCSVType = Field(
        default=None,
        description="Optional list of WFOs to limit the results to",
    )

    @field_validator("sts", "ets", mode="before")
    @classmethod
    def legacy_timestamps(cls, value, _info):
        """Allow the junky old way to work."""
        if isinstance(value, list):
            raise ValueError("Cannot be a list")
        fmt = "%Y%m%d%H%M"
        if value.find("T") > 0 and len(value) >= 16:
            fmt = "%Y-%m-%dT%H:%M"
            value = value[:16]
        return datetime.strptime(value, fmt).replace(tzinfo=timezone.utc)

    @model_validator(mode="after")
    def ensure_some_timestamp(self):
        """Ensure that we have some timestamp set."""
        if self.ts is None and self.sts is None and self.ets is None:
            self.ts = utc()
        if (self.sts is not None and self.ets is None) or (
            self.sts is None and self.ets is not None
        ):
            raise ValueError("Both sts and ets must be provided")
        return self


def df(val: datetime | None):
    """Format a datetime object"""
    return None if val is None else val.strftime(ISO8601)


def run(query: Schema):
    """Actually do the hard work of getting the current SBW in geojson"""
    wfos = query.wfos
    if wfos is not None and wfos[0] == "":
        wfos = None
    if wfos is None and query.wfo is not None:
        wfos = [query.wfo]
    if wfos:
        wfos = [unrectify_wfo(wfo) for wfo in wfos]

    params = {
        "wfos": wfos,
        "ts": query.ts,
        "sts": query.sts,
        "ets": query.ets,
    }

    res = json_response_dict(
        {
            "type": "FeatureCollection",
            "features": [],
        }
    )

    wfo_limiter = ""
    if wfos:
        wfo_limiter = " and s.wfo = ANY(:wfos) "
    time_limiter = " s.expire > :sts and s.issue < :ets "
    status_limiter = " and s.status = 'NEW' "
    # Mode 2, ts is provided, so we get polygons valid at this time
    if query.ts is not None:
        time_limiter = (
            " s.polygon_begin <= :ts and s.polygon_end > :ts and "
            " s.polygon_begin > :pastts "
        )
        status_limiter = ""
        params["pastts"] = query.ts - timedelta(days=14)  # arb
        params["sts"] = query.ts
        params["ets"] = query.ts
        res["valid_at"] = query.ts.strftime(ISO8601)

    ss1 = ""
    ss2 = ""
    if query.states is not None and query.states:
        ss1 = ", states t "
        ss2 = (
            " and ST_Intersects(s.geom, t.the_geom) and "
            " t.state_abbr = ANY(:states) "
        )
        params["states"] = query.states

    # NOTE: we dropped checking for products valid in the future (FL.W)
    # NOTE: we have an arb offset check for child table exclusion
    with get_sqlalchemy_conn("postgis") as conn:
        rs = conn.execute(
            sql_helper(
                """
    with tagmaxes as (
        select wfo, phenomena, vtec_year, eventid, significance,
        max(windtag) as max_windtag,
        max(hailtag) as max_hailtag,
        bool_or(is_pds) as max_is_pds,
        bool_or(is_emergency) as max_is_emergency,
        bool_or(floodtag_damage = 'CATASTROPHIC')
                as floodtag_damage_catastrophic,
        bool_or(floodtag_damage = 'CONSIDERABLE')
                as floodtag_damage_considerable
        from sbw s {ss1} WHERE s.expire > :sts and s.issue <= :ets
        {wfo_limiter} {ss2}
        GROUP by wfo, phenomena, vtec_year, eventid, significance
    )
            SELECT ST_asGeoJson(geom) as geojson, s.phenomena, s.eventid,
            s.wfo, s.significance,
            s.polygon_end at time zone 'UTC' as utc_polygon_end,
            s.polygon_begin at time zone 'UTC' as utc_polygon_begin,
            s.status,
            s.hvtec_nwsli, s.vtec_year, s.product_id,
            s.issue at time zone 'UTC' as utc_issue,
            s.expire at time zone 'UTC' as utc_expire, s.windtag, s.hailtag,
            s.tornadotag, s.damagetag, s.waterspouttag, s.is_emergency,
            s.is_pds, s.windthreat, s.hailthreat, s.product_signature,
            s.floodtag_damage, s.squalltag,
            t.max_windtag, t.max_hailtag, t.max_is_pds, t.max_is_emergency,
            case when t.floodtag_damage_catastrophic then 'CATASTROPHIC'
                    when t.floodtag_damage_considerable then 'CONSIDERABLE'
                    else null end as max_floodtag_damage
            from sbw s, tagmaxes t WHERE s.vtec_year = t.vtec_year and
            s.wfo = t.wfo and s.phenomena = t.phenomena and
            s.significance = t.significance and s.eventid = t.eventid and
            {time_limiter} {status_limiter} {wfo_limiter}
        """,
                wfo_limiter=wfo_limiter,
                time_limiter=time_limiter,
                status_limiter=status_limiter,
                ss1=ss1,
                ss2=ss2,
            ),
            params,
        )
        for row in rs.mappings():
            sid = (
                f"{row['wfo']}.{row['phenomena']}.{row['significance']}."
                f"{row['eventid']:04.0f}"
            )
            ets = row["utc_polygon_end"].strftime(ISO8601)
            sts = row["utc_polygon_begin"].strftime(ISO8601)
            sid += "." + sts
            href = (
                "https://mesonet.agron.iastate.edu/vtec/"
                f"?year={row['vtec_year']}&wfo={rectify_wfo(row['wfo'])}&"
                f"phenomena={row['phenomena']}"
                f"&significance={row['significance']}&"
                f"eventid={row['eventid']:04.0f}"
            )
            label = get_ps_string(row["phenomena"], row["significance"])
            link = f"<a href='{href}'>{label} {row['eventid']}</a> &nbsp; "
            res["features"].append(
                dict(
                    type="Feature",
                    id=sid,
                    properties=dict(
                        status=row["status"],
                        phenomena=row["phenomena"],
                        significance=row["significance"],
                        ps=label,
                        wfo=row["wfo"],
                        eventid=row["eventid"],
                        polygon_begin=sts,
                        polygon_end=ets,
                        expire=df(row["utc_expire"]),
                        hvtec_nwsli=row["hvtec_nwsli"],
                        year=row["vtec_year"],
                        product_id=row["product_id"],
                        issue=df(row["utc_issue"]),
                        expire_utc=df(row["utc_expire"]),
                        href=href,
                        link=link,
                        windtag=row["windtag"],
                        hailtag=row["hailtag"],
                        tornadotag=row["tornadotag"],
                        damagetag=row["damagetag"],
                        waterspouttag=row["waterspouttag"],
                        is_emergency=row["is_emergency"],
                        is_pds=row["is_pds"],
                        windthreat=row["windthreat"],
                        hailthreat=row["hailthreat"],
                        product_signature=row["product_signature"],
                        floodtag_damage=row["floodtag_damage"],
                        squalltag=row["squalltag"],
                        max_windtag=row["max_windtag"],
                        max_hailtag=row["max_hailtag"],
                        max_is_pds=row["max_is_pds"],
                        max_is_emergency=row["max_is_emergency"],
                        max_floodtag_damage=row["max_floodtag_damage"],
                    ),
                    geometry=json.loads(row["geojson"]),
                )
            )
    res["count"] = len(res["features"])
    return json.dumps(res, ensure_ascii=False)


def get_mcexpire(environ: dict) -> int:
    """Compute the cache expiration time."""
    return 15 if environ["ts"] is None else 600


def get_mckey(environ):
    """Compute the key."""
    wfos = environ["wfos"]
    if wfos is None and environ["wfo"] is not None:
        wfos = [environ["wfo"]]
    if wfos is None:
        wfos = []
    states = [] if environ["states"] is None else environ["states"]
    if environ["sts"] is not None:
        ts = f"{environ['sts']:%Y%m%d%H%M}_to_{environ['ets']:%Y%m%d%H%M}"
    else:
        ts = f"{environ['ts']:%Y%m%d%H%M}"
    return (
        f"/geojson/sbw.geojson|{ts}|{','.join(wfos)[:100]}|"
        f"{','.join(states)[:100]}"
    )


@iemapp(
    help=__doc__,
    schema=Schema,
    content_type=get_ct,
    memcachekey=get_mckey,
    memcacheexpire=get_mcexpire,
    parse_times=False,
)
def application(environ, start_response):
    """Main Workflow"""
    query: Schema = environ["_cgimodel_schema"]
    headers = [("Content-type", get_ct(environ))]
    res = run(query)
    start_response("200 OK", headers)
    return res.encode("utf-8")
