Rewrite fitbit sensor API response value parsing (#100782)

* Cleanup fitbit sensor API parsing

* Remove API code that is not used yet

* Remove dead code for battery levels

Small API parsing cleanup

* Address PR feedback

* Update homeassistant/components/fitbit/sensor.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

---------

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
pull/100826/head
Allen Porter 2023-09-24 13:37:48 -07:00 committed by GitHub
parent 5549f697cf
commit 66ebb479ea
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 198 additions and 81 deletions

View File

@ -0,0 +1,65 @@
"""API for fitbit bound to Home Assistant OAuth."""
import logging
from typing import Any
from fitbit import Fitbit
from homeassistant.core import HomeAssistant
from .model import FitbitDevice, FitbitProfile
_LOGGER = logging.getLogger(__name__)
class FitbitApi:
"""Fitbit client library wrapper base class."""
def __init__(
self,
hass: HomeAssistant,
client: Fitbit,
) -> None:
"""Initialize Fitbit auth."""
self._hass = hass
self._profile: FitbitProfile | None = None
self._client = client
@property
def client(self) -> Fitbit:
"""Property to expose the underlying client library."""
return self._client
def get_user_profile(self) -> FitbitProfile:
"""Return the user profile from the API."""
response: dict[str, Any] = self._client.user_profile_get()
_LOGGER.debug("user_profile_get=%s", response)
profile = response["user"]
return FitbitProfile(
encoded_id=profile["encodedId"],
full_name=profile["fullName"],
locale=profile.get("locale"),
)
def get_devices(self) -> list[FitbitDevice]:
"""Return available devices."""
devices: list[dict[str, str]] = self._client.get_devices()
_LOGGER.debug("get_devices=%s", devices)
return [
FitbitDevice(
id=device["id"],
device_version=device["deviceVersion"],
battery_level=int(device["batteryLevel"]),
battery=device["battery"],
type=device["type"],
)
for device in devices
]
def get_latest_time_series(self, resource_type: str) -> dict[str, Any]:
"""Return the most recent value from the time series for the specified resource type."""
response: dict[str, Any] = self._client.time_series(resource_type, period="7d")
_LOGGER.debug("time_series(%s)=%s", resource_type, response)
key = resource_type.replace("/", "-")
dated_results: list[dict[str, Any]] = response[key]
return dated_results[-1]

View File

@ -0,0 +1,37 @@
"""Data representation for fitbit API responses."""
from dataclasses import dataclass
@dataclass
class FitbitProfile:
"""User profile from the Fitbit API response."""
encoded_id: str
"""The ID representing the Fitbit user."""
full_name: str
"""The first name value specified in the user's account settings."""
locale: str | None
"""The locale defined in the user's Fitbit account settings."""
@dataclass
class FitbitDevice:
"""Device from the Fitbit API response."""
id: str
"""The device ID."""
device_version: str
"""The product name of the device."""
battery_level: int
"""The battery level as a percentage."""
battery: str
"""Returns the battery level of the device."""
type: str
"""The type of the device such as TRACKER or SCALE."""

View File

@ -1,6 +1,7 @@
"""Support for the Fitbit API."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
import datetime
import logging
@ -40,6 +41,7 @@ from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util.json import load_json_object
from homeassistant.util.unit_system import METRIC_SYSTEM
from .api import FitbitApi
from .const import (
ATTR_ACCESS_TOKEN,
ATTR_LAST_SAVED_AT,
@ -56,6 +58,7 @@ from .const import (
FITBIT_DEFAULT_RESOURCES,
FITBIT_MEASUREMENTS,
)
from .model import FitbitDevice, FitbitProfile
_LOGGER: Final = logging.getLogger(__name__)
@ -64,11 +67,42 @@ _CONFIGURING: dict[str, str] = {}
SCAN_INTERVAL: Final = datetime.timedelta(minutes=30)
def _default_value_fn(result: dict[str, Any]) -> str:
"""Parse a Fitbit timeseries API responses."""
return cast(str, result["value"])
def _distance_value_fn(result: dict[str, Any]) -> int | str:
"""Format function for distance values."""
return format(float(_default_value_fn(result)), ".2f")
def _body_value_fn(result: dict[str, Any]) -> int | str:
"""Format function for body values."""
return format(float(_default_value_fn(result)), ".1f")
def _clock_format_12h(result: dict[str, Any]) -> str:
raw_state = result["value"]
if raw_state == "":
return "-"
hours_str, minutes_str = raw_state.split(":")
hours, minutes = int(hours_str), int(minutes_str)
setting = "AM"
if hours > 12:
setting = "PM"
hours -= 12
elif hours == 0:
hours = 12
return f"{hours}:{minutes:02d} {setting}"
@dataclass
class FitbitSensorEntityDescription(SensorEntityDescription):
"""Describes Fitbit sensor entity."""
unit_type: str | None = None
value_fn: Callable[[dict[str, Any]], Any] = _default_value_fn
FITBIT_RESOURCES_LIST: Final[tuple[FitbitSensorEntityDescription, ...]] = (
@ -96,6 +130,7 @@ FITBIT_RESOURCES_LIST: Final[tuple[FitbitSensorEntityDescription, ...]] = (
unit_type="distance",
icon="mdi:map-marker",
device_class=SensorDeviceClass.DISTANCE,
value_fn=_distance_value_fn,
),
FitbitSensorEntityDescription(
key="activities/elevation",
@ -115,6 +150,7 @@ FITBIT_RESOURCES_LIST: Final[tuple[FitbitSensorEntityDescription, ...]] = (
name="Resting Heart Rate",
native_unit_of_measurement="bpm",
icon="mdi:heart-pulse",
value_fn=lambda result: int(result["value"]["restingHeartRate"]),
),
FitbitSensorEntityDescription(
key="activities/minutesFairlyActive",
@ -168,6 +204,7 @@ FITBIT_RESOURCES_LIST: Final[tuple[FitbitSensorEntityDescription, ...]] = (
unit_type="distance",
icon="mdi:map-marker",
device_class=SensorDeviceClass.DISTANCE,
value_fn=_distance_value_fn,
),
FitbitSensorEntityDescription(
key="activities/tracker/elevation",
@ -222,6 +259,7 @@ FITBIT_RESOURCES_LIST: Final[tuple[FitbitSensorEntityDescription, ...]] = (
native_unit_of_measurement="BMI",
icon="mdi:human",
state_class=SensorStateClass.MEASUREMENT,
value_fn=_body_value_fn,
),
FitbitSensorEntityDescription(
key="body/fat",
@ -229,6 +267,7 @@ FITBIT_RESOURCES_LIST: Final[tuple[FitbitSensorEntityDescription, ...]] = (
native_unit_of_measurement=PERCENTAGE,
icon="mdi:human",
state_class=SensorStateClass.MEASUREMENT,
value_fn=_body_value_fn,
),
FitbitSensorEntityDescription(
key="body/weight",
@ -237,6 +276,7 @@ FITBIT_RESOURCES_LIST: Final[tuple[FitbitSensorEntityDescription, ...]] = (
icon="mdi:human",
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.WEIGHT,
value_fn=_body_value_fn,
),
FitbitSensorEntityDescription(
key="sleep/awakeningsCount",
@ -279,11 +319,6 @@ FITBIT_RESOURCES_LIST: Final[tuple[FitbitSensorEntityDescription, ...]] = (
icon="mdi:sleep",
device_class=SensorDeviceClass.DURATION,
),
FitbitSensorEntityDescription(
key="sleep/startTime",
name="Sleep Start Time",
icon="mdi:clock",
),
FitbitSensorEntityDescription(
key="sleep/timeInBed",
name="Sleep Time in Bed",
@ -293,6 +328,19 @@ FITBIT_RESOURCES_LIST: Final[tuple[FitbitSensorEntityDescription, ...]] = (
),
)
# Different description depending on clock format
SLEEP_START_TIME = FitbitSensorEntityDescription(
key="sleep/startTime",
name="Sleep Start Time",
icon="mdi:clock",
)
SLEEP_START_TIME_12HR = FitbitSensorEntityDescription(
key="sleep/startTime",
name="Sleep Start Time",
icon="mdi:clock",
value_fn=_clock_format_12h,
)
FITBIT_RESOURCE_BATTERY = FitbitSensorEntityDescription(
key="devices/battery",
name="Battery",
@ -300,7 +348,8 @@ FITBIT_RESOURCE_BATTERY = FitbitSensorEntityDescription(
)
FITBIT_RESOURCES_KEYS: Final[list[str]] = [
desc.key for desc in (*FITBIT_RESOURCES_LIST, FITBIT_RESOURCE_BATTERY)
desc.key
for desc in (*FITBIT_RESOURCES_LIST, FITBIT_RESOURCE_BATTERY, SLEEP_START_TIME)
]
PLATFORM_SCHEMA: Final = PARENT_PLATFORM_SCHEMA.extend(
@ -438,9 +487,10 @@ def setup_platform(
if int(time.time()) - cast(int, expires_at) > 3600:
authd_client.client.refresh_token()
user_profile = authd_client.user_profile_get()["user"]
api = FitbitApi(hass, authd_client)
user_profile = api.get_user_profile()
if (unit_system := config[CONF_UNIT_SYSTEM]) == "default":
authd_client.system = user_profile["locale"]
authd_client.system = user_profile.locale
if authd_client.system != "en_GB":
if hass.config.units is METRIC_SYSTEM:
authd_client.system = "metric"
@ -449,34 +499,38 @@ def setup_platform(
else:
authd_client.system = unit_system
registered_devs = authd_client.get_devices()
clock_format = config[CONF_CLOCK_FORMAT]
monitored_resources = config[CONF_MONITORED_RESOURCES]
resource_list = [
*FITBIT_RESOURCES_LIST,
SLEEP_START_TIME_12HR if clock_format == "12H" else SLEEP_START_TIME,
]
entities = [
FitbitSensor(
authd_client,
api,
user_profile,
config_path,
description,
hass.config.units is METRIC_SYSTEM,
clock_format,
)
for description in FITBIT_RESOURCES_LIST
for description in resource_list
if description.key in monitored_resources
]
if "devices/battery" in monitored_resources:
devices = api.get_devices()
entities.extend(
[
FitbitSensor(
authd_client,
api,
user_profile,
config_path,
FITBIT_RESOURCE_BATTERY,
hass.config.units is METRIC_SYSTEM,
clock_format,
dev_extra,
device,
)
for dev_extra in registered_devs
for device in devices
]
)
add_entities(entities, True)
@ -591,30 +645,30 @@ class FitbitSensor(SensorEntity):
def __init__(
self,
client: Fitbit,
user_profile: dict[str, Any],
api: FitbitApi,
user_profile: FitbitProfile,
config_path: str,
description: FitbitSensorEntityDescription,
is_metric: bool,
clock_format: str,
extra: dict[str, str] | None = None,
device: FitbitDevice | None = None,
) -> None:
"""Initialize the Fitbit sensor."""
self.entity_description = description
self.client = client
self.api = api
self.config_path = config_path
self.is_metric = is_metric
self.clock_format = clock_format
self.extra = extra
self.device = device
self._attr_unique_id = f"{user_profile['encodedId']}_{description.key}"
if self.extra is not None:
self._attr_name = f"{self.extra.get('deviceVersion')} Battery"
self._attr_unique_id = f"{self._attr_unique_id}_{self.extra.get('id')}"
self._attr_unique_id = f"{user_profile.encoded_id}_{description.key}"
if device is not None:
self._attr_name = f"{device.device_version} Battery"
self._attr_unique_id = f"{self._attr_unique_id}_{device.id}"
if description.unit_type:
try:
measurement_system = FITBIT_MEASUREMENTS[self.client.system]
measurement_system = FITBIT_MEASUREMENTS[self.api.client.system]
except KeyError:
if self.is_metric:
measurement_system = FITBIT_MEASUREMENTS["metric"]
@ -629,9 +683,8 @@ class FitbitSensor(SensorEntity):
"""Icon to use in the frontend, if any."""
if (
self.entity_description.key == "devices/battery"
and self.extra is not None
and (extra_battery := self.extra.get("battery")) is not None
and (battery_level := BATTERY_LEVELS.get(extra_battery)) is not None
and self.device is not None
and (battery_level := BATTERY_LEVELS.get(self.device.battery)) is not None
):
return icon_for_battery_level(battery_level=battery_level)
return self.entity_description.icon
@ -641,72 +694,34 @@ class FitbitSensor(SensorEntity):
"""Return the state attributes."""
attrs: dict[str, str | None] = {}
if self.extra is not None:
attrs["model"] = self.extra.get("deviceVersion")
extra_type = self.extra.get("type")
attrs["type"] = extra_type.lower() if extra_type is not None else None
if self.device is not None:
attrs["model"] = self.device.device_version
device_type = self.device.type
attrs["type"] = device_type.lower() if device_type is not None else None
return attrs
def update(self) -> None:
"""Get the latest data from the Fitbit API and update the states."""
resource_type = self.entity_description.key
if resource_type == "devices/battery" and self.extra is not None:
registered_devs: list[dict[str, Any]] = self.client.get_devices()
device_id = self.extra.get("id")
self.extra = list(
filter(lambda device: device.get("id") == device_id, registered_devs)
)[0]
self._attr_native_value = self.extra.get("battery")
if resource_type == "devices/battery" and self.device is not None:
device_id = self.device.id
registered_devs: list[FitbitDevice] = self.api.get_devices()
self.device = next(
device for device in registered_devs if device.id == device_id
)
self._attr_native_value = self.device.battery
else:
container = resource_type.replace("/", "-")
response = self.client.time_series(resource_type, period="7d")
raw_state = response[container][-1].get("value")
if resource_type == "activities/distance":
self._attr_native_value = format(float(raw_state), ".2f")
elif resource_type == "activities/tracker/distance":
self._attr_native_value = format(float(raw_state), ".2f")
elif resource_type == "body/bmi":
self._attr_native_value = format(float(raw_state), ".1f")
elif resource_type == "body/fat":
self._attr_native_value = format(float(raw_state), ".1f")
elif resource_type == "body/weight":
self._attr_native_value = format(float(raw_state), ".1f")
elif resource_type == "sleep/startTime":
if raw_state == "":
self._attr_native_value = "-"
elif self.clock_format == "12H":
hours, minutes = raw_state.split(":")
hours, minutes = int(hours), int(minutes)
setting = "AM"
if hours > 12:
setting = "PM"
hours -= 12
elif hours == 0:
hours = 12
self._attr_native_value = f"{hours}:{minutes:02d} {setting}"
else:
self._attr_native_value = raw_state
elif self.is_metric:
self._attr_native_value = raw_state
else:
try:
self._attr_native_value = int(raw_state)
except TypeError:
self._attr_native_value = raw_state
result = self.api.get_latest_time_series(resource_type)
self._attr_native_value = self.entity_description.value_fn(result)
if resource_type == "activities/heart":
self._attr_native_value = (
response[container][-1].get("value").get("restingHeartRate")
)
token = self.client.client.session.token
token = self.api.client.client.session.token
config_contents = {
ATTR_ACCESS_TOKEN: token.get("access_token"),
ATTR_REFRESH_TOKEN: token.get("refresh_token"),
CONF_CLIENT_ID: self.client.client.client_id,
CONF_CLIENT_SECRET: self.client.client.client_secret,
CONF_CLIENT_ID: self.api.client.client.client_id,
CONF_CLIENT_SECRET: self.api.client.client.client_secret,
ATTR_LAST_SAVED_AT: int(time.time()),
}
save_json(self.config_path, config_contents)