forked from hansij66/dsmr2mqtt
-
Notifications
You must be signed in to change notification settings - Fork 0
/
P1_parser.py
executable file
·228 lines (178 loc) · 7.71 KB
/
P1_parser.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
"""
Parses P1 telegrams to MQTT messages
Select MQTT messages based on dsmr50.py
Queue MQTT messages
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
import threading
import copy
import re
import time
import json
import config as cfg
import dsmr50 as dsmr
# Logging
import __main__
import logging
import os
script = os.path.basename(__main__.__file__)
script = os.path.splitext(script)[0]
logger = logging.getLogger(script + "." + __name__)
class ParseTelegrams(threading.Thread):
"""
"""
def __init__(self, trigger, stopper, mqtt, telegram):
"""
Args:
:param threading.Event() trigger: signals that new telegram is available
:param threading.Event() stopper: stops thread
:param mqtt.mqttclient() mqtt: reference to mqtt worker
:param list() telegram: dsmr telegram
"""
logger.debug(">>")
super().__init__()
self.__trigger = trigger
self.__stopper = stopper
self.__telegram = telegram
self.__mqtt = mqtt
self.__prevjsondict = {}
# Count number of topics which will always be included in MQTT json
# timestamp key:value topic
self.__nroftopics = 1
logger.debug(f"NROF = {self.__nroftopics}")
def __del__(self):
logger.debug(">>")
def __publish_telegram(self, listofjsondicts):
# publish the dictionaries per topic
for d in listofjsondicts:
topic = d["topic"]
# remove topic key:value pair, otherwise it will be packed in the mqtt json
d.pop("topic")
# There is always a timestamp key:value and a database:influxdb in the dictionary (len = 2)
# If there are no other key-value pairs, skip publishing to MQTT
if len(d) > self.__nroftopics:
topic = cfg.MQTT_TOPIC_PREFIX + "/" + topic
# make resilient against double forward slashes in topic
topic = topic.replace('//', '/')
message = json.dumps(d, sort_keys=True, separators=(',', ':'))
self.__mqtt.do_publish(topic, message, retain=False)
return
def __decode_telegram_element(self, index, element, ts, listofjsondicts):
# logger.debug(f">> index={index}; element={element}")
try:
# Extract result from telegram element, based on dsmr definition
# Cast result to type defined in dsmr50
# .....this is normally "dangerous", as you can define any python function....but it is hardcoded
# in dsmr50.py file, which should be ro for regular users
cast = dsmr.definition[index][dsmr.DATATYPE]
dsmr_data = re.match(dsmr.definition[index][dsmr.REGEX], element).group(1)
# multiplication factor for data, eg to convert kW to W
multiply = dsmr.definition[index][dsmr.MULTIPLICATION]
# dict & json pair
# tag:data
# If type is string, there is no multiplication factor
if cast != "str":
data = eval(cast)(dsmr_data) * eval(cast)(multiply)
# Check if data is zero while not allowed
# TODO: seems not to work? For el_consumed
if dsmr.DATAVALIDATION == "1" and data == 0:
# throw exception
logger.warning(f"Throw exception, data == 0, not allowed")
raise ValueError("data == 0, not allowed, skip telegram")
else:
data = eval(cast)(dsmr_data)
# logger.debug(f"CAST = {}")
# dict & json pair
# tag:data
tag = str(dsmr.definition[index][dsmr.MQTT_TAG])
# Calculate, based on specified maxrate, the minimum time between to mqtt messages for a specific tag
# MAXRATE = 0 --> don't publish
if int(dsmr.definition[index][dsmr.MAXRATE]) == 0:
mintimeinterval = -1
else:
mintimeinterval = int(3600/int(dsmr.definition[index][dsmr.MAXRATE]))
# MQTT topic
topic = str(dsmr.definition[index][dsmr.MQTT_TOPIC])
# TODO we should use the timestamp from dsmr telegram...but not sure yet
# about timezone & daylight saving
# influxdb timestamp should be UTC epoch
# https://stackoverflow.com/questions/8777753/converting-datetime-date-to-utc-timestamp-in-python
# if index == "0-0:1.0.0":
# t = datetime.strptime(data, '%y%m%d%H%M%S')
# If topic does not exist yet, create & initialize dictionary for this topic;
# Add tag:data pairs; which will be converted to mqtt json later on.
# Add dictionary to list
if not any(dictionary['topic'] == topic for dictionary in listofjsondicts):
# TODO
# Is this correct?
# Append per single dictionary element?
dict_element = {}
dict_element["topic"] = topic
dict_element["timestamp"] = ts
# if cfg.INFLUXDB:
# d["database"] = cfg.INFLUXDB
listofjsondicts.append(dict_element)
for dictionary in listofjsondicts:
if dictionary['topic'] == topic:
# get previous tag:data ts, which is mqtt broadcasted
try:
prevts = self.__prevjsondict[topic+tag]
except KeyError:
prevts = 0
# Calculate time in seconds between between previous mqtt transmission and current
timeelapsed = int(ts - prevts)
# Only broadcast this data:tag if sufficient time has elapsed based on maxrate
if (mintimeinterval != -1) and (timeelapsed > mintimeinterval):
dictionary[tag] = data
# and store current ts
self.__prevjsondict[topic+tag] = ts
except Exception as e:
pass
def __decode_telegrams(self, telegram):
"""
Args:
:param list telegram:
Returns:
"""
logger.debug(f">>")
# list of dictionaries of mqtt messages which will be converted to json format
listofjsondicts = list()
# global UTC timestamp for all mqtt message for telegraf
# this has nanoseconds accuracy
# TODO...use timestamp from dsmr telegram
# ts = int(round(time.time() * 1000)) * 1000000
# epoch
ts = int(time.time())
for element in telegram:
try:
# Extract the identifier (eg "1-0:1.8.1") of the element
# and use this as index for dsmr.definition
index = re.match(r"(\d{0,3}-\d{0,3}:\d{0,3}\.\d{0,3}\.\d{0,3}).*", element).group(1)
self.__decode_telegram_element(index, element, ts, listofjsondicts)
except Exception as e:
# To handle empty lines or lines not matching dsmr definitions (checksum, header, empty line)
pass
self.__publish_telegram(listofjsondicts)
def run(self):
logger.debug(">>")
while not self.__stopper.is_set():
# block till event is set, but implement timeout to allow stopper
self.__trigger.wait(timeout=1)
if self.__trigger.is_set():
# Make copy of the telegram, for further parsing
telegram = copy.deepcopy(self.__telegram)
# Clear telegram list for next capture by ReadSerial class
self.__telegram.clear()
# Clear trigger, serial can continue
self.__trigger.clear()
self.__decode_telegrams(telegram)
logger.debug("<<")