diff --git a/custom_components/owlet/__init__.py b/custom_components/owlet/__init__.py new file mode 100644 index 0000000..9f75324 --- /dev/null +++ b/custom_components/owlet/__init__.py @@ -0,0 +1,68 @@ +"""The Owlet Smart Sock integration.""" +from __future__ import annotations + +import logging + +from pyowletapi.owlet import Owlet +from pyowletapi.exceptions import OwletAuthenticationError, OwletDevicesError + +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import Platform +from homeassistant.core import HomeAssistant +from homeassistant.helpers.aiohttp_client import async_get_clientsession +from .const import DOMAIN, CONF_OWLET_REGION, CONF_OWLET_USERNAME, CONF_OWLET_PASSWORD +from .coordinator import OwletCoordinator + +PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.SENSOR] + +_LOGGER = logging.getLogger(__name__) + + +async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Set up Owlet Smart Sock from a config entry.""" + hass.data.setdefault(DOMAIN, {}) + + owlet = Owlet( + entry.data[CONF_OWLET_REGION], + entry.data[CONF_OWLET_USERNAME], + entry.data[CONF_OWLET_PASSWORD], + async_get_clientsession(hass), + ) + + try: + await owlet.authenticate() + except OwletAuthenticationError as err: + _LOGGER.error("Login failed %s", err) + return False + + existing_socks = entry.data["devices"] + new_socks = [] + try: + socks = await owlet.get_devices() + except OwletDevicesError: + pass + + [new_socks.append(sock) for sock in socks if sock not in existing_socks] + + if new_socks: + hass.config_entries.async_update_entry( + entry, data={**entry.data, **{"devices": socks}} + ) + + coordinators = [OwletCoordinator(hass, sock) for sock in socks.values()] + + for coordinator in coordinators: + await coordinator.async_config_entry_first_refresh() + hass.data[DOMAIN][entry.entry_id] = coordinator + + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) + + return True + + +async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Unload a config entry.""" + if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS): + hass.data[DOMAIN].pop(entry.entry_id) + + return unload_ok diff --git a/custom_components/owlet/binary_sensor.py b/custom_components/owlet/binary_sensor.py new file mode 100644 index 0000000..7233407 --- /dev/null +++ b/custom_components/owlet/binary_sensor.py @@ -0,0 +1,125 @@ +"""Support for Owlet binary sensors.""" +from __future__ import annotations + +from dataclasses import dataclass + +from homeassistant.components.binary_sensor import ( + BinarySensorDeviceClass, + BinarySensorEntity, + BinarySensorEntityDescription, +) +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity_platform import AddEntitiesCallback + +from .const import DOMAIN +from .coordinator import OwletCoordinator +from .entity import OwletBaseEntity + + +@dataclass +class OwletBinarySensorEntityMixin: + """Owlet binary sensor element mixin""" + + element: str + + +@dataclass +class OwletBinarySensorEntityDescription( + BinarySensorEntityDescription, OwletBinarySensorEntityMixin +): + """Represent the owlet binary sensor entity description.""" + + +SENSOR_TYPES: tuple[OwletBinarySensorEntityDescription, ...] = ( + OwletBinarySensorEntityDescription( + key="charging", + name="Charging", + device_class=BinarySensorDeviceClass.BATTERY_CHARGING, + element="charging", + ), + OwletBinarySensorEntityDescription( + key="highhr", + name="High heart rate alert", + device_class=BinarySensorDeviceClass.SOUND, + element="high_heart_rate_alert", + ), + OwletBinarySensorEntityDescription( + key="lowhr", + name="Low Heart Rate Alert", + device_class=BinarySensorDeviceClass.SOUND, + element="low_heart_rate_alert", + ), + OwletBinarySensorEntityDescription( + key="higho2", + name="High oxygen alert", + device_class=BinarySensorDeviceClass.SOUND, + element="high_oxygen_alert", + ), + OwletBinarySensorEntityDescription( + key="lowo2", + name="Low oxygen alert", + device_class=BinarySensorDeviceClass.SOUND, + element="low_oxygen_alert", + ), + OwletBinarySensorEntityDescription( + key="lowbattery", + name="Low Battery alert", + device_class=BinarySensorDeviceClass.SOUND, + element="low_battery_alert", + ), + OwletBinarySensorEntityDescription( + key="lostpower", + name="Lost power alert", + device_class=BinarySensorDeviceClass.SOUND, + element="lost_power_alert", + ), + OwletBinarySensorEntityDescription( + key="sockdisconnected", + name="Sock disconnected alert", + device_class=BinarySensorDeviceClass.SOUND, + element="sock_disconnected", + ), + OwletBinarySensorEntityDescription( + key="sock_off", + name="Sock off", + device_class=BinarySensorDeviceClass.POWER, + element="sock_off", + ), +) + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up the owlet sensors from config entry.""" + + coordinator: OwletCoordinator = hass.data[DOMAIN][config_entry.entry_id] + + entities = [ + OwletBinarySensor(coordinator, description) for description in SENSOR_TYPES + ] + + async_add_entities(entities) + + +class OwletBinarySensor(OwletBaseEntity, BinarySensorEntity): + """Representation of an Owlet binary sensor.""" + + def __init__( + self, + coordinator: OwletCoordinator, + description: OwletBinarySensorEntityDescription, + ) -> None: + """Initialize the binary sensor.""" + self.entity_description = description + self._attr_unique_id = ( + f"{coordinator.config_entry.entry_id}-{self.entity_description.name}" + ) + super().__init__(coordinator) + + @property + def is_on(self) -> bool: + return self.sock.properties[self.entity_description.element] diff --git a/custom_components/owlet/config_flow.py b/custom_components/owlet/config_flow.py new file mode 100644 index 0000000..f93bdf5 --- /dev/null +++ b/custom_components/owlet/config_flow.py @@ -0,0 +1,97 @@ +"""Config flow for Owlet Smart Sock integration.""" +from __future__ import annotations + +import logging +from typing import Any + +from pyowletapi.owlet import Owlet +from pyowletapi.exceptions import ( + OwletConnectionError, + OwletAuthenticationError, + OwletDevicesError, +) + +import voluptuous as vol + +from homeassistant import config_entries +from homeassistant.data_entry_flow import FlowResult +from homeassistant.config_entries import ConfigEntry +from homeassistant.helpers.aiohttp_client import async_get_clientsession + +from .const import ( + DOMAIN, + CONF_OWLET_REGION, + CONF_OWLET_USERNAME, + CONF_OWLET_PASSWORD, +) + +_LOGGER = logging.getLogger(__name__) + +STEP_USER_DATA_SCHEMA = vol.Schema( + { + vol.Required(CONF_OWLET_REGION): vol.In(["europe", "world"]), + vol.Required(CONF_OWLET_USERNAME): str, + vol.Required(CONF_OWLET_PASSWORD): str, + } +) + + +class OwletConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): + """Handle a config flow for Owlet Smart Sock.""" + + VERSION = 1 + + def __init__(self) -> None: + self._entry: ConfigEntry + self._region: str + self._username: str + self._password: str + + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Handle the initial step.""" + errors: dict[str, str] = {} + if user_input is not None: + self._region = user_input[CONF_OWLET_REGION] + self._username = user_input[CONF_OWLET_USERNAME] + self._password = user_input[CONF_OWLET_PASSWORD] + + owlet = Owlet( + self._region, + self._username, + self._password, + session=async_get_clientsession(self.hass), + ) + + await self.async_set_unique_id(self._username.lower()) + self._abort_if_unique_id_configured() + + try: + await owlet.authenticate() + try: + devices = await owlet.get_devices() + except OwletDevicesError: + errors["base"] = "no_devices" + + except OwletConnectionError: + errors["base"] = "cannot_connect" + except OwletAuthenticationError: + errors["base"] = "invalid_auth" + except Exception: # pylint: disable=broad-except + _LOGGER.exception("Unexpected exception") + errors["base"] = "unknown" + else: + return self.async_create_entry( + title="Owlet", + data={ + CONF_OWLET_REGION: self._region, + CONF_OWLET_USERNAME: self._username, + CONF_OWLET_PASSWORD: self._password, + "devices": list(devices.keys()), + }, + ) + + return self.async_show_form( + step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors + ) diff --git a/custom_components/owlet/const.py b/custom_components/owlet/const.py new file mode 100644 index 0000000..e41ed82 --- /dev/null +++ b/custom_components/owlet/const.py @@ -0,0 +1,11 @@ +"""Constants for the Owlet Smart Sock integration.""" + +DOMAIN = "owlet" + +CONF_OWLET_REGION = "region" +CONF_OWLET_USERNAME = "username" +CONF_OWLET_PASSWORD = "password" +CONF_SOCK_SERIAL = "sock_serial" + +POLLING_INTERVAL = 60 +MANUFACTURER = "Owlet Baby Care" diff --git a/custom_components/owlet/coordinator.py b/custom_components/owlet/coordinator.py new file mode 100644 index 0000000..eeabe8a --- /dev/null +++ b/custom_components/owlet/coordinator.py @@ -0,0 +1,52 @@ +"""Owlet integration.""" +from __future__ import annotations + +from datetime import timedelta +import logging + +from pyowletapi.sock import Sock +from pyowletapi.exceptions import OwletError + +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers.entity import DeviceInfo +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed + +from .const import ( + DOMAIN, + POLLING_INTERVAL, + MANUFACTURER, +) + +_LOGGER = logging.getLogger(__name__) + + +class OwletCoordinator(DataUpdateCoordinator): + """Coordinator is responsible for querying the device at a specified route.""" + + def __init__(self, hass: HomeAssistant, sock: Sock) -> None: + """Initialise a custom coordinator.""" + super().__init__( + hass, + _LOGGER, + name=DOMAIN, + update_interval=timedelta(seconds=POLLING_INTERVAL), + ) + assert self.config_entry is not None + self._device_unique_id = sock.serial + self._model = sock.model + self._sw_version = sock.sw_version + self.sock = sock + self.device_info = DeviceInfo( + identifiers={(DOMAIN, self._device_unique_id)}, + name="Owlet Baby Care Sock", + manufacturer=MANUFACTURER, + model=self._model, + sw_version=self._sw_version, + ) + + async def _async_update_data(self) -> None: + """Fetch the data from the device.""" + try: + await self.sock.update_properties() + except OwletError as err: + raise UpdateFailed(err) from err diff --git a/custom_components/owlet/entity.py b/custom_components/owlet/entity.py new file mode 100644 index 0000000..f5633a6 --- /dev/null +++ b/custom_components/owlet/entity.py @@ -0,0 +1,20 @@ +"""Base class for Owlet entities.""" + +from homeassistant.helpers.entity import Entity +from homeassistant.helpers.update_coordinator import CoordinatorEntity + +from .coordinator import OwletCoordinator + + +class OwletBaseEntity(CoordinatorEntity[OwletCoordinator], Entity): + """Base class for Owlet Sock entities.""" + + def __init__( + self, + coordinator: OwletCoordinator, + ) -> None: + """Initialize the base entity.""" + super().__init__(coordinator) + self.sock = coordinator.sock + self._attr_device_info = coordinator.device_info + self._attr_has_entity_name = True diff --git a/custom_components/owlet/manifest.json b/custom_components/owlet/manifest.json new file mode 100644 index 0000000..0a34229 --- /dev/null +++ b/custom_components/owlet/manifest.json @@ -0,0 +1,16 @@ +{ + "domain": "owlet", + "name": "Owlet Smart Sock", + "codeowners": [ + "@RyanClark123" + ], + "config_flow": true, + "dependencies": [], + "documentation": "https://www.home-assistant.io/integrations/owlet", + "homekit": {}, + "iot_class": "cloud_polling", + "requirements": [ + "pyowletapi==2023.5.7" + ], + "version":"1.0.0" +} diff --git a/custom_components/owlet/sensor.py b/custom_components/owlet/sensor.py new file mode 100644 index 0000000..ff1a33b --- /dev/null +++ b/custom_components/owlet/sensor.py @@ -0,0 +1,106 @@ +"""Support for Android IP Webcam binary sensors.""" +from __future__ import annotations +from dataclasses import dataclass + +from homeassistant.components.sensor import ( + SensorDeviceClass, + SensorEntity, + SensorEntityDescription, +) +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.const import ( + PERCENTAGE, + SIGNAL_STRENGTH_DECIBELS_MILLIWATT, + UnitOfTime, +) + +from .const import DOMAIN +from .coordinator import OwletCoordinator +from .entity import OwletBaseEntity + + +@dataclass +class OwletSensorEntityDescriptionMixin: + """Owlet sensor description mix in""" + + element: str + + +@dataclass +class OwletSensorEntityDescription( + SensorEntityDescription, OwletSensorEntityDescriptionMixin +): + """Represent the owlet sensor entity description.""" + + +SENSOR_TYPES: tuple[OwletSensorEntityDescription, ...] = ( + OwletSensorEntityDescription( + key="batterypercentage", + name="Battery", + native_unit_of_measurement=PERCENTAGE, + device_class=SensorDeviceClass.BATTERY, + element="battery_percentage", + ), + OwletSensorEntityDescription( + key="oxygensaturation", + name="O2 Saturation", + native_unit_of_measurement=PERCENTAGE, + element="oxygen_saturation", + icon="mdi:leaf", + ), + OwletSensorEntityDescription( + key="heartrate", + name="Heart rate", + element="heart_rate", + icon="mdi:heart-pulse", + ), + OwletSensorEntityDescription( + key="batteryminutes", + name="Battery Minutes Remaining", + native_unit_of_measurement=UnitOfTime.MINUTES, + device_class=SensorDeviceClass.DURATION, + element="battery_minutes", + ), + OwletSensorEntityDescription( + key="signalstrength", + name="Singal Strength", + native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT, + device_class=SensorDeviceClass.SIGNAL_STRENGTH, + element="signal_strength", + ), +) + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up the IP Webcam sensors from config entry.""" + + coordinator: OwletCoordinator = hass.data[DOMAIN][config_entry.entry_id] + + entities = [OwletSensor(coordinator, description) for description in SENSOR_TYPES] + + async_add_entities(entities) + + +class OwletSensor(OwletBaseEntity, SensorEntity): + """Representation of an Owlet sensor.""" + + def __init__( + self, coordinator: OwletCoordinator, description: OwletSensorEntityDescription + ) -> None: + """Initialize the binary sensor.""" + self.entity_description = description + self._attr_unique_id = ( + f"{coordinator.config_entry.entry_id}-{self.entity_description.name}" + ) + super().__init__(coordinator) + + @property + def native_value(self) -> float: + """Return if motion is detected.""" + return self.sock.properties[self.entity_description.element] diff --git a/custom_components/owlet/strings.json b/custom_components/owlet/strings.json new file mode 100644 index 0000000..544726c --- /dev/null +++ b/custom_components/owlet/strings.json @@ -0,0 +1,21 @@ +{ + "config": { + "step": { + "user": { + "data": { + "region": "[%key:common::config_flow::data::host%]", + "username": "[%key:common::config_flow::data::username%]", + "password": "[%key:common::config_flow::data::password%]" + } + } + }, + "error": { + "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", + "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]", + "unknown": "[%key:common::config_flow::error::unknown%]" + }, + "abort": { + "already_configured": "[%key:common::config_flow::abort::already_configured_device%]" + } + } +} diff --git a/custom_components/owlet/translations/en.json b/custom_components/owlet/translations/en.json new file mode 100644 index 0000000..d7d0160 --- /dev/null +++ b/custom_components/owlet/translations/en.json @@ -0,0 +1,21 @@ +{ + "config": { + "abort": { + "already_configured": "Device is already configured" + }, + "error": { + "cannot_connect": "Failed to connect", + "invalid_auth": "Invalid authentication", + "unknown": "Unexpected error" + }, + "step": { + "user": { + "data": { + "password": "Password", + "region": "Host", + "username": "Username" + } + } + } + } +} \ No newline at end of file