""".. title:: ASOS/METAR Backend Service

Return to `API Mainpage </api/#cgi>`_ or the
`Download Portal </request/download.phtml>`_.

Documentation on /cgi-bin/request/asos.py
-----------------------------------------

This cgi-bin script provides METAR/ASOS data.  It has a IP-based rate limit for
requests to prevent abuse.  A `503 Service Unavailable` response will be
returned if the server is under heavy load.

Changelog:

- 2026-04-21: Due to incessant requests against this service a 1 second
  per-IP throttle is now in place.  There is literally no reason to hit this
  service at a higher rate.  Just make an hourly request for all stations and
  be done with it in one shot.
- 2026-01-23: Added a practical limit of 1,000 station years worth of data
  to be allowed with the request prior to the database query for the actual
  data. You will get a HTTP 422 in this instance with a message to reduce size.
- 2024-04-01: Fix recently introduced bug with time sort order.
- 2024-03-29: This service had an intermediate bug whereby if the `tz` value
  was not provided, it would default to `America/Chicago` instead of `UTC`.
- 2024-03-29: Migrated to pydantic based request validation.  Will be
  monitoring for any issues.

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

Get the past 24 hours of air temperature and dew point for Des Moines and
Mason City, Iowa.

https://mesonet.agron.iastate.edu/cgi-bin/request/asos.py\
?data=tmpf&data=dwpf&station=DSM&station=MCW&hours=24

Return all observations for the UTC date on 20 May 2020 for ASOS stations in
Iowa and Illinois.

https://mesonet.agron.iastate.edu/cgi-bin/request/asos.py\
?data=all&network=IA_ASOS,IL_ASOS&sts=2020-05-20T00:00:00Z&\
ets=2020-05-21T00:00:00Z

"""

import re
from datetime import datetime, timedelta
from io import StringIO
from typing import Annotated
from zoneinfo import ZoneInfo

from pydantic import AwareDatetime, Field, field_validator
from pyiem.database import get_dbconn
from pyiem.network import Table as NetworkTable
from pyiem.util import utc
from pyiem.webutil import CGIModel, ListOrCSVType, error_log, iemapp

from iemweb.fields import TZ_FIELD

NETWORK_RE = re.compile(r"^[A-Z0-9_]{3,32}$")
STATION_RE = re.compile(r"^[A-Z0-9_]{3,4}$")
NULLS = {"M": "M", "null": "null", "empty": ""}
TRACE_OPTS = {"T": "T", "null": "null", "empty": "", "0.0001": "0.0001"}
AVAILABLE = [
    "tmpf",
    "dwpf",
    "relh",
    "drct",
    "sknt",
    "p01i",
    "alti",
    "mslp",
    "vsby",
    "gust",
    "skyc1",
    "skyc2",
    "skyc3",
    "skyc4",
    "skyl1",
    "skyl2",
    "skyl3",
    "skyl4",
    "wxcodes",
    "ice_accretion_1hr",
    "ice_accretion_3hr",
    "ice_accretion_6hr",
    "peak_wind_gust",
    "peak_wind_drct",
    "peak_wind_time",
    "feel",
    "metar",
    "snowdepth",
]
# inline is so much faster!
CONV_COLS = {
    "tmpc": "f2c(tmpf) as tmpc",
    "dwpc": "f2c(dwpf) as dwpc",
    "p01m": "p01i * 25.4 as p01m",
    "sped": "sknt * 1.15 as sped",
    "gust_mph": "gust * 1.15 as gust_mph",
    "peak_wind_gust_mph": "peak_wind_gust * 1.15 as peak_wind_gust_mph",
}


