avwx.current.airsigmet

  1"""
  2A SIGMET (Significant Meteorological Information) is a weather advisory for the
  3safety of all aircraft. They are divided into:
  4
  5- Convective - thunderstorms, hail, and cyclones
  6- Non-Convective - turbulence, icing, dust clouds, volcanic activity, and radiation
  7
  8An AIRMET (Airman's Meteorological Information) is a weather advisory for
  9smaller aircraft or VFR navigation. They are divided into:
 10
 11- Sierra - IFR conditions like low ceilings and mountain obscuration
 12- Tango - turbulence and high surface winds
 13- Zulu - icing and freezing levels
 14
 15Both types share a similar report format and therefore are combined into a
 16single handling class. The `Bulletin` and weather type can be used to classify
 17each as a SIGMET or AIRMET for filtering purposes.
 18"""
 19
 20# stdlib
 21from __future__ import annotations
 22
 23import asyncio as aio
 24import re
 25from contextlib import suppress
 26from datetime import date, datetime, timezone
 27from itertools import chain
 28from typing import TypeAlias
 29
 30# library
 31from geopy.distance import distance as geo_distance  # type: ignore
 32
 33# module
 34from avwx import exceptions
 35from avwx.base import AVWXBase
 36from avwx.exceptions import MissingExtraModule
 37from avwx.flight_path import to_coordinates
 38from avwx.load_utils import LazyLoad
 39from avwx.parsing import core
 40from avwx.service.bulk import NoaaBulk, NoaaIntl, Service
 41from avwx.static.airsigmet import BULLETIN_TYPES, INTENSITY, WEATHER_TYPES
 42from avwx.static.core import CARDINAL_DEGREES, CARDINALS
 43from avwx.structs import (
 44    AirSigmetData,
 45    AirSigObservation,
 46    Bulletin,
 47    Code,
 48    Coord,
 49    Movement,
 50    Number,
 51    Timestamp,
 52    Units,
 53)
 54
 55try:
 56    from shapely.geometry import LineString
 57except ModuleNotFoundError:
 58    LineString = TypeAlias  # type: ignore
 59
 60
 61class AirSigmet(AVWXBase):
 62    """
 63    In addition to the manager, you can use the `avwx.AirSigmet` class like any
 64    other report when you supply the report string via `parse` or
 65    `from_report`.
 66
 67    ```python
 68    >>> from avwx import AirSigmet
 69    >>> report = 'WSPR31 SPJC 270529 SPIM SIGMET 3 VALID 270530/270830 SPJC- SPIM LIMA FIR EMBD TS OBS AT 0510Z NE OF LINE S0406 W07103 - S0358 W07225 - S0235 W07432 - S0114 W07503 TOP FL410 MOV SW NC='
 70    >>> sigmet = AirSigmet.from_report(report)
 71    True
 72    >>> sigmet.last_updated
 73    datetime.datetime(2022, 3, 27, 6, 29, 33, 300935, tzinfo=datetime.timezone.utc)
 74    >>> sigmet.data.observation.coords
 75    [Coord(lat=-4.06, lon=-71.03, repr='S0406 W07103'),
 76    Coord(lat=-3.58, lon=-72.25, repr='S0358 W07225'),
 77    Coord(lat=-2.35, lon=-74.32, repr='S0235 W07432'),
 78    Coord(lat=-1.14, lon=-75.03, repr='S0114 W07503')]
 79    >>> sigmet.data.observation.intensity
 80    Code(repr='NC', value='No change')
 81    >>> sigmet.data.observation.ceiling
 82    Number(repr='FL410', value=410, spoken='flight level four one zero')
 83    ```
 84    """
 85
 86    data: AirSigmetData | None = None
 87
 88    def _post_parse(self) -> None:
 89        if self.raw:
 90            self.data, self.units = parse(self.raw, self.issued)
 91
 92    @staticmethod
 93    def sanitize(report: str) -> str:
 94        """Sanitizes the report string"""
 95        return sanitize(report)
 96
 97    def intersects(self, path: LineString) -> bool:
 98        """Returns True if the report area intersects a flight path"""
 99        if LineString is None:
