openrvdas.logger.readers.modbus_reader

  1#!/usr/bin/env python3
  2
  3import logging
  4import time
  5import yaml
  6from threading import Lock
  7
  8# Optional dependency
  9try:
 10    from pyModbusTCP.client import ModbusClient
 11    MODBUS_MODULE_FOUND = True
 12except ModuleNotFoundError:
 13    MODBUS_MODULE_FOUND = False
 14
 15from logger.readers.reader import Reader  # noqa
 16
 17
 18################################################################################
 19class ModBusTCPReader(Reader):
 20    """
 21    Read data from Modbus TCP devices.
 22
 23    Supports:
 24      - single-slave (legacy) mode
 25      - multi-slave / multi-function scan files
 26      - holding/input registers, coils, discrete inputs
 27      - text or raw-bytes output
 28      - partial failure tolerance
 29      - non-fatal connection failures
 30    """
 31
 32    _FUNCTION_MAP = {
 33        "holding_registers": "read_holding_registers",
 34        "input_registers": "read_input_registers",
 35        "coils": "read_coils",
 36        "discrete_inputs": "read_discrete_inputs",
 37    }
 38
 39    ############################
 40    def __init__(self,
 41                 registers=None,
 42                 host="localhost",
 43                 port=502,
 44                 scan_file=None,
 45                 slave=1,
 46                 function="holding_registers",
 47                 interval=10,
 48                 sep=" ",
 49                 encoding="utf-8",
 50                 encoding_errors="ignore",
 51                 **kwargs):
 52
 53        super().__init__(encoding=encoding, encoding_errors=encoding_errors, **kwargs)
 54
 55        if not MODBUS_MODULE_FOUND:
 56            raise RuntimeError(
 57                "Modbus functionality not available. Please install Python module pyModbusTCP."
 58            )
 59
 60        self.sep = sep if encoding is not None else None
 61        self.interval = interval
 62        self._lock = Lock()
 63        self._next_read_time = 0.0
 64
 65        self.client = ModbusClient(
 66            host=host,
 67            port=port,
 68            auto_open=True,
 69            auto_close=True,
 70        )
 71
 72        self.polls = []
 73
 74        if scan_file:
 75            self._load_scan_file(scan_file)
 76        else:
 77            # Legacy single-poll behavior
 78            if registers is None:
 79                raise ValueError("registers required when scan_file not provided")
 80
 81            self.polls.append({
 82                "slave": slave,
 83                "function": function,
 84                "registers": self._parse_registers(registers)
 85            })
 86
 87    ############################
 88    def _parse_register_spec(self, spec):
 89        """Parse a single register spec into (start, count) tuple."""
 90        if ":" in spec:
 91            start, end = spec.split(":", 1)
 92            if end == "":
 93                raise ValueError(f"Ambiguous register range: '{spec}'")
 94
 95            start = int(start) if start else 0
 96            end = int(end)
 97
 98            if start < 0 or end < start:
 99                raise ValueError(f"Invalid register range: '{spec}'")
