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