""".. title:: NWS Watch/Warning/Advisory (WWA) Data Service

Return to `API Services </api/#cgi>`_ or
`Download User Interface </request/gis/watchwarn.phtml>`_.

Documentation for /cgi-bin/request/gis/watchwarn.py
---------------------------------------------------

This service emits shapefiles (with additional csv included),
or even Excel files.  This service is
rather blunt force and perhaps you should review the mountain of adhoc JSON/API
services found at
`IEM API Homepage </api/>`_ or at
`IEM API WebServices </api/1/docs>`_ .

Changelog
---------

- 2026-04-24: KML output support was added via `accept=kml`. The KML include
  "TimeSpan" information for KML clients that support it.
- 2026-03-16: Boolean fields were improved to allow for more truthy inputs
  than just `yes` and `no`.
- 2025-12-17: Improved the error message when provided time information is
  not sufficient.  Additionally, the hour values now default to 0 and minute
  values to 0 as well.
- 2025-07-12: Added ``fcster`` parameter to filter results by forecaster name
  (case-insensitive matching).  This field is also added to the output.
- 2025-05-09: Fixed issue with ``ETN`` DBF column allways being (null).
- 2024-10-28: Optimize the zipfile response by streaming the result.
- 2024-10-22: Fix and better document the ``at`` parameter for when
  ``timeopt=2``.
- 2024-07-03: Added a `accept=csv` option to allow for CSV output.
- 2024-06-26: Added `limitpds` parameter to limit the request to only include
  products that have a PDS (Particularly Dangerous Situation) tag or phrasing.
- 2024-05-14: To mitigate against large requests that overwhelm the server, a
  limit of one year's worth of data is now in place for requests that do not
  limit the request by either state, phenomena, nor wfo.
- 2024-05-09: Migrated to pydantic based CGI input validation.

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

Return all Areal Flood, Flash Flood, Severe Thunderstorm, and Tornado Watch
and Warnings for the state of Mississippi during 2024.  Note how the phenomena
and significance parameters are repeated so that each combination is present.

https://mesonet.agron.iastate.edu/cgi-bin/request/gis/watchwarn.py?\
accept=shapefile&sts=2024-01-01T00:00Z&ets=2025-01-01T00:00Z&\
location_group=states&states=MS&limitps=1&phenomena=FF,FA,SV,TO,FF,FA,SV,TO&\
significance=W,W,W,W,A,A,A,A

Return all Tornado Warnings for the Des Moines WFO in shapefile format during
2023.

https://mesonet.agron.iastate.edu/cgi-bin/request/gis/watchwarn.py\
?accept=shapefile&sts=2023-01-01T00:00Z&ets=2024-01-01T00:00Z&wfo[]=DMX\
&limitps=1&phenomena=TO&significance=W

Provide all Tornado Warnings valid at 2120 UTC on 21 May 2024

https://mesonet.agron.iastate.edu/cgi-bin/request/gis/watchwarn.py\
?accept=shapefile&at=2024-05-21T21:20Z&timeopt=2&limitps=1&phenomena=TO\
&significance=W

Same request, but return an excel file

https://mesonet.agron.iastate.edu/cgi-bin/request/gis/watchwarn.py\
?accept=excel&at=2024-05-21T21:20Z&timeopt=2&limitps=1&phenomena=TO\
&significance=W

Same request, but return csv

https://mesonet.agron.iastate.edu/cgi-bin/request/gis/watchwarn.py\
?accept=csv&at=2024-05-21T21:20Z&timeopt=2&limitps=1&phenomena=TO\
&significance=W

Same request, but return KML

https://mesonet.agron.iastate.edu/cgi-bin/request/gis/watchwarn.py\
?accept=kml&at=2024-05-21T21:20Z&timeopt=2&limitps=1&phenomena=TO\
&significance=W

Same request, but using the more verbose parameterization for the timestamp
and also filtering the result by text products signed by john

https://mesonet.agron.iastate.edu/cgi-bin/request/gis/watchwarn.py\
?accept=shapefile&year3=2024&month3=5&day3=21&hour3=21&minute3=20&timeopt=2\
&limitps=1&phenomena=TO&significance=W&fcster=john

"""

