commit 9d7179308fd58d8d6582ce00d716844c3786360b Author: Richard Kacerek Date: Sun Sep 6 10:37:34 2026 +0200 Initial pump monitor with Coolify deployment diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..02f3baa --- /dev/null +++ b/.dockerignore @@ -0,0 +1,7 @@ +.env +.git +.pytest_cache +__pycache__ +*.pyc +*.sqlite3 +tests diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..def3742 --- /dev/null +++ b/.env.example @@ -0,0 +1,16 @@ +# Add these values as Coolify environment variables/secrets. +TAPO_HOST=192.168.1.50 +TAPO_USERNAME=your-tapo-account@example.com +TAPO_PASSWORD=replace-with-a-coolify-secret + +TZ=Europe/Prague +POLL_INTERVAL_SECONDS=2 +START_WATTS=100 +STOP_WATTS=30 +CONFIRM_SAMPLES=2 +MAX_GAP_SECONDS=20 +SAMPLE_RETENTION_DAYS=30 + +DASHBOARD_PORT=8000 +DASHBOARD_USERNAME=water-monitor +DASHBOARD_PASSWORD=replace-with-a-long-random-password diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..bcc8290 --- /dev/null +++ b/.gitignore @@ -0,0 +1,16 @@ +.env +.env.* +!.env.example +.DS_Store +.idea/ +.vscode/ +.venv/ +venv/ +__pycache__/ +.pytest_cache/ +*.py[cod] +*.sqlite +*.sqlite3 +*.sqlite3-shm +*.sqlite3-wal +data/ diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..13e3aa8 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,24 @@ +FROM python:3.12.11-slim + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + DATABASE_PATH=/data/pump-monitor.sqlite3 + +WORKDIR /srv/app + +COPY requirements.txt . +RUN pip install --no-cache-dir --requirement requirements.txt \ + && addgroup --system --gid 10001 monitor \ + && adduser --system --uid 10001 --ingroup monitor monitor \ + && mkdir -p /data \ + && chown monitor:monitor /data + +COPY --chown=monitor:monitor app ./app + +USER monitor +EXPOSE 8000 + +HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \ + CMD python -c "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/health', timeout=3)" || exit 1 + +CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1", "--proxy-headers", "--forwarded-allow-ips", "*"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..ae27585 --- /dev/null +++ b/README.md @@ -0,0 +1,95 @@ +# Tapo P110 pump-cycle monitor + +A small, private monitor for a domestic waterworks pump connected through a Tapo P110. It reads power locally, records the first and last threshold crossing for every run, and presents the results in a phone-friendly dashboard. + +## What it records + +- Start and stop time for every pump cycle +- Duration, average power, peak power, and estimated energy per cycle +- A rolling 24-hour power trace +- Daily cycle count, total runtime, average duration, and longest duration +- Partial-cycle markers after restarts or communication gaps +- CSV export of completed cycles + +With the default two-second polling interval, event times have approximately two-second resolution. Two consecutive readings confirm a transition, but the saved event time is the first threshold crossing. + +## Run locally with Docker + +1. Open the local `.env` file in this directory. +2. Enter `TAPO_USERNAME` and `TAPO_PASSWORD` using the account from the Tapo app. +3. Run `docker compose up --detach --build` from this directory. +4. Open [http://localhost:8000](http://localhost:8000). + +The dashboard is bound to `127.0.0.1`, so it is only available on this computer. Recorded data is stored in a named Docker volume and survives container rebuilds. + +Required variables: + +| Variable | Value | +| --- | --- | +| `TAPO_HOST` | The plug's reserved LAN address, for example `192.168.1.50` | +| `TAPO_USERNAME` | The email address used for the Tapo account | +| `TAPO_PASSWORD` | The Tapo account password; keep it only in the ignored local `.env` file | +| `DASHBOARD_PASSWORD` | Optional locally; use a long password if the dashboard is ever exposed beyond localhost | + +Useful defaults already included in `compose.yaml`: + +| Variable | Default | Meaning | +| --- | ---: | --- | +| `TZ` | `Europe/Prague` | Boundary used for daily totals | +| `POLL_INTERVAL_SECONDS` | `2` | Time between readings | +| `START_WATTS` | `100` | Two readings at or above this start a cycle | +| `STOP_WATTS` | `30` | Two readings at or below this stop a cycle | +| `CONFIRM_SAMPLES` | `2` | Consecutive readings required | +| `MAX_GAP_SECONDS` | `20` | Longer data gaps split and mark partial cycles | +| `SAMPLE_RETENTION_DAYS` | `30` | Raw chart history retention; cycle records remain | +| `DASHBOARD_PORT` | `8000` | Local dashboard port | +| `DASHBOARD_USERNAME` | `water-monitor` | Dashboard login name when a password is configured | + +Do not paste real credentials into `.env.example`, `compose.yaml`, Git, or application logs. The `.env` file is excluded by `.gitignore` and `.dockerignore`. + +## Deploy with Coolify and Gitea + +Use `compose.coolify.yaml` for the Coolify deployment. It exposes container port 8000 to Coolify's proxy without publishing a host port and stores SQLite data in the persistent `tapo-pump-monitor-data` Docker volume. + +1. Create an application in Coolify from this Gitea repository. +2. Select Docker Compose and set the compose file to `/compose.coolify.yaml`. +3. Add `TAPO_HOST`, `TAPO_USERNAME`, `TAPO_PASSWORD`, and `DASHBOARD_PASSWORD` as Coolify environment variables. Do not save their real values in Git. +4. Add a domain in Coolify and route it to the `pump-monitor` service on port 8000. +5. Deploy and confirm that the service becomes healthy and can reach the plug on the home LAN. + +Keep only one instance running. Stop the local Docker deployment at cutover so the same pump cycle is not recorded twice. The local SQLite history can be copied into the Coolify volume before the first server start if it needs to be preserved. + +## Before starting + +- Reserve the plug's LAN address in the router so DHCP does not change it. +- In the Tapo app, enable **Me > Third-Party Services > Third-Party Compatibility**. Newer P110 firmware otherwise advertises the unsupported TPAP-only local protocol. +- Keep exactly one instance of this service. Multiple instances would record duplicate cycles. + +Docker's normal bridge network can initiate connections to devices on the home LAN, so host networking is not required. + +## Calibrate the thresholds + +The defaults are intentionally conservative. After deployment: + +1. Watch the live wattage while the pump is idle. +2. Open a tap and note the stable running wattage. +3. Set `START_WATTS` comfortably below the running value but above all idle noise. +4. Set `STOP_WATTS` above idle consumption and below `START_WATTS`. + +For example, if idle is 1 W and the pump is around 750 W, the defaults of 100 W to start and 30 W to stop are suitable. + +## Useful commands + +```bash +docker compose up --detach --build # build and start +docker compose ps # show health and port +docker compose logs --follow # watch connection and cycle events +docker compose restart # restart without deleting data +docker compose down # stop; recorded data remains +``` + +Do not add `--volumes` to `docker compose down` unless you intentionally want to erase all recorded history. + +## Interpreting likely problems + +Once a normal baseline is established, useful warning signals include unusually short repeated cycles, a run much longer than normal, a sharp increase in cycles per hour, or a meaningful change in average running wattage. The dashboard records the evidence first; alert rules can be added after several days of normal data show appropriate limits. diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..b08e047 --- /dev/null +++ b/app/__init__.py @@ -0,0 +1 @@ +"""Tapo P110 pump-cycle monitor.""" diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..1409b30 --- /dev/null +++ b/app/config.py @@ -0,0 +1,84 @@ +from __future__ import annotations + +import os +from dataclasses import dataclass +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + + +def _float_env(name: str, default: float, minimum: float) -> float: + raw = os.getenv(name, str(default)) + try: + value = float(raw) + except ValueError as exc: + raise ValueError(f"{name} must be a number") from exc + if value < minimum: + raise ValueError(f"{name} must be at least {minimum}") + return value + + +def _int_env(name: str, default: int, minimum: int) -> int: + raw = os.getenv(name, str(default)) + try: + value = int(raw) + except ValueError as exc: + raise ValueError(f"{name} must be an integer") from exc + if value < minimum: + raise ValueError(f"{name} must be at least {minimum}") + return value + + +@dataclass(frozen=True) +class Settings: + tapo_host: str + tapo_username: str + tapo_password: str + timezone_name: str + poll_interval_seconds: float + start_watts: float + stop_watts: float + confirm_samples: int + max_gap_seconds: float + sample_retention_days: int + database_path: str + dashboard_username: str + dashboard_password: str + + @classmethod + def from_env(cls) -> "Settings": + timezone_name = os.getenv("TZ", "Europe/Prague") + try: + ZoneInfo(timezone_name) + except ZoneInfoNotFoundError as exc: + raise ValueError(f"TZ is not a known timezone: {timezone_name}") from exc + + start_watts = _float_env("START_WATTS", 100.0, 0.1) + stop_watts = _float_env("STOP_WATTS", 30.0, 0.0) + if stop_watts >= start_watts: + raise ValueError("STOP_WATTS must be lower than START_WATTS") + + poll_interval = _float_env("POLL_INTERVAL_SECONDS", 2.0, 0.5) + max_gap = _float_env("MAX_GAP_SECONDS", max(20.0, poll_interval * 5), poll_interval * 2) + + return cls( + tapo_host=os.getenv("TAPO_HOST", "").strip(), + tapo_username=os.getenv("TAPO_USERNAME", "").strip(), + tapo_password=os.getenv("TAPO_PASSWORD", ""), + timezone_name=timezone_name, + poll_interval_seconds=poll_interval, + start_watts=start_watts, + stop_watts=stop_watts, + confirm_samples=_int_env("CONFIRM_SAMPLES", 2, 1), + max_gap_seconds=max_gap, + sample_retention_days=_int_env("SAMPLE_RETENTION_DAYS", 30, 1), + database_path=os.getenv("DATABASE_PATH", "/data/pump-monitor.sqlite3"), + dashboard_username=os.getenv("DASHBOARD_USERNAME", "water-monitor").strip(), + dashboard_password=os.getenv("DASHBOARD_PASSWORD", ""), + ) + + @property + def credentials_configured(self) -> bool: + return bool(self.tapo_username and self.tapo_password) + + @property + def timezone(self) -> ZoneInfo: + return ZoneInfo(self.timezone_name) diff --git a/app/detector.py b/app/detector.py new file mode 100644 index 0000000..d5eca02 --- /dev/null +++ b/app/detector.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from typing import Any + +from .storage import Storage + + +@dataclass(frozen=True) +class PowerSample: + sampled_at: datetime + watts: float + + +class CycleDetector: + """Detect pump runs using two thresholds and consecutive-sample confirmation.""" + + def __init__(self, storage: Storage, start_watts: float, stop_watts: float, confirm_samples: int): + self.storage = storage + self.start_watts = start_watts + self.stop_watts = stop_watts + self.confirm_samples = confirm_samples + self.running = False + self.cycle_id: int | None = None + self.pending_start: list[PowerSample] = [] + self.pending_stop: list[PowerSample] = [] + self.first_observation = True + self.pending_incomplete_start = False + self.last_sample_at: datetime | None = None + + def process(self, sample: PowerSample) -> list[dict[str, Any]]: + events: list[dict[str, Any]] = [] + self.last_sample_at = sample.sampled_at + + if not self.running: + if sample.watts >= self.start_watts: + if not self.pending_start: + self.pending_incomplete_start = self.first_observation + self.pending_start.append(sample) + if len(self.pending_start) >= self.confirm_samples: + started_at = self.pending_start[0].sampled_at + self.cycle_id = self.storage.start_cycle(started_at, self.pending_incomplete_start) + self.running = True + events.append( + { + "type": "started", + "cycle_id": self.cycle_id, + "at": started_at, + "incomplete_start": self.pending_incomplete_start, + } + ) + self.pending_start.clear() + self.pending_incomplete_start = False + else: + self.pending_start.clear() + self.pending_incomplete_start = False + else: + if sample.watts <= self.stop_watts: + self.pending_stop.append(sample) + if len(self.pending_stop) >= self.confirm_samples: + ended_at = self.pending_stop[0].sampled_at + cycle_id = self.cycle_id + if cycle_id is None: + raise RuntimeError("Running detector has no cycle id") + cycle = self.storage.finish_cycle(cycle_id, ended_at) + events.append({"type": "stopped", "cycle": cycle, "at": ended_at}) + self.running = False + self.cycle_id = None + self.pending_stop.clear() + else: + self.pending_stop.clear() + + self.first_observation = False + return events + + def interrupt(self, ended_at: datetime | None = None) -> dict[str, Any] | None: + event = None + if self.running and self.cycle_id is not None: + final_time = ended_at or self.last_sample_at + if final_time is not None: + cycle = self.storage.finish_cycle(self.cycle_id, final_time, incomplete_end=True) + event = {"type": "interrupted", "cycle": cycle, "at": final_time} + self.running = False + self.cycle_id = None + self.pending_start.clear() + self.pending_stop.clear() + self.pending_incomplete_start = False + self.first_observation = True + self.last_sample_at = None + return event diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..982b8bb --- /dev/null +++ b/app/main.py @@ -0,0 +1,155 @@ +from __future__ import annotations + +import base64 +import csv +import io +import logging +import secrets +from contextlib import asynccontextmanager +from datetime import UTC, datetime, time, timedelta + +from fastapi import FastAPI, Query, Request +from fastapi.responses import HTMLResponse, PlainTextResponse, Response + +from .config import Settings +from .monitor import TapoMonitor +from .storage import Storage, from_db_time, to_db_time +from .web import DASHBOARD_HTML + +logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s") +settings = Settings.from_env() +storage = Storage(settings.database_path) +monitor = TapoMonitor(settings, storage) + + +@asynccontextmanager +async def lifespan(_: FastAPI): + storage.open() + await monitor.start() + try: + yield + finally: + await monitor.stop() + storage.close() + + +app = FastAPI(title="Tapo Pump Monitor", docs_url=None, redoc_url=None, lifespan=lifespan) + + +@app.middleware("http") +async def protect_dashboard(request: Request, call_next): + if request.url.path == "/health" or not settings.dashboard_password: + return await call_next(request) + + header = request.headers.get("Authorization", "") + valid = False + if header.startswith("Basic "): + try: + decoded = base64.b64decode(header[6:], validate=True).decode("utf-8") + username, password = decoded.split(":", 1) + valid = secrets.compare_digest(username, settings.dashboard_username) and secrets.compare_digest( + password, settings.dashboard_password + ) + except (ValueError, UnicodeDecodeError): + valid = False + if not valid: + return Response(status_code=401, headers={"WWW-Authenticate": 'Basic realm="Pump monitor"'}) + return await call_next(request) + + +def local_day_bounds() -> tuple[datetime, datetime]: + now_local = datetime.now(settings.timezone) + start_local = datetime.combine(now_local.date(), time.min, tzinfo=settings.timezone) + return start_local.astimezone(UTC), (start_local + timedelta(days=1)).astimezone(UTC) + + +@app.get("/", response_class=HTMLResponse) +async def dashboard() -> str: + return DASHBOARD_HTML + + +@app.get("/health") +async def health() -> dict: + return {"service": "ok", "plug_state": monitor.state} + + +@app.get("/api/status") +async def api_status() -> dict: + day_start, day_end = local_day_bounds() + return { + "monitor": monitor.snapshot(), + "today": storage.summary_between(day_start, day_end), + "settings": { + "start_watts": settings.start_watts, + "stop_watts": settings.stop_watts, + "poll_interval_seconds": settings.poll_interval_seconds, + "timezone": settings.timezone_name, + }, + "security": {"dashboard_auth_enabled": bool(settings.dashboard_password)}, + } + + +@app.get("/api/cycles") +async def api_cycles(limit: int = Query(default=100, ge=1, le=1000)) -> dict: + return {"cycles": storage.recent_cycles(limit)} + + +@app.get("/api/history") +async def api_history(hours: int = Query(default=24, ge=1, le=168)) -> dict: + window_end = datetime.now(UTC) + window_start = window_end - timedelta(hours=hours) + cycles = storage.cycles_between(window_start, window_end) + preserve_ranges = [ + ( + from_db_time(cycle["started_at"]), + from_db_time(cycle["ended_at"]) if cycle["ended_at"] else window_end, + ) + for cycle in cycles + ] + return { + "window_start": to_db_time(window_start), + "window_end": to_db_time(window_end), + "samples": storage.sample_history( + window_start, + window_end, + max_points=800, + preserve_ranges=preserve_ranges, + ), + "cycles": cycles, + } + + +@app.get("/api/cycles.csv") +async def export_cycles(days: int = Query(default=30, ge=1, le=3650)) -> Response: + end = datetime.now(UTC) + rows = storage.cycles_between(end - timedelta(days=days), end) + output = io.StringIO() + writer = csv.writer(output) + writer.writerow( + [ + "started_at_utc", + "ended_at_utc", + "duration_seconds", + "average_watts", + "maximum_watts", + "energy_wh", + "partial_cycle", + ] + ) + for row in reversed(rows): + writer.writerow( + [ + row["started_at"], + row["ended_at"], + round(float(row["duration_seconds"] or 0), 3), + round(float(row["average_watts"] or 0), 3), + round(float(row["maximum_watts"] or 0), 3), + round(float(row["energy_wh"] or 0), 5), + bool(row["incomplete_start"] or row["incomplete_end"]), + ] + ) + return PlainTextResponse( + output.getvalue(), + media_type="text/csv", + headers={"Content-Disposition": 'attachment; filename="pump-cycles.csv"'}, + ) diff --git a/app/monitor.py b/app/monitor.py new file mode 100644 index 0000000..acb3c8a --- /dev/null +++ b/app/monitor.py @@ -0,0 +1,186 @@ +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), + } diff --git a/app/storage.py b/app/storage.py new file mode 100644 index 0000000..281761e --- /dev/null +++ b/app/storage.py @@ -0,0 +1,245 @@ +from __future__ import annotations + +import math +import sqlite3 +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any + + +def utc_now() -> datetime: + return datetime.now(UTC) + + +def to_db_time(value: datetime) -> str: + return value.astimezone(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z") + + +def from_db_time(value: str) -> datetime: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + + +class Storage: + def __init__(self, path: str): + self.path = path + self.db: sqlite3.Connection | None = None + + def open(self) -> None: + Path(self.path).parent.mkdir(parents=True, exist_ok=True) + self.db = sqlite3.connect(self.path, check_same_thread=False) + self.db.row_factory = sqlite3.Row + self.db.execute("PRAGMA journal_mode=WAL") + self.db.execute("PRAGMA synchronous=NORMAL") + self.db.executescript( + """ + CREATE TABLE IF NOT EXISTS samples ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + sampled_at TEXT NOT NULL, + watts REAL NOT NULL, + voltage REAL, + current_amps REAL, + socket_on INTEGER NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_samples_at ON samples(sampled_at); + + CREATE TABLE IF NOT EXISTS cycles ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + started_at TEXT NOT NULL, + ended_at TEXT, + duration_seconds REAL, + average_watts REAL, + maximum_watts REAL, + energy_wh REAL, + sample_count INTEGER, + incomplete_start INTEGER NOT NULL DEFAULT 0, + incomplete_end INTEGER NOT NULL DEFAULT 0 + ); + CREATE INDEX IF NOT EXISTS idx_cycles_started ON cycles(started_at); + """ + ) + self.db.commit() + + def close(self) -> None: + if self.db is not None: + self.db.close() + self.db = None + + def _conn(self) -> sqlite3.Connection: + if self.db is None: + raise RuntimeError("Database is not open") + return self.db + + def record_sample( + self, + sampled_at: datetime, + watts: float, + voltage: float | None, + current_amps: float | None, + socket_on: bool, + ) -> None: + db = self._conn() + db.execute( + "INSERT INTO samples(sampled_at, watts, voltage, current_amps, socket_on) VALUES (?, ?, ?, ?, ?)", + (to_db_time(sampled_at), watts, voltage, current_amps, int(socket_on)), + ) + db.commit() + + def start_cycle(self, started_at: datetime, incomplete_start: bool = False) -> int: + db = self._conn() + cursor = db.execute( + "INSERT INTO cycles(started_at, incomplete_start) VALUES (?, ?)", + (to_db_time(started_at), int(incomplete_start)), + ) + db.commit() + return int(cursor.lastrowid) + + def finish_cycle(self, cycle_id: int, ended_at: datetime, incomplete_end: bool = False) -> dict[str, Any]: + db = self._conn() + cycle = db.execute("SELECT * FROM cycles WHERE id = ?", (cycle_id,)).fetchone() + if cycle is None: + raise RuntimeError(f"Cycle {cycle_id} does not exist") + + started_at = from_db_time(cycle["started_at"]) + if ended_at < started_at: + ended_at = started_at + start_text = to_db_time(started_at) + end_text = to_db_time(ended_at) + rows = db.execute( + "SELECT sampled_at, watts FROM samples WHERE sampled_at >= ? AND sampled_at < ? ORDER BY sampled_at", + (start_text, end_text), + ).fetchall() + + values = [(from_db_time(row["sampled_at"]), float(row["watts"])) for row in rows] + duration = max(0.0, (ended_at - started_at).total_seconds()) + average = sum(watts for _, watts in values) / len(values) if values else 0.0 + maximum = max((watts for _, watts in values), default=0.0) + energy_wh = self._integrate_energy(values, ended_at) + + db.execute( + """ + UPDATE cycles + SET ended_at = ?, duration_seconds = ?, average_watts = ?, maximum_watts = ?, + energy_wh = ?, sample_count = ?, incomplete_end = ? + WHERE id = ? + """, + (end_text, duration, average, maximum, energy_wh, len(values), int(incomplete_end), cycle_id), + ) + db.commit() + return self.get_cycle(cycle_id) + + @staticmethod + def _integrate_energy(values: list[tuple[datetime, float]], ended_at: datetime) -> float: + if not values: + return 0.0 + watt_seconds = 0.0 + for index, (sampled_at, watts) in enumerate(values): + segment_end = values[index + 1][0] if index + 1 < len(values) else ended_at + seconds = max(0.0, (segment_end - sampled_at).total_seconds()) + watt_seconds += watts * seconds + return watt_seconds / 3600.0 + + def get_cycle(self, cycle_id: int) -> dict[str, Any]: + row = self._conn().execute("SELECT * FROM cycles WHERE id = ?", (cycle_id,)).fetchone() + if row is None: + raise RuntimeError(f"Cycle {cycle_id} does not exist") + return dict(row) + + def close_open_cycles(self) -> int: + db = self._conn() + open_rows = db.execute("SELECT id, started_at FROM cycles WHERE ended_at IS NULL").fetchall() + closed = 0 + for row in open_rows: + last_sample = db.execute( + "SELECT sampled_at FROM samples WHERE sampled_at >= ? ORDER BY sampled_at DESC LIMIT 1", + (row["started_at"],), + ).fetchone() + ended_at = from_db_time(last_sample["sampled_at"]) if last_sample else from_db_time(row["started_at"]) + self.finish_cycle(int(row["id"]), ended_at, incomplete_end=True) + closed += 1 + return closed + + def recent_cycles(self, limit: int = 100) -> list[dict[str, Any]]: + rows = self._conn().execute( + "SELECT * FROM cycles WHERE ended_at IS NOT NULL ORDER BY started_at DESC LIMIT ?", + (max(1, min(limit, 1000)),), + ).fetchall() + return [dict(row) for row in rows] + + def cycles_between(self, start: datetime, end: datetime) -> list[dict[str, Any]]: + rows = self._conn().execute( + """ + SELECT * FROM cycles + WHERE started_at < ? AND COALESCE(ended_at, ?) >= ? + ORDER BY started_at DESC + """, + (to_db_time(end), to_db_time(end), to_db_time(start)), + ).fetchall() + return [dict(row) for row in rows] + + def summary_between(self, start: datetime, end: datetime) -> dict[str, Any]: + cycles = self.cycles_between(start, end) + durations: list[float] = [] + energy_wh = 0.0 + for cycle in cycles: + cycle_start = max(start, from_db_time(cycle["started_at"])) + cycle_end_raw = from_db_time(cycle["ended_at"]) if cycle["ended_at"] else min(end, utc_now()) + cycle_end = min(end, cycle_end_raw) + durations.append(max(0.0, (cycle_end - cycle_start).total_seconds())) + energy_wh += float(cycle["energy_wh"] or 0.0) + return { + "cycle_count": len(cycles), + "runtime_seconds": sum(durations), + "average_duration_seconds": sum(durations) / len(durations) if durations else 0.0, + "longest_duration_seconds": max(durations, default=0.0), + "energy_wh": energy_wh, + } + + def sample_history( + self, + since: datetime, + until: datetime | None = None, + max_points: int = 800, + preserve_ranges: list[tuple[datetime, datetime]] | None = None, + ) -> list[dict[str, Any]]: + until = until or utc_now() + rows = self._conn().execute( + "SELECT sampled_at, watts FROM samples WHERE sampled_at >= ? AND sampled_at <= ? ORDER BY sampled_at", + (to_db_time(since), to_db_time(until)), + ).fetchall() + + if len(rows) <= max_points: + selected = list(rows) + else: + # First/minimum/maximum/last preserves short power spikes that a + # simple stride can skip entirely. All readings inside known pump + # cycles are also retained for accurate hover details. + bucket_count = max(1, max_points // 4) + bucket_size = max(1, math.ceil(len(rows) / bucket_count)) + selected_by_time: dict[str, sqlite3.Row] = {} + for offset in range(0, len(rows), bucket_size): + bucket = rows[offset : offset + bucket_size] + for row in ( + bucket[0], + min(bucket, key=lambda item: float(item["watts"])), + max(bucket, key=lambda item: float(item["watts"])), + bucket[-1], + ): + selected_by_time[row["sampled_at"]] = row + + ranges = preserve_ranges or [] + if ranges: + normalized = [(start.astimezone(UTC), end.astimezone(UTC)) for start, end in ranges] + for row in rows: + sampled_at = from_db_time(row["sampled_at"]) + if any(start <= sampled_at <= end for start, end in normalized): + selected_by_time[row["sampled_at"]] = row + + selected = sorted(selected_by_time.values(), key=lambda row: row["sampled_at"]) + return [{"sampled_at": row["sampled_at"], "watts": row["watts"]} for row in selected] + + def cleanup_samples(self, retention_days: int) -> int: + cutoff = to_db_time(utc_now() - timedelta(days=retention_days)) + db = self._conn() + cursor = db.execute("DELETE FROM samples WHERE sampled_at < ?", (cutoff,)) + db.commit() + return int(cursor.rowcount) diff --git a/app/web.py b/app/web.py new file mode 100644 index 0000000..01e06ee --- /dev/null +++ b/app/web.py @@ -0,0 +1,456 @@ +DASHBOARD_HTML = r""" + + + + + + Waterworks · Pump monitor + + + +
+
+
+
Domestic waterworks / live telemetry
+

