SPAN Panel API

A Python client library for the SPAN Panel v2 API, using MQTT/Homie for real-time push-based panel state.

v1.x Sunset Notice

Package versions prior to 2.0.0 are deprecated. These versions depend on the SPAN v1 REST API, which will be retired when SPAN sunsets v1 firmware at the end of 2026. Users should upgrade to v2.0.0 or later, which requires v2 firmware (spanos2/r202603/05 or later) and a panel passphrase.

Installation

Two packages: the transport, and a parser for your panel's schema. span-panel-api contains no parser — installing it alone gives a client that connects and then raises SpanPanelAdapterMissingError.

pip install span-panel-api span-panel-api-schema-0

span-panel-api-schema-0 parses the flat schema used by firmware r202603 through r202627, which is every panel in the field today. Panels reporting a data-model-version need the adapter for that schema major instead; the error names the one it could not find and lists what is installed.

Parsers are discovered through the span_panel_api.schema_adapters entry-point group, so support for a new panel schema arrives by installing a package rather than by upgrading the transport. The two version independently — see RELEASE.md.

Dependencies

• httpx — v2 authentication and detection endpoints
• paho-mqtt — MQTT/Homie transport (real-time push)
• pyyaml — YAML parsing for configuration and API payloads

Architecture

Transport

The SpanMqttClient connects to the panel's MQTT broker (MQTTS or WebSocket) and subscribes to the Homie device tree. A two-layer architecture separates generic Homie v5 protocol handling from SPAN-specific interpretation:

• HomiePropertyAccumulator — handles message routing, property and $target storage, dirty-node tracking, and an explicit lifecycle state machine (HomieLifecycle). Protocol-only; no SPAN domain knowledge.
• HomieDeviceConsumer — reads from the accumulator via a query API and builds typed SpanPanelSnapshot dataclasses. Handles power sign normalization, DSM derivation, unmapped tab synthesis, and dirty-node-aware snapshot caching.

Changes are pushed to consumers via callbacks. Dirty-node tracking allows the snapshot builder to skip unchanged nodes, reducing per-scan CPU cost on constrained hardware.

Event-Loop-Driven I/O (Home Assistant Compatible)

The MQTT transport is designed around the Home Assistant core async pattern — all paho-mqtt I/O runs on the asyncio event loop with no background threads:

• NullLock replacement — paho-mqtt's seven internal threading locks are replaced with no-op NullLock instances at setup time, eliminating lock contention since all access is single-threaded on the event loop.
• add_reader / add_writer — AsyncMqttBridge registers the MQTT socket with the event loop via loop.add_reader() and loop.add_writer(), calling paho's loop_read() / loop_write() directly from I/O callbacks rather than from a loop_start() background thread.
• Periodic misc — A loop.call_at() timer fires every second to call loop_misc() for keepalive and timeout housekeeping.
• Executor bridge for connect — The initial TLS handshake and TCP connect are blocking operations, so they run in loop.run_in_executor(). Once the executor returns, socket callbacks are immediately switched from sync bridges (call_soon_threadsafe) back to the async-only versions.

This means the library can be dropped into any asyncio application — including Home Assistant — without spawning threads or requiring thread-safe wrappers.

Circuit Name Synchronization

Circuit names arrive as MQTT retained messages that may land after the Homie device transitions to $state=ready. The client handles this with a bounded wait during connect():

1. After the device reaches ready state, the client polls HomieDeviceConsumer.circuit_nodes_missing_names() every 250ms.
2. As retained name properties arrive, the consumer stores them. Once all circuit-type nodes have a name, the wait returns immediately.
3. If names have not all arrived within 10 seconds, the timeout expires (non-fatal) and the client proceeds — circuits without names will use fallback identifiers.

This ensures that the first get_snapshot() after connect returns human-readable circuit names in the common case, while never blocking indefinitely on a missing retained message.

Protocols

The library defines three structural subtyping protocols (PEP 544) that both the MQTT transport and the simulation engine implement:

Integration code programs against these protocols, not transport-specific classes.

Snapshots

All panel state is represented as immutable, frozen dataclasses:

Usage

Factory Pattern (Recommended)

The create_span_client() factory handles v2 registration and returns a configured SpanMqttClient:

import asyncio
from span_panel_api import create_span_client

async def main():
    client = await create_span_client(
        host="192.168.1.100",
        passphrase="your-panel-passphrase",
    )

    try:
        await client.connect()

        # Get a point-in-time snapshot
        snapshot = await client.get_snapshot()
        # The upstream lugs' own meter. That is grid flow only where the lugs are
        # the utility connection point; a BESS wired ahead of them, or a panel fed
        # by another panel, makes it this panel's feed instead. `power_flow_grid`
        # is the site-level figure in every topology.
        if snapshot.lugs_at_service_entrance:
            print(f"Grid power: {snapshot.instant_grid_power_w}W")
        else:
            print(f"Panel feed: {snapshot.instant_grid_power_w}W")
            print(f"Grid power: {snapshot.power_flow_grid}W")
        print(f"Firmware: {snapshot.firmware_version}")
        print(f"Circuits: {len(snapshot.circuits)}")

        for cid, circuit in snapshot.circuits.items():
            print(f"  {circuit.name}: {circuit.instant_power_w}W ({circuit.relay_state})")

    finally:
        await client.close()

