Initial commit

custom_components/.github/workflows/hacs.yml

@@ -0,0 +1,18 @@
+name: HACS Action
+
+on:
+  push:
+  pull_request:
+  schedule:
+    - cron: "0 0 * * *"
+
+jobs:
+  hacs:
+    name: HACS Action
+    runs-on: "ubuntu-latest"
+    steps:
+      - uses: "actions/checkout@v2"
+      - name: HACS Action
+        uses: "hacs/action@main"
+        with:
+          category: "integration"

custom_components/.github/workflows/hassfest.yml

@@ -0,0 +1,14 @@
+name: Validate with hassfest
+
+on:
+  push:
+  pull_request:
+  schedule:
+    - cron:  '0 0 * * *'
+
+jobs:
+  validate:
+    runs-on: "ubuntu-latest"
+    steps:
+        - uses: "actions/checkout@v2"
+        - uses: "home-assistant/actions/hassfest@master"

custom_components/rpi_gpio/__init__.py

@@ -0,0 +1,68 @@
+"""Support for controlling GPIO pins of a Raspberry Pi."""
+import logging
+
+from RPi import GPIO  # pylint: disable=import-error
+
+from homeassistant.const import (
+    EVENT_HOMEASSISTANT_START,
+    EVENT_HOMEASSISTANT_STOP,
+    Platform,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.typing import ConfigType
+
+DOMAIN = "rpi_gpio"
+PLATFORMS = [
+    Platform.BINARY_SENSOR,
+    Platform.COVER,
+    Platform.SWITCH,
+]
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def setup(hass: HomeAssistant, config: ConfigType) -> bool:
+    """Set up the Raspberry PI GPIO component."""
+    _LOGGER.warning(
+        "The Raspberry Pi GPIO integration is deprecated and will be removed "
+        "in Home Assistant Core 2022.6; this integration is removed under "
+        "Architectural Decision Record 0019, more information can be found here: "
+        "https://github.com/home-assistant/architecture/blob/master/adr/0019-GPIO.md"
+    )
+
+    def cleanup_gpio(event):
+        """Stuff to do before stopping."""
+        GPIO.cleanup()
+
+    def prepare_gpio(event):
+        """Stuff to do when Home Assistant starts."""
+        hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, cleanup_gpio)
+
+    hass.bus.listen_once(EVENT_HOMEASSISTANT_START, prepare_gpio)
+    GPIO.setmode(GPIO.BCM)
+    return True
+
+
+def setup_output(port):
+    """Set up a GPIO as output."""
+    GPIO.setup(port, GPIO.OUT)
+
+
+def setup_input(port, pull_mode):
+    """Set up a GPIO as input."""
+    GPIO.setup(port, GPIO.IN, GPIO.PUD_DOWN if pull_mode == "DOWN" else GPIO.PUD_UP)
+
+
+def write_output(port, value):
+    """Write a value to a GPIO."""
+    GPIO.output(port, value)
+
+
+def read_input(port):
+    """Read a value from a GPIO."""
+    return GPIO.input(port)
+
+
+def edge_detect(port, event_callback, bounce):
+    """Add detection for RISING and FALLING events."""
+    GPIO.add_event_detect(port, GPIO.BOTH, callback=event_callback, bouncetime=bounce)

custom_components/rpi_gpio/binary_sensor.py