Pump monitor

+
+
Connecting…
+
+ +
+ +
+
+
Last completed run
+
+
—
+
duration
+
+
+
Started—
+
Stopped—
+
Average—
+
Peak—
+
+
+ +
+
Cycles today
—
+
Runtime today
—
+
Average cycle
—
+
Longest cycle
—
+
+
+ +
+
+
Last 24 hours

Power trace

+
+
+
+ +
+ + + + + + + + +
+
+ +
No readings yet
+
+
+ +
+
+
Recorded events

Recent pump cycles

+ Download CSV +
+
+ + + +
StartedStoppedDurationAveragePeak
No completed cycles yet
+
+
+
Awaiting first readingTimes shown in your browser timezone
+
+ + + +""" diff --git a/compose.coolify.yaml b/compose.coolify.yaml new file mode 100644 index 0000000..b1ce913 --- /dev/null +++ b/compose.coolify.yaml @@ -0,0 +1,28 @@ +services: + pump-monitor: + build: + context: . + dockerfile: Dockerfile + restart: unless-stopped + expose: + - "8000" + environment: + - TAPO_HOST=${TAPO_HOST} + - TAPO_USERNAME=${TAPO_USERNAME} + - TAPO_PASSWORD=${TAPO_PASSWORD} + - TZ=${TZ:-Europe/Prague} + - POLL_INTERVAL_SECONDS=${POLL_INTERVAL_SECONDS:-2} + - START_WATTS=${START_WATTS:-100} + - STOP_WATTS=${STOP_WATTS:-30} + - CONFIRM_SAMPLES=${CONFIRM_SAMPLES:-2} + - MAX_GAP_SECONDS=${MAX_GAP_SECONDS:-20} + - SAMPLE_RETENTION_DAYS=${SAMPLE_RETENTION_DAYS:-30} + - DATABASE_PATH=/data/pump-monitor.sqlite3 + - DASHBOARD_USERNAME=${DASHBOARD_USERNAME:-water-monitor} + - DASHBOARD_PASSWORD=${DASHBOARD_PASSWORD} + volumes: + - pump-monitor-data:/data + +volumes: + pump-monitor-data: + name: tapo-pump-monitor-data diff --git a/compose.yaml b/compose.yaml new file mode 100644 index 0000000..c35a8f9 --- /dev/null +++ b/compose.yaml @@ -0,0 +1,25 @@ +services: + pump-monitor: + build: . + restart: unless-stopped + ports: + - "127.0.0.1:${DASHBOARD_PORT:-8000}:8000" + environment: + - TAPO_HOST=${TAPO_HOST} + - TAPO_USERNAME=${TAPO_USERNAME} + - TAPO_PASSWORD=${TAPO_PASSWORD} + - TZ=${TZ:-Europe/Prague} + - POLL_INTERVAL_SECONDS=${POLL_INTERVAL_SECONDS:-2} + - START_WATTS=${START_WATTS:-100} + - STOP_WATTS=${STOP_WATTS:-30} + - CONFIRM_SAMPLES=${CONFIRM_SAMPLES:-2} + - MAX_GAP_SECONDS=${MAX_GAP_SECONDS:-20} + - SAMPLE_RETENTION_DAYS=${SAMPLE_RETENTION_DAYS:-30} + - DATABASE_PATH=/data/pump-monitor.sqlite3 + - DASHBOARD_USERNAME=${DASHBOARD_USERNAME:-water-monitor} + - DASHBOARD_PASSWORD=${DASHBOARD_PASSWORD} + volumes: + - pump-monitor-data:/data + +volumes: + pump-monitor-data: diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..c5bdedf --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +fastapi==0.141.1 +python-kasa==0.10.2 +uvicorn==0.52.4 diff --git a/tests/test_detector.py b/tests/test_detector.py new file mode 100644 index 0000000..ea2fe37 --- /dev/null +++ b/tests/test_detector.py @@ -0,0 +1,107 @@ +from datetime import UTC, datetime, timedelta + +from app.detector import CycleDetector, PowerSample +from app.storage import Storage, to_db_time + + +def at(second: int) -> datetime: + return datetime(2026, 9, 5, 10, 0, second, tzinfo=UTC) + + +def make_storage(tmp_path) -> Storage: + storage = Storage(str(tmp_path / "monitor.sqlite3")) + storage.open() + return storage + + +def feed(storage: Storage, detector: CycleDetector, second: int, watts: float): + sampled_at = at(second) + storage.record_sample(sampled_at, watts, 230.0, 0.0, True) + return detector.process(PowerSample(sampled_at, watts)) + + +def test_records_confirmed_cycle_using_first_threshold_crossings(tmp_path): + storage = make_storage(tmp_path) + detector = CycleDetector(storage, start_watts=100, stop_watts=30, confirm_samples=2) + + feed(storage, detector, 0, 2) + assert feed(storage, detector, 1, 120) == [] + started = feed(storage, detector, 2, 710) + assert started[0]["type"] == "started" + assert detector.running + + feed(storage, detector, 3, 700) + assert feed(storage, detector, 4, 10) == [] + stopped = feed(storage, detector, 5, 3) + + assert stopped[0]["type"] == "stopped" + cycle = stopped[0]["cycle"] + assert cycle["started_at"] == "2026-09-05T10:00:01.000Z" + assert cycle["ended_at"] == "2026-09-05T10:00:04.000Z" + assert cycle["duration_seconds"] == 3.0 + assert cycle["maximum_watts"] == 710.0 + assert cycle["incomplete_start"] == 0 + assert cycle["incomplete_end"] == 0 + + +def test_hysteresis_ignores_noise_between_thresholds(tmp_path): + storage = make_storage(tmp_path) + detector = CycleDetector(storage, start_watts=100, stop_watts=30, confirm_samples=2) + + feed(storage, detector, 0, 0) + feed(storage, detector, 1, 105) + feed(storage, detector, 2, 80) + assert not detector.running + + feed(storage, detector, 3, 150) + feed(storage, detector, 4, 160) + assert detector.running + feed(storage, detector, 5, 45) + assert detector.running + feed(storage, detector, 6, 25) + feed(storage, detector, 7, 40) + assert detector.running + + +def test_initial_running_state_and_interruption_are_marked_partial(tmp_path): + storage = make_storage(tmp_path) + detector = CycleDetector(storage, start_watts=100, stop_watts=30, confirm_samples=2) + + feed(storage, detector, 0, 600) + feed(storage, detector, 1, 610) + event = detector.interrupt(at(2)) + + assert event is not None + cycle = event["cycle"] + assert cycle["incomplete_start"] == 1 + assert cycle["incomplete_end"] == 1 + + +def test_open_cycle_summary_uses_current_time_not_end_of_day(tmp_path, monkeypatch): + storage = make_storage(tmp_path) + start = datetime.now(UTC) - timedelta(seconds=12) + storage.start_cycle(start) + summary = storage.summary_between(start - timedelta(minutes=1), start + timedelta(days=1)) + + assert 11 <= summary["runtime_seconds"] <= 14 + + +def test_history_downsampling_preserves_peak_and_cycle_samples(tmp_path): + storage = make_storage(tmp_path) + start = at(0) + for index in range(120): + sampled_at = start + timedelta(seconds=index) + watts = 1700.0 if index == 61 else (1100.0 if 60 <= index <= 64 else 0.0) + storage.record_sample(sampled_at, watts, 230.0, 0.0, True) + + history = storage.sample_history( + start, + start + timedelta(seconds=119), + max_points=20, + preserve_ranges=[(start + timedelta(seconds=60), start + timedelta(seconds=64))], + ) + + retained = {row["sampled_at"]: row["watts"] for row in history} + assert max(retained.values()) == 1700.0 + for second in range(60, 65): + assert to_db_time(start + timedelta(seconds=second)) in retained