Add integration fressnapf_tracker (home-assistant#157480)

Co-authored-by: Joost Lekkerkerker <[email protected]>

Author: eifinger

CODEOWNERS

@@ -0,0 +1,49 @@
+"""The Fressnapf Tracker integration."""
+
+from fressnapftracker import AuthClient
+
+from homeassistant.const import CONF_ACCESS_TOKEN, Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.httpx_client import get_async_client
+
+from .const import CONF_USER_ID
+from .coordinator import (
+    FressnapfTrackerConfigEntry,
+    FressnapfTrackerDataUpdateCoordinator,
+)
+
+PLATFORMS: list[Platform] = [Platform.DEVICE_TRACKER]
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: FressnapfTrackerConfigEntry
+) -> bool:
+    """Set up Fressnapf Tracker from a config entry."""
+    auth_client = AuthClient(client=get_async_client(hass))
+    devices = await auth_client.get_devices(
+        user_id=entry.data[CONF_USER_ID],
+        user_access_token=entry.data[CONF_ACCESS_TOKEN],
+    )
+
+    coordinators: list[FressnapfTrackerDataUpdateCoordinator] = []
+    for device in devices:
+        coordinator = FressnapfTrackerDataUpdateCoordinator(
+            hass,
+            entry,
+            device,
+        )
+        await coordinator.async_config_entry_first_refresh()
+        coordinators.append(coordinator)
+
+    entry.runtime_data = coordinators
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(
+    hass: HomeAssistant, entry: FressnapfTrackerConfigEntry
+) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/fressnapf_tracker/__init__.py

