""".. title:: Local Storm Reports GeoJSON and More

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

This service does a number of different things with Local Storm Reports.

Changelog
---------

- 2025-05-21: Added `product_id` to output attributes.
- 2024-12-22: Added `hours` parameter to allow for a set time window prior
  to the `ets` parameter, which defaults to now.
- 2024-10-24: Added crude spatial bounds based parameters of ``east``,
  ``west``, ``north``, and ``south``.
- 2024-09-23: Added `qualifier` to output attributes, this represents the
  Measured, Estimated, or Unknonw qualifier for the report.

Example Requests
----------------

Provide all Local Storm Reports valid within the past 72 hours for Illinois.

https://mesonet.agron.iastate.edu/geojson/lsr.geojson?states=IL&hours=72

Provide all Local Storm Reports for Wisconsin on 13 July 2024. Note that a
UTC date period is specified that equates to the US Central date.

https://mesonet.agron.iastate.edu/geojson/lsr.geojson?states=WI&\
sts=2024-07-13T05:00Z&ets=2024-07-14T05:00Z

Provide the LSRs associated with Des Moines Tornado Warning 47 and include
any coincident warnings with each LSR report.

https://mesonet.agron.iastate.edu/geojson/lsr.geojson?phenomena=TO&\
significance=W&eventid=47&year=2024&wfo=DMX&inc_ap=1

Provide LSRs from NWS Des Moines and Davenport for 21 May 2024 UTC

https://mesonet.agron.iastate.edu/geojson/lsr.geojson?wfos=DMX,DVN&\
sts=2024-05-21T00:00Z&ets=2024-05-22T00:00Z

Provide all 2024 LSRs for a bounding box approximately covering Iowa.

https://mesonet.agron.iastate.edu/geojson/lsr.geojson?\
west=-96.5&east=-90.5&north=43.5&south=40.5&\
sts=2024-01-01T00:00Z&ets=2024-12-31T23:59Z

"""

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

import geopandas as gpd
from pydantic import Field, field_validator, model_validator
from pyiem.database import get_sqlalchemy_conn, sql_helper
from pyiem.nws.vtec import get_ps_string
from pyiem.util import utc
from pyiem.webutil import CGIModel, ListOrCSVType, iemapp

from iemweb.fields import (
    CALLBACK_FIELD,
    LATITUDE_FIELD_OPTIONAL,
    LONGITUDE_FIELD_OPTIONAL,
    VTEC_PH_FIELD_OPTIONAL,
    VTEC_SIG_FIELD,
    VTEC_YEAR_FIELD,
)
from iemweb.mlib import rectify_wfo, unrectify_wfo
from iemweb.util import get_ct


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

    callback: CALLBACK_FIELD = None
    inc_ap: Annotated[
        bool,
        Field(
            description="Include any associated warnings in the output.",
        ),
    ] = False
    eventid: Annotated[
        int,
        Field(
            description="If provided, use the given eventid to find LSRs for.",
        ),
    ] = 103
    phenomena: VTEC_PH_FIELD_OPTIONAL = None
    significance: VTEC_SIG_FIELD = "W"
    states: Annotated[
        ListOrCSVType | None,
        Field(
            description="If provided, use the given states to find LSRs for.",
        ),
    ] = None
    ets: Annotated[
        datetime | None,
        Field(
            description="End timestamp.",
        ),
    ] = None
    sts: Annotated[
        datetime | None,
        Field(
            description="Start Timestamp.",
        ),
    ] = None
    hours: Annotated[
        int | None,
        Field(
            description=(
                "If provided, number of hours prior to `ets` "
                "(default now) to provide LSRs for."
            ),
        ),
    ] = None
    wfo: Annotated[
        str | None,
        Field(
            description="If provided, use the given WFO to find LSRs for.",
            max_length=4,
        ),
    ] = None
    wfos: Annotated[
        ListOrCSVType | None,
        Field(
            description="If provided, use the given WFOs to find LSRs for.",
        ),
    ] = None
    year: VTEC_YEAR_FIELD = 2006
    east: LONGITUDE_FIELD_OPTIONAL = None
    west: LONGITUDE_FIELD_OPTIONAL = None
    north: LATITUDE_FIELD_OPTIONAL = None
    south: LATITUDE_FIELD_OPTIONAL = None
    fmt: Annotated[
        str,
        Field(
            description="Output format (fixed to geojson).",
            pattern=r"^geojson$",
        ),
    ] = "geojson"

    @model_validator(mode="after")
    def validate_spatial_bounds(self):
        """Ensure we have a valid spatial bounds."""
        # Ensure that if we have one field set, we have them all
        vals = [self.east, self.west, self.north, self.south]
        if any(x is not None for x in vals):
            if not all(x is not None for x in vals):
                raise ValueError("Incomplete spatial bounds provided")
            if self.east <= self.west:
                raise ValueError("East is less than West")
            if self.north <= self.south:
                raise ValueError("North is less than South")
        return self

    @field_validator("ets", "sts", mode="before")
    @classmethod
    def parse_valid(cls, value, _info):
        """Ensure we have a valid time."""
        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)


