Improve radar access

CHANGELOG.rst

@@ -7,6 +7,7 @@ Development
 - Fix DWD DMO access again
 - Add lead time argument - one of short, long - for DWD DMO to address two versions of icon
 - Rework dict-like export formats and tests with extensive support for typing
+- Improve radar access
 
 0.65.0 (24.10.2023)
 *******************

docs/usage/python-api.rst

@@ -791,7 +791,7 @@ Retrieve information about all OPERA radar sites.
     print(f"Number of OPERA radar sites: {len(sites)}")
 
     # Acquire information for a specific OPERA site.
-    site_ukdea = OperaRadarSites().by_odimcode("ukdea")
+    site_ukdea = OperaRadarSites().by_odim_code("ukdea")
     print(site_ukdea)
 
 Retrieve information about the DWD radar sites.
@@ -802,7 +802,7 @@ Retrieve information about the DWD radar sites.
     from wetterdienst.provider.dwd.radar.api import DwdRadarSites
 
     # Acquire information for a specific site.
-    site_asb = DwdRadarSites().by_odimcode("ASB")
+    site_asb = DwdRadarSites().by_odim_code("ASB")
     print(site_asb)
 
 

tests/provider/dwd/radar/test_sites.py

@@ -18,6 +18,6 @@ def test_radar_sites_data_all():
 
 
 def test_radar_sites_data_single():
-    site_asb = DwdRadarSites().by_odimcode("ASB")
+    site_asb = DwdRadarSites().by_odim_code("ASB")
 
     assert site_asb["location"] == "Isle of Borkum"

tests/provider/eumetnet/opera/test_sites.py

@@ -11,40 +11,40 @@ def test_radar_sites_sizes():
 
     assert len(ors.all()) == 205
 
-    assert len(ors.asdict()) == 197
+    assert len(ors.to_dict()) == 197
 
 
 def test_radar_sites_by_odimcode():
     ors = OperaRadarSites()
 
-    assert ors.by_odimcode("ukdea")["location"] == "Dean Hill"
+    assert ors.by_odim_code("ukdea")["location"] == "Dean Hill"
 
-    assert ors.by_odimcode("ASB")["location"] == "Isle of Borkum"
-    assert ors.by_odimcode("EMD")["location"] == "Emden"
-    assert ors.by_odimcode("UMD")["location"] == "Ummendorf"
+    assert ors.by_odim_code("ASB")["location"] == "Isle of Borkum"
+    assert ors.by_odim_code("EMD")["location"] == "Emden"
+    assert ors.by_odim_code("UMD")["location"] == "Ummendorf"
 
     with pytest.raises(ValueError) as exec_info:
-        ors.by_odimcode("foobar")
+        ors.by_odim_code("foobar")
     assert exec_info.match("ODIM code must be three or five letters")
 
     with pytest.raises(KeyError) as exec_info:
-        ors.by_odimcode("foo")
+        ors.by_odim_code("foo")
     assert exec_info.match("'Radar site not found'")
 
 
 def test_radar_sites_by_wmocode():
     ors = OperaRadarSites()
 
-    assert ors.by_wmocode(3859)["location"] == "Dean Hill"
-    assert ors.by_wmocode(10103)["location"] == "Isle of Borkum"
+    assert ors.by_wmo_code(3859)["location"] == "Dean Hill"
+    assert ors.by_wmo_code(10103)["location"] == "Isle of Borkum"
 
 
 def test_radar_sites_by_countryname():
     ors = OperaRadarSites()
 
-    sites_uk = ors.by_countryname(name="United Kingdom")
+    sites_uk = ors.by_country_name(country_name="United Kingdom")
     assert len(sites_uk) == 16
 
     with pytest.raises(KeyError) as exec_info:
-        ors.by_countryname(name="foo")
+        ors.by_country_name(country_name="foo")
     assert exec_info.match("'No radar sites for this country'")

wetterdienst/provider/dwd/radar/api.py

