hass-core/homeassistant/components/garmin_connect/sensor.py
Ron Klinkien 4e2737bfb7 Add Garmin Connect integration (#30792)
* Added code files

* Correctly name init file

* Update codeowners

* Update requirements

* Added code files

* Correctly name init file

* Update codeowners

* Update requirements

* Black changes, added to coveragerc

* Removed documentation location for now

* Added documentation url

* Fixed merge

* Fixed flake8 syntax

* Fixed isort

* Removed false check and double throttle, applied time format change

* Renamed email to username, used dict, deleted unused type, changed attr name

* Async and ConfigFlow code

* Fixes

* Added device_class and misc fixes

* isort and pylint fixes

* Removed from test requirements

* Fixed isort checkblack

* Removed host field

* Fixed coveragerc

* Start working test file

* Added more config_flow tests

* Enable only most used sensors by default

* Added more default enabled sensors, fixed tests

* Fixed isort

* Test config_flow  improvements

* Remove unused import

* Removed redundant patch calls

* Fixed mock return value

* Updated to garmin_connect 0.1.8 fixed exceptions

* Quick fix test patch to see if rest is error free

* Fixed mock routine

* Code improvements from PR feedback

* Fix entity indentifier

* Reverted device identifier

* Fixed abort message

* Test fix

* Fixed unique_id MockConfigEntry
2020-01-27 09:12:18 -08:00

177 lines
4.9 KiB
Python

"""Platform for Garmin Connect integration."""
import logging
from typing import Any, Dict
from garminconnect import (
GarminConnectAuthenticationError,
GarminConnectConnectionError,
GarminConnectTooManyRequestsError,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_ATTRIBUTION, CONF_ID
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.typing import HomeAssistantType
from .const import ATTRIBUTION, DOMAIN, GARMIN_ENTITY_LIST
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities
) -> None:
"""Set up Garmin Connect sensor based on a config entry."""
garmin_data = hass.data[DOMAIN][entry.entry_id]
unique_id = entry.data[CONF_ID]
try:
await garmin_data.async_update()
except (
GarminConnectConnectionError,
GarminConnectAuthenticationError,
GarminConnectTooManyRequestsError,
) as err:
_LOGGER.error("Error occured during Garmin Connect Client update: %s", err)
except Exception: # pylint: disable=broad-except
_LOGGER.error("Unknown error occured during Garmin Connect Client update.")
entities = []
for (
sensor_type,
(name, unit, icon, device_class, enabled_by_default),
) in GARMIN_ENTITY_LIST.items():
_LOGGER.debug(
"Registering entity: %s, %s, %s, %s, %s, %s",
sensor_type,
name,
unit,
icon,
device_class,
enabled_by_default,
)
entities.append(
GarminConnectSensor(
garmin_data,
unique_id,
sensor_type,
name,
unit,
icon,
device_class,
enabled_by_default,
)
)
async_add_entities(entities, True)
class GarminConnectSensor(Entity):
"""Representation of a Garmin Connect Sensor."""
def __init__(
self,
data,
unique_id,
sensor_type,
name,
unit,
icon,
device_class,
enabled_default: bool = True,
):
"""Initialize."""
self._data = data
self._unique_id = unique_id
self._type = sensor_type
self._name = name
self._unit = unit
self._icon = icon
self._device_class = device_class
self._enabled_default = enabled_default
self._available = True
self._state = None
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def icon(self):
"""Return the icon to use in the frontend."""
return self._icon
@property
def state(self):
"""Return the state of the sensor."""
return self._state
@property
def unique_id(self) -> str:
"""Return the unique ID for this sensor."""
return f"{self._unique_id}_{self._type}"
@property
def unit_of_measurement(self):
"""Return the unit of measurement."""
return self._unit
@property
def device_state_attributes(self):
"""Return attributes for sensor."""
attributes = {}
if self._data.data:
attributes = {
"source": self._data.data["source"],
"last_synced": self._data.data["lastSyncTimestampGMT"],
ATTR_ATTRIBUTION: ATTRIBUTION,
}
return attributes
@property
def device_info(self) -> Dict[str, Any]:
"""Return device information."""
return {
"identifiers": {(DOMAIN, self._unique_id)},
"name": "Garmin Connect",
"manufacturer": "Garmin Connect",
}
@property
def entity_registry_enabled_default(self) -> bool:
"""Return if the entity should be enabled when first added to the entity registry."""
return self._enabled_default
@property
def available(self) -> bool:
"""Return True if entity is available."""
return self._available
@property
def device_class(self):
"""Return the device class of the sensor."""
return self._device_class
async def async_update(self):
"""Update the data from Garmin Connect."""
if not self.enabled:
return
await self._data.async_update()
if not self._data.data:
_LOGGER.error("Didn't receive data from Garmin Connect")
return
data = self._data.data
if "Duration" in self._type:
self._state = data[self._type] // 60
elif "Seconds" in self._type:
self._state = data[self._type] // 60
else:
self._state = data[self._type]
_LOGGER.debug(
"Entity %s set to state %s %s", self._type, self._state, self._unit
)