avwx.parsing.translate.taf

  1"""TAF data translation handlers."""
  2
  3# stdlib
  4from __future__ import annotations
  5
  6# module
  7import avwx.parsing.translate.base as _trans
  8from avwx.parsing import core
  9from avwx.parsing.translate import remarks
 10from avwx.static.taf import ICING_CONDITIONS, TURBULENCE_CONDITIONS
 11from avwx.structs import TafData, TafLineTrans, TafTrans, Units
 12
 13
 14def wind_shear(
 15    shear: str | None,
 16    unit_alt: str = "ft",
 17    unit_wind: str = "kt",
 18    *,
 19    spoken: bool = False,
 20) -> str:
 21    """Translate wind shear into a readable string.
 22
 23    Ex: Wind shear 2000ft from 140 at 30kt
 24    """
 25    if not shear or "WS" not in shear or "/" not in shear:
 26        return ""
 27    altitude, wind = shear[2:].rstrip(unit_wind.upper()).split("/")
 28    wdir = core.spoken_number(wind[:3], literal=True) if spoken else wind[:3]
 29    return f"Wind shear {int(altitude)*100}{unit_alt} from {wdir} at {wind[3:]}{unit_wind}"
 30
 31
 32def turb_ice(values: list[str], unit: str = "ft") -> str:
 33    """Translate the list of turbulence or icing into a readable sentence.
 34
 35    Ex: Occasional moderate turbulence in clouds from 3000ft to 14000ft
 36    """
 37    if not values:
 38        return ""
 39    # Determine turbulence or icing
 40    if values[0][0] == "5":
 41        conditions = TURBULENCE_CONDITIONS
 42    elif values[0][0] == "6":
 43        conditions = ICING_CONDITIONS
 44    else:
 45        return ""
 46    # Create list of split items (type, floor, height)
 47    split = [[item[1:2], item[2:5], item[5]] for item in values if len(item) == 6]
 48    # Combine items that cover a layer greater than 9000ft
 49    for i in reversed(range(len(split) - 1)):
 50        if (
 51            split[i][2] == "9"
 52            and split[i][0] == split[i + 1][0]
 53            and int(split[i + 1][1]) == (int(split[i][1]) + int(split[i][2]) * 10)
 54        ):
 55            split[i][2] = str(int(split[i][2]) + int(split[i + 1][2]))
 56            split.pop(i + 1)
 57    # Return joined, formatted string from split items
 58    return ", ".join(
 59        f"{conditions[item[0]]} from {int(item[1]) * 100}{unit} to {int(item[1]) * 100 + int(item[2]) * 1000}{unit}"
 60        for item in split
 61    )
 62
 63
 64def min_max_temp(temp: str | None, unit: str = "C") -> str:
 65    """Format the Min and Max temp elements into a readable string.
 66
 67    Ex: Maximum temperature of 23°C (73°F) at 18-15:00Z
 68    """
 69    if not temp or len(temp) < 7:
 70        return ""
 71    if temp[:2] == "TX":
 72        temp_type = "Maximum"
 73    elif temp[:2] == "TN":
 74        temp_type = "Minimum"
 75    else:
 76        return ""
 77    value, time = temp[2:].replace("M", "-").replace("Z", "").replace("//", "/").strip("/").split("/")
 78    if len(time) > 2:
 79        time = f"{time[:2]}-{time[2:]}"
 80    translation = _trans.temperature(core.make_number(value), unit)
 81    return f"{temp_type} temperature of {translation} at {time}:00Z"
 82
 83
 84def translate_taf(wxdata: TafData, units: Units) -> TafTrans:
 85    """Return translations for a TafData object."""
 86    forecast: list[TafLineTrans] = []
 87    for line in wxdata.forecast:
 88        shared = _trans.current_shared(line, units)
 89        # Remove false 'Sky Clear' if line type is 'BECMG'
 90        clouds = shared.clouds
 91        if line.type == "BECMG" and clouds == "Sky clear":
 92            clouds = ""
 93        struct = TafLineTrans(
 94            altimeter=shared.altimeter,
 95            clouds=clouds,
 96            wx_codes=shared.wx_codes,
 97            visibility=shared.visibility,
 98            wind=_trans.wind(
 99                line.wind_direction,
100                line.wind_speed,
101                line.wind_gust,
102                line.wind_variable_direction,
103                unit=units.wind_speed,
104            ),
105            wind_shear=wind_shear(line.wind_shear, units.altitude, units.wind_speed),
106            turbulence=turb_ice(line.turbulence, units.altitude),
107            icing=turb_ice(line.icing, units.altitude),
108        )
109        forecast.append(struct)
110    return TafTrans(
111        forecast=forecast,
112        max_temp=min_max_temp(wxdata.max_temp, units.temperature),
113        min_temp=min_max_temp(wxdata.min_temp, units.temperature),
114        remarks=remarks.translate(wxdata.remarks, wxdata.remarks_info),
115    )