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 __future__ import annotations
 10
 11from contextlib import suppress
 12from typing import TYPE_CHECKING
 13
 14# module
 15from avwx.current.base import Report, get_wx_codes
 16from avwx.parsing import core, speech, summary
 17from avwx.parsing.remarks import parse as parse_remarks
 18from avwx.parsing.sanitization.taf import clean_taf_list, clean_taf_string
 19from avwx.parsing.translate.taf import translate_taf
 20from avwx.static.core import FLIGHT_RULES
 21from avwx.static.taf import TAF_NEWLINE, TAF_NEWLINE_STARTSWITH, TAF_RMK
 22from avwx.station import uses_na_format, valid_station
 23from avwx.structs import (
 24    Cloud,
 25    Number,
 26    Sanitization,
 27    TafData,
 28    TafLineData,
 29    TafTrans,
 30    Timestamp,
 31    Units,
 32)
 33
 34if TYPE_CHECKING:
 35    from datetime import date
 36
 37
 38class Taf(Report):
 39    """
 40    The Taf class offers an object-oriented approach to managing TAF data for a
 41    single station.
 42
 43    ```python
 44    >>> from avwx import Taf
 45    >>> kjfk = Taf("KJFK")
 46    >>> kjfk.station.name
 47    'John F Kennedy International Airport'
 48    >>> kjfk.update()
 49    True
 50    >>> kjfk.last_updated
 51    datetime.datetime(2018, 3, 4, 23, 43, 26, 209644, tzinfo=datetime.timezone.utc)
 52    >>> kjfk.raw
 53    'KJFK 042030Z 0421/0524 33016G27KT P6SM BKN045 FM051600 36016G22KT P6SM BKN040 FM052100 35013KT P6SM SCT035'
 54    >>> len(kjfk.data.forecast)
 55    3
 56    >>> kjfk.data.forecast[0].flight_rules
 57    'VFR'
 58    >>> kjfk.translations.forecast[0].wind
 59    'NNW-330 at 16kt gusting to 27kt'
 60    >>> kjfk.speech
 61    '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'
 62    ```
 63
 64    The `parse` and `from_report` methods can parse a report string if you want
 65    to override the normal fetching process.
 66
 67    ```python
 68    >>> from avwx import Taf
 69    >>> 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"
 70    >>> zyhb = Taf.from_report(report)
 71    True
 72    >>> zyhb.station.city
 73    'Hulan'
 74    >>> zyhb.data.remarks
 75    'RMK FCST BASED ON AUTO OBS. NXT FCST BY 090600Z'
 76    >>> zyhb.summary[-1]
 77    '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'
 78    ```
 79    """
 80
 81    data: TafData | None = None
 82    translations: TafTrans | None = None  # type: ignore
 83
 84    async def _post_update(self) -> None:
 85        if self.code is None or self.raw is None:
 86            return
 87        self.data, self.units, self.sanitization = parse(self.code, self.raw, self.issued)
 88        if self.data is None or self.units is None:
 89            return
 90        self.translations = translate_taf(self.data, self.units)
 91
 92    def _post_parse(self) -> None:
 93        if self.code is None or self.raw is None:
 94            return
 95        self.data, self.units, self.sanitization = parse(self.code, self.raw, self.issued)
 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) -> str | None:
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    """Fix 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    """Return 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], Number | None, list[str], list[str]]:
166    """Return 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 is_normal_time(item: str) -> bool:
184    """Return if the item looks like a valid TAF (1200/1400) time range."""
185    return len(item) == 9 and item[4] == "/" and item[:4].isdigit() and item[5:].isdigit()
186
187
188def starts_new_line(item: str) -> bool:
189    """Returns True if the given element should start a new report line"""
190    if item in TAF_NEWLINE:
191        return True
192    return any(item.startswith(start) for start in TAF_NEWLINE_STARTSWITH)
193
194
195def split_taf(txt: str) -> list[str]:
196    """Split a TAF report into each distinct time period."""
197    lines = []
198    split = txt.split()
199    last_index = 0
200    e_splits = enumerate(split)
201    next(e_splits)
202    for i, item in e_splits:
203        if (starts_new_line(item) and not split[i - 1].startswith("PROB")) or (
204            is_normal_time(item) and not starts_new_line(split[i - 1])
205        ):
206            lines.append(" ".join(split[last_index:i]))
207            last_index = i
208    lines.append(" ".join(split[last_index:]))
209    return lines
210
211
212# TAF line report type and start/end times
213def get_type_and_times(
214    data: list[str],
215) -> tuple[list[str], str, str | None, str | None, str | None]:
216    """Extract the report type string, start time string, and end time string."""
217    report_type, start_time, end_time, transition = "FROM", None, None, None
218    # TEMPO, BECMG, INTER
219    if data and data[0] in TAF_NEWLINE or len(data[0]) == 6 and data[0].startswith("PROB"):
220        report_type = data.pop(0)
221    if data:
222        item, length = data[0], len(data[0])
223        # 1200/1306
224        if is_normal_time(item):
225            start_time, end_time = data.pop(0).split("/")
226
227        # 1200 1306
228        elif len(data) == 8 and length == 4 and len(data[1]) == 4 and item.isdigit() and data[1].isdigit():
229            start_time = data.pop(0)
230            end_time = data.pop(0)
231
232        # 120000
233        elif length == 6 and item.isdigit() and item[-2:] == "00":
234            start_time = data.pop(0)[:4]
235        # FM120000
236        elif length > 7 and item.startswith("FM"):
237            report_type = "FROM"
238            if "/" in item and item[2:].split("/")[0].isdigit() and item[2:].split("/")[1].isdigit():
239                start_time, end_time = data.pop(0)[2:].split("/")
240            elif item[2:8].isdigit():
241                start_time = data.pop(0)[2:6]
242            # TL120600
243            if data and length > 7 and data[0].startswith("TL") and data[0][2:8].isdigit():
244                end_time = data.pop(0)[2:6]
245        elif report_type == "BECMG" and length == 5:
246            # 1200/
247            if item[-1] == "/" and item[:4].isdigit():
248                start_time = data.pop(0)[:4]
249            # /1200
250            elif item[0] == "/" and item[1:].isdigit():
251                end_time = data.pop(0)[1:]
252    if report_type == "BECMG":
253        transition, start_time, end_time = start_time, end_time, None
254    return data, report_type, start_time, end_time, transition
255
256
257def _is_tempo_or_prob(line: TafLineData) -> bool:
258    """Return True if report type is TEMPO or non-null probability."""
259    return line.type == "TEMPO" or line.probability is not None
260
261
262def _get_next_time(lines: list[TafLineData], target: str) -> Timestamp | None:
263    """Returns the next normal time target value or empty"""
264    for line in lines:
265        if _is_tempo_or_prob(line):
266            continue
267        time = line.transition_start or getattr(line, target) if target == "start_time" else getattr(line, target)
268        if time:
269            return time  # type: ignore
270    return None
271
272
273def find_missing_taf_times(
274    lines: list[TafLineData], start: Timestamp | None, end: Timestamp | None
275) -> list[TafLineData]:
276    """Fix any missing time issues except for error/empty lines."""
277    if not lines:
278        return lines
279    # Assign start time
280    lines[0].start_time = start
281    # Fix other times
282    last_fm_line = 0
283    for i, line in enumerate(lines):
284        if _is_tempo_or_prob(line):
285            continue
286        last_fm_line = i
287        # Search remaining lines to fill empty end or previous for empty start
288        for target, other, direc in (("start", "end", -1), ("end", "start", 1)):
289            target += "_time"  # noqa: PLW2901
290            if not getattr(line, target):
291                setattr(line, target, _get_next_time(lines[i::direc][1:], f"{other}_time"))
292    # Special case for final forcast
293    if last_fm_line:
294        lines[last_fm_line].end_time = end
295    # Reset original end time if still empty
296    if lines and not lines[0].end_time:
297        lines[0].end_time = end
298    return lines
299
300
301def get_wind_shear(data: list[str]) -> tuple[list[str], str | None]:
302    """Return the report list and the remove wind shear."""
303    shear = None
304    for i, item in reversed(list(enumerate(data))):
305        if len(item) > 6 and item.startswith("WS") and item[5] == "/":
306            shear = data.pop(i).replace("KT", "")
307    return data, shear
308
309
310def get_temp_min_and_max(
311    data: list[str],
312) -> tuple[list[str], str | None, str | None]:
313    """Pull out Max temp at time and Min temp at time items from wx list."""
314    temp_max, temp_min = "", ""
315    for i, item in reversed(list(enumerate(data))):
316        if len(item) > 6 and item[0] == "T" and "/" in item:
317            # TX12/1316Z
318            if item[1] == "X":
319                temp_max = data.pop(i)
320            # TNM03/1404Z
321            elif item[1] == "N":
322                temp_min = data.pop(i)
323            # TM03/1404Z T12/1316Z -> Will fix TN/TX
324            elif item[1] == "M" or item[1].isdigit():
325                if temp_min:
326                    if int(temp_min[2 : temp_min.find("/")].replace("M", "-")) > int(
327                        item[1 : item.find("/")].replace("M", "-")
328                    ):
329                        temp_max, temp_min = f"TX{temp_min[2:]}", f"TN{item[1:]}"
330                    else:
331                        temp_max = f"TX{item[1:]}"
332                else:
333                    temp_min = f"TN{item[1:]}"
334                data.pop(i)
335    return data, temp_max or None, temp_min or None
336
337
338def get_oceania_temp_and_alt(data: list[str]) -> tuple[list[str], list[str], list[str]]:
339    """Get Temperature and Altimeter lists for Oceania TAFs."""
340    tlist: list[str] = []
341    qlist: list[str] = []
342    if "T" in data:
343        data, tlist = core.get_digit_list(data, data.index("T"))
344    if "Q" in data:
345        data, qlist = core.get_digit_list(data, data.index("Q"))
346    return data, tlist, qlist
347
348
349def get_taf_flight_rules(lines: list[TafLineData]) -> list[TafLineData]:
350    """Get flight rules by looking for missing data in prior reports."""
351    for i, line in enumerate(lines):
352        temp_vis, temp_cloud, is_clear = line.visibility, line.clouds, False
353        for report in reversed(lines[: i + 1]):
354            if not _is_tempo_or_prob(report):
355                if not temp_vis:
356                    temp_vis = report.visibility
357                # SKC or CLR should force no clouds instead of looking back
358                if "SKC" in report.other or "CLR" in report.other or temp_vis and temp_vis.repr == "CAVOK":
359                    is_clear = True
360                elif temp_cloud == []:
361                    temp_cloud = report.clouds
362                if temp_vis and temp_cloud != []:
363                    break
364        if is_clear:
365            temp_cloud = []
366        line.flight_rules = FLIGHT_RULES[core.get_flight_rules(temp_vis, core.get_ceiling(temp_cloud))]
367    return lines
368
369
370def fix_report_header(report: str) -> str:
371    """Correct the header order for key elements."""
372    split_report = report.split()
373
374    # Limit scope to only the first few elements. Remarks may include similar tokens
375    header_length = min(len(split_report), 6)
376    headers = split_report[:header_length]
377
378    fixed_headers = []
379    for target in ("TAF", "AMD", "COR"):
380        with suppress(ValueError):
381            headers.remove(target)
382            fixed_headers.append(target)
383
384    return " ".join(fixed_headers + headers + split_report[header_length:])
385
386
387def _is_possible_start_end_time_slash(item: str) -> bool:
388    """Return True if item is a possible period start or end with missing element."""
389    return len(item) == 5 and (
390        # 1200/
391        (item[-1] == "/" and item[:4].isdigit())
392        or
393        # /1200
394        (item[0] == "/" and item[1:].isdigit())
395    )
396
397
398def parse(
399    station: str, report: str, issued: date | None = None
400) -> tuple[TafData | None, Units | None, Sanitization | None]:
401    """Return TafData and Units dataclasses with parsed data and their associated units."""
402    if not report:
403        return None, None, None
404    valid_station(station)
405    report = fix_report_header(report)
406    while len(report) > 3 and report[:4] in ("TAF ", "AMD ", "COR "):
407        report = report[4:]
408    start_time: Timestamp | None = None
409    end_time: Timestamp | None = None
410    sans = Sanitization()
411    sanitized = clean_taf_string(report, sans)
412    _, new_station, time = core.get_station_and_time(sanitized[:20].split())
413    if new_station is not None:
414        station = new_station
415    sanitized = sanitized.replace(station, "")
416    if time:
417        sanitized = sanitized.replace(time, "").strip()
418    units = Units.north_american() if uses_na_format(station) else Units.international()
419    # Find and remove remarks
420    sanitized, remarks = get_taf_remarks(sanitized)
421    # Split and parse each line
422    lines = split_taf(sanitized)
423    parsed_lines = parse_lines(lines, units, sans, issued)
424    # Perform additional info extract and corrections
425    max_temp: str | None = None
426    min_temp: str | None = None
427    if parsed_lines:
428        (
429            parsed_lines[-1].other,
430            max_temp,
431            min_temp,
432        ) = get_temp_min_and_max(parsed_lines[-1].other)
433        if not (max_temp or min_temp):
434            (
435                parsed_lines[0].other,
436                max_temp,
437                min_temp,
438            ) = get_temp_min_and_max(parsed_lines[0].other)
439        # Set start and end times based on the first line
440        start_time, end_time = parsed_lines[0].start_time, parsed_lines[0].end_time
441        parsed_lines[0].end_time = None
442        parsed_lines = find_missing_taf_times(parsed_lines, start_time, end_time)
443        parsed_lines = get_taf_flight_rules(parsed_lines)
444    # Extract Oceania-specific data
445    alts: list[str] | None = None
446    temps: list[str] | None = None
447    if station[0] == "A":
448        (
449            parsed_lines[-1].other,
450            alts,
451            temps,
452        ) = get_oceania_temp_and_alt(parsed_lines[-1].other)
453    # Convert wx codes
454    for line in parsed_lines:
455        line.other, line.wx_codes = get_wx_codes(line.other)
456    sanitized = " ".join(i for i in (station, time, sanitized) if i)
457    struct = TafData(
458        raw=report,
459        sanitized=sanitized,
460        station=station,
461        time=core.make_timestamp(time, target_date=issued),
462        remarks=remarks,
463        remarks_info=parse_remarks(remarks),
464        forecast=parsed_lines,
465        start_time=start_time,
466        end_time=end_time,
467        max_temp=max_temp,
468        min_temp=min_temp,
469        alts=alts,
470        temps=temps,
471    )
472    return struct, units, sans
473
474
475def parse_lines(lines: list[str], units: Units, sans: Sanitization, issued: date | None = None) -> list[TafLineData]:
476    """Return a list of parsed line dictionaries."""
477    parsed_lines: list[TafLineData] = []
478    prob = ""
479    while lines:
480        raw_line = lines[0].strip()
481        line = sanitize_line(raw_line, sans)
482        # Remove prob from the beginning of a line
483        if line.startswith("PROB"):
484            # Add standalone prob to next line
485            if len(line) == 6:
486                prob = line
487                line = ""
488            # Add to current line
489            elif len(line) > 6:
490                prob = line[:6]
491                line = line[6:].strip()
492        if line:
493            parsed_line = parse_line(line, units, sans, issued)
494            parsed_line.probability = None if " " in prob else core.make_number(prob[4:])
495            parsed_line.raw = raw_line
496            if prob:
497                parsed_line.sanitized = f"{prob} {parsed_line.sanitized}"
498            prob = ""
499            parsed_lines.append(parsed_line)
500        lines.pop(0)
501    return parsed_lines
502
503
504def parse_line(line: str, units: Units, sans: Sanitization, issued: date | None = None) -> TafLineData:
505    """Parser for the International TAF forcast variant."""
506    data: list[str] = core.dedupe(line.split())
507    # Grab original time piece under certain conditions to preserve a useful slash
508    old_time = data[1] if len(data) > 1 and _is_possible_start_end_time_slash(data[1]) else None
509    data = clean_taf_list(data, sans)
510    if old_time and len(data) > 1 and data[1] == old_time.strip("/"):
511        data[1] = old_time
512    sanitized = " ".join(data)
513    data, report_type, start_time, end_time, transition = get_type_and_times(data)
514    data, wind_shear = get_wind_shear(data)
515    (
516        data,
517        wind_direction,
518        wind_speed,
519        wind_gust,
520        wind_variable_direction,
521    ) = core.get_wind(data, units)
522    if "CAVOK" in data:
523        visibility = core.make_number("CAVOK")
524        clouds: list[Cloud] = []
525        data.pop(data.index("CAVOK"))
526    else:
527        data, visibility = core.get_visibility(data, units)
528        data, clouds = core.get_clouds(data)
529    other, altimeter, icing, turbulence = get_alt_ice_turb(data)
530    return TafLineData(
531        altimeter=altimeter,
532        clouds=clouds,
533        flight_rules="",
534        other=other,
535        visibility=visibility,
536        wind_direction=wind_direction,
537        wind_gust=wind_gust,
538        wind_speed=wind_speed,
539        wx_codes=[],
540        end_time=core.make_timestamp(end_time, target_date=issued),
541        icing=icing,
542        probability=None,
543        raw=line,
544        sanitized=sanitized,
545        start_time=core.make_timestamp(start_time, target_date=issued),
546        transition_start=core.make_timestamp(transition, target_date=issued),
547        turbulence=turbulence,
548        type=report_type,
549        wind_shear=wind_shear,
550        wind_variable_direction=wind_variable_direction,
551    )