avwx.current.taf

  1"""
  2A TAF (Terminal Aerodrome Forecast) is a 24-hour weather forecast for the area
  35 statute miles from the reporting station. They are update once every three or
  4six hours or when significant changes warrant an update, and the observations
  5are valid for six hours or until the next report is issued
  6"""
  7
  8# stdlib
  9from contextlib import suppress
 10from datetime import date
 11from typing import List, Tuple, Optional
 12
 13# module
 14from avwx.current.base import Report, get_wx_codes
 15from avwx.parsing import core, speech, summary
 16from avwx.parsing.remarks import parse as parse_remarks
 17from avwx.parsing.sanitization.taf import clean_taf_list, clean_taf_string
 18from avwx.parsing.translate.taf import translate_taf
 19from avwx.static.core import FLIGHT_RULES
 20from avwx.static.taf import TAF_RMK, TAF_NEWLINE, TAF_NEWLINE_STARTSWITH
 21from avwx.station import uses_na_format, valid_station
 22from avwx.structs import (
 23    Cloud,
 24    Number,
 25    Sanitization,
 26    TafData,
 27    TafLineData,
 28    TafTrans,
 29    Timestamp,
 30    Units,
 31)
 32
 33
 34class Taf(Report):
 35    """
 36    The Taf class offers an object-oriented approach to managing TAF data for a
 37    single station.
 38
 39    ```python
 40    >>> from avwx import Taf
 41    >>> kjfk = Taf("KJFK")
 42    >>> kjfk.station.name
 43    'John F Kennedy International Airport'
 44    >>> kjfk.update()
 45    True
 46    >>> kjfk.last_updated
 47    datetime.datetime(2018, 3, 4, 23, 43, 26, 209644, tzinfo=datetime.timezone.utc)
 48    >>> kjfk.raw
 49    'KJFK 042030Z 0421/0524 33016G27KT P6SM BKN045 FM051600 36016G22KT P6SM BKN040 FM052100 35013KT P6SM SCT035'
 50    >>> len(kjfk.data.forecast)
 51    3
 52    >>> kjfk.data.forecast[0].flight_rules
 53    'VFR'
 54    >>> kjfk.translations.forecast[0].wind
 55    'NNW-330 at 16kt gusting to 27kt'
 56    >>> kjfk.speech
 57    'Starting on March 4th - From 21 to 16 zulu, Winds three three zero at 16kt gusting to 27kt. Visibility greater than six miles. Broken layer at 4500ft. From 16 to 21 zulu, Winds three six zero at 16kt gusting to 22kt. Visibility greater than six miles. Broken layer at 4000ft. From 21 to midnight zulu, Winds three five zero at 13kt. Visibility greater than six miles. Scattered clouds at 3500ft'
 58    ```
 59
 60    The `parse` and `from_report` methods can parse a report string if you want
 61    to override the normal fetching process.
 62
 63    ```python
 64    >>> from avwx import Taf
 65    >>> report = "TAF ZYHB 082300Z 0823/0911 VRB03KT 9999 SCT018 BKN120 TX14/0907Z TN04/0921Z FM090100 09015KT 9999 -SHRA WS020/13045KT SCT018 BKN120 BECMG 0904/0906 34008KT PROB30 TEMPO 0906/0911 7000 -RA SCT020 650104 530804 RMK FCST BASED ON AUTO OBS. NXT FCST BY 090600Z"
 66    >>> zyhb = Taf.from_report(report)
 67    True
 68    >>> zyhb.station.city
 69    'Hulan'
 70    >>> zyhb.data.remarks
 71    'RMK FCST BASED ON AUTO OBS. NXT FCST BY 090600Z'
 72    >>> zyhb.summary[-1]
 73    'Vis 7km, Light Rain, Scattered clouds at 2000ft, Frequent moderate turbulence in clear air from 8000ft to 12000ft, Moderate icing in clouds from 1000ft to 5000ft'
 74    ```
 75    """
 76
 77    data: Optional[TafData] = None
 78    translations: Optional[TafTrans] = None  # type: ignore
 79
 80    async def _post_update(self) -> None:
 81        if self.code is None or self.raw is None:
 82            return
 83        self.data, self.units, self.sanitization = parse(
 84            self.code, self.raw, self.issued
 85        )
 86        if self.data is None or self.units is None:
 87            return
 88        self.translations = translate_taf(self.data, self.units)
 89
 90    def _post_parse(self) -> None:
 91        if self.code is None or self.raw is None:
 92            return
 93        self.data, self.units, self.sanitization = parse(
 94            self.code, self.raw, self.issued
 95        )
 96        if self.data is None or self.units is None:
 97            return
 98        self.translations = translate_taf(self.data, self.units)
 99
