1#!/usr/bin/env python3
  2"""
  3'''
  4ModBusSerialReader
  5==================
  6
  7This module provides a thread-safe reader for polling Modbus RTU devices
  8over a serial connection. It supports single-slave and multi-slave setups,
  9multiple Modbus function types, and flexible output formats (binary or text).
 10
 11Key Features
 12------------
 13- Reads holding registers, input registers, coils, and discrete inputs.
 14- Supports configuration via direct register specification or YAML scan file.
 15- Partial failure tolerance: failed polls return placeholder values without
 16  stopping other reads.
 17- Exponential backoff for serial connection retries (1s → 30s max).
 18- Thread-safe access via internal Lock.
 19- Output can be raw bytes (binary) or encoded text with customizable separator.
 20- Automatic cleanup of the serial port on `stop()` or object deletion.
 21
 22Typical Usage
 23-------------
 24    reader = ModBusSerialReader(
 25        registers="0:5,10:15",
 26        port="/dev/ttyUSB0",
 27        baudrate=19200,
 28        interval=5,
 29        encoding="utf-8"
 30    )
 31    data = reader.read()
 32    print(data)
 33    reader.stop()
 34
 35Dependencies
 36------------
 37- pymodbus (optional; raises RuntimeError if missing)
 38- PyYAML
 39- Standard Python libraries: logging, time, threading, sys, os.path
 40'''
 41"""
 42
 43import logging
 44import yaml
 45import time
 46from threading import Lock
 47
 48try:
 49    from pymodbus.client import ModbusSerialClient
 50    from pymodbus import ModbusException
 51    MODBUS_MODULE_FOUND = True
 52except ModuleNotFoundError:
 53    MODBUS_MODULE_FOUND = False
 54
 55from logger.readers.reader import Reader  # noqa
 56
 57
 58###############################################################################
 59class ModBusSerialReader(Reader):
 60    """
 61    Read data from Modbus RTU devices over a serial connection.
 62    """
 63
 64    ############################
 65    def __init__(self,
 66                 registers=None,
 67                 port="/dev/ttyUSB0",
 68                 baudrate=9600,
 69                 parity="N",
 70                 stopbits=1,
 71                 bytesize=8,
 72                 scan_file=None,
 73                 slave=1,
 74                 function='holding_registers',
 75                 interval=1.0,
 76                 sep=" ",
 77                 encoding="utf-8",
 78                 encoding_errors="ignore",
 79                 timeout=None,
 80                 **kwargs):
 81        """
 82        ```
 83        registers - Comma-separated string (e.g. '0:5,10:15') or list of
 84                    tuples [(start,count), ...] specifying which registers
 85                    to poll. Required if scan_file is not provided.
 86
 87        port - Serial port device (e.g. '/dev/ttyUSB0').
 88
 89        baudrate - Serial connection baud rate (default 9600).
 90
 91        parity - Serial parity, one of 'N', 'E', 'O' (default 'N').
 92
 93        stopbits - Number of stop bits (default 1).
 94
 95        bytesize - Number of data bits (default 8).
 96
 97        scan_file - Optional YAML file specifying multiple slaves, functions,
 98                    and register blocks. Overrides registers/slave/function
 99                    parameters if provided.
100
101        slave - Modbus slave ID (default 1). Ignored if scan_file is provided.
102
103        function - Modbus function type to read: 'holding_registers',
104                   'input_registers', 'coils', or 'discrete_inputs'.
105                   Ignored if scan_file is provided.
106
107        interval - Seconds between consecutive reads. Must be >= 0.1.
108
109        sep - Separator string used when encoding output as text. Ignored
110              if encoding is None.
111
112        encoding - Character encoding for text output (default 'utf-8').
113                   If None, raw bytes are returned.
114
115        encoding_errors - Strategy for handling encoding errors ('ignore'
116                          by default). Other options: 'strict', 'replace',
117                          'backslashreplace'.
118
119        timeout - Max time in seconds to wait for serial response. Defaults
120                  to 2 seconds if None. Minimum timeout is 1s.
121        ```
122        """
123        super().__init__(encoding=encoding, encoding_errors=encoding_errors, **kwargs)
124
125        if not MODBUS_MODULE_FOUND:
126            raise RuntimeError(
127                "Modbus functionality not available. Install pymodbus: "
128                'pip install "pymodbus[serial]"'
129            )
130
131        self._FUNCTION_MAP = {
132            "holding_registers": "read_holding_registers",
133            "input_registers": "read_input_registers",
134            "coils": "read_coils",
135            "discrete_inputs": "read_discrete_inputs",
136        }
137
138        if interval < 0.1:
139            raise ValueError('Interval must be greater or equal to 0.1 seconds (10Hz)')
140
141        timeout = timeout or 2.0
142        if timeout < 1.0:
143            raise ValueError('Timeout must be greater or equal to 1 seconds')
144
145        self.polls = []
146        if scan_file:
147            self._load_scan_file(scan_file)
148        else:
149            if registers is None:
150                raise ValueError("registers required when scan_file not provided")
151            self.polls = [{
152                "slave": slave,
153                "function": function,
154                "registers": self._parse_registers(registers)
155            }]
156
157        self.sep = sep if encoding is not None else None
158        self.interval = interval
159        self._next_read_time = 0.0
160        self._connected = False
161        self._reconnect_delay = timeout
162        self._reconnect_delay_max = 30.0
163        self._next_connect_time = 0.0
164        self._lock = Lock()
165
166        self.client = ModbusSerialClient(
167            port=port,
168            baudrate=baudrate,
169            parity=parity,
170            stopbits=stopbits,
171            bytesize=bytesize,
172            timeout=timeout,
173        )
174
175    ############################
176    def _reset_client(self):
177        """Hard-reset of the serial client."""
178        try:
179            self.client.close()
180        except Exception:
181            pass
182
183        self._connected = False
184
185        now = time.monotonic()
186        self._next_connect_time = now + self._reconnect_delay
187        self._reconnect_delay = min(
188            self._reconnect_delay * 2,
189            self._reconnect_delay_max
190        )
191        logging.info(
192            "Next Modbus reconnect attempt in %.1f seconds",
193            self._reconnect_delay
194        )
195
196    ############################
197    def _parse_register_spec(self, spec):
198        """Parse a single register spec into (start, count)."""
199        if ":" in spec:
200            start, end = spec.split(":", 1)
201            if end == "":
202                raise ValueError(f"Ambiguous register range: '{spec}'")
203            start = int(start) if start else 0
204            end = int(end)
205            if start < 0 or end < start:
206                raise ValueError(f"Invalid register range: '{spec}'")
207            return (start, end - start + 1)
208        addr = int(spec)
209        if addr < 0:
210            raise ValueError(f"Invalid register address: '{spec}'")
211        return (addr, 1)
212
213    ############################
214    def _validate_blocks(self, blocks):
215        """Validate and individual register block"""
216        validated = []
217        for block in blocks:
218            if not (
219                isinstance(block, tuple)
220                and len(block) == 2
221                and all(isinstance(x, int) and x >= 0 for x in block)
222            ):
223                raise ValueError(f"Invalid register block: {block}")
224            validated.append(block)
225        return validated
226
227    ############################
228    def _parse_registers(self, registers):
229        """Parse comma-separated string or list into [(start, count), ...] blocks."""
230        if registers is None:
231            return [(0, 10)]
232        if isinstance(registers, list):
233            return self._validate_blocks(registers)
234        if not isinstance(registers, str):
235            raise TypeError("registers must be str, list, or None")
236
237        blocks = []
238        for spec in registers.split(","):
239            spec = spec.strip()
240            if spec:
241                blocks.append(self._parse_register_spec(spec))
242        if not blocks:
243            raise ValueError("No valid registers specified")
244        return blocks
245
246    ############################
247    def _client_connected(self) -> bool:
248        """Connect if not already connected. Returns True if OK."""
249        if self._connected:
250            return True
251
252        now = time.monotonic()
253        if now < self._next_connect_time:
254            return False
255
256        try:
257            self._connected = self.client.connect()
258        except ModbusException as exc:
259            logging.error(f"ModbusException: connect() failed: {exc}")
260            self._reset_client()
261            return False
262        except Exception as exc:
263            logging.error(f"Exception: connect() failed: {exc}")
264            self._reset_client()
265            return False
266
267        if not self._connected:
268            logging.error("ModBusSerialReader: unable to open serial port")
269            self._reset_client()
270            return False
271
272        self._reconnect_delay = 1.0
273        self._next_connect_time = 0.0
274
275        return self._connected
276
277    ############################
278    def _load_scan_file(self, path):
279        """Load the YAML-formatted scan file"""
280        with open(path, "r") as f:
281            cfg = yaml.safe_load(f)
282
283        if "polls" not in cfg:
284            raise ValueError("scan_file must contain 'polls' list")
285
286        for poll in cfg["polls"]:
287            slave = poll["slave"]
288            function = poll.get("function", "holding_registers")
289            registers = poll["registers"]
290
291            self.polls.append({
292                "slave": slave,
293                "function": function,
294                "registers": self._parse_registers(registers)
295            })
296
297    ############################
298    def _format_record(self, readings: list[list[int | bool | None]], function: str) -> bytes | str:
299        """
300        Flatten nested register/coil values into a single record.
301
302        Parameters
303        ----------
304        readings : list[list[int | bool | None]]
305            Nested lists of values returned from Modbus reads.
306
307        function : str
308            Modbus function type. Determines formatting behavior.
309            Expected values:
310              - "holding_registers"
311              - "input_registers"
312              - "coils"
313              - "discrete_inputs"
314
315        Behavior
316        --------
317        - encoding is None:
318            - Registers → 16-bit unsigned big-endian
319            - Coils/discrete_inputs → packed bits (LSB-first)
320            - None values → 0
321        - Otherwise convert to text and encode with _encode_str, using self.sep
322        """
323
324        flat_values: list[int | bool | None] = []
325        for block in readings:
326            if block:
327                flat_values.extend(block)
328
329        is_coils = function in ("coils", "discrete_inputs")
330
331        # ------------------------------------------------------------------
332        # Binary
333        # ------------------------------------------------------------------
334        if self.encoding is None:
335            if not flat_values:
336                return b""
337
338            if is_coils:
339                # Pack bits LSB-first, per Modbus spec
340                byte_vals = []
341                current_byte = 0
342                bit_index = 0
343
344                for bit in flat_values:
345                    bit_val = 1 if bit else 0
346                    current_byte |= bit_val << bit_index
347                    bit_index += 1
348
349                    if bit_index == 8:
350                        byte_vals.append(current_byte)
351                        current_byte = 0
352                        bit_index = 0
353
354                if bit_index:
355                    byte_vals.append(current_byte)
356
357                return bytes(byte_vals)
358
359            # Registers: 16-bit unsigned big-endian
360            return b"".join(
361                int(val or 0).to_bytes(2, byteorder="big", signed=False)
362                for val in flat_values
363            )
364
365        # ------------------------------------------------------------------
366        # Encoded text
367        # ------------------------------------------------------------------
368        text_values = [
369            str(val) if val is not None else "nan"
370            for val in flat_values
371        ]
372        return self._encode_str(self.sep.join(text_values))
373
374
375    ############################
376    def read(self):
377        """Read all configured polls and return a list of formatted records per slave.
378
379        Returns:
380            list[bytes|str|None]: None for polls that cannot be read; otherwise formatted output.
381        """
382        with self._lock:
383            now = time.monotonic()
384            if now < self._next_read_time:
385                time.sleep(self._next_read_time - now)
386            start_time = time.monotonic()
387
388            if not self._client_connected():
389                return None
390
391            records = []
392
393            for poll in self.polls:
394                slave = poll["slave"]
395                func_name = self._FUNCTION_MAP.get(poll["function"])
396
397                if not func_name:
398                    logging.warning("Invalid function for slave=%s: %s", slave, poll["function"])
399                    total_count = sum(count for _, count in poll["registers"])
400                    readings = [[None] * total_count]
401                else:
402                    func = getattr(self.client, func_name)
403                    readings = []
404                    for start, count in poll["registers"]:
405                        try:
406                            resp = func(address=start, count=count, unit=slave)
407                            if resp is None or getattr(resp, "isError", lambda: False)():
408                                logging.warning("Failed to read slave=%d addr=%d:%d", slave, start, count)
409                                readings.append([None] * count)
410                            else:
411                                values = getattr(resp, "registers", getattr(resp, "bits", None))
412                                readings.append(values or [None] * count)
413                        except (ModbusException, OSError, ConnectionError, ValueError, AttributeError) as exc:
414                            logging.warning("Modbus read error slave=%d addr=%d:%d: %s", slave, start, count, exc)
415                            readings.append([None] * count)
416                            self._reset_client()
417                            break
418
419                record = self._format_record(readings, poll["function"])
420                if self.encoding is not None and isinstance(record, bytes):
421                    record = record.decode(self.encoding, errors=self.encoding_errors)
422                    record = self._encode_str(f"slave {slave}:{self.sep}{record}")
423                records.append(record)
424
425            self._next_read_time = start_time + self.interval
426            return records
427
428    ############################
429    def stop(self):
430        """Explicitly close the serial port."""
431        with self._lock:
432            try:
433                self.client.close()
434            except Exception:
435                pass
436            self._connected = False
437
438    ############################
439    def __del__(self):
440        """Destructor — ensures port is closed safely."""
441        try:
442            self.stop()
443        except Exception:
444            pass