diff --git a/docs/RE_DART_FINDINGS.md b/docs/RE_DART_FINDINGS.md
index 21b32dc..ee68dfd 100644
--- a/docs/RE_DART_FINDINGS.md
+++ b/docs/RE_DART_FINDINGS.md
@@ -366,6 +366,30 @@ consistent with all-silent probes.
   socket/HTTP hotspot APIs (`/terminal/*`, `wifi_iot` join) take it from
   there. Same envelope + cipher on both transports (§5, §9).
 
+## 16. Live power readings
+
+- `queryDeviceRealTimeData` — what the home page polls
+  (`DeviceRealTimeDataMixin`, `startPollingDeviceRealTimeData`): cmd
+  `deviceDetailInfo` (socket `1301`), no arguments on socket
+  (`{"refreshData": 2}` is added only for mqtt mode). Reply `dataArea`:
+  `apowerCount`, **`totalPower`** (the household-draw headline).
+- `queryLoadStatusInfo` — cmd `loadStatus` (socket `1411`), no arguments.
+  Reply per smart switch: `Sw1Volt`/`Sw2Volt`, `SW1Curr`, `SW1ExpPower`
+  (`SW2…`), `SW1ExpEnergy`, `index`. (`loadStatus2`, socket `1415`, is the
+  newer variant.)
+- `queryEmsInfoBySocket` — cmd `ems` (socket `1707`), no arguments. Reply
+  carries `gridVol1/2`, `gridCurr1/2`, `loadCurr1/2`, `gridFreq`, relay
+  states. Household draw ≈
+  `(loadCurr1 + loadCurr2) × (gridVol1 + gridVol2) / 2`
+  (library: `power.ems_load_watts`).
+- `InstallSunRunSummaryEntity` (`homeloadPower`, `solarPower`,
+  `gridPower`, `apowersPower`) is parsed in the bean factory layer, not
+  off the local socket — cloud/HTTP-side entity.
+
+Library: `power.get_power()` tries detail → load → EMS in order
+(`franklinwh_agate_client/power.py`); `LocalDeviceClient` gained
+`query_detail_info()` / `query_load_status()`.
+
 ## 9. Bluetooth (BLE) transport — `communication_center/bluetooth`
 
 - `BluetoothManager` (singleton), `BluetoothStatus`: `closed / connected / connecting / interrupted`.
diff --git a/franklinwh_agate_client/device_local.py b/franklinwh_agate_client/device_local.py
index 99a5681..38e74a1 100644
--- a/franklinwh_agate_client/device_local.py
+++ b/franklinwh_agate_client/device_local.py
@@ -289,6 +289,23 @@ class LocalDeviceClient:
     def query_device_info(self) -> Dict[str, Any]:
         return self.send_rpc(CommunicationCmd.QUERY_DEVICE_INFO)
 
