avwx.current.base

  1"""Current report shared resources."""
  2
  3# stdlib
  4from __future__ import annotations
  5
  6import asyncio as aio
  7from typing import TYPE_CHECKING
  8
  9# module
 10from avwx.base import ManagedReport
 11from avwx.service import get_service
 12from avwx.static.core import WX_TRANSLATIONS
 13from avwx.structs import Code, Coord, ReportData, ReportTrans, Sanitization, Units
 14
 15if TYPE_CHECKING:
 16    from datetime import date
 17
 18
 19def wx_code(code: str) -> Code | str:
 20    """Translate weather codes into readable strings.
 21
 22    Returns translated string of variable length
 23    """
 24    if not code:
 25        return ""
 26    ret, code_copy = "", code
 27    if code[0] == "+":
 28        ret = "Heavy "
 29        code = code[1:]
 30    elif code[0] == "-":
 31        ret = "Light "
 32        code = code[1:]
 33    # Return code if code is not a code, ex R03/03002V03
 34    if len(code) not in [2, 4, 6] or code.isdigit():
 35        return code
 36    is_code = False
 37    while code:
 38        try:
 39            ret += f"{WX_TRANSLATIONS[code[:2]]} "
 40            is_code = True
 41        except KeyError:
 42            ret += code[:2]
 43        code = code[2:]
 44    # Return Code if any part was able to be translated
 45    return Code(code_copy, ret.strip()) if is_code else code_copy
 46
 47
 48def get_wx_codes(codes: list[str]) -> tuple[list[str], list[Code]]:
 49    """Separate parsed WX codes."""
 50    other: list[str] = []
 51    ret: list[Code] = []
 52    for item in codes:
 53        code = wx_code(item)
 54        if isinstance(code, Code):
 55            ret.append(code)
 56        else:
 57            other.append(code)
 58    return other, ret
 59
 60
 61class Report(ManagedReport):
 62    """Base report to take care of service assignment and station info."""
 63
 64    #: ReportTrans dataclass of translation strings from data. Parsed on update()
 65    translations: ReportTrans | None = None
 66
 67    sanitization: Sanitization | None = None
 68
 69    def __init__(self, code: str):
 70        """Add doc string to show constructor."""
 71        super().__init__(code)
 72        if self.station is not None:
 73            service = get_service(code, self.station.country)
 74            self.service = service(self.__class__.__name__.lower())  # type: ignore
 75
 76
 77class Reports(ManagedReport):
 78    """Base class containing multiple reports."""
 79
 80    coord: Coord | None = None
 81    raw: list[str] | None = None  # type: ignore
 82    data: list[ReportData] | None = None  # type: ignore
 83    units: Units = Units.north_american()
 84    sanitization: list[Sanitization] | None = None
 85
 86    def __init__(self, code: str | None = None, coord: Coord | None = None):
 87        if code:
 88            super().__init__(code)
 89            if self.station is not None:
 90                coord = self.station.coord
 91        elif coord is None:
 92            msg = "No station or coordinate given"
 93            raise ValueError(msg)
 94        self.coord = coord
 95
 96    def __repr__(self) -> str:
 97        if self.code:
 98            return f"<avwx.{self.__class__.__name__} code={self.code}>"
 99        return f"<avwx.{self.__class__.__name__} coord={self.coord}>"
100
101    @staticmethod
102    def _report_filter(reports: list[str]) -> list[str]:
103        """Apply any report filtering before updating raw_reports."""
104        return reports
105
106    async def _update(  # type: ignore
107        self, reports: list[str], issued: date | None, *, disable_post: bool
108    ) -> bool:
109        if not reports:
110            return False
111        reports = self._report_filter(reports)
112        return await super()._update(reports, issued, disable_post=disable_post)
113
114    def parse(self, reports: str | list[str], issued: date | None = None) -> bool:
115        """Update report data by parsing a given report.
116
117        Can accept a report issue date if not a recent report string
118        """
119        return aio.run(self.async_parse(reports, issued))
120
121    async def async_parse(self, reports: str | list[str], issued: date | None = None) -> bool:
122        """Async update report data by parsing a given report.
123
124        Can accept a report issue date if not a recent report string
125        """
126        self.source = None
127        if isinstance(reports, str):
128            reports = [reports]
129        return await self._update(reports, issued, disable_post=False)
130
131    def update(self, timeout: int = 10, *, disable_post: bool = False) -> bool:
132        """Update report data by fetching and parsing the report.
133
134        Returns True if new reports are available, else False
135        """
136        return aio.run(self.async_update(timeout, disable_post=disable_post))
137
138    async def async_update(self, timeout: int = 10, *, disable_post: bool = False) -> bool:
139        """Async update report data by fetching and parsing the report."""
140        reports = await self.service.async_fetch(coord=self.coord, timeout=timeout)  # type: ignore
141        self.source = self.service.root
142        return await self._update(reports, None, disable_post=disable_post)