avwx.station.station

  1"""
  2Station handling and coordinate search
  3"""
  4
  5# pylint: disable=invalid-name,too-many-arguments,too-many-instance-attributes
  6
  7# stdlib
  8from contextlib import suppress
  9from copy import copy
 10from dataclasses import dataclass
 11from functools import lru_cache
 12from typing import Any, Dict, List, Optional, Tuple, Type, TypeVar, Union
 13
 14# library
 15import httpx
 16from geopy.distance import great_circle, Distance  # type: ignore
 17
 18# module
 19from avwx.exceptions import BadStation
 20from avwx.load_utils import LazyCalc
 21from avwx.station.meta import STATIONS
 22from avwx.structs import Coord
 23
 24
 25def _get_ip_location() -> Coord:
 26    """Returns the current location according to ipinfo.io"""
 27    lat, lon = httpx.get("https://ipinfo.io/loc").text.strip().split(",")
 28    return Coord(float(lat), float(lon))
 29
 30
 31@dataclass
 32class Runway:
 33    """Represents a runway at an airport"""
 34
 35    length_ft: int
 36    width_ft: int
 37    surface: str
 38    lights: bool
 39    ident1: str
 40    ident2: str
 41    bearing1: float
 42    bearing2: float
 43
 44
 45T = TypeVar("T", bound="Station")
 46_ICAO = LazyCalc(lambda: {v["icao"]: k for k, v in STATIONS.items() if v["icao"]})
 47_IATA = LazyCalc(lambda: {v["iata"]: k for k, v in STATIONS.items() if v["iata"]})
 48_GPS = LazyCalc(lambda: {v["gps"]: k for k, v in STATIONS.items() if v["gps"]})
 49_LOCAL = LazyCalc(lambda: {v["local"]: k for k, v in STATIONS.items() if v["local"]})
 50
 51
 52@dataclass
 53class Station:
 54    """
 55    The Station dataclass stores basic info about the desired station and
 56    available Runways.
 57
 58    The easiest way to get a station is to supply the ICAO, IATA, or GPS code.
 59    The example below uses `from_code` which checks against all three types,
 60    but you can also use `from_icao`, `from_iata`, or `from_gps` if you know
 61    what type of code you are using. This can be important if you may be using
 62    a code used by more than one station depending on the context. ICAO and
 63    IATA codes are guarenteed unique, but not all airports have them. That
 64    said, all stations available in AVWX have either an ICAO or GPS code.
 65
 66    ```python
 67    >>> from avwx import Station
 68    >>> klex = Station.from_code("KLEX")
 69    >>> f"{klex.name} in {klex.city}, {klex.state}"
 70    'Blue Grass Airport in Lexington, KY'
 71    >>> coord = round(klex.latitude, 3), round(klex.longitude, 3)
 72    >>> f"Located at {coord} at {klex.elevation_ft} feet ({klex.elevation_m} meters)"
 73    'Located at (38.036, -84.606) at 979 feet (298 meters)'
 74    >>> rw = max(klex.runways, key=lambda r: r.length_ft)
 75    >>> f"Its longest runway is {rw.ident1}/{rw.ident2} at {rw.length_ft} feet"
 76    'Its longest runway is 04/22 at 7003 feet'
 77    ```
 78
 79    This is also the same information you'd get from calling Report.station.
 80
 81    ```python
 82    >>> from avwx import Metar
 83    >>> klex = Metar('KLEX')
 84    >>> klex.station.name
 85    'Blue Grass Airport'
 86    ```
 87    """
 88
 89    # pylint: disable=too-many-instance-attributes
 90
 91    city: Optional[str]
 92    country: str
 93    elevation_ft: Optional[int]
 94    elevation_m: Optional[int]
 95    gps: Optional[str]
 96    iata: Optional[str]
 97    icao: Optional[str]
 98    latitude: float
 99    local: Optional[str]
