Metadata-Version: 2.4
Name: py-solar-assistant
Version: 0.4.0
Summary: Python client for SolarAssistant — cloud API and real-time WebSocket
Author: Solar-Assistant
License-Expression: Apache-2.0
Project-URL: Homepage, https://github.com/Solar-Assistant/py_solar_assistant
Project-URL: Repository, https://github.com/Solar-Assistant/py_solar_assistant
Project-URL: Issues, https://github.com/Solar-Assistant/py_solar_assistant/issues
Keywords: solar-assistant,solar,inverter,battery,homeassistant
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Home Automation
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.12
Description-Content-Type: text/markdown
License-File: LICENSE
License-File: NOTICE
Requires-Dist: aiohttp<4,>=3.9
Dynamic: license-file

# py_solar_assistant

Python client for SolarAssistant. Also available in [Go](https://github.com/Solar-Assistant/go_solar_assistant).

## Installation

```bash
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](https://solar-assistant.io/user/edit#api).

```python
from py_solar_assistant import SolarAssistantClient, list_sites, authorize_site

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

### List sites

```python
sites = await list_sites(client)
```

Filter by inverter, battery, name, and more:

```python
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.

```python
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

```python
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](#authorize-a-site):

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

### Read metrics

```python
# 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

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

### Read system metrics

The unit's own metrics (Site ID, software version, CPU temperature and usage, free storage and memory) are the `system/*` topics, in
the same row shape as `get_metrics`. They are read from `GET /api/v1/metrics?topic=system/*`; a unit running a build that doesn't list
them there is read from the older `GET /api/v1/system` instead. A build that predates both responds `404` to the second request, which
raises `SolarAssistantError` like any other non-200 — catch it and check `.status == 404` to detect old firmware.

```python
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:

```python
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`](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](#authorize-a-site):

```python
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:

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

### Topic filters

Subscribe to specific topics with optional server-side throttling:

```python
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/*
grid/*
weather/*
battery_*/voltage
battery_*/state_of_charge
battery_*/power
battery_*/temperature
inverter_*/pv_power
inverter_*/load_power
inverter_*/grid_power
inverter_*/device_mode
inverter_*/temperature
ev_charger_*/*
```

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

### System metrics

The unit's own diagnostics (`system/*`: Site ID, software version, CPU temperature and usage, free storage and memory) are not part of
the default set. Name them like any other topic and they arrive through the same handler:

```python
await sock.subscribe_metrics(handler, TopicFilter(topic="total/*"), TopicFilter(topic="system/*"))
```

### Write a setting

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

```python
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

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

## Examples

Runnable scripts are in the [`examples/`](examples/) folder:

| Script                                            | Description                                                      |
| ------------------------------------------------- | ---------------------------------------------------------------- |
| [`rest_read.py`](examples/rest_read.py)           | Fetch all metrics once via REST and print them grouped by device |
| [`rest_system.py`](examples/rest_system.py)       | Read a unit's system metrics, handling the 404/old-firmware case  |
| [`rest_set.py`](examples/rest_set.py)             | Write a metric value via REST                                    |
| [`websocket_read.py`](examples/websocket_read.py) | Stream live metrics via WebSocket until Ctrl+C                   |
| [`websocket_set.py`](examples/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](https://peps.python.org/pep-0735/) dependency group, kept
out of the published wheel, and need `pip >= 25.1`:

```bash
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):

```bash
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](https://www.conventionalcommits.org) via commitizen. See [RELEASING.md](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](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](NOTICE) for the copyright and scope
statement.
