smart-energy-monitor/box-pc/application/inputs/snap7_connect.py

from snap7.client import Client
import time
from datetime import datetime
import struct

from structures.plant import *
from inputs.common import Input

localtz = datetime.now().astimezone().tzinfo

class SiemensCPU(Input):

  interval = 0.05

  def __init__(self, address) -> None:
    super().__init__(self.read_handler)
    self.address = address
    self.cpu = Client()

  def start(self):
    self.cpu.connect(self.address, rack=0, slot=0)
    self.cpu.get_connected()
    super().start()

  def read_handler(self):
    timestamp = datetime.now(localtz)
    cpu_state = self.cpu.get_cpu_state() == "S7CpuStatusRun"

    status = self.cpu.db_read(db_number=3, start=0, size=4)

    hydraulics_powered = status[0] & 1 > 0

    data = struct.unpack(">BBBB", status)

    cooling_enabled = tuple([data[1] & (1 << i) > 0 for i in range(8)])
    heating_enabled = tuple([data[2] & (1 << i) > 0 for i in range(8)])

    self._q.put(PlantState(timestamp, "S7", cpu_state, hydraulics_powered, cooling_enabled, heating_enabled))

  def get_timestamp(self, cpu_time):
    if not self.cpu_start_time:
      self.synchronize()
    cpu_diff = cpu_time - self.cpu_start_time
    date = datetime.fromtimestamp(self.local_start_time + cpu_diff, localtz)
    return date