100    longitude: float
101    name: str
102    note: Optional[str]
103    reporting: bool
104    runways: List[Runway]
105    state: Optional[str]
106    type: str
107    website: Optional[str]
108    wiki: Optional[str]
109
110    @classmethod
111    def _from_code(cls: Type[T], ident: str) -> T:
112        try:
113            info: Dict[str, Any] = copy(STATIONS[ident])
114            if info["runways"]:
115                info["runways"] = [Runway(**r) for r in info["runways"]]
116            return cls(**info)
117        except (KeyError, AttributeError) as not_found:
118            raise BadStation(
119                f"Could not find station with ident {ident}"
120            ) from not_found
121
122    @classmethod
123    def from_code(cls: Type[T], ident: str) -> T:
124        """Load a Station from ICAO, GPS, or IATA code in that order"""
125        if ident and isinstance(ident, str):
126            if len(ident) == 4:
127                with suppress(BadStation):
128                    return cls.from_icao(ident)
129                with suppress(BadStation):
130                    return cls.from_gps(ident)
131            if len(ident) == 3:
132                with suppress(BadStation):
133                    return cls.from_iata(ident)
134            with suppress(BadStation):
135                return cls.from_local(ident)
136        raise BadStation(f"Could not find station with ident {ident}")
137
138    @classmethod
139    def from_icao(cls: Type[T], ident: str) -> T:
140        """Load a Station from an ICAO station ident"""
141        try:
142            return cls._from_code(_ICAO.value[ident.upper()])
143        except (KeyError, AttributeError) as not_found:
144            raise BadStation(
145                f"Could not find station with ICAO ident {ident}"
146            ) from not_found
147
148    @classmethod
149    def from_iata(cls: Type[T], ident: str) -> T:
150        """Load a Station from an IATA code"""
151        try:
152            return cls._from_code(_IATA.value[ident.upper()])
153        except (KeyError, AttributeError) as not_found:
154            raise BadStation(
155                f"Could not find station with IATA ident {ident}"
156            ) from not_found
157
158    @classmethod
159    def from_gps(cls: Type[T], ident: str) -> T:
160        """Load a Station from a GPS code"""
161        try:
162            return cls._from_code(_GPS.value[ident.upper()])
163        except (KeyError, AttributeError) as not_found:
164            raise BadStation(
165                f"Could not find station with GPS ident {ident}"
166            ) from not_found
167
168    @classmethod
169    def from_local(cls: Type[T], ident: str) -> T:
170        """Load a Station from a local code"""
171        try:
172            return cls._from_code(_LOCAL.value[ident.upper()])
173        except (KeyError, AttributeError) as not_found:
174            raise BadStation(
175                f"Could not find station with local ident {ident}"
176            ) from not_found
177
178    @classmethod
179    def nearest(
180        cls: Type[T],
181        lat: Optional[float] = None,
182        lon: Optional[float] = None,
183        is_airport: bool = False,
184        sends_reports: bool = True,
185        max_coord_distance: float = 10,
186    ) -> Optional[Tuple[T, dict]]:
187        """Load the Station nearest to your location or a lat,lon coordinate pair
188
189        Returns the Station and distances from source
190
191        NOTE: Becomes less accurate toward poles and doesn't cross +/-180
192        """
193        if not (lat and lon):
194            lat, lon = _get_ip_location().pair
195        ret = nearest(lat, lon, 1, is_airport, sends_reports, max_coord_distance)
196        if not isinstance(ret, dict):
197            return None
198        station = ret.pop("station")
199        return station, ret
200
201    @property
202    def lookup_code(self) -> str:
203        """Returns the ICAO or GPS code for report fetch"""
204        if self.icao:
205            return self.icao
206        if self.gps:
207            return self.gps
208        raise BadStation("Station does not have a valid lookup code")
209
210    @property
211    def storage_code(self) -> str:
212        """Returns the first unique-ish code from what's available"""
213        if self.icao:
214            return self.icao
215        if self.iata:
216            return self.iata
217        if self.gps:
218            return self.gps
219        if self.local:
220            return self.local
221        raise BadStation("Station does not have any useable codes")
222
223    @property
224    def sends_reports(self) -> bool:
225        """Returns whether or not a Station likely sends weather reports"""
226        return self.reporting is True
227
228    @property
229    def coord(self) -> Coord:
230        """Returns the station location as a Coord"""
231        return Coord(lat=self.latitude, lon=self.longitude, repr=self.icao)
232
233    def distance(self, lat: float, lon: float) -> Distance:
234        """Returns a geopy Distance using the great circle method"""
235        return great_circle((lat, lon), (self.latitude, self.longitude))
236
237    def nearby(
238        self,
239        is_airport: bool = False,
240        sends_reports: bool = True,
241        max_coord_distance: float = 10,
242    ) -> List[Tuple[T, dict]]:
243        """Returns Stations nearest to current station and their distances
244
245        NOTE: Becomes less accurate toward poles and doesn't cross +/-180
246        """
247        stations = nearest(
248            self.latitude,
249            self.longitude,
250            11,
251            is_airport,
252            sends_reports,
253            max_coord_distance,
254        )
255        if isinstance(stations, dict):
256            return []
257        return [(s.pop("station"), s) for s in stations[1:]]
258
259
260# Coordinate search and resources
261
262
263def _make_coords() -> List[Tuple]:
264    return [
265        (
266            s["icao"] or s["gps"] or s["iata"] or s["local"],
267            s["latitude"],
268            s["longitude"],
269        )
270        for s in STATIONS.values()
271    ]
272
273
274_COORDS = LazyCalc(_make_coords)
275
276
277def _make_coord_tree():  # type: ignore
278    # pylint: disable=import-outside-toplevel
279    try:
280        from scipy.spatial import KDTree  # type: ignore
281
282        return KDTree([c[1:] for c in _COORDS.value])
283    except (NameError, ModuleNotFoundError) as name_error:
284        raise ModuleNotFoundError(
285            'scipy must be installed to use coordinate lookup. Run "pip install avwx-engine[scipy]" to enable this feature'
286        ) from name_error
287
288
289_COORD_TREE = LazyCalc(_make_coord_tree)
290
291
292def _query_coords(lat: float, lon: float, n: int, d: float) -> List[Tuple[str, float]]:
293    """Returns <= n number of ident, dist tuples <= d coord distance from lat,lon"""
294    dist, index = _COORD_TREE.value.query([lat, lon], n, distance_upper_bound=d)
295    if n == 1:
296        dist, index = [dist], [index]
297    # NOTE: index == len of list means Tree ran out of items
298    return [
299        (_COORDS.value[i][0], d) for i, d in zip(index, dist) if i < len(_COORDS.value)
300    ]
301
302
303def station_filter(station: Station, is_airport: bool, reporting: bool) -> bool:
304    """Return True if station matches given criteria"""
305    if is_airport and "airport" not in station.type:
306        return False
307    return bool(not reporting or station.sends_reports)
308
309
310@lru_cache(maxsize=128)
311def _query_filter(
312    lat: float, lon: float, n: int, d: float, is_airport: bool, reporting: bool
313) -> List[Tuple[Station, float]]:
314    """Returns <= n number of stations <= d distance from lat,lon matching the query params"""
315    k = n * 20
316    last = 0
317    stations: List[Tuple[Station, float]] = []
318    while True:
319        nodes = _query_coords(lat, lon, k, d)[last:]
320        # Ran out of new stations
321        if not nodes:
322            return stations
323        for code, dist in nodes:
324            if not code:
325                continue
326            stn = Station.from_code(code)
327            if station_filter(stn, is_airport, reporting):
328                stations.append((stn, dist))
329            # Reached the desired number of stations
330            if len(stations) >= n:
331                return stations
332        last = k
333        k += n * 100
334
335
336def nearest(
337    lat: float,
338    lon: float,
339    n: int = 1,
340    is_airport: bool = False,
341    sends_reports: bool = True,
342    max_coord_distance: float = 10,
343) -> Union[dict, List[dict]]:
344    """Finds the nearest n Stations to a lat,lon coordinate pair
345
346    Returns the Station and coordinate distance from source
347
348    NOTE: Becomes less accurate toward poles and doesn't cross +/-180
349    """
350    # Default state includes all, no filtering necessary
351    if is_airport or sends_reports:
352        stations = _query_filter(
353            lat, lon, n, max_coord_distance, is_airport, sends_reports
354        )
355    else:
356        data = _query_coords(lat, lon, n, max_coord_distance)
357        stations = [(Station.from_code(code), d) for code, d in data]
358    if not stations:
359        return []
360    ret = []
361    for station, coord_dist in stations:
362        dist = station.distance(lat, lon)
363        ret.append(
364            {
365                "station": station,
366                "coordinate_distance": coord_dist,
367                "nautical_miles": dist.nautical,
368                "miles": dist.miles,
369                "kilometers": dist.kilometers,
370            }
371        )
372    if n == 1:
373        return ret[0]
374    ret.sort(key=lambda x: x["miles"])
375    return ret