Skip to content

Commit

Permalink
rewrite sensor platform to use SensorEntityDescription
Browse files Browse the repository at this point in the history
add pending_requests as a new sensor (#33)
  • Loading branch information
pantherale0 authored Nov 8, 2024
1 parent 6c2659c commit ee0158a
Show file tree
Hide file tree
Showing 2 changed files with 107 additions and 96 deletions.
2 changes: 1 addition & 1 deletion custom_components/family_safety/entity_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ def _account(self) -> Account:
@property
def unique_id(self) -> str:
"""Return a unique ID for the entity."""
return f"familysafety_{self._account_id}_{self._entity_id}"
return f"{self._account_id}_{self._entity_id}"

@property
def device_info(self):
Expand Down
201 changes: 106 additions & 95 deletions custom_components/family_safety/sensor.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,17 @@
"""Sensors for family safety."""

from collections.abc import Mapping
from collections.abc import Mapping, Callable
from dataclasses import dataclass
from datetime import datetime
import logging
from typing import Any
from typing import Generic, Any

import voluptuous as vol

from pyfamilysafety import Account
from pyfamilysafety.application import Application

from homeassistant.components.sensor import SensorEntity, SensorDeviceClass
from homeassistant.components.sensor import SensorEntity, SensorEntityDescription, SensorDeviceClass
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import (
Expand All @@ -20,11 +23,48 @@

from .const import DOMAIN

from .entity_base import ManagedAccountEntity, ApplicationEntity
from .entity_base import ManagedAccountEntity

_LOGGER = logging.getLogger(__name__)


@dataclass(frozen=True, kw_only=True)
class FamilySafetySensorEntityDescription(SensorEntityDescription):
"""Describes family_safety sensor entity."""
value_fn: Callable[[ManagedAccountEntity], str | int | datetime]
name_fn: Callable[[ManagedAccountEntity], str]
native_unit_of_measurement_fn: Callable[[ManagedAccountEntity], str]


GEN_SENSORS: dict[str, FamilySafetySensorEntityDescription] = {
"account_balance": FamilySafetySensorEntityDescription(
key="account_balance",
value_fn=lambda data: data._account.account_balance,
device_class=SensorDeviceClass.MONETARY,
name_fn=lambda data: f"{data._account.first_name} Available Balance",
native_unit_of_measurement_fn=lambda data: data._account.account_currency,
),
"pending_requests": FamilySafetySensorEntityDescription(
key="pending_requests",
value_fn=lambda data: len(
[d for d in data.coordinator.api.pending_requests if d["puid"] == data._account_id]),
name_fn=lambda data: f"{data._account.first_name} Pending Requests",
native_unit_of_measurement_fn=lambda data: None
)
}

TIME_SENSORS: dict[str, FamilySafetySensorEntityDescription] = {
"screentime": FamilySafetySensorEntityDescription(
key="screentime",
value_fn=lambda data: (
data._account.today_screentime_usage / 1000) / 60,
device_class=SensorDeviceClass.DURATION,
native_unit_of_measurement_fn=lambda data: "min",
name_fn=lambda data: f"{data._account.first_name} Used Screen Time"
)
}


async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
Expand All @@ -37,29 +77,27 @@ async def async_setup_entry(
if (account.user_id in config_entry.options.get("accounts", [])) or (
len(config_entry.options.get("accounts", [])) == 0
):
entities.append(
AccountScreentimeSensor(
for app in config_entry.options.get("tracked_applications", []):
entities.append(ScreentimeSensor(
coordinator=hass.data[DOMAIN][config_entry.entry_id],
description=FamilySafetySensorEntityDescription(
key=app,
device_class=SensorDeviceClass.DURATION,
native_unit_of_measurement_fn=lambda data: "min",
value_fn=lambda data: data._account.get_application(
app).usage,
name_fn=lambda data: f"{data._account.first_name} {data._account.get_application(app).name} Used Screen Time"),
idx=None,
account_id=account.user_id,
)
account_id=account.user_id
))
entities.extend(
[ScreentimeSensor(coordinator=hass.data[DOMAIN]
[config_entry.entry_id], idx=None, account_id=account.user_id, description=desc) for desc in TIME_SENSORS.values()]
)
entities.append(
AccountBalanceSensor(
coordinator=hass.data[DOMAIN][config_entry.entry_id],
idx=None,
account_id=account.user_id,
)
entities.extend(
[GenericSensor(coordinator=hass.data[DOMAIN]
[config_entry.entry_id], idx=None, account_id=account.user_id, description=desc) for desc in GEN_SENSORS.values()]
)
for app in config_entry.options.get("tracked_applications", []):
entities.append(
ApplicationScreentimeSensor(
coordinator=hass.data[DOMAIN][config_entry.entry_id],
idx=None,
account_id=account.user_id,
app_id=app,
)
)

async_add_entities(entities, True)
# register services
Expand All @@ -76,100 +114,73 @@ async def async_setup_entry(
)


class AccountBalanceSensor(ManagedAccountEntity, SensorEntity):
"""A balance sensor for the account."""
class GenericSensor(ManagedAccountEntity, SensorEntity):
"""Generic Sensor."""

def __init__(self, coordinator: FamilySafetyCoordinator, idx, account_id) -> None:
"""Account Balance Sensor."""
super().__init__(coordinator, idx, account_id, "balance")
def __init__(self, coordinator: FamilySafetyCoordinator, description: FamilySafetySensorEntityDescription, idx, account_id) -> None:
"""Generic Sensor."""
super().__init__(coordinator, idx, account_id, description.key)
self.entity_description = description

@property
def name(self) -> str:
"""Return name of entity."""
return f"{self._account.first_name} Available Balance"

@property
def native_value(self) -> float:
"""Return balance."""
return self._account.account_balance

@property
def native_unit_of_measurement(self) -> str | None:
"""Return unit of measurement."""
return self._account.account_currency

@property
def device_class(self) -> SensorDeviceClass | None:
"""Return device class."""
return SensorDeviceClass.MONETARY


class AccountScreentimeSensor(ManagedAccountEntity, SensorEntity):
"""Aggregate screentime sensor."""

def __init__(self, coordinator: FamilySafetyCoordinator, idx, account_id) -> None:
"""Screentime Sensor."""
super().__init__(coordinator, idx, account_id, "screentime")
return self.entity_description.name_fn(self)

@property
def name(self) -> str:
"""Return entity name."""
return f"{self._account.first_name} Used Screen Time"

@property
def native_value(self) -> float:
"""Return duration (minutes)."""
return (self._account.today_screentime_usage / 1000) / 60
def native_value(self):
"""Return the native value of the entity."""
return self.entity_description.value_fn(self)

@property
def native_unit_of_measurement(self) -> str | None:
"""Return unit of measurement."""
return "min"
def native_unit_of_measurement(self):
"""Return UOM."""
return self.entity_description.native_unit_of_measurement_fn(self)

@property
def device_class(self) -> SensorDeviceClass | None:
def device_class(self):
"""Return device class."""
return SensorDeviceClass.DURATION
return self.entity_description.device_class

@property
def extra_state_attributes(self) -> Mapping[str, Any] | None:
"""Return additional state attributes."""
devices = {}
for device in self._account.devices:
if device.today_time_used:
devices[device.device_name] = (device.today_time_used / 1000) / 60
else:
devices[device.device_name] = 0
applications = {}
for app in self._account.applications:
applications[app.name] = app.usage
return {"application_usage": applications, "device_usage": devices}

if self.entity_description.key == "pending_requests":
return {
"requests": [d for d in self.coordinator.api.pending_requests if d["puid"] == self._account_id]
}

class ApplicationScreentimeSensor(ApplicationEntity, SensorEntity):
"""Application specific screentime sensor."""

@property
def name(self) -> str:
"""Return entity name."""
return f"{self._account.first_name} {self._application.name} Used Screen Time"

@property
def native_value(self) -> float:
"""Return duration (minutes)."""
return self._application.usage
class ScreentimeSensor(GenericSensor, SensorEntity):
"""Aggregate screentime sensor."""

@property
def native_unit_of_measurement(self) -> str | None:
"""Return native unit of measurement."""
return "min"
def __init__(self, coordinator: FamilySafetyCoordinator, description: FamilySafetySensorEntityDescription, idx, account_id) -> None:
"""Screentime Sensor."""
super().__init__(coordinator, description, idx, account_id)
if description.key == "screentime":
self.app_id = None
else:
self.app_id = description.key

@property
def device_class(self) -> SensorDeviceClass | None:
"""Return device class."""
return SensorDeviceClass.DURATION
def _application(self) -> Application:
"""Get the application."""
return self._account.get_application(self.app_id)

@property
def extra_state_attributes(self) -> Mapping[str, Any] | None:
"""Return additional state attributes."""
return {"blocked": self._application.blocked}
if self.entity_description.key == "screentime":
devices = {}
for device in self._account.devices:
if device.today_time_used:
devices[device.device_name] = (
device.today_time_used / 1000) / 60
else:
devices[device.device_name] = 0
applications = {}
for app in self._account.applications:
applications[app.name] = app.usage
return {"application_usage": applications, "device_usage": devices}
elif self.app_id is not None:
return {"blocked": self._application.blocked}

0 comments on commit ee0158a

Please sign in to comment.