""".. title:: Special Weather Statements (SPS) by Point

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

This service returns any Special Weather Statements (SPS) that included a
polygon threat area.

Changelog
---------

- 2026-02-26: Renamed top level metadata `generation_time` to `generated_at`
  for better consistency across IEM services.
- 2024-07-06: The `sdate` and `edate` parameters were rectified to be in the
  format of `YYYY-mm-dd` instead of `YYYY/mm/dd`.

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

Provide SPS metadata for those with a polygon that covered Newport, NC.

https://mesonet.agron.iastate.edu/json/sps_by_point.py?lat=34.77&lon=-76.88

Return the same, in Excel format this time and only those valid at 6z on
10 Aug 2024.

https://mesonet.agron.iastate.edu/json/sps_by_point.py\
?lat=34.77&lon=-76.88&valid=2024-08-10T06:00:00Z&fmt=xlsx

The same, but CSV this time

https://mesonet.agron.iastate.edu/json/sps_by_point.py\
?lat=34.77&lon=-76.88&valid=2024-08-10T06:00:00Z&fmt=csv

"""

import json
from datetime import date
from io import BytesIO, StringIO
from typing import Annotated

import numpy as np
import pandas as pd
from pydantic import AwareDatetime, Field
from pyiem.database import get_sqlalchemy_conn, sql_helper
from pyiem.reference import ISO8601
from pyiem.webutil import CGIModel, iemapp

from iemweb.fields import (
    CALLBACK_FIELD,
    LATITUDE_FIELD,
    LONGITUDE_FIELD,
)
from iemweb.util import json_response_dict

EXL = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"


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

    callback: CALLBACK_FIELD = None
    fmt: Annotated[
        str,
        Field(
            pattern="^(json|csv|xlsx)$",
            description="The format of the output, either json, csv, or xlsx",
        ),
    ] = "json"
    lat: LATITUDE_FIELD = 41.99
    lon: LONGITUDE_FIELD = -92.0
    sdate: Annotated[
        date,
        Field(
            description="Start date of search",
        ),
    ] = date(2002, 1, 1)
    edate: Annotated[
        date,
        Field(
            description="End date of search",
        ),
    ] = date(2099, 1, 1)
    valid: Annotated[
        AwareDatetime | None,
        Field(
            description="If provided, only include events valid at this time.",
        ),
    ] = None


def get_events(environ):
    """Get Events"""
    data = json_response_dict(
        {
            "data": [],
            "lon": environ["lon"],
            "lat": environ["lat"],
            "valid": environ["valid"],
        }
    )
    valid_limiter = ""
    params = {
        "lon": environ["lon"],
        "lat": environ["lat"],
        "sdate": environ["sdate"],
        "edate": environ["edate"],
        "valid": environ["valid"],
    }
    if environ["valid"] is not None:
        valid_limiter = " and issue <= :valid and expire > :valid "
        data["valid"] = environ["valid"].strftime(ISO8601)

    with get_sqlalchemy_conn("postgis") as conn:
        df = pd.read_sql(
            sql_helper(
                """
    select wfo, landspout, product_id, waterspout, max_hail_size,
    max_wind_gust,
    to_char(issue at time zone 'UTC', 'YYYY-MM-DDThh24:MIZ') as issue,
    to_char(expire at time zone 'UTC', 'YYYY-MM-DDThh24:MIZ') as expire
    from sps where
    ST_Contains(geom, ST_Point(:lon, :lat, 4326)) and
    issue > :sdate and expire < :edate
    {valid_limiter} ORDER by issue ASC
        """,
                valid_limiter=valid_limiter,
            ),
            conn,
            params=params,
        )
    if df.empty:
        return data, df
    df = df.replace({np.nan: None})
    df["uri"] = "/p.php?pid=" + df["product_id"]
    return data, df


def to_json(data, df):
    """Make JSON."""
    for _, row in df.iterrows():
        data["data"].append(
            {
                "wfo": row["wfo"],
                "landspout": row["landspout"],
                "waterspout": row["waterspout"],
                "uri": row["uri"],
                "issue": row["issue"],
                "expire": row["expire"],
                "max_hail_size": row["max_hail_size"],
                "max_wind_gust": row["max_wind_gust"],
            }
        )
    return data


@iemapp(
    help=__doc__,
    schema=Schema,
)
def application(environ, start_response):
    """Answer request."""
    fmt = environ.get("fmt", "json")
    data, df = get_events(environ)
    if fmt == "xlsx":
        fn = f"sps_{environ['lat']:.4f}N_{(0 - environ['lon']):.4f}W.xlsx"
        headers = [
            ("Content-type", EXL),
            ("Content-disposition", f"attachment; Filename={fn}"),
        ]
        bio = BytesIO()
        df.to_excel(bio, index=False)
        start_response("200 OK", headers)
        return [bio.getvalue()]
    if fmt == "csv":
        fn = f"sps_{environ['lat']:.4f}N_{(0 - environ['lon']):.4f}W.csv"
        headers = [
            ("Content-type", "application/octet-stream"),
            ("Content-disposition", f"attachment; Filename={fn}"),
        ]
        bio = StringIO()
        df.to_csv(bio, index=False)
        start_response("200 OK", headers)
        return [bio.getvalue().encode("utf-8")]
    res = to_json(data, df)
    payload = json.dumps(res).replace(" NaN", " null").encode("ascii")
    start_response("200 OK", [("Content-type", "application/json")])
    return payload