+    def query_detail_info(self) -> Dict[str, Any]:
+        """``queryDeviceRealTimeData`` — realtime headline (socket id 1301).
+
+        No arguments.  The reply ``dataArea`` carries ``totalPower``
+        (the household draw the home page polls) and ``apowerCount``.
+        """
+        return self.send_rpc(CommunicationCmd.DEVICE_DETAIL_INFO)
+
+    def query_load_status(self) -> Dict[str, Any]:
+        """``queryLoadStatusInfo`` — per smart-switch watts (socket id 1411).
+
+        No arguments.  The reply ``dataArea`` carries per-channel
+        ``Sw1Volt``/``SW1Curr``/``SW1ExpPower`` (and ``SW2…``) plus
+        ``index``.
+        """
+        return self.send_rpc(CommunicationCmd.LOAD_STATUS)
+
     def query_bms_info(self, pack_id: int = 1) -> Dict[str, Any]:
         """``queryBmsInfoBySocket`` — BMS pack query (``arguments={"id"}``).
 
diff --git a/franklinwh_agate_client/power.py b/franklinwh_agate_client/power.py
new file mode 100644
index 0000000..1e5b573
--- /dev/null
+++ b/franklinwh_agate_client/power.py
@@ -0,0 +1,181 @@
+"""Live household power over the local socket.
+
+The app's home page polls ``queryDeviceRealTimeData`` (socket
+``DEVICE_DETAIL_INFO``) for its energy-flow screen; per-switch watts come
+from ``queryLoadStatusInfo`` (socket ``LOAD_STATUS``); grid/load
+current+voltage telemetry comes from ``queryEmsInfoBySocket``.  See
+``docs/RE_DART_FINDINGS.md`` §4 (ids) and §16.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import socket
+from typing import Any, Dict, List, Optional, Tuple
+
+from .battery import _walk
+from .device_local import LocalDeviceClient
+
+log = logging.getLogger(__name__)
+
+HOST = "10.100.1.1"
+PORTS: Tuple[int, ...] = (9000, 18000)
+
+#: Reply keys holding watt readings, in priority order.
+POWER_KEYS: Tuple[str, ...] = (
+    "totalPower",
+    "homeloadPower",
+    "gridPower",
+    "solarPower",
+    "SW1ExpPower",
+    "SW2ExpPower",
+)
+
+
+def _to_float(value: Any) -> Optional[float]:
+    try:
+        if isinstance(value, bool):
+            return None
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def extract_power(reply: Dict[str, Any]) -> Dict[str, Any]:
+    """Pull watt readings out of a parsed reply envelope.
+
+    Returns ``{"value": float|None, "source": str|None, "detail": {...}}``.
+    ``value`` is the first ``POWER_KEYS`` hit (headline: ``totalPower``);
+    ``detail`` carries the supporting fields (per-switch watts/currents,
+    grid volts/amps) when present.
+    """
+    data = reply.get("dataArea", reply)
+    if isinstance(data, str):
+        try:
+            data = json.loads(data)
+        except (ValueError, TypeError):
+            data = {}
+    for key in POWER_KEYS:
+        for hit in _walk(reply, key):
+            num = _to_float(hit)
+            if num is not None:
+                return {"value": num, "source": key,
+                        "detail": _power_detail(data)}
+    return {"value": None, "source": None, "detail": _power_detail(data)}
+
+
+def _power_detail(data: Any) -> Dict[str, Any]:
+    detail: Dict[str, Any] = {}
+    if not isinstance(data, dict):
+        return detail
+    for key in (
+        "totalPower", "apowerCount", "homeloadPower", "solarPower",
+        "gridPower", "SW1ExpPower", "SW2ExpPower", "Sw1Volt", "Sw2Volt",
+        "SW1Curr", "gridVol1", "gridVol2", "gridCurr1", "gridCurr2",
+        "loadCurr1", "loadCurr2", "gridLineVol", "gridFreq",
+    ):
+        hits = _walk(data, key)
+        if hits:
+            detail[key] = hits[0] if len(hits) == 1 else hits
+    return detail
+
+
+def ems_load_watts(reply: Dict[str, Any]) -> Optional[float]:
+    """Estimate household draw from an EMS reply, when present.
+
+    ``(loadCurr1 + loadCurr2) * (gridVol1 + gridVol2) / 2`` — per-leg
+    current times mean leg voltage.  Returns ``None`` if the fields are
+    absent (the EMS reply on the tested unit carries live grid Vol/Curr).
+    """
+    data = reply.get("dataArea", reply)
+    if not isinstance(data, dict):
+        return None
+    try:
+        amps = float(data["loadCurr1"]) + float(data["loadCurr2"])
+        volts = (float(data["gridVol1"]) + float(data["gridVol2"])) / 2
+    except (KeyError, TypeError, ValueError):
+        return None
+    return amps * volts
+
+
+def get_power(
+    sn: str,
+    host: str = HOST,
+    ports: Tuple[int, ...] = PORTS,
+    key_radix: int = 16,
+    timeout: float = 5.0,
+) -> Dict[str, Any]:
+    """Read live household wattage over the local TCP socket.
+
+    Tries realtime detail (``DEVICE_DETAIL_INFO`` → ``totalPower``),
+    then per-switch load status, then EMS telemetry, probing each port.
+    Returns the ``extract_power`` dict plus ``cmd``/``port``/``reply``.
+    """
+    last_err: Optional[Exception] = None
+    for port in ports:
+        client = LocalDeviceClient(
+            sn=sn, host=host, socket_port=port,
+            key_radix=key_radix, timeout=timeout,
+        )
+        for name, call in (
+            ("detail", client.query_detail_info),
+            ("load", client.query_load_status),
+            ("ems", client.query_ems_info),
+        ):
+            try:
+                reply = call()
+            except (ConnectionError, socket.error, TimeoutError) as exc:
+                last_err = exc
+                log.debug("port %d %s failed: %s", port, name, exc)
+                break
+            except Exception as exc:
+                last_err = exc
+                log.debug("port %d %s error: %s", port, name, exc)
+                continue
+            result = extract_power(reply)
+            if result["value"] is None and name == "ems":
+                watts = ems_load_watts(reply)
+                if watts is not None:
+                    result = {"value": watts, "source": "ems(loadCurr*V)",
+                              "detail": _power_detail(
+                                  reply.get("dataArea", reply))}
+            result.update({"cmd": name, "port": port, "reply": reply})
+            if result["value"] is not None:
+                return result
+            last_err = ValueError(
+                f"reply on port {port} {name} has no power field")
+    raise ConnectionError(f"no power reply via {host}:{list(ports)}") from last_err
+
+
+def main(argv: Optional[List[str]] = None) -> int:
+    parser = argparse.ArgumentParser(
+        description="Read live household wattage over the device SoftAP socket.")
+    parser.add_argument("--sn", required=True, help="20-char device serial")
+    parser.add_argument("--host", default=HOST, help="aGate gateway IP")
+    parser.add_argument("--ports", default=",".join(map(str, PORTS)),
+                        help="comma-separated socket ports to probe")
+    parser.add_argument("--timeout", type=float, default=5.0)
+    parser.add_argument("--raw", action="store_true",
+                        help="print full reply JSON")
+    args = parser.parse_args(argv)
+
+    ports = tuple(int(p) for p in args.ports.split(",") if p.strip())
+    try:
+        result = get_power(args.sn, host=args.host, ports=ports,
+                           timeout=args.timeout)
+    except Exception as exc:
+        print(f"ERROR: {exc}")
+        return 1
+    if args.raw:
+        print(json.dumps(result["reply"], indent=2, default=str))
+    else:
+        value = result["value"]
+        print(f"{value:g}W ({result['source']})"
+              if value is not None else "unknown")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/host_probe8.py b/scripts/host_probe8.py
new file mode 100644
index 0000000..5b4f86b
--- /dev/null
+++ b/scripts/host_probe8.py
@@ -0,0 +1,127 @@
+#!/usr/bin/env python3
+"""Host probe v8 — live household wattage (stdlib only).
+
+Three socket APIs (all verified in disassembly; detail/load pending live
+confirmation on this unit, EMS already answered live):
+* DEVICE_DETAIL_INFO (1301, no args) -> totalPower  [home-page poll]
+* LOAD_STATUS        (1411, no args) -> SW1ExpPower [per-switch watts]
+* EMS                (1707, no args) -> grid/load V+A [watts via math]
+
+Run on the SoftAP-joined host:
+    python3 host_probe8.py --sn A03X2516013210060005 --ports 9000
+"""
+
+import argparse
+import json
+import socket
+import time
+import zlib
+
+DETAIL, LOAD, EMS = 0x515, 0x583, 0x6AB
+SNNO = [0]
+
+
+def secret_key(sn):
+    return format(sum(sn.encode()) & 0xFF, "x")
+
+
+def build_envelope(sn, cmd, args=None, snno=None, ts=None, equip=None):
+    if snno is None:
+        SNNO[0] = (SNNO[0] % 9999) + 1
+        snno = SNNO[0]
+    data = {"opt": 0}
+    if args:
+        data.update(args)
+    compact = json.dumps(data, separators=(",", ":"))
+    env = {"cmdType": cmd, "equipNo": equip if equip is not None else sn,
+           "type": 0,
+           "timeStamp": ts if ts is not None else int(time.time()),
+           "snno": snno,
+           "len": len(compact.encode()),
+           "crc": format(zlib.crc32(compact.encode()), "X"),
+           "dataArea": data}
+    return json.dumps(env, separators=(",", ":"))
+
+
+def encrypt(json_str, key):
+    parts = json_str.split(",")
+    a = (parts[0] + "," + parts[1] + ",").encode()
+    b = ",".join(parts[2:]).encode()
+    n = int(key, 16)
+    return a + bytes((b[i] + i + len(a) + n) & 0xFF
+                     for i in range(len(b)))
+
+
+def decrypt_wire(wire):
+    first = wire.find(b",")
+    second = wire.find(b",", first + 1)
+    if first < 0 or second < 0:
+        return ""
+    a, b = wire[:second + 1], wire[second + 1:]
+    try:
+        probe = json.loads(a[:-1].decode() + "}")
+    except Exception:
+        return ""
+    if not isinstance(probe, dict):
+        return ""
+    n = int(secret_key(str(probe.get("equipNo", ""))), 16)
+    try:
+        return (a + bytes((t - i - len(a) - n) & 0xFF
+                          for i, t in enumerate(b))).decode()
+    except Exception:
+        return ""
+
+
+def recv_all(sock, idle=8.0):
+    out = bytearray()
+    sock.settimeout(idle)
+    while True:
+        try:
+            chunk = sock.recv(65536)
+        except (socket.timeout, TimeoutError):
+            break
+        if not chunk:
+            break
+        out.extend(chunk)
+    return bytes(out)
+
+
+def run(label, host, port, tx, timeout):
+    rx, err, reply = b"", None, None
+    try:
+        with socket.create_connection((host, port),
+                                      timeout=timeout) as sock:
+            sock.sendall(tx)
+            rx = recv_all(sock)
+            if rx:
+                dec = decrypt_wire(rx)
+                reply = json.loads(dec) if dec else {"_raw": rx.hex()}
+    except (ConnectionError, socket.error, TimeoutError) as exc:
+        err = "%s: %s" % (type(exc).__name__, exc)
+    print("--- %s -> %s:%d" % (label, host, port))
+    print("  rx (%d B)" % len(rx))
+    if reply is not None:
+        print("  reply: " + json.dumps(reply, default=str)[:2000])
+    if err:
+        print("  error: %s" % err)
+    return reply
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--sn", required=True)
+    ap.add_argument("--host", default="10.100.1.1")
+    ap.add_argument("--ports", default="9000")
+    ap.add_argument("--timeout", type=float, default=5.0)
+    args = ap.parse_args()
+    key = secret_key(args.sn)
+    print("secret key for %s: %s (radix 16)" % (args.sn, key))
+    for port in [int(p) for p in args.ports.split(",") if p.strip()]:
+        for label, cmd in (("detail/1301", DETAIL), ("load/1411", LOAD),
+                           ("ems/1707", EMS)):
+            tx = encrypt(build_envelope(args.sn, cmd), key)
+            run(label, args.host, port, tx, args.timeout)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/test_power.py b/tests/test_power.py
new file mode 100644
index 0000000..d474e7a
--- /dev/null
+++ b/tests/test_power.py
@@ -0,0 +1,44 @@
+"""Offline tests for live-power extraction (no radio/hardware needed)."""
+
+import unittest
+
+from franklinwh_agate_client.power import (
+    ems_load_watts,
+    extract_power,
+)
+
+
+class TestExtractPower(unittest.TestCase):
+    def test_total_power_wins(self):
+        reply = {"cmdType": 1301,
+                 "dataArea": {"apowerCount": 2, "totalPower": 1872.5}}
+        out = extract_power(reply)
+        self.assertEqual(out["value"], 1872.5)
+        self.assertEqual(out["source"], "totalPower")
+
+    def test_switch_power_fallback(self):
+        reply = {"dataArea": {"index": 1, "Sw1Volt": 120.1,
+                              "SW1Curr": 3.2, "SW1ExpPower": 384.0}}
+        out = extract_power(reply)
+        self.assertEqual(out["value"], 384.0)
+        self.assertEqual(out["source"], "SW1ExpPower")
+
+    def test_missing_power(self):
+        out = extract_power({"dataArea": {"result": 1, "reason": -1}})
+        self.assertIsNone(out["value"])
+        self.assertIsNone(out["source"])
+
+    def test_ems_load_watts_math(self):
+        reply = {"dataArea": {"gridVol1": 120.4, "gridVol2": 120.5,
+                              "loadCurr1": 7.8, "loadCurr2": 7.8,
+                              "gridLineVol": 2410}}
+        watts = ems_load_watts(reply)
+        self.assertAlmostEqual(watts, 15.6 * 120.45, places=1)
+
+    def test_ems_load_watts_missing(self):
+        self.assertIsNone(ems_load_watts({"dataArea": {}}))
+        self.assertIsNone(ems_load_watts({}))
+
+
+if __name__ == "__main__":
+    unittest.main()