from __future__ import annotations import asyncio import logging import time from datetime import UTC, datetime from typing import Any from kasa import Discover, Module from .config import Settings from .detector import CycleDetector, PowerSample from .storage import Storage, to_db_time logger = logging.getLogger("pump-monitor") class TapoMonitor: def __init__(self, settings: Settings, storage: Storage): self.settings = settings self.storage = storage self.detector = CycleDetector( storage, start_watts=settings.start_watts, stop_watts=settings.stop_watts, confirm_samples=settings.confirm_samples, ) self.device: Any = None self.task: asyncio.Task[None] | None = None self.state = "starting" self.error: str | None = None self.watts: float | None = None self.voltage: float | None = None self.current_amps: float | None = None self.socket_on: bool | None = None self.last_success: datetime | None = None self.started_at = datetime.now(UTC) self._last_error_key: str | None = None async def start(self) -> None: stale_count = self.storage.close_open_cycles() if stale_count: logger.warning("Closed %d interrupted cycle(s) left open by a previous run", stale_count) self.storage.cleanup_samples(self.settings.sample_retention_days) self.task = asyncio.create_task(self._run(), name="tapo-power-poller") async def stop(self) -> None: if self.task is not None: self.task.cancel() try: await self.task except asyncio.CancelledError: pass self.task = None self.detector.interrupt(self.last_success) await self._disconnect() async def _connect(self) -> None: if not self.settings.tapo_host: raise RuntimeError("Tapo host has not been configured") if not self.settings.credentials_configured: raise RuntimeError("Tapo credentials have not been configured") self.device = await Discover.discover_single( self.settings.tapo_host, username=self.settings.tapo_username, password=self.settings.tapo_password, discovery_timeout=5, timeout=5, ) if self.device is None: raise RuntimeError("Tapo plug was not found at the configured address") await self.device.update() if Module.Energy not in self.device.modules: raise RuntimeError("The connected Tapo device does not expose energy monitoring") logger.info("Connected to %s (%s)", self.device.alias or "Tapo plug", self.device.model or "unknown model") async def _disconnect(self) -> None: if self.device is not None: try: await self.device.disconnect() except Exception: pass self.device = None async def _run(self) -> None: reconnect_delay = 2.0 cleanup_at = time.monotonic() while True: iteration_started = time.monotonic() try: if self.device is None: await self._connect() await self.device.update() energy = self.device.modules[Module.Energy] watts = float(energy.current_consumption) voltage = self._optional_float(energy.voltage) current_amps = self._optional_float(energy.current) sampled_at = datetime.now(UTC) if self.last_success is not None: gap = (sampled_at - self.last_success).total_seconds() if gap > self.settings.max_gap_seconds: event = self.detector.interrupt(self.last_success) if event: logger.warning("Cycle interrupted by a %.1f-second monitoring gap", gap) socket_on = bool(self.device.is_on) self.storage.record_sample(sampled_at, watts, voltage, current_amps, socket_on) events = self.detector.process(PowerSample(sampled_at, watts)) for event in events: if event["type"] == "started": logger.info("Pump started at %s", to_db_time(event["at"])) elif event["type"] == "stopped": cycle = event["cycle"] logger.info("Pump stopped after %.1f seconds", cycle["duration_seconds"]) self.watts = watts self.voltage = voltage self.current_amps = current_amps self.socket_on = socket_on self.last_success = sampled_at self.state = "running" if self.detector.running else "idle" self.error = None self._last_error_key = None reconnect_delay = 2.0 if time.monotonic() >= cleanup_at: self.storage.cleanup_samples(self.settings.sample_retention_days) cleanup_at = time.monotonic() + 86400 except asyncio.CancelledError: raise except Exception as exc: error_key = f"{type(exc).__name__}:{exc}" if error_key != self._last_error_key: logger.warning("Cannot read Tapo power data: %s", exc) self._last_error_key = error_key tapo_configured = bool(self.settings.tapo_host and self.settings.credentials_configured) self.state = "configuration_required" if not tapo_configured else "offline" self.error = self._friendly_error(exc) self.watts = None self.voltage = None self.current_amps = None self.socket_on = None if self.last_success is not None: gap = (datetime.now(UTC) - self.last_success).total_seconds() if gap > self.settings.max_gap_seconds and self.detector.running: self.detector.interrupt(self.last_success) await self._disconnect() await asyncio.sleep(reconnect_delay) reconnect_delay = min(30.0, reconnect_delay * 1.7) elapsed = time.monotonic() - iteration_started await asyncio.sleep(max(0.1, self.settings.poll_interval_seconds - elapsed)) @staticmethod def _optional_float(value: Any) -> float | None: return None if value is None else float(value) @staticmethod def _friendly_error(exc: Exception) -> str: text = str(exc) lowered = text.lower() if "tpap" in lowered and "unsupported device" in lowered: return ( "This P110 firmware requires Third-Party Compatibility. " "In the Tapo app, open Me > Third-Party Services > Third-Party Compatibility and enable it." ) if "host has not been configured" in lowered: return "The plug address has not been configured. Add TAPO_HOST in Coolify." if "authentication" in lowered or "credentials" in lowered or "login" in lowered: return "Tapo authentication failed. Check the Coolify username and password secrets." if "not found" in lowered or "timed out" in lowered or "timeout" in lowered: return "The plug is not reachable from the server. Check its address and LAN access." return text[:240] or type(exc).__name__ def snapshot(self) -> dict[str, Any]: return { "state": self.state, "error": self.error, "watts": self.watts, "voltage": self.voltage, "current_amps": self.current_amps, "socket_on": self.socket_on, "last_success": to_db_time(self.last_success) if self.last_success else None, "monitor_started_at": to_db_time(self.started_at), }