avwx.current.notam

  1"""
  2A NOTAM (Notice to Air Missions) is a report detailing special events or
  3conditions affecting airport and flight operations. These can include, but are
  4in no way limitted to:
  5
  6- Runway closures
  7- Lack of radar services
  8- Rocket launches
  9- Hazard locations
 10- Airspace restrictions
 11- Construction updates
 12- Unusual aircraft activity
 13
 14NOTAMs have varius classifications and apply to certain types or size of
 15aircraft. Some apply only to IFR operations, like when an ILS is out of
 16service. Others apply only to airport operations the en route aircraft can
 17ignore.
 18
 19Every NOTAM has a start and end date and time. Additional NOTAMs may be issued
 20to update, replace, or cancel existing NOTAMs as well. Some NOTAMs may still be
 21served up to 10 days after the end date, so it's up to the developer to include
 22or filter these reports.
 23"""
 24
 25# stdlib
 26from __future__ import annotations
 27
 28import re
 29from contextlib import suppress
 30from datetime import datetime, timezone
 31
 32# library
 33from dateutil.tz import gettz
 34
 35# module
 36from avwx import exceptions
 37from avwx.current.base import Reports
 38from avwx.parsing import core
 39
 40# from avwx.service import FaaNotam
 41from avwx.static.core import SPECIAL_NUMBERS
 42from avwx.static.notam import (
 43    CODES,
 44    CONDITION,
 45    MISTYPED_SUBJECT,
 46    PURPOSE,
 47    REPORT_TYPE,
 48    SCOPE,
 49    SUBJECT,
 50    TRAFFIC_TYPE,
 51    TRANSPOSED_SUBJECT,
 52)
 53from avwx.structs import (
 54    Altitude,
 55    Code,
 56    Coord,
 57    NotamData,
 58    Number,
 59    Qualifiers,
 60    Timestamp,
 61    Units,
 62)
 63
 64# https://www.navcanada.ca/en/briefing-on-the-transition-to-icao-notam-format.pdf
 65# https://www.faa.gov/air_traffic/flight_info/aeronav/notams/media/2021-09-07_ICAO_NOTAM_101_Presentation_for_Airport_Operators.pdf
 66
 67
 68_DEP_MSG = "This method is temporarily deprecated until non-auth source can be found."
 69
 70
 71class Notams(Reports):
 72    '''
 73    The Notams class provides two ways of requesting all applicable NOTAMs in
 74    an area: airport code and coordinate. The service will fetch all reports
 75    within 10 nautical miles of the desired center point.
 76
 77    *Update methods are temporarily deprecated until non-auth source can be found.*
 78    You can change the distance by updating the `Notams.radius` member before calling `update()`.
 79
 80    ```python
 81    >>> from pprint import pprint
 82    >>> from avwx import Notams
 83    >>> from avwx.structs import Coord
 84    >>>
 85    >>> kjfk = Notams("KJFK")
 86    >>> kjfk.update()
 87    True
 88    >>> kjfk.last_updated
 89    datetime.datetime(2022, 5, 26, 0, 43, 22, 44753, tzinfo=datetime.timezone.utc)
 90    >>> print(kjfk.data[0].raw)
 91    01/113 NOTAMN
 92    Q) ZNY/QMXLC/IV/NBO/A/000/999/4038N07346W005
 93    A) KJFK
 94    B) 2101081328
 95    C) 2209301100
 96
 97    E) TWY TB BTN TERMINAL 8 RAMP AND TWY A CLSD
 98    >>> pprint(kjfk.data[0].qualifiers)
 99    Qualifiers(repr='ZNY/QMXLC/IV/NBO/A/000/999/4038N07346W005',
100            fir='ZNY',
101            subject=Code(repr='MX', value='Taxiway'),
102            condition=Code(repr='LC', value='Closed'),
103            traffic=Code(repr='IV', value='IFR and VFR'),
104            purpose=[Code(repr='N', value='Immediate'),
105                        Code(repr='B', value='Briefing'),
106                        Code(repr='O', value='Flight Operations')],
107            scope=[Code(repr='A', value='Aerodrome')],
108            lower=Number(repr='000', value=0, spoken='zero'),
109            upper=Number(repr='999', value=999, spoken='nine nine nine'),
110            coord=Coord(lat=40.38, lon=-73.46, repr='4038N07346W'),
111            radius=Number(repr='005', value=5, spoken='five'))
112    >>>
113    >>> coord = Notams(coord=Coord(lat=52, lon=-0.23))
114    >>> coord.update()
115    True
116    >>> coord.data[0].station
117    'EGSS'
118    >>> print(coord.data[0].body)
119    LONDON STANSTED ATC SURVEILLANCE MINIMUM ALTITUDE CHART - IN
120    FREQUENCY BOX RENAME ESSEX RADAR TO STANSTED RADAR.
121    UK AIP AD 2.EGSS-5-1 REFERS
122    ```
123
124    The `parse` and `from_report` methods can parse a report string if you want
125    to override the normal fetching process.
126
127    ```python
128    >>> from avwx import Notams
129    >>> report = """
130    05/295 NOTAMR
131    Q) ZNY/QMNHW/IV/NBO/A/000/999/4038N07346W005
132    A) KJFK
133    B) 2205201527
134    C) 2205271100
135
136    E) APRON TERMINAL 4 RAMP CONST WIP S SIDE TAXILANE G LGTD AND BARRICADED
137    """
138    >>> kjfk = Notams.from_report(report)
139    >>> kjfk.data[0].type
140    Code(repr='NOTAMR', value='Replace')
141    >>> kjfk.data[0].start_time
142    Timestamp(repr='2205201527', dt=datetime.datetime(2022, 5, 20, 15, 27, tzinfo=datetime.timezone.utc))
143    ```
144    '''
145
146    data: list[NotamData] | None = None  # type: ignore
147    radius: int = 10
148
149    def __init__(self, code: str | None = None, coord: Coord | None = None):
150        super().__init__(code, coord)
151        # self.service = FaaNotam("notam")
152
153    async def _post_update(self) -> None:
154        self._post_parse()
155
156    def _post_parse(self) -> None:
157        self.data, units = [], None
158        if self.raw is None:
159            return
160        for report in self.raw:
161            if "||" in report:
162                issue_text, report = report.split("||")  # noqa: PLW2901
163                issued_value = datetime.strptime(issue_text, r"%m/%d/%Y %H%M").replace(tzinfo=timezone.utc)
164                issued = Timestamp(issue_text, issued_value)
165            else:
166                issued = None
167            try:
168                data, units = parse(report, issued=issued)
169                self.data.append(data)
170            except Exception as exc:  # noqa: BLE001
171                exceptions.exception_intercept(exc, raw=report)  # type: ignore
172        if units:
173            self.units = units
174
175    @staticmethod
176    def sanitize(report: str) -> str:
177        """Sanitize a NOTAM string."""
178        return sanitize(report)
179
180    # @deprecated(_DEP_MSG)
181    def update(self, timeout: int = 10, *, disable_post: bool = False) -> bool:
182        raise NotImplementedError(_DEP_MSG)
183
184    # @deprecated(_DEP_MSG)
185    async def async_update(self, timeout: int = 10, *, disable_post: bool = False) -> bool:
186        """Async updates report data by fetching and parsing the report."""
187        raise NotImplementedError(_DEP_MSG)
188        # reports = await self.service.async_fetch(  # type: ignore
189        #     icao=self.code, coord=self.coord, radius=self.radius, timeout=timeout
190        # )
191        # self.source = self.service.root
192        # return await self._update(reports, None, disable_post=disable_post)
193
194
195# A key starts the report or follows whitespace, so "AR112H(E) " in body text is not a key
196_KEY_START = r"(?:^|(?<=\s))"
197ALL_KEYS_PATTERN = re.compile(_KEY_START + r"[A-GQ]\) ")
198MISSING_KEY_SPACE_PATTERN = re.compile(r"(^|\s)([A-GQ])\)(?=\S)")
199Q_CODE_SUBJECT_PATTERN = re.compile(r"(\bQ\)\s*[A-Z]{3,4}\s*/\s*Q)([A-Z]{2})")
200SUBJECT_CORRECTIONS = TRANSPOSED_SUBJECT | MISTYPED_SUBJECT
201REVERSED_FLIGHT_LEVEL_PATTERN = re.compile("(" + _KEY_START + r"[FG]\)\s*)(\d+)FL\b")
202# The start of an ICAO NOTAM, ie "A3230/26 NOTAMN", at the start of a line or after "("
203NOTAM_START_PATTERN = re.compile(r"(\(\s*|^)(?=[A-Z]\d{4}/\d{2}\s+NOTAM[NRC]\b)", re.MULTILINE)
204# "FL150" and the shorthand "F150" both name a flight level
205FLIGHT_LEVEL_PATTERN = re.compile(r"^FL?\d+$")
206KEY_PATTERNS = {
207    "Q": re.compile(_KEY_START + r"[A-G]\) "),
208    "A": re.compile(_KEY_START + r"[B-G]\) "),
209    "B": re.compile(_KEY_START + r"[C-G]\) "),
210    "C": re.compile(_KEY_START + r"[D-G]\) "),
211    "D": re.compile(_KEY_START + r"[E-G]\) "),
212    "E": re.compile(_KEY_START + r"[FG]\) "),
213    "F": re.compile(_KEY_START + r"G\) "),
214    # No "G"
215}
216
217
218def _rear_coord(value: str) -> Coord | None:
219    """Convert coord strings with direction characters at the end: 5126N00036W."""
220    if len(value) != 11:
221        return None
222    try:
223        lat = float(f"{value[:2]}.{value[2:4]}")
224        lon = float(f"{value[5:8]}.{value[8:10]}")
225    except ValueError:
226        return None
227    if value[4] == "S":
228        lat *= -1
229    if value[10] == "W":
230        lon *= -1
231    return Coord(lat, lon, value)
232
233
234def _split_location(
235    location: str | None,
236) -> tuple[Coord | None, Number | None]:
237    """Identify coordinate and radius from location element."""
238    if not location:
239        return None, None
240    coord, radius = None, None
241    if len(location) == 14 and location[-3:].isdigit():
242        radius = core.make_number(location[-3:])
243        location = location[:-3]
244    if len(location) == 11 and location[-1] in {"E", "W"}:
245        coord = _rear_coord(location)
246    return coord, radius
247
248
249def _header(value: str) -> tuple[str, Code | None, str | None]:
250    """Parse pre-tag headers."""
251    header = value.strip().split()
252    replaces = None
253    if len(header) == 3:
254        number, type_text, replaces = header
255    else:
256        number, type_text = header
257    report_type = Code.from_dict(type_text, REPORT_TYPE)
258    return number, report_type, replaces
259
260
261def _find_q_codes(
262    codes: list[str],
263) -> tuple[
264    Code | None,
265    list[Code],
266    list[Code],
267    str | None,
268    str | None,
269    str | None,
270]:
271    """Identify traffic, purpose, and scope codes."""
272    # The 'K' code can be both purpose and scope, but they have the same value
273    traffic, lower, upper, location = None, None, None, None
274    purpose: list[Code] = []
275    scope: list[Code] = []
276    for code in codes:
277        if not code:
278            continue
279        # Altitudes can be int or float values
280        with suppress(ValueError):
281            float(code)
282            if not lower:
283                lower = code
284            else:
285                upper = code
286            continue
287        # location will always be the longest element if available
288        if len(code) > 10:
289            location = code
290            continue
291        # Remaining elements must match known value dictionary combinations
292        if not traffic and code in TRAFFIC_TYPE:
293            traffic = Code.from_dict(code, TRAFFIC_TYPE)
294            continue
295        if not purpose:
296            purpose = Code.from_list(code, PURPOSE, exclusive=True)
297        if not scope:
298            scope = Code.from_list(code, SCOPE, exclusive=True)
299    return traffic, purpose, scope, lower, upper, location
300
301
302def _qualifiers(value: str, units: Units) -> Qualifiers:
303    """Parse the NOTAM Q) line into components."""
304    fir, q_code, *codes = (i.strip() for i in re.split("/| ", value.strip()))
305    traffic, purpose, scope, lower, upper, location = _find_q_codes(codes)
306    subject, condition = None, None
307    if q_code.startswith("Q") and len(q_code) >= 5:
308        subject = Code(q_code[1:3], "Other") if q_code[1] == "Q" else Code.from_dict(q_code[1:3], SUBJECT)
309        condition_code = q_code[3:]
310        if condition_code.startswith("XX"):
311            condition = Code("XX", (condition_code[2:] or "Unknown").strip())
312        else:
313            condition = Code.from_dict(condition_code, CONDITION, error=False)
314    coord, radius = _split_location(location)
315    return Qualifiers(
316        repr=value,
317        fir=fir,
318        subject=subject,
319        condition=condition,
320        traffic=traffic,
321        purpose=purpose,
322        scope=scope,
323        lower=make_altitude(lower, units),
324        upper=make_altitude(upper, units),
325        coord=coord,
326        radius=radius,
327    )
328
329
330def _tz_offset_for(name: str | None) -> timezone | None:
331    """Generate a timezone from tz string name."""
332    if not name:
333        return None
334    if tz := gettz(name):  # noqa: SIM102
335        if offset := tz.utcoffset(datetime.now(timezone.utc)):
336            return timezone(offset)
337    return None
338
339
340def make_year_timestamp(
341    value: str,
342    repr: str,  # noqa: A002
343    tzname: str | None = None,
344) -> Timestamp | Code | None:
345    """Convert NOTAM timestamp which includes year and month."""
346    values = value.strip().split()
347    if not values:
348        return None
349    value = values[0]
350    if code := CODES.get(value):
351        return Code(value, code)
352    tz = _tz_offset_for(tzname) or timezone.utc
353    # Keep the original value when it is not a real time, like the FAA checklist
354    # placeholder hour "2604018000" or an originator's 29 Feb in a non-leap year
355    try:
356        raw = datetime.strptime(value[:10], r"%y%m%d%H%M")  # noqa: DTZ007
357    except ValueError:
358        return Timestamp(repr, None)
359    date = datetime(raw.year, raw.month, raw.day, raw.hour, raw.minute, tzinfo=tz)
360    return Timestamp(repr, date)
361
362
363def parse_linked_times(start: str, end: str) -> tuple[Timestamp | Code | None, Timestamp | Code | None]:
364    """Parse start and end times sharing any found timezone."""
365    start, end = start.strip(), end.strip()
366    start_raw, end_raw, tzname = start, end, None
367    if len(start) > 10:
368        start, tzname = start[:-3], start[-3:]
369    if len(end) > 10:
370        end, tzname = end[:-3], end[-3:]
371    return make_year_timestamp(start, start_raw, tzname), make_year_timestamp(end, end_raw, tzname)
372
373
374def make_altitude(value: str | None, units: Units) -> Altitude | None:
375    """Parse NOTAM altitudes.
376
377    A value is only a flight level when it says so, ie "FL150" or "F150" in a G) line.
378
379    ICAO Annex 15 defines the Q) line limits as flight levels, but they are not treated
380    as such here. Producers fill them with the F) and G) values rounded up to the next
381    hundred feet against those items' own datum, which is AMSL far more often than it is
382    a pressure altitude, and the spec's own "000"/"999" default means the subject carries
383    no height information at all rather than FL000 to FL999.
384    """
385    if not value:
386        return None
387    trimmed = value.split()[0].strip(" .")
388    if not trimmed:
389        return None
390    if "(" in trimmed:
391        trimmed = trimmed[trimmed.find("(") + 1 :]
392    is_flight_level = FLIGHT_LEVEL_PATTERN.match(trimmed) is not None
393    if not (is_flight_level or trimmed in SPECIAL_NUMBERS or trimmed[0].isdigit()):
394        return None
395    number = core.make_altitude(trimmed, units, repr=value)[0]
396    if number is None:
397        return None
398    return Altitude(
399        repr=number.repr,
400        value=number.value,
401        spoken=number.spoken,
402        flight_level=is_flight_level,
403    )
404
405
406def parse(report: str, issued: Timestamp | None = None) -> tuple[NotamData, Units]:
407    """Parse NOTAM report string."""
408    units = Units.international()
409    sanitized = sanitize(report)
410    qualifiers, station, start_time, end_time = None, None, None, None
411    body, number, replaces, report_type = "", None, None, None
412    schedule, lower, upper, text = None, None, None, sanitized
413    # US domestic NOTAMs ("!DCA 09/123 ...") have no ICAO items, but their text can still
414    # look like one, ie "(35FT N TWY A) 98FT"
415    match = None if text.startswith("!") else ALL_KEYS_PATTERN.search(text)
416    # Type and number here
417    if match and match.start() > 0:
418        number, report_type, replaces = _header(text[: match.start()])
419    start_text, end_text = "", ""
420    while match:
421        tag = match.group()[0]
422        text = text[match.end() :]
423        try:
424            match = KEY_PATTERNS[tag].search(text)
425        except KeyError:
426            match = None
427        item = (text[: match.start()] if match else text).strip()
428        if tag == "Q":
429            qualifiers = _qualifiers(item, units)
430        elif tag == "A":
431            station = item
432        elif tag == "B":
433            start_text = item
434        elif tag == "C":
435            end_text = item
436        elif tag == "D":
437            schedule = item
438        elif tag == "E":
439            body = item
440        elif tag == "F":
441            lower = make_altitude(item, units)
442        elif tag == "G":
443            upper = make_altitude(item, units)
444    start_time, end_time = parse_linked_times(start_text, end_text)
445    return (
446        NotamData(
447            raw=report,
448            sanitized=sanitized,
449            station=station,
450            time=issued,
451            remarks=None,
452            number=number,
453            replaces=replaces,
454            type=report_type,
455            qualifiers=qualifiers,
456            start_time=start_time,
457            end_time=end_time,
458            schedule=schedule,
459            body=body,
460            lower=lower,
461            upper=upper,
462        ),
463        units,
464    )
465
466
467def _unwrap(report: str) -> str:
468    """Return the NOTAM without an enclosing AFTN message.
469
470    Relayed NOTAMs can arrive as the whole AFTN message, ie a heading like
471    "ORIGINAL NOTAM" or "RQR LTAA A1750/26" followed by the NOTAM in parentheses.
472    This anchors on the NOTAM's own number and type rather than on any heading.
473    """
474    if report.startswith("!"):
475        return report
476    match = NOTAM_START_PATTERN.search(report)
477    if match is None or (match.start() == 0 and not match.group(1)):
478        return report
479    body = report[match.end() :].strip()
480    # The closing parenthesis can be missing, so only drop a trailing one left unmatched
481    # once item keys like "Q)" are set aside
482    if match.group(1) and body.endswith(")"):
483        text = ALL_KEYS_PATTERN.sub("", MISSING_KEY_SPACE_PATTERN.sub(r"\1\2) ", body))
484        if text.count("(") < text.count(")"):
485            body = body[:-1].rstrip()
486    return body
487
488
489def sanitize(report: str) -> str:
490    """Retun a sanitized report ready for parsing."""
491    # Relayed NOTAMs can use CRLF or bare CR line endings
492    report = _unwrap(report.replace("\r\n", "\n").replace("\r", "\n").strip())
493    # Some sources omit the space after a key, ie "E)TWY CLSD" instead of "E) TWY CLSD",
494    # which the key patterns above require. Only repair a key that starts a line or
495    # follows whitespace so keys quoted inside body text are left alone.
496    report = MISSING_KEY_SPACE_PATTERN.sub(r"\1\2) ", report)
497    # Some originators swap or mistype the Q-code subject letters, ie QXMLC for QMXLC
498    report = Q_CODE_SUBJECT_PATTERN.sub(
499        lambda m: m.group(1) + SUBJECT_CORRECTIONS.get(m.group(2), m.group(2)), report, count=1
500    )
501    # Some originators give a flight level limit number first, ie "G) 120FL" for "G) FL120"
502    return REVERSED_FLIGHT_LEVEL_PATTERN.sub(r"\1FL\2", report)