avwx.parsing.remarks

  1"""Contains functions for handling and translating remarks."""
  2
  3# stdlib
  4from __future__ import annotations
  5
  6from contextlib import suppress
  7
  8# module
  9from avwx.parsing import core
 10from avwx.static.core import REMARKS_ELEMENTS, REMARKS_GROUPS, WX_TRANSLATIONS
 11from avwx.static.taf import PRESSURE_TENDENCIES
 12from avwx.structs import Code, FiveDigitCodes, Number, PressureTendency, RemarksData
 13
 14Codes = list[str]
 15
 16
 17def decimal_code(code: str, repr: str | None = None) -> Number | None:  # noqa: A002
 18    """Parse a 4-digit decimal temperature representation.
 19
 20    Ex: 1045 -> -4.5    0237 -> 23.7
 21    """
 22    if not code:
 23        return None
 24    number = f"{'-' if code[0] == '1' else ''}{int(code[1:3])}.{code[3]}"
 25    return core.make_number(number, repr or code)
 26
 27
 28def temp_dew_decimal(codes: Codes) -> tuple[Codes, Number | None, Number | None]:
 29    """Return the decimal temperature and dewpoint values."""
 30    temp, dew = None, None
 31    for i, code in reversed(list(enumerate(codes))):
 32        if len(code) in {5, 9} and code[0] == "T" and code[1:].isdigit():
 33            codes.pop(i)
 34            temp, dew = decimal_code(code[1:5]), decimal_code(code[5:])
 35            break
 36    return codes, temp, dew
 37
 38
 39def temp_minmax(codes: Codes) -> tuple[Codes, Number | None, Number | None]:
 40    """Return the 24-hour minimum and maximum temperatures."""
 41    maximum, minimum = None, None
 42    for i, code in enumerate(codes):
 43        if len(code) == 9 and code[0] == "4" and code.isdigit():
 44            maximum, minimum = decimal_code(code[1:5]), decimal_code(code[5:])
 45            codes.pop(i)
 46            break
 47    return codes, maximum, minimum
 48
 49
 50def precip_snow(codes: Codes) -> tuple[Codes, Number | None, Number | None]:
 51    """Return the hourly precipitation and snow depth."""
 52    precip, snow = None, None
 53    for i, code in reversed(list(enumerate(codes))):
 54        if len(code) != 5:
 55            continue
 56        # P0213
 57        if code[0] == "P" and code[1:].isdigit():
 58            precip = core.make_number(f"{code[1:3]}.{code[3:]}", code)
 59            codes.pop(i)
 60        # 4/012
 61        elif code[:2] == "4/" and code[2:].isdigit():
 62            snow = core.make_number(code[2:], code)
 63            codes.pop(i)
 64    return codes, precip, snow
 65
 66
 67def sea_level_pressure(codes: Codes) -> tuple[Codes, Number | None]:
 68    """Return the sea level pressure always in hPa."""
 69    sea = None
 70    for i, code in enumerate(codes):
 71        if len(code) == 6 and code.startswith("SLP") and code[-3:].isdigit():
 72            value = f"{'9' if int(code[-3]) > 4 else '10'}{code[-3:-1]}.{code[-1]}"
 73            sea = core.make_number(value, code)
 74            codes.pop(i)
 75            break
 76    return codes, sea
 77
 78
 79def parse_pressure(code: str) -> PressureTendency:
 80    """Parse a 5-digit pressure tendency."""
 81    return PressureTendency(
 82        repr=code,
 83        tendency=PRESSURE_TENDENCIES[code[1]],
 84        change=float(f"{code[2:4]}.{code[4]}"),
 85    )
 86
 87
 88def parse_precipitation(code: str) -> Number | None:
 89    """Parse a 5-digit precipitation amount."""
 90    return core.make_number(f"{code[1:3]}.{code[3:]}", code)
 91
 92
 93def five_digit_codes(codes: Codes) -> tuple[Codes, FiveDigitCodes]:
 94    """Return  a 5-digit min/max temperature code."""
 95    values = FiveDigitCodes()
 96    for i, code in reversed(list(enumerate(codes))):
 97        if len(code) == 5 and code.isdigit():
 98            key = int(code[0])
 99            if key == 1:
100                values.maximum_temperature_6 = decimal_code(code[1:], code)
101            elif key == 2:
102                values.minimum_temperature_6 = decimal_code(code[1:], code)
103            elif key == 5:
104                values.pressure_tendency = parse_pressure(code)
105            elif key == 6:
106                values.precip_36_hours = parse_precipitation(code)
107            elif key == 7:
108                values.precip_24_hours = parse_precipitation(code)
109            elif key == 9:
110                values.sunshine_minutes = core.make_number(code[2:], code)
111            else:
112                continue
113            codes.pop(i)
114    return codes, values
115
116
117def find_codes(rmk: str) -> tuple[Codes, list[Code]]:
118    """Find a remove known static codes from the starting remarks list."""
119    ret = []
120    for key, value in REMARKS_GROUPS.items():
121        if key in rmk:
122            ret.append(Code(key, value))
123            rmk.replace(key, "")
124    codes = [i for i in rmk.split() if i]
125    for i, code in reversed(list(enumerate(codes))):
126        with suppress(KeyError):
127            ret.append(Code(code, REMARKS_ELEMENTS[code]))
128            codes.pop(i)
129        # Weather began/ended
130        if len(code) == 5 and code[2] in ("B", "E") and code[3:].isdigit() and code[:2] in WX_TRANSLATIONS:
131            state = "began" if code[2] == "B" else "ended"
132            value = f"{WX_TRANSLATIONS[code[:2]]} {state} at :{code[3:]}"
133            ret.append(Code(code, value))
134            codes.pop(i)
135    ret.sort(key=lambda x: x.repr)
136    return codes, ret
137
138
139def parse(rmk: str) -> RemarksData | None:
140    """Find temperature and dewpoint decimal values from the remarks."""
141    if not rmk:
142        return None
143    codes, parsed_codes = find_codes(rmk)
144    codes, temperature, dewpoint = temp_dew_decimal(codes)
145    codes, max_temp, min_temp = temp_minmax(codes)
146    codes, precip, snow = precip_snow(codes)
147    codes, sea = sea_level_pressure(codes)
148    codes, fivedigits = five_digit_codes(codes)
149    return RemarksData(
150        codes=parsed_codes,
151        dewpoint_decimal=dewpoint,
152        temperature_decimal=temperature,
153        minimum_temperature_6=fivedigits.minimum_temperature_6,
154        minimum_temperature_24=min_temp,
155        maximum_temperature_6=fivedigits.maximum_temperature_6,
156        maximum_temperature_24=max_temp,
157        pressure_tendency=fivedigits.pressure_tendency,
158        precip_36_hours=fivedigits.precip_36_hours,
159        precip_24_hours=fivedigits.precip_24_hours,
160        sunshine_minutes=fivedigits.sunshine_minutes,
161        precip_hourly=precip,
162        snow_depth=snow,
163        sea_level_pressure=sea,
164    )