Add rpi_power integration (#35527)

Co-authored-by: Toast <swetoast@users.noreply.github.com>
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

CODEOWNERS

@@ -357,6 +357,7 @@ homeassistant/components/rmvtransport/* @cgtobi
 homeassistant/components/roku/* @ctalkington
 homeassistant/components/roomba/* @pschmitt @cyr-ius @shenxn
 homeassistant/components/roon/* @pavoni
+homeassistant/components/rpi_power/* @shenxn @swetoast
 homeassistant/components/safe_mode/* @home-assistant/core
 homeassistant/components/saj/* @fredericvl
 homeassistant/components/salt/* @bjornorri

homeassistant/components/rpi_power/__init__.py

@@ -0,0 +1,21 @@
+"""The Raspberry Pi Power Supply Checker integration."""
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+
+
+async def async_setup(hass: HomeAssistant, config: dict):
+    """Set up the Raspberry Pi Power Supply Checker component."""
+    return True
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
+    """Set up Raspberry Pi Power Supply Checker from a config entry."""
+    hass.async_create_task(
+        hass.config_entries.async_forward_entry_setup(entry, "binary_sensor")
+    )
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
+    """Unload a config entry."""
+    return await hass.config_entries.async_forward_entry_unload(entry, "binary_sensor")

homeassistant/components/rpi_power/binary_sensor.py

@@ -0,0 +1,73 @@
+"""
+A sensor platform which detects underruns and capped status from the official Raspberry Pi Kernel.
+
+Minimal Kernel needed is 4.14+
+"""
+import logging
+
+from rpi_bad_power import new_under_voltage
+
+from homeassistant.components.binary_sensor import (
+    DEVICE_CLASS_PROBLEM,
+    BinarySensorEntity,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+
+_LOGGER = logging.getLogger(__name__)
+
+DESCRIPTION_NORMALIZED = "Voltage normalized. Everything is working as intended."
+DESCRIPTION_UNDER_VOLTAGE = "Under-voltage was detected. Consider getting a uninterruptible power supply for your Raspberry Pi."
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, config_entry: ConfigEntry, async_add_entities
+):
+    """Set up rpi_power binary sensor."""
+    under_voltage = await hass.async_add_executor_job(new_under_voltage)
+    async_add_entities([RaspberryChargerBinarySensor(under_voltage)], True)
+
+
+class RaspberryChargerBinarySensor(BinarySensorEntity):
+    """Binary sensor representing the rpi power status."""
+
+    def __init__(self, under_voltage):
+        """Initialize the binary sensor."""
+        self._under_voltage = under_voltage
+        self._is_on = None
+        self._last_is_on = False
+
+    def update(self):
+        """Update the state."""
+        self._is_on = self._under_voltage.get()
+        if self._is_on != self._last_is_on:
+            if self._is_on:
+                _LOGGER.warning(DESCRIPTION_UNDER_VOLTAGE)
+            else:
+                _LOGGER.info(DESCRIPTION_NORMALIZED)
+            self._last_is_on = self._is_on
+
+    @property
+    def unique_id(self):
+        """Return the unique id of the sensor."""
+        return "rpi_power"  # only one sensor possible
+
+    @property
+    def name(self):
+        """Return the name of the sensor."""
+        return "RPi Power status"
+
+    @property
+    def is_on(self):
+        """Return if there is a problem detected."""
+        return self._is_on
+
+    @property
+    def icon(self):
+        """Return the icon of the sensor."""
+        return "mdi:raspberry-pi"
+
+    @property
+    def device_class(self):
+        """Return the class of this device."""
+        return DEVICE_CLASS_PROBLEM

homeassistant/components/rpi_power/config_flow.py

@@ -0,0 +1,22 @@
+"""Config flow for Raspberry Pi Power Supply Checker."""
+from rpi_bad_power import new_under_voltage
+
+from homeassistant import config_entries
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import config_entry_flow
+
+from .const import DOMAIN
+
+
+async def _async_supported(hass: HomeAssistant) -> bool:
+    """Return if the system supports under voltage detection."""
+    under_voltage = await hass.async_add_executor_job(new_under_voltage)
+    return under_voltage is not None
+
+
+config_entry_flow.register_discovery_flow(
+    DOMAIN,
+    "Raspberry Pi Power Supply Checker",
+    _async_supported,
+    config_entries.CONN_CLASS_LOCAL_POLL,
+)

homeassistant/components/rpi_power/const.py

@@ -0,0 +1,3 @@
+"""Constants for Raspberry Pi Power Supply Checker."""
+
+DOMAIN = "rpi_power"

homeassistant/components/rpi_power/manifest.json

@@ -0,0 +1,13 @@
+{
+  "domain": "rpi_power",
+  "name": "Raspberry Pi Power Supply Checker",
+  "documentation": "https://www.home-assistant.io/integrations/rpi_power",
+  "codeowners": [
+    "@shenxn",
+    "@swetoast"
+  ],
+  "requirements": [
+    "rpi-bad-power==0.0.3"
+  ],
+  "config_flow": true
+}

homeassistant/components/rpi_power/strings.json

@@ -0,0 +1,14 @@
+{
+  "title": "Raspberry Pi Power Supply Checker",
+  "config": {
+    "step": {
+      "confirm": {
+        "description": "[%key:common::config_flow::description::confirm_setup%]"
+      }
+    },
+    "abort": {
+      "single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]",
+      "no_devices_found": "Can't find the system class needed for this component, make sure that your kernel is recent and the hardware is supported"
+    }
+  }
+}

homeassistant/generated/config_flows.py

@@ -152,6 +152,7 @@ FLOWS = [
     "roku",
     "roomba",
     "roon",
+    "rpi_power",
     "samsungtv",
     "sense",
     "sentry",

requirements_all.txt

@@ -1923,6 +1923,9 @@ roonapi==0.0.21
 # homeassistant.components.rova
 rova==0.1.0
 
+# homeassistant.components.rpi_power
+rpi-bad-power==0.0.3
+
 # homeassistant.components.rpi_rf
 # rpi-rf==0.9.7
 

requirements_test_all.txt

@@ -901,6 +901,9 @@ roombapy==1.6.1
 # homeassistant.components.roon
 roonapi==0.0.21
 
+# homeassistant.components.rpi_power
+rpi-bad-power==0.0.3
+
 # homeassistant.components.yamaha
 rxv==0.6.0
 

tests/components/rpi_power/__init__.py

@@ -0,0 +1 @@
+"""Tests for rpi_power."""

tests/components/rpi_power/test_binary_sensor.py

@@ -0,0 +1,73 @@
+"""Tests for rpi_power binary sensor."""
+from datetime import timedelta
+import logging
+
+from homeassistant.components.rpi_power.binary_sensor import (
+    DESCRIPTION_NORMALIZED,
+    DESCRIPTION_UNDER_VOLTAGE,
+)
+from homeassistant.components.rpi_power.const import DOMAIN
+from homeassistant.const import STATE_OFF, STATE_ON
+from homeassistant.setup import async_setup_component
+from homeassistant.util import dt as dt_util
+
+from tests.async_mock import MagicMock
+from tests.common import MockConfigEntry, async_fire_time_changed, patch
+
+ENTITY_ID = "binary_sensor.rpi_power_status"
+
+MODULE = "homeassistant.components.rpi_power.binary_sensor.new_under_voltage"
+
+
+async def _async_setup_component(hass, detected):
+    mocked_under_voltage = MagicMock()
+    type(mocked_under_voltage).get = MagicMock(return_value=detected)
+    entry = MockConfigEntry(domain=DOMAIN)
+    entry.add_to_hass(hass)
+    with patch(MODULE, return_value=mocked_under_voltage):
+        await async_setup_component(hass, DOMAIN, {DOMAIN: {}})
+        await hass.async_block_till_done()
+    return mocked_under_voltage
+
+
+async def test_new(hass, caplog):
+    """Test new entry."""
+    await _async_setup_component(hass, False)
+    state = hass.states.get(ENTITY_ID)
+    assert state.state == STATE_OFF
+    assert not any(x.levelno == logging.WARNING for x in caplog.records)
+
+
+async def test_new_detected(hass, caplog):
+    """Test new entry with under voltage detected."""
+    mocked_under_voltage = await _async_setup_component(hass, True)
+    state = hass.states.get(ENTITY_ID)
+    assert state.state == STATE_ON
+    assert (
+        len(
+            [
+                x
+                for x in caplog.records
+                if x.levelno == logging.WARNING
+                and x.message == DESCRIPTION_UNDER_VOLTAGE
+            ]
+        )
+        == 1
+    )
+
+    # back to normal
+    type(mocked_under_voltage).get = MagicMock(return_value=False)
+    future = dt_util.utcnow() + timedelta(minutes=1)
+    async_fire_time_changed(hass, future)
+    await hass.async_block_till_done()
+    state = hass.states.get(ENTITY_ID)
+    assert (
+        len(
+            [
+                x
+                for x in caplog.records
+                if x.levelno == logging.INFO and x.message == DESCRIPTION_NORMALIZED
+            ]
+        )
+        == 1
+    )

tests/components/rpi_power/test_config_flow.py

@@ -0,0 +1,42 @@
+"""Tests for rpi_power config flow."""
+from homeassistant.components.rpi_power.const import DOMAIN
+from homeassistant.config_entries import SOURCE_USER
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import (
+    RESULT_TYPE_ABORT,
+    RESULT_TYPE_CREATE_ENTRY,
+    RESULT_TYPE_FORM,
+)
+
+from tests.async_mock import MagicMock
+from tests.common import patch
+
+MODULE = "homeassistant.components.rpi_power.config_flow.new_under_voltage"
+
+
+async def test_setup(hass: HomeAssistant):
+    """Test setting up manually."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": SOURCE_USER},
+    )
+    assert result["type"] == RESULT_TYPE_FORM
+    assert result["step_id"] == "confirm"
+    assert not result["errors"]
+
+    with patch(MODULE, return_value=MagicMock()):
+        result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+    assert result["type"] == RESULT_TYPE_CREATE_ENTRY
+
+
+async def test_not_supported(hass: HomeAssistant):
+    """Test setting up on not supported system."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": SOURCE_USER},
+    )
+
+    with patch(MODULE, return_value=None):
+        result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+    assert result["type"] == RESULT_TYPE_ABORT
+    assert result["reason"] == "no_devices_found"