@@ -0,0 +1,109 @@
+"""Support for binary sensor using RPi GPIO."""
+from __future__ import annotations
+
+import asyncio
+
+import voluptuous as vol
+
+from homeassistant.components import rpi_gpio
+from homeassistant.components.binary_sensor import PLATFORM_SCHEMA, BinarySensorEntity
+from homeassistant.const import DEVICE_DEFAULT_NAME
+from homeassistant.core import HomeAssistant
+import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.reload import setup_reload_service
+from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
+
+from . import DOMAIN, PLATFORMS
+
+CONF_BOUNCETIME = "bouncetime"
+CONF_INVERT_LOGIC = "invert_logic"
+CONF_PORTS = "ports"
+CONF_PULL_MODE = "pull_mode"
+
+DEFAULT_BOUNCETIME = 50
+DEFAULT_INVERT_LOGIC = False
+DEFAULT_PULL_MODE = "UP"
+
+_SENSORS_SCHEMA = vol.Schema({cv.positive_int: cv.string})
+
+PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
+    {
+        vol.Required(CONF_PORTS): _SENSORS_SCHEMA,
+        vol.Optional(CONF_BOUNCETIME, default=DEFAULT_BOUNCETIME): cv.positive_int,
+        vol.Optional(CONF_INVERT_LOGIC, default=DEFAULT_INVERT_LOGIC): cv.boolean,
+        vol.Optional(CONF_PULL_MODE, default=DEFAULT_PULL_MODE): cv.string,
+    }
+)
+
+
+def setup_platform(
+    hass: HomeAssistant,
+    config: ConfigType,
+    add_entities: AddEntitiesCallback,
+    discovery_info: DiscoveryInfoType | None = None,
+) -> None:
+    """Set up the Raspberry PI GPIO devices."""
+    setup_reload_service(hass, DOMAIN, PLATFORMS)
+
+    pull_mode = config[CONF_PULL_MODE]
+    bouncetime = config[CONF_BOUNCETIME]
+    invert_logic = config[CONF_INVERT_LOGIC]
+
+    binary_sensors = []
+    ports = config[CONF_PORTS]
+    for port_num, port_name in ports.items():
+        binary_sensors.append(
+            RPiGPIOBinarySensor(
+                port_name, port_num, pull_mode, bouncetime, invert_logic
+            )
+        )
+    add_entities(binary_sensors, True)
+
+
+class RPiGPIOBinarySensor(BinarySensorEntity):
+    """Represent a binary sensor that uses Raspberry Pi GPIO."""
+
+    async def async_read_gpio(self):
+        """Read state from GPIO."""
+        await asyncio.sleep(float(self._bouncetime) / 1000)
+        self._state = await self.hass.async_add_executor_job(
+            rpi_gpio.read_input, self._port
+        )
+        self.async_write_ha_state()
+
+    def __init__(self, name, port, pull_mode, bouncetime, invert_logic):
+        """Initialize the RPi binary sensor."""
+        self._name = name or DEVICE_DEFAULT_NAME
+        self._port = port
+        self._pull_mode = pull_mode
+        self._bouncetime = bouncetime
+        self._invert_logic = invert_logic
+        self._state = None
+
+        rpi_gpio.setup_input(self._port, self._pull_mode)
+
+        def edge_detected(port):
+            """Edge detection handler."""
+            self.hass.add_job(self.async_read_gpio)
+
+        rpi_gpio.edge_detect(self._port, edge_detected, self._bouncetime)
+
+    @property
+    def should_poll(self):
+        """No polling needed."""
+        return False
+
+    @property
+    def name(self):
+        """Return the name of the sensor."""
+        return self._name
+
+    @property
+    def is_on(self):
+        """Return the state of the entity."""
+        return self._state != self._invert_logic
+
+    def update(self):
+        """Update the GPIO state."""
+        self._state = rpi_gpio.read_input(self._port)

custom_components/rpi_gpio/cover.py