@@ -547,4 +547,4 @@ def __init__(self):
         super().__init__()
 
         # Restrict available sites to the list of OPERA radar sites in Germany.
-        self.sites = self.by_countryname(name="Germany")
+        self.sites = self.by_country_name(country_name="Germany")

wetterdienst/provider/eumetnet/opera/sites.py

@@ -2,10 +2,10 @@
 # Copyright (C) 2018-2021, earthobservations developers.
 # Distributed under the MIT License. See LICENSE for more info.
 import gzip
+import importlib.resources
 import json
-from typing import Dict, List
+from typing import Any, Dict, List, Union
 
-import pkg_resources
 import requests
 
 
@@ -14,7 +14,7 @@ class OperaRadarSites:
     Provide information about all European OPERA radar sites.
     """
 
-    data_file = pkg_resources.resource_filename(__name__, "sites.json.gz")
+    data_file = importlib.resources.files(__package__) / "sites.json.gz"
 
     def __init__(self):
         self.sites = self.load()
@@ -23,16 +23,17 @@ def load(self) -> List[Dict]:
         """
         Load and decode JSON file from filesystem.
         """
-        with gzip.open(self.data_file, mode="rt") as fp:
-            return json.load(fp)
+        with importlib.resources.as_file(self.data_file) as rf:
+            with gzip.open(rf, mode="rb") as f:
+                return json.load(f)
 
     def all(self) -> List[Dict]:  # noqa: A003
         """
         The whole list of OPERA radar sites.
         """
         return self.sites
 
-    def asdict(self) -> Dict:
+    def to_dict(self) -> Dict:
         """
         Dictionary of sites, keyed by ODIM code.
         """
@@ -43,47 +44,42 @@ def asdict(self) -> Dict:
             result[site["odimcode"]] = site
         return result
 
-    def by_odimcode(self, odimcode: str) -> Dict:
+    def by_odim_code(self, odim_code: str) -> Dict:
         """
         Return radar site by ODIM code.
 
-        :param odimcode:
-        :return:
+        :param odim_code: The ODIM code, e.g. "atrau".
+        :return:          Single site information.
         """
-        if not (len(odimcode) == 3 or len(odimcode) == 5):
+        if len(odim_code) not in (3, 5):
             raise ValueError("ODIM code must be three or five letters")
         for site in self.sites:
-            if site["odimcode"] and odimcode.lower() in site["odimcode"]:
+            if site["odimcode"] and odim_code.lower() in site["odimcode"]:
                 return site
         else:
             raise KeyError("Radar site not found")
 
-    def by_wmocode(self, wmocode: int) -> Dict:
+    def by_wmo_code(self, wmo_code: int) -> Dict:
         """
         Return radar site by WMO code.
 
-        :param wmocode:
-        :return:
+        :param wmo_code: The WMO code, e.g. 11038.
+        :return:        Single site information.
         """
         for site in self.sites:
-            if site["wmocode"] == wmocode:
+            if site["wmocode"] == wmo_code:
                 return site
         else:
             raise KeyError("Radar site not found")
 
-    def by_countryname(self, name: str) -> List[Dict]:
+    def by_country_name(self, country_name: str) -> List[Dict]:
         """
         Filter list of radar sites by country name.
 
