avwx.base

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