add config file

30_Implementierung.md

@@ -28,4 +28,24 @@ Zur Erfassung der Energiewerte kommen verschiedene Systeme in Frage. Alle diese
   - IFM elektronische Sicherung mit IO-Link
 - Testaufbau
   - Energiemonitor, Feldbus, SPS, Netzwerk, IPC
-- Einbau in eine fertige Anlage
+
+
+## Software
+
+Die Software für dieses Projekt übernimmt die Aufgaben der Datenverarbeitung und -speicherung. Zudem müssen die aufgenommenen Messwerte von der zentralen Steuerung der Kernschießmaschine eingelesen werden. Das selbst geschriebene Programm "PLC-Connector" ist in diesem Abschnitt genauer beschrieben. Die Benutzerschnittstelle zum Benutzer des Systems kann im nachhinein durch ein generisches Interface zur Datenbank geschehen, da die dort enthaltenen Daten schon durch das Programm "PLC-Connector" und dessen entsprechende Signalverarbeitungsmodule aufbereitet sind. 
+
+*PLC-Connector* ist modular aufgebaut, so dass verschiedene Komponenten einfach ausgetauscht werden können. Die Module sind in drei Kategorien unterteilt:
+
+1. Inputs
+
+    Ein *Input*-Modul stellt die Verbindung zu einer bestimmten Komponente der Anlage auf und bezieht über diese periodisch die Messwerte eines oder mehrerer Sensoren. Die Komponenten können beispielsweise Feldbusverteiler, netzwerkfähige Sensoren oder die zentrale Steuerung einer Anlage sein. Je nach Modul kommen unterschiedliche Protokolle zur Kommunikation zum Einsatz. Unter diesen sind zum Beispiel EtherNet/IP und das S7-Protokoll vertreten.
+
+2. Middlewares
+
+    Eine *Middleware* ist ein Algorithmus, der strukturierte Datensätze der aktiven *Input*-Module entgegen nimmt und die für die Auswertung interessante Informationen extrahiert. Eine erste *Middleware* nimmt beispielsweise die Werte mehrerer *Inputs* entgegen und führt eine zeitliche Korrelation durch. Die dadurch entstandenen Datenpakete können dann optional an weitere *Middlewares* weitergegeben werden, welche andere Analysen durchführen. Schließlich kann eine Middleware ihre Ergebnisse an die *Outputs* übergeben.
+
+3. Outputs
+
+    *Outputs* sind Datensenken, welche Datensätze von *Middlewares* entgegen nehmen und abspeichern. Ein *Output* archiviert beispielsweise die Datensätze in CSV Dateien und ein anderes sendet sie an eine Influxdb-Datenbank.
+
+Die Module werden durch ein zentrales Python-Programm geladen, welches auch die initiale Konfiguration und die Datenübertragung zwischen den Modulen orchestriert. Ansonsten arbeiten die Module komplett autonom. Die Verbindungen und Parameter der Module sind in einer zentralen Konfigurationsdatei `config.yml` definiert. 

box-pc/plc-connector/config.yml

@@ -0,0 +1,46 @@
+
+Inputs:
+  - SiemensCPU: siemens.snap7_connect
+    enabled: False
+    host: "192.168.0.10"
+
+  - SiemensServer: siemens.snap7_server
+    enabled: True
+    port: 102
+
+  - Balluff: balluff.balluff_html
+    enabled: False
+
+  - AllenBradleyCPU: rockwell.allen_bradley_connect
+    host: "192.168.1.15"
+    enabled: True
+  
+  - Input: dummy
+    message: Hello World!
+
+  - Replay: replay_influxdb
+    url: "http://influxdb:8086"
+    token: "XPBViJ3s4JL9_wPffwd5M2EgXj5hcUgT0n4jNhv7m6-NC-6SSxQ3run4kXtWBvOk-FYr1VG5Tj5WcoHgjge9jw=="
+    org: "laempe"
+    bucket: "energy-monitor"
+    start_time: 03.05.2022 07:00:00
+
+Middlewares:
+  - PrintStats: print_stats
+    enabled: False
+  - TimeCorrelation: time_correlation
+    submodules:
+    - PrintStats: print_stats
+
+Outputs:
+  - CSVStorage: csv_file
+    path: logs
+
+  - InfluxDB: influxdb
+    url: "http://influxdb:8086"
+    token: "XPBViJ3s4JL9_wPffwd5M2EgXj5hcUgT0n4jNhv7m6-NC-6SSxQ3run4kXtWBvOk-FYr1VG5Tj5WcoHgjge9jw=="
+    org: "laempe"
+    bucket: "energy-monitor"
+
+  - SqliteDB: sqlite_db
+    enabled: False

