py-solar-assistant 0.1.1

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)

Common filters:

Key	Example
name	name="my-site"
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")

Standalone functions

For one-off calls without connection reuse:

from py_solar_assistant import get_device_metrics, set_metric

metrics = await get_device_metrics("192.168.1.100", password="<web-password>")
await set_metric("192.168.1.100", "inverter_1/charge_current_limit", "20", password="<web-password>")

---

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_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

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.

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.