import tempfile
import zlib
from datetime import datetime, timedelta
from io import BytesIO
from stat import S_IFREG
from typing import Annotated
from xml.sax.saxutils import escape as xml_escape

import fiona
import pandas as pd
from pydantic import AwareDatetime, Field, model_validator
from pyiem.database import get_sqlalchemy_conn, sql_helper
from pyiem.exceptions import IncompleteWebRequest
from pyiem.nws.vtec import NWS_COLORS, get_ps_string
from pyiem.util import utc
from pyiem.webutil import CGIModel, ListOrCSVType, iemapp
from shapely.geometry import mapping
from shapely.wkb import loads
from sqlalchemy.engine import Connection
from stream_zip import ZIP_32, stream_zip

from iemweb.fields import VTEC_YEAR_FIELD_OPTIONAL

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


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

    accept: Annotated[
        str,
        Field(
            pattern="^(shapefile|excel|csv|kml)$",
            description=(
                "Materialize the data as the given format, either "
                "shapefile, excel, csv, or kml.  Default is shapefile."
            ),
        ),
    ] = "shapefile"
    at: AwareDatetime = Field(
        default=None,
        description=(
            "The timestamp to use when ``timeopt=2``, which the service "
            "provides events valid at the specified time."
        ),
    )
    addsvs: Annotated[
        bool,
        Field(
            description=(
                "Include polygons that were found within any followup "
                "statements after issuance."
            ),
        ),
    ] = False
    ets: Annotated[
        AwareDatetime | None,
        Field(
            description=(
                "The end timestamp in UTC. The format is ISO8601, e.g. "
                "2010-06-01T00:00Z."
            ),
            ge=utc(1986, 1, 1),
        ),
    ] = None
    limit0: Annotated[
        bool,
        Field(
            description="If true, only include Tornado, Severe Thunderstorm, "
            "Flash Flood, and Marine Warnings.",
        ),
    ] = False
    limit1: Annotated[
        bool,
        Field(
            description="If true, only include Storm Based Warnings.",
        ),
    ] = False
    limit2: Annotated[
        bool,
        Field(
            description="If true, only include Emergency Warnings.",
        ),
    ] = False
    limitpds: Annotated[
        bool,
        Field(
            description=(
                "If true, only include products that have a PDS "
                "(Particularly Dangerous Situation) tag or phrasing."
            ),
        ),
    ] = False
    limitps: Annotated[
        bool,
        Field(
            description="If true, only include the specified phenomena and "
            "significance.",
        ),
    ] = False
    location_group: str = Field(
        default="wfo",
        pattern="^(wfo|states)$",
        description="The location group to use, either wfo or states.",
    )
    phenomena: ListOrCSVType = Field(
        default=["TO"],
        description="The two character VTEC phenomena(s) to include. If you "
        "provide more than one value, the length must correspond and align "
        "with the ``significance`` parameter.",
    )
    simple: Annotated[
        bool,
        Field(
            description="If true, use a simplified geometry for the UGC "
            "counties/zones.",
        ),
    ] = True
    significance: ListOrCSVType = Field(
        default=["W"],
        description="The one character VTEC significance to include, if you "
        "provide more than one value, the length must correspond "
        "and align with the ``phenomena`` parameter.",
    )
    states: ListOrCSVType = Field(
        default=None, description="List of states to include data for."
    )
    sts: AwareDatetime = Field(
        default=None,
        description="The start timestamp in UTC. The format is ISO8601, e.g. "
        "2010-06-01T00:00Z.",
        ge=utc(1986, 1, 1),
    )
    timeopt: int = Field(
        default=1,
        description="The time option to use, either 1 or 2, default is 1, "
        "which uses the start and end timestamps to determine "
        "which events to include. Option 2 uses the ``at`` timestamp "
        "to determine which events to include.",
        ge=1,
        le=2,
    )
    wfo: ListOrCSVType = Field(
        default=None, description="List of WFOs to include data for."
    )
    wfos: ListOrCSVType = Field(
        default=None, description="Legacy parameter, update to use ``wfo``."
    )
    year1: VTEC_YEAR_FIELD_OPTIONAL = None
    year2: VTEC_YEAR_FIELD_OPTIONAL = None
    year3: VTEC_YEAR_FIELD_OPTIONAL = None
    month1: int = Field(
        default=None,
        description="The start timestamp components in UTC, if you specify a "
        "sts parameter, these are ignored.",
        ge=1,
        le=12,
    )
    month2: int = Field(
        default=None,
        description="The end timestamp components in UTC, if you specify a "
        "ets parameter, these are ignored.",
        ge=1,
        le=12,
    )
    month3: int = Field(
        default=None,
        description=(
            "The ``at`` timestamp components in UTC.  When timeopt is 2, "
            "this is used to find all events that were valid at this "
            "time."
        ),
        ge=1,
        le=12,
    )
    day1: int = Field(
        default=None,
        description="The start timestamp components in UTC, if you specify a "
        "sts parameter, these are ignored.",
        ge=1,
        le=31,
    )
    day2: int = Field(
        default=None,
        description="The end timestamp components in UTC, if you specify a "
        "ets parameter, these are ignored.",
        ge=1,
        le=31,
    )
    day3: int = Field(
        default=None,
        description=(
            "The ``at`` timestamp components in UTC.  When timeopt is 2, "
            "this is used to find all events that were valid at this "
            "time."
        ),
        ge=1,
        le=31,
    )
    hour1: int = Field(
        default=0,
        description="The start timestamp components in UTC, if you specify a "
        "sts parameter, these are ignored.",
        ge=0,
        le=23,
    )
    hour2: int = Field(
        default=0,
        description="The end timestamp components in UTC, if you specify a "
        "ets parameter, these are ignored.",
        ge=0,
        le=23,
    )
    hour3: int = Field(
        default=0,
        description=(
            "The ``at`` timestamp components in UTC.  When timeopt is 2, "
            "this is used to find all events that were valid at this "
            "time."
        ),
        ge=0,
        le=23,
    )
    minute1: int = Field(
        default=0,
        description="The start timestamp components in UTC, if you specify a "
        "sts parameter, these are ignored.",
        ge=0,
        le=59,
    )
    minute2: int = Field(
        default=0,
        description="The end timestamp components in UTC, if you specify a "
        "ets parameter, these are ignored.",
        ge=0,
        le=59,
    )
    minute3: int = Field(
        default=0,
        description=(
            "The ``at`` timestamp components in UTC.  When timeopt is 2, "
            "this is used to find all events that were valid at this "
            "time."
        ),
        ge=0,
        le=59,
    )
    fcster: str = Field(
        default=None,
        description="Optional forecaster filter to limit results to events "
        "where the forecaster field matches this value (case-insensitive).",
        max_length=100,
    )

    @model_validator(mode="after")
    def validate_timeopt(self):
        """Ensure that timeopt and times jive."""
        if self.timeopt == 1:
            if self.sts is None and (
                self.year1 is None or self.month1 is None or self.day1 is None
            ):
                raise ValueError("Either sts or {year,month,day}1 needs set")
            if self.ets is None and (
                self.year2 is None or self.month2 is None or self.day2 is None
            ):
                raise ValueError("Either ets or {year,month,day}2 needs set")
        elif self.timeopt == 2:
            if self.at is None and (
                self.year3 is None or self.month3 is None or self.day3 is None
            ):
                raise ValueError("Either at or {year,month,day}3 needs set")
        return self