@@ -0,0 +1,139 @@
+"""Support for controlling a Raspberry Pi cover."""
+from __future__ import annotations
+
+from time import sleep
+
+import voluptuous as vol
+
+from homeassistant.components import rpi_gpio
+from homeassistant.components.cover import PLATFORM_SCHEMA, CoverEntity
+from homeassistant.const import CONF_COVERS, CONF_NAME
+from homeassistant.core import HomeAssistant
+import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.reload import setup_reload_service
+from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
+
+from . import DOMAIN, PLATFORMS
+
+CONF_RELAY_PIN = "relay_pin"
+CONF_RELAY_TIME = "relay_time"
+CONF_STATE_PIN = "state_pin"
+CONF_STATE_PULL_MODE = "state_pull_mode"
+CONF_INVERT_STATE = "invert_state"
+CONF_INVERT_RELAY = "invert_relay"
+
+DEFAULT_RELAY_TIME = 0.2
+DEFAULT_STATE_PULL_MODE = "UP"
+DEFAULT_INVERT_STATE = False
+DEFAULT_INVERT_RELAY = False
+_COVERS_SCHEMA = vol.All(
+    cv.ensure_list,
+    [
+        vol.Schema(
+            {
+                CONF_NAME: cv.string,
+                CONF_RELAY_PIN: cv.positive_int,
+                CONF_STATE_PIN: cv.positive_int,
+            }
+        )
+    ],
+)
+
+PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
+    {
+        vol.Required(CONF_COVERS): _COVERS_SCHEMA,
+        vol.Optional(CONF_STATE_PULL_MODE, default=DEFAULT_STATE_PULL_MODE): cv.string,
+        vol.Optional(CONF_RELAY_TIME, default=DEFAULT_RELAY_TIME): cv.positive_int,
+        vol.Optional(CONF_INVERT_STATE, default=DEFAULT_INVERT_STATE): cv.boolean,
+        vol.Optional(CONF_INVERT_RELAY, default=DEFAULT_INVERT_RELAY): cv.boolean,
+    }
+)
+
+
+def setup_platform(
+    hass: HomeAssistant,
+    config: ConfigType,
+    add_entities: AddEntitiesCallback,
+    discovery_info: DiscoveryInfoType | None = None,
+) -> None:
+    """Set up the RPi cover platform."""
+    setup_reload_service(hass, DOMAIN, PLATFORMS)
+
+    relay_time = config[CONF_RELAY_TIME]
+    state_pull_mode = config[CONF_STATE_PULL_MODE]
+    invert_state = config[CONF_INVERT_STATE]
+    invert_relay = config[CONF_INVERT_RELAY]
+    covers = []
+    covers_conf = config[CONF_COVERS]
+
+    for cover in covers_conf:
+        covers.append(
+            RPiGPIOCover(
+                cover[CONF_NAME],
+                cover[CONF_RELAY_PIN],
+                cover[CONF_STATE_PIN],
+                state_pull_mode,
+                relay_time,
+                invert_state,
+                invert_relay,
+            )
+        )
+    add_entities(covers)
+
+
+class RPiGPIOCover(CoverEntity):
+    """Representation of a Raspberry GPIO cover."""
+
+    def __init__(
+        self,
+        name,
+        relay_pin,
+        state_pin,
+        state_pull_mode,
+        relay_time,
+        invert_state,
+        invert_relay,
+    ):
+        """Initialize the cover."""
+        self._name = name
+        self._state = False
+        self._relay_pin = relay_pin
+        self._state_pin = state_pin
+        self._state_pull_mode = state_pull_mode
+        self._relay_time = relay_time
+        self._invert_state = invert_state
+        self._invert_relay = invert_relay
+        rpi_gpio.setup_output(self._relay_pin)
+        rpi_gpio.setup_input(self._state_pin, self._state_pull_mode)
+        rpi_gpio.write_output(self._relay_pin, 0 if self._invert_relay else 1)
+
+    @property
+    def name(self):
+        """Return the name of the cover if any."""
+        return self._name
+
+    def update(self):
+        """Update the state of the cover."""
+        self._state = rpi_gpio.read_input(self._state_pin)
+
+    @property
+    def is_closed(self):
+        """Return true if cover is closed."""
+        return self._state != self._invert_state
+
+    def _trigger(self):
+        """Trigger the cover."""
+        rpi_gpio.write_output(self._relay_pin, 1 if self._invert_relay else 0)
+        sleep(self._relay_time)
+        rpi_gpio.write_output(self._relay_pin, 0 if self._invert_relay else 1)
+
+    def close_cover(self, **kwargs):
+        """Close the cover."""
+        if not self.is_closed:
+            self._trigger()
+
+    def open_cover(self, **kwargs):
+        """Open the cover."""
+        if self.is_closed:
+            self._trigger()