def do_vtec(environ: dict) -> gpd.GeoDataFrame:
    """Actually do the hard work of getting the current SBW in geojson"""
    with get_sqlalchemy_conn("postgis") as conn:
        lsrdf = gpd.read_postgis(
            sql_helper(
                """
            SELECT l.wfo, l.type,
            l.magnitude as magf, l,county, l.typetext,
            l.state, l.remark, l.city, l.source, l.unit, l.geom,
            to_char(valid at time zone 'UTC', 'YYYY-MM-DDThh24:MI:SSZ')
                as valid,
            ST_x(l.geom) as lon, ST_y(l.geom) as lat, qualifier
            from sbw w, lsrs l
            WHERE w.vtec_year = :year and w.wfo = :wfo and
            w.phenomena = :phenomena and
            w.eventid = :eventid and w.significance = :significance
            and w.geom && l.geom and l.valid BETWEEN w.issue and w.expire
            and w.status = 'NEW'
                """
            ),
            conn,
            params={
                "year": environ["year"],
                "wfo": environ["wfo"],
                "phenomena": environ["phenomena"],
                "eventid": environ["eventid"],
                "significance": environ["significance"],
            },
            geom_col="geom",
        )  # type: ignore
    return lsrdf


def do_states(environ: dict) -> gpd.GeoDataFrame:
    """Actually do the hard work of getting the current SBW in geojson"""
    with get_sqlalchemy_conn("postgis") as conn:
        lsrdf = gpd.read_postgis(
            sql_helper(
                """
            SELECT distinct l.wfo, l.type,
            l.magnitude as magf, l,county, l.typetext,
            l.state, l.remark, l.city, l.source, l.unit, l.geom,
            to_char(valid at time zone 'UTC', 'YYYY-MM-DDThh24:MI:SSZ')
                as valid,
            ST_x(l.geom) as lon, ST_y(l.geom) as lat, qualifier,
            coalesce(l.product_id_summary, l.product_id) as product_id
            FROM lsrs l, states s WHERE
            valid BETWEEN :sts and :ets and state_abbr = ANY(:states)
            and ST_Intersects(l.geom, s.the_geom)
            LIMIT 10000
                """
            ),
            conn,
            params={
                "states": environ["states"],
                "ets": environ["ets"],
                "sts": environ["sts"],
            },
            geom_col="geom",
        )  # type: ignore
    return lsrdf


def do_default(environ: dict) -> gpd.GeoDataFrame:
    """Actually do the hard work of getting the current SBW in geojson"""
    wfo_limiter = ""
    if environ["wfos"]:
        wfo_limiter = " and wfo = ANY(:wfos) "
    if environ["west"] is not None:
        wfo_limiter = (
            " and ST_Contains(ST_MakeEnvelope(:west, :south, :east, :north, "
            "4326), geom) "
        )
    with get_sqlalchemy_conn("postgis") as conn:
        lsrdf = gpd.read_postgis(
            sql_helper(
                """
            SELECT distinct wfo, type, magnitude as magf,
            county, typetext, state, remark, city, source, unit,
            to_char(valid at time zone 'UTC', 'YYYY-MM-DDThh24:MI:SSZ')
                as valid,
            ST_x(geom) as lon, ST_y(geom) as lat, qualifier, geom,
            coalesce(product_id_summary, product_id) as product_id
            FROM lsrs WHERE
            valid BETWEEN :sts and :ets {wfo_limiter}
            LIMIT 10000
                """,
                wfo_limiter=wfo_limiter,
            ),
            conn,
            params=environ,
            geom_col="geom",
        )  # type: ignore
    return lsrdf


