🎉 initial commit

.github/workflows/validate.yaml

@@ -0,0 +1,17 @@
+name: Validate
+
+on:
+  push: 
+  pull_request:
+  workflow_dispatch:
+
+jobs:
+  validate-hacs:
+    runs-on: "ubuntu-latest"
+    steps:
+      - uses: "actions/checkout@v3"
+      - name: HACS validation
+        uses: "hacs/action@main"
+        with:
+          category: "integration"
+      - uses: "home-assistant/actions/hassfest@master"

.gitignore

@@ -0,0 +1 @@
+__pycache__

README.md

@@ -0,0 +1,16 @@
+# Find My Integration for Home Assistant
+
+[![Open your Home Assistant instance and add this repository to the Home Assistant Community Store.](https://my.home-assistant.io/badges/hacs_repository.svg)](https://my.home-assistant.io/redirect/hacs_repository/?owner=krmax44&repository=homeassistant-findmy&category=integration) [![Open your Home Assistant instance and set up this integration.](https://my.home-assistant.io/badges/config_flow_start.svg)](https://my.home-assistant.io/redirect/config_flow_start/?domain=findmy)
+
+> [!WARNING]
+> This integration is still work-in-progress.
+
+## Installation
+
+You need to have a running [Anisette](https://github.com/Dadoum/anisette-v3-server) and a decrypted plist file of your tracker. The latter can be challenging to obtain, see [this issue](https://github.com/malmeloo/FindMy.py/issues/31)).
+
+With those two things ready, setup should be easy using the config flow.
+
+## Acknowledgements
+
+Thanks to [FindMy.py](https://github.com/malmeloo/FindMy.py/), on which this integration is based on.

custom_components/findmy/__init__.py

@@ -0,0 +1,37 @@
+"""The Find My integration."""
+
+from __future__ import annotations
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_URL, Platform
+from homeassistant.core import HomeAssistant
+
+from .const import CONF_ACCOUNT, CONF_PLIST
+from .coordinator import FindMyUpdateCoordinator
+from .findmy_hub import FindMyHub
+
+PLATFORMS: list[Platform] = [Platform.DEVICE_TRACKER]
+
+type FindMyConfigEntry = ConfigEntry[FindMyUpdateCoordinator]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: FindMyConfigEntry) -> bool:
+    """Set up Find My from a config entry."""
+
+    hub = FindMyHub(entry.data[CONF_URL])
+    hub.restore_account(entry.data[CONF_ACCOUNT])
+    hub.load_plist(entry.data[CONF_PLIST][0])
+
+    coordinator = FindMyUpdateCoordinator(hass=hass, hub=hub)
+    await coordinator.async_config_entry_first_refresh()
+
+    entry.runtime_data = coordinator
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: FindMyConfigEntry) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

custom_components/findmy/config_flow.py

@@ -0,0 +1,182 @@
+"""Config flow for the Find My integration."""
+
+from __future__ import annotations
+
+import logging
+from plistlib import InvalidFileException
+from typing import Any
+
+import aiohttp
+from findmy.errors import (
+    InvalidCredentialsError,
+    UnauthorizedError,
+    UnhandledProtocolError,
+)
+from findmy.reports.state import LoginState
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_EMAIL, CONF_PASSWORD, CONF_URL
+from homeassistant.helpers.selector import (
+    SelectOptionDict,
+    SelectSelector,
+    SelectSelectorConfig,
+    TextSelector,
+    TextSelectorConfig,
+    TextSelectorType,
+)
+
+from .const import CONF_2FA_CODE, CONF_2FA_METHOD, CONF_ACCOUNT, CONF_PLIST, DOMAIN
+from .findmy_hub import FindMyHub
+
+_LOGGER = logging.getLogger(__name__)
+
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(
+            CONF_URL,
+            default="http://localhost:6969",
+        ): TextSelector(TextSelectorConfig(type=TextSelectorType.URL)),
+        vol.Required(CONF_EMAIL): str,
+        vol.Required(CONF_PASSWORD): str,
+    }
+)
+
+STEP_2FA_CODE_DATA_SCHEMA = vol.Schema({vol.Required(CONF_2FA_CODE): str})
+
+STEP_PLIST_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_PLIST): TextSelector(TextSelectorConfig(multiline=True)),
+    }
+)
+
+
+class ConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Find My."""
+
+    VERSION = 1
+
+    def __init__(self):
+        """Initialize config flow."""
+        self.url: str | None = None
+        self.email: str | None = None
+        self.hub: FindMyHub | None = None
+        self.two_factor_method: int | None = 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:
+            try:
+                self.url = user_input[CONF_URL]
+                self.email = user_input[CONF_EMAIL]
+                self.hub = FindMyHub(user_input[CONF_URL])
+                state = await self.hub.authenticate(
+                    user_input[CONF_EMAIL], user_input[CONF_PASSWORD]
+                )
+
+                _LOGGER.debug("Login state: %s", state)
+
+                if state == LoginState.REQUIRE_2FA:
+                    methods = [
+                        SelectOptionDict(label=label, value=value)
+                        for (value, label) in await self.hub.get_2fa_methods()
+                    ]
+
+                    two_factor_schema = vol.Schema(
+                        {
+                            vol.Required(CONF_2FA_METHOD): SelectSelector(
+                                SelectSelectorConfig(options=methods)
+                            )
+                        }
+                    )
+
+                    return self.async_show_form(
+                        step_id="2fa_method", data_schema=two_factor_schema
+                    )
+                if state == LoginState.LOGGED_IN:
+                    return self.async_show_form(
+                        step_id="plist", data_schema=STEP_PLIST_DATA_SCHEMA
+                    )
+
+                errors["base"] = "invalid_auth"
+            except aiohttp.ClientConnectorError:
+                errors["base"] = "cannot_connect"
+            except (InvalidCredentialsError, UnauthorizedError):
+                errors["base"] = "invalid_auth"
+            except Exception as e:
+                _LOGGER.exception("Unexpected exception", exc_info=e)
+                errors["base"] = "unknown"
+
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )
+
+    async def async_step_2fa_method(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the two-factor method step."""
+
+        if user_input is not None and self.hub is not None:
+            self.two_factor_method = int(user_input[CONF_2FA_METHOD])
+
+            try:
+                await self.hub.request_two_factor(self.two_factor_method)
+            except UnhandledProtocolError:
+                return self.async_abort(reason="unknown")
+
+            return self.async_show_form(
+                step_id="2fa_code", data_schema=STEP_2FA_CODE_DATA_SCHEMA
+            )
+
+        return self.async_abort(reason="invalid input")
+
+    async def async_step_2fa_code(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the two-factor step."""
+
+        if user_input and self.hub and self.two_factor_method is not None:
+            state = await self.hub.submit_two_factor(
+                self.two_factor_method, user_input[CONF_2FA_CODE]
+            )
+
+            if state not in (LoginState.LOGGED_IN, LoginState.AUTHENTICATED):
+                return self.async_abort(reason="invalid_2fa_code")
+
+            return self.async_show_form(
+                step_id="plist", data_schema=STEP_PLIST_DATA_SCHEMA
+            )
+
+        return self.async_abort(reason="invalid input")
+
+    async def async_step_plist(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the plist step."""
+        errors: dict[str, str] = {}
+
+        if user_input and self.hub and self.email:
+            try:
+                self.hub.load_plist(user_input[CONF_PLIST])
+
+                return self.async_create_entry(
+                    title=self.email,
+                    data={
+                        CONF_URL: self.url,
+                        CONF_ACCOUNT: self.hub.get_account_credentials(),
+                        CONF_PLIST: user_input[CONF_PLIST],
+                    },
+                )
+            except InvalidFileException:
+                errors["base"] = "invalid_file"
+
+                return self.async_show_form(
+                    step_id="plist", data_schema=STEP_PLIST_DATA_SCHEMA, errors=errors
+                )
+
+        return self.async_abort(reason="invalid input")

custom_components/findmy/const.py

@@ -0,0 +1,12 @@
+"""Constants for the Find My integration."""
+
+from typing import Final
+
+DOMAIN = "findmy"
+
+CONF_2FA_METHOD: Final = "2fa_method"
+CONF_2FA_CODE: Final = "2fa_code"
+CONF_PLIST: Final = "plist"
+CONF_ACCOUNT: Final = "account"
+
+DEFAULT_UPDATE_INTERVAL = 1800  # 30 minutes

custom_components/findmy/coordinator.py

@@ -0,0 +1,37 @@
+"""DataUpdateCoordinator for the FindMy integration."""
+
+from __future__ import annotations
+
+from datetime import timedelta
+import logging
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import DEFAULT_UPDATE_INTERVAL
+from .findmy_hub import FindMyHub, FindMyReport
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class FindMyUpdateCoordinator(DataUpdateCoordinator[FindMyReport]):
+    """The FindMy update coordinator."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        hub: FindMyHub,
+    ) -> None:
+        """Initialize the FindMy coordinator."""
+        self.hub = hub
+
+        super().__init__(
+            hass,
+            _LOGGER,
+            name="FindMy Accessory",
+            update_interval=timedelta(seconds=DEFAULT_UPDATE_INTERVAL),
+        )
+
+    async def _async_update_data(self) -> FindMyReport:
+        """Trigger position update."""
+        return await self.hub.get_position()

custom_components/findmy/device_tracker.py

@@ -0,0 +1,49 @@
+"""Support for tracking FindMy devices."""
+
+from config.custom_components.findmy import FindMyConfigEntry
+from config.custom_components.findmy.coordinator import FindMyUpdateCoordinator
+from homeassistant.components.device_tracker import TrackerEntity
+from homeassistant.components.device_tracker.const import SourceType
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: FindMyConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up a Ping config entry."""
+    async_add_entities([FindMyDeviceTracker(entry, entry.runtime_data)])
+
+
+class FindMyDeviceTracker(CoordinatorEntity[FindMyUpdateCoordinator], TrackerEntity):
+    """Representation of a FindMy device tracker."""
+
+    def __init__(
+        self, config_entry: ConfigEntry, coordinator: FindMyUpdateCoordinator
+    ) -> None:
+        """Initialize the Ping device tracker."""
+        super().__init__(coordinator)
+
+        self.config_entry = config_entry
+        self._attr_unique_id = coordinator.hub.accessory.identifier
+        self._attr_name = coordinator.hub.accessory.name
+        self._attr_source_type = SourceType.GPS
+
+    @property
+    def location_accuracy(self):
+        """Return the location accuracy of the device."""
+        return self.coordinator.data.accuracy
+
+    @property
+    def latitude(self):
+        """Return latitude value of the device."""
+        return self.coordinator.data.latitude
+
+    @property
+    def longitude(self):
+        """Return longitude value of the device."""
+        return self.coordinator.data.longitude