Initial commit

Initial commit
This commit is contained in:
RyanClark123
2023-05-04 16:33:06 +01:00
parent f5ed126eaa
commit fe34bd63af
10 changed files with 537 additions and 0 deletions
+68
View File
@@ -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
+125
View File
@@ -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]
+97
View File
@@ -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
)
+11
View File
@@ -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"
+52
View File
@@ -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
+20
View File
@@ -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
+16
View File
@@ -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"
}
+106
View File
@@ -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]
+21
View File
@@ -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%]"
}
}
}
@@ -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"
}
}
}
}
}