Add FAA airspace data sets (#496)

Author: sfo

src/traffic/data/faa/__init__.py

@@ -1,64 +1,33 @@
-import importlib
-import json
-import logging
-from pathlib import Path
-from typing import Any, Dict
-
-import httpx
-
-import pandas as pd
-
-from ... import cache_expiration, cache_path
-from .. import client
-
-__all__ = list(p.stem[1:] for p in Path(__file__).parent.glob("_[a-z]*py"))
-
-_log = logging.getLogger(__name__)
-
-
-class ADDS_FAA_OpenData:
-    id_: str
-    filename: str
-    website = "https://adds-faa.opendata.arcgis.com/datasets/{}"
-    json_url = "https://opendata.arcgis.com/datasets/{}.geojson"
-
-    def __init__(self) -> None:
-        self.cache_file = cache_path / self.filename
-        self.website = self.website.format(self.id_)
-        self.json_url = self.json_url.format(self.id_)
-
-    def download_data(self) -> None:
-        _log.warning(
-            f"Downloading data from {self.website}. Please check terms of use."
-        )
-        c = client.get(self.json_url)
-        c.raise_for_status()
-        json_contents = c.json()
-        with self.cache_file.open("w") as fh:
-            json.dump(json_contents, fh)
-
-    def json_contents(self) -> Dict[str, Any]:
-        if self.cache_file.exists():
-            last_modification = (self.cache_file).lstat().st_mtime
-            delta = pd.Timestamp("now") - pd.Timestamp(last_modification * 1e9)
-
-            if cache_expiration is not None and delta > cache_expiration:
-                try:
-                    self.download_data()
-                except httpx.TransportError:
-                    pass
-        else:
-            self.download_data()
-
-        with self.cache_file.open("r") as fh:
-            json_contents = json.load(fh)
-
-        return json_contents  # type: ignore
-
-
+from typing import Any
+
+from ._airspaces import (
+    Airspace_Boundary,
+    Class_Airspace,
+    Prohibited_Airspace,
+    Route_Airspace,
+    Special_Use_Airspace,
+)
+from ._ats_route import Ats_Route
+from ._designated_points import Designated_Points
+from ._navaid_components import Navaid_Components
+
+__all__ = [
+    "Airspace_Boundary",
+    "Ats_Route",
+    "Class_Airspace",
+    "Designated_Points",
+    "Navaid_Components",
+    "Prohibited_Airspace",
+    "Route_Airspace",
+    "Special_Use_Airspace",
+]
+
+
+# this is here for compatibility with the <=2.12.0 API to allow for stuff like
+# `from traffic.data.faa import class_airspace`
 def __getattr__(name: str) -> Any:
-    if name in __all__:
-        mod = importlib.import_module("._" + name, package="traffic.data.faa")
-        return getattr(mod, name.title())()
+    for cls in __all__:
+        if cls.lower() == name.lower():
+            return globals()[cls]()
 
     raise AttributeError()

src/traffic/data/faa/_adds_faa_opendata.py

@@ -0,0 +1,52 @@
+import json
+import logging
+from typing import Any, Dict
+
+import httpx
+
+import pandas as pd
+
+from ... import cache_expiration, cache_path
+from .. import client
+
+_log = logging.getLogger(__name__)
+
+
+class ADDS_FAA_OpenData:
+    id_: str
+    filename: str
+    website = "https://adds-faa.opendata.arcgis.com/datasets/{}"
+    json_url = "https://opendata.arcgis.com/datasets/{}.geojson"
+
+    def __init__(self) -> None:
+        self.cache_file = cache_path / self.filename
+        self.website = self.website.format(self.id_)
+        self.json_url = self.json_url.format(self.id_)
+
+    def download_data(self) -> None:
+        _log.warning(
+            f"Downloading data from {self.website}. Please check terms of use."
+        )
+        c = client.get(self.json_url)
+        c.raise_for_status()
+        json_contents = c.json()
+        with self.cache_file.open("w") as fh:
+            json.dump(json_contents, fh)
+
+    def json_contents(self) -> Dict[str, Any]:
+        if self.cache_file.exists():
+            last_modification = (self.cache_file).lstat().st_mtime
+            delta = pd.Timestamp("now") - pd.Timestamp(last_modification * 1e9)
+
+            if cache_expiration is not None and delta > cache_expiration:
+                try:
+                    self.download_data()
+                except httpx.TransportError:
+                    pass
+        else:
+            self.download_data()
+
+        with self.cache_file.open("r") as fh:
+            json_contents = json.load(fh)
+
+        return json_contents  # type: ignore

…c/traffic/data/faa/_airspace_boundary.py src/traffic/data/faa/_airspaces.pysrc/traffic/data/faa/_airspace_boundary.py renamed to src/traffic/data/faa/_airspaces.py src/traffic/data/faa/_airspace_boundary.py renamed to src/traffic/data/faa/_airspaces.py

@@ -9,15 +9,12 @@
 from shapely.ops import orient
 
 from ...core.airspace import Airspace, Airspaces, ExtrudedPolygon
-from . import ADDS_FAA_OpenData
+from ._adds_faa_opendata import ADDS_FAA_OpenData
 
 _log = logging.getLogger(__name__)
 
 
-class Airspace_Boundary(ADDS_FAA_OpenData, Airspaces):
-    id_ = "67885972e4e940b2aa6d74024901c561_0"
-    filename = "faa_airspace_boundary.json"
-
+class FAA_Airspace(ADDS_FAA_OpenData, Airspaces):
     def __init__(self, data: gpd.GeoDataFrame | None = None) -> None:
         super().__init__()
         self.data = data
@@ -78,3 +75,28 @@ def back(self) -> Dict[str, Airspace]:
                 airspaces[name] = airspace
 
         return airspaces
+
+
+class Airspace_Boundary(FAA_Airspace):
+    id_ = "67885972e4e940b2aa6d74024901c561_0"
+    filename = "faa_airspace_boundary.json"
+
+
+class Class_Airspace(FAA_Airspace):
+    id_ = "c6a62360338e408cb1512366ad61559e_0"
+    filename = "faa_class_airspace.json"
+
+
+class Special_Use_Airspace(FAA_Airspace):
+    id_ = "dd0d1b726e504137ab3c41b21835d05b_0"
+    filename = "faa_special_use_airspace.json"
+
+
+class Route_Airspace(FAA_Airspace):
+    id_ = "8bf861bb9b414f4ea9f0ff2ca0f1a851_0"
+    filename = "faa_route_airspace.json"
+
+
+class Prohibited_Airspace(FAA_Airspace):
+    id_ = "354ee0c77484461198ebf728a2fca50c_0"
+    filename = "faa_prohibited_airspace.json"

src/traffic/data/faa/_ats_route.py

@@ -1,6 +1,6 @@
 from typing import Any, Dict
 
-from . import ADDS_FAA_OpenData
+from ._adds_faa_opendata import ADDS_FAA_OpenData
 
 
 class Ats_Route(ADDS_FAA_OpenData):

src/traffic/data/faa/_designated_points.py

@@ -1,7 +1,7 @@
 import pandas as pd
 
 from ..basic.navaid import Navaids
-from . import ADDS_FAA_OpenData
+from ._adds_faa_opendata import ADDS_FAA_OpenData
 
 
 class FAA_Designated_Points(Navaids):

src/traffic/data/faa/_navaid_components.py

@@ -1,7 +1,7 @@
 import pandas as pd
 
 from ..basic.navaid import Navaids
-from . import ADDS_FAA_OpenData
+from ._adds_faa_opendata import ADDS_FAA_OpenData
 
 
 class Navaid_Components(ADDS_FAA_OpenData):