Delete sensor.py

sensor.py

@@ -1,237 +0,0 @@
-"""Sensor platform for Uster Waste."""
-import asyncio
-import logging
-from datetime import datetime, timedelta
-from typing import Optional
-
-import aiohttp
-from bs4 import BeautifulSoup
-from homeassistant.components.sensor import SensorEntity
-from homeassistant.config_entries import ConfigEntry
-from homeassistant.core import HomeAssistant
-from homeassistant.helpers.entity_platform import AddEntitiesCallback
-from homeassistant.helpers.aiohttp_client import async_get_clientsession
-from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
-
-from .const import (
-    DOMAIN,
-    ATTR_NEXT_COLLECTION,
-    ATTR_DATE,
-    ATTR_TYPE,
-    ATTR_DAYS_UNTIL,
-    ATTR_ENTRIES,
-    ATTR_ERROR,
-    MANUAL_REFRESH,
-)
-
-_LOGGER = logging.getLogger(__name__)
-
-SCAN_INTERVAL = timedelta(days=1)
-
-# Swiss date helpers
-MONTH_MAP = {
-    "Jan": "01", "Feb": "02", "Mrz": "03", "Mär": "03",
-    "Apr": "04", "Mai": "05", "Jun": "06", "Jul": "07",
-    "Aug": "08", "Sep": "09", "Okt": "10", "Nov": "11", "Dez": "12"
-}
-
-
-def _parse_date(date_str: str) -> Optional[datetime]:
-    """Convert Swiss date string (e.g., '24.10.2023' or '24. Okt. 2023') to datetime."""
-    date_str = date_str.strip()
-    # Normalize Swiss month abbreviations
-    for key, value in MONTH_MAP.items():
-        date_str = date_str.replace(key, value)
-    # Try formats: dd.mm.yyyy, d.m.yy
-    for fmt in ["%d.%m.%Y", "%d.%m.%y"]:
-        try:
-            return datetime.strptime(date_str, fmt)
-        except ValueError:
-            continue
-    _LOGGER.warning(f"Could not parse date: '{date_str}'")
-    return None
-
-
-async def async_setup_entry(
-    hass: HomeAssistant,
-    entry: ConfigEntry,
-    async_add_entities: AddEntitiesCallback,
-) -> None:
-    """Set up the sensor."""
-    config = entry.data
-    token = config["token"]
-    waste_id = config["id"]
-    name = config.get("name", "Uster Waste")
-
-    session = async_get_clientsession(hass)
-    coordinator = UsterWasteDataUpdateCoordinator(hass, session, token, waste_id)
-
-    entity = UsterWasteSensor(
-        entry_id=entry.entry_id,
-        coordinator=coordinator,
-        name=name
-    )
-    async_add_entities([entity], update_before_add=True)
-
-
-class UsterWasteDataUpdateCoordinator:
-    """Fetch data from Uster website."""
-
-    def __init__(
-        self,
-        hass: HomeAssistant,
-        session: aiohttp.ClientSession,
-        token: str,
-        waste_id: str,
-    ):
-        self.hass = hass
-        self.session = session
-        self.token = token
-        self.waste_id = waste_id
-        self.data = None
-        self.last_error = None
-        self.last_updated = None
-
-    async def async_update(self) -> dict:
-        """Fetch data (cache if valid)."""
-        # 1. Check cache (valid for 24h)
-        if self.last_updated and (datetime.now() - self.last_updated) < SCAN_INTERVAL:
-            return self.data
-
-        url = (
-            "https://www.uster.ch/abfallstrassenabschnitt"
-            f"?strassenabschnitt%5B_token%5D={self.token}"
-            f"&strassenabschnitt%5BstrassenabschnittId%5D={self.waste_id}"
-        )
-
-        try:
-            async with self.session.get(url, timeout=10) as response:
-                if response.status == 403 or response.status == 404:
-                    raise Exception(
-                        "Token expired or invalid. "
-                        "Please get a fresh URL from https://www.uster.ch/abfallstrassenabschnitt"
-                    )
-                response.raise_for_status()
-                html = await response.text()
-
-            # Parse HTML
-            soup = BeautifulSoup(html, "html.parser")
-            table = soup.find("table", class_="table table-striped")
-            if not table:
-                table = soup.find("table")
-                if not table:
-                    raise ValueError("No table found on page.")
-
-            rows = table.find_all("tr")
-            if len(rows) < 2:
-                raise ValueError("Table has no data rows.")
-
-            entries = []
-            now = datetime.now()
-
-            for row in rows[1:4]:  # Next 3 entries
-                cols = row.find_all("td")
-                if len(cols) < 2:
-                    continue
-
-                collection_type = cols[0].get_text(strip=True)
-                date_str = cols[1].get_text(strip=True).replace(" \u00a0", " ")  # Clean no-break space
-                dt = _parse_date(date_str)
-                if not dt:
-                    _LOGGER.warning(f"Skipping row with invalid date: {date_str}")
-                    continue
-
-                entries.append({
-                    "Sammlung": collection_type,
-                    "Wann?": date_str,
-                    "date_obj": dt,
-                    "days_until": (dt - now).days
-                })
-
-            # Sort by date (ascending)
-            entries.sort(key=lambda x: x["date_obj"])
-
-            self.data = {
-                "next_collection": entries[0]["Sammlung"] if entries else None,
-                "date": entries[0]["Wann?"] if entries else None,
-                "type": entries[0]["Sammlung"] if entries else None,
-                "days_until": entries[0]["days_until"] if entries else None,
-                "entries": [
-                    {
-                        "type": e["Sammlung"],
-                        "date": e["Wann?"],
-                        "days_until": e["days_until"]
-                    }
-                    for e in entries[:3]
-                ]
-            }
-            self.last_updated = datetime.now()
-
-        except Exception as e:
-            _LOGGER.error("Error fetching Uster data: %s", e)
-            self.data = {
-                ATTR_ERROR: str(e),
-                "next_collection": None,
-                "date": None,
-                "entries": []
-            }
-            self.last_error = str(e)
-
-        return self.data
-
-
-class UsterWasteSensor(SensorEntity):
-    """Uster Waste Sensor Entity."""
-
-    _attr_has_entity_name = True
-    _attr_icon = "mdi:recycle"
-    _attr_device_class = None
-
-    def __init__(
-        self,
-        entry_id: str,
-        coordinator: UsterWasteDataUpdateCoordinator,
-        name: str
-    ):
-        self._entry_id = entry_id
-        self.coordinator = coordinator
-        self._attr_name = f"{name} Schedule"
-        self._attr_unique_id = f"uster_waste_{entry_id}"
-        self._attr_extra_state_attributes = {
-            ATTR_ENTRIES: [],
-        }
-
-    async def async_update(self):
-        """Update sensor state."""
-        self._attr_native_value = len(self.coordinator.data.get("entries", []))
-        self._attr_extra_state_attributes.update(
-            {
-                ATTR_NEXT_COLLECTION: self.coordinator.data.get("next_collection"),
-                ATTR_DATE: self.coordinator.data.get("date"),
-                ATTR_TYPE: self.coordinator.data.get("type"),
-                ATTR_DAYS_UNTIL: self.coordinator.data.get("days_until"),
-                ATTR_ENTRIES: self.coordinator.data.get("entries", []),
-                ATTR_ERROR: self.coordinator.data.get("error")
-            }
-        )
-
-    async def async_added_to_hass(self):
-        """When entity is added to hass."""
-        await super().async_added_to_hass()
-        self.async_on_remove(
-            self.coordinator.async_add_listener(
-                self.async_write_ha_state, self.coordinator.last_updated
-            )
-        )
-        # Force first update
-        await self.async_update()
-
-    @property
-    def available(self) -> bool:
-        """Return if entity is available."""
-        return self.coordinator.last_updated is not None
-
-    async def async_press(self):
-        """Handle the button press (manual refresh)."""
-        await self.async_update()
-        self.async_write_ha_state()