@@ -0,0 +1,193 @@
+"""Config flow for the Fressnapf Tracker integration."""
+
+import logging
+from typing import Any
+
+from fressnapftracker import (
+    AuthClient,
+    FressnapfTrackerInvalidPhoneNumberError,
+    FressnapfTrackerInvalidTokenError,
+)
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_ACCESS_TOKEN
+from homeassistant.helpers.httpx_client import get_async_client
+
+from .const import CONF_PHONE_NUMBER, CONF_SMS_CODE, CONF_USER_ID, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_PHONE_NUMBER): str,
+    }
+)
+STEP_SMS_CODE_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_SMS_CODE): int,
+    }
+)
+
+
+class FressnapfTrackerConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Fressnapf Tracker."""
+
+    VERSION = 1
+
+    def __init__(self) -> None:
+        """Init Config Flow."""
+        self._context: dict[str, Any] = {}
+        self._auth_client: AuthClient | None = None
+
+    @property
+    def auth_client(self) -> AuthClient:
+        """Return the auth client, creating it if needed."""
+        if self._auth_client is None:
+            self._auth_client = AuthClient(client=get_async_client(self.hass))
+        return self._auth_client
+
+    async def _async_request_sms_code(
+        self, phone_number: str
+    ) -> tuple[dict[str, str], bool]:
+        """Request SMS code and return errors dict and success flag."""
+        errors: dict[str, str] = {}
+        try:
+            response = await self.auth_client.request_sms_code(
+                phone_number=phone_number
+            )
+        except FressnapfTrackerInvalidPhoneNumberError:
+            errors["base"] = "invalid_phone_number"
+        except Exception:
+            _LOGGER.exception("Unexpected exception")
+            errors["base"] = "unknown"
+        else:
+            _LOGGER.debug("SMS code request response: %s", response)
+            self._context[CONF_USER_ID] = response.id
+            self._context[CONF_PHONE_NUMBER] = phone_number
+            return errors, True
+        return errors, False
+
+    async def _async_verify_sms_code(
+        self, sms_code: int
+    ) -> tuple[dict[str, str], str | None]:
+        """Verify SMS code and return errors and access_token."""
+        errors: dict[str, str] = {}
+        try:
+            verification_response = await self.auth_client.verify_phone_number(
+                user_id=self._context[CONF_USER_ID],
+                sms_code=sms_code,
+            )
+        except FressnapfTrackerInvalidTokenError:
+            errors["base"] = "invalid_sms_code"
+        except Exception:
+            _LOGGER.exception("Unexpected exception during SMS code verification")
+            errors["base"] = "unknown"
+        else:
+            _LOGGER.debug(
+                "Phone number verification response: %s", verification_response
+            )
+            return errors, verification_response.user_token.access_token
+        return errors, None
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            self._async_abort_entries_match(
+                {CONF_PHONE_NUMBER: user_input[CONF_PHONE_NUMBER]}
+            )
+            errors, success = await self._async_request_sms_code(
+                user_input[CONF_PHONE_NUMBER]
+            )
+            if success:
+                await self.async_set_unique_id(str(self._context[CONF_USER_ID]))
+                self._abort_if_unique_id_configured()
+                return await self.async_step_sms_code()
+
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )
+
+    async def async_step_sms_code(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the SMS code step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            errors, access_token = await self._async_verify_sms_code(
+                user_input[CONF_SMS_CODE]
+            )
+            if access_token:
+                return self.async_create_entry(
+                    title=self._context[CONF_PHONE_NUMBER],
+                    data={
+                        CONF_PHONE_NUMBER: self._context[CONF_PHONE_NUMBER],
+                        CONF_USER_ID: self._context[CONF_USER_ID],
+                        CONF_ACCESS_TOKEN: access_token,
+                    },
+                )
+
+        return self.async_show_form(
+            step_id="sms_code",
+            data_schema=STEP_SMS_CODE_DATA_SCHEMA,
+            errors=errors,
+        )
+
+    async def async_step_reconfigure(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle reconfiguration of the integration."""
+        errors: dict[str, str] = {}
+        reconfigure_entry = self._get_reconfigure_entry()
+
+        if user_input is not None:
+            errors, success = await self._async_request_sms_code(
+                user_input[CONF_PHONE_NUMBER]
+            )
+            if success:
+                if reconfigure_entry.data[CONF_USER_ID] != self._context[CONF_USER_ID]:
+                    errors["base"] = "account_change_not_allowed"
+                else:
+                    return await self.async_step_reconfigure_sms_code()
+
+        return self.async_show_form(
+            step_id="reconfigure",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(
+                        CONF_PHONE_NUMBER,
+                        default=reconfigure_entry.data.get(CONF_PHONE_NUMBER),
+                    ): str,
+                }
+            ),
+            errors=errors,
+        )
+
+    async def async_step_reconfigure_sms_code(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the SMS code step during reconfiguration."""
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            errors, access_token = await self._async_verify_sms_code(
+                user_input[CONF_SMS_CODE]
+            )
+            if access_token:
+                return self.async_update_reload_and_abort(
+                    self._get_reconfigure_entry(),
+                    data={
+                        CONF_PHONE_NUMBER: self._context[CONF_PHONE_NUMBER],
+                        CONF_USER_ID: self._context[CONF_USER_ID],
+                        CONF_ACCESS_TOKEN: access_token,
+                    },
+                )
+
+        return self.async_show_form(
+            step_id="reconfigure_sms_code",
+            data_schema=STEP_SMS_CODE_DATA_SCHEMA,
+            errors=errors,
+        )

homeassistant/components/fressnapf_tracker/config_flow.py

@@ -0,0 +1,6 @@
+"""Constants for the Fressnapf Tracker integration."""
+
+DOMAIN = "fressnapf_tracker"
+CONF_PHONE_NUMBER = "phone_number"
+CONF_SMS_CODE = "sms_code"
+CONF_USER_ID = "user_id"

homeassistant/components/fressnapf_tracker/const.py

@@ -0,0 +1,50 @@
+"""Data update coordinator for Fressnapf Tracker integration."""
+
+from datetime import timedelta
+import logging
+
+from fressnapftracker import ApiClient, Device, FressnapfTrackerError, Tracker
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.httpx_client import get_async_client
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+type FressnapfTrackerConfigEntry = ConfigEntry[
+    list[FressnapfTrackerDataUpdateCoordinator]
+]
+
+
+class FressnapfTrackerDataUpdateCoordinator(DataUpdateCoordinator[Tracker]):
+    """Class to manage fetching data from the API."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        config_entry: FressnapfTrackerConfigEntry,
+        device: Device,
+    ) -> None:
+        """Initialize."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=DOMAIN,
+            update_interval=timedelta(minutes=15),
+            config_entry=config_entry,
+        )
+        self.device = device
+        self.client = ApiClient(
+            serial_number=device.serialnumber,
+            device_token=device.token,
+            client=get_async_client(hass),
+        )
+
+    async def _async_update_data(self) -> Tracker:
+        try:
+            return await self.client.get_tracker()
+        except FressnapfTrackerError as exception:
+            raise UpdateFailed(exception) from exception

homeassistant/components/fressnapf_tracker/coordinator.py

@@ -0,0 +1,69 @@
+"""Device tracker platform for fressnapf_tracker."""
+
+from homeassistant.components.device_tracker import SourceType
+from homeassistant.components.device_tracker.config_entry import TrackerEntity
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+
+from . import FressnapfTrackerConfigEntry, FressnapfTrackerDataUpdateCoordinator
+from .entity import FressnapfTrackerBaseEntity
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: FressnapfTrackerConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up the fressnapf_tracker device_trackers."""
+    async_add_entities(
+        FressnapfTrackerDeviceTracker(coordinator) for coordinator in entry.runtime_data
+    )
+
+
+class FressnapfTrackerDeviceTracker(FressnapfTrackerBaseEntity, TrackerEntity):
+    """fressnapf_tracker device tracker."""
+
+    _attr_name = None
+    _attr_translation_key = "pet"
+
+    def __init__(
+        self,
+        coordinator: FressnapfTrackerDataUpdateCoordinator,
+    ) -> None:
+        """Initialize the device tracker."""
+        super().__init__(coordinator)
+        self._attr_unique_id = coordinator.device.serialnumber
+
+    @property
+    def available(self) -> bool:
+        """Return if entity is available."""
+        return super().available and self.coordinator.data.position is not None
+
+    @property
+    def latitude(self) -> float | None:
+        """Return latitude value of the device."""
+        if self.coordinator.data.position is not None:
+            return self.coordinator.data.position.lat
+        return None
+
+    @property
+    def longitude(self) -> float | None:
+        """Return longitude value of the device."""
+        if self.coordinator.data.position is not None:
+            return self.coordinator.data.position.lng
+        return None
+
+    @property
+    def source_type(self) -> SourceType:
+        """Return the source type, eg gps or router, of the device."""
+        return SourceType.GPS
+
+    @property
+    def location_accuracy(self) -> float:
+        """Return the location accuracy of the device.
+
+        Value in meters.
+        """
+        if self.coordinator.data.position is not None:
+            return float(self.coordinator.data.position.accuracy)
+        return 0