box-pc/plc-connector/database/__init__.py

@@ -1,4 +0,0 @@
-
-from .csvFile import *
-from .influxdb import *
-from .sqliteDb import *

box-pc/plc-connector/inputs/__main__.py

@@ -2,8 +2,8 @@ import time
 
 from snap7.client import Client
 
-from .allen_bradley_connect import AllenBradleyCPU
-from .snap7_server import SiemensServer
+from rockwell.allen_bradley_connect import AllenBradleyCPU
+from siemens.snap7_server import SiemensServer
 
 
 cpu = SiemensServer()

box-pc/plc-connector/inputs/dummy.py

@@ -0,0 +1,13 @@
+import logging
+from inputs.common import Input as Inp
+
+logger = logging.getLogger(__name__)
+
+class Input(Inp):
+  def __init__(self, message) -> None:
+    super().__init__(self.read_handler)
+    logger.info(message)
+    self.interval = 1
+  
+  def read_handler(self):
+    pass

box-pc/plc-connector/inputs/replay_influxdb.py

@@ -0,0 +1,43 @@
+import logging, time
+from datetime import datetime, timedelta
+from influxdb_client import InfluxDBClient
+
+from inputs.common import Input
+
+logger = logging.getLogger(__name__)
+
+class Replay(Input):
+  def __init__(self, url, token, org, bucket, start_time) -> None:
+    super().__init__(self.read_handler)
+    self.interval = 1.0
+    self.client = InfluxDBClient(url, token, org=org)
+    self.bucket = bucket
+
+    self.query_api = self.client.query_api()
+    self.current_time = datetime.strptime(start_time, "%d.%m.%Y %H:%M:%S")
+
+  
+  def read_handler(self):
+    start = self.current_time
+    logger.info(start)
+    end = start + timedelta(seconds=1)
+    for result in self.query(start, end):
+      self._q.put(result)
+    self.current_time = end
+
+  def query(self, start, stop):
+    query = f'from(bucket:"{self.bucket}")\
+      |> range(start: {start}, stop: {stop})'
+    result = self.query_api.query(query=query)
+    results = []
+    for table in result:
+      for record in table.records:
+        res = {
+          'series': table,
+          'timestamp': record.get_time(),
+        }
+        res.extend(record.values)
+        results.append(res)
+    return results
+
+    

box-pc/plc-connector/inputs/allen_bradley_connect.py → box-pc/plc-connector/inputs/rockwell/allen_bradley_connect.py

@@ -9,6 +9,7 @@ from structures.plant import *
 from inputs.common import Input
 
 localtz = datetime.now().astimezone().tzinfo
+logger = logging.getLogger(__name__)
 
 class AllenBradleyCPU(Input):
  
@@ -81,6 +82,7 @@ class AllenBradleyCPU(Input):
       "B18[72]", # vertical_mixer_sand_slide_gate_open
       "B18[73]", # sand_sender
     ]
+    
   def read_handler(self):
     timestamp = datetime.now(localtz)
     ret = self.comm.Read(self.tags)
@@ -88,4 +90,4 @@ class AllenBradleyCPU(Input):
       values = [r.Value for r in ret]
       self._q.put(PlantState(timestamp, "AB", *values))
     else:
-      logging.error("CPU read: " + ret[0].Status)
+      logger.error("CPU read: " + ret[0].Status)

box-pc/plc-connector/inputs/snap7_connect.py → box-pc/plc-connector/inputs/siemens/snap7_connect.py

@@ -12,6 +12,7 @@ from structures.plant import *
 from inputs.common import Input
 
 localtz = datetime.now().astimezone().tzinfo
+logger = logging.getLogger(__name__)
 
 class SiemensCPU(Input):
 
@@ -26,7 +27,7 @@ class SiemensCPU(Input):
     try:
       self.cpu.connect(self.address, rack=0, slot=0)
     except Snap7Exception as ex:
-      logging.exception(ex)
+      logger.exception(ex)
     super().start()
 
   def read_handler(self):

box-pc/plc-connector/inputs/snap7_server.py → box-pc/plc-connector/inputs/siemens/snap7_server.py

@@ -1,4 +1,3 @@
-from psutil import cpu_times
 import snap7
 import logging
 import struct
@@ -8,13 +7,14 @@ from datetime import datetime, tzinfo
 from inputs.common import Input
 
 localtz = datetime.now().astimezone().tzinfo
+logger = logging.getLogger(__name__)
 
 class SiemensServer(Input):
   interval = 0.02
 
   time_offset = None
 