custom_components/rpi_gpio/manifest.json

@@ -0,0 +1,10 @@
+{
+  "domain": "rpi_gpio",
+  "name": "Raspberry Pi GPIO",
+  "documentation": "https://github.com/thecode/ha-rpi_gpio",
+  "issue_tracker": "https://github.com/thecode/ha-rpi_gpio/issues",
+  "requirements": ["RPi.GPIO==0.7.1a4"],
+  "codeowners": ["@thecode"],
+  "iot_class": "local_push",
+  "version": "0.1.0"
+}

custom_components/rpi_gpio/services.yaml

@@ -0,0 +1,3 @@
+reload:
+  name: Reload
+  description: Reload all rpi_gpio entities.

custom_components/rpi_gpio/switch.py

@@ -0,0 +1,88 @@
+"""Allows to configure a switch using RPi GPIO."""
+from __future__ import annotations
+
+import voluptuous as vol
+
+from homeassistant.components import rpi_gpio
+from homeassistant.components.switch import PLATFORM_SCHEMA, SwitchEntity
+from homeassistant.const import DEVICE_DEFAULT_NAME
+from homeassistant.core import HomeAssistant
+import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.reload import setup_reload_service
+from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
+
+from . import DOMAIN, PLATFORMS
+
+CONF_PULL_MODE = "pull_mode"
+CONF_PORTS = "ports"
+CONF_INVERT_LOGIC = "invert_logic"
+
+DEFAULT_INVERT_LOGIC = False
+
+_SWITCHES_SCHEMA = vol.Schema({cv.positive_int: cv.string})
+
+PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
+    {
+        vol.Required(CONF_PORTS): _SWITCHES_SCHEMA,
+        vol.Optional(CONF_INVERT_LOGIC, default=DEFAULT_INVERT_LOGIC): cv.boolean,
+    }
+)
+
+
+def setup_platform(
+    hass: HomeAssistant,
+    config: ConfigType,
+    add_entities: AddEntitiesCallback,
+    discovery_info: DiscoveryInfoType | None = None,
+) -> None:
+    """Set up the Raspberry PI GPIO devices."""
+    setup_reload_service(hass, DOMAIN, PLATFORMS)
+
+    invert_logic = config[CONF_INVERT_LOGIC]
+
+    switches = []
+    ports = config[CONF_PORTS]
+    for port, name in ports.items():
+        switches.append(RPiGPIOSwitch(name, port, invert_logic))
+    add_entities(switches)
+
+
+class RPiGPIOSwitch(SwitchEntity):
+    """Representation of a  Raspberry Pi GPIO."""
+
+    def __init__(self, name, port, invert_logic):
+        """Initialize the pin."""
+        self._name = name or DEVICE_DEFAULT_NAME
+        self._port = port
+        self._invert_logic = invert_logic
+        self._state = False
+        rpi_gpio.setup_output(self._port)
+        rpi_gpio.write_output(self._port, 1 if self._invert_logic else 0)
+
+    @property
+    def name(self):
+        """Return the name of the switch."""
+        return self._name
+
+    @property
+    def should_poll(self):
+        """No polling needed."""
+        return False
+
+    @property
+    def is_on(self):
+        """Return true if device is on."""
+        return self._state
+
+    def turn_on(self, **kwargs):
+        """Turn the device on."""
+        rpi_gpio.write_output(self._port, 0 if self._invert_logic else 1)
+        self._state = True
+        self.schedule_update_ha_state()
+
+    def turn_off(self, **kwargs):
+        """Turn the device off."""
+        rpi_gpio.write_output(self._port, 1 if self._invert_logic else 0)
+        self._state = False
+        self.schedule_update_ha_state()

hacs.json

@@ -0,0 +1,6 @@
+{
+  "name": "Raspberry Pi GPIO",
+  "domains": ["binary_sensor", "cover", "switch"],
+  "iot_class": "Local Push",
+  "homeassistant": "2022.2.0"
+}

info.md

@@ -0,0 +1 @@
+Home Assistant Raspberry Pi GPIO Integration