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