WIP: Troubleshooting ModBus bit_sensor

homeassistant/components/modbus/__init__.py

@@ -37,6 +37,7 @@
 )
 import homeassistant.helpers.config_validation as cv
 
+from .bit_sensor import CONF_BIT_SENSORS
 from .const import (
     ATTR_ADDRESS,
     ATTR_HUB,
@@ -217,6 +218,16 @@ def number(value: Any) -> Union[int, float]:
     }
 )
 
+BIT_SENSOR_SCHEMA = BASE_COMPONENT_SCHEMA.extend(
+    {
+        vol.Required(CONF_ADDRESS): cv.positive_int,
+        vol.Optional(CONF_DEVICE_CLASS): BINARY_SENSOR_DEVICE_CLASSES_SCHEMA,
+        vol.Optional(CONF_INPUT_TYPE, default=CALL_TYPE_COIL): vol.In(
+            [CALL_TYPE_COIL, CALL_TYPE_DISCRETE]
+        ),
+    }
+)
+
 MODBUS_SCHEMA = vol.Schema(
     {
         vol.Optional(CONF_NAME, default=DEFAULT_HUB): cv.string,
@@ -225,6 +236,7 @@ def number(value: Any) -> Union[int, float]:
         vol.Optional(CONF_BINARY_SENSORS): vol.All(
             cv.ensure_list, [BINARY_SENSOR_SCHEMA]
         ),
+        vol.Optional(CONF_BIT_SENSORS): vol.All(cv.ensure_list, [BIT_SENSOR_SCHEMA]),
         vol.Optional(CONF_CLIMATES): vol.All(cv.ensure_list, [CLIMATE_SCHEMA]),
         vol.Optional(CONF_COVERS): vol.All(cv.ensure_list, [COVERS_SCHEMA]),
         vol.Optional(CONF_SENSORS): vol.All(cv.ensure_list, [SENSOR_SCHEMA]),

homeassistant/components/modbus/bit_sensor.py

@@ -0,0 +1,130 @@
+"""Support for Modbus Coil and Discrete Input sensors."""
+from __future__ import annotations
+
+from datetime import timedelta
+import logging
+
+from pymodbus.exceptions import ConnectionException, ModbusException
+from pymodbus.pdu import ExceptionResponse
+
+from homeassistant.components.binary_sensor import BinarySensorEntity
+from homeassistant.const import (
+    CONF_ADDRESS,
+    CONF_DEVICE_CLASS,
+    CONF_NAME,
+    CONF_SCAN_INTERVAL,
+    CONF_SLAVE,
+)
+from homeassistant.helpers.event import async_track_time_interval
+from homeassistant.helpers.typing import (
+    ConfigType,
+    DiscoveryInfoType,
+    HomeAssistantType,
+)
+
+from .const import CALL_TYPE_COIL, CONF_BIT_SENSORS, CONF_INPUT_TYPE, MODBUS_DOMAIN
+from .modbus import ModbusHub
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_platform(
+    hass: HomeAssistantType,
+    config: ConfigType,
+    async_add_entities,
+    discovery_info: DiscoveryInfoType | None = None,
+):
+    """Set up the Modbus bit sensors."""
+    sensors = []
+    _LOGGER.error("\t!! async_setup_platform called !!")
+    if discovery_info is None:
+        return
+
+    for entry in discovery_info[CONF_BIT_SENSORS]:
+        hub: ModbusHub = hass.data[MODBUS_DOMAIN][discovery_info[CONF_NAME]]
+        sensors.append(
+            ModbusBitSensor(
+                hub,
+                entry[CONF_NAME],
+                entry.get(CONF_SLAVE),
+                entry[CONF_ADDRESS],
+                entry.get(CONF_DEVICE_CLASS),
+                entry[CONF_INPUT_TYPE],
+                entry[CONF_SCAN_INTERVAL],
+            )
+        )
+
+    async_add_entities(sensors)
+
+
+class ModbusBitSensor(BinarySensorEntity):
+    """Modbus bit sensor."""
+
+    def __init__(
+        self, hub, name, slave, address, device_class, input_type, scan_interval
+    ):
+        """Initialize the Modbus binary sensor."""
+        self._hub = hub
+        self._name = name
+        self._slave = int(slave) if slave else None
+        self._address = int(address)
+        self._device_class = device_class
+        self._input_type = input_type
+        self._value = None
+        self._available = True
+        self._scan_interval = timedelta(seconds=scan_interval)
+
+    async def async_added_to_hass(self):
+        """Handle entity which will be added."""
+        async_track_time_interval(
+            self.hass, lambda arg: self._update(), self._scan_interval
+        )
+
+    @property
+    def name(self):
+        """Return the name of the sensor."""
+        return self._name
+
+    @property
+    def is_on(self):
+        """Return the state of the sensor."""
+        return self._value
+
+    @property
+    def device_class(self) -> str | None:
+        """Return the device class of the sensor."""
+        return self._device_class
+
+    @property
+    def should_poll(self):
+        """Return True if entity has to be polled for state.
+
+        False if entity pushes its state to HA.
+        """
+
+        # Handle polling directly in this entity
+        return False
+
+    @property
+    def available(self) -> bool:
+        """Return True if entity is available."""
+        return self._available
+
+    def _update(self):
+        """Update the state of the sensor."""
+        try:
+            if self._input_type == CALL_TYPE_COIL:
+                result = self._hub.read_coils(self._slave, self._address, 1)
+            else:
+                result = self._hub.read_discrete_inputs(self._slave, self._address, 1)
+        except ConnectionException:
+            self._available = False
+            return
+
+        if isinstance(result, (ModbusException, ExceptionResponse)):
+            self._available = False
+            return
+
+        self._value = result.bits[0] & 1
+        self._available = True
+        self.schedule_update_ha_state()

homeassistant/components/modbus/const.py

@@ -53,6 +53,10 @@
 CONF_BINARY_SENSORS = "binary_sensors"
 CONF_BINARY_SENSOR = "binary_sensor"
 
+# bit_sensor.py
+CONF_BIT_SENSORS = "bit_sensors"
+CONF_BIT_SENSOR = "bit_sensor"
+
 # sensor.py
 # CONF_DATA_TYPE = "data_type"
 DEFAULT_STRUCT_FORMAT = {

homeassistant/components/modbus/modbus.py

@@ -27,6 +27,8 @@
     CONF_BAUDRATE,
     CONF_BINARY_SENSOR,
     CONF_BINARY_SENSORS,
+    CONF_BIT_SENSOR,
+    CONF_BIT_SENSORS,
     CONF_BYTESIZE,
     CONF_CLIMATE,
     CONF_CLIMATES,
@@ -65,6 +67,7 @@ def modbus_setup(
             (CONF_BINARY_SENSOR, CONF_BINARY_SENSORS),
             (CONF_SENSOR, CONF_SENSORS),
             (CONF_SWITCH, CONF_SWITCHES),
+            (CONF_BIT_SENSOR, CONF_BIT_SENSORS),
         ):
             if conf_key in conf_hub:
                 load_platform(hass, component, DOMAIN, conf_hub, config)

tests/components/modbus/test_modbus_bit_sensor.py

@@ -0,0 +1,80 @@
+"""The tests for the Modbus sensor component."""
+import pytest
+
+from homeassistant.components.binary_sensor import DOMAIN as SENSOR_DOMAIN
+from homeassistant.components.modbus.const import (
+    CALL_TYPE_DISCRETE,
+    CONF_BIT_SENSORS,
+    CONF_INPUT_TYPE,
+    CONF_INPUTS,
+)
+from homeassistant.const import (
+    CONF_ADDRESS,
+    CONF_DEVICE_CLASS,
+    CONF_NAME,
+    CONF_SLAVE,
+    STATE_ON,
+)
+
+from .conftest import base_config_test, base_test
+
+
+@pytest.mark.parametrize(
+    "do_options",
+    [
+        {},
+        {
+            CONF_SLAVE: 10,
+            CONF_INPUT_TYPE: CALL_TYPE_DISCRETE,
+            CONF_DEVICE_CLASS: "door",
+        },
+    ],
+)
+async def test_config_binary_sensor(hass, do_options):
+    """Run test for binary sensor."""
+    sensor_name = "test_sensor"
+    config_sensor = {
+        CONF_NAME: sensor_name,
+        CONF_ADDRESS: 51,
+        **do_options,
+    }
+    await base_config_test(
+        hass,
+        config_sensor,
+        sensor_name,
+        SENSOR_DOMAIN,
+        CONF_BIT_SENSORS,
+        CONF_INPUTS,
+        method_discovery=True,
+    )
+
+
+@pytest.mark.parametrize(
+    "regs,expected",
+    [
+        (
+            [0xFF],
+            STATE_ON,
+        )
+    ],
+)
+async def test_all_bit_sensor(hass, regs, expected):
+    """Run test for given config."""
+    sensor_name = "modbus_test_bit_sensor"
+    state = await base_test(
+        hass,
+        {
+            CONF_NAME: sensor_name,
+            CONF_ADDRESS: 1234,
+            CONF_INPUT_TYPE: CALL_TYPE_DISCRETE,
+        },
+        sensor_name,
+        SENSOR_DOMAIN,
+        CONF_BIT_SENSORS,
+        CONF_INPUTS,
+        regs,
+        expected,
+        method_discovery=True,
+        scan_interval=5,
+    )
+    assert state == expected