Migrated sica_lu to new API

custom_components/waste_collection_schedule/waste_collection_schedule/source/sica_lu.py

@@ -1,88 +1,80 @@
 import datetime
-
 import requests
 from waste_collection_schedule import Collection
+from waste_collection_schedule.exceptions import SourceArgumentNotFoundWithSuggestions
 
 TITLE = "SICA"
 DESCRIPTION = "Source script for sica.lu served municipalities"
 URL = "https://sica.lu"
 TEST_CASES = {
-    "Bertrange": {"municipality": "Bertrange"},
-    "Capellen": {"municipality": "capellen"},
-    "Garnich": {"municipality": "Garnich"},
-    "Holzem": {"municipality": "holzem"},
-    # For testing purposes:
-    # "Kehlen": {"municipality": "Kehlen"},
-    # "Koerich": {"municipality": "Koerich"},
-    # "Kopstal": {"municipality": "Kopstal"},
-    # "Mamer": {"municipality": "Mamer"},
-    # "Septfontaines": {"municipality": "Septfontaines"},
-    # "Steinfort": {"municipality": "Steinfort"},
+    "Habscht": {"municipality": "habscht"},
+    "Steinfort": {"municipality": "Steinfort"},
 }
 
-API_URL = "http://sicaapp.lu/wp-json/wp/v2"
+BASE_URL = "https://dashboard.sicaapp.lu"
+API_URL = f"{BASE_URL}/api/api/app"
+HEADERS = {"User-Agent": "SicaAPP", "Accept": "application/json"}
 ICON_MAP = {
     "Residual waste": "mdi:trash-can",
-    "Valorlux packaging": "mdi:recycle",
+    "Valorlux - blue bag": "mdi:recycle",
     "Organic waste": "mdi:leaf",
     "Glass": "mdi:bottle-wine-outline",
-    "Paper - Cartons": "mdi:newspaper",
+    "Paper /Carton": "mdi:newspaper",
     "Scrap and electrical appliances": "mdi:washing-machine",
     "Clothing and Shoes": "mdi:tshirt-crew",
-    "Trees, shrubs and hedges": "mdi:tree",
-}
-
-MUNICIPALITIES = {
-    "bertrange": 28,
-    "capellen": 138,
-    "garnich": 29,
-    "holzem": 139,
-    "kehlen": 30,
-    "koerich": 31,
-    "kopstal": 24,
-    "mamer": 137,
-    "septfontaines": 26,
-    "steinfort": 27,
+    "Hedges, Shrubs and Trees": "mdi:tree",
+    "Bulky waste": "mdi:sofa",
 }
 
 
 class Source:
-    def __init__(self, municipality: str):
-        # https://sicaapp.lu/wp-json/wp/v2/locations/
-
-        self._municipality = MUNICIPALITIES.get(municipality.lower())
-        if not self._municipality:
-            raise ValueError(
-                f"Unknown municipality: {municipality}, use one of {list(MUNICIPALITIES.keys())}"
-            )
+    def __init__(self, municipality: str) -> None:
+        self._municipality = municipality.lower()
+        self._municipality_id = None
 
-    def fetch(self):
-        headers = {"User-Agent": "SicaAPP", "Accept": "application/json"}
-        year = datetime.datetime.now().year
-        url = f"{API_URL}/calendaryear/{self._municipality}/{year}"
-
-        # Retrieve specified municipality from API as JSON
+    @staticmethod
+    def _fetch_json(url: str, headers: dict) -> dict[str, any]:
         r = requests.get(url, headers)
         if r.status_code != 200:
             r.raise_for_status()
         try:
             data = r.json()
         except ValueError as e:
-            raise ValueError(f"Error decoding JSON from SICA API: {e} - {r.text}")
+            raise ValueError(f"Error decoding JSON from API: {e} - {r.text}")
+        return data
+
+    def _get_municipality_id(self) -> int | None:
+        url = f"{API_URL}/community"
+        data = self._fetch_json(url, HEADERS)
+        _municipalities: dict[str, int] = {}
+        _municipalities.update(
+            {
+                item["name"]["en"].lower(): item["id"]
+                for m in data.get("data", [])
+                for item in [m] + m.get("children", [])
+                if not item["isDisabled"] and not item.get("children", [])
+            }
+        )
+        self._municipality_id = _municipalities.get(self._municipality)
+        if not self._municipality_id:
+            raise SourceArgumentNotFoundWithSuggestions(
+                "municipality", self._municipality, list(self._municipalities.keys())
+            )
 
-        # Extract collection dates
-        entries = []
-        for month in data:
-            for day in month["schedule"]:
-                pickup_date = datetime.datetime.strptime(day["date"], "%Y%m%d").date()
-                for pickup in day["pickupTypes"]:
-                    entries.append(
-                        Collection(
-                            date=pickup_date,
-                            t=pickup["name"],
-                            icon=ICON_MAP.get(pickup["name"]),
-                            picture=pickup["img"],
-                        )
-                    )
+    def fetch(self) -> list[Collection]:
+        if not self._municipality_id:
+            self._get_municipality_id()
 
+        url = f"{API_URL}/pickup-date"
+        data = self._fetch_json(url, HEADERS)
+        entries = [
+            Collection(
+                date=datetime.date.fromisoformat(e["date"].split(" ")[0]),
+                t=e["pickup_type"]["name"]["en"],
+                picture=f"{BASE_URL}{e['pickup_type']['icon']['url']}",
+                icon=ICON_MAP.get(e["pickup_type"]["name"]["en"]),
+            )
+            for e in data
+            if not e["isDisabled"] and e["community_id"] == self._municipality_id
+        ]
         return entries

doc/source/sica_lu.md

@@ -24,21 +24,22 @@ Valid values is one of the following:
 - Bertrange
 - Capellen
 - Garnich
+- Habscht
 - Holzem
 - Kehlen
 - Koerich
 - Kopstal
 - Mamer
-- Septfontaines
 - Steinfort
 
 ## Included collection types
 
+- Bulky waste
 - Clothing and Shoes
 - Glass
 - Organic waste
-- Paper - Cartons
+- Paper /Carton
 - Residual waste
 - Scrap and electrical appliances
-- Trees, shrubs and hedges
-- Valorlux packaging
+- Hedges, Shrubs and Trees
+- Valorlux - blue bag