#!/usr/bin/env python3 """Local-only SGCC configuration kit. Python 3.11+, standard library only.""" import argparse import datetime as dt import hashlib import json import math import os from pathlib import Path import re import sys import tempfile import time import urllib.error import urllib.request OWNED = "sgcc-blog-kit" METRICS = { "balance": ("Balance", "CNY", "monetary"), "month_usage": ("Last billed month energy", "kWh", "energy"), "month_charge": ("Last billed month cost", "CNY", "monetary"), "year_usage": ("Year to date energy", "kWh", "energy"), "year_charge": ("Year to date cost", "CNY", "monetary"), } def digest(data): return hashlib.sha256(data).hexdigest() def merge_day(rows, date, value, keep=120): """Reference rule for dated daily observations; never invent missing days.""" day = dt.date.fromisoformat(date) if day.isoformat() != date or day > dt.date.today(): raise ValueError("Invalid or future business date") number = float(value) if not math.isfinite(number) or number < 0: raise ValueError("Daily energy must be finite and nonnegative") values = {r["date"]: float(r["kwh"]) for r in rows} values[date] = number return [{"date": d, "kwh": values[d]} for d in sorted(values)[-keep:]] def load_mapping(filename): data = json.loads(Path(filename).read_text(encoding="utf-8")) alias = data.get("alias", "home") if not re.fullmatch(r"[a-z][a-z0-9_]{0,25}", alias): raise ValueError("alias must use lowercase ASCII letters, digits, underscore") sources = data.get("sources", {}) if set(sources) != {"daily", *METRICS}: raise ValueError("sources must contain exactly daily and the five summary metrics") if len(set(sources.values())) != 6: raise ValueError("Each metric needs a distinct source entity") for entity in sources.values(): if not re.fullmatch(r"sensor\.[a-z0-9_]+", entity): raise ValueError("Invalid source entity ID") attr = data.get("daily_date_attribute", "last_reset") if not re.fullmatch(r"[a-z][a-z0-9_]*", attr): raise ValueError("Invalid daily date attribute") keep = data.get("keep_days", 120) if type(keep) is not int or not 14 <= keep <= 366: raise ValueError("keep_days must be between 14 and 366") if type(data.get("yearly_counter_verified", False)) is not bool: raise ValueError("yearly_counter_verified must be true or false") return {**data, "alias": alias, "keep_days": keep, "daily_date_attribute": attr} def generated(mapping): alias, sources = mapping["alias"], mapping["sources"] daily = sources["daily"] prefix = "sgcc_kit_" + alias attr = mapping["daily_date_attribute"] valid_date = ( "{% set d = state_attr('" + daily + "', '" + attr + "') %}" "{% set parsed = as_datetime(d, none) %}" "{{ d is string and d | regex_match('^[0-9]{4}-[0-9]{2}-[0-9]{2}$') " "and parsed is not none and parsed.date() <= now().date() " "and is_number(states('" + daily + "')) and states('" + daily + "') | float >= 0 }}" ) rows_template = ( "{% set d = state_attr('" + daily + "', '" + attr + "') %}\n" "{% set v = states('" + daily + "') | float %}\n" "{% set old = this.attributes.history | default([], true) %}\n" "{% set rows = (old | rejectattr('date', 'eq', d) | list " "+ [{'date': d, 'kwh': v}]) | sort(attribute='date') | list %}\n" ) blocks = [ "# Generated by SGCC local kit. Daily snapshots deliberately have no state_class.\n" "template:\n - triggers:\n" f" - trigger: state\n entity_id: {daily}\n" " - trigger: time_pattern\n minutes: '/5'\n" " conditions:\n - condition: template\n value_template: >-\n" " " + valid_date + "\n sensor:\n" f" - name: 'SGCC {alias} dated daily energy'\n" f" default_entity_id: sensor.{prefix}_daily\n" f" unique_id: {prefix}_daily\n" " unit_of_measurement: kWh\n device_class: energy\n" " state: >-\n" + indent(rows_template + "{{ rows[-1].kwh }}", 10) + "\n" " attributes:\n data_date: >-\n" + indent(rows_template + "{{ rows[-1].date }}", 12) + "\n" " history: >-\n" + indent(rows_template + "{{ rows[-" + str(mapping["keep_days"]) + ":] }}", 12) + "\n" ] for key, (label, unit, device_class) in METRICS.items(): src = sources[key] # A failed/empty source cannot overwrite a valid cached reading with zero. state = "{{ states('" + src + "') if is_number(states('" + src + "')) else this.state }}" block = ( " - triggers:\n" f" - trigger: state\n entity_id: {src}\n" " - trigger: time_pattern\n minutes: '/5'\n" " sensor:\n" f" - name: 'SGCC {alias} {label}'\n" f" default_entity_id: sensor.{prefix}_{key}\n" f" unique_id: {prefix}_{key}\n" f" unit_of_measurement: {unit}\n device_class: {device_class}\n" " state: >-\n " + state + "\n" " attributes:\n" " source_present: \"{{ is_number(states('" + src + "')) }}\"\n" ) if key == "year_usage" and mapping.get("yearly_counter_verified", False): block += " state_class: total_increasing\n" blocks.append(block) blocks.append( "lovelace:\n dashboards:\n sgcc-kit:\n mode: yaml\n" " title: Electricity\n icon: mdi:lightning-bolt\n" " show_in_sidebar: true\n require_admin: false\n" f" filename: {OWNED}/dashboard.yaml\n" ) dashboard = ( "title: Electricity\nviews:\n - title: Electricity\n path: home\n cards:\n" " - type: entities\n title: SGCC cached readings\n entities:\n" ) for key in ["daily", *METRICS]: dashboard += f" - sensor.{prefix}_{key}\n" dashboard += ( " - type: markdown\n title: Dated daily readings\n content: |-\n" " Daily energy is delayed billing data, not live power.\n\n" " Date | Energy (kWh)\n --- | ---:\n" " {% for row in (state_attr('sensor." + prefix + "_daily', 'history') or []) | reverse %}\n" " {{ row.date }} | {{ row.kwh }}\n {% endfor %}\n" " - type: markdown\n content: |-\n" " Latest utility date: {{ state_attr('sensor." + prefix + "_daily', 'data_date') }}\n\n" " Cached values are not proof of a successful new collection. Check collector logs too.\n" ) return "".join(blocks), dashboard def indent(text, size): return "\n".join(" " * size + line for line in text.splitlines()) def attach(config): """Preserve existing YAML byte structure; refuse unsupported include layouts.""" line = f" sgcc_blog_kit: !include {OWNED}/package.yaml\n" if line in config: return config, f"{OWNED}/package.yaml" if re.search(r"(?m)^\s*sgcc_blog_kit:", config): raise ValueError("Existing sgcc_blog_kit key is not owned by this kit") if re.search(r"(?m)^homeassistant:\s*!", config): raise ValueError("External homeassistant include: add package manually") roots = list(re.finditer(r"(?m)^homeassistant:\s*(?:#.*)?$", config)) if len(roots) > 1: raise ValueError("Duplicate homeassistant blocks") if not roots: return config.rstrip() + "\n\nhomeassistant:\n packages:\n" + line, f"{OWNED}/package.yaml" start = roots[0].end() following = re.search(r"(?m)^\S[^\n]*:", config[start:]) end = start + following.start() if following else len(config) section = config[start:end] named = re.search(r"(?m)^ packages: !include_dir_named ([A-Za-z0-9_./-]+)\s*$", section) if named: folder = Path(named[1]) if folder.is_absolute() or ".." in folder.parts: raise ValueError("Package directory must stay inside HA config") return config, (folder / "sgcc_blog_kit.yaml").as_posix() inline = re.search(r"(?m)^ packages:\s*(?:#.*)?$", section) if inline: pos = start + inline.end() return config[:pos] + "\n" + line.rstrip("\n") + config[pos:], f"{OWNED}/package.yaml" if re.search(r"\bpackages:", section): raise ValueError("Unsupported packages layout: add include manually") return config[:end].rstrip() + "\n packages:\n" + line + config[end:], f"{OWNED}/package.yaml" def atomic(path, data): path.parent.mkdir(parents=True, exist_ok=True) fd, name = tempfile.mkstemp(prefix=".sgcc-", dir=path.parent) try: with os.fdopen(fd, "wb") as f: f.write(data) os.chmod(name, 0o600) os.replace(name, path) finally: if os.path.exists(name): os.unlink(name) def api(path, body=None): base = os.environ.get("HOME_ASSISTANT_URL", "").rstrip("/") token = os.environ.get("HOME_ASSISTANT_TOKEN", "") if not base.startswith(("http://", "https://")) or not token: raise ValueError("Set HOME_ASSISTANT_URL and HOME_ASSISTANT_TOKEN privately") req = urllib.request.Request(base + path, data=None if body is None else json.dumps(body).encode(), headers={"Authorization": "Bearer " + token, "Content-Type": "application/json"}) # Talk directly to the user's HA; do not send HA credentials through a system proxy. with urllib.request.build_opener(urllib.request.ProxyHandler({})).open(req, timeout=20) as r: return json.load(r) def audit(mapping): states = {x["entity_id"]: x for x in api("/api/states")} report = {"sources": {}, "mirrors": {}} for key, entity in mapping["sources"].items(): src = states.get(entity, {}) valid = False try: valid = math.isfinite(float(src.get("state", "nan"))) except (ValueError, TypeError): pass expected_unit = "kWh" if key in ["daily", "month_usage", "year_usage"] else "CNY" report["sources"][key] = {"present": bool(src), "numeric": valid, "unit_matches": src.get("attributes", {}).get("unit_of_measurement") == expected_unit, "unit": src.get("attributes", {}).get("unit_of_measurement"), "state_class": src.get("attributes", {}).get("state_class")} for key in ["daily", *METRICS]: src = states.get(f"sensor.sgcc_kit_{mapping['alias']}_{key}", {}) report["mirrors"][key] = {"present": bool(src), "available": bool(src) and src.get("state") not in ("unknown", "unavailable")} daily = states.get(mapping["sources"]["daily"], {}) date = daily.get("attributes", {}).get(mapping["daily_date_attribute"]) try: if not isinstance(date, str) or not re.fullmatch(r"[0-9]{4}-[0-9]{2}-[0-9]{2}", date): raise ValueError("Daily date must use YYYY-MM-DD") report["daily_age_days"] = (dt.date.today() - dt.date.fromisoformat(date)).days except (ValueError, TypeError): report["daily_age_days"] = None report["collector_login_verified"] = False return report def apply(config_dir, mapping): root = Path(config_dir).resolve() cfg = root / "configuration.yaml" before = cfg.read_bytes() updated, package_path = attach(before.decode("utf-8")) package, dashboard = generated(mapping) files = {"configuration.yaml": updated.encode(), package_path: package.encode(), f"{OWNED}/dashboard.yaml": dashboard.encode()} for relative in files: target = root / relative if not target.resolve().is_relative_to(root) or target.is_symlink(): raise ValueError("Refusing an out-of-scope or symlinked target") if target.exists() and relative != "configuration.yaml": # An exact match is an idempotent rerun. Never overwrite someone else's files. if target.read_bytes() != files[relative]: raise ValueError("A generated target already exists with different content") if all((root / p).exists() and (root / p).read_bytes() == data for p, data in files.items()): print("UNCHANGED: configuration already matches; no files changed") return None snapshot = root / ".sgcc-blog-kit-backups" / (dt.datetime.now().strftime("%Y%m%dT%H%M%S") + "-" + os.urandom(3).hex()) snapshot.mkdir(parents=True, mode=0o700) os.chmod(snapshot.parent, 0o700) manifest = {} for relative, data in files.items(): target = root / relative previous = target.read_bytes() if target.exists() else None manifest[relative] = {"existed": previous is not None, "applied_sha256": digest(data), "before_sha256": digest(previous) if previous is not None else None} if previous is not None: atomic(snapshot / relative, previous) atomic(snapshot / "manifest.json", json.dumps(manifest, indent=2).encode()) for relative, data in files.items(): target = root / relative entry = manifest[relative] current = target.read_bytes() if target.exists() else None if (digest(current) if current is not None else None) != entry["before_sha256"]: raise ValueError("Concurrent configuration edit detected before write") written = [] try: for relative, data in files.items(): atomic(root / relative, data) written.append(relative) check = api("/api/config/core/check_config", {}) if check.get("result") != "valid": raise ValueError("HA rejected configuration") except Exception: for relative in reversed(written): if manifest[relative]["existed"]: atomic(root / relative, (snapshot / relative).read_bytes()) else: (root / relative).unlink() raise print("APPLIED and HA configuration valid. Rollback snapshot:", snapshot) return snapshot def rollback(root, snapshot): root, snapshot = Path(root).resolve(), Path(snapshot).resolve() if snapshot.parent != root / ".sgcc-blog-kit-backups": raise ValueError("Rollback snapshot must belong to this HA configuration directory") manifest = json.loads((snapshot / "manifest.json").read_text()) for relative, entry in manifest.items(): target = root / relative if not target.resolve().is_relative_to(root) or target.is_symlink(): raise ValueError("Unsafe rollback target") if not target.exists() or digest(target.read_bytes()) != entry["applied_sha256"]: raise ValueError("Files changed since apply; refusing to overwrite later edits") if entry["existed"] and digest((snapshot / relative).read_bytes()) != entry["before_sha256"]: raise ValueError("Backup checksum mismatch") for relative, entry in manifest.items(): if entry["existed"]: atomic(root / relative, (snapshot / relative).read_bytes()) else: (root / relative).unlink() print("ROLLED BACK owned configuration; reload or restart HA separately") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("mode", choices=["plan", "apply", "check", "rollback"]) parser.add_argument("--mapping", default="mapping.local.json") parser.add_argument("--config-dir", default=os.environ.get("HA_CONFIG_DIR")) parser.add_argument("--snapshot") parser.add_argument("--restart", action="store_true", help="Explicitly restart HA after validated apply") parser.add_argument("--require-mirrors", action="store_true", help="check also requires all cached sensors to have data") args = parser.parse_args() if args.mode == "rollback": if not args.config_dir or not args.snapshot: raise ValueError("rollback requires --config-dir and --snapshot") rollback(args.config_dir, args.snapshot) return mapping = load_mapping(args.mapping) if args.mode == "plan": package, dashboard = generated(mapping) target = Path("generated") target.mkdir(exist_ok=True) atomic(target / "package.yaml", package.encode()) atomic(target / "dashboard.yaml", dashboard.encode()) print("PLAN: generated/package.yaml and generated/dashboard.yaml; HA unchanged") elif args.mode == "check": report = audit(mapping) print(json.dumps(report, indent=2)) if not all(x["present"] and x["numeric"] and x["unit_matches"] for x in report["sources"].values()): raise ValueError("Some source states are missing, nonnumeric, or have the wrong unit") if report["daily_age_days"] is None or report["daily_age_days"] < 0: raise ValueError("Daily business date is missing, invalid, or in the future") if args.require_mirrors and not all(x["available"] for x in report["mirrors"].values()): raise ValueError("Some mirrors have not received a valid reading yet") elif args.mode == "apply": if not args.config_dir: raise ValueError("Set HA_CONFIG_DIR to your writable HA configuration mount") # Authenticate before writing any file. api("/api/") snapshot = apply(args.config_dir, mapping) if args.restart: try: api("/api/services/homeassistant/restart", {}) except (urllib.error.URLError, ConnectionError, TimeoutError): pass # A disconnect can accompany a successful restart; never infer success. time.sleep(3) for _ in range(60): try: if api("/api/config").get("state") == "RUNNING": print("HA returned RUNNING. Run check after the collector's next push.") break except (urllib.error.URLError, ConnectionError, TimeoutError): pass time.sleep(3) else: raise ValueError("HA restart did not become healthy; inspect HA and snapshot") elif snapshot: print("Restart HA when ready to load the new sidebar dashboard and package") if __name__ == "__main__": try: main() except Exception as error: # Never dump an HTTP response, URL, token, or household readings into a public log. print("ERROR:", str(error) if isinstance(error, ValueError) else type(error).__name__, file=sys.stderr) sys.exit(1)