modbus-utils/galion_modbus/poll.py

"""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):
    """Poll all configured PLCs for a site and return {tag: value}."""
    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


def write_coil(host, port=502, unit=1, address=0, value=True):
    """Write single coil — used for pump enable/disable commands."""
    c = ModbusTcpClient(host, port=port)
    c.connect()
    try:
        rr = c.write_coil(address, value, unit)
        return not rr.isError()
    finally:
        c.close()