""".. title:: SPC Outlook Service

Return to `API Services </api/>`_.

Documentation for /json/spcoutlook.py
-------------------------------------

This service provides access to the Storm Prediction Center's Convective
Outlook products.  This service is a nasty bit of confusion for the three or
four different modes that it can be run in.  Those modes being:

- Provide the current day X outlook for a given point (`current=1` param).
- Provide all day X outlooks for a given point.
- Provide the day X outlook for a given point and given valid time
  (`time` param).
- Provide the most recent X number of outlooks for each threshold level for
  a given lat/lon point (`last` param).

There are examples below or feel free to yell at daryl for big a mess this is.

Changelog
---------

- 2026-04-17: Correct returned threshold value to not include CIGs.
- 2026-03-22: Add a new field to the JSON response called `conditional`,
  which will indicate the CIG1-3 or legacy SIGN for current queries.
- 2026-03-06: Sadly, the recently changed `time` parameter handling created
  additional service ambiguity.  There is now an additional boolean
  parameter called `current`, which when enabled causes the service to
  ignore the `time` parameter and instead return the current outlook for the
  given day.
- 2026-03-03: The ill conceived support for time being `now` or `current`
  was removed.
- 2026-02-26: Renamed top level metadata `generation_time` to `generated_at`
  for better consistency across IEM services.

Example Usage
-------------

Get all the day 1 outlooks for Pierre, South Dakota.  First in JSON, then
CSV, and finally Excel.

https://mesonet.agron.iastate.edu/json/spcoutlook.py\
?lat=44.368&lon=-100.336&day=1
https://mesonet.agron.iastate.edu/json/spcoutlook.py\
?lat=44.368&lon=-100.336&day=1&fmt=csv
https://mesonet.agron.iastate.edu/json/spcoutlook.py\
?lat=44.368&lon=-100.336&day=1&fmt=excel

Provide the day 1 outlook for Pierre, SD valid at 8 UTC on 3 Aug 2024

https://mesonet.agron.iastate.edu/json/spcoutlook.py\
?lat=44.368&lon=-100.336&day=1&time=2024-08-03T08:00Z

Provide the most recent outlook for each categorical threshold for day 2
for Washington, DC

https://mesonet.agron.iastate.edu/json/spcoutlook.py\
?lat=38.907&lon=-77.037&day=2&last=1

Return the last outlook of each threshold for day 4 for Washington, DC

https://mesonet.agron.iastate.edu/json/spcoutlook.py\
?lat=38.907&lon=-77.037&day=4&last=1&cat=ANY%20SEVERE

Provide the current day 2 outlook for Washington, DC

https://mesonet.agron.iastate.edu/json/spcoutlook.py\
?lat=38.907&lon=-77.037&day=2&current=1

Provide the day 2 outlook that was valid at 20 UTC on 6 March 2026.  This is
returning the day 2 forecast made on 5 March 2026.

https://mesonet.agron.iastate.edu/json/spcoutlook.py\
?lat=38.907&lon=-77.037&day=2&time=2026-03-06T20:00:00Z

"""

import json
from datetime import datetime, timedelta
from io import BytesIO
from typing import Annotated

import pandas as pd
from pydantic import AwareDatetime, Field
from pyiem.database import get_sqlalchemy_conn, sql_helper
from pyiem.nws.products.spcpts import THRESHOLD_ORDER
from pyiem.reference import ISO8601
from pyiem.util import utc
from pyiem.webutil import CGIModel, iemapp

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