100
101            return (start, end - start + 1)
102
103        addr = int(spec)
104        if addr < 0:
105            raise ValueError(f"Invalid register address: '{spec}'")
106
107        return (addr, 1)
108
109    ############################
110    def _validate_blocks(self, blocks):
111        validated = []
112        for block in blocks:
113            if not (
114                isinstance(block, tuple)
115                and len(block) == 2
116                and all(isinstance(x, int) and x >= 0 for x in block)
117            ):
118                raise ValueError(f"Invalid register block: {block}")
119            validated.append(block)
120        return validated
121
122    ############################
123    def _parse_registers(self, registers):
124        """Parse register specification into [(start, count), ...] blocks."""
125        if isinstance(registers, list):
126            return self._validate_blocks(registers)
127
128        if not isinstance(registers, str):
129            raise TypeError("registers must be a string or list")
130
131        blocks = []
132        for spec in registers.split(","):
133            spec = spec.strip()
134            if spec:
135                blocks.append(self._parse_register_spec(spec))
136
137        if not blocks:
138            raise ValueError("No valid registers specified")
139
140        return blocks
141
142    ############################
143    def _load_scan_file(self, path):
144        with open(path, "r") as f:
145            cfg = yaml.safe_load(f)
146
147        if "polls" not in cfg:
148            raise ValueError("scan_file must contain 'polls' list")
149
150        for poll in cfg["polls"]:
151            self.polls.append({
152                "slave": poll["slave"],
153                "function": poll.get("function", "holding_registers"),
154                "registers": self._parse_registers(poll["registers"]),
155            })
156
157    ############################
158    def _format_record(self, readings):
159        """
160        Flatten nested register/coil values into a single record.
161
162        - encoding=None:
163            - registers -> 16-bit unsigned big-endian
164            - coils -> packed bits
165        - otherwise:
166            - text with sep, 'nan' placeholders
167        """
168        if readings is None:
169            return None
170
171        flat_values = [val for block in readings if block for val in block]
172
173        if self.encoding is None:
174            if not flat_values:
175                return b""
176
177            first_val = next((v for v in flat_values if v is not None), 0)
178
179            # Coils / discrete inputs
180            if isinstance(first_val, (bool, int)) and all(
181                v in (0, 1, True, False, None) for v in flat_values
182            ):
183                byte_vals = []
184                current = 0
185                for i, bit in enumerate(flat_values):
186                    current |= (1 if bit else 0) << (i % 8)
187                    if i % 8 == 7:
188                        byte_vals.append(current)
189                        current = 0
190                if len(flat_values) % 8:
191                    byte_vals.append(current)
192                return bytes(byte_vals)
193
194            # Registers
195            return b"".join(
196                int(val or 0).to_bytes(2, byteorder="big", signed=False)
197                for val in flat_values
198            )
199
200        # Text mode
201        text_values = [str(val) if val is not None else "nan" for val in flat_values]
202        return self._encode_str(self.sep.join(text_values))
203
204    ############################
205    def read(self):
206        """
207        Read all configured polls.
208
209        Returns:
210          - list[str|bytes|None] for each poll
211          - None in place of any poll that cannot be read
212        """
213        with self._lock:
214            now = time.monotonic()
215            if now < self._next_read_time:
216                time.sleep(self._next_read_time - now)
217
218            start_time = time.monotonic()
219            results = []
220
221            for poll in self.polls:
222                slave = poll["slave"]
223                func_name = self._FUNCTION_MAP.get(poll["function"])
224
225                if not func_name:
226                    logging.warning(
227                        "Invalid function for slave=%s: %s",
228                        slave, poll["function"]
229                    )
230                    total = sum(c for _, c in poll["registers"])
231                    results.append([[None] * total])
232                    continue
233
234                func = getattr(self.client, func_name)
235                poll_readings = []
236
237                try:
238                    for start, count in poll["registers"]:
239                        block = func(start, count, unit_id=slave)
240                        if block is None:
241                            logging.warning(
242                                "Failed to read slave=%d addr=%d:%d",
243                                slave, start, count
244                            )
245                            poll_readings.append([None] * count)
246                        else:
247                            poll_readings.append(block)
248
249                    results.append(poll_readings)
250
251                except (OSError, AttributeError, ValueError, ConnectionError) as exc:
252                    logging.warning(
253                        "Modbus TCP connection error for slave=%d: %s", slave, exc
254                    )
255                    results.append(None)  # Entire poll failed, not fatal
256
257            self._next_read_time = start_time + self.interval
258
259            # Format records, preserving None for failed polls
260            formatted = [self._format_record(r) if r is not None else None for r in results]
261
262            # Prepend slave info for text mode
263            if self.encoding is not None:
264                formatted = [
265                    self._encode_str(f"slave {poll['slave']}:{self.sep}{line.decode(self.encoding) if isinstance(line, bytes) else line}")
266                    if line is not None else None
267                    for poll, line in zip(self.polls, formatted)
268                ]
269
270            return formatted