Energyclient 【HD 2027】

import requests import sqlite3 import time from dataclasses import dataclass from datetime import datetime, timezone @dataclass class EnergyClient: api_root: str meter_id: str token: str

def __post_init__(self): self._init_db() self.session = requests.Session() self.session.headers.update("Authorization": f"Bearer self.token") energyclient

def fetch_latest(self): resp = self.session.get(f"self.api_root/meters/self.meter_id/reading") resp.raise_for_status() data = resp.json() # Cache it self.conn.execute( "INSERT OR REPLACE INTO readings (ts, power_w, cumulative_wh) VALUES (?, ?, ?)", (datetime.now(timezone.utc).isoformat(), data["power_w"], data["cumulative_wh"]) ) self.conn.commit() return data import requests import sqlite3 import time from dataclasses

def control_load(self, state: bool): """Turn load on/off (demand response)""" resp = self.session.post( f"self.api_root/meters/self.meter_id/control", json="load_control": state ) return resp.status_code == 200 cumulative_wh) VALUES (?

def _init_db(self): self.conn = sqlite3.connect("energy_cache.db") self.conn.execute(""" CREATE TABLE IF NOT EXISTS readings ( ts TEXT PRIMARY KEY, power_w REAL, cumulative_wh REAL ) """)