def dfmt(txt):
    """Produce a prettier format for CSV."""
    if txt is None or len(txt) != 12:
        return ""
    return f"{txt[:4]}-{txt[4:6]}-{txt[6:8]} {txt[8:10]}:{txt[10:12]}"


def kml_stamp(txt: str | None) -> str:
    """Convert YYYYmmddHHMM into KML timestamp format."""
    if txt is None or len(txt) != 12:
        return ""
    return f"{txt[:4]}-{txt[4:6]}-{txt[6:8]}T{txt[8:10]}:{txt[10:12]}:00Z"


def kml_nice_date(txt: str | None) -> str:
    """Convert YYYYmmddHHMM into user-friendly UTC time text."""
    if txt is None or len(txt) != 12:
        return "Not available"
    return f"{txt[:4]}-{txt[4:6]}-{txt[6:8]} {txt[8:10]}:{txt[10:12]} UTC"


def get_kml_inline_style(
    phenomena: str, significance: str
) -> tuple[str, str, int]:
    """Return line color, polygon color, and line width for KML output.

    KML colors are ABGR (alpha, blue, green, red).
    Replace this default logic with your phenomena/significance color mapping.
    """
    hexrgb = NWS_COLORS.get(f"{phenomena}.{significance}", "#ffffff")
    hexbgr = hexrgb[5:7] + hexrgb[3:5] + hexrgb[1:3]
    return ("ff000000", f"7d{hexbgr}", 1)


