py-solar-assistant 0.3.0

Python client for SolarAssistant — cloud API and real-time WebSocket

py_solar_assistant

Python client for SolarAssistant. Also available in Go.

Installation

pip install py-solar-assistant

Requires Python 3.12+.

Cloud API

Interact with the SolarAssistant cloud API. All endpoints require an API key — generate one at solar-assistant.io/user/edit#api.

from py_solar_assistant import SolarAssistantClient, list_sites, authorize_site

async with SolarAssistantClient("<api_key>") as client:
    sites = await list_sites(client)

List sites

sites = await list_sites(client)

Filter by inverter, battery, name, and more:

sites = await list_sites(client, inverter="srne", limit=50, offset=20)

Filters match exactly; use search for a prefix + full-text match.

Common filters:

Key	Example
search	search="my-s" (name prefix + text)
name	name="my-site" (exact)
inverter	inverter="srne"
battery	battery="daly"
inverter_params_output_power	inverter_params_output_power="5000"
last_seen_after	last_seen_after="2026-01-01"
build_date_after	build_date_after="2026-02-26"
limit	limit=50
offset	offset=20

Authorize a site

Returns a short-lived token and connection details for a site. The token works for both cloud and local connections.

auth = await authorize_site(client, site_id)
# auth.host, auth.site_id, auth.site_key, auth.token, auth.local_ip

---

Device — REST

Read and write metrics directly on a SolarAssistant unit via REST.

Local connection

from py_solar_assistant import DeviceClient

async with DeviceClient("192.168.1.100", password="<web-password>") as c:
    ...

Cloud-proxied connection

First obtain connection details via authorize_site:

async with DeviceClient(
    auth.host,
    token=auth.token,
    site_id=auth.site_id,
    site_key=auth.site_key,
    scheme="https",
) as c:
    ...

Read metrics

# All metrics
metrics = await c.get_metrics()

# Filtered by topic glob — multiple topics are fetched and deduplicated
metrics = await c.get_metrics("battery_1/*", "total/pv_power")

Write a metric

await c.set_metric("inverter_1/charge_current_limit", "20")

Read system metrics

GET /api/v1/system exposes unit-level metrics (Site ID, software version, CPU temperature, free storage) in the same row shape as get_metrics. A unit running a SolarAssistant build that predates the endpoint responds 404, which raises SolarAssistantError like any other non-200 — catch it and check .status == 404 to detect old firmware.

rows = await c.get_system_metrics()  # list[DeviceMetric]

# Typed convenience accessors - each returns None only if the value is unset/unreadable.
# Each does its own request, so to read several at once prefer get_system_metrics() above.
site_id = await c.get_site_id()                  # int | None
version = await c.get_software_version()         # str | None
cpu_temperature = await c.get_cpu_temperature()  # int | None  (°C)
free_storage = await c.get_free_storage()        # int | None  (MB)

Standalone functions

Every DeviceClient method has a module-level twin that opens and closes its own connection — handy for one-off calls without managing a context:

from py_solar_assistant import get_device_site_id

site_id = await get_device_site_id("192.168.1.100", password="<web-password>")  # int | None

The rest follow the same shape: get_device_metrics, set_metric, get_device_system_metrics, get_device_software_version, get_device_cpu_temperature, get_device_free_storage. See examples/rest_system.py for reading the system metrics (including the 404/old-firmware case).

---

Device — WebSocket

Stream live metrics via WebSocket.

Cloud connection

First obtain connection details via authorize_site:

import asyncio
from py_solar_assistant import SolarAssistantClient, authorize_site, connect, Options

async def main():
    async with SolarAssistantClient("<api_key>") as client:
        auth = await authorize_site(client, site_id)

    sock = await connect(Options(
        host=auth.host,
        token=auth.token,
        site_id=auth.site_id,
        site_key=auth.site_key,
        local_ip=auth.local_ip,  # if set, tries local network first and falls back to cloud
    ))

    await sock.subscribe_metrics(
        lambda m: print(f"{m.device}/{m.name} = {m.value} {m.unit}")
    )
    try:
        await sock.listen()  # blocks until disconnected
    finally:
        await sock.close()

