Add Ituran integration

Author: shmuelzon

CODEOWNERS

@@ -753,6 +753,8 @@ build.json @home-assistant/supervisor
 /tests/components/ista_ecotrend/ @tr4nt0r
 /homeassistant/components/isy994/ @bdraco @shbatm
 /tests/components/isy994/ @bdraco @shbatm
+/homeassistant/components/ituran/ @shmuelzon
+/tests/components/ituran/ @shmuelzon
 /homeassistant/components/izone/ @Swamp-Ig
 /tests/components/izone/ @Swamp-Ig
 /homeassistant/components/jellyfin/ @j-stienstra @ctalkington

homeassistant/components/ituran/__init__.py

@@ -0,0 +1,31 @@
+"""The Ituran integration."""
+
+from __future__ import annotations
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from .coordinator import IturanDataUpdateCoordinator
+
+PLATFORMS: list[Platform] = [
+    Platform.DEVICE_TRACKER,
+]
+
+type IturanConfigEntry = ConfigEntry[IturanDataUpdateCoordinator]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: IturanConfigEntry) -> bool:
+    """Set up Ituran from a config entry."""
+
+    coordinator = IturanDataUpdateCoordinator(hass, entry=entry)
+    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: IturanConfigEntry) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/ituran/config_flow.py

@@ -0,0 +1,119 @@
+"""Config flow for Ituran integration."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+import logging
+from typing import Any
+
+from pyituran import Ituran
+from pyituran.exceptions import IturanApiError, IturanAuthError
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+
+from .const import (
+    CONF_ID_OR_PASSPORT,
+    CONF_MOBILE_ID,
+    CONF_OTP,
+    CONF_PHONE_NUMBER,
+    DOMAIN,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_ID_OR_PASSPORT): str,
+        vol.Required(CONF_PHONE_NUMBER): str,
+        vol.Optional(CONF_MOBILE_ID): str,
+    }
+)
+
+STEP_OTP_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_OTP, description="OTP"): str,
+    }
+)
+
+
+class IturanConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Ituran."""
+
+    _user_info: dict[str, Any]
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the inial step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            if self.unique_id is None:
+                await self.async_set_unique_id(user_input[CONF_ID_OR_PASSPORT])
+                self._abort_if_unique_id_configured()
+
+            try:
+                ituran = Ituran(
+                    user_input[CONF_ID_OR_PASSPORT],
+                    user_input[CONF_PHONE_NUMBER],
+                    user_input.get(CONF_MOBILE_ID),
+                )
+                user_input[CONF_MOBILE_ID] = ituran.mobile_id
+                authenticated = await ituran.is_authenticated()
+                if not authenticated:
+                    await ituran.request_otp()
+            except IturanApiError:
+                errors["base"] = "cannot_connect"
+            except IturanAuthError:
+                errors["base"] = "invalid_auth"
+            except Exception:
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            else:
+                if authenticated:
+                    return self.async_create_entry(
+                        title=f"Ituran {user_input[CONF_ID_OR_PASSPORT]}",
+                        data=user_input,
+                    )
+                self._user_info = user_input
+                return await self.async_step_otp()
+
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )
+
+    async def async_step_reauth(
+        self, entry_data: Mapping[str, Any]
+    ) -> ConfigFlowResult:
+        """Handle configuration by re-auth."""
+        return await self.async_step_user()
+
+    async def async_step_otp(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the inial step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            try:
+                ituran = Ituran(
+                    self._user_info[CONF_ID_OR_PASSPORT],
+                    self._user_info[CONF_PHONE_NUMBER],
+                    self._user_info[CONF_MOBILE_ID],
+                )
+                await ituran.authenticate(user_input[CONF_OTP])
+            except IturanApiError:
+                errors["base"] = "cannot_connect"
+            except IturanAuthError:
+                errors["base"] = "invalid_otp"
+            except Exception:
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            else:
+                return self.async_create_entry(
+                    title=f"Ituran {self._user_info[CONF_ID_OR_PASSPORT]}",
+                    data=self._user_info,
+                )
+
+        return self.async_show_form(
+            step_id="otp", data_schema=STEP_OTP_DATA_SCHEMA, errors=errors
+        )

homeassistant/components/ituran/const.py

@@ -0,0 +1,14 @@
+"""Constants for the Ituran integration."""
+
+from typing import Final
+
+DOMAIN = "ituran"
+
+ATTR_HEADING: Final = "heading"
+ATTR_ADDRESS: Final = "address"
+ATTR_LAST_UPDATE: Final = "last_update"
+
+CONF_ID_OR_PASSPORT: Final = "id_or_passport"
+CONF_PHONE_NUMBER: Final = "phone_number"
+CONF_MOBILE_ID: Final = "mobile_id"
+CONF_OTP: Final = "otp"

homeassistant/components/ituran/coordinator.py

@@ -0,0 +1,67 @@
+"""Coordinator for Ituran."""
+
+from datetime import timedelta
+import logging
+
+from pyituran import Ituran, Vehicle
+from pyituran.exceptions import IturanApiError, IturanAuthError
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed
+from homeassistant.helpers import device_registry as dr
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import CONF_ID_OR_PASSPORT, CONF_MOBILE_ID, CONF_PHONE_NUMBER, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class IturanDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Vehicle]]):
+    """Class to manage fetching Ituran data."""
+
+    ituran: Ituran
+
+    def __init__(self, hass: HomeAssistant, *, entry: ConfigEntry) -> None:
+        """Initialize account-wide BMW data updater."""
+        self.ituran = Ituran(
+            entry.data[CONF_ID_OR_PASSPORT],
+            entry.data[CONF_PHONE_NUMBER],
+            entry.data[CONF_MOBILE_ID],
+        )
+        self._entry = entry
+
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=f"{DOMAIN}-{entry.data[CONF_ID_OR_PASSPORT]}",
+            update_interval=timedelta(seconds=300),
+            config_entry=entry,
+        )
+
+    async def _async_update_data(self) -> dict[str, Vehicle]:
+        """Fetch data from Ituran."""
+
+        try:
+            vehicles = await self.ituran.get_vehicles()
+        except IturanApiError as e:
+            raise UpdateFailed(e) from e
+        except IturanAuthError as e:
+            raise ConfigEntryAuthFailed(e) from e
+
+        updated_data = {vehicle.license_plate: vehicle for vehicle in vehicles}
+        self._cleanup_removed_vehicles(updated_data)
+
+        return updated_data
+
+    def _cleanup_removed_vehicles(self, data: dict[str, Vehicle]) -> None:
+        account_vehicles = {(DOMAIN, license_plate) for license_plate in data}
+        device_registry = dr.async_get(self.hass)
+        device_entries = dr.async_entries_for_config_entry(
+            device_registry, config_entry_id=self._entry.entry_id
+        )
+        for device in device_entries:
+            if not device.identifiers.intersection(account_vehicles):
+                device_registry.async_update_device(
+                    device.id, remove_config_entry_id=self._entry.entry_id
+                )

