-
Notifications
You must be signed in to change notification settings - Fork 0
/
prometheus_enocean_exporter.py
executable file
·324 lines (268 loc) · 10.9 KB
/
prometheus_enocean_exporter.py
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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
#!/usr/bin/env python3
import argparse
from collections import defaultdict
import logging
import queue
from threading import Lock
import time
from typing import Any, DefaultDict, Dict, Iterable, NamedTuple, Optional, Tuple
from enocean.communicators.serialcommunicator import SerialCommunicator
from enocean.protocol.constants import RORG
from prometheus_client import Metric, REGISTRY, start_http_server
from prometheus_client.core import GaugeMetricFamily
import yaml
class MetaValueAtTime(NamedTuple):
value: Any
mono_timestamp: float
unix_timestamp: float
class ValueAtTime(NamedTuple):
meta: MetaValueAtTime
raw_value: float
description: str
unit: str
LOGGER = logging.getLogger('prometheus_enocean_exporter')
DEFAULT_MAX_VALUE_AGE_S = 15.0 * 60.0
def obtain_mono_timestamp() -> float:
return time.monotonic()
def obtain_unix_timestamp() -> float:
return time.time()
class EnOceanCollector:
def __init__(self) -> None:
self.serial_port: Optional[str] = None
self.known_device_file: Optional[str] = None
self.learning: bool = True
self.max_value_age_s: float = DEFAULT_MAX_VALUE_AGE_S
self._data_lock = Lock()
# SenderID -> (RorgFunc, RorgType)
self.known_4bs_eeps: Dict[str, Tuple[int, int]] = {}
# SenderID -> ValueShortcut -> value
self.values: DefaultDict[str, Dict[str, ValueAtTime]] = defaultdict(dict)
# SenderID -> dBm
self.sender_dbm: Dict[str, MetaValueAtTime] = {}
def collect(self) -> Iterable[Metric]:
value_metric = GaugeMetricFamily(
'enocean_value',
'A value obtained from an EnOcean device.',
labels=['source', 'key'],
)
raw_value_metric = GaugeMetricFamily(
'enocean_value_raw',
'A raw value obtained from an EnOcean device.',
labels=['source', 'key'],
)
value_description_metric = GaugeMetricFamily(
'enocean_value_description',
'The description for a value obtained from an EnOcean device.',
labels=['source', 'key', 'description'],
)
value_unit_metric = GaugeMetricFamily(
'enocean_value_unit',
'The unit for a value obtained from an EnOcean device.',
labels=['source', 'key', 'unit'],
)
value_timestamp_metric = GaugeMetricFamily(
'enocean_value_timestamp',
(
'The Unix timestamp of the last update of a value obtained from'
' an EnOcean device.'
),
labels=['source', 'key'],
)
dbm_metric = GaugeMetricFamily(
'enocean_last_transmission_dBm',
(
'The strength of the last transmission of an EnOcean device'
' in decibels referenced to one milliwatt (dBm).'
),
labels=['source'],
)
transmission_timestamp_metric = GaugeMetricFamily(
'enocean_last_transmission_timestamp',
(
'The Unix timestamp of the last transmission of an EnOcean'
' device.'
),
labels=['source'],
)
with self._data_lock:
mono_now = obtain_mono_timestamp()
for sender, shortcut_value in self.values.items():
for shortcut, value in shortcut_value.items():
if mono_now - value.meta.mono_timestamp > self.max_value_age_s:
# value has aged out
continue
value_timestamp_metric.add_metric(
[sender, shortcut], value.meta.unix_timestamp,
)
if isinstance(value.meta.value, (int, float)):
value_metric.add_metric(
[sender, shortcut], value.meta.value,
)
raw_value_metric.add_metric(
[sender, shortcut], value.raw_value,
)
value_description_metric.add_metric(
[sender, shortcut, value.description], 1,
)
value_unit_metric.add_metric(
[sender, shortcut, value.unit], 1,
)
for sender, dbm_at_time in self.sender_dbm.items():
if mono_now - dbm_at_time.mono_timestamp > self.max_value_age_s:
# value has aged out
continue
dbm_metric.add_metric(
[sender], dbm_at_time.value
)
transmission_timestamp_metric.add_metric(
[sender], dbm_at_time.unix_timestamp,
)
yield value_metric
yield raw_value_metric
yield value_description_metric
yield value_unit_metric
yield value_timestamp_metric
yield dbm_metric
yield transmission_timestamp_metric
def _load_known_devices(self) -> None:
if self.known_device_file is None:
return
# load known-device file
try:
with open(self.known_device_file, 'r', encoding='utf-8') as f:
known_devices = yaml.safe_load(f)
self.known_4bs_eeps.update(
known_devices.get('known_4bs_eeps', {}),
)
except FileNotFoundError:
pass
def _write_known_devices(self) -> None:
if self.known_device_file is None:
return
known_devices = {
'known_4bs_eeps': self.known_4bs_eeps,
}
with open(self.known_device_file, 'w', encoding='utf-8') as f:
yaml.safe_dump(known_devices, f)
def run(self) -> None:
self._load_known_devices()
cereal = SerialCommunicator(port=self.serial_port)
cereal.start()
# pre-obtain the base ID
base_id = cereal.base_id
LOGGER.info("communicator base ID: %s", base_id)
try:
LOGGER.info("ready to read")
while cereal.is_alive():
try:
packet = cereal.receive.get(block=True, timeout=1)
except queue.Empty:
continue
LOGGER.debug('obtained packet: %s', packet)
sender_hex = getattr(packet, 'sender_hex', None)
if sender_hex is None:
LOGGER.debug('no sender_hex; skipping packet')
continue
dbm = getattr(packet, 'dBm', None)
if dbm is not None:
with self._data_lock:
self.sender_dbm[sender_hex] = MetaValueAtTime(
value=dbm,
mono_timestamp=obtain_mono_timestamp(),
unix_timestamp=obtain_unix_timestamp(),
)
if packet.learn and packet.rorg != RORG.VLD:
# new device to learn
if not self.learning:
# but we don't want to
LOGGER.debug("learning packet but we are not learning")
continue
if not getattr(packet, 'contains_eep', False):
# TODO: handle packets without EEP
LOGGER.debug("learning packet without EEP")
continue
# know this device
with self._data_lock:
self.known_4bs_eeps[packet.sender_hex] = (
packet.rorg_func, packet.rorg_type
)
LOGGER.info(
"learned via ROrg 0x%02x that sender %s has func 0x%02x type 0x%02x",
packet.rorg, packet.sender_hex, packet.rorg_func, packet.rorg_type
)
# update known-device file
self._write_known_devices()
continue
# non-learning packet
rorg_func_type = self.known_4bs_eeps.get(packet.sender_hex)
if rorg_func_type is None:
# unknown source
LOGGER.debug("source %s is not known", packet.sender_hex)
continue
rorg_func, rorg_type = rorg_func_type
packet.parse_eep(rorg_func, rorg_type)
LOGGER.debug("parsed packet: %r", packet.parsed)
# add timestamps
shortcut_to_value: Dict[str, ValueAtTime] = {}
for short, val in packet.parsed.items():
shortcut_to_value[short] = ValueAtTime(
meta=MetaValueAtTime(
value=val['value'],
mono_timestamp=obtain_mono_timestamp(),
unix_timestamp=obtain_unix_timestamp(),
),
raw_value=val['raw_value'],
description=val['description'],
unit=val['unit'],
)
with self._data_lock:
self.values[packet.sender_hex].update(shortcut_to_value)
LOGGER.debug(
"values for %s updated to %r",
packet.sender_hex, self.values[packet.sender_hex],
)
finally:
cereal.stop()
def main():
parser = argparse.ArgumentParser()
parser.add_argument(
"--web.listen-port",
dest='web_listen_port', type=int, required=True,
)
parser.add_argument(
"--web.listen-address",
dest='web_listen_address', type=str, default="",
)
parser.add_argument(
"--serial.port",
dest='serial_port', type=str, default=None,
)
parser.add_argument(
"--config.known-device-file",
dest='known_device_file', type=str, default=None,
)
parser.add_argument(
"--config.max-value-age",
dest='max_value_age', metavar='SECONDS', type=float,
default=DEFAULT_MAX_VALUE_AGE_S
)
args = parser.parse_args()
# empty string now raises exceptions; fold it to None
# (keep accepting empty strings to allow args --web.listen-address "")
if not args.web_listen_address:
args.web_listen_address = None
logging.basicConfig(level=logging.DEBUG)
logging.getLogger('enocean').setLevel(logging.WARNING)
if args.serial_port == '':
args.serial_port = None
if args.known_device_file == '':
args.known_device_file = None
enocean_collector = EnOceanCollector()
enocean_collector.serial_port = args.serial_port
enocean_collector.known_device_file = args.known_device_file
enocean_collector.max_value_age_s = args.max_value_age
REGISTRY.register(enocean_collector)
start_http_server(args.web_listen_port, args.web_listen_address)
enocean_collector.run()
if __name__ == '__main__':
main()