avwx.base

  1"""Report parent classes."""
  2
  3# stdlib
  4from __future__ import annotations
  5
  6import asyncio as aio
  7from abc import ABCMeta, abstractmethod
  8from contextlib import suppress
  9from datetime import date, datetime, timezone
 10from typing import TYPE_CHECKING
 11
 12# module
 13from avwx.exceptions import BadStation
 14from avwx.station import Station
 15
 16if TYPE_CHECKING:
 17    from avwx.service import Service
 18    from avwx.structs import ReportData, Units
 19
 20try:
 21    from typing import Self
 22except ImportError:
 23    from typing_extensions import Self
 24
 25
 26# Report header words that are never station codes
 27_REPORT_HEADERS = {"METAR", "SPECI", "TAF", "COR", "AMD", "RTD"}
 28
 29
 30def find_station(report: str) -> Station | None:
 31    """Returns the first Station found in a report string"""
 32    for item in report.split():
 33        if item.upper() in _REPORT_HEADERS:
 34            continue
 35        with suppress(BadStation):
 36            return Station.from_code(item.upper())
 37    return None
 38
 39
 40class AVWXBase(metaclass=ABCMeta):
 41    """Abstract base class for AVWX report types."""
 42
 43    #: UTC datetime object when the report was last updated
 44    last_updated: datetime | None = None
 45
 46    #: UTC date object when the report was issued
 47    issued: date | None = None
 48
 49    #: Root URL used to retrieve the current report
 50    source: str | None = None
 51
 52    #: The original report string
 53    raw: str | None = None
 54
 55    #: ReportData dataclass of parsed data values and units
 56    data: ReportData | None = None
 57
 58    #: Units inferred from the station location and report contents
 59    units: Units | None = None
 60
 61    def __repr__(self) -> str:
 62        return f"<avwx.{self.__class__.__name__}>"
 63
 64    def _set_meta(self) -> None:
 65        """Update timestamps after parsing."""
 66        self.last_updated = datetime.now(tz=timezone.utc)
 67        with suppress(AttributeError):
 68            self.issued = self.data.time.dt.date()  # type: ignore
 69
 70    @abstractmethod
 71    def _post_parse(self) -> None:
 72        pass
 73
 74    @classmethod
 75    def from_report(cls, report: str, issued: date | None = None) -> Self | None:
 76        """Return an updated report object based on an existing report."""
 77        report = report.strip()
 78        obj = cls()
 79        obj.parse(report, issued=issued)
 80        return obj
 81
 82    def parse(self, report: str, issued: date | None = None) -> bool:
 83        """Update report data by parsing a given report.
 84
 85        Can accept a report issue date if not a recent report string.
 86        """
 87        self.source = None
 88        if not report or report == self.raw:
 89            return False
 90        self.raw = report
 91        self.issued = issued
 92        self._post_parse()
 93        self._set_meta()
 94        return True
 95
 96    @staticmethod
 97    def sanitize(report: str) -> str:
 98        """Sanitize the report string.
 99
100        This has not been overridden and returns the raw report.
101        """
102        return report
103
104
105class ManagedReport(AVWXBase, metaclass=ABCMeta):
106    """Abstract base class for reports types associated with a single station."""
107
108    #: 4-character station code the report was initialized with
109    code: str | None = None
110
111    #: Provide basic station info if given at init
112    station: Station | None = None
113
114    #: Service object used to fetch the report string
115    service: Service
116
117    def __init__(self, code: str):
118        code = code.upper()
119        self.code = code
120        self.station = Station.from_code(code)
121
122    def __repr__(self) -> str:
123        return f"<avwx.{self.__class__.__name__} code={self.code}>"
124
125    @abstractmethod
126    async def _post_update(self) -> None:
127        pass
128
129    @classmethod
130    def from_report(cls, report: str, issued: date | None = None) -> Self | None:
131        """Return an updated report object based on an existing report."""
132        report = report.strip()
133        station = find_station(report)
134        if not station:
135            return None
136        obj = cls(station.lookup_code)
137        obj.parse(report, issued=issued)
138        return obj
139
140    async def _update(self, report: str | list[str], issued: date | None, *, disable_post: bool) -> bool:
141        if not report or report == self.raw:
142            return False
143        self.raw = report  # type: ignore
144        self.issued = issued
145        if not disable_post:
146            await self._post_update()
147        self._set_meta()
148        return True
149
150    def update(self, timeout: int = 10, *, disable_post: bool = False) -> bool:
151        """Update. report data by fetching and parsing the report.
152
153        Returns True if a new report is available, else False.
154        """
155        report = self.service.fetch(self.code, timeout=timeout)  # type: ignore
156        self.source = self.service.root
157        return aio.run(self._update(report, None, disable_post=disable_post))
158
159    async def async_update(self, timeout: int = 10, *, disable_post: bool = False) -> bool:
160        """Async update report data by fetching and parsing the report.
161
162        Returns True if a new report is available, else False.
163        """
164        report = await self.service.async_fetch(self.code, timeout=timeout)  # type: ignore
165        self.source = self.service.root
166        return await self._update(report, None, disable_post=disable_post)