avwx.current.base

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