""".. title:: VTEC Events by WFO

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

Changelog
---------

- 2024-08-14: Initial documentation and pydantic validation

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

Return all Des Moines Tornado Warnings for 2024

https://mesonet.agron.iastate.edu/json/vtec_events_bywfo.py?\
wfo=DMX&year=2024&phenomena=TO&significance=W

Same request, csv format

https://mesonet.agron.iastate.edu/json/vtec_events_bywfo.py?\
wfo=DMX&year=2024&phenomena=TO&significance=W&fmt=csv

Same request, xlsx format

https://mesonet.agron.iastate.edu/json/vtec_events_bywfo.py?\
wfo=DMX&year=2024&phenomena=TO&significance=W&fmt=xlsx

"""

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

import pandas as pd
from pydantic import AwareDatetime, Field
from pyiem.database import get_sqlalchemy_conn, sql_helper
from pyiem.util import utc
from pyiem.webutil import CGIModel, iemapp

from iemweb.fields import (
    CALLBACK_FIELD,
    VTEC_PH_FIELD_OPTIONAL,
    VTEC_SIG_FIELD_OPTIONAL,
    VTEC_YEAR_FIELD,
    WFO3_FIELD,
)
from iemweb.mlib import rectify_wfo
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(
            description=(
                "The format of the response, either json or csv or xlsx"
            ),
            pattern="^(json|csv|xlsx)$",
        ),
    ] = "json"
    start: Annotated[
        AwareDatetime | None,
        Field(
            description="Start of period of interest",
        ),
    ] = None
    end: Annotated[
        AwareDatetime | None,
        Field(
            description="End of period of interest",
        ),
    ] = None
    phenomena: VTEC_PH_FIELD_OPTIONAL = None
    significance: VTEC_SIG_FIELD_OPTIONAL = None
    wfo: WFO3_FIELD = "DMX"
    year: VTEC_YEAR_FIELD = utc().year


def make_url(row):
    """Build URL."""
    return (
        f"https://mesonet.agron.iastate.edu/vtec/?year={row['vtec_year']}&"
        f"wfo={rectify_wfo(row['wfo'])}&"
        f"phenomena={row['phenomena']}&"
        f"significance={row['significance']}&"
        f"eventid={row['eventid']:04.0f}"
    )


def get_df(wfo, start, end, phenomena, significance):
    """Answer the request!"""
    params = {
        "wfo": wfo,
        "start": start,
        "end": end,
    }
    plimiter = ""
    if phenomena is not None:
        params["phenomena"] = phenomena
        plimiter = " and phenomena = :phenomena "
    if significance is not None:
        params["significance"] = significance
        plimiter += " and significance = :significance "
    with get_sqlalchemy_conn("postgis") as conn:
        df = pd.read_sql(
            sql_helper(
                """
            SELECT
            to_char(product_issue at time zone 'UTC',
                'YYYY-MM-DDThh24:MI:SSZ') as product_issued,
            to_char(init_expire at time zone 'UTC',
                'YYYY-MM-DDThh24:MI:SSZ') as init_expired,
            to_char(issue at time zone 'UTC',
                'YYYY-MM-DDThh24:MI:SSZ') as issued,
            to_char(expire at time zone 'UTC',
                'YYYY-MM-DDThh24:MI:SSZ') as expired,
            to_char(updated at time zone 'UTC',
                'YYYY-MM-DDThh24:MI:SSZ') as updated,
            to_char(purge_time at time zone 'UTC',
                'YYYY-MM-DDThh24:MI:SSZ') as purge_time,
            eventid, phenomena, significance, hvtec_nwsli, wfo, ugc,
            product_ids[1] as product_id,
            vtec_year, product_ids[cardinality(product_ids)] as last_product_id
            from warnings WHERE wfo = :wfo and issue < :end
            and (expire > :start or init_expire > :start)
            {plimiter} ORDER by issue ASC
            """,
                plimiter=plimiter,
            ),
            conn,
            params=params,
        )
    if df.empty:
        return df
    # Construct a URL
    df["url"] = df.apply(make_url, axis=1)
    return df


def as_json(df: pd.DataFrame) -> str:
    """Materialize this df as JSON."""
    res = json_response_dict({"events": []})
    for _, row in df.iterrows():
        res["events"].append(
            {
                "url": row["url"],
                "product_issued": row["product_issued"],
                "issue": row["issued"],
                "expire": row["expired"],
                "init_expired": row["init_expired"],
                "purge_time": row["purge_time"],
                "updated": row["updated"],
                "eventid": row["eventid"],
                "phenomena": row["phenomena"],
                "hvtec_nwsli": row["hvtec_nwsli"],
                "significance": row["significance"],
                "last_product_id": row["last_product_id"],
                "product_id": row["product_id"],
                "wfo": row["wfo"],
                "ugc": row["ugc"],
            }
        )

    return json.dumps(res).replace(" NaN", " null")


@iemapp(help=__doc__, schema=Schema)
def application(environ, start_response):
    """Answer request."""
    query: Schema = environ["_cgimodel_schema"]
    year = query.year
    start = query.start
    end = query.end
    if start is None or end is None:
        start = utc(year)
        end = utc(year + 1)
    phenomena = query.phenomena
    significance = query.significance

    df = get_df(query.wfo, start, end, phenomena, significance)
    if query.fmt == "xlsx":
        fn = f"vtec_{query.wfo}_{start:%Y%m%d%H%M}_{end:%Y%m%d%H%M}.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 query.fmt == "csv":
        fn = f"vtec_{query.wfo}_{start:%Y%m%d%H%M}_{end:%Y%m%d%H%M}.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 = as_json(df)

    headers = [("Content-type", "application/json")]
    start_response("200 OK", headers)
    return res.encode("ascii")
