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