VALID_CATS = [
    "ANY SEVERE",
    "CATEGORICAL",
    "CRITICAL FIRE WEATHER AREA",
    "DRY THUNDERSTORM CRITICAL FIRE WEATHER AREA",
    "FIRE WEATHER CATEGORICAL",
    "HAIL",
    "TORNADO",
    "WIND",
]


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

    callback: CALLBACK_FIELD = None
    current: Annotated[
        bool,
        Field(
            description=(
                "Return the currently valid outlook for the given day and "
                "category. The `time` parameter is ignored when this is "
                "enabled."
            )
        ),
    ] = False
    fmt: Annotated[
        str,
        Field(
            description=(
                "The format to return data in, either json, excel, or csv"
            ),
            pattern="^(json|excel|csv)$",
        ),
    ] = "json"
    lat: LATITUDE_FIELD = 42.0
    lon: LONGITUDE_FIELD = -95.0
    last: Annotated[
        int, Field(description="Limit to last N outlooks, 0 for all", ge=0)
    ] = 0
    day: OUTLOOK_DAY_FIELD = 1
    time: Annotated[
        AwareDatetime | None,
        Field(
            description=(
                "Optional specification for a valid timestamp to query "
                "outlooks for."
            )
        ),
    ] = None
    cat: Annotated[
        str,
        Field(
            description=(
                "A label for the category of the outlook type. Valid values "
                f"are: {', '.join(VALID_CATS)}"
            )
        ),
    ] = "CATEGORICAL"


def get_order(threshold):
    """Lookup a threshold and get its rank, higher is more extreme"""
    if threshold not in THRESHOLD_ORDER:
        return -1
    return THRESHOLD_ORDER.index(threshold)


def process_df(outlooks: pd.DataFrame) -> pd.DataFrame:
    """Condition the dataframe."""
    if outlooks.empty:
        return outlooks
    outlooks["threshold_rank"] = outlooks["threshold"].apply(get_order)
    return outlooks


def dotime(
    ts: datetime, lon: float, lat: float, day: int, cat: str
) -> tuple[pd.DataFrame, datetime]:
    """Query for Outlook based on some timestamp"""
    with get_sqlalchemy_conn("postgis") as conn:
        outlooks = pd.read_sql(
            sql_helper("""
        SELECT issue at time zone 'UTC' as i,
        expire at time zone 'UTC' as e,
        product_issue at time zone 'UTC' as v,
        threshold, category from spc_outlooks where
        product_issue = (
            select product_issue from spc_outlook where
            issue <= :ts and expire > :ts and day = :day
            and outlook_type = 'C' ORDER by product_issue DESC LIMIT 1)
        and ST_Contains(geom, ST_Point(:lon, :lat, 4326))
        and day = :day and outlook_type = 'C' and category = :cat
        """),
            conn,
            params={"lon": lon, "lat": lat, "day": day, "cat": cat, "ts": ts},
            index_col=None,
        )
    return process_df(outlooks), ts


def dowork(lon: float, lat: float, day: int, cat: str) -> pd.DataFrame:
    """Actually do stuff"""
    with get_sqlalchemy_conn("postgis") as conn:
        # Need to compute SIGN seperately
        outlooks = pd.read_sql(
            sql_helper("""
        WITH data as (
            SELECT issue at time zone 'UTC' as i,
            expire at time zone 'UTC' as e,
            product_issue at time zone 'UTC' as v,
            o.threshold, category, t.priority,
            row_number() OVER (PARTITION by expire
                ORDER by priority DESC NULLS last, issue ASC) as rank
            from spc_outlooks o, spc_outlook_thresholds t
            where o.threshold = t.threshold and
            ST_Contains(geom, ST_Point(:lon, :lat, 4326))
            and day = :day and outlook_type = 'C' and category = :cat
            and o.threshold not in ('TSTM', 'SIGN', 'CIG1', 'CIG2', 'CIG3')
            ORDER by issue DESC),
        agg as (
            select i, e, v, threshold, category from data where rank = 1),
        sign as (
            SELECT issue at time zone 'UTC' as i,
            expire at time zone 'UTC' as e,
            product_issue at time zone 'UTC' as v,
            threshold, category from spc_outlooks
            where ST_Contains(geom, ST_Point(:lon, :lat, 4326))
            and day = :day and outlook_type = 'C' and category = :cat
            and threshold in ('SIGN', 'CIG1', 'CIG2', 'CIG3')
            ORDER by expire DESC, issue ASC LIMIT 1)

        (SELECT i, e, v, threshold, category from agg
        ORDER by e DESC, threshold desc) UNION ALL
        (SELECT i, e, v, threshold, category from sign
        ORDER by e DESC, threshold desc)
        """),
            conn,
            params={"lon": lon, "lat": lat, "day": day, "cat": cat},
        )
    return outlooks


