avwx.parsing.core

  1"""Contains the core parsing and indent functions of avwx."""
  2
  3# stdlib
  4from __future__ import annotations
  5
  6import datetime as dt
  7import math
  8import re
  9from calendar import monthrange
 10from contextlib import suppress
 11from copy import copy
 12from typing import TYPE_CHECKING, Any
 13
 14# library
 15from dateutil.relativedelta import relativedelta
 16
 17# module
 18from avwx.static.core import (
 19    CARDINALS,
 20    CLOUD_LIST,
 21    FRACTIONS,
 22    NUMBER_REPL,
 23    SPECIAL_NUMBERS,
 24    WIND_UNITS,
 25)
 26from avwx.structs import Cloud, Fraction, Number, Timestamp, Units
 27
 28if TYPE_CHECKING:
 29    from collections.abc import Iterable
 30
 31
 32def dedupe(items: Iterable[Any], *, only_neighbors: bool = False) -> list[Any]:
 33    """Deduplicate a list while keeping order.
 34
 35    If only_neighbors is True, dedupe will only check neighboring values.
 36    """
 37    ret: list[Any] = []
 38    for item in items:
 39        if (only_neighbors and ret and ret[-1] != item) or item not in ret:
 40            ret.append(item)
 41    return ret
 42
 43
 44def is_unknown(value: str) -> bool:
 45    """Return True if val represents and unknown value."""
 46    if not isinstance(value, str):
 47        raise TypeError
 48    if not value or value.upper() in {"UNKN", "UNK", "UKN"}:
 49        return True
 50    for char in value:
 51        if char not in ("/", "X", "."):
 52            break
 53    else:
 54        return True
 55    return False
 56
 57
 58def get_digit_list(data: list[str], from_index: int) -> tuple[list[str], list[str]]:
 59    """Return a list of items removed from a given list of strings
 60    that are all digits from 'from_index' until hitting a non-digit item.
 61    """
 62    ret = []
 63    data.pop(from_index)
 64    while len(data) > from_index and data[from_index].isdigit():
 65        ret.append(data.pop(from_index))
 66    return data, ret
 67
 68
 69def unpack_fraction(num: str) -> str:
 70    """Return unpacked fraction string 5/2 -> 2 1/2."""
 71    numbers = [int(n) for n in num.split("/") if n]
 72    if len(numbers) != 2 or numbers[0] <= numbers[1]:
 73        return num
 74    numerator, denominator = numbers
 75    over = numerator // denominator
 76    rem = numerator % denominator
 77    return f"{over} {rem}/{denominator}"
 78
 79
 80def remove_leading_zeros(num: str) -> str:
 81    """Strip zeros while handling -, M, and empty strings."""
 82    if not num:
 83        return num
 84    if num.startswith("M"):
 85        ret = "M" + num[1:].lstrip("0")
 86    elif num.startswith("-"):
 87        ret = "-" + num[1:].lstrip("0")
 88    else:
 89        ret = num.lstrip("0")
 90    return "0" if ret in ("", "M", "-") else ret
 91
 92
 93SPOKEN_POSTFIX = (
 94    (" zero zero zero", " thousand"),
 95    (" zero zero", " hundred"),
 96)
 97
 98
 99def spoken_number(num: str, *, literal: bool = False) -> str:
100    """Return the spoken version of a number.
101
102    If literal, no conversion to hundreds/thousands
103
104    Ex: 1.2 -> one point two
105        1 1/2 -> one and one half
106        25000 -> two five thousand
107    """
108    ret = []
109    for part in num.split():
110        if part in FRACTIONS:
111            ret.append(FRACTIONS[part])
112        else:
113            val = " ".join(NUMBER_REPL[char] for char in part if char in NUMBER_REPL)
114            if not literal:
115                for target, replacement in SPOKEN_POSTFIX:
116                    if val.endswith(target):
117                        val = val[: -len(target)] + replacement
118            ret.append(val)
119    return " and ".join(ret)
120
121
122def make_fraction(
123    num: str,
124    repr: str | None = None,  # noqa: A002
125    *,
126    literal: bool = False,
127    speak_prefix: str = "",
128) -> Fraction:
129    """Return a fraction dataclass for numbers with / in them."""
130    num_str, den_str = num.split("/")
131    # 2-1/2 but not -2 1/2
132    if "-" in num_str and not num_str.startswith("-"):
133        num_str = num_str.replace("-", " ")
134    denominator = int(den_str)
135    # Multiply multi-digit numerator
136    if len(num_str) > 1:
137        numerator = int(num_str[:-1]) * denominator + int(num_str[-1])
138        num = f"{numerator}/{denominator}"
139    else:
140        numerator = int(num_str)
141    value = numerator / denominator
142    unpacked = unpack_fraction(num)
143    spoken = speak_prefix + spoken_number(unpacked, literal=literal)
144    return Fraction(repr or num, value, spoken, numerator, denominator, unpacked)
145
146
147def make_number(
148    num: str | None,
149    repr: str | None = None,  # noqa: A002
150    speak: str | None = None,
151    *,
152    literal: bool = False,
153    special: dict | None = None,
154    m_minus: bool = True,
155) -> Number | Fraction | None:
156    """Return a Number or Fraction dataclass for a number string.
157
158    If literal, spoken string will not convert to hundreds/thousands.
159
160    NOTE: Numerators are assumed to have a single digit. Additional are whole numbers.
161    """
162    if not num or is_unknown(num):
163        return None
164    # Check special
165    with suppress(KeyError):
166        item = (special or {}).get(num) or SPECIAL_NUMBERS[num]
167        if isinstance(item, tuple):
168            value, spoken = item
169        else:
170            value = item
171            spoken = spoken_number(str(value), literal=literal)
172        return Number(repr or num, value, spoken)
173    # Check cardinal direction
174    if num in CARDINALS:
175        if not repr:
176            repr = num  # noqa: A001
177        num = str(CARDINALS[num])
178    val_str = num
179    # Remove unit suffixes
180    if val_str.endswith("SM"):
181        repr = val_str[:]  # noqa: A001
182        val_str = val_str[:-2]
183    # Remove spurious characters from the end
184    num = num.rstrip("M.")
185    num = num.replace("O", "0")
186    num = num.replace("+", "")
187    num = num.replace(",", "")
188    # The value is parsed from val_str, which was snapshotted above, so the
189    # cleanup has to reach it too. Previously it only did when the m_minus
190    # branch below happened to reassign val_str from num, which meant "M1O"
191    # parsed as -10 while "1O" raised ValueError out of int().
192    val_str = val_str.rstrip("M.").replace("O", "0").replace("+", "").replace(",", "")
193    # Handle Minus values with errors like 0M04
194    if m_minus and "M" in num:
195        val_str = num.replace("MM", "-").replace("M", "-")
196        while val_str[0] != "-":
197            val_str = val_str[1:]
198    # Check value prefixes
199    speak_prefix = ""
200    if val_str.startswith("ABV "):
201        speak_prefix += "above "
202        val_str = val_str[4:]
203    if val_str.startswith("BLW "):
204        speak_prefix += "below "
205        val_str = val_str[4:]
206    if val_str.startswith("FL"):
207        speak_prefix += "flight level "
208        val_str, literal = val_str[2:], True
209    if val_str.startswith("M"):
210        speak_prefix += "less than "
211        repr = repr or val_str  # noqa: A001
212        val_str = val_str[1:]
213    if val_str.startswith("P"):
214        speak_prefix += "greater than "
215        repr = repr or val_str  # noqa: A001
216        val_str = val_str[1:]
217    # Create Number
218    if not val_str:
219        return None
220    ret: Number | Fraction | None = None
221    # Create Fraction
222    if "/" in val_str:
223        ret = make_fraction(val_str, repr, literal=literal, speak_prefix=speak_prefix)
224    else:
225        val_str = val_str.replace(",", "")
226        # Overwrite float 0 due to "0.0" literal
227        value = float(val_str) or 0 if "." in num else int(val_str)
228        spoken = speak_prefix + spoken_number(speak or str(value), literal=literal)
229        ret = Number(repr or num, value, spoken)
230    # Null the value if "greater than"/"less than"
231    if ret and not m_minus and repr and repr.startswith(("M", "P")):
232        ret.value = None
233    return ret
234
235
236def find_first_in_list(txt: str, str_list: list[str]) -> int:
237    """Return the index of the earliest occurrence of an item from a list in a string.
238
239    Ex: find_first_in_list('foobar', ['bar', 'fin']) -> 3
240    """
241    start = len(txt) + 1
242    for item in str_list:
243        if start > txt.find(item) > -1:
244            start = txt.find(item)
245    return start if len(txt) + 1 > start > -1 else -1
246
247
248def is_timestamp(item: str) -> bool:
249    """Return True if the item matches the timestamp format."""
250    return len(item) == 7 and item[-1] == "Z" and item[:-1].isdigit()
251
252
253def is_timerange(item: str) -> bool:
254    """Return True if the item is a TAF to-from time range."""
255    return len(item) == 9 and item[4] == "/" and item[:4].isdigit() and item[5:].isdigit()
256
257
258def is_possible_temp(temp: str) -> bool:
259    """Return True if all characters are digits or 'M' for minus."""
260    return all((char.isdigit() or char == "M") for char in temp)
261
262
263_Numeric = int | float
264
265
266def relative_humidity(temperature: _Numeric, dewpoint: _Numeric, unit: str = "C") -> float:
267    """Calculate the relative humidity as a 0 to 1 percentage."""
268
269    def saturation(value: _Numeric) -> float:
270        """Return the saturation vapor pressure without the C constant for humidity calc."""
271        return math.exp((17.67 * value) / (243.5 + value))
272
273    if unit == "F":
274        dewpoint = (dewpoint - 32) * 5 / 9
275        temperature = (temperature - 32) * 5 / 9
276    return saturation(dewpoint) / saturation(temperature)
277
278
279# https://aviation.stackexchange.com/questions/47971/how-do-i-calculate-density-altitude-by-hand
280
281
282def pressure_altitude(pressure: float, altitude: _Numeric, unit: str = "inHg") -> int:
283    """Calculate the pressure altitude in feet. Converts pressure units."""
284    if unit == "hPa":
285        pressure *= 0.02953
286    return round((29.92 - pressure) * 1000 + altitude)
287
288
289def density_altitude(pressure: float, temperature: _Numeric, altitude: _Numeric, units: Units) -> int:
290    """Calculate the density altitude in feet. Converts pressure and temperature units."""
291    if units.temperature == "F":
292        temperature = (temperature - 32) * 5 / 9
293    if units.altimeter == "hPa":
294        pressure *= 0.02953
295    pressure_alt = pressure_altitude(pressure, altitude)
296    standard = 15 - (2 * altitude / 1000)
297    return round(((temperature - standard) * 120) + pressure_alt)
298
299
300def get_station_and_time(
301    data: list[str],
302) -> tuple[list[str], str | None, str | None]:
303    """Return the report list and removed station ident and time strings."""
304    if not data:
305        return data, None, None
306    station = data.pop(0)
307    if not data:
308        return data, station, None
309    q_time, r_time = data[0], None
310    if data and q_time.endswith("Z") and q_time[:-1].isdigit():
311        r_time = data.pop(0)
312    elif data and len(q_time) == 6 and q_time.isdigit():
313        r_time = f"{data.pop(0)}Z"
314    return data, station, r_time
315
316
317def is_wind(text: str) -> bool:
318    """Return True if the text is likely a normal wind element."""
319    # Ignore wind shear
320    if text.startswith("WS"):
321        return False
322    # 09010KT, 09010G15KT
323    if len(text) > 4:
324        for ending in WIND_UNITS:
325            unit_index = text.find(ending)
326            if text.endswith(ending) and text[unit_index - 2 : unit_index].isdigit():
327                return True
328    # 09010  09010G15 VRB10
329    if len(text) != 5 and (len(text) < 8 or "G" not in text or "/" in text):
330        return False
331    return text[:5].isdigit() or (text.startswith("VRB") and text[3:5].isdigit())
332
333
334VARIABLE_DIRECTION_PATTERN = re.compile(r"\d{3}V\d{3}")
335
336
337def is_variable_wind_direction(text: str) -> bool:
338    """Return True if element looks like 350V040."""
339    if len(text) < 7:
340        return False
341    return VARIABLE_DIRECTION_PATTERN.match(text[:7]) is not None
342
343
344def separate_wind(text: str) -> tuple[str, str, str]:
345    """Extract the direction, speed, and gust from a wind element."""
346    direction, speed, gust = "", "", ""
347    # Remove gust
348    if "G" in text:
349        g_index = text.find("G")
350        start, end = g_index + 1, g_index + 3
351        # 16006GP99KT ie gust greater than
352        if "GP" in text:
353            end += 1
354        gust = text[start:end]
355        text = text[:g_index] + text[end:]
356    if text:
357        # 10G18KT
358        if len(text) == 2:
359            speed = text
360        else:
361            direction = text[:3]
362            speed = text[3:]
363    return direction, speed, gust
364
365
366def get_wind(
367    data: list[str], units: Units
368) -> tuple[
369    list[str],
370    Number | None,
371    Number | None,
372    Number | None,
373    list[Number],
374]:
375    """Return the report list, direction string, speed string, gust string, and variable direction list."""
376    direction, speed, gust = "", "", ""
377    variable: list[Number] = []
378    # Remove unit and split elements
379    if data:
380        item = copy(data[0])
381        if is_wind(item):
382            for key, unit in WIND_UNITS.items():
383                if item.endswith(key):
384                    units.wind_speed = unit
385                    item = item.replace(key, "")
386                    break
387            direction, speed, gust = separate_wind(item)
388            data.pop(0)
389    # Separated Gust
390    if data and 1 < len(data[0]) < 4 and data[0][0] == "G" and data[0][1:].isdigit():
391        gust = data.pop(0)[1:]
392    # Variable Wind Direction
393    if data and is_variable_wind_direction(data[0]):
394        for item in data.pop(0).split("V"):
395            value = make_number(item, speak=item, literal=True)
396            if value is not None:
397                variable.append(value)
398    # Convert to Number
399    direction_value = make_number(direction, speak=direction, literal=True)
400    speed_value = make_number(speed.strip("BV"), m_minus=False)
401    gust_value = make_number(gust, m_minus=False)
402    return data, direction_value, speed_value, gust_value, variable
403
404
405def get_visibility(data: list[str], units: Units) -> tuple[list[str], Number | None]:
406    """Return the report list and removed visibility string."""
407    visibility = ""
408    if data:
409        item = copy(data[0])
410        # Vis reported in statue miles
411        if item.endswith("SM"):  # 10SM
412            if item[:-2].isdigit():
413                visibility = str(int(item[:-2]))
414            elif "/" in item:
415                visibility = item[: item.find("SM")]  # 1/2SM
416            else:
417                visibility = item[:-2]
418            data.pop(0)
419            units.visibility = "sm"
420        # Vis reported in meters
421        elif len(item) == 4 and item.isdigit():
422            visibility = data.pop(0)
423            units.visibility = "m"
424        elif 7 >= len(item) >= 5 and item[:4].isdigit() and (item[4] in ["M", "N", "S", "E", "W"] or item[4:] == "NDV"):
425            visibility = data.pop(0)[:4]
426            units.visibility = "m"
427        elif len(item) == 5 and item[1:].isdigit() and item[0] in ["M", "P", "B"]:
428            visibility = data.pop(0)[1:]
429            units.visibility = "m"
430        elif item.endswith("KM"):
431            visibility = f"{item[:-2]}000"
432            data.pop(0)
433            units.visibility = "m"
434        # Vis statute miles but split Ex: 2 1/2SM
435        elif len(data) > 1 and data[1].endswith("SM") and "/" in data[1] and item.isdigit():
436            vis1 = data.pop(0)  # 2
437            vis2 = data.pop(0).replace("SM", "")  # 1/2
438            visibility = str(int(vis1) * int(vis2[2]) + int(vis2[0])) + vis2[1:]  # 5/2
439            units.visibility = "sm"
440    return data, make_number(visibility, m_minus=False)
441
442
443def sanitize_cloud(cloud: str) -> str:
444    """Fix rare cloud layer issues."""
445    if len(cloud) < 4:
446        return cloud
447    if not cloud[3].isdigit() and cloud[3] not in ("/", "-"):
448        # Bad "O": FEWO03 -> FEW003
449        if cloud[3] == "O":
450            cloud = f"{cloud[:3]}0{cloud[4:]}"
451        # Move modifiers to end: BKNC015 -> BKN015C
452        elif cloud[3] != "U" and cloud[:4] not in {"BASE", "UNKN"}:
453            cloud = cloud[:3] + cloud[4:] + cloud[3]
454    return cloud
455
456
457def _null_or_int(val: str | None) -> int | None:
458    """Nullify unknown elements and convert ints."""
459    return None if not isinstance(val, str) or is_unknown(val) else int(val)
460
461
462_TOP_OFFSETS = ("-TOPS", "-TOP")
463
464
465def make_cloud(cloud: str) -> Cloud:
466    """Return a Cloud dataclass for a cloud string.
467
468    This function assumes the input is potentially valid.
469    """
470    raw_cloud = cloud
471    cloud_type = ""
472    base: str | None = None
473    top: str | None = None
474    cloud = sanitize_cloud(cloud).replace("/", "")
475    # Separate top
476    for target in _TOP_OFFSETS:
477        topi = cloud.find(target)
478        if topi > -1:
479            top, cloud = cloud[topi + len(target) :], cloud[:topi]
480            break
481    # Separate type
482    ## BASE027
483    if cloud.startswith("BASES"):
484        cloud = cloud[5:]
485    elif cloud.startswith("BASE"):
486        cloud = cloud[4:]
487    ## VV003
488    elif cloud.startswith("VV"):
489        cloud_type, cloud = cloud[:2], cloud[2:]
490    ## FEW010
491    elif len(cloud) >= 3 and cloud[:3] in CLOUD_LIST:
492        cloud_type, cloud = cloud[:3], cloud[3:]
493    ## BKN-OVC065
494    if len(cloud) > 4 and cloud[0] == "-" and cloud[1:4] in CLOUD_LIST:
495        cloud_type += cloud[:4]
496        cloud = cloud[4:]
497    # Separate base
498    if len(cloud) >= 3 and cloud[:3].isdigit():
499        base, cloud = cloud[:3], cloud[3:]
500    elif len(cloud) >= 4 and cloud[:4] == "UNKN":
501        cloud = cloud[4:]
502    # Remainder is considered modifiers
503    modifier = cloud or None
504    # Make Cloud
505    return Cloud(raw_cloud, cloud_type or None, _null_or_int(base), _null_or_int(top), modifier)
506
507
508def get_clouds(data: list[str]) -> tuple[list[str], list]:
509    """Return the report list and removed list of split cloud layers."""
510    clouds = []
511    for i, item in reversed(list(enumerate(data))):
512        if item[:3] in CLOUD_LIST or item[:2] == "VV":
513            cloud = data.pop(i)
514            clouds.append(make_cloud(cloud))
515    # Attempt cloud sort. Fails if None values are present
516    try:
517        clouds.sort(key=lambda cloud: (cloud.base, cloud.type))
518    except TypeError:
519        clouds.reverse()  # Restores original report order
520    return data, clouds
521
522
523def get_flight_rules(visibility: Number | None, ceiling: Cloud | None) -> int:
524    """Return int based on current flight rules from parsed METAR data.
525
526    0=VFR, 1=MVFR, 2=IFR, 3=LIFR
527
528    Note: Common practice is to report no higher than IFR if visibility unavailable.
529    """
530    # Parse visibility
531    vis: _Numeric
532    if visibility is None:
533        vis = 2
534    elif visibility.repr == "CAVOK" or visibility.repr.startswith("P6"):
535        vis = 10
536    elif visibility.repr.startswith("M"):
537        vis = 0
538    elif visibility.value is None:
539        vis = 2
540    # Convert meters to miles
541    elif len(visibility.repr) == 4:
542        vis = (visibility.value or 0) * 0.000621371
543    else:
544        vis = visibility.value or 0
545    # Parse ceiling
546    cld = (ceiling.base if ceiling else 99) or 99
547    # Determine flight rules
548    if (vis <= 5) or (cld <= 30):
549        if (vis < 3) or (cld < 10):
550            if (vis < 1) or (cld < 5):
551                return 3  # LIFR
552            return 2  # IFR
553        return 1  # MVFR
554    return 0  # VFR
555
556
557def get_ceiling(clouds: list[Cloud]) -> Cloud | None:
558    """Return ceiling layer from Cloud-List or None if none found.
559
560    Assumes that the clouds are already sorted lowest to highest.
561
562    Only 'Broken', 'Overcast', and 'Vertical Visibility' are considered ceilings.
563
564    Prevents errors due to lack of cloud information (eg. '' or 'FEW///')
565    """
566    return next((c for c in clouds if c.base and c.type in {"OVC", "BKN", "VV"}), None)
567
568
569def is_altitude(value: str) -> bool:
570    """Return True if the value is a possible altitude."""
571    if len(value) < 5:
572        return False
573    if value.startswith("SFC/"):
574        return True
575    if value.startswith("FL") and value[2:5].isdigit():
576        return True
577    first, *_ = value.split("/")
578    return bool(first[-2:] == "FT" and first[-5:-2].isdigit())
579
580
581def make_altitude(
582    value: str,
583    units: Units,
584    repr: str | None = None,  # noqa: A002
585    *,
586    force_fl: bool = False,
587) -> tuple[Number | None, Units]:
588    """Convert altitude string into a number."""
589    if not value:
590        return None, units
591    raw = repr or value
592    for end in ("FT", "M"):
593        if value.endswith(end):
594            force_fl = False
595            units.altitude = end.lower()
596            value = value.removesuffix(end)
597    # F430
598    if value[0] == "F" and value[1:].isdigit():
599        value = f"FL{value[1:]}"
600    if force_fl and value[:2] != "FL":
601        value = f"FL{value}"
602    return make_number(value, repr=raw), units
603
604
605def parse_date(
606    date: str,
607    hour_threshold: int = 200,
608    *,
609    time_only: bool = False,
610    target: dt.date | None = None,
611) -> dt.datetime | None:
612    """Parse a report timestamp in ddhhZ or ddhhmmZ format.
613
614    If time_only, assumes hhmm format with current or previous day.
615
616    This function assumes the given timestamp is within the hour threshold from current date.
617    """
618    # Format date string
619    date = date.strip("Z")
620    if not date.isdigit():
621        return None
622    if time_only:
623        if len(date) != 4:
624            return None
625        index_hour = 0
626    else:
627        if len(date) == 4:
628            date += "00"
629        if len(date) != 6:
630            return None
631        index_hour = 2
632    # Create initial guess
633    if target:
634        target = dt.datetime(target.year, target.month, target.day, tzinfo=dt.timezone.utc)
635    else:
636        target = dt.datetime.now(tz=dt.timezone.utc)
637    day = target.day if time_only else int(date[:2])
638    hour = int(date[index_hour : index_hour + 2])
639    # Handle situation where next month has less days than current month
640    # Shifted value makes sure that a month shift doesn't happen twice
641    shifted = False
642    if day > monthrange(target.year, target.month)[1]:
643        target += relativedelta(months=-1)
644        shifted = True
645    try:
646        guess = target.replace(
647            day=day,
648            hour=hour % 24,
649            minute=int(date[index_hour + 2 : index_hour + 4]) % 60,
650            second=0,
651            microsecond=0,
652        )
653    except ValueError:
654        return None
655    # Handle overflow hour
656    if hour > 23:
657        guess += dt.timedelta(days=1)
658    # Handle changing months if not already shifted
659    if not shifted:
660        hourdiff = (guess - target) / dt.timedelta(minutes=1) / 60
661        if hourdiff > hour_threshold:
662            guess += relativedelta(months=-1)
663        elif hourdiff < -hour_threshold:
664            guess += relativedelta(months=+1)
665    return guess
666
667
668def make_timestamp(
669    timestamp: str | None,
670    *,
671    time_only: bool = False,
672    target_date: dt.date | None = None,
673) -> Timestamp | None:
674    """Return a Timestamp dataclass for a report timestamp in ddhhZ or ddhhmmZ format."""
675    if not timestamp:
676        return None
677    date_obj = parse_date(timestamp, time_only=time_only, target=target_date)
678    return Timestamp(timestamp, date_obj)
679
680
681def is_runway_visibility(item: str) -> bool:
682    """Return True if the item is a runway visibility range string."""
683    return (
684        len(item) > 4
685        and item[0] == "R"
686        and (item[3] == "/" or item[4] == "/")
687        and item[1:3].isdigit()
688        and "CLRD" not in item  # R28/CLRD70 Runway State
689    )