Metadata-Version: 2.5
Name: pydaze
Version: 0.1.0
Summary: Unofficial Async Python client library for the Daze Wallbox API
Author-email: Marco Caimi <mcaimi@redhat.com>
License: MIT
Requires-Python: >=3.12
Requires-Dist: aiohttp>=3.9.3
Description-Content-Type: text/markdown

# pydaze

**pydaze** is an unofficial async Python client library for the [Daze Wallbox](https://dazeservice.com) web API.

It wraps the Daze REST endpoints behind a clean, typed, `async`/`await` interface and handles the full OAuth token lifecycle (authentication, validation, automatic refresh on `401`) so you only have to think about the charging data.

> **Disclaimer:** pydaze is not affiliated with, endorsed by, or maintained by Daze. It interacts with Daze's private web API and may break when the API changes.

## Features

- Fully asynchronous, built on [`aiohttp`](https://docs.aiohttp.org/)
- **Dual authentication mechanisms**:
  - Username/password login via Amazon Cognito (`USER_PASSWORD_AUTH`)
  - Token-based auth using access/refresh tokens harvested from the browser (DevTools)
- Automatic access-token refresh: a `401` triggers a transparent token refresh and retry; if the refresh token itself is invalid, an `ApiAuthError` is raised so you can re-authenticate
- Typed data models (`dataclass`) for all API responses, tolerant of both legacy v3 and current v4 API shapes
- Python ≥ 3.12, MIT licensed

## Installation

```bash
# uv
uv add pydaze
```

Requires Python 3.12 or later.

## Authentication

pydaze supports two authentication mechanisms. Both produce an auth client that can be passed to `DazeApiClient`.

### Option 1 — Username/password (Cognito IDP)

Use the credentials of your Daze web portal account:

```python
from pydaze import DazeCognitoAuthClient

auth = await DazeCognitoAuthClient.async_authenticate(
    session,          # aiohttp.ClientSession
    email="user@example.com",
    password="s3cret",
)
```

`async_authenticate` is a classmethod that performs the `USER_PASSWORD_AUTH` flow against Amazon Cognito and returns a fully initialized client with valid access and refresh tokens.

### Option 2 — Harvested tokens

Harvest the `access_token` and `refresh_token` from Firefox/Chrome DevTools while logged in to the Daze web portal, then construct the client directly:

```python
from pydaze import DazeAuthClient

auth = DazeAuthClient(
    access_token="eyJ...",
    refresh_token="eyJ...",
    token_expiry=1756290000.0,  # optional: UNIX expiry timestamp of the access token
)

# optional: verify the tokens are still valid (raises AuthError otherwise)
await auth.async_validate_tokens(session)
```

Notes:

- If `token_expiry` is omitted, `is_token_expired()` always returns `False` and expiry tracking is disabled.
- `auth.get_tokens_for_store()` returns a plain dict of the current tokens, convenient for persisting them in a config store.
- Both auth clients implement `async_refresh_access_token(session)`, which the API client calls automatically when needed.

## API Reference

### `DazeApiClient`

```python
client = DazeApiClient(auth_client, session)
```

| Method | Description |
| --- | --- |
| `async_get_user_info()` | Fetch user info from the Cognito `userInfo` endpoint. Returns `dict`. |
| `async_get_networks(email)` | Fetch all networks (installations) for a user. Returns `list[Network]`. |
| `async_get_evses(network_uid)` | Fetch all EVSEs (chargers) in a network. Returns `list[Evse]`. |
| `async_get_socket_remote_info(serial)` | Fetch live socket metrics and state (power, energy, per-phase voltage/current, temperatures, schedule). Returns `SocketRemoteInfo`. |
| `async_set_max_charging_current(serial, current_ma)` | Set the maximum charging current for a wallbox. `current_ma` is in **milliamps** (e.g. `16000` = 16 A). |
| `async_set_eco_mode(serial, eco_mode_enabled)` | Enable or disable eco mode on a wallbox. |
| `async_start_charge(serial)` | Start charging on a wallbox. |
| `async_stop_charge(serial)` | Stop charging on a wallbox. |
| `async_get_recharge_sessions(network_uid, limit=1000)` | Fetch recharge session history for a network. Returns `list[RechargeSession]`. |

All methods are `async`. Write commands return the raw API response dict.

### `DazeAuthClient`

| Method / property | Description |
| --- | --- |
| `get_headers()` | Return the `authorization` header dict for API calls. |
| `is_token_expired()` | `True` if the access token is expired (or within the expiry buffer). |
| `async_refresh_access_token(session)` | Refresh the access token via the Cognito OAuth2 token endpoint. |
| `async_validate_tokens(session)` | Validate the stored access token against Cognito `userInfo`. Raises `AuthError` on failure. |
| `token_expiry` | UNIX timestamp of the access token expiry (or `None`). |
| `get_tokens_for_store()` | Tokens as a plain dict, safe for persistence. |

`DazeCognitoAuthClient` extends `DazeAuthClient` and additionally provides:

| Method | Description |
| --- | --- |
| `async_authenticate(session, email, password)` *(classmethod)* | Log in via Cognito IDP and return an initialized client. |

## Usage Examples

### Full workflow: login → discover → live status

```python
import asyncio

import aiohttp
from pydaze import DazeApiClient, DazeCognitoAuthClient


async def main() -> None:
    async with aiohttp.ClientSession() as session:
        # 1. Authenticate with Daze web portal credentials
        auth = await DazeCognitoAuthClient.async_authenticate(
            session, "user@example.com", "s3cret"
        )
        client = DazeApiClient(auth, session)

        # 2. Discover networks and chargers
        networks = await client.async_get_networks("user@example.com")
        for network in networks:
            print(f"Network: {network.name} ({network.uid})")

        evses = await client.async_get_evses(networks[0].uid)
        for evse in evses:
            print(f"  EVSE: {evse.evse_name}, serial {evse.serial_number}")

        # 3. Read live socket metrics
        info = await client.async_get_socket_remote_info(evses[0].serial_number)
        print(f"State:       {info.evse_status}")
        print(f"Power:       {info.instant_power_as_watt} W")
        print(f"Energy:      {info.delivered_energy_as_watt_hour} Wh")
        print(f"Voltage L1:  {info.last_ac_voltage_l1} V")
        print(f"Eco mode:    {info.eco_mode_enabled}")


if __name__ == "__main__":
    asyncio.run(main())
```

### Controlling the wallbox

```python
import aiohttp
from pydaze import DazeApiClient, DazeCognitoAuthClient

SERIAL = "99XZ0402185"  # EVSE serial number


async def main() -> None:
    async with aiohttp.ClientSession() as session:
        auth = await DazeCognitoAuthClient.async_authenticate(
            session, "user@example.com", "s3cret"
        )
        client = DazeApiClient(auth, session)

        # Limit charging to 16 A (value is in milliamps)
        await client.async_set_max_charging_current(SERIAL, 16000)

        # Enable eco mode
        await client.async_set_eco_mode(SERIAL, True)

        await client.async_start_charge(SERIAL)
        # ... charge ...
        await client.async_stop_charge(SERIAL)
```

### Recharge session history

```python
import aiohttp
from pydaze import DazeApiClient, DazeCognitoAuthClient


async def main() -> None:
    async with aiohttp.ClientSession() as session:
        auth = await DazeCognitoAuthClient.async_authenticate(
            session, "user@example.com", "s3cret"
        )
        client = DazeApiClient(auth, session)

        networks = await client.async_get_networks("user@example.com")
        sessions = await client.async_get_recharge_sessions(
            networks[0].uid, limit=10
        )
        for s in sessions:
            state = "in progress" if s.is_in_progress else "completed"
            print(
                f"{s.start_time} → {s.end_time} | "
                f"{s.energy_wh} Wh | {s.cost} {s.currency} ({state})"
            )
```

### Handling authentication errors

The API client automatically refreshes the access token and retries once on a `401`. If the refresh token itself is invalid or expired, it raises `ApiAuthError` — at that point you must re-authenticate (e.g. by calling `DazeCognitoAuthClient.async_authenticate` again and reusing the new client, or by harvesting fresh tokens):

```python
from pydaze import ApiAuthError, ApiError

try:
    info = await client.async_get_socket_remote_info(serial)
except ApiAuthError as err:
    # Refresh token invalid/expired — re-authenticate and retry
    ...
except ApiError as err:
    # Other 4xx/5xx API error or network problem
    ...
```

### Exception hierarchy

| Exception | Raised when |
| --- | --- |
| `AuthError` | Authentication or token refresh fails (invalid/expired tokens, network error during auth). |
| `CognitoAuthError` | *(subclass of `AuthError`)* Cognito IDP-specific auth failure; carries an `error_type` attribute. |
| `ApiError` | Non-auth API errors (4xx/5xx responses, network issues). |
| `ApiAuthError` | The API returned `401` even after a token refresh — re-authentication required. |

## Data Models

All models are `dataclass`es with a `from_dict()` classmethod and a `raw` attribute holding the original API response:

| Model | Description | Key fields |
| --- | --- | --- |
| `Network` | An installation (site) with one or more chargers | `uid`, `name`, `city`, `is_photovoltaic`, `num_evses_in_network`, `eco_mode_enabled` |
| `Evse` | A charger | `serial_number`, `evse_name`, `firmware_version`, `active`, `max_external_charging_current_in_milli_amps`, `eco_mode_enabled` |
| `SocketRemoteInfo` | Live socket metrics and state | `evse_status`, `instant_power_as_watt`, `delivered_energy_as_watt_hour`, `last_ac_voltage_l1/l2/l3`, `board_temperature`, `active`, `is_paused`, `next_scheduled_charge` |
| `RechargeSession` | A charging session (v3/v4 compatible) | `session_uid`, `start_time`, `end_time`, `energy_wh`, `cost`, `currency`, `evse_serial`, `is_in_progress` (property) |

## Development

```bash
uv sync          # install project + dev dependencies
uv run pytest    # run the test suite (synthetic data, no live API calls)
```

## License

[MIT](https://opensource.org/licenses/MIT)
