Gear service to update default gear per activity on garmin connect

custom_components/garmin_connect/__init__.py

@@ -14,10 +14,16 @@ from garminconnect import (
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
 from homeassistant.core import HomeAssistant
-from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.exceptions import ConfigEntryNotReady, IntegrationError
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
-from .const import DATA_COORDINATOR, DEFAULT_UPDATE_INTERVAL, DOMAIN
+from .const import (
+    DATA_COORDINATOR,
+    DEFAULT_UPDATE_INTERVAL,
+    DOMAIN,
+    GEAR,
+    SERVICE_SETTING,
+)
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -35,12 +41,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     await coordinator.async_config_entry_first_refresh()
 
     hass.data.setdefault(DOMAIN, {})
-    hass.data[DOMAIN][entry.entry_id] = {
-        DATA_COORDINATOR: coordinator,
-    }
+    hass.data[DOMAIN][entry.entry_id] = {DATA_COORDINATOR: coordinator}
 
     hass.config_entries.async_setup_platforms(entry, PLATFORMS)
-
     return True
 
 
@@ -100,15 +103,18 @@ class GarminConnectDataUpdateCoordinator(DataUpdateCoordinator):
             )
             alarms = await self.hass.async_add_executor_job(self._api.get_device_alarms)
             gear = await self.hass.async_add_executor_job(
-                self._api.get_gear, summary["userProfileId"]
+                self._api.get_gear, summary[GEAR.USERPROFILE_ID]
             )
             tasks: list[Awaitable] = [
                 self.hass.async_add_executor_job(
-                    self._api.get_gear_stats, gear_item["uuid"]
+                    self._api.get_gear_stats, gear_item[GEAR.UUID]
                 )
                 for gear_item in gear
             ]
             gear_stats = await asyncio.gather(*tasks)
+            activity_types = await self.hass.async_add_executor_job(
+                self._api.get_activity_types
+            )
         except (
             GarminConnectAuthenticationError,
             GarminConnectTooManyRequestsError,
@@ -125,4 +131,46 @@ class GarminConnectDataUpdateCoordinator(DataUpdateCoordinator):
             "nextAlarm": alarms,
             "gear": gear,
             "gear_stats": gear_stats,
+            "activity_types": activity_types,
         }
+
+    async def set_active_gear(self, entity, service_data):
+        """Update Garmin Gear settings"""
+        if not await self.async_login():
+            raise IntegrationError(
+                "Failed to login to Garmin Connect, unable to update"
+            )
+
+        setting = service_data.data["setting"]
+        activity_type_id = next(
+            filter(
+                lambda a: a[GEAR.TYPE_KEY] == service_data.data["activity_type"],
+                self.data["activity_types"],
+            )
+        )[GEAR.TYPE_ID]
+        if setting != SERVICE_SETTING.ONLY_THIS_AS_DEFAULT:
+            await self._api.set_gear_default(
+                activity_type_id, entity.uuid, setting == SERVICE_SETTING.DEFAULT
+            )
+        else:
+            old_default_state = await self.hass.async_add_executor_job(
+                self._api.get_gear_defaults, self.data[GEAR.USERPROFILE_ID]
+            )
+            to_deactivate = list(
+                filter(
+                    lambda o: o[GEAR.ACTIVITY_TYPE_PK] == activity_type_id
+                    and o[GEAR.UUID] != entity.uuid,
+                    old_default_state,
+                )
+            )
+
+            for active_gear in to_deactivate:
+                await self.hass.async_add_executor_job(
+                    self._api.set_gear_default,
+                    activity_type_id,
+                    active_gear[GEAR.UUID],
+                    False,
+                )
+            await self.hass.async_add_executor_job(
+                self._api.set_gear_default, activity_type_id, entity.uuid, True
+            )

custom_components/garmin_connect/const.py

@@ -1,5 +1,7 @@
 """Constants for the Garmin Connect integration."""
 from datetime import timedelta
