fluvius-energy-api 0.2.2
Python wrapper for the Fluvius Energy API
Requires Python: >=3.12
Keywords: api, belgium, electricity, energy, esco, fluvius, gas, smart-meter

# Fluvius Energy API

Python wrapper for the Fluvius Energy API v3 (ESCO - Energy Service Company API).

## Installation

```bash
pip install fluvius-energy-api
```

Or with uv:

```bash
uv add fluvius-energy-api
```

## Configuration

The client requires OAuth2 credentials for Azure AD authentication. Two authentication methods are supported:

- **Certificate-based** (production): Uses certificate thumbprint + private key
- **Client secret** (sandbox): Uses a simple client secret

### Common Environment Variables

These are shared between production and sandbox:

| Variable | Description |
|----------|-------------|
| `FLUVIUS_SUBSCRIPTION_KEY` | Azure API Management subscription key |
| `FLUVIUS_CLIENT_ID` | Azure AD application (client) ID |
| `FLUVIUS_TENANT_ID` | Azure AD tenant ID |
| `FLUVIUS_SCOPE` | OAuth2 scope (e.g., `api://your-app-id/.default`) |

### Production (Certificate-based auth)

| Variable | Description |
|----------|-------------|
| `FLUVIUS_CERTIFICATE_THUMBPRINT` | Certificate thumbprint (hex format) |
| `FLUVIUS_PRIVATE_KEY` | RSA private key in PEM format |
| `FLUVIUS_PRIVATE_KEY_PATH` | Alternative: path to private key file |

### Sandbox (Client secret auth)

| Variable | Description |
|----------|-------------|
| `FLUVIUS_SANDBOX_CLIENT_SECRET` | Client secret for sandbox |

Note: For sandbox, common values (subscription_key, client_id, tenant_id, scope, data_access_contract_number) automatically fall back to `FLUVIUS_*` if not set with `FLUVIUS_SANDBOX_*` prefix.

### Data Access Contract

| Variable | Description |
|----------|-------------|
| `FLUVIUS_DATA_ACCESS_CONTRACT_NUMBER` | Data access contract number (required, typically constant per service provider) |

### Private Key Formats

The private key can be provided in three ways:
1. **Inline PEM**: Set `FLUVIUS_PRIVATE_KEY` with the full PEM content
2. **Base64-encoded**: Set `FLUVIUS_PRIVATE_KEY` with the PEM content encoded as base64
3. **File path**: Set `FLUVIUS_PRIVATE_KEY_PATH` to point to a `.pem` file

## Usage

### Production (Certificate auth)

```python
from fluvius_energy_api import FluviusEnergyClient, Environment

# Credentials loaded from FLUVIUS_* environment variables
with FluviusEnergyClient(environment=Environment.PRODUCTION) as client:
    response = client.get_mandates()
    for mandate in response.data.mandates:
        print(f"EAN: {mandate.ean}, Status: {mandate.status}")
```

### Sandbox (Client secret auth)

```python
from fluvius_energy_api import FluviusEnergyClient, FluviusCredentials, Environment

# Load credentials - falls back to FLUVIUS_* for common values,
# uses FLUVIUS_SANDBOX_CLIENT_SECRET for auth
sandbox_credentials = FluviusCredentials.from_env(prefix="FLUVIUS_SANDBOX")

with FluviusEnergyClient(credentials=sandbox_credentials, environment=Environment.SANDBOX) as client:
    response = client.get_mandates()
    print(f"Found {len(response.data.mandates)} mandates in sandbox")
```

### Explicit Credentials

```python
from fluvius_energy_api import FluviusEnergyClient, FluviusCredentials, Environment

# Production with certificate
prod_credentials = FluviusCredentials(
    subscription_key="your-subscription-key",
    client_id="your-client-id",
    tenant_id="your-tenant-id",
    scope="api://your-scope/.default",
    certificate_thumbprint="YOUR_CERTIFICATE_THUMBPRINT",
    private_key="-----BEGIN RSA PRIVATE KEY-----\n...\n-----END RSA PRIVATE KEY-----",
)

# Sandbox with client secret
sandbox_credentials = FluviusCredentials(
    subscription_key="your-subscription-key",
    client_id="your-client-id",
    tenant_id="your-tenant-id",
    scope="api://your-scope/.default",
    client_secret="your-client-secret",
)

with FluviusEnergyClient(credentials=prod_credentials, environment=Environment.PRODUCTION) as client:
    mandates = client.get_mandates()
```

### Create Client Session

Create a session for end-users to grant data access:

```python
from fluvius_energy_api import (
    FluviusEnergyClient,
    ClientSessionDataService,
    DataServiceType,
    Environment,
)

with FluviusEnergyClient(environment=Environment.PRODUCTION) as client:
    response = client.create_client_session(
        data_access_contract_number="3599900040",
        reference_number="my-unique-reference",
        flow="B2B",
        data_services=[
            ClientSessionDataService(
                data_service_type=DataServiceType.VH_DAG,
                data_period_from="2024-01-01T00:00:00Z",
            ),
            ClientSessionDataService(
                data_service_type=DataServiceType.VH_KWARTIER_UUR,
                data_period_from="2024-01-01T00:00:00Z",
            ),
        ],
        number_of_eans=5,
    )

    if response.data.status == "success":
        # Build the consent URL for end-users
        consent_url = f"https://mijn.fluvius.be/verbruik/dienstverlener?id={response.data.short_url_identifier}"
        print(f"Consent URL: {consent_url}")
        print(f"Valid until: {response.data.valid_to}")
```

### Get Mandates

Retrieve mandates with optional filters:

```python
from fluvius_energy_api import (
    FluviusEnergyClient,
    EnergyType,
    MandateStatus,
    DataServiceType,
    Environment,
)

with FluviusEnergyClient(environment=Environment.PRODUCTION) as client:
    # Get all mandates
    response = client.get_mandates()

    # Get mandates with filters
    response = client.get_mandates(
        ean="541448800000004312",
        energy_type=EnergyType.ELECTRICITY,
        status=MandateStatus.APPROVED,
        data_service_types=[DataServiceType.VH_DAG, DataServiceType.VH_KWARTIER_UUR],
    )

    for mandate in response.data.mandates:
        print(f"EAN: {mandate.ean}")
        print(f"Status: {mandate.status}")
        print(f"Energy type: {mandate.energy_type}")
        print(f"Data service: {mandate.data_service_type}")
```

### Get Energy Data

Retrieve energy consumption/production data:

```python
from datetime import datetime
from fluvius_energy_api import FluviusEnergyClient, PeriodType, Environment

with FluviusEnergyClient(environment=Environment.PRODUCTION) as client:
    response = client.get_energy(
        ean="541448800000004312",
        period_type=PeriodType.READ_TIME,
        granularity="daily",
        from_date=datetime(2024, 1, 1),
        to_date=datetime(2024, 6, 1),
    )

    headpoint = response.data.headpoint
    print(f"EAN: {headpoint.ean}")
    print(f"Energy type: {headpoint.energy_type}")
```

## Postman Collection

A hand-crafted Postman collection is included in the `postman/` directory covering authentication, client sessions, mandates, and energy data endpoints.

To generate the environment files from your `.env`:

```bash
python postman/generate_postman_env.py
```

This creates two files (gitignored):
- `postman/fluvius-sandbox.postman_environment.json` -- client-secret auth
- `postman/fluvius-production.postman_environment.json` -- certificate / JWT auth

Import the collection (`postman/fluvius-energy-api.postman_collection.json`) and the appropriate environment into Postman. The collection's pre-request script acquires tokens automatically.

The production client assertion is valid for 1 hour. Re-run the script for a fresh one -- the JWT is printed to stdout so you can copy-paste it directly into your already-imported Postman environment without re-importing the file.

## Environments

The client supports two environments:

| Environment | Base URL | Auth Method |
|-------------|----------|-------------|
| `Environment.PRODUCTION` | `https://apihub.fluvius.be/esco-live/v3` | Certificate |
| `Environment.SANDBOX` | `https://apihub.fluvius.be/esco-sbx/v3` | Client secret |

```python
from fluvius_energy_api import FluviusEnergyClient, Environment

# Production
client = FluviusEnergyClient(environment=Environment.PRODUCTION)

# Sandbox
client = FluviusEnergyClient(environment=Environment.SANDBOX)
```

## Error Handling

The client raises specific exceptions for different error scenarios:

```python
from fluvius_energy_api import (
    FluviusEnergyClient,
    FluviusAPIError,
    AuthenticationError,
    ForbiddenError,
    NotFoundError,
    ValidationError,
    ServerError,
    ServiceUnavailableError,
    ConfigurationError,
    Environment,
)

try:
    with FluviusEnergyClient(environment=Environment.PRODUCTION) as client:
        response = client.get_mandates()
except ConfigurationError as e:
    print(f"Configuration error: {e}")
except AuthenticationError as e:
    print(f"Authentication failed: {e}")
except ForbiddenError as e:
    print(f"Access forbidden: {e}")
except NotFoundError as e:
    print(f"Resource not found: {e}")
except ValidationError as e:
    print(f"Validation error: {e}")
    if e.validation_errors:
        for error in e.validation_errors:
            print(f"  - {error}")
except ServerError as e:
    print(f"Server error: {e}")
except ServiceUnavailableError as e:
    print(f"Service unavailable: {e}")
except FluviusAPIError as e:
    print(f"API error: {e}")
```

## Data Service Types

Available data service types:

| Type | Description |
|------|-------------|
| `VH_DAG` | Daily consumption history |
| `VH_KWARTIER_UUR` | Quarter-hourly/hourly consumption history |
| `VH_ONBEPAALD` | Indefinite consumption history |
| `IG` | Installation data |
## License

GNU AFFERO GENERAL PUBLIC LICENSE