Tapo P110M × FastAPIで電力監視ダッシュボードを作る【SQLite記録・リアルタイム堅牢化編】

import sqlite3
from datetime import datetime
from pathlib import Path

DB_PATH = Path(__file__).parent / "power_log.db"


def init_db():
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS power_readings (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp TEXT NOT NULL,
            device_name TEXT NOT NULL,
            power_w REAL,
            energy_today_kwh REAL,
            energy_month_kwh REAL,
            voltage_v REAL,
            current_a REAL
        )
        """
    )
    conn.commit()
    return conn

def log_reading(conn, device_name, dev):
    """dev(更新済みのDeviceオブジェクト)のEnergy情報をDBに記録し、Energyモジュールを返す。"""
    energy = dev.modules["Energy"]
    conn.execute(
        """
        INSERT INTO power_readings
            (timestamp, device_name, power_w, energy_today_kwh, energy_month_kwh, voltage_v, current_a)
        VALUES (?, ?, ?, ?, ?, ?, ?)
        """,
        (
            datetime.now().isoformat(timespec="seconds"),
            device_name,
            energy.current_consumption,
            energy.consumption_today,
            energy.consumption_this_month,
            energy.voltage,
            energy.current,
        ),
    )
    conn.commit()
    return energy

import asyncio
import sys
from datetime import datetime

from power_store import init_db, log_reading
from tapo_devices import discover_with_retry, disconnect_all

DEFAULT_INTERVAL_SEC = 60


async def main():
    once = len(sys.argv) > 1 and sys.argv[1] == "once"
    interval = DEFAULT_INTERVAL_SEC
    if not once and len(sys.argv) > 1:
        interval = int(sys.argv[1])

    conn = init_db()
    devices = await discover_with_retry()
    print(f"{len(devices)}台のデバイスを検出しました。")

    try:
        while True:
            for name, dev in devices.items():
                await dev.update()
                energy = log_reading(conn, name, dev)
                now = datetime.now().strftime("%H:%M:%S")
                print(f"[{now}] {dev.alias}: {energy.current_consumption}W / 本日 {energy.consumption_today}kWh")

            if once:
                break

            await asyncio.sleep(interval)
    except (KeyboardInterrupt, asyncio.CancelledError):
        print("ポーリングを停止しました。")
    finally:
        await disconnect_all(devices)
        conn.close()


if __name__ == "__main__":
    asyncio.run(main())

python power_logger.py once

python power_logger.py 15

2台のデバイスを検出しました。
[09:11:59] Tapo P110M 2: 0.0W / 本日 0.0kWh
[09:11:59] Tapo P110M 1: 1.591W / 本日 0.01kWh
[09:12:14] Tapo P110M 2: 0.0W / 本日 0.0kWh
[09:12:15] Tapo P110M 1: 1.875W / 本日 0.01kWh
[09:12:30] Tapo P110M 2: 0.0W / 本日 0.0kWh
[09:12:30] Tapo P110M 1: 1.539W / 本日 0.01kWh
^C
ポーリングを停止しました。

import sqlite3
import sys
from pathlib import Path

DB_PATH = Path(__file__).parent / "power_log.db"


def main():
    limit = int(sys.argv[1]) if len(sys.argv) > 1 else 20

    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    rows = conn.execute(
        "SELECT * FROM power_readings ORDER BY id DESC LIMIT ?", (limit,)
    ).fetchall()
    conn.close()

    if not rows:
        print("記録がまだありません。")
        return

    for row in reversed(rows):
        print(
            f"{row['timestamp']} | {row['device_name']:6} | "
            f"{row['power_w']:>6.2f} W | 本日 {row['energy_today_kwh']:.3f} kWh | "
            f"{row['voltage_v']:.1f} V | {row['current_a']:.3f} A"
        )


if __name__ == "__main__":
    main()

python view_log.py

import asyncio
import os
from contextlib import asynccontextmanager
from datetime import datetime
from pathlib import Path

from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse, JSONResponse

from power_store import DB_PATH, init_db, log_reading
from tapo_devices import DEVICES, discover_known_devices, discover_with_retry, disconnect_all

BASE_DIR = Path(__file__).parent

POLL_INTERVAL_SEC = int(os.environ.get("TAPO_POLL_INTERVAL_SEC", "15"))  # 電力値はそこまで急変しないため既定15秒
REDISCOVER_AFTER_FAILURES = 2  # この回数だけ連続で通信に失敗したら再discoverする

devices: dict = {}
latest_status: dict = {}
consecutive_failures: dict = {}


def _status_from_device(dev, online: bool = True) -> dict:
    energy = dev.modules["Energy"]
    return {
        "label": dev.alias,
        "online": online,
        "is_on": dev.is_on,
        "power_w": energy.current_consumption,
        "energy_today_kwh": energy.consumption_today,
        "energy_month_kwh": energy.consumption_this_month,
        "voltage_v": energy.voltage,
        "current_a": energy.current,
        "updated_at": datetime.now().isoformat(timespec="seconds"),
    }


async def _redisover_one(name: str):
    """1台だけを対象にdiscoverし直す(IPアドレスが変わった場合の追従用)。"""
    old = devices.get(name)
    if old is not None:
        try:
            await old.disconnect()
        except Exception:
            pass

    fresh = await discover_known_devices([name])
    if name in fresh:
        devices[name] = fresh[name]
        consecutive_failures[name] = 0
        print(f"{DEVICES[name]['label']} を再discoverし、接続を復旧しました。")
        return devices[name]
    return None


async def poll_loop():
    conn = init_db()
    try:
        while True:
            for name in list(devices.keys()):
                dev = devices[name]
                try:
                    await dev.update()
                    log_reading(conn, name, dev)
                    latest_status[name] = _status_from_device(dev, online=True)
                    consecutive_failures[name] = 0
                except Exception as exc:
                    consecutive_failures[name] = consecutive_failures.get(name, 0) + 1
                    prev = latest_status.get(name, {})
                    latest_status[name] = {
                        **prev,
                        "label": DEVICES[name]["label"],
                        "online": False,
                        "error": str(exc),
                        "updated_at": datetime.now().isoformat(timespec="seconds"),
                    }
                    if consecutive_failures[name] >= REDISCOVER_AFTER_FAILURES:
                        await _redisover_one(name)

            await asyncio.sleep(POLL_INTERVAL_SEC)
    finally:
        conn.close()


@asynccontextmanager
async def lifespan(app: FastAPI):
    global devices
    devices = await discover_with_retry()
    for name, dev in devices.items():
        await dev.update()
        consecutive_failures[name] = 0
        latest_status[name] = _status_from_device(dev, online=True)
    task = asyncio.create_task(poll_loop())
    try:
        yield
    finally:
        task.cancel()
        await disconnect_all(devices)


app = FastAPI(lifespan=lifespan)

@app.get("/")
async def index():
    return FileResponse(BASE_DIR / "static" / "dashboard.html")


@app.get("/api/status")
async def api_status():
    return JSONResponse(latest_status)


@app.get("/api/history")
async def api_history(device: str | None = None, limit: int = 50):
    import sqlite3
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    if device:
        rows = conn.execute(
            "SELECT * FROM power_readings WHERE device_name = ? ORDER BY id DESC LIMIT ?",
            (device, limit),
        ).fetchall()
    else:
        rows = conn.execute(
            "SELECT * FROM power_readings ORDER BY id DESC LIMIT ?", (limit,)
        ).fetchall()
    conn.close()
    return [dict(row) for row in reversed(rows)]


async def _set_power(name: str, on: bool) -> dict:
    dev = devices.get(name)
    if dev is None:
        raise HTTPException(status_code=404, detail="device not found")

    try:
        if on:
            await dev.turn_on()
        else:
            await dev.turn_off()
        await dev.update()
    except Exception:
        dev = await _redisover_one(name)  # 通信失敗時はIPが変わった可能性を疑い、再discoverしてから1度だけ再試行する
        if dev is None:
            raise HTTPException(
                status_code=503,
                detail=f"{DEVICES[name]['label']} に接続できませんでした。ネットワークや電源を確認してください。",
            )
        if on:
            await dev.turn_on()
        else:
            await dev.turn_off()
        await dev.update()

    consecutive_failures[name] = 0
    latest_status[name] = _status_from_device(dev, online=True)
    return latest_status[name]


@app.post("/api/device/{name}/on")
async def device_on(name: str):
    if name not in DEVICES:
        raise HTTPException(status_code=404, detail="unknown device")
    return await _set_power(name, True)


@app.post("/api/device/{name}/off")
async def device_off(name: str):
    if name not in DEVICES:
        raise HTTPException(status_code=404, detail="unknown device")
    return await _set_power(name, False)


@app.post("/api/bulk/on")
async def bulk_on():
    return {name: await _set_power(name, True) for name in devices}


@app.post("/api/bulk/off")
async def bulk_off():
    return {name: await _set_power(name, False) for name in devices}

async function refresh() {
  const status = await api('/api/status');
  const cardsEl = document.getElementById('cards');
  cardsEl.innerHTML = '';

  for (const [name, s] of Object.entries(status)) {
    const card = el('div', { class: 'card' });
    const online = s.online !== false;
    const badgeClass = !online ? 'offline' : (s.is_on ? 'on' : 'off');
    const badgeText = !online ? 'OFFLINE' : (s.is_on ? 'ON' : 'OFF');
    const badge = el('span', { class: `badge ${badgeClass}`, text: badgeText });
    card.appendChild(el('h2', {}, [document.createTextNode(s.label + ' '), badge]));

    const metrics = el('div', { class: 'metrics' });
    metrics.innerHTML = `
      <div><span>瞬間電力</span><b>${online ? s.power_w.toFixed(2) + ' W' : '--'}</b></div>
      <div><span>本日の電力量</span><b>${online ? s.energy_today_kwh.toFixed(3) + ' kWh' : '--'}</b></div>
    `;
    card.appendChild(metrics);

    const canvas = el('canvas', {});
    card.appendChild(canvas);
    cardsEl.appendChild(card);

    api(`/api/history?device=${name}&limit=30`).then(history => {
      drawChart(canvas, history.map(r => r.power_w));
    });
  }
}

refresh();
setInterval(refresh, 5000);

python web_dashboard.py

Tapo P110M PDU ダッシュボードを起動します...
このPCのブラウザで http://127.0.0.1:8000/ を開いてください(PC専用運用)。
終了するにはこのウィンドウでCtrl+Cを押してください。
INFO:     Started server process [xxxxx]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

.badge.offline { background: #fdecea; color: #b3261e; }

@echo off
cd /d "%~dp0"
python web_dashboard.py
pause