+from enum import Enum
+from typing import NamedTuple
 
 from homeassistant.const import (
     DEVICE_CLASS_TIMESTAMP,
@@ -361,3 +363,19 @@ GEAR_ICONS = {
     "Other": "mdi:basketball",
     "Golf Clubs": "mdi:golf",
 }
+
+
+class SERVICE_SETTING(NamedTuple):
+    """Options for the service settings, see services.yaml"""
+
+    ONLY_THIS_AS_DEFAULT = "set this as default, unset others"
+    DEFAULT = "set as default"
+    UNSET_DEFAULT = "unset default"
+
+
+class GEAR(NamedTuple):
+    UUID = "uuid"
+    TYPE_KEY = "typeKey"
+    TYPE_ID = "typeId"
+    USERPROFILE_ID = "userProfileId"
+    ACTIVITY_TYPE_PK = "activityTypePk"

custom_components/garmin_connect/sensor.py

@@ -2,27 +2,33 @@
 from __future__ import annotations
 
 import logging
+import voluptuous as vol
 from numbers import Number
 
+
 from homeassistant.components.sensor import SensorEntity, SensorStateClass
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import (
+    ATTR_ENTITY_ID,
     CONF_ID,
     DEVICE_CLASS_TIMESTAMP,
     LENGTH_KILOMETERS,
 )
 from homeassistant.core import HomeAssistant
+from homeassistant.helpers import entity_platform
 from homeassistant.helpers.entity import DeviceInfo
 from homeassistant.helpers.update_coordinator import (
     CoordinatorEntity,
     DataUpdateCoordinator,
 )
 
+
 from .alarm_util import calculate_next_active_alarms
 from .const import (
     DATA_COORDINATOR,
     DOMAIN as GARMIN_DOMAIN,
     GARMIN_ENTITY_LIST,
+    GEAR,
     GEAR_ICONS,
 )
 
@@ -65,21 +71,35 @@ async def async_setup_entry(
                 enabled_by_default,
             )
         )
-
-    for gear_item in coordinator.data["gear"]:
-        entities.append(
-            GarminConnectGearSensor(
-                coordinator,
-                unique_id,
-                gear_item["uuid"],
-                gear_item["gearTypeName"],
-                gear_item["displayName"],
-                None,
-                True,
+    if "gear" in coordinator.data:
+        for gear_item in coordinator.data["gear"]:
+            entities.append(
+                GarminConnectGearSensor(
+                    coordinator,
+                    unique_id,
+                    gear_item[GEAR.UUID],
+                    gear_item["gearTypeName"],
+                    gear_item["displayName"],
+                    None,
+                    True,
+                )
             )
-        )
 
     async_add_entities(entities)
+    platform = entity_platform.async_get_current_platform()
+
+    platform.async_register_entity_service(
+        "set_active_gear", ENTITY_SERVICE_SCHEMA, coordinator.set_active_gear
+    )
+
+
+ENTITY_SERVICE_SCHEMA = vol.Schema(
+    {
+        vol.Required(ATTR_ENTITY_ID): str,
+        vol.Required("activity_type"): str,
+        vol.Required("setting"): str,
+    }
+)
 
 
 class GarminConnectSensor(CoordinatorEntity, SensorEntity):
@@ -204,6 +224,12 @@ class GarminConnectGearSensor(CoordinatorEntity, SensorEntity):
         self._attr_native_unit_of_measurement = LENGTH_KILOMETERS
         self._attr_unique_id = f"{self._unique_id}_{self._uuid}"
         self._attr_state_class = SensorStateClass.TOTAL
+        self._attr_device_class = "Gear"
+
+    @property
+    def uuid(self):
+        """Return the entity uuid"""
+        return self._uuid
 
     @property
     def native_value(self):
@@ -255,19 +281,16 @@ class GarminConnectGearSensor(CoordinatorEntity, SensorEntity):
     @property
     def available(self) -> bool:
         """Return True if entity is available."""
-        return (
-            super().available
-            and self.coordinator.data
-            and self.get_gear()
-            # and any(g["uuid"] == self._uuid for g in self.coordinator.data["gear"])
-        )
+        return super().available and self.coordinator.data and self.get_gear()
 
     def get_stats(self):
+        """Get gear statistics from garmin"""
         for gear_stats_item in self.coordinator.data["gear_stats"]:
-            if gear_stats_item["uuid"] == self._uuid:
+            if gear_stats_item[GEAR.UUID] == self._uuid:
                 return gear_stats_item
 
     def get_gear(self):
+        """Get gear from garmin"""
         for gear_item in self.coordinator.data["gear"]:
-            if gear_item["uuid"] == self._uuid:
+            if gear_item[GEAR.UUID] == self._uuid:
                 return gear_item

custom_components/garmin_connect/services.yaml

@@ -0,0 +1,39 @@
+set_active_gear:
+  name: Set active gear for activity
+  # target:
+  #   entity:
+  #     integration: "garmin_connect"
+  fields:
+    activity_type:
+      required: true
+      name: activity type
+      description: garmin activity type
+      example: running
+      default: running
+      selector:
+        select:
+          options:
+            - running
+            - cycling
+            - hiking
+            - other
+            - walking
+            - swimming
+    setting:
+      required: true
+      name: setting
+      description: gear setting to apply
+      default: set this as default, unset others
+      selector:
+        select:
+          options:
+            - set this as default, unset others
+            - set as default
+            - unset default
+    entity_id:
+      description: entity
+      required: true
+      selector:
+        entity:
+          integration: garmin_connect
+          device_class: Gear