100    @property
101    def summary(self) -> List[str]:
102        """Condensed summary for each forecast created from translations"""
103        if not self.translations:
104            self.update()
105        if self.translations is None or self.translations.forecast is None:
106            return []
107        return [summary.taf(trans) for trans in self.translations.forecast]
108
109    @property
110    def speech(self) -> Optional[str]:
111        """Report summary designed to be read by a text-to-speech program"""
112        if not self.data:
113            self.update()
114        if self.data is None or self.units is None:
115            return None
116        return speech.taf(self.data, self.units)
117
118
119LINE_FIXES = {
120    "TEMP0": "TEMPO",
121    "TEMP O": "TEMPO",
122    "TMPO": "TEMPO",
123    "TE MPO": "TEMPO",
124    "TEMP ": "TEMPO ",
125    "T EMPO": "TEMPO",
126    " EMPO": " TEMPO",
127    "TEMO": "TEMPO",
128    "BECM G": "BECMG",
129    "BEMCG": "BECMG",
130    "BE CMG": "BECMG",
131    "B ECMG": "BECMG",
132    " BEC ": " BECMG ",
133    "BCEMG": "BECMG",
134    "BEMG": "BECMG",
135}
136
137
138def sanitize_line(txt: str, sans: Sanitization) -> str:
139    """Fixes common mistakes with 'new line' signifiers so that they can be recognized"""
140    for key, fix in LINE_FIXES.items():
141        if key in txt:
142            txt = txt.replace(key, fix)
143            sans.log(key, fix)
144    # Fix when space is missing following new line signifiers
145    for item in ["BECMG", "TEMPO"]:
146        if item in txt and f"{item} " not in txt:
147            index = txt.find(item) + len(item)
148            txt = f"{txt[:index]} {txt[index:]}"
149            sans.extra_spaces_needed = True
150    return txt
151
152
153def get_taf_remarks(txt: str) -> Tuple[str, str]:
154    """Returns report and remarks separated if found"""
155    remarks_start = core.find_first_in_list(txt, TAF_RMK)
156    if remarks_start == -1:
157        return txt, ""
158    remarks = txt[remarks_start:]
159    txt = txt[:remarks_start].strip()
160    return txt, remarks
161
162
163def get_alt_ice_turb(
164    data: List[str],
165) -> Tuple[List[str], Optional[Number], List[str], List[str]]:
166    """Returns the report list and removed: Altimeter string, Icing list, Turbulence list"""
167    altimeter_number = None
168    icing, turbulence = [], []
169    for i, item in reversed(list(enumerate(data))):
170        if len(item) > 6 and item.startswith("QNH") and item[3:7].isdigit():
171            altimeter = data.pop(i)[3:7]
172            if altimeter[0] in ("2", "3"):
173                altimeter = f"{altimeter[:2]}.{altimeter[2:]}"
174            altimeter_number = core.make_number(altimeter, literal=True)
175        elif item.isdigit():
176            if item[0] == "6":
177                icing.append(data.pop(i))
178            elif item[0] == "5":
179                turbulence.append(data.pop(i))
180    return data, altimeter_number, icing, turbulence
181
182
183def starts_new_line(item: str) -> bool:
184    """Returns True if the given element should start a new report line"""
185    if item in TAF_NEWLINE:
186        return True
187    return any(item.startswith(start) for start in TAF_NEWLINE_STARTSWITH)
188
189
190def split_taf(txt: str) -> List[str]:
191    """Splits a TAF report into each distinct time period"""
192    lines = []
193    split = txt.split()
194    last_index = 0
195    for i, item in enumerate(split):
196        if starts_new_line(item) and i != 0 and not split[i - 1].startswith("PROB"):
197            lines.append(" ".join(split[last_index:i]))
198            last_index = i
199    lines.append(" ".join(split[last_index:]))
200    return lines
201
202
203# TAF line report type and start/end times
204def get_type_and_times(
205    data: List[str],
206) -> Tuple[List[str], str, Optional[str], Optional[str], Optional[str]]:
207    """Returns the report list and removed:
208
209    Report type string, start time string, end time string
210    """
211    report_type, start_time, end_time, transition = "FROM", None, None, None
212    if data:
213        # TEMPO, BECMG, INTER
214        if data[0] in TAF_NEWLINE:
215            report_type = data.pop(0)
216        # PROB[30,40]
217        elif len(data[0]) == 6 and data[0].startswith("PROB"):
218            report_type = data.pop(0)
219    if data:
220        # 1200/1306
221        if (
222            len(data[0]) == 9
223            and data[0][4] == "/"
224            and data[0][:4].isdigit()
225            and data[0][5:].isdigit()
226        ):
227            start_time, end_time = data.pop(0).split("/")
228
229        # 1200 1306
230        elif (
231            len(data) == 8
232            and len(data[0]) == 4
233            and len(data[1]) == 4
234            and data[0].isdigit()
235            and data[1].isdigit()
236        ):
237            start_time = data.pop(0)
238            end_time = data.pop(0)
239
240        # 120000
241        elif len(data[0]) == 6 and data[0].isdigit() and data[0][-2:] == "00":
242            start_time = data.pop(0)[:4]
243        # FM120000
244        elif len(data[0]) > 7 and data[0].startswith("FM"):
245            report_type = "FROM"
246            if (
247                "/" in data[0]
248                and data[0][2:].split("/")[0].isdigit()
249                and data[0][2:].split("/")[1].isdigit()
250            ):
251                start_time, end_time = data.pop(0)[2:].split("/")
252            elif data[0][2:8].isdigit():
253                start_time = data.pop(0)[2:6]
254            # TL120600
255            if (
256                data
257                and len(data[0]) > 7
258                and data[0].startswith("TL")
259                and data[0][2:8].isdigit()
260            ):
261                end_time = data.pop(0)[2:6]
262    if report_type == "BECMG":
263        transition, start_time, end_time = start_time, end_time, None
264    return data, report_type, start_time, end_time, transition
265
266
267def _is_tempo_or_prob(line: TafLineData) -> bool:
268    """Returns True if report type is TEMPO or non-null probability"""
269    return line.type == "TEMPO" or line.probability is not None
270
271
272def _get_next_time(lines: List[TafLineData], target: str) -> Optional[Timestamp]:
273    """Returns the next normal time target value or empty"""
274    for line in lines:
275        if _is_tempo_or_prob(line):
276            continue
277        if target == "start_time":
278            time = line.transition_start or getattr(line, target)
279        else:
280            time = getattr(line, target)
281        if time:
282            return time
283    return None
284
285
286def find_missing_taf_times(
287    lines: List[TafLineData], start: Optional[Timestamp], end: Optional[Timestamp]
288) -> List[TafLineData]:
289    """Fix any missing time issues (except for error/empty lines)"""
290    if not lines:
291        return lines
292    # Assign start time
293    lines[0].start_time = start
294    # Fix other times
295    last_fm_line = 0
296    for i, line in enumerate(lines):
297        if _is_tempo_or_prob(line):
298            continue
299        last_fm_line = i
300        # Search remaining lines to fill empty end or previous for empty start
301        for target, other, direc in (("start", "end", -1), ("end", "start", 1)):
302            target += "_time"
303            if not getattr(line, target):
304                setattr(
305                    line, target, _get_next_time(lines[i::direc][1:], f"{other}_time")
306                )
307    # Special case for final forcast
308    if last_fm_line:
309        lines[last_fm_line].end_time = end
310    # Reset original end time if still empty
311    if lines and not lines[0].end_time:
312        lines[0].end_time = end
313    return lines
314
315
316def get_wind_shear(data: List[str]) -> Tuple[List[str], Optional[str]]:
317    """Returns the report list and the remove wind shear"""
318    shear = None
319    for i, item in reversed(list(enumerate(data))):
320        if len(item) > 6 and item.startswith("WS") and item[5] == "/":
321            shear = data.pop(i).replace("KT", "")
322    return data, shear
323
324
325def get_temp_min_and_max(
326    data: List[str],
327) -> Tuple[List[str], Optional[str], Optional[str]]:
328    """Pull out Max temp at time and Min temp at time items from wx list"""
329    temp_max, temp_min = "", ""
330    for i, item in reversed(list(enumerate(data))):
331        if len(item) > 6 and item[0] == "T" and "/" in item:
332            # TX12/1316Z
333            if item[1] == "X":
334                temp_max = data.pop(i)
335            # TNM03/1404Z
336            elif item[1] == "N":
337                temp_min = data.pop(i)
338            # TM03/1404Z T12/1316Z -> Will fix TN/TX
339            elif item[1] == "M" or item[1].isdigit():
340                if temp_min:
341                    if int(temp_min[2 : temp_min.find("/")].replace("M", "-")) > int(
342                        item[1 : item.find("/")].replace("M", "-")
343                    ):
344                        temp_max, temp_min = f"TX{temp_min[2:]}", f"TN{item[1:]}"
345                    else:
346                        temp_max = f"TX{item[1:]}"
347                else:
348                    temp_min = f"TN{item[1:]}"
349                data.pop(i)
350    return data, temp_max or None, temp_min or None
351
352
353def get_oceania_temp_and_alt(data: List[str]) -> Tuple[List[str], List[str], List[str]]:
354    """Get Temperature and Altimeter lists for Oceania TAFs"""
355    tlist: List[str] = []
356    qlist: List[str] = []
357    if "T" in data:
358        data, tlist = core.get_digit_list(data, data.index("T"))
359    if "Q" in data:
360        data, qlist = core.get_digit_list(data, data.index("Q"))
361    return data, tlist, qlist
362
363
364def get_taf_flight_rules(lines: List[TafLineData]) -> List[TafLineData]:
365    """Get flight rules by looking for missing data in prior reports"""
366    for i, line in enumerate(lines):
367        temp_vis, temp_cloud, is_clear = line.visibility, line.clouds, False
368        for report in reversed(lines[: i + 1]):
369            if not _is_tempo_or_prob(report):
370                if not temp_vis:
371                    temp_vis = report.visibility
372                # SKC or CLR should force no clouds instead of looking back
373                if "SKC" in report.other or "CLR" in report.other:
374                    is_clear = True
375                elif temp_vis and temp_vis.repr == "CAVOK":
376                    is_clear = True
377                elif temp_cloud == []:
378                    temp_cloud = report.clouds
379                if temp_vis and temp_cloud != []:
380                    break
381        if is_clear:
382            temp_cloud = []
383        line.flight_rules = FLIGHT_RULES[
384            core.get_flight_rules(temp_vis, core.get_ceiling(temp_cloud))
385        ]
386    return lines
387
388
389def fix_report_header(report: str) -> str:
390    """Corrects the header order for key elements"""
391    split_report = report.split()
392
393    # Limit scope to only the first few elements. Remarks may include similar tokens
394    header_length = min(len(split_report), 6)
395    headers = split_report[:header_length]
396
397    fixed_headers = []
398    for target in ("TAF", "AMD", "COR"):
399        with suppress(ValueError):
400            headers.remove(target)
401            fixed_headers.append(target)
402
403    return " ".join(fixed_headers + headers + split_report[header_length:])
404
405
406def parse(
407    station: str, report: str, issued: Optional[date] = None
408) -> Tuple[Optional[TafData], Optional[Units], Optional[Sanitization]]:
409    """Returns TafData and Units dataclasses with parsed data and their associated units"""
410    # pylint: disable=too-many-locals
411    if not report:
412        return None, None, None
413    valid_station(station)
414    report = fix_report_header(report)
415    while len(report) > 3 and report[:4] in ("TAF ", "AMD ", "COR "):
416        report = report[4:]
417    start_time: Optional[Timestamp] = None
418    end_time: Optional[Timestamp] = None
419    sans = Sanitization()
420    sanitized = clean_taf_string(report, sans)
421    _, new_station, time = core.get_station_and_time(sanitized[:20].split())
422    if new_station is not None:
423        station = new_station
424    sanitized = sanitized.replace(station, "")
425    if time:
426        sanitized = sanitized.replace(time, "").strip()
427    units = Units.north_american() if uses_na_format(station) else Units.international()
428    # Find and remove remarks
429    sanitized, remarks = get_taf_remarks(sanitized)
430    # Split and parse each line
431    lines = split_taf(sanitized)
432    parsed_lines = parse_lines(lines, units, sans, issued)
433    # Perform additional info extract and corrections
434    max_temp: Optional[str] = None
435    min_temp: Optional[str] = None
436    if parsed_lines:
437        (
438            parsed_lines[-1].other,
439            max_temp,
440            min_temp,
441        ) = get_temp_min_and_max(parsed_lines[-1].other)
442        if not (max_temp or min_temp):
443            (
444                parsed_lines[0].other,
445                max_temp,
446                min_temp,
447            ) = get_temp_min_and_max(parsed_lines[0].other)
448        # Set start and end times based on the first line
449        start_time, end_time = parsed_lines[0].start_time, parsed_lines[0].end_time
450        parsed_lines[0].end_time = None
451        parsed_lines = find_missing_taf_times(parsed_lines, start_time, end_time)
452        parsed_lines = get_taf_flight_rules(parsed_lines)
453    # Extract Oceania-specific data
454    alts: Optional[List[str]] = None
455    temps: Optional[List[str]] = None
456    if station[0] == "A":
457        (
458            parsed_lines[-1].other,
459            alts,
460            temps,
461        ) = get_oceania_temp_and_alt(parsed_lines[-1].other)
462    # Convert wx codes
463    for line in parsed_lines:
464        line.other, line.wx_codes = get_wx_codes(line.other)
465    sanitized = " ".join(i for i in (station, time, sanitized) if i)
466    struct = TafData(
467        raw=report,
468        sanitized=sanitized,
469        station=station,
470        time=core.make_timestamp(time, target_date=issued),
471        remarks=remarks,
472        remarks_info=parse_remarks(remarks),
473        forecast=parsed_lines,
474        start_time=start_time,
475        end_time=end_time,
476        max_temp=max_temp,
477        min_temp=min_temp,
478        alts=alts,
479        temps=temps,
480    )
481    return struct, units, sans
482
483
484def parse_lines(
485    lines: List[str], units: Units, sans: Sanitization, issued: Optional[date] = None
486) -> List[TafLineData]:
487    """Returns a list of parsed line dictionaries"""
488    parsed_lines: List[TafLineData] = []
489    prob = ""
490    while lines:
491        raw_line = lines[0].strip()
492        line = sanitize_line(raw_line, sans)
493        # Remove prob from the beginning of a line
494        if line.startswith("PROB"):
495            # Add standalone prob to next line
496            if len(line) == 6:
497                prob = line
498                line = ""
499            # Add to current line
500            elif len(line) > 6:
501                prob = line[:6]
502                line = line[6:].strip()
503        if line:
504            parsed_line = parse_line(line, units, sans, issued)
505            parsed_line.probability = (
506                None if " " in prob else core.make_number(prob[4:])
507            )
508            parsed_line.raw = raw_line
509            if prob:
510                parsed_line.sanitized = f"{prob} {parsed_line.sanitized}"
511            prob = ""
512            parsed_lines.append(parsed_line)
513        lines.pop(0)
514    return parsed_lines
515
516
517def parse_line(
518    line: str, units: Units, sans: Sanitization, issued: Optional[date] = None
519) -> TafLineData:
520    """Parser for the International TAF forcast variant"""
521    # pylint: disable=too-many-locals
522    data = core.dedupe(line.split())
523    data = clean_taf_list(data, sans)
524    sanitized = " ".join(data)
525    data, report_type, start_time, end_time, transition = get_type_and_times(data)
526    data, wind_shear = get_wind_shear(data)
527    (
528        data,
529        wind_direction,
530        wind_speed,
531        wind_gust,
532        wind_variable_direction,
533    ) = core.get_wind(data, units)
534    if "CAVOK" in data:
535        visibility = core.make_number("CAVOK")
536        clouds: List[Cloud] = []
537        data.pop(data.index("CAVOK"))
538    else:
539        data, visibility = core.get_visibility(data, units)
540        data, clouds = core.get_clouds(data)
541    other, altimeter, icing, turbulence = get_alt_ice_turb(data)
542    return TafLineData(
543        altimeter=altimeter,
544        clouds=clouds,
545        flight_rules="",
546        other=other,
547        visibility=visibility,
548        wind_direction=wind_direction,
549        wind_gust=wind_gust,
550        wind_speed=wind_speed,
551        wx_codes=[],
552        end_time=core.make_timestamp(end_time, target_date=issued),
553        icing=icing,
554        probability=None,
555        raw=line,
556        sanitized=sanitized,
557        start_time=core.make_timestamp(start_time, target_date=issued),
558        transition_start=core.make_timestamp(transition, target_date=issued),
559        turbulence=turbulence,
560        type=report_type,
561        wind_shear=wind_shear,
562        wind_variable_direction=wind_variable_direction,
563    )