Implement TAF speech and tests

avwx/__init__.py

@@ -145,3 +145,12 @@ def summary(self) -> str:
         if not self.translations:
             self.update()
         return [summary.taf(trans) for trans in self.translations.forecast]
+
+    @property
+    def speech(self) -> str:
+        """
+        Report summary designed to be read by a text-to-speech program
+        """
+        if not self.data:
+            self.update()
+        return speech.taf(self.data, self.units)

avwx/core.py

@@ -93,7 +93,7 @@ def spoken_number(num: str) -> str:
     return ' and '.join(ret)
 
 
-def make_number(num: str, repr: str = None) -> Number:
+def make_number(num: str, repr: str = None, speak: str = None) -> Number:
     """
     Returns a Number or Fraction dataclass for a number string
     """
@@ -111,7 +111,7 @@ def make_number(num: str, repr: str = None) -> Number:
     # Create Number
     val = num.replace('M', '-')
     val = float(val) if '.' in num else int(val)
-    return Number(repr or num, val, spoken_number(str(val)))
+    return Number(repr or num, val, spoken_number(speak or str(val)))
 
 
 def find_first_in_list(txt: str, str_list: [str]) -> int:
@@ -486,9 +486,9 @@ def get_wind(wxdata: [str], units: Units) -> ([str], Number, Number, Number, [Nu
     #Variable Wind Direction
     if wxdata and len(wxdata[0]) == 7 and wxdata[0][:3].isdigit() \
         and wxdata[0][3] == 'V' and wxdata[0][4:].isdigit():
-        variable = [make_number(i) for i in wxdata.pop(0).split('V')]
+        variable = [make_number(i, speak=i) for i in wxdata.pop(0).split('V')]
     # Convert to Number
-    direction = make_number(direction)
+    direction = make_number(direction, speak=direction)
     speed = make_number(speed)
     gust = make_number(gust)
     return wxdata, direction, speed, gust, variable
@@ -609,9 +609,12 @@ def find_missing_taf_times(lines: [dict], start: Timestamp, end: Timestamp) -> [
             target += '_time'
             if not line[target]:
                 line[target] = _get_next_time(lines[i::direc][1:], other+'_time')
-    #Special case for final forcast
+    # Special case for final forcast
     if last_fm_line:
         lines[last_fm_line]['end_time'] = end
+    # Reset original end time if still empty
+    if lines and not lines[0]['end_time']:
+        lines[0]['end_time'] = end
     return lines
 
 

avwx/speech.py

@@ -3,12 +3,13 @@
 Currently only supports METAR
 """
 
-# stdlib
-from copy import deepcopy
 # module
 from avwx import core, translate
 from avwx.static import SPOKEN_UNITS, NUMBER_REPL, FRACTIONS
-from avwx.structs import Cloud, MetarData, Number, Units
+from avwx.structs import Cloud, MetarData, Number, TafData, TafLineData, Timestamp, Units
+
+
+ordinal = lambda n: "%d%s" % (n,"tsnrhtdd"[(n/10%10!=1)*(n%10<4)*n%10::4])
 
 
 def wind(direction: Number,
@@ -82,36 +83,97 @@ def other(wxcodes: [str]) -> str:
     Format wx codes into a spoken word string
     """
     ret = []
-    for item in wxcodes:
-        item = translate.wxcode(item)
+    for code in wxcodes:
+        item = translate.wxcode(code)
         if item.startswith('Vicinity'):
             item = item.lstrip('Vicinity ') + ' in the Vicinity'
         ret.append(item)
     return '. '.join(ret)
 
 
-def metar(wxdata: MetarData, units: Units) -> str:
+def type_and_times(type: str, start: Timestamp, end: Timestamp, probability: Number = None) -> str:
+    """
+    Format line type and times into the beginning of a spoken line string
+    """
+    if not type:
+        return ''    
+    if type == 'BECMG':
+        return f"At {start.dt.hour or 'midnight'} zulu becoming"
+    ret = f"From {start.dt.hour or 'midnight'} to {end.dt.hour or 'midnight'} zulu,"
+    if probability and probability.value:
+        ret += f" there's a {probability.value}% chance for"
+    if type == 'INTER':
+        ret += ' intermittent'
+    elif type == 'TEMPO':
+        ret += ' temporary'
+    return ret
+
+def wind_shear(shear: str, unit_alt: str = 'ft', unit_wind: str = 'kt') -> str:
+    """
+    Format wind shear string into a spoken word string
+    """
+    unit_alt = SPOKEN_UNITS.get(unit_alt, unit_alt)
+    unit_wind = SPOKEN_UNITS.get(unit_wind, unit_wind)
+    return translate.wind_shear(shear, unit_alt, unit_wind, spoken=True) or 'Wind shear unknown'
+
+def metar(data: MetarData, units: Units) -> str:
     """
-    Convert wxdata into a string for text-to-speech
+    Convert MetarData into a string for text-to-speech
     """
-    # We make copies here because the functions may change the original values
-    _data = deepcopy(wxdata)
-    units = deepcopy(units)
     speech = []
-    if _data.wind_direction and _data.wind_speed:
-        speech.append(wind(_data.wind_direction, _data.wind_speed,
-                           _data.wind_gust, _data.wind_variable_direction,
+    if data.wind_direction and data.wind_speed:
+        speech.append(wind(data.wind_direction, data.wind_speed,
+                           data.wind_gust, data.wind_variable_direction,
                            units.wind_speed))
-    if _data.visibility:
-        speech.append(visibility(_data.visibility, units.visibility))
-    if _data.temperature:
-        speech.append(temperature('Temperature', _data.temperature, units.temperature))
-    if _data.dewpoint:
-        speech.append(temperature('Dew point', _data.dewpoint, units.temperature))
-    if _data.altimeter:
-        speech.append(altimeter(_data.altimeter, units.altimeter))
-    if _data.other:
-        speech.append(other(_data.other))
-    speech.append(translate.clouds(_data.clouds,
+    if data.visibility:
+        speech.append(visibility(data.visibility, units.visibility))
+    if data.temperature:
+        speech.append(temperature('Temperature', data.temperature, units.temperature))
+    if data.dewpoint:
+        speech.append(temperature('Dew point', data.dewpoint, units.temperature))
+    if data.altimeter:
+        speech.append(altimeter(data.altimeter, units.altimeter))
+    if data.other:
+        speech.append(other(data.other))
+    speech.append(translate.clouds(data.clouds,
                                    units.altitude).replace(' - Reported AGL', ''))
     return ('. '.join([l for l in speech if l])).replace(',', '.')
+
+
+def taf_line(line: TafLineData, units: Units) -> str:
+    """
+    Convert TafLineData into a string for text-to-speech
+    """
+    speech = []
+    start = type_and_times(line.type, line.start_time, line.end_time, line.probability)
+    if line.wind_direction and line.wind_speed:
+        speech.append(wind(line.wind_direction, line.wind_speed,
+                           line.wind_gust, unit=units.wind_speed))
+    if line.wind_shear:
+        speech.append(wind_shear(line.wind_shear, units.altimeter, units.wind_speed))
+    if line.visibility:
+        speech.append(visibility(line.visibility, units.visibility))
+    if line.altimeter:
+        speech.append(altimeter(line.altimeter, units.altimeter))
+    if line.other:
+        speech.append(other(line.other))
+    speech.append(translate.clouds(line.clouds,
+                                   units.altitude).replace(' - Reported AGL', ''))
+    if line.turbulance:
+        speech.append(translate.turb_ice(line.turbulance, units.altitude))
+    if line.icing:
+        speech.append(translate.turb_ice(line.icing, units.altitude))
+    return start + ' ' + ('. '.join([l for l in speech if l])).replace(',', '.')
+
+
+def taf(data: TafData, units: Units) -> str:
+    """
+    Convert TafData into a string for text-to-speech
+    """
+    try:
+        month = data.start_time.dt.strftime(r'%B')
+        day = ordinal(data.start_time.dt.day)
+        ret = f"Starting on {month} {day} - "
+    except AttributeError:
+        ret = ''
+    return ret + '. '.join([taf_line(line, units) for line in data.forecast])

avwx/structs.py

@@ -97,7 +97,7 @@ class MetarData(ReportData, SharedData):
 class TafLineData(SharedData):
     end_time: Timestamp
     icing: [str]
-    probability: str
+    probability: Number
     raw: str
     start_time: Timestamp
     turbulance: [str]

avwx/translate.py

@@ -86,6 +86,7 @@ def wind(direction: Number,
     Ex: NNE-020 (variable 010 to 040) at 14kt gusting to 20kt
     """
     ret = ''
+    target = 'spoken' if spoken else 'repr'
     # Wind direction
     if direction:
         if direction.repr in WIND_DIR_REPR:
@@ -95,13 +96,10 @@ def wind(direction: Number,
         else:
             if cardinals:
                 ret += get_cardinal_direction(direction.value) + '-'
-            ret += core.spoken_number(direction.repr) if spoken else direction.repr
+            ret += getattr(direction, target)
     # Variable direction
     if vardir and isinstance(vardir, list):
-        if spoken:
-            vardir = [core.spoken_number(d.repr) for d in vardir]
-        else:
-            vardir = [d.repr for d in vardir]
+        vardir = [getattr(var, target) for var in vardir]
         ret += ' (variable {} to {})'.format(*vardir)
     # Speed
     if speed and speed.value:
@@ -187,6 +185,8 @@ def clouds(clds: [Cloud], unit: str = 'ft') -> str:
 
     Ex: Broken layer at 2200ft (Cumulonimbus), Overcast layer at 3600ft - Reported AGL
     """
+    if clds is None:
+        return ''
     ret = []
     for cloud in clds:
         if cloud.altitude is None:
@@ -236,18 +236,17 @@ def other_list(wxcodes: [str]) -> str:
     return ', '.join([wxcode(code) for code in wxcodes])
 
 
-def wind_shear(shear: str, unit_alt: str = 'ft', unit_wnd: str = 'kt') -> str:
+def wind_shear(shear: str, unit_alt: str = 'ft', unit_wind: str = 'kt', spoken: bool = False) -> str:
     """
     Translate wind shear into a readable string
 
     Ex: Wind shear 2000ft from 140 at 30kt
     """
     if not shear or 'WS' not in shear or '/' not in shear:
         return ''
-    shear = shear[2:].rstrip(unit_wnd.upper()).split('/')
-    return 'Wind shear {alt}{unit_alt} from {winddir} at {speed}{unit_wind}'.format(
-        alt=int(shear[0]) * 100, unit_alt=unit_alt, winddir=shear[1][:3],
-        speed=shear[1][3:], unit_wind=unit_wnd)
+    shear = shear[2:].rstrip(unit_wind.upper()).split('/')
+    wdir = core.spoken_number(shear[1][:3]) if spoken else shear[1][:3]
+    return f'Wind shear {int(shear[0])*100}{unit_alt} from {wdir} at {shear[1][3:]}{unit_wind}'
 
 
 def turb_ice(turbice: [str], unit: str = 'ft') -> str:

tests/metar/EGLL.json

@@ -1,45 +1,33 @@
 {
     "data": {
         "altimeter": {
-            "repr": "1014",
-            "spoken": "one zero one four",
-            "value": 1014
+            "repr": "1020",
+            "spoken": "one zero two zero",
+            "value": 1020
         },
         "clouds": [
             {
-                "altitude": 18,
+                "altitude": 31,
                 "modifier": null,
-                "repr": "BKN018",
-                "type": "BKN"
-            },
-            {
-                "altitude": 24,
-                "modifier": null,
-                "repr": "BKN024",
-                "type": "BKN"
-            },
-            {
-                "altitude": 28,
-                "modifier": null,
-                "repr": "OVC028",
+                "repr": "OVC031",
                 "type": "OVC"
             }
         ],
         "dewpoint": {
-            "repr": "15",
-            "spoken": "one five",
-            "value": 15
+            "repr": "16",
+            "spoken": "one six",
+            "value": 16
         },
-        "flight_rules": "MVFR",
+        "flight_rules": "VFR",
         "other": [],
-        "raw": "EGLL 160250Z AUTO 24010KT 9999 BKN018 BKN024 OVC028 19/15 Q1014 NOSIG",
+        "raw": "EGLL 192350Z AUTO 28008KT 9999 OVC031 19/16 Q1020 NOSIG",
         "remarks": "NOSIG",
         "remarks_info": {
             "dewpoint_decimal": null,
             "temperature_decimal": null
         },
         "runway_visibility": [],
-        "sanitized": "EGLL 160250Z AUTO 24010KT 9999 BKN018 BKN024 OVC028 19/15 Q1014 NOSIG",
+        "sanitized": "EGLL 192350Z AUTO 28008KT 9999 OVC031 19/16 Q1020 NOSIG",
         "station": "EGLL",
         "temperature": {
             "repr": "19",
@@ -53,19 +41,19 @@
             "value": 9999
         },
         "wind_direction": {
-            "repr": "240",
-            "spoken": "two four zero",
-            "value": 240
+            "repr": "280",
+            "spoken": "two eight zero",
+            "value": 280
         },
         "wind_gust": null,
         "wind_speed": {
-            "repr": "10",
-            "spoken": "one zero",
-            "value": 10
+            "repr": "08",
+            "spoken": "eight",
+            "value": 8
         },
         "wind_variable_direction": []
     },
-    "speech": "Winds two four zero at 10kt. Visibility one zero kilometers. Temperature one nine degrees Celsius. Dew point one five degrees Celsius. Altimeter one zero one four. Broken layer at 1800ft. Broken layer at 2400ft. Overcast layer at 2800ft",
+    "speech": "Winds two eight zero at 8kt. Visibility one zero kilometers. Temperature one nine degrees Celsius. Dew point one six degrees Celsius. Altimeter one zero two zero. Overcast layer at 3100ft",
     "station_info": {
         "city": "London",
         "country": "GBR",
@@ -78,15 +66,15 @@
         "priority": 6,
         "state": "ENG"
     },
-    "summary": "Winds WSW-240 at 10kt, Vis 10km, Temp 19\u00b0C, Dew 15\u00b0C, Alt 1014 hPa, Broken layer at 1800ft, Broken layer at 2400ft, Overcast layer at 2800ft",
+    "summary": "Winds W-280 at 8kt, Vis 10km, Temp 19\u00b0C, Dew 16\u00b0C, Alt 1020 hPa, Overcast layer at 3100ft",
     "translations": {
-        "altimeter": "1014 hPa (29.94 inHg)",
-        "clouds": "Broken layer at 1800ft, Broken layer at 2400ft, Overcast layer at 2800ft - Reported AGL",
-        "dewpoint": "15\u00b0C (59\u00b0F)",
+        "altimeter": "1020 hPa (30.12 inHg)",
+        "clouds": "Overcast layer at 3100ft - Reported AGL",
+        "dewpoint": "16\u00b0C (61\u00b0F)",
         "other": "",
         "remarks": {},
         "temperature": "19\u00b0C (66\u00b0F)",
         "visibility": "10km (6.2sm)",
-        "wind": "WSW-240 at 10kt"
+        "wind": "W-280 at 8kt"
     }
 }