asyncio.run(main())

Streaming Pattern

For real-time push updates without polling:

import asyncio
from span_panel_api import create_span_client, SpanPanelSnapshot

async def on_snapshot(snapshot: SpanPanelSnapshot) -> None:
    print(f"Grid: {snapshot.instant_grid_power_w}W, Circuits: {len(snapshot.circuits)}")

async def main():
    client = await create_span_client(
        host="192.168.1.100",
        passphrase="your-panel-passphrase",
    )

    try:
        await client.connect()

        # Register callback and start streaming
        unsubscribe = client.register_snapshot_callback(on_snapshot)
        await client.start_streaming()

        # Run until interrupted
        await asyncio.Event().wait()

    finally:
        await client.stop_streaming()
        await client.close()

asyncio.run(main())

Connection State Monitoring

Push consumers that need to react to broker disconnect/reconnect events — for example, to mark downstream entities offline within a second of a dropped connection rather than waiting on a fallback poll — can register a connection callback. The callback fires False on disconnect and True on reconnect, edge-only (no synthetic call at registration time):

def on_connection_change(connected: bool) -> None:
    if connected:
        print("Broker connection restored")
    else:
        print("Broker connection lost")

unsubscribe_connection = client.register_connection_callback(on_connection_change)

# Later, during teardown:
unsubscribe_connection()

To check the current connection state on demand (for example, just after registering), call await client.ping().

When the client is not fully live (broker disconnected, or Homie device not yet ready), await client.get_snapshot() raises SpanPanelStaleDataError instead of returning cached data. Treat that exception as the canonical "panel currently unreachable" signal — see Error Handling below.

Pre-Built Config Pattern

If you already have MQTT broker credentials (e.g., stored from a previous registration):

from span_panel_api import create_span_client, MqttClientConfig

config = MqttClientConfig(
    broker_host="192.168.1.100",
    username="stored-username",
    password="stored-password",
    mqtts_port=8883,
    ws_port=9001,
    wss_port=443,
)

client = await create_span_client(
    host="192.168.1.100",
    mqtt_config=config,
    serial_number="nj-2316-XXXX",
)

Direct Client Construction

Consumers that manage their own registration and broker configuration can instantiate SpanMqttClient directly:

from span_panel_api import SpanMqttClient, MqttClientConfig

config = MqttClientConfig(
    broker_host="192.168.1.100",
    username="stored-username",
    password="stored-password",
    mqtts_port=8883,
    ws_port=9001,
    wss_port=443,
)

client = SpanMqttClient(
    host="192.168.1.100",
    serial_number="nj-2316-XXXX",
    broker_config=config,
    snapshot_interval=1.0,
)
await client.connect()

Scan Frequency

set_snapshot_interval() controls how often push-mode snapshot callbacks fire. Lower values mean lower latency; higher values reduce CPU usage on constrained hardware. Dirty-node caching (v2.5.0) further reduces per-scan cost by skipping unchanged nodes.

Passing 0 (or any non-positive value) disables debounce and dispatches a snapshot for every incoming property message — real-time mode, intended for fast consumers.

# Reduce snapshot frequency to every 2 seconds
client.set_snapshot_interval(2.0)

# Real-time dispatch — every property update triggers a callback
client.set_snapshot_interval(0)

Circuit Control

# Set circuit relay (OPEN/CLOSED)
await client.set_circuit_relay("circuit-uuid", "OPEN")
await client.set_circuit_relay("circuit-uuid", "CLOSED")

# Set circuit shed priority (NEVER / SOC_THRESHOLD / OFF_GRID)
await client.set_circuit_priority("circuit-uuid", "NEVER")

Pending-State Detection

When the panel publishes Homie $target properties, SpanCircuitSnapshot exposes the desired state alongside the actual state:

for cid, circuit in snapshot.circuits.items():
    if circuit.relay_state_target and circuit.relay_state_target != circuit.relay_state:
        print(f"  {circuit.name}: relay transitioning {circuit.relay_state} → {circuit.relay_state_target}")
    if circuit.priority_target and circuit.priority_target != circuit.priority:
        print(f"  {circuit.name}: priority pending {circuit.priority} → {circuit.priority_target}")

API Version Detection

Detect whether a panel supports v2 (unauthenticated probe):

from span_panel_api import detect_api_version