def char3(wfos):
    """Make sure we don't have any 4 char IDs."""
    res = []
    for wfo in wfos:
        res.append(wfo[1:] if len(wfo) == 4 else wfo)  # noqa
    return res


def parse_wfo_location_group(environ, params):
    """Parse wfoLimiter"""
    limiter = ""
    wfos = environ["wfo"]
    if environ["wfos"]:
        wfos = environ["wfos"]
    if wfos is not None and "ALL" not in wfos:
        if len(wfos) == 1:
            wfo = wfos[0]
            wfo = wfo[1:] if len(wfo) == 4 else wfo
            limiter = " and w.wfo = :wfo "
            params["wfo"] = wfo
        else:
            limiter = " and w.wfo = ANY(:wfos) "
            params["wfos"] = char3(wfos)

    return limiter


def build(environ: dict) -> tuple[str, str, dict]:
    """Build the SQL statement."""
    params = {}
    sts = environ["sts"]
    ets = environ["ets"]
    table_extra = ""
    if environ["location_group"] == "states":
        if environ["states"]:
            params["states"] = [x[:2].upper() for x in environ["states"]]
            wfo_limiter = (
                " and ST_Intersects(s.the_geom, w.geom) "
                "and s.state_abbr = ANY(:states) "
            )
            wfo_limiter2 = " and substr(w.ugc, 1, 2) = ANY(:states) "
            table_extra = " , states s "
        else:
            raise IncompleteWebRequest("No state specified")
    else:  # wfo
        wfo_limiter = parse_wfo_location_group(environ, params)
        wfo_limiter2 = wfo_limiter

    if environ["timeopt"] == 1:
        # Keep size low
        if wfo_limiter == "" and (ets - sts) > timedelta(days=366):
            raise IncompleteWebRequest("Please shorten request to <1 year.")
        # Change to postgis db once we have the wfo list
        fn = f"wwa_{sts:%Y%m%d%H%M}_{ets:%Y%m%d%H%M}"
    else:
        if environ["at"] is None:
            year3 = int(environ.get("year3"))
            month3 = int(environ.get("month3"))
            day3 = int(environ.get("day3"))
            hour3 = int(environ.get("hour3"))
            minute3 = int(environ.get("minute3"))
            sts = utc(year3, month3, day3, hour3, minute3)
        else:
            sts = environ["at"]
        ets = sts
        environ["sts"] = sts
        environ["ets"] = ets
        fn = f"wwa_{sts:%Y%m%d%H%M}"

    limiter = ""
    if environ["limit0"]:
        limiter = (
            " and phenomena IN ('TO','SV','FF','MA') and significance = 'W' "
        )
    if environ["limitps"]:
        phenom = environ["phenomena"]
        sig = environ["significance"]
        parts = []
        for _i, (p, s) in enumerate(zip(phenom, sig, strict=False)):
            parts.append(f"(phenomena = :p{_i} and significance = :s{_i}) ")
            params[f"p{_i}"] = p[:2]
            params[f"s{_i}"] = s[:1]
        limiter = f" and ({' or '.join(parts)}) "

    sbwlimiter = " WHERE gtype = 'P' " if environ["limit1"] else ""

    elimiter = " and is_emergency " if environ["limit2"] else ""
    pdslimiter = " and is_pds " if environ["limitpds"] else ""

    # Forecaster filter
    fcsterlimiter_warnings = ""
    fcsterlimiter_sbw = ""
    if environ["fcster"]:
        fcsterlimiter_warnings = " and w.fcster ~* :fcster "
        fcsterlimiter_sbw = " and w.product_signature ~* :fcster "
        params["fcster"] = environ["fcster"]

    warnings_table = "warnings"
    sbw_table = "sbw"
    if sts.year == ets.year:
        warnings_table = f"warnings_{sts.year}"
        sbw_table = f"sbw_{sts.year}"

    geomcol = "geom"
    sbw_geosql = "ST_AsBinary(w.geom)"
    if environ["simple"]:
        geomcol = "simple_geom"
    cbw_geosql = f"ST_AsBinary(u.{geomcol})"
    if environ["accept"] == "kml":
        sbw_geosql = "ST_AsKML(w.geom)"
        cbw_geosql = f"ST_AsKML(u.{geomcol})"

    cols = (
        "wfo, utc_issue, utc_expire, utc_prodissue, utc_init_expire, "
        "phenomena, gtype, significance, eventid,  status, ugc, area2d, "
        "utc_updated, hvtec_nwsli, hvtec_severity, hvtec_cause, hvtec_record, "
        "is_emergency, utc_polygon_begin, utc_polygon_end, windtag, hailtag, "
        "tornadotag, damagetag, product_id, fcster, vtec_year "
    )
    if environ["accept"] not in ["excel", "csv"]:
        cols = f"geo, {cols}"

    timelimit = f"issue >= '{sts}' and issue < '{ets}'"
    if environ["timeopt"] == 2:
        timelimit = "issue <= :sts and issue > :sts30 and expire > :sts"
        params["sts"] = sts
        params["sts30"] = sts + timedelta(days=-30)
    else:
        if wfo_limiter == "" and limiter == "" and (ets - sts).days > 366:
            raise IncompleteWebRequest(
                "You must limit your request to a year or less."
            )
    sbwtimelimit = timelimit
    statuslimit = " status = 'NEW' "
    if environ["addsvs"]:
        statuslimit = " status != 'CAN' "
        sbwtimelimit = timelimit.replace(
            "issue",
            "coalesce(issue, polygon_begin)",
        )
    # NB: need distinct since state join could return multiple
    return (
        f"""
    WITH stormbased as (
     SELECT distinct {sbw_geosql} as geo, 'P'::text as gtype,
     significance, wfo, status, eventid, ''::text as ugc,
     phenomena,
     ST_area( ST_transform(w.geom,9311) ) / 1000000.0 as area2d,
     to_char(expire at time zone 'UTC', 'YYYYMMDDHH24MI') as utc_expire,
     to_char(issue at time zone 'UTC', 'YYYYMMDDHH24MI') as utc_issue,
     to_char(issue at time zone 'UTC', 'YYYYMMDDHH24MI') as utc_prodissue,
     to_char(polygon_begin at time zone 'UTC', 'YYYYMMDDHH24MI')
        as utc_polygon_begin,
     to_char(polygon_end at time zone 'UTC', 'YYYYMMDDHH24MI')
        as utc_polygon_end,
     to_char(init_expire at time zone 'UTC',
             'YYYYMMDDHH24MI') as utc_init_expire,
     to_char(updated at time zone 'UTC',
             'YYYYMMDDHH24MI') as utc_updated,
     hvtec_nwsli, hvtec_severity, hvtec_cause, hvtec_record, is_emergency,
     windtag, hailtag, tornadotag,
     coalesce(damagetag, floodtag_damage) as damagetag,
     product_id, product_signature as fcster, vtec_year
     from {sbw_table} w {table_extra}
     WHERE {statuslimit} and {sbwtimelimit}
     {wfo_limiter} {limiter} {elimiter} {pdslimiter} {fcsterlimiter_sbw}
    ),
    countybased as (
     SELECT {cbw_geosql} as geo, 'C'::text as gtype,
     significance,
     w.wfo, status, eventid, u.ugc, phenomena,
     u.area2163 as area2d,
     to_char(expire at time zone 'UTC', 'YYYYMMDDHH24MI') as utc_expire,
     to_char(issue at time zone 'UTC', 'YYYYMMDDHH24MI') as utc_issue,
     to_char(product_issue at time zone 'UTC',
             'YYYYMMDDHH24MI') as utc_prodissue,
     null as utc_polygon_begin,
     null as utc_polygon_end,
     to_char(init_expire at time zone 'UTC',
             'YYYYMMDDHH24MI') as utc_init_expire,
     to_char(updated at time zone 'UTC',
             'YYYYMMDDHH24MI') as utc_updated,
     hvtec_nwsli, hvtec_severity, hvtec_cause, hvtec_record, is_emergency,
     null::real as windtag, null::real as hailtag, null::varchar as tornadotag,
     null::varchar as damagetag,
     product_ids[1] as product_id, fcster, vtec_year
     from {warnings_table} w JOIN ugcs u on (u.gid = w.gid) WHERE
     {timelimit} {wfo_limiter2} {limiter} {elimiter} {pdslimiter}
     {fcsterlimiter_warnings}
     )
     SELECT {cols} from stormbased UNION ALL
     SELECT {cols} from countybased {sbwlimiter}
    """,
        fn,
        params,
    )