class MyModel(CGIModel):
    """Request Model."""

    data: Annotated[
        ListOrCSVType | None,
        Field(
            description=(
                "The data columns to return, defaults to all.  The available "
                "options are: tmpf, dwpf, relh, drct, sknt, p01i, alti, mslp, "
                "vsby, gust, skyc1, skyc2, skyc3, skyc4, skyl1, skyl2, skyl3, "
                "skyl4, wxcodes, ice_accretion_1hr, ice_accretion_3hr, "
                "ice_accretion_6hr, peak_wind_gust, peak_wind_drct, "
                "peak_wind_time, feel, metar, snowdepth"
            ),
        ),
    ] = None
    direct: Annotated[
        bool,
        Field(
            description=(
                "If set to 'yes', the data will be directly "
                "downloaded as a file."
            ),
        ),
    ] = False
    elev: Annotated[
        bool,
        Field(
            description=(
                "If set to 'yes', the elevation (m) of the station will be "
                "included in the output."
            ),
        ),
    ] = False
    ets: Annotated[
        AwareDatetime | None,
        Field(
            description=("The end time of the data request."),
        ),
    ] = None
    format: Annotated[
        str,
        Field(
            description=(
                "The format of the data, defaults to onlycomma. "
                "The available options are: onlycomma, tdf."
            ),
        ),
    ] = "onlycomma"
    hours: Annotated[
        int | None,
        Field(
            description=(
                "The number of hours of data to return prior to the current "
                "timestamp.  Can not be more than 24 if no stations are "
                "specified."
            ),
        ),
    ] = None
    latlon: Annotated[
        bool,
        Field(
            description=(
                "If set to 'yes', the latitude and longitude of the station "
                "will be included in the output."
            ),
        ),
    ] = False
    missing: Annotated[
        str,
        Field(
            description=(
                "How to represent missing values, defaults to M.  Other "
                "options are 'null' and 'empty'."
            ),
            pattern="^(M|null|empty)$",
        ),
    ] = "M"
    nometa: Annotated[
        bool,
        Field(
            description=(
                "If set to 'yes', the column headers will not be included "
                "in the output and the station and valid output files will "
                "not be included by default.  This is often used with the "
                "data=metar parameter to get just METARs."
            ),
        ),
    ] = False
    network: Annotated[
        ListOrCSVType | None,
        Field(
            description="The network to query, defaults to all networks.",
        ),
    ] = None
    report_type: ListOrCSVType = Field(
        default_factory=list,
        description=(
            "The report type to query, defaults to all.  The available "
            "options are: 1 (HFMETAR), 3 (Routine), 4 (Specials)."
        ),
    )
    station: Annotated[
        ListOrCSVType | None,
        Field(
            description=(
                "The station identifier to query, defaults to all stations "
                "and if you do not specify any stations, you can only "
                "request 24 hours of data."
            ),
        ),
    ] = None
    sts: Annotated[
        AwareDatetime | None,
        Field(
            description=("The start time of the data request."),
        ),
    ] = None
    trace: Annotated[
        str,
        Field(
            description=(
                "How to represent trace values, defaults to 0.0001.  Other "
                "options are 'null' and 'empty'."
            ),
            pattern="^(0.0001|null|empty|T)$",
        ),
    ] = "0.0001"
    tz: TZ_FIELD = "UTC"
    year1: Annotated[
        int | None,
        Field(
            description=(
                "The year of the start time, defaults to the time zone "
                "provided by `tzname`. If `sts` is not provided."
            ),
        ),
    ] = None
    month1: Annotated[
        int | None,
        Field(
            description=(
                "The month of the start time, defaults to the time zone "
                "provided by `tzname`. If `sts` is not provided."
            ),
        ),
    ] = None
    day1: Annotated[
        int | None,
        Field(
            description=(
                "The day of the start time, defaults to the time zone "
                "provided by `tzname`. If `sts` is not provided."
            ),
        ),
    ] = None
    hour1: Annotated[
        int,
        Field(
            description=(
                "The hour of the start time, defaults to the time zone "
                "provided by `tzname`. If `sts` is not provided."
            ),
        ),
    ] = 0
    minute1: Annotated[
        int,
        Field(
            description=(
                "The minute of the start time, defaults to the time zone "
                "provided by `tzname`. If `sts` is not provided."
            ),
        ),
    ] = 0
    year2: Annotated[
        int | None,
        Field(
            description=(
                "The year of the end time, defaults to the time zone "
                "provided by `tzname`. If `ets` is not provided."
            ),
        ),
    ] = None
    month2: Annotated[
        int | None,
        Field(
            description=(
                "The month of the end time, defaults to the time zone "
                "provided by `tzname`. If `ets` is not provided."
            ),
        ),
    ] = None
    day2: Annotated[
        int | None,
        Field(
            description=(
                "The day of the end time, defaults to the time zone "
                "provided by `tzname`. If `ets` is not provided."
            ),
        ),
    ] = None
    hour2: Annotated[
        int,
        Field(
            description=(
                "The hour of the end time, defaults to the time zone "
                "provided by `tzname`. If `ets` is not provided."
            ),
        ),
    ] = 0
    minute2: Annotated[
        int,
        Field(
            default=0,
            description=(
                "The minute of the end time, defaults to the time zone "
                "provided by `tzname`. If `ets` is not provided."
            ),
        ),
    ] = 0

    @field_validator("station")
    @classmethod
    def station_validator(cls, value):
        """Ensure the station is valid."""
        for station in value:
            if not STATION_RE.fullmatch(station):
                raise ValueError(f"Invalid station identifier: {station}")
        return value

    @field_validator("network")
    @classmethod
    def network_validator(cls, value):
        """Ensure the network is valid."""
        for network in value:
            if not NETWORK_RE.fullmatch(network):
                raise ValueError(f"Invalid network identifier: {network}")
        return value

    @field_validator("report_type", mode="after")
    @classmethod
    def ensure_ints(cls, value: list):
        """Ensure that each value is a valid integer."""
        _ = [int(v) for v in value]  # skipcq
        return value


