initial modbus polling utils

README.md

@@ -0,0 +1,2 @@
+# modbus-utils
+Internal Modbus TCP polling scripts for Galion Systems client sites.

config/sites.yaml.example

@@ -0,0 +1,17 @@
+sites:
+  walnut_creek:
+    plcs:
+      - host: 10.10.1.5
+        unit: 1
+        start: 0
+        count: 20
+        tags:
+          "0": flow_rate_gpm
+          "1": tank_level_ft
+          "2": pump1_status
+  gary_in:
+    plcs:
+      - host: 10.20.1.10
+        unit: 2
+        start: 100
+        count: 10

galion_modbus/__init__.py

@@ -0,0 +1 @@
+"""Galion Systems Modbus polling utilities."""

galion_modbus/poll.py

@@ -0,0 +1,32 @@
+"""Simple Modbus TCP poller for client site PLCs."""
+from pymodbus.client import ModbusTcpClient
+import logging
+
+log = logging.getLogger(__name__)
+
+
+def read_holding_registers(host, port=502, unit=1, address=0, count=10):
+    c = ModbusTcpClient(host, port=port)
+    c.connect()
+    try:
+        rr = c.read_holding_registers(address, count, unit)
+        if rr.isError():
+            log.warning("read error from %s unit=%d", host, unit)
+            return None
+        return rr.registers
+    finally:
+        c.close()
+
+
+def poll_site(site_cfg):
+    results = {}
+    for plc in site_cfg.get("plcs", []):
+        regs = read_holding_registers(
+            plc["host"], plc.get("port", 502),
+            plc.get("unit", 1), plc.get("start", 0), plc.get("count", 20)
+        )
+        if regs is not None:
+            for i, val in enumerate(regs):
+                tag = plc.get("tags", {}).get(str(i), f"{plc['host']}:r{i}")
+                results[tag] = val
+    return results

requirements.txt

@@ -0,0 +1,2 @@
+pymodbus>=3.5.0
+pyyaml>=6.0