Energyclient • Full & Extended

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

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 energyclient

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 ) """) "measurements": "power_w": 1243.5

def close(self): self.conn.close() | Problem | Strategy | |---------|----------| | Network failure | Retry with backoff (3–5 attempts) | | Rate limiting | Parse Retry-After header, queue commands | | Invalid token | Re‑authenticate automatically once | | Data gaps | Interpolate or flag missing samples | | Meter offline | Cache commands; apply when reconnected | energyclient

"timestamp": "2025-03-15T10:00:00Z", "device_id": "meter_001", "measurements": "power_w": 1243.5, "energy_wh_cumulative": 1203400, "voltage_v": 239.2, "current_a": 5.2, "frequency_hz": 50.02 , "tariff": "off_peak"

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