Tapo P110Mの消費電力をExcelレポート化する【年次集計・電気代試算・セキュリティ判断編】

from io import BytesIO
from openpyxl import Workbook
from fastapi.responses import StreamingResponse

@app.get("/api/export")
async def export_excel(device: str | None = None):
    """消費電力量の記録をExcel(.xlsx)としてダウンロードする。省エネ参考資料としての利用を想定。"""
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    if device:
        if device not in DEVICES:
            raise HTTPException(status_code=404, detail="unknown device")
        rows = conn.execute(
            "SELECT * FROM power_readings WHERE device_name = ? ORDER BY id", (device,)
        ).fetchall()
    else:
        rows = conn.execute("SELECT * FROM power_readings ORDER BY id").fetchall()
    conn.close()

    wb = Workbook()
    ws = wb.active
    ws.title = "消費電力履歴"
    headers = ["日時", "デバイス", "瞬間電力(W)", "本日の電力量(kWh)", "当月の電力量(kWh)", "電圧(V)", "電流(A)"]
    ws.append(headers)
    for row in rows:
        label = DEVICES.get(row["device_name"], {}).get("label", row["device_name"])
        ws.append([
            row["timestamp"], label, row["power_w"],
            row["energy_today_kwh"], row["energy_month_kwh"],
            row["voltage_v"], row["current_a"],
        ])

    for col, width in zip("ABCDEFG", [20, 16, 14, 18, 18, 12, 12]):
        ws.column_dimensions[col].width = width
    ws.freeze_panes = "A2"

    buf = BytesIO()
    wb.save(buf)
    buf.seek(0)

    filename = f"tapo_power_history_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
    return StreamingResponse(
        buf,
        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
    )

query = """
    SELECT
        date(timestamp) AS day,
        device_name,
        MAX(energy_today_kwh) AS daily_kwh,
        AVG(power_w) AS avg_w,
        MAX(power_w) AS max_w
    FROM power_readings
    WHERE strftime('%Y', timestamp) = ? AND strftime('%m', timestamp) = ?
    GROUP BY day, device_name
    ORDER BY day, device_name
"""

DEFAULT_UNIT_PRICE_YEN = 31.0

@app.get("/api/export/yearly")
async def export_yearly(year: int, device: str | None = None, unit_price: float = DEFAULT_UNIT_PRICE_YEN):
    """指定した年の1〜12月タブを持つExcelを生成する。各タブは日次集計(電力量合計・平均/最大電力・参考金額)。"""
    if device is not None and device not in DEVICES:
        raise HTTPException(status_code=404, detail="unknown device")

    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row

    wb = Workbook()
    wb.remove(wb.active)
    note = (
        f"※参考単価: 1kWhあたり{unit_price:g}円(目安)で試算した金額です。"
        "実際の電気料金は契約している電力会社・プランにより異なります。"
    )
    headers = ["日付", "デバイス", "電力量合計(kWh)", "平均電力(W)", "最大電力(W)", "参考金額(円)"]

    for month in range(1, 13):
        sheet = wb.create_sheet(title=f"{year}年{month}月")
        sheet.append([note])
        sheet.merge_cells(start_row=1, start_column=1, end_row=1, end_column=len(headers))
        sheet.append(headers)

        # (このあと日次集計クエリを実行し、1日1行ずつ書き込む)

        sheet.freeze_panes = "A3"
    conn.close()
    # ...(Excelをメモリ上に保存してダウンロードさせる処理は/api/exportと同様)

daily_kwh = round(row["daily_kwh"], 3) if row["daily_kwh"] is not None else None
sheet.append([
    row["day"], label, daily_kwh,
    round(row["avg_w"], 2) if row["avg_w"] is not None else None,
    round(row["max_w"], 2) if row["max_w"] is not None else None,
    round(daily_kwh * unit_price) if daily_kwh is not None else None,
])

<input type="number" id="yearInput">
<label>参考単価
  <input type="number" id="unitPriceInput" value="31" step="0.1"> 円/kWh
</label>
<button onclick="exportYearly()">年別レポート(日次集計・全12ヶ月タブ)</button>
<button onclick="exportExcel()">全期間の生データをエクスポート</button>