moving to HACS format
This commit is contained in:
Rolf
237
custom_components/uster_waste/sensor.py
Normal file
237
custom_components/uster_waste/sensor.py
Normal file
@@ -0,0 +1,237 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user