def fmt_time(val, missing, _trace, tzinfo):
    """Format timestamp."""
    if val is None:
        return missing
    return (val.astimezone(tzinfo)).strftime("%Y-%m-%d %H:%M")


def fmt_trace(val, missing, trace, _tzinfo):
    """Format precip."""
    if val is None:
        return missing
    # careful with this comparison
    if 0 < val < 0.009999:
        return trace
    return f"{val:.2f}"


def fmt_simple(val, missing, _trace, _tzinfo):
    """Format simplely."""
    if val is None:
        return missing
    return dance(val).replace(",", " ").replace("\n", " ")


def fmt_wxcodes(val, missing, _trace, _tzinfo):
    """Format weather codes."""
    if val is None:
        return missing
    return " ".join(val)


def fmt_f2(val, missing, _trace, _tzinfo):
    """Simple 2 place formatter."""
    if val is None:
        return missing
    return f"{val:.2f}"


def fmt_f0(val, missing, _trace, _tzinfo):
    """Simple 0 place formatter."""
    if val is None:
        return missing
    return f"{val:.0f}"


def dance(val):
    """Force the val to ASCII."""
    return val.encode("ascii", "ignore").decode("ascii")


def overloaded(environ: dict):
    """Prevent automation from overwhelming the server"""

    with get_dbconn("asos") as pgconn:
        cursor = pgconn.cursor()
        cursor.execute("select one::float from system_loadavg")
        val = cursor.fetchone()[0]
    if val > 40:  # Cut back on logging
        error_log(environ, f"/cgi-bin/request/asos.py over cpu thres: {val}")
    return val > 30


def get_stations(form):
    """Figure out the requested station"""
    if not form["station"]:
        if form["network"] is not None:
            nt = NetworkTable(form["network"], only_online=False)
            return list(nt.sts.keys())
        return []
    stations = form["station"]
    if not stations:
        return []
    # allow folks to specify the ICAO codes for K*** sites
    for i, station in enumerate(stations):
        if len(station) == 4 and station[0] == "K":
            stations[i] = station[1:]
    return stations


def get_time_bounds(form, tzinfo):
    """Figure out the exact time bounds desired"""
    if form["hours"] is not None:
        ets = utc()
        sts = ets - timedelta(hours=int(form.get("hours")))
        return sts, ets
    # Here lie dragons, so tricky to get a proper timestamp
    try:

        def _get(num):
            return datetime(
                form[f"year{num}"],
                form[f"month{num}"],
                form[f"day{num}"],
                form[f"hour{num}"],
                form[f"minute{num}"],
            )

        if form["sts"] is None:
            form["sts"] = _get("1").replace(tzinfo=tzinfo)
        if form["ets"] is None:
            form["ets"] = _get("2").replace(tzinfo=tzinfo)
    except Exception:
        return None, None

    if form["sts"] == form["ets"]:
        form["ets"] += timedelta(days=1)
    if form["sts"] > form["ets"]:
        form["sts"], form["ets"] = form["ets"], form["sts"]
    return form["sts"], form["ets"]


