1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
# energyid-webhooks
Python package for interfacing with EnergyID Webhooks (supports both V1 and V2 APIs)
## Installation
```bash
pip install energyid-webhooks
```
## Features
- **V2 API support** with device provisioning and claiming
- Authentication with automatic token renewal
- Smart sensor bundling for efficient data transmission
- Async support
- Full type hints
## Type Checking
This package is fully type-hinted and checked with strict mypy settings.
## V2 API Usage (Recommended)
The V2 API includes device provisioning, authentication, and smart bundling:
```python
import asyncio
from energyid_webhooks import WebhookClient
# Create the client
client = WebhookClient(
client_id="your_provisioning_key",
client_secret="your_provisioning_secret",
device_id="unique_device_id",
device_name="My Device",
reauth_interval=24 # Hours between token refresh
)
async def main():
# Authenticate with EnergyID
is_claimed = await client.authenticate()
if not is_claimed:
# Device needs to be claimed
claim_info = client.get_claim_info()
print(f"Claim your device at {claim_info['claim_url']} with code {claim_info['claim_code']}")
# Wait for claiming...
return
# Add and update sensors
client.update_sensor("el", 1250.5) # Electricity consumption
client.update_sensor("pv", 3560.2) # Solar production
# Synchronize with EnergyID
await client.synchronize_sensors()
# Or enable automatic synchronization
client.start_auto_sync(interval_seconds=300) # Every 5 minutes
# Don't forget to close when done
await client.close()
asyncio.run(main())
```
## V1 API Usage (Legacy)
The original API is still available for backward compatibility:
```python
from energyid_webhooks import WebhookClientV1, WebhookPayload
url = "https://app.energyid.eu/integrations/WebhookIn/..."
client = WebhookClientV1(url)
# Post some data to the webhook
data = {
'remoteId': 'my-solar-inverter',
'remoteName': 'My Solar Panels',
'metric': 'solarPhotovoltaicProduction',
'metricKind': 'total',
'unit': 'kWh',
'interval': 'P1D',
'data': [['2022-10-05T08:00+0200', 0.004]]
}
client.post(data)
```
## V1 Async Usage (Legacy)
```python
import asyncio
from energyid_webhooks import WebhookClientAsyncV1
url = "https://app.energyid.eu/integrations/WebhookIn/..."
client = WebhookClientAsyncV1(url)
async def main():
data = {
'remoteId': 'my-solar-inverter',
'remoteName': 'My Solar Panels',
'metric': 'solarPhotovoltaicProduction',
'metricKind': 'total',
'unit': 'kWh',
'interval': 'P1D',
'data': [['2022-10-05T08:00+0200', 0.004]]
}
await client.post(data)
asyncio.run(main())
```
## Demo Notebook
See [energyid_webhook_demo.ipynb](demos/energyid_webhook_demo.ipynb) for a complete V2 API demo.
## Development
### Setup
```bash
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
```
|