def do_kml(sql: str, params: dict, conn: Connection) -> iter[bytes]:
    """Materialize as KML."""
    res = conn.execute(sql_helper(sql), params)
    # Careful that the first byte is the XML
    yield b"""<?xml version="1.0" encoding="UTF-8"?>
<kml xmlns="http://www.opengis.net/kml/2.2">
<Document>
 <name>IEM NWS WWA Export</name>
 <open>1</open>
"""
    for row in res.mappings():
        if row["geo"] is None:  # Is this possible?
            continue
        uri = (
            f"https://mesonet.agron.iastate.edu/vtec/?year={row['vtec_year']}&"
            f"phenomena={row['phenomena']}&sig={row['significance']}&"
            f"eventid={row['eventid']}&wfo={row['wfo']}"
        )
        pname = (
            f"{row['wfo']}.{row['phenomena']}.{row['significance']}."
            f"{row['eventid']} "
            f"{get_ps_string(row['phenomena'], row['significance'])}"
        )
        pname_xml = xml_escape(pname)
        issued = kml_stamp(row["utc_issue"])
        expired = kml_stamp(row["utc_expire"])
        status = row["status"] or ""
        gtype = row["gtype"] or ""
        area2d = row["area2d"] if row["area2d"] is not None else 0.0
        utce = kml_nice_date(row["utc_expire"])
        line_color, poly_color, lw = get_kml_inline_style(
            row["phenomena"], row["significance"]
        )
        yield (
            f"""
<Placemark>
    <name>{pname_xml}</name>
    <TimeSpan><begin>{issued}</begin><end>{expired}</end></TimeSpan>
    <Style>
      <LineStyle><width>{lw}</width><color>{line_color}</color></LineStyle>
      <PolyStyle><color>{poly_color}</color></PolyStyle>
    </Style>
    <description>
        <![CDATA[
  <p>
    <font color="red"><i>Issued:</i></font> {kml_nice_date(row["utc_issue"])}
    <br /><font color="red"><i>Expires:</i></font> {utce}
    <br /><font color="red"><i>Status:</i></font> {status}
    <br /><font color="red"><i>Geometry Type:</i></font> {gtype}
    <br /><font color="red"><i>Area:</i></font> {area2d:.2f} km^2
    <br /><a href="{uri}">IEM VTEC Browser Link</a>
  </p>
        ]]>
    </description>
    {row["geo"]}
</Placemark>
"""
        ).encode()
    yield b"</Document></kml>"