100            extra = "shape"
101            raise MissingExtraModule(extra)
102        if not self.data:
103            return False
104        for data in (self.data.observation, self.data.forecast):
105            if data:
106                poly = data.poly
107                if poly and path.intersects(poly):
108                    return True
109        return False
110
111    def contains(self, coord: Coord) -> bool:
112        """Returns True if the report area contains a coordinate"""
113        if not self.data:
114            return False
115        for data in (self.data.observation, self.data.forecast):
116            if data:
117                poly = data.poly
118                if poly and coord.point.within(poly):
119                    return True
120        return False
121
122
123class AirSigManager:
124    """
125    Because of the global nature of these report types, we don't initialize a
126    report class with a station ident like the other report types. Instead, we
127    use a class to manage and update the list of all active SIGMET and AIRMET
128    reports.
129
130    ```python
131    >>> from avwx import AirSigManager
132    >>> from avwx.structs import Coord
133    >>> manager = AirSigManager()
134    >>> manager.update()
135    True
136    >>> manager.last_updated
137    datetime.datetime(2022, 3, 27, 5, 54, 21, 516741, tzinfo=datetime.timezone.utc)
138    >>> len(manager.reports)
139    113
140    >>> len(manager.contains(Coord(lat=33.12, lon=-105)))
141    5
142    >>> manager.reports[0].data.bulletin.type
143    Code(repr='WA', value='airmet')
144    >>> manager.reports[0].data.type
145    'AIRMET SIERRA FOR IFR AND MTN OBSCN'
146    ```
147    """
148
149    _services: list[Service]
150    _raw: list[tuple[str, str | None]]
151    last_updated: datetime | None = None
152    raw: list[str]
153    reports: list[AirSigmet] | None = None
154
155    def __init__(self):  # type: ignore
156        self._services = [NoaaBulk("airsigmet"), NoaaIntl("airsigmet")]
157        self._raw, self.raw = [], []
158
159    async def _update(self, index: int, timeout: int) -> list[tuple[str, str | None]]:
160        source = self._services[index].root
161        reports = await self._services[index].async_fetch(timeout=timeout)  # type: ignore
162        raw: list[tuple[str, str | None]] = [(report, source) for report in reports if report]
163        return raw
164
165    def update(self, timeout: int = 10, *, disable_post: bool = False) -> bool:
166        """Updates fetched reports and returns whether they've changed"""
167        return aio.run(self.async_update(timeout, disable_post=disable_post))
168
169    async def async_update(self, timeout: int = 10, *, disable_post: bool = False) -> bool:
170        """Updates fetched reports and returns whether they've changed"""
171        coros = [self._update(i, timeout) for i in range(len(self._services))]
172        data = await aio.gather(*coros)
173        raw = list(chain.from_iterable(data))
174        reports = [i[0] for i in raw]
175        if raw == self._raw:
176            return False
177        self._raw, self.raw = raw, reports
178        self.last_updated = datetime.now(tz=timezone.utc)
179        # Parse reports if not disabled
180        if not disable_post:
181            parsed = []
182            for report, source in raw:
183                try:
184                    if obj := AirSigmet.from_report(report):
185                        obj.source = source
186                        parsed.append(obj)
187                except Exception as exc:  # noqa: BLE001
188                    exceptions.exception_intercept(exc, raw={"report": report})
189            self.reports = parsed
190        return True
191
192    def along(self, coords: list[Coord]) -> list[AirSigmet]:
193        """Returns available reports the intersect a flight path"""
194        if LineString is None:
195            extra = "shape"
196            raise MissingExtraModule(extra)
197        if self.reports is None:
198            return []
199        path = LineString([c.pair for c in coords])
200        return [r for r in self.reports if r.intersects(path)]
201
202    def contains(self, coord: Coord) -> list[AirSigmet]:
203        """Returns available reports that contain a coordinate"""
204        if self.reports is None:
205            return []
206        return [r for r in self.reports if r.contains(coord)]
207
208
209# N1429 W09053 - N1427 W09052 - N1411 W09139 - N1417 W09141
210_COORD_PATTERN = re.compile(r"\b[NS]\d{4} [EW]\d{5}\b( -)?")
211
212# FROM 60NW ISN-INL-TVC-GIJ-UIN-FSD-BIL-60NW ISN
213# FROM 70SSW ISN TO 20NNW FAR TO 70E DLH TO 40SE EAU TO 80SE RAP TO 40NNW BFF TO 70SSW
214_NAVAID_PATTERN = re.compile(r"\b(\d{1,3}[NESW]{1,3} [A-z]{3}-?\b)|((-|(TO )|(FROM ))[A-z]{3}\b)")
215
216# N OF N2050 AND S OF N2900
217_LATTERAL_PATTERN = re.compile(r"\b([NS] OF [NS]\d{2,4})|([EW] OF [EW]\d{3,5})( AND)?\b")
218
219NAVAIDS = LazyLoad("navaids")
220
221# Used to assist parsing after sanitized. Removed after parse
222_FLAGS = {
223    "...": " <elip> ",
224    "..": " <elip> ",
225    ". ": " <break> ",
226    "/VIS ": " <vis> VIS ",
227}
228
229
230def _parse_prep(report: str) -> list[str]:
231    """Prepares sanitized string by replacing elements with flags"""
232    report = report.rstrip(".")
233    for key, val in _FLAGS.items():
234        report = report.replace(key, val)
235    return report.split()
236
237
238def _clean_flags(data: list[str]) -> list[str]:
239    return [i for i in data if i[0] != "<"]
240
241
242def _bulletin(value: str) -> Bulletin:
243    # if len(value) != 6:
244    #     return None
245    type_key = value[:2]
246    return Bulletin(
247        repr=value,
248        type=Code(repr=type_key, value=BULLETIN_TYPES[type_key]),
249        country=value[2:4],
250        number=int(value[4:]),
251    )
252
253
254def _header(data: list[str]) -> tuple[list[str], Bulletin, str, str, str | None]:
255    bulletin = _bulletin(data[0])
256    correction, end = (data[3], 4) if len(data[3]) == 3 else (None, 3)
257    return data[end:], bulletin, data[1], data[2], correction
258
259
260def _spacetime(
261    data: list[str],
262) -> tuple[list[str], str, str, str | None, str, str | None]:
263    area = data.pop(0)
264    # Skip airmet type + time repeat
265    if data[0] == "WA" and data[1].isdigit():
266        data = data[2:]
267        area = area[:-1]  # Remove type label from 3-letter ident
268    valid_index = data.index("VALID")
269    report_type = " ".join(data[:valid_index])
270    data = data[valid_index + 1 :]
271    if data[0] == "UNTIL":
272        start_time = None
273        end_time = data[1]
274        data = data[2:]
275    else:
276        target = "-" if "-" in data[0] else "/"
277        start_time, end_time = data.pop(0).split(target)
278    # KMCO- ORL FIR
279    if data[0][-1] == "-":
280        station = data.pop(0)[:-1]
281    # KMCO - KMCO
282    elif data[1] == "-" and len(data[0]) == 4:
283        station = data.pop(0)
284        data.pop(0)
285    else:
286        station = None
287    return data, area, report_type, start_time, end_time, station
288
289
290def _first_index(data: list[str], *targets: str) -> int:
291    for target in targets:
292        with suppress(ValueError):
293            return data.index(target)
294    return -1
295
296
297def _region(data: list[str]) -> tuple[list[str], str]:
298    # FIR/CTA region name
299    # Or non-standard name using lookahead Ex: FL CSTL WTRS FROM 100SSW
300    name_end = _first_index(data, "FIR", "CTA") + 1 or _first_index(data, "FROM")
301    # State list
302    if not name_end:
303        for item in data:
304            if len(item) == 2:
305                name_end += 1
306            else:
307                break
308    name = " ".join(data[:name_end])
309    return data[name_end:], name
310
311
312def _time(data: list[str], issued: date | None = None) -> tuple[list[str], Timestamp | None, Timestamp | None]:
313    """Extracts the start and/or end time based on a couple starting elements"""
314    index = _first_index(data, "AT", "FCST", "UNTIL", "VALID", "OUTLOOK", "OTLK")
315    if index == -1:
316        return data, None, None
317    start_item = data.pop(index)
318    start, end, observed = None, None, None
319    if "-" in data[index]:
320        start_item, end_item = data.pop(index).split("-")
321        start = core.make_timestamp(start_item, time_only=len(start_item) < 6, target_date=issued)
322        end = core.make_timestamp(end_item, time_only=len(end_item) < 6, target_date=issued)
323    elif len(data[index]) >= 4 and data[index][:4].isdigit():
324        observed = core.make_timestamp(data.pop(index), time_only=True, target_date=issued)
325        if index > 0 and data[index - 1] == "OBS":
326            data.pop(index - 1)
327    for remv in ("FCST", "OUTLOOK", "OTLK", "VALID"):
328        with suppress(ValueError):
329            data.remove(remv)
330    if observed:
331        if start_item in ("UNTIL", "VALID"):
332            end = observed
333        else:
334            start = observed
335    return data, start, end
336
337
338def _coord_value(value: str) -> float:
339    if value[0] in ("N", "S"):
340        index, strip, replace = 3, "N", "S"
341    else:
342        index, strip, replace = 4, "E", "W"
343    num = f"{value[:index]}.{value[index:]}".lstrip(strip).replace(replace, "-")
344    return float(num)
345
346
347def _position(data: list[str]) -> tuple[list[str], Coord | None]:
348    try:
349        index = data.index("PSN")
350    except ValueError:
351        return data, None
352    data.pop(index)
353    raw = f"{data[index]} {data[index + 1]}"
354    lat = _coord_value(data.pop(index))
355    lon = _coord_value(data.pop(index))
356    return data, Coord(lat=lat, lon=lon, repr=raw)
357
358
359def _movement(data: list[str], units: Units) -> tuple[list[str], Units, Movement | None]:
360    with suppress(ValueError):
361        data.remove("STNR")
362        speed = core.make_number("STNR")
363        return data, units, Movement(repr="STNR", direction=None, speed=speed)
364    try:
365        index = data.index("MOV")
366    except ValueError:
367        return data, units, None
368    raw = data.pop(index)
369    direction_str = data.pop(index)
370    # MOV CNL
371    if direction_str == "CNL":
372        return data, units, None
373    # MOV SLOW ESE / MOV SLOW "SLOW" is a speed qualifier, not a direction
374    if direction_str == "SLOW":
375        raw += f" {direction_str}"
376        if index < len(data):
377            direction_str = data.pop(index)
378            raw += f" {direction_str}"
379        direction = core.make_number(direction_str.replace("/", ""), literal=True, special=CARDINAL_DEGREES)
380        return data, units, Movement(repr=raw.strip(), direction=direction, speed=None)
381    raw += f" {direction_str} "
382    # MOV FROM 23040KT
383    if direction_str == "FROM":
384        value = data[index][:3]
385        raw += value
386        direction = core.make_number(value)
387        data[index] = data[index][3:]
388    # MOV E 45KMH
389    else:
390        direction = core.make_number(direction_str.replace("/", ""), literal=True, special=CARDINAL_DEGREES)
391    speed = None
392    with suppress(IndexError):
393        kt_unit, kmh_unit = data[index].endswith("KT"), data[index].endswith("KMH")
394        if kt_unit or kmh_unit:
395            units.wind_speed = "kmh" if kmh_unit else "kt"
396            speed_str = data.pop(index)
397            raw += speed_str
398            # Remove bottom speed Ex: MOV W 05-10KT
399            if "-" in speed_str:
400                speed_str = speed_str[speed_str.find("-") + 1 :]
401            speed = core.make_number(speed_str[: -3 if kmh_unit else -2])
402    return data, units, Movement(repr=raw.strip(), direction=direction, speed=speed)
403
404
405def _info_from_match(match: re.Match, start: int) -> tuple[str, int]:
406    """Returns the matching text and starting location if none yet available"""
407    if start == -1:
408        start = match.start()
409    return match.group(), start
410
411
412def _pre_break(report: str) -> str:
413    break_index = report.find(" <break> ")
414    return report[:break_index] if break_index != -1 else report
415
416
417def _bounds_from_latterals(report: str, start: int) -> tuple[str, list[str], int]:
418    """Extract coordinate latterals from report Ex: N OF N2050"""
419    bounds = []
420    for match in _LATTERAL_PATTERN.finditer(_pre_break(report)):
421        group, start = _info_from_match(match, start)
422        bounds.append(group.removesuffix(" AND"))
423        report = report.replace(group, " ")
424    return report, bounds, start
425
426
427def _coords_from_text(report: str, start: int) -> tuple[str, list[Coord], int]:
428    """Extract raw coordinate values from report Ex: N4409 E01506"""
429    coords = []
430    for match in _COORD_PATTERN.finditer(_pre_break(report)):
431        group, start = _info_from_match(match, start)
432        text = group.strip(" -")
433        lat, lon = text.split()
434        coord = Coord(lat=_coord_value(lat), lon=_coord_value(lon), repr=text)
435        coords.append(coord)
436        report = report.replace(group, " ")
437    return report, coords, start
438
439
440def _coords_from_navaids(report: str, start: int) -> tuple[str, list[Coord], int]:
441    """Extract navaid referenced coordinates from report Ex: 30SSW BNA"""
442    coords, navs = [], []
443    for match in _NAVAID_PATTERN.finditer(_pre_break(report)):
444        group, start = _info_from_match(match, start)
445        report = report.replace(group, " ")
446        group = group.strip("-").removeprefix("FROM ").removeprefix("TO ")
447        navs.append((group, *group.split()))
448    locs = to_coordinates([n[2 if len(n) == 3 else 1] for n in navs])
449    for i, nav in enumerate(navs):
450        value = nav[0]
451        if len(nav) == 3:
452            vector, num_index = nav[1], 0
453            while vector[num_index].isdigit():
454                num_index += 1
455            distance, bearing = (
456                int(vector[:num_index]),
457                CARDINAL_DEGREES[vector[num_index:]],
458            )
459            loc = geo_distance(nautical=distance).destination(locs[i].pair, bearing=bearing)
460            coord = Coord(lat=loc.latitude, lon=loc.longitude, repr=value)
461        else:
462            coord = locs[i]
463            coord.repr = value
464        coords.append(coord)
465    return report, coords, start
466
467
468def _bounds(data: list[str]) -> tuple[list[str], list[Coord], list[str]]:
469    """Extract coordinate bounds by coord, navaid, and latterals"""
470    report, start = " ".join(data), -1
471    report, bounds, start = _bounds_from_latterals(report, start)
472    report, coords, start = _coords_from_text(report, start)
473    report, navs, start = _coords_from_navaids(report, start)
474    coords += navs
475    for target in ("FROM", "WI", "BOUNDED", "OBS"):
476        index = report.find(f"{target} ")
477        if index != -1 and index < start:
478            start = index
479    report = report[:start] + report[report.rfind("  ") :]
480    data = [s for s in report.split() if s]
481    return data, coords, bounds
482
483
484def _altitudes(data: list[str], units: Units) -> tuple[list[str], Units, Number | None, Number | None]:
485    """Extract the floor and ceiling altitudes"""
486    floor, ceiling = None, None
487    for i, item in enumerate(data):
488        # BTN FL180 AND FL330
489        if item == "BTN" and len(data) > i + 2 and data[i + 2] == "AND":
490            floor, units = core.make_altitude(data[i + 1], units)
491            ceiling, units = core.make_altitude(data[i + 3], units)
492            data = data[:i] + data[i + 4 :]
493            break
494        # TOPS ABV FL450
495        if item in ("TOP", "TOPS", "BLW"):
496            if data[i + 1] == "ABV":
497                ceiling = core.make_number(f"ABV {data[i + 2]}")
498                data = data[:i] + data[i + 3 :]
499                break
500            if data[i + 1] == "BLW":
501                ceiling = core.make_number(f"BLW {data[i + 2]}")
502                data = data[:i] + data[i + 3 :]
503                break
504            # TOPS TO FL310
505            if data[i + 1] == "TO":
506                data.pop(i)
507            ceiling, units = core.make_altitude(data[i + 1], units)
508            data = data[:i] + data[i + 2 :]
509            # CIG BLW 010
510            if data[i - 1] == "CIG":
511                data.pop(i - 1)
512            break
513        # FL060/300 SFC/FL160
514        if core.is_altitude(item):
515            if "/" in item:
516                floor_val, ceiling_val = item.split("/")
517                floor, units = core.make_altitude(floor_val, units)
518                if (floor_val == "SFC" or floor_val[:2] == "FL") and ceiling_val[:2] != "FL":
519                    ceiling, units = core.make_altitude(ceiling_val, units, force_fl=True)
520                else:
521                    ceiling, units = core.make_altitude(ceiling_val, units)
522            else:
523                ceiling, units = core.make_altitude(item, units)
524            data.pop(i)
525            break
526    return data, units, floor, ceiling
527
528
529def _weather_type(data: list[str]) -> tuple[list[str], Code | None]:
530    weather = None
531    report = " ".join(data)
532    for key, val in WEATHER_TYPES.items():
533        if key in report:
534            weather = Code(repr=key, value=val)
535            data = [i for i in report.replace(key, "").split() if i]
536            break
537    return data, weather
538
539
540def _intensity(data: list[str]) -> tuple[list[str], Code | None]:
541    if not data:
542        return data, None
543    try:
544        value = INTENSITY[data[-1]]
545        code = data.pop()
546        return data, Code(repr=code, value=value)
547    except KeyError:
548        return data, None
549
550
551def _sigmet_observation(data: list[str], units: Units, issued: date | None = None) -> tuple[AirSigObservation, Units]:
552    data, start_time, end_time = _time(data, issued)
553    data, position = _position(data)
554    data, coords, bounds = _bounds(data)
555    data, units, movement = _movement(data, units)
556    data, intensity = _intensity(data)
557    data, units, floor, ceiling = _altitudes(data, units)
558    data, weather = _weather_type(data)
559    struct = AirSigObservation(
560        type=weather,
561        start_time=start_time,
562        end_time=end_time,
563        position=position,
564        floor=floor,
565        ceiling=ceiling,
566        coords=coords,
567        bounds=bounds,
568        movement=movement,
569        intensity=intensity,
570        other=_clean_flags(data),
571    )
572    return struct, units
573
574
575def _observations(
576    data: list[str], units: Units, issued: date | None = None
577) -> tuple[Units, AirSigObservation | None, AirSigObservation | None]:
578    observation, forecast, forecast_index = None, None, -1
579    forecast_index = _first_index(data, "FCST", "OUTLOOK", "OTLK")
580    if forecast_index == -1:
581        observation, units = _sigmet_observation(data, units, issued)
582    # 6 is arbitrary. Will likely change or be more precise later
583    elif forecast_index < 6:
584        forecast, units = _sigmet_observation(data, units, issued)
585    else:
586        observation, units = _sigmet_observation(data[:forecast_index], units, issued)
587        forecast, units = _sigmet_observation(data[forecast_index:], units, issued)
588    return units, observation, forecast
589
590
591_REPLACE = {
592    " MO V ": " MOV ",
593    " STNRY": " STNR",
594    " STCNRY": " STNR",
595    " N-NE ": " NNE ",
596    " N-NW ": " NNW ",
597    " E-NE ": " ENE ",
598    " E-SE ": " ESE ",
599    " S-SE ": " SSE ",
600    " S-SW ": " SSW ",
601    " W-SW ": " WSW ",
602    " W-NW ": " WNW ",
603}
604
605
606def _find_first_digit(item: str) -> int:
607    return next((i for i, char in enumerate(item) if char.isdigit()), -1)
608
609
610def sanitize(report: str) -> str:
611    """Sanitized AIRMET / SIGMET report string"""
612    report = report.strip(" =")
613    for key, val in _REPLACE.items():
614        report = report.replace(key, val)
615    data = report.split()
616    for i, item in reversed(list(enumerate(data))):
617        # Remove extra element on altitude Ex: FL450Z skip 1000FT
618        if (
619            len(item) > 4
620            and not item[-1].isdigit()
621            and item[-2:] != "FT"
622            and item[-1] != "M"
623            and core.is_altitude(item[:-1])
624        ):
625            data[i] = item[:-1]
626        # Split attached movement direction Ex: NE05KT
627        if len(item) >= 4 and item.endswith(("KT", "KMH")) and item[: _find_first_digit(item)] in CARDINALS:
628            index = _find_first_digit(item)
629            direction = item[:index]
630            data.insert(i + 1, item[index:])
631            data[i] = direction
632    return " ".join(data)
633
634
635def parse(report: str, issued: date | None = None) -> tuple[AirSigmetData, Units]:
636    """Parse AIRMET / SIGMET report string"""
637    units = Units.international()
638    sanitized = sanitize(report)
639    data, bulletin, issuer, time, correction = _header(_parse_prep(sanitized))
640    data, area, report_type, start_time, end_time, station = _spacetime(data)
641    body = sanitized[sanitized.find(" ".join(data[:2])) :]
642    # Trim AIRMET type
643    if data[0] == "AIRMET":
644        with suppress(ValueError):
645            data = data[data.index("<elip>") + 1 :]
646    data, region = _region(data)
647    units, observation, forecast = _observations(data, units, issued)
648    struct = AirSigmetData(
649        raw=report,
650        sanitized=sanitized,
651        station=station,
652        time=core.make_timestamp(time, target_date=issued),
653        remarks=None,
654        bulletin=bulletin,
655        issuer=issuer,
656        correction=correction,
657        area=area,
658        type=report_type,
659        start_time=core.make_timestamp(start_time, target_date=issued),
660        end_time=core.make_timestamp(end_time, target_date=issued),
661        body=body,
662        region=region,
663        observation=observation,
664        forecast=forecast,
665    )
666    return struct, units