-  def __init__(self):
+  def __init__(self, port = 102):
     super().__init__(self.read_handler)
     self.server = snap7.server.Server(True)
     size = 100
@@ -22,19 +22,22 @@ class SiemensServer(Input):
     self.DB2 = (snap7.types.wordlen_to_ctypes[snap7.types.WordLen.Byte.value] * size)()
     self.server.register_area(snap7.types.srvAreaDB, 1, self.DB1)
     self.server.register_area(snap7.types.srvAreaDB, 2, self.DB2)
-    self.server.start(102)
+    self.server.start(port)
 
   def read_handler(self):
     event : snap7.types.SrvEvent
-    while event := self.server.pick_event():
+    while True:
+      event = self.server.pick_event()
+      if not event:
+        break
       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:
-        logging.warn(text)
+        logger.warn(text)
         continue
       
       if match.group("type") != "Write request":
-        logging.warn(text)
+        logger.warn(text)
         continue
       
       if int(match.group("start")) + int(match.group("size")) <= 4:

box-pc/plc-connector/main.py

@@ -1,66 +1,83 @@
-from inputs.snap7_server import SiemensServer
-from inputs.snap7_connect import SiemensCPU
-from inputs.balluff_html import Balluff
-from inputs.allen_bradley_connect import AllenBradleyCPU
-from database import *
 import logging
 import time
+import yaml
+from importlib import import_module
 
-logging.basicConfig(level=logging.WARNING)
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
 
-logging.warning("starting")
+logger.info("starting")
 
-sources = [
-  #SiemensCPU("192.168.0.10"),
-  SiemensServer(),
-  #Balluff(),
-  AllenBradleyCPU("192.168.1.15"),
-]
+# read config
+with open("config.yml", "r") as f:
+  config = yaml.load(f, Loader=yaml.FullLoader)
 
-sinks = [
-  InfluxDB("http://influxdb:8086"),
-  CSVStorage("logs"),
-]
+def createModules(configItems, type):
+  for item in configItems:
+    cls = next(iter(item))
+    module = import_module(f"{type}s.{item[cls]}")
+    if item.get('enabled') == False:
+      continue
+    params = item.copy()
+    params.pop(cls, None)
+    params.pop('enabled', None)
+    params.pop('submodules', None)
+    try:
+      yield getattr(module, cls)(**params)
+    except Exception as ex:
+      logger.exception(F"{type} {cls} couldn't be initialized.")
+      raise
 
-for source in sources:
-  source.start()
+# setup input modules
+inputs = list(createModules(config['Inputs'], "input"))
 
-logging.warning("started sources")
+# setup middlewares recursively
+def createMiddlewares(configItems, parent = None):
+  items = [dict(x, parent=parent) for x in configItems if x.get('enabled') != False]
+  middlewares = list(createModules(items, "middleware"))
+  for (item, middleware) in zip(items, middlewares):
+    if 'submodules' in item:
+      middleware.submodules = list(createMiddlewares(item['submodules'], middleware))
+  return middlewares
 
-startTime = 0
+middlewares = createMiddlewares(config['Middlewares'])
 
-def printStats(values):
-  global startTime
-  counts = {}
-  dt = time.monotonic() - startTime
-  startTime = time.monotonic()
-  text = ""
-  for meas in values:
-    id = "{} {}".format(meas.series, meas.source)
-    if id in counts:
-      counts[id] += 1
-    else:
-      counts[id] = 1
-  if counts:
-    ids = list(counts.keys())
-    ids.sort()
-    for id in ids:
-      text += "{}: {:4d} in {:.03f}s, {:.1f}/s    ".format(id, counts[id], dt, counts[id] / dt)
-  else:
-    text = "0 Messungen in {:.03f}s               ".format(dt)
+# setup output modules
+outputs = list(createModules(config['Outputs'], "output"))
 
-  if not counts or len(ids) < 3:
-    logging.warning(text)
+
+for source in inputs:
+  source.start()
+
+logger.info("started sources")
+
+def executeMiddleware(middleware, values):
+  submodules = getattr(middleware, 'submodules', [])
+  result = middleware.execute(values)
+  if not submodules:
+    return result
   else:
-    logging.info(text)
+    subResults = set()
+    for submodule in submodules:
+      tmp = executeMiddleware(submodule, result)
+      if tmp:
+        subResults.update(tmp)
+    return subResults
 
 while True:
-  values = []
-  for source in sources:
-    values.extend(source.read())
+  values = set()
+  for input in inputs:
+    values.update(input.read())
 