homeassistant/components/ituran/device_tracker.py

@@ -0,0 +1,62 @@
+"""Device tracker for Ituran vehicles."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from homeassistant.components.device_tracker import TrackerEntity
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from . import IturanConfigEntry
+from .const import ATTR_ADDRESS, ATTR_HEADING, ATTR_LAST_UPDATE
+from .coordinator import IturanDataUpdateCoordinator
+from .entity import IturanBaseEntity
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: IturanConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up the Ituran tracker from config entry."""
+    coordinator = config_entry.runtime_data
+    async_add_entities(
+        IturanDeviceTracker(coordinator, license_plate)
+        for license_plate in coordinator.data
+    )
+
+
+class IturanDeviceTracker(IturanBaseEntity, TrackerEntity):
+    """Ituran device tracker."""
+
+    _attr_force_update = False
+    _attr_translation_key = "car"
+    _attr_name = None
+
+    def __init__(
+        self,
+        coordinator: IturanDataUpdateCoordinator,
+        license_plate: str,
+    ) -> None:
+        """Initialize the device tracker."""
+        super().__init__(coordinator, license_plate, "device_tracker")
+
+    @property
+    def extra_state_attributes(self) -> dict[str, Any]:
+        """Return entity specific state attributes."""
+        return {
+            ATTR_ADDRESS: self.vehicle.address,
+            ATTR_HEADING: self.vehicle.heading,
+            ATTR_LAST_UPDATE: self.vehicle.last_update,
+        }
+
+    @property
+    def latitude(self) -> float | None:
+        """Return latitude value of the device."""
+        return self.vehicle.gps_coordinates[0]
+
+    @property
+    def longitude(self) -> float | None:
+        """Return longitude value of the device."""
+        return self.vehicle.gps_coordinates[1]

homeassistant/components/ituran/entity.py

@@ -0,0 +1,47 @@
+"""Base for all turan entities."""
+
+from __future__ import annotations
+
+from pyituran import Vehicle
+
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN
+from .coordinator import IturanDataUpdateCoordinator
+
+
+class IturanBaseEntity(CoordinatorEntity[IturanDataUpdateCoordinator]):
+    """Common base for Ituran entities."""
+
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        coordinator: IturanDataUpdateCoordinator,
+        license_plate: str,
+        unique_key: str,
+    ) -> None:
+        """Initialize the entity."""
+        super().__init__(coordinator)
+
+        self._license_plate = license_plate
+        self._attr_unique_id = f"{license_plate}-{unique_key}"
+
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, self.vehicle.license_plate)},
+            manufacturer=self.vehicle.make,
+            model=self.vehicle.model,
+            name=self.vehicle.model,
+            serial_number=self.vehicle.license_plate,
+        )
+
+    @property
+    def available(self) -> bool:
+        """Return True if vehicle is still included in the account."""
+        return self._license_plate in self.coordinator.data
+
+    @property
+    def vehicle(self) -> Vehicle:
+        """Return the vehicle information associated with this entity."""
+        return self.coordinator.data[self._license_plate]

homeassistant/components/ituran/icons.json

@@ -0,0 +1,9 @@
+{
+  "entity": {
+    "device_tracker": {
+      "car": {
+        "default": "mdi:car"
+      }
+    }
+  }
+}

homeassistant/components/ituran/manifest.json

@@ -0,0 +1,10 @@
+{
+  "domain": "ituran",
+  "name": "Ituran",
+  "codeowners": ["@shmuelzon"],
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/ituran",
+  "integration_type": "hub",
+  "iot_class": "cloud_polling",
+  "requirements": ["pyituran==0.1.2"]
+}