-        :param name: The country name, e.g. "Germany", "United Kingdom".
-        :return:     List of site information.
+        :param country_name: The country name, e.g. "Germany", "United Kingdom".
+        :return:             List of site information.
         """
-        sites = list(
-            filter(
-                lambda site: site["country"] and site["country"].lower() == name.lower(),
-                self.sites,
-            )
-        )
+        sites = [site for site in self.sites if site["country"] and site["country"].lower() == country_name.lower()]
         if not sites:
             raise KeyError("No radar sites for this country")
         return sites
@@ -118,7 +114,7 @@ def get_opera_radar_sites(self) -> List[Dict]:  # pragma: no cover
         ]
         boolean_values = ["doppler", "status"]
 
-        def asbool(obj):
+        def asbool(obj: Any) -> bool:
             # from sqlalchemy.util.asbool
             if isinstance(obj, str):
                 obj = obj.strip().lower()
@@ -130,7 +126,7 @@ def asbool(obj):
                     raise ValueError("String is not true/false: %r" % obj)
             return bool(obj)
 
-        def convert_types(element):
+        def convert_types(element: Dict) -> Dict[str, Union[int, float, bool, None]]:
             converted = {}
             for key, value in element.items():
                 try:
@@ -142,7 +138,7 @@ def convert_types(element):
                         value = asbool(value)
                     if value == "":
                         value = None
-                except Exception:  # noqa: S110
+                except ValueError:
                     value = None
                 converted[key] = value
             return converted
@@ -154,20 +150,21 @@ def filter_and_convert(elements):
 
         return list(filter_and_convert(data))
 
-    def to_json(self) -> str:
+    def to_json(self, indent: int = 4) -> str:
         """
         Return JSON representation of all sites.
         """
         sites = self.get_opera_radar_sites()
-        return json.dumps(sites, indent=4)
+        return json.dumps(sites, indent=indent)
 
-    def export(self):
+    def export(self, indent: int = 4):
         """
         Generate "sites.json.gz".
         """
         sites = self.get_opera_radar_sites()
-        with gzip.open(OperaRadarSites.data_file, mode="wt", compresslevel=9, encoding="utf-8") as fp:
-            json.dump(sites, fp, indent=4)
+        with importlib.resources.as_file(OperaRadarSites.data_file) as rf:
+            with gzip.open(rf, mode="wt", compresslevel=9, encoding="utf-8") as f:
+                json.dump(sites, f, indent=indent)
 
 
 if __name__ == "__main__":  # pragma: no cover

wetterdienst/ui/cli.py

@@ -801,7 +801,7 @@ def values(
     parameter: List[str],
     resolution: str,
     period: List[str],
-    lead_time: str,
+    lead_time: Literal["short", "long"],
     date: str,
     issue: str,
     all_: bool,
@@ -909,7 +909,7 @@ def interpolate(
     parameter: List[str],
     resolution: str,
     period: List[str],
-    lead_time: str,
+    lead_time: Literal["short", "long"],
     use_nearby_station_distance: float,
     date: str,
     issue: str,
@@ -996,7 +996,7 @@ def summarize(
     parameter: List[str],
     resolution: str,
     period: List[str],
-    lead_time: str,
+    lead_time: Literal["short", "long"],
     date: str,
     issue: str,
     station: str,
@@ -1059,12 +1059,14 @@ def summarize(
     RequireExactly(1),
     ["dwd", "all_", "odim_code", "wmo_code", "country_name"],
 )
+@cloup.option("--indent", type=click.INT, default=4)
 def radar(
     dwd: bool,
     all_: bool,
     odim_code: str,
-    wmo_code: str,
+    wmo_code: int,
     country_name: str,
+    indent: int,
 ):
     from wetterdienst.provider.dwd.radar.api import DwdRadarSites
     from wetterdienst.provider.eumetnet.opera.sites import OperaRadarSites
@@ -1075,13 +1077,15 @@ def radar(
         if all_:
             data = OperaRadarSites().all()
         elif odim_code:
-            data = OperaRadarSites().by_odimcode(odim_code)
+            data = OperaRadarSites().by_odim_code(odim_code)
         elif wmo_code:
-            data = OperaRadarSites().by_wmocode(wmo_code)
+            data = OperaRadarSites().by_wmo_code(wmo_code)
         elif country_name:
-            data = OperaRadarSites().by_countryname(country_name)
+            data = OperaRadarSites().by_country_name(country_name)
+        else:
+            raise KeyError("No valid option provided")
 
-    output = json.dumps(data, indent=2)
+    output = json.dumps(data, indent=indent)
 
     print(output)  # noqa: T201