44 lines
1.4 KiB
Python
44 lines
1.4 KiB
Python
import os
|
|
from dsmr_parser import telegram_specifications
|
|
from dsmr_parser.clients import SerialReader, SERIAL_SETTINGS_V2_2, SERIAL_SETTINGS_V4, SERIAL_SETTINGS_V5
|
|
|
|
from app_settings import Settings
|
|
from device_dsmr import Device_DSMR
|
|
from serial import SerialException
|
|
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def main():
|
|
logging.basicConfig(format='%(asctime)s (%(levelname)s) [%(name)s] %(message)s', level=logging.INFO)
|
|
cfg = Settings()
|
|
protocol_version: str = str(cfg.dsmr_protocol)
|
|
dev = str(cfg.dsmr_port)
|
|
if 'V2' in protocol_version:
|
|
devsettings = SERIAL_SETTINGS_V2_2
|
|
spec = telegram_specifications.V2_2
|
|
elif 'V3' in protocol_version:
|
|
devsettings = SERIAL_SETTINGS_V4
|
|
spec = telegram_specifications.V3
|
|
elif 'V4' in protocol_version:
|
|
devsettings = SERIAL_SETTINGS_V4
|
|
spec = telegram_specifications.V4
|
|
else:
|
|
devsettings = SERIAL_SETTINGS_V5
|
|
spec = telegram_specifications.V5
|
|
|
|
device = Device_DSMR("dsmr", name="Digitale Slimme Meter")
|
|
|
|
serial_reader = SerialReader(dev, devsettings, spec)
|
|
while True:
|
|
try:
|
|
telegram = next(serial_reader.read_as_object())
|
|
except SerialException as serial_exc:
|
|
logger.warning("Serial exception", exc_info=serial_exc)
|
|
for telegram in serial_reader.read_as_object():
|
|
device.update(telegram)
|
|
|
|
if __name__ == "__main__":
|
|
main() |