def do_excel(sql, fmt, params):
    """Generate an Excel format response."""
    with get_sqlalchemy_conn("postgis") as conn:
        df = pd.read_sql(sql_helper(sql), conn, params=params, index_col=None)
    if fmt == "excel" and len(df.index) >= 1048576:
        raise IncompleteWebRequest("Result too large for Excel download")
    # Back-convert datetimes :/
    for col in (
        "utc_issue utc_expire utc_prodissue utc_updated utc_polygon_begin "
        "utc_polygon_end"
    ).split():
        df[col] = pd.to_datetime(
            df[col],
            errors="coerce",
            format="%Y%m%d%H%M",
        ).dt.strftime("%Y-%m-%d %H:%M")
    if fmt == "csv":
        return df.to_csv(index=False).encode("ascii")
    bio = BytesIO()
    with pd.ExcelWriter(bio, engine="xlsxwriter") as writer:
        df.to_excel(writer, sheet_name="VTEC WaWA", index=False)
    return bio.getvalue()


def process_results_yield_records(cursor, csv):
    """Do something with the results."""
    # Filenames are racy, so we need to have a temp folder
    csv.write(
        "WFO,ISSUED,EXPIRED,INIT_ISS,INIT_EXP,PHENOM,GTYPE,SIG,ETN,"
        "STATUS,NWS_UGC,AREA_KM2,UPDATED,HVTEC_NWSLI,HVTEC_SEVERITY,"
        "HVTEC_CAUSE,HVTEC_RECORD,IS_EMERGENCY,POLYBEGIN,POLYEND,"
        "WINDTAG,HAILTAG,TORNADOTAG,DAMAGETAG,PRODUCT_ID,FCSTER,VTEC_YEAR\n"
    )
    for row in cursor.mappings():
        if row["geo"] is None:  # Is this possible?
            continue
        mp = loads(row["geo"])
        csv.write(
            f"{row['wfo']},{dfmt(row['utc_issue'])},"
            f"{dfmt(row['utc_expire'])},"
            f"{dfmt(row['utc_prodissue'])},"
            f"{dfmt(row['utc_init_expire'])},"
            f"{row['phenomena']},{row['gtype']},"
            f"{row['significance']},{row['eventid']},"
            f"{row['status']},"
            f"{row['ugc']},{row['area2d']:.2f},"
            f"{dfmt(row['utc_updated'])},"
            f"{row['hvtec_nwsli']},{row['hvtec_severity']},"
            f"{row['hvtec_cause']},{row['hvtec_record']},"
            f"{row['is_emergency']},"
            f"{dfmt(row['utc_polygon_begin'])},"
            f"{dfmt(row['utc_polygon_end'])},{row['windtag']},"
            f"{row['hailtag']},{row['tornadotag']},"
            f"{row['damagetag']},{row['product_id']},"
            f"{row['fcster']},{row['vtec_year']}\n"
        )
        yield {
            "properties": {
                "WFO": row["wfo"],
                "ISSUED": row["utc_issue"],
                "EXPIRED": row["utc_expire"],
                "INIT_ISS": row["utc_prodissue"],
                "INIT_EXP": row["utc_init_expire"],
                "PHENOM": row["phenomena"],
                "GTYPE": row["gtype"],
                "SIG": row["significance"],
                "ETN": row["eventid"],
                "STATUS": row["status"],
                "NWS_UGC": row["ugc"],
                "AREA_KM2": row["area2d"],
                "UPDATED": row["utc_updated"],
                "HV_NWSLI": row["hvtec_nwsli"],
                "HV_SEV": row["hvtec_severity"],
                "HV_CAUSE": row["hvtec_cause"],
                "HV_REC": row["hvtec_record"],
                "EMERGENC": row["is_emergency"],
                "POLY_BEG": row["utc_polygon_begin"],
                "POLY_END": row["utc_polygon_end"],
                "WINDTAG": row["windtag"],
                "HAILTAG": row["hailtag"],
                "TORNTAG": row["tornadotag"],
                "DAMAGTAG": row["damagetag"],
                "PROD_ID": row["product_id"],
                "FCSTER": row["fcster"],
                "VTEC_YR": row["vtec_year"],
            },
            "geometry": mapping(mp),
        }