def add_warnings(lsrdf: gpd.GeoDataFrame) -> None:
    """Add any associated warnings to the LSRs."""
    lsrdf["prodlinks"] = ""
    with get_sqlalchemy_conn("postgis") as conn:
        for idx, row in lsrdf.iterrows():
            res = conn.execute(
                sql_helper("""
                SELECT distinct phenomena, significance, eventid, vtec_year
                from sbw
                WHERE vtec_year = :year and wfo = :wfo and issue <= :valid
                and issue > :valid  - '7 days'::interval and expire > :valid
                and ST_POINT(:lon, :lat, 4326) && geom
                and ST_contains(geom, ST_POINT(:lon, :lat, 4326))
                """),
                {
                    "year": row["valid"][:4],
                    "wfo": row["wfo"],
                    "valid": row["valid"],
                    "lon": row["lon"],
                    "lat": row["lat"],
                },
            )
            products = ""
            for row2 in res.mappings():
                url = (
                    f"/vtec/?year={row2['vtec_year']}"
                    f"&wfo={rectify_wfo(row['wfo'])}"
                    f"&phenomena={row2['phenomena']}"
                    f"&significance={row2['significance']}"
                    f"&eventid={row2['eventid']:04.0f}"
                )
                products += (
                    f'<a href="{url}">'
                    f"{get_ps_string(row2['phenomena'], row2['significance'])}"
                    f" {row2['eventid']}</a><br />"
                )
            if products == "":
                products = "No Warnings Found"
            lsrdf.at[idx, "prodlinks"] = products


def get_mckey(environ: dict) -> str | None:
    """Figure out the key for this request."""
    if environ["hours"] is None or environ["ets"] is not None:
        return None
    return (
        f"geojson_lsr_{environ['hours']}_{environ['states']}_{environ['wfo']}_"
        f"{environ['inc_ap']}"
    ).replace(" ", "_")


@iemapp(
    content_type=get_ct,
    help=__doc__,
    schema=Schema,
    parse_times=False,
    memcacheexpire=60,
    memcachekey=get_mckey,
)
def application(environ, start_response):
    """Do Something"""
    # Quirk unhandled properly yet
    if environ["wfos"]:
        environ["wfos"] = list(
            filter(lambda x: 2 < len(x) < 5, environ["wfos"])
        )
        environ["wfos"] = [unrectify_wfo(x) for x in environ["wfos"]]
    if environ["states"]:
        environ["states"] = list(
            filter(lambda x: len(x) == 2, environ["states"])
        )
    if environ["hours"] is not None:
        if environ["ets"] is None:
            environ["ets"] = utc()
        environ["sts"] = environ["ets"] - timedelta(hours=environ["hours"])
    # Go Main Go
    headers = [("Content-type", get_ct(environ))]
    if environ["phenomena"] is not None:
        lsrdf = do_vtec(environ)
    elif environ["states"]:
        lsrdf = do_states(environ)
    else:
        lsrdf = do_default(environ)
    if environ["inc_ap"]:
        add_warnings(lsrdf)
    # legacy stuff
    lsrdf["st"] = lsrdf["state"]
    lsrdf = lsrdf.rename(
        columns={
            "iso_valid": "valid",
        }
    )
    # Keep magnitude as a string
    lsrdf["magnitude"] = ""
    lsrdf.loc[lsrdf["magf"].isna(), "magnitude"] = ""
    lsrdf.loc[lsrdf["magf"].notna(), "magnitude"] = lsrdf["magf"].astype(str)
    # Trim any strings ending in .0
    lsrdf["magnitude"] = lsrdf["magnitude"].str.replace(
        r"\.0$", "", regex=True
    )
    payload = lsrdf.to_json()
    start_response("200 OK", headers)
    return payload.encode("utf-8")
