| 1234567891011121314151617181920212223242526272829303132333435363738394041 |
- import snap7
- import time
- import re
- from datetime import datetime
- from inputs.common import Input
- localtz = datetime.now().astimezone().tzinfo
- class SiemensCPU(Input):
- interval = 0.02
- def __init__(self):
- super().__init__(self.read_handler)
- self.server = snap7.server.Server(True)
- size = 100
- self.DBdata = (snap7.types.wordlen_to_ctypes[snap7.types.WordLen.Byte.value] * size)()
- self.server.register_area(snap7.types.srvAreaDB, 1, self.DBdata)
- self.server.start(102)
- def read_handler(self):
- event : snap7.types.SrvEvent
- while event := self.server.pick_event():
- timestamp = datetime.now(localtz)
- text = self.server.event_text(event)
- match = re.match("^(?P<datetime>\d+-\d+-\d+ \d+:\d+:\d+) \[(?P<host>[\w\.:]+)\] (?P<type>[\w ]+), Area : (?P<area>.+), Start : (?P<start>\d+), Size : (?P<size>\d+) --> (?P<response>.+)$", text)
- if not match:
- print(text)
- continue
-
- if match.group("type") != "Write request" or match.group("area") != "DB1":
- print(text)
- continue
-
- raw = bytearray(self.DBdata)
-
- #print("{} {:4d} {:4d}".format(timestamp.microsecond, int(match.group("start")), int(match.group("size"))))
- if int(match.group("start")) < 30:
- self.queue_ifm_from_bytes("S7", timestamp, raw[0:30])
- if int(match.group("start")) < 68 and int(match.group("start")) + int(match.group("size")) > 32:
- self.queue_energy_meter_from_bytes("S7", timestamp, raw[32:68])
|
