Sure Petcare new features various improvements (#31437)
* add typing * 100% battery_level is enough * human-friendly datetime * better enum usage * add online and learning mode attrs * use max two decimals in attrs * use legacy style debug logging * remove str usage of enums * add feeder * add feeder and adapt to new surepy version * use ProductID instead of ThingID * various changes and improvements * add connectivity sensors for all devices & proper support for multiple hubs * remove "side effects"/exception catching in attribs * correct unique ids, reorder classes * move Flap class from binary_sensor to sensor and add a sensore base class * comments cleanup, minor typing and logging fixes * remove commented code * remove commented code * add typing * 100% battery_level is enough * human-friendly datetime * better enum usage * add online and learning mode attrs * use max two decimals in attrs * use legacy style debug logging * remove str usage of enums * add feeder * add feeder and adapt to new surepy version * use ProductID instead of ThingID * various changes and improvements * add connectivity sensors for all devices & proper support for multiple hubs * remove "side effects"/exception catching in attribs * correct unique ids, reorder classes * move Flap class from binary_sensor to sensor and add a sensore base class * comments cleanup, minor typing and logging fixes * remove commented code * remove commented code * fix spelling in comment to make the CI happy (seriously?!) * fix manifest file * fix requirements_all.txt file * add missing docstrings * fix available property * remove typing from self * remove commented code * remove is_on property from sensor * jump to new surepy version * remove useles init methodspull/32128/head
Ben
GitHub
parent
150b376cf9
commit
fb2e120563
homeassistant/components/surepetcare
|
|
@ -1,17 +1,17 @@
|
|||
"""Support for Sure Petcare cat/pet flaps."""
|
||||
import logging
|
||||
from typing import Any, Dict, List
|
||||
|
||||
from surepy import (
|
||||
SurePetcare,
|
||||
SurePetcareAuthenticationError,
|
||||
SurePetcareError,
|
||||
SureThingID,
|
||||
SureProductID,
|
||||
)
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.const import (
|
||||
CONF_ID,
|
||||
CONF_NAME,
|
||||
CONF_PASSWORD,
|
||||
CONF_SCAN_INTERVAL,
|
||||
CONF_TYPE,
|
||||
|
|
@ -23,9 +23,11 @@ from homeassistant.helpers.dispatcher import async_dispatcher_send
|
|||
from homeassistant.helpers.event import async_track_time_interval
|
||||
|
||||
from .const import (
|
||||
CONF_FEEDERS,
|
||||
CONF_FLAPS,
|
||||
CONF_HOUSEHOLD_ID,
|
||||
CONF_PARENT,
|
||||
CONF_PETS,
|
||||
CONF_PRODUCT_ID,
|
||||
DATA_SURE_PETCARE,
|
||||
DEFAULT_SCAN_INTERVAL,
|
||||
DOMAIN,
|
||||
|
|
@ -36,23 +38,19 @@ from .const import (
|
|||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
FLAP_SCHEMA = vol.Schema(
|
||||
{vol.Required(CONF_ID): cv.positive_int, vol.Required(CONF_NAME): cv.string}
|
||||
)
|
||||
|
||||
PET_SCHEMA = vol.Schema(
|
||||
{vol.Required(CONF_ID): cv.positive_int, vol.Required(CONF_NAME): cv.string}
|
||||
)
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema(
|
||||
{
|
||||
DOMAIN: vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_USERNAME): cv.string,
|
||||
vol.Required(CONF_PASSWORD): cv.string,
|
||||
vol.Required(CONF_HOUSEHOLD_ID): cv.positive_int,
|
||||
vol.Required(CONF_FLAPS): vol.All(cv.ensure_list, [FLAP_SCHEMA]),
|
||||
vol.Required(CONF_PETS): vol.All(cv.ensure_list, [PET_SCHEMA]),
|
||||
vol.Optional(CONF_FEEDERS, default=[]): vol.All(
|
||||
cv.ensure_list, [cv.positive_int]
|
||||
),
|
||||
vol.Optional(CONF_FLAPS, default=[]): vol.All(
|
||||
cv.ensure_list, [cv.positive_int]
|
||||
),
|
||||
vol.Optional(CONF_PETS): vol.All(cv.ensure_list, [cv.positive_int]),
|
||||
vol.Optional(
|
||||
CONF_SCAN_INTERVAL, default=DEFAULT_SCAN_INTERVAL
|
||||
): cv.time_period,
|
||||
|
|
@ -63,7 +61,7 @@ CONFIG_SCHEMA = vol.Schema(
|
|||
)
|
||||
|
||||
|
||||
async def async_setup(hass, config):
|
||||
async def async_setup(hass, config) -> bool:
|
||||
"""Initialize the Sure Petcare component."""
|
||||
conf = config[DOMAIN]
|
||||
|
||||
|
|
@ -78,11 +76,10 @@ async def async_setup(hass, config):
|
|||
surepy = SurePetcare(
|
||||
conf[CONF_USERNAME],
|
||||
conf[CONF_PASSWORD],
|
||||
conf[CONF_HOUSEHOLD_ID],
|
||||
hass.loop,
|
||||
async_get_clientsession(hass),
|
||||
)
|
||||
await surepy.refresh_token()
|
||||
await surepy.get_data()
|
||||
except SurePetcareAuthenticationError:
|
||||
_LOGGER.error("Unable to connect to surepetcare.io: Wrong credentials!")
|
||||
return False
|
||||
|
|
@ -90,32 +87,44 @@ async def async_setup(hass, config):
|
|||
_LOGGER.error("Unable to connect to surepetcare.io: Wrong %s!", error)
|
||||
return False
|
||||
|
||||
# add flaps
|
||||
# add feeders
|
||||
things = [
|
||||
{
|
||||
CONF_NAME: flap[CONF_NAME],
|
||||
CONF_ID: flap[CONF_ID],
|
||||
CONF_TYPE: SureThingID.FLAP.name,
|
||||
}
|
||||
for flap in conf[CONF_FLAPS]
|
||||
{CONF_ID: feeder, CONF_TYPE: SureProductID.FEEDER}
|
||||
for feeder in conf[CONF_FEEDERS]
|
||||
]
|
||||
|
||||
# add pets
|
||||
# add flaps (don't differentiate between CAT and PET for now)
|
||||
things.extend(
|
||||
[
|
||||
{
|
||||
CONF_NAME: pet[CONF_NAME],
|
||||
CONF_ID: pet[CONF_ID],
|
||||
CONF_TYPE: SureThingID.PET.name,
|
||||
}
|
||||
for pet in conf[CONF_PETS]
|
||||
{CONF_ID: flap, CONF_TYPE: SureProductID.PET_FLAP}
|
||||
for flap in conf[CONF_FLAPS]
|
||||
]
|
||||
)
|
||||
|
||||
spc = hass.data[DATA_SURE_PETCARE][SPC] = SurePetcareAPI(
|
||||
hass, surepy, things, conf[CONF_HOUSEHOLD_ID]
|
||||
# discover hubs the flaps/feeders are connected to
|
||||
for device in things.copy():
|
||||
device_data = await surepy.device(device[CONF_ID])
|
||||
if (
|
||||
CONF_PARENT in device_data
|
||||
and device_data[CONF_PARENT][CONF_PRODUCT_ID] == SureProductID.HUB
|
||||
and device_data[CONF_PARENT][CONF_ID] not in things
|
||||
):
|
||||
things.append(
|
||||
{
|
||||
CONF_ID: device_data[CONF_PARENT][CONF_ID],
|
||||
CONF_TYPE: SureProductID.HUB,
|
||||
}
|
||||
)
|
||||
|
||||
# add pets
|
||||
things.extend(
|
||||
[{CONF_ID: pet, CONF_TYPE: SureProductID.PET} for pet in conf[CONF_PETS]]
|
||||
)
|
||||
|
||||
_LOGGER.debug("Devices and Pets to setup: %s", things)
|
||||
|
||||
spc = hass.data[DATA_SURE_PETCARE][SPC] = SurePetcareAPI(hass, surepy, things)
|
||||
|
||||
# initial update
|
||||
await spc.async_update()
|
||||
|
||||
|
|
@ -135,16 +144,18 @@ async def async_setup(hass, config):
|
|||
class SurePetcareAPI:
|
||||
"""Define a generic Sure Petcare object."""
|
||||
|
||||
def __init__(self, hass, surepy, ids, household_id):
|
||||
def __init__(self, hass, surepy: SurePetcare, ids: List[Dict[str, Any]]) -> None:
|
||||
"""Initialize the Sure Petcare object."""
|
||||
self.hass = hass
|
||||
self.surepy = surepy
|
||||
self.household_id = household_id
|
||||
self.ids = ids
|
||||
self.states = {}
|
||||
self.states: Dict[str, Any] = {}
|
||||
|
||||
async def async_update(self, args=None):
|
||||
async def async_update(self, arg: Any = None) -> None:
|
||||
"""Refresh Sure Petcare data."""
|
||||
|
||||
await self.surepy.get_data()
|
||||
|
||||
for thing in self.ids:
|
||||
sure_id = thing[CONF_ID]
|
||||
sure_type = thing[CONF_TYPE]
|
||||
|
|
@ -152,10 +163,15 @@ class SurePetcareAPI:
|
|||
try:
|
||||
type_state = self.states.setdefault(sure_type, {})
|
||||
|
||||
if sure_type == SureThingID.FLAP.name:
|
||||
type_state[sure_id] = await self.surepy.get_flap_data(sure_id)
|
||||
elif sure_type == SureThingID.PET.name:
|
||||
type_state[sure_id] = await self.surepy.get_pet_data(sure_id)
|
||||
if sure_type in [
|
||||
SureProductID.CAT_FLAP,
|
||||
SureProductID.PET_FLAP,
|
||||
SureProductID.FEEDER,
|
||||
SureProductID.HUB,
|
||||
]:
|
||||
type_state[sure_id] = await self.surepy.device(sure_id)
|
||||
elif sure_type == SureProductID.PET:
|
||||
type_state[sure_id] = await self.surepy.pet(sure_id)
|
||||
|
||||
except SurePetcareError as error:
|
||||
_LOGGER.error("Unable to retrieve data from surepetcare.io: %s", error)
|
||||
|
|
|
|||
|
|
@ -1,23 +1,28 @@
|
|||
"""Support for Sure PetCare Flaps/Pets binary sensors."""
|
||||
from datetime import datetime
|
||||
import logging
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from surepy import SureLocationID, SureLockStateID, SureThingID
|
||||
from surepy import SureLocationID, SureProductID
|
||||
|
||||
from homeassistant.components.binary_sensor import (
|
||||
DEVICE_CLASS_LOCK,
|
||||
DEVICE_CLASS_CONNECTIVITY,
|
||||
DEVICE_CLASS_PRESENCE,
|
||||
BinarySensorDevice,
|
||||
)
|
||||
from homeassistant.const import CONF_ID, CONF_NAME, CONF_TYPE
|
||||
from homeassistant.const import CONF_ID, CONF_TYPE
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
||||
|
||||
from .const import DATA_SURE_PETCARE, DEFAULT_DEVICE_CLASS, SPC, TOPIC_UPDATE
|
||||
from . import SurePetcareAPI
|
||||
from .const import DATA_SURE_PETCARE, SPC, TOPIC_UPDATE
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
|
||||
async def async_setup_platform(
|
||||
hass, config, async_add_entities, discovery_info=None
|
||||
) -> None:
|
||||
"""Set up Sure PetCare Flaps sensors based on a config entry."""
|
||||
if discovery_info is None:
|
||||
return
|
||||
|
|
@ -30,10 +35,20 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
|
|||
sure_id = thing[CONF_ID]
|
||||
sure_type = thing[CONF_TYPE]
|
||||
|
||||
if sure_type == SureThingID.FLAP.name:
|
||||
entity = Flap(sure_id, thing[CONF_NAME], spc)
|
||||
elif sure_type == SureThingID.PET.name:
|
||||
entity = Pet(sure_id, thing[CONF_NAME], spc)
|
||||
# connectivity
|
||||
if sure_type in [
|
||||
SureProductID.CAT_FLAP,
|
||||
SureProductID.PET_FLAP,
|
||||
SureProductID.FEEDER,
|
||||
]:
|
||||
entities.append(DeviceConnectivity(sure_id, sure_type, spc))
|
||||
|
||||
if sure_type == SureProductID.PET:
|
||||
entity = Pet(sure_id, spc)
|
||||
elif sure_type == SureProductID.HUB:
|
||||
entity = Hub(sure_id, spc)
|
||||
else:
|
||||
continue
|
||||
|
||||
entities.append(entity)
|
||||
|
||||
|
|
@ -44,57 +59,67 @@ class SurePetcareBinarySensor(BinarySensorDevice):
|
|||
"""A binary sensor implementation for Sure Petcare Entities."""
|
||||
|
||||
def __init__(
|
||||
self, _id: int, name: str, spc, device_class: str, sure_type: SureThingID
|
||||
self,
|
||||
_id: int,
|
||||
spc: SurePetcareAPI,
|
||||
device_class: str,
|
||||
sure_type: SureProductID,
|
||||
):
|
||||
"""Initialize a Sure Petcare binary sensor."""
|
||||
self._id = _id
|
||||
self._name = name
|
||||
self._spc = spc
|
||||
self._device_class = device_class
|
||||
self._sure_type = sure_type
|
||||
self._state = {}
|
||||
self._device_class = device_class
|
||||
|
||||
self._spc: SurePetcareAPI = spc
|
||||
self._spc_data: Dict[str, Any] = self._spc.states[self._sure_type].get(self._id)
|
||||
self._state: Dict[str, Any] = {}
|
||||
|
||||
# cover special case where a device has no name set
|
||||
if "name" in self._spc_data:
|
||||
name = self._spc_data["name"]
|
||||
else:
|
||||
name = f"Unnamed {self._sure_type.name.capitalize()}"
|
||||
|
||||
self._name = f"{self._sure_type.name.capitalize()} {name.capitalize()}"
|
||||
|
||||
self._async_unsub_dispatcher_connect = None
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
def is_on(self) -> Optional[bool]:
|
||||
"""Return true if entity is on/unlocked."""
|
||||
return bool(self._state)
|
||||
|
||||
@property
|
||||
def should_poll(self):
|
||||
def should_poll(self) -> bool:
|
||||
"""Return true."""
|
||||
return False
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
def name(self) -> str:
|
||||
"""Return the name of the device if any."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return the state attributes of the device."""
|
||||
return self._state
|
||||
|
||||
@property
|
||||
def device_class(self):
|
||||
def device_class(self) -> str:
|
||||
"""Return the device class."""
|
||||
return DEFAULT_DEVICE_CLASS if not self._device_class else self._device_class
|
||||
return None if not self._device_class else self._device_class
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
def unique_id(self: BinarySensorDevice) -> str:
|
||||
"""Return an unique ID."""
|
||||
return f"{self._spc.household_id}-{self._id}"
|
||||
return f"{self._spc_data['household_id']}-{self._id}"
|
||||
|
||||
async def async_update(self):
|
||||
async def async_update(self) -> None:
|
||||
"""Get the latest data and update the state."""
|
||||
self._state = self._spc.states[self._sure_type][self._id].get("data")
|
||||
self._spc_data = self._spc.states[self._sure_type].get(self._id)
|
||||
self._state = self._spc_data.get("status")
|
||||
_LOGGER.debug("%s -> self._state: %s", self._name, self._state)
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
async def async_added_to_hass(self) -> None:
|
||||
"""Register callbacks."""
|
||||
|
||||
@callback
|
||||
def update():
|
||||
def update() -> None:
|
||||
"""Update the state."""
|
||||
self.async_schedule_update_ha_state(True)
|
||||
|
||||
|
|
@ -102,54 +127,38 @@ class SurePetcareBinarySensor(BinarySensorDevice):
|
|||
self.hass, TOPIC_UPDATE, update
|
||||
)
|
||||
|
||||
async def async_will_remove_from_hass(self):
|
||||
async def async_will_remove_from_hass(self) -> None:
|
||||
"""Disconnect dispatcher listener when removed."""
|
||||
if self._async_unsub_dispatcher_connect:
|
||||
self._async_unsub_dispatcher_connect()
|
||||
|
||||
|
||||
class Flap(SurePetcareBinarySensor):
|
||||
"""Sure Petcare Flap."""
|
||||
class Hub(SurePetcareBinarySensor):
|
||||
"""Sure Petcare Pet."""
|
||||
|
||||
def __init__(self, _id: int, name: str, spc):
|
||||
"""Initialize a Sure Petcare Flap."""
|
||||
super().__init__(
|
||||
_id,
|
||||
f"Flap {name.capitalize()}",
|
||||
spc,
|
||||
DEVICE_CLASS_LOCK,
|
||||
SureThingID.FLAP.name,
|
||||
)
|
||||
def __init__(self, _id: int, spc: SurePetcareAPI) -> None:
|
||||
"""Initialize a Sure Petcare Hub."""
|
||||
super().__init__(_id, spc, DEVICE_CLASS_CONNECTIVITY, SureProductID.HUB)
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
"""Return true if entity is on/unlocked."""
|
||||
try:
|
||||
return bool(self._state["locking"]["mode"] == SureLockStateID.UNLOCKED)
|
||||
except (KeyError, TypeError):
|
||||
return None
|
||||
def available(self) -> bool:
|
||||
"""Return true if entity is available."""
|
||||
return bool(self._state["online"])
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
def is_on(self) -> bool:
|
||||
"""Return true if entity is online."""
|
||||
return self.available
|
||||
|
||||
@property
|
||||
def device_state_attributes(self) -> Optional[Dict[str, Any]]:
|
||||
"""Return the state attributes of the device."""
|
||||
attributes = None
|
||||
if self._state:
|
||||
try:
|
||||
attributes = {
|
||||
"battery_voltage": self._state["battery"] / 4,
|
||||
"locking_mode": self._state["locking"]["mode"],
|
||||
"device_rssi": self._state["signal"]["device_rssi"],
|
||||
"hub_rssi": self._state["signal"]["hub_rssi"],
|
||||
}
|
||||
|
||||
except (KeyError, TypeError) as error:
|
||||
_LOGGER.error(
|
||||
"Error getting device state attributes from %s: %s\n\n%s",
|
||||
self._name,
|
||||
error,
|
||||
self._state,
|
||||
)
|
||||
attributes = self._state
|
||||
attributes = {
|
||||
"led_mode": int(self._state["led_mode"]),
|
||||
"pairing_mode": bool(self._state["pairing_mode"]),
|
||||
}
|
||||
|
||||
return attributes
|
||||
|
||||
|
|
@ -157,20 +166,76 @@ class Flap(SurePetcareBinarySensor):
|
|||
class Pet(SurePetcareBinarySensor):
|
||||
"""Sure Petcare Pet."""
|
||||
|
||||
def __init__(self, _id: int, name: str, spc):
|
||||
def __init__(self, _id: int, spc: SurePetcareAPI) -> None:
|
||||
"""Initialize a Sure Petcare Pet."""
|
||||
super().__init__(
|
||||
_id,
|
||||
f"Pet {name.capitalize()}",
|
||||
spc,
|
||||
DEVICE_CLASS_PRESENCE,
|
||||
SureThingID.PET.name,
|
||||
)
|
||||
super().__init__(_id, spc, DEVICE_CLASS_PRESENCE, SureProductID.PET)
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
def is_on(self) -> bool:
|
||||
"""Return true if entity is at home."""
|
||||
try:
|
||||
return bool(self._state["where"] == SureLocationID.INSIDE)
|
||||
return bool(SureLocationID(self._state["where"]) == SureLocationID.INSIDE)
|
||||
except (KeyError, TypeError):
|
||||
return False
|
||||
|
||||
@property
|
||||
def device_state_attributes(self) -> Optional[Dict[str, Any]]:
|
||||
"""Return the state attributes of the device."""
|
||||
attributes = None
|
||||
if self._state:
|
||||
attributes = {
|
||||
"since": str(
|
||||
datetime.fromisoformat(self._state["since"]).replace(tzinfo=None)
|
||||
),
|
||||
"where": SureLocationID(self._state["where"]).name.capitalize(),
|
||||
}
|
||||
|
||||
return attributes
|
||||
|
||||
async def async_update(self) -> None:
|
||||
"""Get the latest data and update the state."""
|
||||
self._spc_data = self._spc.states[self._sure_type].get(self._id)
|
||||
self._state = self._spc_data.get("position")
|
||||
_LOGGER.debug("%s -> self._state: %s", self._name, self._state)
|
||||
|
||||
|
||||
class DeviceConnectivity(SurePetcareBinarySensor):
|
||||
"""Sure Petcare Pet."""
|
||||
|
||||
def __init__(
|
||||
self, _id: int, sure_type: SureProductID, spc: SurePetcareAPI,
|
||||
) -> None:
|
||||
"""Initialize a Sure Petcare Device."""
|
||||
super().__init__(_id, spc, DEVICE_CLASS_CONNECTIVITY, sure_type)
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
"""Return the name of the device if any."""
|
||||
return f"{self._name}_connectivity"
|
||||
|
||||
@property
|
||||
def unique_id(self: BinarySensorDevice) -> str:
|
||||
"""Return an unique ID."""
|
||||
return f"{self._spc_data['household_id']}-{self._id}-connectivity"
|
||||
|
||||
@property
|
||||
def available(self) -> bool:
|
||||
"""Return true if entity is available."""
|
||||
return bool(self._state)
|
||||
|
||||
@property
|
||||
def is_on(self) -> bool:
|
||||
"""Return true if entity is online."""
|
||||
return self.available
|
||||
|
||||
@property
|
||||
def device_state_attributes(self) -> Optional[Dict[str, Any]]:
|
||||
"""Return the state attributes of the device."""
|
||||
attributes = None
|
||||
if self._state:
|
||||
attributes = {
|
||||
"device_rssi": f'{self._state["signal"]["device_rssi"]:.2f}',
|
||||
"hub_rssi": f'{self._state["signal"]["hub_rssi"]:.2f}',
|
||||
}
|
||||
|
||||
return attributes
|
||||
|
|
|
|||
|
|
@ -11,8 +11,11 @@ SPC = "spc"
|
|||
SUREPY = "surepy"
|
||||
|
||||
CONF_HOUSEHOLD_ID = "household_id"
|
||||
CONF_FEEDERS = "feeders"
|
||||
CONF_FLAPS = "flaps"
|
||||
CONF_PARENT = "parent"
|
||||
CONF_PETS = "pets"
|
||||
CONF_PRODUCT_ID = "product_id"
|
||||
CONF_DATA = "data"
|
||||
|
||||
SURE_IDS = "sure_ids"
|
||||
|
|
|
|||
|
|
@ -4,5 +4,5 @@
|
|||
"documentation": "https://www.home-assistant.io/integrations/surepetcare",
|
||||
"dependencies": [],
|
||||
"codeowners": ["@benleb"],
|
||||
"requirements": ["surepy==0.1.10"]
|
||||
"requirements": ["surepy==0.2.3"]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,19 +1,15 @@
|
|||
"""Support for Sure PetCare Flaps/Pets sensors."""
|
||||
import logging
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from surepy import SureThingID
|
||||
from surepy import SureLockStateID, SureProductID
|
||||
|
||||
from homeassistant.const import (
|
||||
ATTR_VOLTAGE,
|
||||
CONF_ID,
|
||||
CONF_NAME,
|
||||
CONF_TYPE,
|
||||
DEVICE_CLASS_BATTERY,
|
||||
)
|
||||
from homeassistant.const import ATTR_VOLTAGE, CONF_ID, CONF_TYPE, DEVICE_CLASS_BATTERY
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
||||
from homeassistant.helpers.entity import Entity
|
||||
|
||||
from . import SurePetcareAPI
|
||||
from .const import (
|
||||
DATA_SURE_PETCARE,
|
||||
SPC,
|
||||
|
|
@ -30,97 +26,82 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
|
|||
if discovery_info is None:
|
||||
return
|
||||
|
||||
entities = []
|
||||
|
||||
spc = hass.data[DATA_SURE_PETCARE][SPC]
|
||||
async_add_entities(
|
||||
[
|
||||
FlapBattery(entity[CONF_ID], entity[CONF_NAME], spc)
|
||||
for entity in spc.ids
|
||||
if entity[CONF_TYPE] == SureThingID.FLAP.name
|
||||
],
|
||||
True,
|
||||
)
|
||||
|
||||
for entity in spc.ids:
|
||||
sure_type = entity[CONF_TYPE]
|
||||
|
||||
if sure_type in [
|
||||
SureProductID.CAT_FLAP,
|
||||
SureProductID.PET_FLAP,
|
||||
SureProductID.FEEDER,
|
||||
]:
|
||||
entities.append(SureBattery(entity[CONF_ID], sure_type, spc))
|
||||
|
||||
if sure_type in [
|
||||
SureProductID.CAT_FLAP,
|
||||
SureProductID.PET_FLAP,
|
||||
]:
|
||||
entities.append(Flap(entity[CONF_ID], sure_type, spc))
|
||||
|
||||
async_add_entities(entities, True)
|
||||
|
||||
|
||||
class FlapBattery(Entity):
|
||||
"""Sure Petcare Flap."""
|
||||
class SurePetcareSensor(Entity):
|
||||
"""A binary sensor implementation for Sure Petcare Entities."""
|
||||
|
||||
def __init__(
|
||||
self, _id: int, sure_type: SureProductID, spc: SurePetcareAPI,
|
||||
):
|
||||
"""Initialize a Sure Petcare sensor."""
|
||||
|
||||
def __init__(self, _id: int, name: str, spc):
|
||||
"""Initialize a Sure Petcare Flap battery sensor."""
|
||||
self._id = _id
|
||||
self._name = f"Flap {name.capitalize()} Battery Level"
|
||||
self._sure_type = sure_type
|
||||
|
||||
self._spc = spc
|
||||
self._state = self._spc.states[SureThingID.FLAP.name][self._id].get("data")
|
||||
self._spc_data: Dict[str, Any] = self._spc.states[self._sure_type].get(self._id)
|
||||
self._state: Dict[str, Any] = {}
|
||||
|
||||
self._name = (
|
||||
f"{self._sure_type.name.capitalize()} "
|
||||
f"{self._spc_data['name'].capitalize()}"
|
||||
)
|
||||
|
||||
self._async_unsub_dispatcher_connect = None
|
||||
|
||||
@property
|
||||
def should_poll(self):
|
||||
"""Return true."""
|
||||
return False
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
def name(self) -> str:
|
||||
"""Return the name of the device if any."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
"""Return battery level in percent."""
|
||||
try:
|
||||
per_battery_voltage = self._state["battery"] / 4
|
||||
voltage_diff = per_battery_voltage - SURE_BATT_VOLTAGE_LOW
|
||||
battery_percent = int(voltage_diff / SURE_BATT_VOLTAGE_DIFF * 100)
|
||||
except (KeyError, TypeError):
|
||||
battery_percent = None
|
||||
|
||||
return battery_percent
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
def unique_id(self) -> str:
|
||||
"""Return an unique ID."""
|
||||
return f"{self._spc.household_id}-{self._id}"
|
||||
return f"{self._spc_data['household_id']}-{self._id}"
|
||||
|
||||
@property
|
||||
def device_class(self):
|
||||
"""Return the device class."""
|
||||
return DEVICE_CLASS_BATTERY
|
||||
def available(self) -> bool:
|
||||
"""Return true if entity is available."""
|
||||
return bool(self._state)
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return state attributes."""
|
||||
attributes = None
|
||||
if self._state:
|
||||
try:
|
||||
voltage_per_battery = float(self._state["battery"]) / 4
|
||||
attributes = {
|
||||
ATTR_VOLTAGE: f"{float(self._state['battery']):.2f}",
|
||||
f"{ATTR_VOLTAGE}_per_battery": f"{voltage_per_battery:.2f}",
|
||||
}
|
||||
except (KeyError, TypeError) as error:
|
||||
attributes = self._state
|
||||
_LOGGER.error(
|
||||
"Error getting device state attributes from %s: %s\n\n%s",
|
||||
self._name,
|
||||
error,
|
||||
self._state,
|
||||
)
|
||||
def should_poll(self) -> bool:
|
||||
"""Return true."""
|
||||
return False
|
||||
|
||||
return attributes
|
||||
|
||||
@property
|
||||
def unit_of_measurement(self):
|
||||
"""Return the unit of measurement."""
|
||||
return "%"
|
||||
|
||||
async def async_update(self):
|
||||
async def async_update(self) -> None:
|
||||
"""Get the latest data and update the state."""
|
||||
self._state = self._spc.states[SureThingID.FLAP.name][self._id].get("data")
|
||||
self._spc_data = self._spc.states[self._sure_type].get(self._id)
|
||||
self._state = self._spc_data.get("status")
|
||||
_LOGGER.debug("%s -> self._state: %s", self._name, self._state)
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
async def async_added_to_hass(self) -> None:
|
||||
"""Register callbacks."""
|
||||
|
||||
@callback
|
||||
def update():
|
||||
def update() -> None:
|
||||
"""Update the state."""
|
||||
self.async_schedule_update_ha_state(True)
|
||||
|
||||
|
|
@ -128,7 +109,77 @@ class FlapBattery(Entity):
|
|||
self.hass, TOPIC_UPDATE, update
|
||||
)
|
||||
|
||||
async def async_will_remove_from_hass(self):
|
||||
async def async_will_remove_from_hass(self) -> None:
|
||||
"""Disconnect dispatcher listener when removed."""
|
||||
if self._async_unsub_dispatcher_connect:
|
||||
self._async_unsub_dispatcher_connect()
|
||||
|
||||
|
||||
class Flap(SurePetcareSensor):
|
||||
"""Sure Petcare Flap."""
|
||||
|
||||
@property
|
||||
def state(self) -> Optional[int]:
|
||||
"""Return battery level in percent."""
|
||||
return SureLockStateID(self._state["locking"]["mode"]).name.capitalize()
|
||||
|
||||
@property
|
||||
def device_state_attributes(self) -> Optional[Dict[str, Any]]:
|
||||
"""Return the state attributes of the device."""
|
||||
attributes = None
|
||||
if self._state:
|
||||
attributes = {
|
||||
"learn_mode": bool(self._state["learn_mode"]),
|
||||
}
|
||||
|
||||
return attributes
|
||||
|
||||
|
||||
class SureBattery(SurePetcareSensor):
|
||||
"""Sure Petcare Flap."""
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
"""Return the name of the device if any."""
|
||||
return f"{self._name} Battery Level"
|
||||
|
||||
@property
|
||||
def state(self) -> Optional[int]:
|
||||
"""Return battery level in percent."""
|
||||
battery_percent: Optional[int]
|
||||
try:
|
||||
per_battery_voltage = self._state["battery"] / 4
|
||||
voltage_diff = per_battery_voltage - SURE_BATT_VOLTAGE_LOW
|
||||
battery_percent = min(int(voltage_diff / SURE_BATT_VOLTAGE_DIFF * 100), 100)
|
||||
except (KeyError, TypeError):
|
||||
battery_percent = None
|
||||
|
||||
return battery_percent
|
||||
|
||||
@property
|
||||
def unique_id(self) -> str:
|
||||
"""Return an unique ID."""
|
||||
return f"{self._spc_data['household_id']}-{self._id}-battery"
|
||||
|
||||
@property
|
||||
def device_class(self) -> str:
|
||||
"""Return the device class."""
|
||||
return DEVICE_CLASS_BATTERY
|
||||
|
||||
@property
|
||||
def device_state_attributes(self) -> Optional[Dict[str, Any]]:
|
||||
"""Return state attributes."""
|
||||
attributes = None
|
||||
if self._state:
|
||||
voltage_per_battery = float(self._state["battery"]) / 4
|
||||
attributes = {
|
||||
ATTR_VOLTAGE: f"{float(self._state['battery']):.2f}",
|
||||
f"{ATTR_VOLTAGE}_per_battery": f"{voltage_per_battery:.2f}",
|
||||
}
|
||||
|
||||
return attributes
|
||||
|
||||
@property
|
||||
def unit_of_measurement(self) -> str:
|
||||
"""Return the unit of measurement."""
|
||||
return "%"
|
||||
|
|
|
|||
|
|
@ -1928,7 +1928,7 @@ sucks==0.9.4
|
|||
sunwatcher==0.2.1
|
||||
|
||||
# homeassistant.components.surepetcare
|
||||
surepy==0.1.10
|
||||
surepy==0.2.3
|
||||
|
||||
# homeassistant.components.swiss_hydrological_data
|
||||
swisshydrodata==0.0.3
|
||||
|
|
|
|||
Loading…
Reference in New Issue