avwx.parsing.speech

  1"""Contains functions for converting translations into a speech string.
  2Currently only supports METAR.
  3"""
  4
  5# stdlib
  6from __future__ import annotations
  7
  8import re
  9from typing import TYPE_CHECKING
 10
 11# module
 12import avwx.parsing.translate.base as translate_base
 13import avwx.parsing.translate.taf as translate_taf
 14from avwx.parsing import core
 15from avwx.static.core import SPOKEN_UNITS
 16
 17if TYPE_CHECKING:
 18    from avwx.structs import Code, MetarData, Number, TafData, TafLineData, Timestamp, Units
 19
 20
 21def ordinal(n: int) -> str | None:
 22    """Convert an int to it spoken ordinal representation."""
 23    if n < 0:
 24        return None
 25    return str(n) + "tsnrhtdd"[(n / 10 % 10 != 1) * (n % 10 < 4) * n % 10 :: 4]
 26
 27
 28def _format_plural_unit(value: str, unit: str) -> str:
 29    spoken = SPOKEN_UNITS.get(unit, unit)
 30    value = re.sub(r"(?<=\b1)" + unit, f" {spoken}", value)  # 1 knot
 31    return re.sub(r"(?<=\d)+" + unit, f" {spoken}s", value)  # 2 knots
 32
 33
 34def wind(
 35    direction: Number,
 36    speed: Number,
 37    gust: Number | None,
 38    vardir: list[Number] | None = None,
 39    unit: str = "kt",
 40) -> str:
 41    """Format wind details into a spoken word string."""
 42    val = translate_base.wind(direction, speed, gust, vardir, unit, cardinals=False, spoken=True)
 43    if val and unit in SPOKEN_UNITS:
 44        val = _format_plural_unit(val, unit)
 45    return "Winds " + (val or "unknown")
 46
 47
 48def temperature(header: str, temp: Number | None, unit: str = "C") -> str:
 49    """Format temperature details into a spoken word string."""
 50    if not temp or temp.value is None:
 51        return f"{header} unknown"
 52    unit = SPOKEN_UNITS.get(unit, unit)
 53    use_s = "" if temp.spoken in ("one", "minus one") else "s"
 54    return " ".join((header, temp.spoken, f"degree{use_s}", unit))
 55
 56
 57def visibility(vis: Number | None, unit: str = "m") -> str:
 58    """Format visibility details into a spoken word string."""
 59    if not vis:
 60        return "Visibility unknown"
 61    if vis.value is None or "/" in vis.repr:
 62        ret_vis = vis.spoken
 63    else:
 64        ret_vis = translate_base.visibility(vis, unit=unit)
 65        if unit == "m":
 66            unit = "km"
 67        ret_vis = ret_vis[: ret_vis.find(" (")].lower().replace(unit, "").strip()
 68        ret_vis = core.spoken_number(core.remove_leading_zeros(ret_vis))
 69    ret = f"Visibility {ret_vis}"
 70    if unit in SPOKEN_UNITS:
 71        if "/" in vis.repr and "half" not in ret:
 72            ret += " of a"
 73        ret += f" {SPOKEN_UNITS[unit]}"
 74        if ("one half" not in ret or " and " in ret) and "of a" not in ret:
 75            ret += "s"
 76    else:
 77        ret += unit
 78    return ret
 79
 80
 81def altimeter(alt: Number | None, unit: str = "inHg") -> str:
 82    """Format altimeter details into a spoken word string."""
 83    ret = "Altimeter "
 84    if not alt:
 85        ret += "unknown"
 86    elif unit == "inHg":
 87        ret += core.spoken_number(str(alt.value).ljust(5, "0"), literal=True)
 88    elif unit == "hPa":
 89        ret += core.spoken_number(str(alt.value).zfill(4), literal=True)
 90    return ret
 91
 92
 93def wx_codes(codes: list[Code]) -> str:
 94    """Format wx codes into a spoken word string."""
 95    ret = []
 96    for code in codes:
 97        item = code.value
 98        if item.startswith("Vicinity"):
 99            item = item.removeprefix("Vicinity ") + " in the Vicinity"
