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