def local_files(names):
    """Return a generator of local files."""
    now = datetime.now()

    def contents(name: str):
        """Return a generator of file contents."""
        with open(name, "rb") as f:  # skipcq
            while chunk := f.read(65536):
                yield chunk

    return (
        (name.split("/")[-1], now, S_IFREG | 0o644, ZIP_32, contents(name))
        for name in names
    )


@iemapp(default_tz="UTC", help=__doc__, schema=Schema)
def application(environ, start_response):
    """Go Main Go"""
    try:
        if environ["timeopt"] != 2 and environ["sts"] is None:
            raise IncompleteWebRequest("Missing start time parameter")
        sql, fn, params = build(environ)
    except (IncompleteWebRequest, ValueError) as exp:
        start_response(
            "422 Unprocessable Content", [("Content-type", "text/plain")]
        )
        yield str(exp).encode("ascii")
        return
    if environ["accept"] == "kml":
        start_response(
            "200 OK",
            [
                ("Content-type", "application/vnd.google-earth.kml+xml"),
                ("Content-disposition", f"attachment; Filename={fn}.kml"),
            ],
        )
        with get_sqlalchemy_conn("postgis") as conn:
            yield from do_kml(sql, params, conn)
        return

    if environ["accept"] == "excel":
        headers = [
            ("Content-type", EXL),
            ("Content-disposition", f"attachment; Filename={fn}.xlsx"),
        ]
        start_response("200 OK", headers)
        yield do_excel(sql, environ["accept"], params)
        return
    if environ["accept"] == "csv":
        headers = [
            ("Content-type", "text/csv"),
            ("Content-disposition", f"attachment; Filename={fn}.csv"),
        ]
        start_response("200 OK", headers)
        yield do_excel(sql, environ["accept"], params)
        return

    with tempfile.TemporaryDirectory() as tmpdir:
        with (
            get_sqlalchemy_conn("postgis") as conn,
            fiona.open(
                f"{tmpdir}/{fn}.shp",
                "w",
                crs="EPSG:4326",
                driver="ESRI Shapefile",
                schema={
                    "geometry": "MultiPolygon",
                    "properties": {
                        "WFO": "str:3",
                        "ISSUED": "str:12",
                        "EXPIRED": "str:12",
                        "INIT_ISS": "str:12",
                        "INIT_EXP": "str:12",
                        "PHENOM": "str:2",
                        "GTYPE": "str:1",
                        "SIG": "str:1",
                        "ETN": "int:4",
                        "STATUS": "str:3",
                        "NWS_UGC": "str:6",
                        "AREA_KM2": "float",
                        "UPDATED": "str:12",
                        "HV_NWSLI": "str:5",
                        "HV_SEV": "str:1",
                        "HV_CAUSE": "str:2",
                        "HV_REC": "str:2",
                        "EMERGENC": "bool",
                        "POLY_BEG": "str:12",
                        "POLY_END": "str:12",
                        "WINDTAG": "float",
                        "HAILTAG": "float",
                        "TORNTAG": "str:16",
                        "DAMAGTAG": "str:16",
                        "PROD_ID": "str:36",
                        "FCSTER": "str:24",
                        "VTEC_YR": "int:4",
                    },
                },
            ) as output,
            open(f"{tmpdir}/{fn}.csv", "w", encoding="ascii") as csv,
            conn.execution_options(
                stream_results=True, max_row_buffer=10000
            ).execute(sql_helper(sql), params) as cursor,
        ):
            output.writerecords(process_results_yield_records(cursor, csv))

        headers = [
            ("Content-type", "application/octet-stream"),
            ("Content-Disposition", f"attachment; filename={fn}.zip"),
        ]
        start_response("200 OK", headers)
        zstream = stream_zip(
            local_files(
                (
                    f"{tmpdir}/{fn}.shp",
                    f"{tmpdir}/{fn}.shx",
                    f"{tmpdir}/{fn}.dbf",
                    f"{tmpdir}/{fn}.prj",
                    f"{tmpdir}/{fn}.cpg",
                    f"{tmpdir}/{fn}.csv",
                )
            ),
            get_compressobj=lambda: zlib.compressobj(
                wbits=-zlib.MAX_WBITS, level=1
            ),
        )
        yield from zstream