def build_querycols(form):
    """Which database columns correspond to our query."""
    req = form["data"]
    if not req or "all" in req:
        return AVAILABLE
    res = []
    for col in req:
        if col == "presentwx":
            res.append("wxcodes")
        elif col in AVAILABLE:
            res.append(col)
        elif col in CONV_COLS:
            res.append(CONV_COLS[col])
    if not res:
        res.append("tmpf")
    return res


def toobusy(pgconn, environ, name):
    """Check internal logging..."""
    cursor = pgconn.cursor()
    cursor.execute(
        "SELECT pid from pg_stat_activity where query ~* %s",
        (name,),
    )
    over = cursor.rowcount > 6
    if over and cursor.rowcount > 9:  # cut back on logging
        error_log(environ, f"asos.py cursors {cursor.rowcount}: {name}")
    cursor.close()
    return over


# NOTE This app can't be cached via iemapp since it is a generator
@iemapp(
    help=__doc__,
    parse_times=False,
    schema=MyModel,
    ip_throttle_secs=1,
)
def application(environ, start_response):
    """Go main"""
    if environ["REQUEST_METHOD"] == "OPTIONS":
        start_response("400 Bad Request", [("Content-type", "text/plain")])
        yield b"Allow: GET,POST,OPTIONS"
        return
    if overloaded(environ):
        start_response(
            "503 Service Unavailable", [("Content-type", "text/plain")]
        )
        yield b"ERROR: server over capacity, please try later"
        return
    tzinfo = ZoneInfo(environ["tz"])
    pgconn = get_dbconn("asos")
    ip = environ.get("HTTP_X_FORWARDED_FOR", environ.get("REMOTE_ADDR"))
    if ip is not None:
        ip = ip.split(",")[0].strip()
        # Run the check at the class C level, since the requests could be
        # coming from a network of systems
        ip = ".".join(ip.split(".")[:2])
    cursor_name = f"mystream_{ip}"
    if toobusy(pgconn, environ, cursor_name):
        pgconn.close()
        start_response(
            "503 Service Unavailable", [("Content-type", "text/plain")]
        )
        yield b"ERROR: server over capacity, please try later"
        return
    acursor = pgconn.cursor(cursor_name, scrollable=False)
    acursor.itersize = 20_000

    report_types = [int(x) for x in environ["report_type"]]
    sts, ets = get_time_bounds(environ, tzinfo)
    if sts is None:
        pgconn.close()
        start_response(
            "422 Unprocessable Entity", [("Content-type", "text/plain")]
        )
        yield b"Invalid times provided."
        return
    stations = get_stations(environ)
    if not stations:
        # We are asking for all-data.  We limit the amount of data returned to
        # one day or less
        if (ets - sts) > timedelta(hours=24):
            pgconn.close()
            start_response("400 Bad Request", [("Content-type", "text/plain")])
            yield b"When requesting all-stations, must be less than 24 hours."
            return
    else:
        # Limit to 1_000 station years of data
        if (ets.year - sts.year) * len(stations) > 1_000:
            pgconn.close()
            start_response("400 Bad Request", [("Content-type", "text/plain")])
            yield (
                b"Requested too much data, reduce stations or time period. "
                b"Currently limited to 1,000 station_years worth of data."
            )
            return
    delim = environ["format"]
    headers = []
    if environ["direct"]:
        headers.append(("Content-type", "application/octet-stream"))
        suffix = "tsv" if delim in ["tdf", "onlytdf"] else "csv"
        if not stations or len(stations) > 1:
            fn = f"asos.{suffix}"
        else:
            fn = f"{stations[0]}.{suffix}"
        headers.append(("Content-Disposition", f"attachment; filename={fn}"))
    else:
        headers.append(("Content-type", "text/plain"))
    start_response("200 OK", headers)

    # How should null values be represented
    missing = NULLS[environ["missing"]]
    # How should trace values be represented
    trace = TRACE_OPTS[environ["trace"]]

    querycols = build_querycols(environ)

    if delim in ["tdf", "onlytdf"]:
        rD = "\t"
    else:
        rD = ","

    gisextra = environ["latlon"]
    elev_extra = environ["elev"]
    table = "alldata"
    metalimiter = ""
    colextra = "0 as lon, 0 as lat, 0 as elev, "
    if gisextra or elev_extra:
        colextra = "ST_X(geom) as lon, ST_Y(geom) as lat, elevation, "
        table = "alldata a JOIN stations t on (a.station = t.id)"
        metalimiter = "t.network ~* 'ASOS' and "

    rlimiter = ""
    # Munge legacy report_type=2 into 2,3,4 see akrherz/iem#104
    if 2 in report_types:
        report_types.extend([3, 4])
    if len(report_types) == 1:
        rlimiter = f" and report_type = {report_types[0]}"
    elif len(report_types) > 1:
        rlimiter = f" and report_type in {tuple(report_types)}"
    sqlcols = ",".join(querycols)
    sorder = "DESC" if environ["hours"] is not None else "ASC"
    if stations:
        acursor.execute(
            f"SELECT station, valid, {colextra} {sqlcols} from {table} "
            f"WHERE {metalimiter} valid >= %s and valid < %s and "
            f"station = ANY(%s) {rlimiter} ORDER by valid {sorder}",
            (sts, ets, stations),
        )
    else:
        acursor.execute(
            f"SELECT station, valid, {colextra} {sqlcols} from {table} "
            f"WHERE {metalimiter} valid >= %s and valid < %s {rlimiter} "
            f"ORDER by valid {sorder}",
            (sts, ets),
        )
    sio = StringIO()
    if delim not in ["onlytdf", "onlycomma"]:
        sio.write(f"#DEBUG: Format Typ    -> {delim}\n")
        sio.write(f"#DEBUG: Time Period   -> {sts} {ets}\n")
        sio.write(f"#DEBUG: Time Zone     -> {tzinfo}\n")
        sio.write(
            "#DEBUG: Data Contact   -> daryl herzmann "
            "akrherz@iastate.edu 515-294-5978\n"
        )
        sio.write(f"#DEBUG: Entries Found -> {acursor.rowcount}\n")
    nometa = environ["nometa"]
    if not nometa:
        sio.write(f"station{rD}valid{rD}")
        if gisextra:
            sio.write(f"lon{rD}lat{rD}")
        if elev_extra:
            sio.write(f"elevation{rD}")
        # hack to convert tmpf as tmpc to tmpc
        sio.write(
            f"{rD.join([c.rsplit(' as ', maxsplit=1)[-1] for c in querycols])}"
        )
        sio.write("\n")

    ff = {
        "wxcodes": fmt_wxcodes,
        "metar": fmt_simple,
        "skyc1": fmt_simple,
        "skyc2": fmt_simple,
        "skyc3": fmt_simple,
        "skyc4": fmt_simple,
        "p01i": fmt_trace,
        "p01i * 25.4 as p01m": fmt_trace,
        "ice_accretion_1hr": fmt_trace,
        "ice_accretion_3hr": fmt_trace,
        "ice_accretion_6hr": fmt_trace,
        "peak_wind_time": fmt_time,
        "snowdepth": fmt_f0,
    }
    # The default is the %.2f formatter
    formatters = [ff.get(col, fmt_f2) for col in querycols]

    for rownum, row in enumerate(acursor):
        if not nometa:
            sio.write(row[0] + rD)
            sio.write(
                (row[1].astimezone(tzinfo)).strftime("%Y-%m-%d %H:%M") + rD
            )
        if gisextra:
            sio.write(f"{row[2]:.4f}{rD}{row[3]:.4f}{rD}")
        if elev_extra:
            sio.write(f"{row[4]:.2f}{rD}")
        sio.write(
            rD.join(
                [
                    func(val, missing, trace, tzinfo)
                    for func, val in zip(formatters, row[5:], strict=False)
                ]
            )
            + "\n"
        )
        if rownum > 0 and rownum % 1000 == 0:
            yield sio.getvalue().encode("ascii", "ignore")
            sio = StringIO()
    acursor.close()
    pgconn.close()
    yield sio.getvalue().encode("ascii", "ignore")
