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