result = await detect_api_version("192.168.1.100")
print(f"API version: {result.api_version}")  # "v1" or "v2"
if result.status_info:
    print(f"Serial: {result.status_info.serial_number}")
    print(f"Firmware: {result.status_info.firmware_version}")

v2 Authentication Functions

Standalone async functions for v2-specific HTTP operations:

from span_panel_api import (
    register_v2, download_ca_cert, get_homie_schema,
    regenerate_passphrase, get_v2_status,
    register_fqdn, get_fqdn, delete_fqdn,
)

# Register and obtain MQTT broker credentials
auth = await register_v2("192.168.1.100", "my-app", passphrase="panel-passphrase")
print(f"Broker: {auth.ebus_broker_host}:{auth.ebus_broker_mqtts_port}")
print(f"Serial: {auth.serial_number}")

# Download the panel's CA certificate (for TLS verification)
pem = await download_ca_cert("192.168.1.100")

# Fetch the Homie property schema (unauthenticated)
schema = await get_homie_schema("192.168.1.100")
print(f"Panel size: {schema.panel_size} spaces")
print(f"Schema hash: {schema.types_schema_hash}")

# Rotate MQTT broker password (invalidates previous password)
new_password = await regenerate_passphrase("192.168.1.100", token=auth.access_token)

# Get panel status (unauthenticated)
status = await get_v2_status("192.168.1.100")
print(f"Serial: {status.serial_number}, Firmware: {status.firmware_version}")

# FQDN management (for panel TLS certificate SAN)
await register_fqdn("192.168.1.100", "panel.local", token=auth.access_token)
fqdn = await get_fqdn("192.168.1.100", token=auth.access_token)
await delete_fqdn("192.168.1.100", token=auth.access_token)

Error Handling

All exceptions inherit from SpanPanelError:

SpanPanelStaleDataError is distinct from SpanPanelConnectionError: the former means the client is running but data cannot be trusted right now (transient disconnect, or panel-declared not-ready); the latter means the initial connect failed and the client cannot be used at all.

from span_panel_api import (
    SpanPanelAuthError,
    SpanPanelConnectionError,
    SpanPanelStaleDataError,
)

try:
    client = await create_span_client(host="192.168.1.100", passphrase="wrong")
except SpanPanelAuthError:
    print("Invalid passphrase")
except SpanPanelConnectionError:
    print("Cannot reach panel")

# Later, during normal operation:
try:
    snapshot = await client.get_snapshot()
except SpanPanelStaleDataError as err:
    # Broker dropped or panel declared not-ready — fall back to last-known
    # data, a grace-period value, or mark downstream state unavailable.
    print(f"Snapshot unavailable: {err}")

Capabilities

The PanelCapability flag enum advertises transport features at runtime:

Reference Payloads

Captures of what a panel actually serves, shipped as package data so a consumer can check its own assumptions against real bytes without vendoring a copy that silently goes stale:

from span_panel_api.reference_payloads import homie_schema, homie_schema_types

document = homie_schema()        # the captured GET /api/v2/homie/schema response
types = homie_schema_types()     # its `types` map, typed as HomieSchemaTypes

homie_schema_types() returns exactly what span_panel_api_schema_0.field_metadata.build_field_metadata accepts, so building real adapter metadata to compare against is two lines and no file handling.

The parent/child device tree is the schema_1 counterpart and ships from that adapter, with the parser that can interpret it:

from span_panel_api_schema_1.reference_payloads import devices_from_tree, parent_child_tree

devices = devices_from_tree(parent_child_tree())

Each payload carries the version of the release it shipped in. Pin a version and you read the bytes that version was written against.

Project Structure

src/span_panel_api/
├── __init__.py          # Public API exports
├── auth.py              # v2 HTTP provisioning (register, cert, schema, passphrase)
├── const.py             # Panel state constants (DSM, relay)
├── detection.py         # detect_api_version() → DetectionResult
├── exceptions.py        # Exception hierarchy
├── factory.py           # create_span_client() → SpanMqttClient
├── models.py            # Snapshot dataclasses (panel, circuit, battery, PV)
├── phase_validation.py  # Electrical phase utilities
├── protocol.py          # PEP 544 protocols + PanelCapability flags
├── reference_payloads/  # Captured wire payloads shipped as package data
└── mqtt/
    ├── __init__.py
    ├── accumulator.py   # HomiePropertyAccumulator (Homie v5 protocol layer)
    ├── async_client.py  # NullLock + AsyncMQTTClient (HA core pattern)
    ├── client.py        # SpanMqttClient (all three protocols)
    ├── connection.py    # AsyncMqttBridge (event-loop-driven, no threads)
    ├── const.py         # MQTT/Homie constants + UUID helpers
    ├── homie.py         # HomieDeviceConsumer (SPAN snapshot builder)
    └── models.py        # MqttClientConfig, MqttTransport

Development

See DEVELOPMENT.md for setup, testing, and contribution guidelines.

License

MIT License - see LICENSE file for details.