Add new CentriConnect component

.strict-typing

@@ -138,6 +138,7 @@ homeassistant.components.cambridge_audio.*
 homeassistant.components.camera.*
 homeassistant.components.canary.*
 homeassistant.components.casper_glow.*
+homeassistant.components.centriconnect.*
 homeassistant.components.cert_expiry.*
 homeassistant.components.clickatell.*
 homeassistant.components.clicksend.*

homeassistant/components/centriconnect/__init__.py

@@ -0,0 +1,33 @@
+"""The CentriConnect/MyPropane API integration."""
+
+from __future__ import annotations
+
+import logging
+
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from .coordinator import CentriConnectConfigEntry, CentriConnectCoordinator
+
+_LOGGER = logging.getLogger(__name__)
+
+PLATFORMS: list[Platform] = [Platform.SENSOR]
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: CentriConnectConfigEntry
+) -> bool:
+    """Set up CentriConnect/MyPropane API from a config entry."""
+    coordinator = CentriConnectCoordinator(hass, entry)
+    await coordinator.async_config_entry_first_refresh()
+    entry.runtime_data = coordinator
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    return True
+
+
+async def async_unload_entry(
+    hass: HomeAssistant, entry: CentriConnectConfigEntry
+) -> bool:
+    """Unload CentriConnect/MyPropane API integration platforms and coordinator."""
+    _LOGGER.info("Unloading CentriConnect/MyPropane API integration")
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/centriconnect/config_flow.py