asyncio.run(main())

Direct local connection (no cloud)

Connect using the unit's web password, no cloud account required:

sock = await connect(Options(
    local_ip="192.168.1.100",
    password="<web-password>",
))

Topic filters

Subscribe to specific topics with optional server-side throttling:

from py_solar_assistant import TopicFilter

await sock.subscribe_metrics(
    handler,
    TopicFilter(topic="total/*"),
    TopicFilter(topic="inverter_*/load_power"),
    TopicFilter(topic="battery_*/voltage", max_frequency_s=10),
)

If no filters are passed the server applies a default set of common metrics:

total/*
battery_*/voltage
battery_*/state_of_charge
battery_*/power
battery_*/temperature
inverter_*/pv_power
inverter_*/load_power
inverter_*/grid_power
inverter_*/device_mode
inverter_*/temperature

Only metrics in groups Info, Status, and Settings are sent.

Write a setting

Send a setting change over the WebSocket and wait for the result:

await sock.set_setting("inverter_1/power_mode", "Off grid with relay")

Raises ValueError if the device rejects the value.

Options

Field	Type	Description
host	str	Hostname or host:port of the cloud proxy.
token	str	JWT from authorize_site. Required for cloud and local-fallback connections.
password	str	Web password for direct local connections.
site_id	int	Required for cloud connections.
site_key	str	Required for cloud connections.
local_ip	str	If set, tries local network first and falls back to host.
verbose	bool	Log all WebSocket frames at DEBUG level (via Python logging).

Advanced: raw channel messages

sock.subscribe("*", "*", lambda msg: print(msg.topic, msg.event, msg.payload))

Examples

Runnable scripts are in the examples/ folder:

Script	Description
rest_read.py	Fetch all metrics once via REST and print them grouped by device
rest_system.py	Read a unit's system metrics, handling the 404/old-firmware case
rest_set.py	Write a metric value via REST
websocket_read.py	Stream live metrics via WebSocket until Ctrl+C
websocket_set.py	Write a setting via WebSocket

Development

Install the package editable (-e, so src/ edits are picked up without reinstalling — the src/ layout otherwise keeps it off sys.path) together with its dev/test dependencies. Those live in a PEP 735 dependency group, kept out of the published wheel, and need pip >= 25.1:

python -m pip install -e . --group dev

The suite uses pytest + pytest-asyncio, driving the client against real local aiohttp test servers (no network, no mocking library):

python -m pytest          # run the tests
ruff check .              # lint
ruff format .             # format (use `ruff format --check .` to verify only)

scripts/release.sh runs ruff and the test suite as a preflight, so releases are gated on a clean lint, clean formatting, and a green test run.

Releasing

Maintainers cut releases from a developer machine with scripts/release.sh. The version, changelog, tag, PyPI upload, and GitHub release are all driven by Conventional Commits via commitizen. See RELEASING.md for the full process.

Roadmap

Upgrade to aiohttp 4.0

The client is capped at aiohttp<4 because three calls it makes are removed in 4.0: DeviceClient's local-password auth (aiohttp.BasicAuth plus the request auth= parameter) and Socket._dial's ws_connect(timeout=ClientTimeout(...)). To lift the cap, migrate those calls (a base64 Authorization header for Basic auth, ClientWSTimeout for the WebSocket timeout) and drop the matching filterwarnings suppressions in pyproject.toml.

Harden Socket.set_setting

Add a reply timeout, raise on error replies instead of hanging, preserve an existing topic filter instead of re-joining empty, and make it safe to call alongside listen().

License

Apache 2.0 - see LICENSE.

This licence covers the Python client library in this repository only. The SolarAssistant platform, including the downloadable device software and cloud infrastructure, is proprietary and distributed under separate terms. See NOTICE for the copyright and scope statement.