blob: e56d83bf84d18f5a5c2b7188e49144980c591d52 (
plain)
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
|
"""
Stream Data from Tibber Pulse API.
Cf. https://github.com/Danielhiversen/pyTibber
"""
import asyncio
import aiohttp
import tibber
# import pprint
import time
import Common.influx as influx
async def _callback(pkg):
data = pkg.get("data")
if data is None:
return
measurements = data.get("liveMeasurement")
# print(type(measurements))
# print(measurements.keys())
# dump some data
# print(int(time.time()))
# # pprint.pprint(data.get("liveMeasurement"))
# pprint.pprint(measurements)
# collect some data
epoch = int(time.time())
lines = []
# accumulated consumption
line = "pulse_accumulatedConsumption value=%.2f %i" % \
(measurements['accumulatedConsumption'], epoch)
lines.append(line)
# accumulated cost
line = "pulse_accumulatedCost value=%.2f %i" % \
(measurements['accumulatedCost'], epoch)
lines.append(line)
# power
line = "pulse_power value=%ii %i" % (measurements['power'], epoch)
lines.append(line)
# voltages per phase
for kk in [ 1, 2, 3 ]:
key = "voltagePhase%i" % kk
if not measurements[key] == None:
line = "pulse_%s value=%.1f %i" % \
(key, measurements[key], epoch)
lines.append(line)
# signal strength
if not measurements['signalStrength'] == None:
line = "pulse_signalStrength value=%ii %i" % \
(measurements['signalStrength'], epoch)
lines.append(line)
# send to database
post_data = "\n".join(lines)
# print(post_data)
influx.post_data(post_data)
async def run():
async with aiohttp.ClientSession() as session:
tibber_connection = tibber.Tibber(ACCESS_TOKEN, websession=session)
await tibber_connection.update_info()
home = tibber_connection.get_homes()[0]
await home.rt_subscribe(_callback)
if __name__ == '__main__':
# ACCESS_TOKEN = tibber.DEMO_TOKEN
with open('userpass_tibber', 'r') as f:
line = f.readline()
line = line.strip()
ACCESS_TOKEN = line
loop = asyncio.get_event_loop()
asyncio.ensure_future(run())
loop.run_forever()
|