-  for sink in sinks:
-    sink.write(values)
+  results = set()
+  for middleware in middlewares:
+    tmp = executeMiddleware(middleware, values)
+    if tmp:
+      results.update(tmp)
+  else:
+    results = values
 
-  printStats(values)
+  for output in outputs:
+    output.write(results)
+    
   time.sleep(1.9)

box-pc/plc-connector/middlewares/print_stats.py

@@ -0,0 +1,32 @@
+import logging
+import time
+
+logger = logging.getLogger(__name__)
+
+class PrintStats:
+  def __init__(self, parent):
+    self.startTime = time.monotonic()
+
+  def execute(self, values):
+    counts = {}
+    dt = time.monotonic() - self.startTime
+    self.startTime = time.monotonic()
+    text = ""
+    for meas in values:
+      id = "{} {}".format(meas.series, meas.source)
+      if id in counts:
+        counts[id] += 1
+      else:
+        counts[id] = 1
+    if counts:
+      ids = list(counts.keys())
+      ids.sort()
+      for id in ids:
+        text += "{}: {:4d} in {:.03f}s, {:.1f}/s    ".format(id, counts[id], dt, counts[id] / dt)
+    else:
+      text = "0 Messungen in {:.03f}s               ".format(dt)
+
+    if not counts or len(ids) < 3:
+      logger.warning(text)
+    else:
+      logger.info(text)

box-pc/plc-connector/middlewares/time_correlation.py

@@ -0,0 +1,10 @@
+import logging
+
+logger = logging.getLogger(__name__)
+
+class TimeCorrelation:
+  def __init__(self, parent):
+    pass
+
+  def execute(self, values):
+    return values

box-pc/plc-connector/database/csvFile.py → box-pc/plc-connector/outputs/csv_file.py

@@ -5,6 +5,8 @@ import dataclasses
 import zipfile
 import logging
 
+logger = logging.getLogger(__name__)
+
 class CSVStorage:
   files = {}
 
@@ -19,7 +21,7 @@ class CSVStorage:
           self.files[meas.series] = CSVFile(self.path, meas.series, self.zipname)
         self.files[meas.series].write(meas)
     except Exception as ex:
-      logging.exception("CSV write failed")
+      logger.exception("CSV write failed")
 
 class CSVFile:
   file = None

box-pc/plc-connector/database/influxdb.py → box-pc/plc-connector/outputs/influxdb.py

@@ -4,11 +4,13 @@ from influxdb_client import InfluxDBClient, Point
 from influxdb_client.client.write_api import SYNCHRONOUS
 import dataclasses
 
+logger = logging.getLogger(__name__)
+
 class InfluxDB:
-  def __init__(self, url):
-    self.client = InfluxDBClient(url=url, token="XPBViJ3s4JL9_wPffwd5M2EgXj5hcUgT0n4jNhv7m6-NC-6SSxQ3run4kXtWBvOk-FYr1VG5Tj5WcoHgjge9jw==", org="laempe")
+  def __init__(self, url, token, org, bucket):
+    self.client = InfluxDBClient(url, token, org=org)
 
-    self.bucket = "energy-monitor"
+    self.bucket = bucket
 
     self.write_api = self.client.write_api(write_options=SYNCHRONOUS)
     self.query_api = self.client.query_api()
@@ -36,4 +38,4 @@ class InfluxDB:
     try:
       self.write_api.write(bucket=self.bucket, record=points)
     except Exception as ex:
-      logging.exception("Influx DB write failed")
+      logger.exception("Influx DB write failed")

box-pc/plc-connector/structures/measurement.py

@@ -5,12 +5,12 @@ from datetime import datetime
 class Measurement24v:
   timestamp: datetime
   source: str
-  current: tuple[float, ...]
-  status: tuple[bool, ...]
-  overload: tuple[bool, ...]
-  short_circuit: tuple[bool, ...]
-  limit: tuple[bool, ...]
-  pushbutton: tuple[bool, ...]
+  current: tuple # [float, ...]
+  status: tuple # [bool, ...]
+  overload: tuple # [bool, ...]
+  short_circuit: tuple # [bool, ...]
+  limit: tuple # [bool, ...]
+  pushbutton: tuple # [bool, ...]
   voltage: float
   series = "24v"
 
@@ -18,7 +18,7 @@ class Measurement24v:
 class Measurement480v:
   timestamp: datetime
   source: str
-  voltage: tuple[float, ...]
-  current: tuple[float, ...]
-  phase: tuple[float, ...]
+  voltage: tuple # [float, ...]
+  current: tuple # [float, ...]
+  phase: tuple # [float, ...]
   series = "480v"