From 056c33fe8c8f83c52d0745163fb8370e748735ce Mon Sep 17 00:00:00 2001 From: rolf Date: Wed, 4 Feb 2026 19:42:44 +0000 Subject: [PATCH] Delete sensor.py --- sensor.py | 237 ------------------------------------------------------ 1 file changed, 237 deletions(-) delete mode 100644 sensor.py diff --git a/sensor.py b/sensor.py deleted file mode 100644 index 09050a2..0000000 --- a/sensor.py +++ /dev/null @@ -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()