A simple driver class for Waveshare's Modbus RTU Relay boards. Uses a reliable minimalmodbus library for low-level communication handling.

waveshare_modbus_rtu_relay.py

import logging
import time

from minimalmodbus import Instrument


logger = logging.getLogger(__name__)


class ModbusRTURelay:
    """
    A simple driver for Waveshare Modbus RTU relay boards.

    The driver communicates with the board using the MinimalModbus library
    and supports turning individual or all relay channels on and off, as well
    as toggling individual channels.

    During initialization, the serial port is opened and communication with
    the board is verified by reading the state of the first relay channel.
    If communication cannot be established, the connection attempt is retried
    according to the configured retry settings.

    Channel numbers exposed by this class start at 1, while Modbus addresses
    used by the board start at 0.

    The Waveshare-specific toggle operation uses MinimalModbus's private
    `_perform_command()` method because the non-standard value 0x5500 is not
    supported by the public `write_bit()` API.

    Example:
        with ModbusRTURelay("/dev/ttyUSB0", 1) as relay:
            relay.turn_on_channel(1)
            relay.toggle_channel(2)
            relay.turn_off_all_channels()

    References:
        https://www.waveshare.com/wiki/Modbus_RTU_Relay
        https://minimalmodbus.readthedocs.io/
    """
    FUNCTION_WRITE_SINGLE_COIL = 0x05

    VALUE_ON = 0xFF00
    VALUE_OFF = 0x0000
    VALUE_TOGGLE = 0x5500

    ALL_CHANNELS_ADDRESS = 0x00FF

    def __init__(
        self,
        port: str,
        slave_address: int,
        channels_count: int = 8,
        retries_count: int = 10,
        retry_delay: float = 1.0,
        timeout: float = 0.2,
    ) -> None:
        if channels_count < 1:
            raise ValueError("channels_count must be positive")

        self.channels_count = channels_count
        self.instrument = self._connect(
            port=port,
            slave_address=slave_address,
            retries_count=retries_count,
            retry_delay=retry_delay,
            timeout=timeout,
        )

    @staticmethod
    def _connect(
        port: str,
        slave_address: int,
        retries_count: int,
        retry_delay: float,
        timeout: float,
    ) -> Instrument:
        """
        Opens and configures the serial port, then verifies communication with
        the relay board by reading the first channel.

        Communication is retried up to `retries_count` times, with `retry_delay`
        seconds between attempts. Raises the last communication error if all
        attempts fail.
        """
        for attempt in range(1, retries_count + 1):
            try:
                instrument = Instrument(
                    port=port,
                    slaveaddress=slave_address,
                )

                instrument.serial.baudrate = 9600
                instrument.serial.timeout = timeout

                # Faktycznie sprawdza komunikację z urządzeniem.
                instrument.read_bit(0, functioncode=1)

                logger.debug(
                    "Connected to Modbus RTU relay on %s, address %d",
                    port,
                    slave_address,
                )
                return instrument

            except OSError as error:
                if attempt == retries_count:
                    raise

                logger.warning(
                    "Failed to communicate with Modbus relay "
                    "(attempt %d/%d): %s",
                    attempt,
                    retries_count,
                    error,
                )
                time.sleep(retry_delay)

        raise RuntimeError("Unreachable code")

    def _validate_channel(self, channel: int) -> None:
        if isinstance(channel, bool) or not isinstance(channel, int):
            raise TypeError("channel must be an integer")

        if not 1 <= channel <= self.channels_count:
            raise ValueError(
                f"channel must be between 1 and {self.channels_count}"
            )

    def _send_vendor_command(self, address: int, value: int) -> bytes:
        payload = address.to_bytes(2, byteorder="big")
        payload += value.to_bytes(2, byteorder="big")

        # Private MinimalModbus API required for Waveshare's 0x5500 toggle.
        return self.instrument._perform_command(
            self.FUNCTION_WRITE_SINGLE_COIL,
            payload,
        )

    def turn_on_channel(self, channel: int) -> None:
        self._validate_channel(channel)
        self.instrument.write_bit(channel - 1, 1, functioncode=5)

    def turn_off_channel(self, channel: int) -> None:
        self._validate_channel(channel)
        self.instrument.write_bit(channel - 1, 0, functioncode=5)

    def toggle_channel(self, channel: int) -> bytes:
        self._validate_channel(channel)
        return self._send_vendor_command(
            channel - 1,
            self.VALUE_TOGGLE,
        )

    def turn_on_all_channels(self) -> None:
        self.instrument.write_bit(
            self.ALL_CHANNELS_ADDRESS,
            1,
            functioncode=5,
        )

    def turn_off_all_channels(self) -> None:
        self.instrument.write_bit(
            self.ALL_CHANNELS_ADDRESS,
            0,
            functioncode=5,
        )

    def close(self) -> None:
        if self.instrument.serial.is_open:
            self.instrument.serial.close()

    def __enter__(self) -> "ModbusRTURelay":
        return self

    def __exit__(self, exc_type, exc_value, traceback) -> None:
        self.close()