avwx.current.metar

  1"""
  2A METAR (Meteorological Aerodrome Report) is the surface weather observed at
  3most controlled (and some uncontrolled) airports. They are updated once per
  4hour or when conditions change enough to warrant an update, and the
  5observations are valid for one hour after the report was issued or until the
  6next report is issued.
  7"""
  8
  9# pylint: disable=invalid-overridden-method
 10
 11# stdlib
 12from contextlib import suppress
 13from datetime import date, datetime, timedelta, timezone
 14from typing import List, Tuple, Optional
 15
 16# module
 17from avwx.current.base import Report, get_wx_codes
 18from avwx.parsing import core, remarks, speech, summary
 19from avwx.parsing.sanitization.metar import clean_metar_list, clean_metar_string
 20from avwx.parsing.translate.metar import translate_metar
 21from avwx.service import NOAA
 22from avwx.static.core import FLIGHT_RULES
 23from avwx.static.metar import METAR_RMK
 24from avwx.station import uses_na_format, valid_station
 25from avwx.structs import (
 26    Code,
 27    MetarData,
 28    MetarTrans,
 29    Number,
 30    RemarksData,
 31    RunwayVisibility,
 32    Sanitization,
 33    Units,
 34)
 35
 36
 37class Metar(Report):
 38    """
 39    The Metar class offers an object-oriented approach to managing METAR data
 40    for a single station.
 41
 42    Below is typical usage for fetching and pulling METAR data for KJFK.
 43
 44    ```python
 45    >>> from avwx import Metar
 46    >>> kjfk = Metar("KJFK")
 47    >>> kjfk.station.name
 48    'John F Kennedy International Airport'
 49    >>> kjfk.update()
 50    True
 51    >>> kjfk.last_updated
 52    datetime.datetime(2018, 3, 4, 23, 36, 6, 62376)
 53    >>> kjfk.raw
 54    'KJFK 042251Z 32023G32KT 10SM BKN060 04/M08 A3008 RMK AO2 PK WND 32032/2251 SLP184 T00441078'
 55    >>> kjfk.data.flight_rules
 56    'VFR'
 57    >>> kjfk.translations.remarks
 58    {'AO2': 'Automated with precipitation sensor', 'SLP184': 'Sea level pressure: 1018.4 hPa', 'T00441078': 'Temperature 4.4°C and dewpoint -7.8°C'}
 59    ```
 60
 61    The `parse` and `from_report` methods can parse a report string if you want
 62    to override the normal fetching process. Here's an example of a really bad
 63    day.
 64
 65    ```python
 66    >>> from avwx import Metar
 67    >>> report = 'KSFO 031254Z 36024G55KT 320V040 1/8SM R06/0200D +TS VCFC OVC050 BKN040TCU 14/10 A2978 RMK AIRPORT CLOSED'
 68    >>> ksfo = Metar.from_report(report)
 69    True
 70    >>> ksfo.station.city
 71    'San Francisco'
 72    >>> ksfo.last_updated
 73    datetime.datetime(2018, 3, 4, 23, 54, 4, 353757, tzinfo=datetime.timezone.utc)
 74    >>> ksfo.data.flight_rules
 75    'LIFR'
 76    >>> ksfo.translations.clouds
 77    'Broken layer at 4000ft (Towering Cumulus), Overcast layer at 5000ft - Reported AGL'
 78    >>> ksfo.summary
 79    'Winds N-360 (variable 320 to 040) at 24kt gusting to 55kt, Vis 0.125sm, Temp 14C, Dew 10C, Alt 29.78inHg, Heavy Thunderstorm, Vicinity Funnel Cloud, Broken layer at 4000ft (Towering Cumulus), Overcast layer at 5000ft'
 80    ```
 81    """
 82
 83    data: Optional[MetarData] = None
 84    translations: Optional[MetarTrans] = None
 85
 86    async def _pull_from_default(self) -> None:
 87        """Checks for a more recent report from NOAA. Only sync"""
 88        service = NOAA(self.__class__.__name__.lower())
 89        if self.code is None:
 90            return
 91        report = await service.async_fetch(self.code)
 92        if report is not None:
 93            data, units, sans = parse(self.code, report, self.issued)
 94            if not data or data.time is None or data.time.dt is None:
 95                return
 96            if (
 97                not self.data
 98                or self.data.time is None
 99                or self.data.time.dt is None
100                or data.time.dt > self.data.time.dt
101            ):
102                self.data, self.units, self.sanitization = data, units, sans
103                self.source = service.root
104
105    @property
106    def _should_check_default(self) -> bool:
107        """Returns True if pulled from regional source and potentially out of date"""
108        if isinstance(self.service, NOAA) or self.source is None:
109            return False
110
111        if self.data is None or self.data.time is None or self.data.time.dt is None:
112            return True
113        time_since = datetime.now(tz=timezone.utc) - self.data.time.dt
114        return time_since > timedelta(minutes=90)
115
116    def _calculate_altitudes(self) -> None:
117        """Adds the pressure and density altitudes to data if all fields are available"""
118        if self.data is None or self.station is None or self.units is None:
119            return
120        # Select decimal temperature if available
121        temp = self.data.temperature
122        if self.data.remarks_info is not None:
123            temp = self.data.remarks_info.temperature_decimal or temp
124        alt = self.data.altimeter
125        if temp is None or temp.value is None or alt is None or alt.value is None:
126            return
127        elev = self.station.elevation_ft
128        if elev is None:
129            return
130        self.data.pressure_altitude = core.pressure_altitude(
131            alt.value, elev, self.units.altimeter
132        )
133        self.data.density_altitude = core.density_altitude(
134            alt.value, temp.value, elev, self.units
135        )
136
137    async def _post_update(self) -> None:
138        if self.code is None or self.raw is None:
139            return
140        self.data, self.units, self.sanitization = parse(
141            self.code, self.raw, self.issued
142        )
143        if self._should_check_default:
144            await self._pull_from_default()
145        if self.data is None or self.units is None:
146            return
147        self._calculate_altitudes()
148        self.translations = translate_metar(self.data, self.units)
149
150    def _post_parse(self) -> None:
151        if self.code is None or self.raw is None:
152            return
153        self.data, self.units, self.sanitization = parse(
154            self.code, self.raw, self.issued
155        )
156        if self.data is None or self.units is None:
157            return
158        self._calculate_altitudes()
159        self.translations = translate_metar(self.data, self.units)
160
161    @staticmethod
162    def sanitize(report: str) -> str:
163        """Sanitizes a METAR string"""
164        return sanitize(report)[0]
165
166    @property
167    def summary(self) -> Optional[str]:
168        """Condensed report summary created from translations"""
169        if not self.translations:
170            self.update()
171        return None if self.translations is None else summary.metar(self.translations)
172
173    @property
174    def speech(self) -> Optional[str]:
175        """Report summary designed to be read by a text-to-speech program"""
176        if not self.data:
177            self.update()
178        if self.data is None or self.units is None:
179            return None
180        return speech.metar(self.data, self.units)
181
182
183def get_remarks(txt: str) -> Tuple[List[str], str]:
184    """Returns the report split into components and the remarks string
185
186    Remarks can include items like RMK and on, NOSIG and on, and BECMG and on
187    """
188    txt = txt.replace("?", "").strip()
189    # First look for Altimeter in txt
190    alt_index = len(txt) + 1
191    for item in [" A2", " A3", " Q1", " Q0", " Q9"]:
192        index = txt.find(item)
193        if len(txt) - 6 > index > -1 and txt[index + 2 : index + 6].isdigit():
194            alt_index = index
195    # Then look for earliest remarks 'signifier'
196    sig_index = core.find_first_in_list(txt, METAR_RMK)
197    if sig_index == -1:
198        sig_index = len(txt) + 1
199    if sig_index > alt_index > -1:
200        return txt[: alt_index + 6].strip().split(), txt[alt_index + 7 :]
201    if alt_index > sig_index > -1:
202        return txt[:sig_index].strip().split(), txt[sig_index + 1 :]
203    return txt.strip().split(), ""
204
205
206_RVR_CODES = {
207    "M": "less than",
208    "A": "greater than",
209    "P": "greater than",
210    "U": "increasing",
211    "I": "increasing",
212    "D": "decreasing",
213    "F": "decreasing",
214    "R": "decreasing",
215    "N": "no change",
216    "V": "variable",
217}
218
219
220def _parse_rvr_number(value: str) -> Optional[Number]:
221    if not value:
222        return None
223    raw, prefix = value, None
224    with suppress(KeyError):
225        prefix = _RVR_CODES[value[0]]
226        value = value[1:]
227    number = core.make_number(value, raw)
228    if number is not None and prefix is not None:
229        number.spoken = f"{prefix} {number.spoken}"
230        number.value = None
231    return number
232
233
234def parse_runway_visibility(value: str) -> RunwayVisibility:
235    """Parse a runway visibility range string"""
236    raw, trend = value, None
237    # TODO: update to check and convert units post visibility parse
238    value = value.replace("FT", "")
239    with suppress(KeyError):
240        trend = Code(value[-1], _RVR_CODES[value[-1]])
241        value = value[:-1]
242    runway, value, *_ = value[1:].split("/")
243    if value:
244        possible_numbers = [_parse_rvr_number(n) for n in value.split("V")]
245        numbers = [n for n in possible_numbers if n is not None]
246        visibility = numbers.pop() if len(numbers) == 1 else None
247    else:
248        visibility, numbers = None, []
249    return RunwayVisibility(
250        repr=raw,
251        runway=runway,
252        visibility=visibility,
253        variable_visibility=numbers,
254        trend=trend,
255    )
256
257
258def get_runway_visibility(data: List[str]) -> Tuple[List[str], List[RunwayVisibility]]:
259    """Returns the report list and the remove runway visibility list"""
260    runway_vis = [
261        parse_runway_visibility(data.pop(i))
262        for i, item in reversed(list(enumerate(data)))
263        if core.is_runway_visibility(item)
264    ]
265    runway_vis.sort(key=lambda x: x.runway)
266    return data, runway_vis
267
268
269def parse_altimeter(value: str) -> Optional[Number]:
270    """Parse an altimeter string into a Number"""
271    if not value or len(value) < 4:
272        return None
273    # QNH3003INS
274    if len(value) >= 7 and value.endswith("INS"):
275        return core.make_number(f"{value[-7:-5]}.{value[-5:-3]}", value, literal=True)
276    number = value.replace(".", "")
277    # Q1000/10
278    if "/" in number:
279        number = number.split("/")[0]
280    if number.startswith("QNH"):
281        number = f"Q{number[1:]}"
282    if not (len(number) in {4, 5} and number[-4:].isdigit()):
283        return None
284    number = number.lstrip("AQ")
285    if number[0] in ("2", "3"):
286        number = f"{number[:2]}.{number[2:]}"
287    elif number[0] not in ("0", "1"):
288        return None
289    return core.make_number(number, value, number, literal=True)
290
291
292def get_altimeter(
293    data: List[str], units: Units, version: str = "NA"
294) -> Tuple[List[str], Optional[Number]]:
295    """Returns the report list and the removed altimeter item
296
297    Version is 'NA' (North American / default) or 'IN' (International)
298    """
299    values: List[Number] = []
300    for _ in range(2):
301        if not data:
302            break
303        value = parse_altimeter(data[-1])
304        if value is None:
305            break
306        values.append(value)
307        data.pop(-1)
308    if not values:
309        return data, None
310    values.sort(key=lambda x: x.value or 0)
311    altimeter = values[0 if version == "NA" else -1]
312    if altimeter.value is not None:
313        units.altimeter = "inHg" if altimeter.value < 100 else "hPa"
314    return data, altimeter
315
316
317def get_temp_and_dew(
318    data: List[str],
319) -> Tuple[List[str], Optional[Number], Optional[Number]]:
320    """Returns the report list and removed temperature and dewpoint strings"""
321    for i, item in reversed(list(enumerate(data))):
322        if "/" in item:
323            # ///07
324            if item[0] == "/":
325                item = "/" + item.lstrip("/")
326            # 07///
327            elif item[-1] == "/":
328                item = item.rstrip("/") + "/"
329            tempdew = item.split("/")
330            if len(tempdew) != 2:
331                continue
332            valid = True
333            for j, temp in enumerate(tempdew):
334                if temp in ["MM", "XX"]:
335                    tempdew[j] = ""
336                elif not core.is_possible_temp(temp):
337                    valid = False
338                    break
339            if valid:
340                data.pop(i)
341                temp, dew = tempdew
342                return data, core.make_number(temp), core.make_number(dew)
343    return data, None, None
344
345
346def get_relative_humidity(
347    temperature: Optional[Number],
348    dewpoint: Optional[Number],
349    remarks_info: Optional[RemarksData],
350    units: Units,
351) -> Optional[float]:
352    """Calculates relative humidity from preferred temperature and dewpoint"""
353    if remarks_info is not None:
354        temp = remarks_info.temperature_decimal or temperature
355        dew = remarks_info.dewpoint_decimal or dewpoint
356    else:
357        temp = temperature
358        dew = dewpoint
359    if temp is None or temp.value is None:
360        return None
361    if dew is None or dew.value is None:
362        return None
363    return core.relative_humidity(temp.value, dew.value, units.temperature)
364
365
366def sanitize(report: str) -> Tuple[str, str, List[str], Sanitization]:
367    """Returns a sanitized report, remarks, and elements ready for parsing"""
368    sans = Sanitization()
369    clean = clean_metar_string(report, sans)
370    data, remark_str = get_remarks(clean)
371    data = core.dedupe(data)
372    data = clean_metar_list(data, sans)
373    clean = " ".join(data)
374    if remark_str:
375        clean += f" {remark_str}"
376    return clean, remark_str, data, sans
377
378
379def parse(
380    station: str,
381    report: str,
382    issued: Optional[date] = None,
383    use_na: Optional[bool] = None,
384) -> Tuple[Optional[MetarData], Optional[Units], Optional[Sanitization]]:
385    """Returns MetarData and Units dataclasses with parsed data and their associated units"""
386    valid_station(station)
387    if not report:
388        return None, None, None
389    if use_na is None:
390        use_na = uses_na_format(station[:2])
391    parser = parse_na if use_na else parse_in
392    return parser(report, issued)
393
394
395def parse_na(
396    report: str, issued: Optional[date] = None
397) -> Tuple[MetarData, Units, Sanitization]:
398    """Parser for the North American METAR variant"""
399    # pylint: disable=too-many-locals
400    units = Units.north_american()
401    sanitized, remarks_str, data, sans = sanitize(report)
402    data, station, time = core.get_station_and_time(data)
403    data, runway_visibility = get_runway_visibility(data)
404    data, clouds = core.get_clouds(data)
405    (
406        data,
407        wind_direction,
408        wind_speed,
409        wind_gust,
410        wind_variable_direction,
411    ) = core.get_wind(data, units)
412    data, altimeter = get_altimeter(data, units, "NA")
413    data, visibility = core.get_visibility(data, units)
414    data, temperature, dewpoint = get_temp_and_dew(data)
415    condition = core.get_flight_rules(visibility, core.get_ceiling(clouds))
416    other, wx_codes = get_wx_codes(data)
417    remarks_info = remarks.parse(remarks_str)
418    humidity = get_relative_humidity(temperature, dewpoint, remarks_info, units)
419    struct = MetarData(
420        altimeter=altimeter,
421        clouds=clouds,
422        dewpoint=dewpoint,
423        flight_rules=FLIGHT_RULES[condition],
424        other=other,
425        raw=report,
426        relative_humidity=humidity,
427        remarks_info=remarks_info,
428        remarks=remarks_str,
429        runway_visibility=runway_visibility,
430        sanitized=sanitized,
431        station=station,
432        temperature=temperature,
433        time=core.make_timestamp(time, target_date=issued),
434        visibility=visibility,
435        wind_direction=wind_direction,
436        wind_gust=wind_gust,
437        wind_speed=wind_speed,
438        wind_variable_direction=wind_variable_direction,
439        wx_codes=wx_codes,
440    )
441    return struct, units, sans
442
443
444def parse_in(
445    report: str, issued: Optional[date] = None
446) -> Tuple[MetarData, Units, Sanitization]:
447    """Parser for the International METAR variant"""
448    # pylint: disable=too-many-locals
449    units = Units.international()
450    sanitized, remarks_str, data, sans = sanitize(report)
451    data, station, time = core.get_station_and_time(data)
452    data, runway_visibility = get_runway_visibility(data)
453    if "CAVOK" not in data:
454        data, clouds = core.get_clouds(data)
455    (
456        data,
457        wind_direction,
458        wind_speed,
459        wind_gust,
460        wind_variable_direction,
461    ) = core.get_wind(data, units)
462    data, altimeter = get_altimeter(data, units, "IN")
463    if "CAVOK" in data:
464        visibility = core.make_number("CAVOK")
465        clouds = []
466        data.remove("CAVOK")
467    else:
468        data, visibility = core.get_visibility(data, units)
469    data, temperature, dewpoint = get_temp_and_dew(data)
470    condition = core.get_flight_rules(visibility, core.get_ceiling(clouds))
471    other, wx_codes = get_wx_codes(data)
472    remarks_info = remarks.parse(remarks_str)
473    humidity = get_relative_humidity(temperature, dewpoint, remarks_info, units)
474    struct = MetarData(
475        altimeter=altimeter,
476        clouds=clouds,
477        dewpoint=dewpoint,
478        flight_rules=FLIGHT_RULES[condition],
479        other=other,
480        raw=report,
481        relative_humidity=humidity,
482        remarks_info=remarks_info,
483        remarks=remarks_str,
484        runway_visibility=runway_visibility,
485        sanitized=sanitized,
486        station=station,
487        temperature=temperature,
488        time=core.make_timestamp(time, target_date=issued),
489        visibility=visibility,
490        wind_direction=wind_direction,
491        wind_gust=wind_gust,
492        wind_speed=wind_speed,
493        wind_variable_direction=wind_variable_direction,
494        wx_codes=wx_codes,
495    )
496    return struct, units, sans