100        ret.append(item)
101    return ". ".join(ret)
102
103
104def type_and_times(
105    type: str | None,  # noqa: A002
106    start: Timestamp | None,
107    end: Timestamp | None,
108    probability: Number | None = None,
109) -> str:
110    """Format line type and times into the beginning of a spoken line string."""
111    if not type:
112        return ""
113    start_time = start.dt.hour if start and start.dt else "an unknown start time"
114    end_time = end.dt.hour if end and end.dt else "an unknown end time"
115    if type == "BECMG":
116        return f"At {start_time or 'midnight'} zulu becoming"
117    ret = f"From {start_time or 'midnight'} to {end_time or 'midnight'} zulu,"
118    if probability and probability.value:
119        ret += f" there's a {probability.value}% chance for"
120    if type == "INTER":
121        ret += " intermittent"
122    elif type == "TEMPO":
123        ret += " temporary"
124    return ret
125
126
127def wind_shear(shear: str, unit_alt: str = "ft", unit_wind: str = "kt") -> str:
128    """Format wind shear string into a spoken word string."""
129    value = translate_taf.wind_shear(shear, unit_alt, unit_wind, spoken=True)
130    if not value:
131        return "Wind shear unknown"
132    for unit in (unit_alt, unit_wind):
133        if unit in SPOKEN_UNITS:
134            value = _format_plural_unit(value, unit)
135    return value
136
137
138def metar(data: MetarData, units: Units) -> str:
139    """Convert MetarData into a string for text-to-speech."""
140    speech = []
141    if data.wind_direction and data.wind_speed:
142        speech.append(
143            wind(
144                data.wind_direction,
145                data.wind_speed,
146                data.wind_gust,
147                data.wind_variable_direction,
148                units.wind_speed,
149            )
150        )
151    if data.visibility:
152        speech.append(visibility(data.visibility, units.visibility))
153    speech.append(translate_base.clouds(data.clouds, units.altitude).replace(" - Reported AGL", ""))
154    if data.wx_codes:
155        speech.append(wx_codes(data.wx_codes))
156    if data.temperature:
157        speech.append(temperature("Temperature", data.temperature, units.temperature))
158    if data.dewpoint:
159        speech.append(temperature("Dew point", data.dewpoint, units.temperature))
160    if data.altimeter:
161        speech.append(altimeter(data.altimeter, units.altimeter))
162    return (". ".join([el for el in speech if el])).replace(",", ".")
163
164
165def taf_line(line: TafLineData, units: Units) -> str:
166    """Convert TafLineData into a string for text-to-speech."""
167    speech = []
168    start = type_and_times(line.type, line.start_time, line.end_time, line.probability)
169    if line.wind_direction and line.wind_speed:
170        speech.append(
171            wind(
172                line.wind_direction,
173                line.wind_speed,
174                line.wind_gust,
175                line.wind_variable_direction,
176                unit=units.wind_speed,
177            )
178        )
179    if line.wind_shear:
180        speech.append(wind_shear(line.wind_shear, units.altitude, units.wind_speed))
181    if line.visibility:
182        speech.append(visibility(line.visibility, units.visibility))
183    if line.altimeter:
184        speech.append(altimeter(line.altimeter, units.altimeter))
185    if line.wx_codes:
186        speech.append(wx_codes(line.wx_codes))
187    speech.append(translate_base.clouds(line.clouds, units.altitude).replace(" - Reported AGL", ""))
188    if line.turbulence:
189        speech.append(translate_taf.turb_ice(line.turbulence, units.altitude))
190    if line.icing:
191        speech.append(translate_taf.turb_ice(line.icing, units.altitude))
192    return f"{start} " + (". ".join([el for el in speech if el])).replace(",", ".")
193
194
195def taf(data: TafData, units: Units) -> str:
196    """Convert TafData into a string for text-to-speech."""
197    try:
198        month = data.start_time.dt.strftime(r"%B")  # type: ignore
199        day = ordinal(data.start_time.dt.day) or "Unknown"  # type: ignore
200        ret = f"Starting on {month} {day} - "
201    except AttributeError:
202        ret = ""
203    return ret + ". ".join([taf_line(line, units) for line in data.forecast])