@@ -0,0 +1,95 @@
+"""Config flow for the CentriConnect/MyPropane API integration."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from aiocentriconnect import CentriConnect
+from aiocentriconnect.exceptions import (
+    CentriConnectConnectionError,
+    CentriConnectDecodeError,
+    CentriConnectEmptyResponseError,
+    CentriConnectNotFoundError,
+    CentriConnectTooManyRequestsError,
+)
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.const import CONF_DEVICE_ID, CONF_PASSWORD, CONF_USERNAME
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import CENTRICONNECT_DEVICE_ID, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_USERNAME): str,
+        vol.Required(CONF_DEVICE_ID): str,
+        vol.Required(CONF_PASSWORD): str,
+    }
+)
+
+
+async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str, Any]:
+    """Validate the user input allows us to connect.
+
+    Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
+    """
+    # Validate the user-supplied data can be used to set up a connection.
+    hub = CentriConnect(
+        data[CONF_USERNAME],
+        data[CONF_DEVICE_ID],
+        data[CONF_PASSWORD],
+        session=async_get_clientsession(hass),
+    )
+
+    tank_data = await hub.async_get_tank_data()
+
+    # Return info to store in the config entry.
+    return {
+        "title": tank_data.device_name,
+        CENTRICONNECT_DEVICE_ID: tank_data.device_id,
+    }
+
+
+class CentriConnectConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for CentriConnect/MyPropane API."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> config_entries.ConfigFlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            try:
+                info = await validate_input(self.hass, user_input)
+            except CentriConnectConnectionError:
+                errors["base"] = "cannot_connect"
+            except CentriConnectTooManyRequestsError:
+                errors["base"] = "cannot_connect"
+            except CentriConnectNotFoundError:
+                errors["base"] = "invalid_auth"
+            except CentriConnectEmptyResponseError:
+                errors["base"] = "unknown"
+            except CentriConnectDecodeError:
+                errors["base"] = "unknown"
+            except Exception:
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            else:
+                await self.async_set_unique_id(
+                    unique_id=info[CENTRICONNECT_DEVICE_ID], raise_on_progress=True
+                )
+                self._abort_if_unique_id_configured(
+                    updates=user_input, reload_on_update=True
+                )
+                return self.async_create_entry(title=info["title"], data=user_input)
+
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )

homeassistant/components/centriconnect/const.py

@@ -0,0 +1,5 @@
+"""Constants for the CentriConnect/MyPropane API integration."""
+
+DOMAIN = "centriconnect"
+
+CENTRICONNECT_DEVICE_ID = "device_id"

homeassistant/components/centriconnect/coordinator.py

@@ -0,0 +1,89 @@
+"""Coordinator for CentriConnect/MyPropane API integration.
+
+Responsible for polling the device API endpoint and normalizing data for entities.
+"""
+
+from dataclasses import dataclass
+from datetime import timedelta
+import logging
+
+from aiocentriconnect import CentriConnect, Tank
+from aiocentriconnect.exceptions import CentriConnectConnectionError, CentriConnectError
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_DEVICE_ID, CONF_PASSWORD, CONF_USERNAME
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+COORDINATOR_NAME = f"{DOMAIN} Coordinator"
+# Maximum update frequency is every 6 hours. The API will return 429 Too Many Requests if polled frequently.
+# The device updates its data every 8-12 hours, so there's no need to poll more frequently.
+UPDATE_INTERVAL = timedelta(hours=6)
+
+type CentriConnectConfigEntry = ConfigEntry[CentriConnectCoordinator]
+
+
+@dataclass
+class CentriConnectDeviceInfo:
+    """Data about the CentriConnect device."""
+
+    device_id: str
+    device_name: str
+    hardware_version: str
+    lte_version: str
+    tank_size: int
+    tank_size_unit: str
+
+
+class CentriConnectCoordinator(DataUpdateCoordinator[Tank]):
+    """Data update coordinator for CentriConnect/MyPropane devices."""
+
+    config_entry: CentriConnectConfigEntry
+    device_info: CentriConnectDeviceInfo
+
+    def __init__(self, hass: HomeAssistant, entry: CentriConnectConfigEntry) -> None:
+        """Initialize the CentriConnect data update coordinator."""
+        super().__init__(
+            hass,
+            logger=_LOGGER,
+            name=COORDINATOR_NAME,
+            update_interval=UPDATE_INTERVAL,
+            config_entry=entry,
+        )
+
+        self.api_client = CentriConnect(
+            entry.data[CONF_USERNAME],
+            entry.data[CONF_DEVICE_ID],
+            entry.data[CONF_PASSWORD],
+            session=async_get_clientsession(hass),
+        )
+
+    async def _async_setup(self) -> None:
+        try:
+            tank_data = await self.api_client.async_get_tank_data()
+            self.device_info = CentriConnectDeviceInfo(
+                device_id=tank_data.device_id,
+                device_name=tank_data.device_name,
+                hardware_version=tank_data.hardware_version,
+                lte_version=tank_data.lte_version,
+                tank_size=tank_data.tank_size,
+                tank_size_unit=tank_data.tank_size_unit,
+            )
+        except CentriConnectError as err:
+            raise ConfigEntryNotReady("Could not fetch device info") from err
+
+    async def _async_update_data(self) -> Tank:
+        """Fetch device state."""
+        try:
+            state = await self.api_client.async_get_tank_data()
+        except CentriConnectConnectionError as err:
+            raise UpdateFailed(f"Error communicating with device: {err}") from err
+        except CentriConnectError as err:
+            raise UpdateFailed(f"Unexpected response: {err}") from err
+        return state

homeassistant/components/centriconnect/entity.py

@@ -0,0 +1,37 @@
+"""Defines a base CentriConnect entity."""
+
+from typing import TYPE_CHECKING
+
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.entity import EntityDescription
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN
+from .coordinator import CentriConnectCoordinator
+
+
+class CentriConnectBaseEntity(CoordinatorEntity[CentriConnectCoordinator]):
+    """Defines a base CentriConnect entity."""
+
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        coordinator: CentriConnectCoordinator,
+        description: EntityDescription,
+    ) -> None:
+        """Initialize the CentriConnect entity."""
+        super().__init__(coordinator)
+        if TYPE_CHECKING:
+            assert coordinator.config_entry.unique_id
+
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, coordinator.config_entry.unique_id)},
+            name=coordinator.device_info.device_name,
+            serial_number=coordinator.device_info.device_id,
+            hw_version=coordinator.device_info.hardware_version,
+            sw_version=coordinator.device_info.lte_version,
+            manufacturer="CentriConnect",
+        )
+        self._attr_unique_id = f"{coordinator.config_entry.unique_id}_{description.key}"
+        self.entity_description = description

homeassistant/components/centriconnect/icons.json

@@ -0,0 +1,84 @@
+{
+  "entity": {
+    "sensor": {
+      "alert_status": {
+        "default": "mdi:alert-circle-outline",
+        "state": {
+          "critical_level": "mdi:alert-circle",
+          "low_level": "mdi:alert-circle-outline",
+          "no_alert": "mdi:check-circle-outline"
+        }
+      },
+      "altitude": {
+        "default": "mdi:altimeter"
+      },
+      "battery_level": {
+        "default": "mdi:battery-unknown",
+        "range": {
+          "0": "mdi:battery-outline",
+          "10": "mdi:battery-10",
+          "20": "mdi:battery-20",
+          "30": "mdi:battery-30",
+          "40": "mdi:battery-40",
+          "50": "mdi:battery-50",
+          "60": "mdi:battery-60",
+          "70": "mdi:battery-70",
+          "80": "mdi:battery-80",
+          "90": "mdi:battery-90",
+          "100": "mdi:battery"
+        }
+      },
+      "battery_voltage": {
+        "default": "mdi:car-battery"
+      },
+      "device_temperature": {
+        "default": "mdi:thermometer"
+      },
+      "last_post_time": {
+        "default": "mdi:clock-end"
+      },
+      "latitude": {
+        "default": "mdi:latitude"
+      },
+      "longitude": {
+        "default": "mdi:longitude"
+      },
+      "lte_signal_level": {
+        "default": "mdi:signal",
+        "range": {
+          "0": "mdi:signal-cellular-outline",
+          "25": "mdi:signal-cellular-1",
+          "50": "mdi:signal-cellular-2",
+          "75": "mdi:signal-cellular-3"
+        }
+      },
+      "lte_signal_strength": {
+        "default": "mdi:signal-variant"
+      },
+      "next_post_time": {
+        "default": "mdi:clock-start"
+      },
+      "solar_level": {
+        "default": "mdi:sun-wireless"
+      },
+      "solar_voltage": {
+        "default": "mdi:solar-power"
+      },
+      "tank_level": {
+        "default": "mdi:gauge",
+        "range": {
+          "0": "mdi:gauge-empty",
+          "25": "mdi:gauge-low",
+          "50": "mdi:gauge",
+          "75": "mdi:gauge-full"
+        }
+      },
+      "tank_remaining_volume": {
+        "default": "mdi:storage-tank-outline"
+      },
+      "tank_size": {
+        "default": "mdi:storage-tank"
+      }
+    }
+  }
+}

homeassistant/components/centriconnect/manifest.json

@@ -0,0 +1,11 @@
+{
+  "domain": "centriconnect",
+  "name": "CentriConnect/MyPropane",
+  "codeowners": ["@gresrun"],
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/centriconnect",
+  "integration_type": "device",
+  "iot_class": "cloud_polling",
+  "quality_scale": "bronze",
+  "requirements": ["aiocentriconnect==0.2.2"]
+}