"""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