avwx.service.bulk

 1"""
 2These services are specifically for returning multiple reports at a time.
 3For example, we'd want to know all SIGMETs currently in circulation.
 4The sources can be FTP, scraping, or any other method. There is no need
 5for specific stations or updating files behind the scenes.
 6
 7The `fetch` and `async_fetch` methods are identical except they return
 8`List[str]` instead.
 9"""
10
11# pylint: disable=invalid-name
12
13# stdlib
14import asyncio as aio
15from contextlib import suppress
16from typing import List
17
18from avwx.service.base import CallsHTTP, Service
19
20
21class NOAA_Bulk(Service, CallsHTTP):
22    """Subclass for extracting current reports from NOAA CSV files
23
24    This class accepts `"metar"`, `"taf"`, `"aircraftreport"`, and
25    `"airsigmet"` as valid report types.
26    """
27
28    _url = "https://aviationweather.gov/data/cache/{}s.cache.csv"
29    _valid_types = ("metar", "taf", "aircraftreport", "airsigmet")
30    _rtype_map = {"airep": "aircraftreport", "pirep": "aircraftreport"}
31    _targets = {"aircraftreport": -2}  # else 0
32
33    def __init__(self, report_type: str):
34        super().__init__(self._rtype_map.get(report_type, report_type))
35
36    @staticmethod
37    def _clean_report(report: str) -> str:
38        report = report.strip(" '\"")
39        for remove in (r"\x07", "\x07"):
40            report = report.replace(remove, " ")
41        return " ".join(report.split())
42
43    def _extract(self, raw: str) -> List[str]:
44        reports = []
45        index = self._targets.get(self.report_type, 0)
46        for line in raw.split("\n")[6:]:
47            with suppress(IndexError):
48                report = self._clean_report(line.split(",")[index])
49                if report:
50                    reports.append(report)
51        return reports
52
53    def fetch(self, timeout: int = 10) -> List[str]:
54        """Bulk fetch report strings from the service"""
55        return aio.run(self.async_fetch(timeout))
56
57    async def async_fetch(self, timeout: int = 10) -> List[str]:
58        """Asynchronously bulk fetch report strings from the service"""
59        url = self._url.format(self.report_type)
60        text = await self._call(url, timeout=timeout)
61        return self._extract(text)
62
63
64class NOAA_Intl(Service, CallsHTTP):
65    """Scrapes international reports from NOAA. Designed to
66    accompany `NOAA_Bulk` for AIRMET / SIGMET fetch.
67
68    Currently, this class only accepts `"airsigmet"` as a valid report type.
69    """
70
71    _url = "https://www.aviationweather.gov/api/data/{}"
72    _valid_types = ("airsigmet",)
73    _url_map = {"airsigmet": "isigmet"}
74
75    @staticmethod
76    def _clean_report(report: str) -> str:
77        lines = report.split()
78        return " ".join([line for line in lines if not line.startswith("Hazard:")])
79
80    def _extract(self, raw: str) -> List[str]:
81        reports = []
82        for line in raw.split("----------------------"):
83            reports.append(self._clean_report(line.strip().strip('"')))
84        return reports
85
86    def fetch(self, timeout: int = 10) -> List[str]:
87        """Bulk fetch report strings from the service"""
88        return aio.run(self.async_fetch(timeout))
89
90    async def async_fetch(self, timeout: int = 10) -> List[str]:
91        """Asynchronously bulk fetch report strings from the service"""
92        url = self._url.format(self._url_map[self.report_type])
93        text = await self._call(url, timeout=timeout)
94        return self._extract(text)