Bump mypy from 1.16.1 to 1.17.0 (#376)

Bumps [mypy](https://github.com/python/mypy) from 1.16.1 to 1.17.0.
- [Changelog](https://github.com/python/mypy/blob/master/CHANGELOG.md)
- [Commits](https://github.com/python/mypy/compare/v1.16.1...v1.17.0)

---
updated-dependencies:
- dependency-name: mypy
  dependency-version: 1.17.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

.github/workflows/stale.yaml

@@ -14,7 +14,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: 🚀 Run stale
-        uses: actions/stale@v9.0.0
+        uses: actions/stale@v9.1.0
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
           days-before-stale: 14

.github/workflows/test.yml

@@ -23,7 +23,7 @@ jobs:
         with:
           fetch-depth: 2
       - name: Set up Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v5.3.0
+        uses: actions/setup-python@v5.6.0
         with:
           python-version: ${{ matrix.python-version }}
       - name: Install dependencies

README.md

@@ -210,4 +210,4 @@ switch:
 ```
 
 # Reporting issues
-*Before* reporting issues please enable debug logging as described [here](https://www.home-assistant.io/docs/configuration/troubleshooting/#enabling-debug-logging), check logs and report issue attaching the log file.
+*Before* reporting issues please enable debug logging as described [here](https://www.home-assistant.io/docs/configuration/troubleshooting/#enabling-debug-logging), check logs and report issue attaching the log file and the relevant YAML section.

custom_components/rpi_gpio/binary_sensor.py

@@ -12,6 +12,7 @@ from homeassistant.helpers.config_validation import PLATFORM_SCHEMA
 from homeassistant.components.binary_sensor import BinarySensorEntity
 from homeassistant.const import CONF_SENSORS, CONF_NAME, CONF_PORT, CONF_UNIQUE_ID
 from .hub import BIAS
+
 CONF_INVERT_LOGIC = "invert_logic"
 DEFAULT_INVERT_LOGIC = False
 CONF_BOUNCETIME = "bouncetime"
@@ -51,17 +52,20 @@ async def async_setup_platform(
 
     sensors = []
     for sensor in config.get(CONF_SENSORS):
-        sensors.append(
+        try:
-            GPIODBinarySensor(
+            sensors.append(
-                hub,
+                GPIODBinarySensor(
-                sensor[CONF_NAME],
+                    hub,
-                sensor[CONF_PORT],
+                    sensor[CONF_NAME],
-                sensor.get(CONF_UNIQUE_ID) or f"{DOMAIN}_{sensor[CONF_PORT]}_{sensor[CONF_NAME].lower().replace(' ', '_')}",
+                    sensor[CONF_PORT],
-                sensor.get(CONF_INVERT_LOGIC),
+                    sensor.get(CONF_UNIQUE_ID) or f"{DOMAIN}_{sensor[CONF_PORT]}_{sensor[CONF_NAME].lower().replace(' ', '_')}",
-                sensor.get(CONF_PULL_MODE),
+                    sensor.get(CONF_INVERT_LOGIC),
-                sensor.get(CONF_BOUNCETIME)
+                    sensor.get(CONF_PULL_MODE),
+                    sensor.get(CONF_BOUNCETIME)
+                )
             )
-        )
+        except Exception as e:
+            _LOGGER.error(f"Failed to add binary sensor {sensor[CONF_NAME]} for port {sensor[CONF_PORT]}: {e}")
 
     async_add_entities(sensors)
 
@@ -78,12 +82,22 @@ class GPIODBinarySensor(BinarySensorEntity):
         self._active_low = active_low
         self._bias = bias
         self._debounce = debounce
+        self._line, current_is_on = self._hub.add_sensor(self._port, self._active_low, self._bias, self._debounce)
+        self._attr_is_on = current_is_on        
 
     async def async_added_to_hass(self) -> None:
         await super().async_added_to_hass()
-        self._hub.add_sensor(self, self._port, self._active_low, self._bias, self._debounce)
+        _LOGGER.debug(f"GPIODBinarySensor async_added_to_hass: Adding fd:{self._line.fd}")
-        self.async_write_ha_state()
+        self._hub._hass.loop.add_reader(self._line.fd, self.handle_event)
+
+    async def async_will_remove_from_hass(self) -> None:
+        await super().async_will_remove_from_hass()
+        _LOGGER.debug(f"GPIODBinarySensor async_will_remove_from_hass: Removing fd:{self._line.fd}")
+        self._hub._hass.loop.remove_reader(self._line.fd)
+        self._line.release()
 
     def handle_event(self):
-        self._attr_is_on = self._hub.get_line_value(self._port)
+        for event in self._line.read_edge_events():
+            self._attr_is_on = True if event.event_type is event.Type.RISING_EDGE else False
+            _LOGGER.debug(f"Event: {event}. New line value: {self._attr_is_on}")
         self.schedule_update_ha_state(False)

custom_components/rpi_gpio/cover.py

@@ -17,6 +17,7 @@ from homeassistant.const import CONF_COVERS, CONF_NAME, CONF_UNIQUE_ID
 from .hub import BIAS, DRIVE
 import homeassistant.helpers.config_validation as cv
 import voluptuous as vol
+import asyncio
 
 CONF_RELAY_PIN = "relay_pin"
 CONF_RELAY_TIME = "relay_time"
@@ -70,21 +71,24 @@ async def async_setup_platform(
     invert_relay = config[CONF_INVERT_RELAY]
     covers = []
     for cover in config.get(CONF_COVERS):
-        covers.append(
+        try:
-            GPIODCover(
+            covers.append(
-                hub,
+                GPIODCover(
-                cover[CONF_NAME],
+                    hub,
-                cover.get(CONF_RELAY_PIN),
+                    cover[CONF_NAME],
-                relay_time,
+                    cover.get(CONF_RELAY_PIN),
-                invert_relay,
+                    relay_time,
-                "AS_IS",
+                    invert_relay,
-                "PUSH_PULL",
+                    "AS_IS",
-                cover.get(CONF_STATE_PIN),
+                    "PUSH_PULL",
-                state_pull_mode,
+                    cover.get(CONF_STATE_PIN),
-                invert_state,
+                    state_pull_mode,
-                cover.get(CONF_UNIQUE_ID) or f"{DOMAIN}_{cover.get(CONF_RELAY_PIN)}_{cover[CONF_NAME].lower().replace(' ', '_')}",
+                    invert_state,
+                    cover.get(CONF_UNIQUE_ID) or f"{DOMAIN}_{cover.get(CONF_RELAY_PIN)}_{cover[CONF_NAME].lower().replace(' ', '_')}",
+                )
             )
-        )
+        except Exception as e:
+            _LOGGER.error(f"Failed to add cover {cover[CONF_NAME]} for port {cover.get(CONF_RELAY_PIN)}:{cover.get(CONF_STATE_PIN)}: {e}")
 
     async_add_entities(covers)
 
@@ -105,51 +109,66 @@ class GPIODCover(CoverEntity):
         self._state_port = state_port
         self._state_bias = state_bias
         self._state_active_low = state_active_low
-        self._attr_is_closed = False != state_active_low
+        self._relay_line, self._state_line, current_is_on = self._hub.add_cover(
+            self._relay_port, self._relay_active_low, self._relay_bias, self._relay_drive,
+            self._state_port, self._state_bias, self._state_active_low)        
+        self._attr_is_closed = current_is_on
+        self.is_on = current_is_on
 
     async def async_added_to_hass(self) -> None:
         await super().async_added_to_hass()
-        self._hub.add_cover(self, self._relay_port, self._relay_active_low, self._relay_bias, 
+        _LOGGER.debug(f"GPIODCover async_added_to_hass: Adding fd:{self._state_line.fd}")
-                            self._relay_drive, self._state_port, self._state_bias, self._state_active_low)
+        self._hub._hass.loop.add_reader(self._state_line.fd, self.handle_event)
-        self.async_write_ha_state()
+
+    async def async_will_remove_from_hass(self) -> None:
+        await super().async_will_remove_from_hass()
+        _LOGGER.debug(f"GPIODCover async_will_remove_from_hass: Removing fd:{self._state_line.fd}")
+        self._hub._hass.loop.remove_reader(self._state_line.fd)
+        self._relay_line.release()
+        self._state_line.release()
 
     def handle_event(self):
-        self._attr_is_closed = self._hub.get_line_value(self._state_port)
+        for event in self._state_line.read_edge_events():
+            self._attr_is_closed = True if event.event_type is event.Type.RISING_EDGE else False
+            _LOGGER.debug(f"Event: {event}. New _attr_is_closed value: {self._attr_is_closed}")
         self.schedule_update_ha_state(False)
 
-    def close_cover(self, **kwargs):
+    async def async_close_cover(self, **kwargs):
+        _LOGGER.debug(f"GPIODCover async_close_cover: is_closed: {self.is_closed}. is_closing: {self.is_closing}, is_opening: {self.is_opening}")
         if self.is_closed:
             return
-        self._hub.turn_on(self._relay_port)
+        self._hub.turn_on(self._relay_line, self._relay_port)
         self._attr_is_closing = True
-        self.schedule_update_ha_state(False)
+        self.async_write_ha_state()
-        sleep(self._relay_time)
+        await asyncio.sleep(self._relay_time)
         if not self.is_closing:
             # closing stopped
             return
-        self._hub.turn_off(self._relay_port)
+        self._hub.turn_off(self._relay_line, self._relay_port)
         self._attr_is_closing = False
-        self.handle_event()
+        self.async_write_ha_state()
 
-    def open_cover(self, **kwargs):
+    async def async_open_cover(self, **kwargs):
+        _LOGGER.debug(f"GPIODCover async_open_cover: is_closed: {self.is_closed}. is_closing: {self.is_closing}, is_opening: {self.is_opening}")
         if not self.is_closed:
             return
-        self._hub.turn_on(self._relay_port)
+        self._hub.turn_on(self._relay_line, self._relay_port)
         self._attr_is_opening = True
-        self.schedule_update_ha_state(False)
+        self.async_write_ha_state()
-        sleep(self._relay_time)
+        await asyncio.sleep(self._relay_time)
         if not self.is_opening:
             # opening stopped
             return
-        self._hub.turn_off(self._relay_port)
+        self._hub.turn_off(self._relay_line, self._relay_port)
         self._attr_is_opening = False
-        self.handle_event()
+        self.async_write_ha_state()
 
-    def stop_cover(self, **kwargs):
+    async def async_stop_cover(self, **kwargs):
+        _LOGGER.debug(f"GPIODCover async_stop_cover: is_closed: {self.is_closed}. is_closing: {self.is_closing}, is_opening: {self.is_opening}")
         if not (self.is_closing or self.is_opening):
             return
-        self._hub.turn_off(self._relay_port)
+        self._hub.turn_off(self._relay_line, self._relay_port)
         self._attr_is_opening = False
         self._attr_is_closing = False
-        self.schedule_update_ha_state(False)
+        self.async_write_ha_state()
 

custom_components/rpi_gpio/hub.py

@@ -7,7 +7,7 @@ _LOGGER = logging.getLogger(__name__)
 
 from homeassistant.core import HomeAssistant
 from homeassistant.const import EVENT_HOMEASSISTANT_STOP, EVENT_HOMEASSISTANT_START
-from homeassistant.exceptions import HomeAssistantError
+from homeassistant.exceptions import HomeAssistantError,ServiceValidationError
 
 from typing import Dict
 from datetime import timedelta
@@ -39,10 +39,6 @@ class Hub:
         self._id = path
         self._hass = hass
         self._online = False
-        self._lines : gpiod.LineRequest = None
-        self._config : Dict[int, gpiod.LineSettings] = {}
-        self._edge_events = False
-        self._entities = {}
 
         if path:
             # use config
@@ -62,13 +58,8 @@ class Hub:
                     break
 
         self.verify_online()
-
         _LOGGER.debug(f"using gpio_device: {self._path}")
 
-        # startup and shutdown triggers of hass
-        self._hass.bus.async_listen_once(EVENT_HOMEASSISTANT_START, self.startup)
-        self._hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self.cleanup)
-
     def verify_online(self):
         if not self._online:
             _LOGGER.error("No gpio device detected, bailing out")
@@ -82,6 +73,7 @@ class Hub:
         _LOGGER.debug(f"verify_gpiochip: {path} is a gpiochip_device")
         self._chip = gpiod.Chip(path)
         info = self._chip.get_info()
+        _LOGGER.debug(f"verify_gpiochip: {path} info is: {info}")
         if not "pinctrl" in info.label:
             _LOGGER.debug(f"verify_gpiochip: {path} no pinctrl {info.label}")
             return False
@@ -91,126 +83,67 @@ class Hub:
 
     def verify_port_ready(self, port: int):
         info = self._chip.get_line_info(port)
-        _LOGGER.debug(f"original port info: {info}")
+        _LOGGER.debug(f"original port {port} info: {info}")
-        if info.used and info.consumer != DOMAIN:
+        if info.used:
-            _LOGGER.error(f"Port {port} already in use by {info.consumer}")
+            if info.consumer != DOMAIN:
-            raise HomeAssistantError(f"Port {port} already in use by {info.consumer}")
+                raise HomeAssistantError(f"Port {port} already in use by {info.consumer}")
-
+            else:
-
+                raise HomeAssistantError(f"Port {port} already in use by another entity, check your config for duplicates port usage")
-    async def startup(self, _):
-        """Stuff to do after starting."""
-        _LOGGER.debug(f"startup {DOMAIN} hub")
-        if not self._online:
-            _LOGGER.debug(f"integration is not online")
-            return
-        if not self._config:
-            _LOGGER.debug(f"gpiod config is empty")
-            return
-        
-        # setup lines
-        try:
-            self.update_lines()
-        except Exception as e:
-            _LOGGER.error(f"Failed to update lines: {e}")
-            return    
-
-        if not self._edge_events:
-            return
-
-        _LOGGER.debug("Start listener")
-        self._hass.loop.add_reader(self._lines.fd, self.handle_events)
-
-    def cleanup(self, _):
-        """Stuff to do before stopping."""
-        _LOGGER.debug(f"cleanup {DOMAIN} hub")
-        if self._config:
-            self._config.clear()
-        if self._lines:
-            self._lines.release()
-        if self._chip:
-            self._chip.close()
-        self._online = False
 
     @property
     def hub_id(self) -> str:
         """ID for hub"""
         return self._id
 
-    def update_lines(self) -> None:
+    def add_switch(self, port, active_low, bias, drive_mode, init_state) -> gpiod.LineRequest:
-        if self._lines:
+        _LOGGER.debug(f"add_switch - port: {port}, active_low: {active_low}, bias: {bias}, drive_mode: {drive_mode}, init_state: {init_state}")
-            self._lines.release()
-
-        _LOGGER.debug(f"updating lines: {self._config}")
-        self._lines = self._chip.request_lines(
-            consumer = DOMAIN,
-            config = self._config
-        )
-        _LOGGER.debug(f"update_lines new lines: {self._lines}")
-
-    def handle_events(self):
-        for event in self._lines.read_edge_events():
-            _LOGGER.debug(f"Event: {event}")
-            self._entities[event.line_offset].handle_event()
-
-    def add_switch(self, entity, port, active_low, bias, drive_mode, init_output_value = True) -> None:
-        _LOGGER.debug(f"in add_switch {port}")
         self.verify_online()
         self.verify_port_ready(port)
 
-        self._entities[port] = entity
+        line_request = self._chip.request_lines(
-        self._config[port] = gpiod.LineSettings(
+            consumer=DOMAIN,
-            direction = Direction.OUTPUT,
+            config={port: gpiod.LineSettings(
-            bias = BIAS[bias],
+                direction = Direction.OUTPUT,
-            drive = DRIVE[drive_mode],
+                bias = BIAS[bias],
-            active_low = active_low,
+                drive = DRIVE[drive_mode],
-            output_value = Value.ACTIVE if init_output_value and entity.is_on else Value.INACTIVE
+                active_low = active_low,
-        )
+                output_value = Value.ACTIVE if init_state is not None and init_state else Value.INACTIVE)})
+        _LOGGER.debug(f"add_switch line_request: {line_request}")
+        return line_request
 
-    def turn_on(self, port) -> None:
+    def turn_on(self, line, port) -> None:
         _LOGGER.debug(f"in turn_on {port}")
         self.verify_online()
-        self._lines.set_value(port, Value.ACTIVE)
+        line.set_value(port, Value.ACTIVE)
 
-    def turn_off(self, port) -> None:
+    def turn_off(self, line, port) -> None:
         _LOGGER.debug(f"in turn_off {port}")
         self.verify_online()
-        self._lines.set_value(port, Value.INACTIVE)
+        line.set_value(port, Value.INACTIVE)
 
-    def add_sensor(self, entity, port, active_low, bias, debounce) -> None:
+    def add_sensor(self, port, active_low, bias, debounce) -> gpiod.LineRequest:
-        _LOGGER.debug(f"in add_sensor {port}")
+        _LOGGER.debug(f"add_sensor - port: {port}, active_low: {active_low}, bias: {bias}, debounce: {debounce}")
         self.verify_online()
         self.verify_port_ready(port)
 
-        self._entities[port] = entity
+        line_request = self._chip.request_lines(
-        self._config[port] = gpiod.LineSettings(
-            direction = Direction.INPUT,
-            edge_detection = Edge.BOTH,
-            bias = BIAS[bias],
-            active_low = active_low,
-            debounce_period = timedelta(milliseconds=debounce),
-            event_clock = Clock.REALTIME
-        )
-
-        # read current status of the sensor
-        with self._chip.request_lines(
             consumer=DOMAIN,
             config={port: gpiod.LineSettings(
                 direction = Direction.INPUT,
+                edge_detection = Edge.BOTH,
                 bias = BIAS[bias],
-                active_low = active_low)},
+                active_low = active_low,
-        ) as request:
+                debounce_period = timedelta(milliseconds=debounce),
-            # Although we prefer to set the _attr_is_on attribute, in cover it does not exists for whatever reason.
+                event_clock = Clock.REALTIME)})
-            entity.is_on = True if request.get_value(port) == Value.ACTIVE else False
+        _LOGGER.debug(f"add_sensor line_request: {line_request}")
+        current_is_on = True if line_request.get_value(port) == Value.ACTIVE else False
+        _LOGGER.debug(f"add_sensor current state: {current_is_on}")
+        return line_request, current_is_on
 
-        _LOGGER.debug(f"current value for port {port}: {entity.is_on}")
+    def add_cover(self, relay_port, relay_active_low, relay_bias, relay_drive, 
-        self._edge_events = True
+                  state_port, state_bias, state_active_low):
-
+        _LOGGER.debug(f"add_cover - relay_port: {relay_port}, state_port: {state_port}")
-    def get_line_value(self, port):
+        relay_line = self.add_switch(relay_port, relay_active_low, relay_bias, relay_drive, False)
-        return self._lines.get_value(port) == Value.ACTIVE
+        state_line, current_is_on = self.add_sensor(state_port, state_active_low, state_bias, 50)
-
+        return relay_line, state_line, current_is_on
-    def add_cover(self, entity, relay_port, relay_active_low, relay_bias, relay_drive, 
-                  state_port, state_bias, state_active_low) -> None:
-        _LOGGER.debug(f"in add_cover {relay_port} {state_port}")
-        self.add_switch(entity, relay_port, relay_active_low, relay_bias, relay_drive, init_output_value = False)
-        self.add_sensor(entity, state_port, state_active_low, state_bias, 50)
 

custom_components/rpi_gpio/manifest.json

@@ -7,5 +7,5 @@
   "iot_class": "local_push",
   "issue_tracker": "https://github.com/thecode/ha-rpi_gpio/issues",
   "requirements": [ "gpiod>=2.2.1" ],
-  "version": "2024.12.1"
+  "version": "2025.2.1"
 }

custom_components/rpi_gpio/switch.py

@@ -56,18 +56,21 @@ async def async_setup_platform(
 
     switches = []
     for switch in config.get(CONF_SWITCHES):
-        switches.append(
+        try:
-            GPIODSwitch(
+            switches.append(
-                hub,
+                GPIODSwitch(
-                switch[CONF_NAME],
+                    hub,
-                switch[CONF_PORT],
+                    switch[CONF_NAME],
-                switch.get(CONF_UNIQUE_ID) or f"{DOMAIN}_{switch[CONF_PORT]}_{switch[CONF_NAME].lower().replace(' ', '_')}",
+                    switch[CONF_PORT],
-                switch.get(CONF_INVERT_LOGIC),
+                    switch.get(CONF_UNIQUE_ID) or f"{DOMAIN}_{switch[CONF_PORT]}_{switch[CONF_NAME].lower().replace(' ', '_')}",
-                switch.get(CONF_PULL_MODE),
+                    switch.get(CONF_INVERT_LOGIC),
-                switch.get(CONF_DRIVE),
+                    switch.get(CONF_PULL_MODE),
-                switch[CONF_PERSISTENT]
+                    switch.get(CONF_DRIVE),
+                    switch[CONF_PERSISTENT]
+                )
             )
-        )
+        except Exception as e:
+            _LOGGER.error(f"Failed to add switch {switch[CONF_NAME]} for port {switch[CONF_PORT]}: {e}")
 
     async_add_entities(switches)
 
@@ -85,6 +88,8 @@ class GPIODSwitch(SwitchEntity, RestoreEntity):
         self._bias = bias
         self._drive_mode = drive
         self._persistent = persistent
+        self._line = None
+        self._hub.verify_port_ready(self._port)
         
     async def async_added_to_hass(self) -> None:
         """Call when the switch is added to hass."""
@@ -95,19 +100,21 @@ class GPIODSwitch(SwitchEntity, RestoreEntity):
         else: 
             _LOGGER.debug(f"setting initial persistent state for: {self._port}. state: {state.state}")
             self._attr_is_on = True if state.state == STATE_ON else False
-        self._hub.add_switch(self, self._port, self._active_low, self._bias, self._drive_mode)
+            self.async_write_ha_state()
-        self.async_write_ha_state()
+        self._line = self._hub.add_switch(self._port, self._active_low, self._bias, self._drive_mode, self._attr_is_on)
+
+    async def async_will_remove_from_hass(self) -> None:
+        await super().async_will_remove_from_hass()
+        _LOGGER.debug(f"GPIODSwitch async_will_remove_from_hass")
+        if self._line:
+            self._line.release()
 
     async def async_turn_on(self, **kwargs: Any) -> None:
-        self._hub.turn_on(self._port)
+        self._hub.turn_on(self._line, self._port)
         self._attr_is_on = True
         self.async_write_ha_state()
 
     async def async_turn_off(self, **kwargs: Any) -> None:
-        self._hub.turn_off(self._port)
+        self._hub.turn_off(self._line, self._port)
         self._attr_is_on = False
         self.async_write_ha_state()
-
-    def handle_event(self):
-        self._attr_is_on = self._hub.get_line_value(self._port)
-        self.schedule_update_ha_state(False)

requirements_lint.txt

@@ -1,7 +1,7 @@
-black==24.10.0
+black==25.1.0
-flake8==7.1.1
+flake8==7.3.0
-isort==5.13.2
+isort==6.0.1
-mypy==1.13.0
+mypy==1.17.0
-pre-commit==4.0.1
+pre-commit==4.2.0
 pydocstyle==6.3.0
-pylint==3.3.2
+pylint==3.3.7