@iemapp(help=__doc__, schema=Schema)
def application(environ: dict, start_response: callable):
    """Answer request."""
    time = environ["time"]
    cat = environ["cat"].upper()
    if cat not in VALID_CATS:
        res = json_response_dict(
            {
                "error": f"Invalid category specified, valid categories are: "
                f"{', '.join(VALID_CATS)}"
            }
        )
        headers = [("Content-type", "application/json")]
        start_response("422 Unprocessable Entity", headers)
        return json.dumps(res)
    fmt = environ["fmt"]
    lon = environ["lon"]
    lat = environ["lat"]
    last = environ["last"]
    day = environ["day"]
    ts = None
    conditional = None
    if time is not None or environ["current"]:
        if environ["current"]:
            # Goose the time by the given day offset
            time = utc() + timedelta(hours=(day - 1) * 24)
        outlooks, ts = dotime(time, lon, lat, day, cat)
        if not outlooks.empty:
            cigs = ["CIG1", "CIG2", "CIG3", "SIGN"]
            # We need to look for SIGN, CIG1, CIG2, and CIG3
            for _cig in cigs:
                if _cig in outlooks["threshold"].values:
                    conditional = _cig
                    break
            # This will represent the max threshold, but...
            outlooks = outlooks[~outlooks["threshold"].isin(cigs)].iloc[[0]]
    else:
        outlooks = dowork(lon, lat, day, cat)

    if fmt == "json" and time is not None:
        res = json_response_dict(
            {
                "query_params": {
                    "time": ts.strftime(ISO8601),
                    "lon": lon,
                    "lat": lat,
                    "cat": cat,
                    "day": day,
                },
                "outlook": {},
            }
        )
        if not outlooks.empty:
            row0 = outlooks.iloc[0]
            res["outlook"] = {
                "threshold": row0["threshold"],
                "conditional": conditional,
                "utc_product_issue": pd.Timestamp(row0["v"]).strftime(ISO8601),
                "utc_issue": pd.Timestamp(row0["i"]).strftime(ISO8601),
                "utc_expire": pd.Timestamp(row0["e"]).strftime(ISO8601),
            }
        start_response("200 OK", [("Content-type", "application/json")])
        return json.dumps(res)
    if fmt == "json":
        running = {}
        res = json_response_dict({"outlooks": []})
        for _, row in outlooks.iterrows():
            if last > 0:
                running.setdefault(row["threshold"], 0)
                running[row["threshold"]] += 1
                if running[row["threshold"]] > last:
                    continue
            res["outlooks"].append(
                dict(
                    day=day,
                    utc_issue=row["i"].strftime(ISO8601),
                    utc_expire=row["e"].strftime(ISO8601),
                    utc_product_issue=row["v"].strftime(ISO8601),
                    threshold=row["threshold"],
                    category=row["category"],
                )
            )
        start_response("200 OK", [("Content-type", "application/json")])
        return json.dumps(res)

    outlooks = outlooks.rename(
        columns={
            "i": "utc_issue",
            "e": "utc_expire",
            "v": "utc_product_issue",
        },
    )
    if fmt == "excel":
        headers = [
            ("Content-type", "application/vnd.ms-excel"),
            ("Content-Disposition", "attachment; filename=outlooks.xls"),
        ]
        with BytesIO() as bio:
            outlooks.to_excel(bio, index=False)
            payload = bio.getvalue()
        start_response("200 OK", headers)
        return payload

    headers = [
        ("Content-type", "text/csv"),
        ("Content-Disposition", "attachment; filename=outlooks.csv"),
    ]
    payload = outlooks.to_csv(index=False)
    start_response("200 OK", headers)
    return payload
