mirror of
https://github.com/cyberjunky/home-assistant-garmin_connect.git
synced 2026-01-08 20:38:00 -05:00
324 lines
11 KiB
Python
324 lines
11 KiB
Python
"""Services for Garmin Connect integration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import TYPE_CHECKING
|
|
|
|
import voluptuous as vol
|
|
from homeassistant.core import HomeAssistant, ServiceCall
|
|
from homeassistant.exceptions import HomeAssistantError
|
|
from homeassistant.helpers import config_validation as cv
|
|
|
|
from .const import DOMAIN
|
|
|
|
if TYPE_CHECKING:
|
|
from .coordinator import GarminConnectDataUpdateCoordinator
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
# Service schemas
|
|
SERVICE_ADD_BODY_COMPOSITION = "add_body_composition"
|
|
SERVICE_ADD_BLOOD_PRESSURE = "add_blood_pressure"
|
|
SERVICE_CREATE_ACTIVITY = "create_activity"
|
|
|
|
ADD_BODY_COMPOSITION_SCHEMA = vol.Schema(
|
|
{
|
|
vol.Required("weight"): vol.Coerce(float),
|
|
vol.Optional("timestamp"): cv.string,
|
|
vol.Optional("percent_fat"): vol.Coerce(float),
|
|
vol.Optional("percent_hydration"): vol.Coerce(float),
|
|
vol.Optional("visceral_fat_mass"): vol.Coerce(float),
|
|
vol.Optional("bone_mass"): vol.Coerce(float),
|
|
vol.Optional("muscle_mass"): vol.Coerce(float),
|
|
vol.Optional("basal_met"): vol.Coerce(float),
|
|
vol.Optional("active_met"): vol.Coerce(float),
|
|
vol.Optional("physique_rating"): vol.Coerce(float),
|
|
vol.Optional("metabolic_age"): vol.Coerce(float),
|
|
vol.Optional("visceral_fat_rating"): vol.Coerce(float),
|
|
vol.Optional("bmi"): vol.Coerce(float),
|
|
}
|
|
)
|
|
|
|
ADD_BLOOD_PRESSURE_SCHEMA = vol.Schema(
|
|
{
|
|
vol.Required("systolic"): int,
|
|
vol.Required("diastolic"): int,
|
|
vol.Required("pulse"): int,
|
|
vol.Optional("timestamp"): cv.string,
|
|
vol.Optional("notes"): cv.string,
|
|
}
|
|
)
|
|
|
|
CREATE_ACTIVITY_SCHEMA = vol.Schema(
|
|
{
|
|
vol.Required("activity_name"): cv.string,
|
|
vol.Required("activity_type"): cv.string,
|
|
vol.Required("start_datetime"): cv.string,
|
|
vol.Required("duration_min"): int,
|
|
vol.Optional("distance_km", default=0.0): vol.Coerce(float),
|
|
vol.Optional("time_zone"): cv.string,
|
|
}
|
|
)
|
|
|
|
SERVICE_UPLOAD_ACTIVITY = "upload_activity"
|
|
UPLOAD_ACTIVITY_SCHEMA = vol.Schema(
|
|
{
|
|
vol.Required("file_path"): cv.string,
|
|
}
|
|
)
|
|
|
|
SERVICE_ADD_GEAR_TO_ACTIVITY = "add_gear_to_activity"
|
|
ADD_GEAR_TO_ACTIVITY_SCHEMA = vol.Schema(
|
|
{
|
|
vol.Required("activity_id"): vol.Coerce(int),
|
|
vol.Required("gear_uuid"): cv.string,
|
|
}
|
|
)
|
|
|
|
|
|
def _get_coordinator(hass: HomeAssistant) -> GarminConnectDataUpdateCoordinator:
|
|
"""Get the first available coordinator from config entries."""
|
|
entries = hass.config_entries.async_entries(DOMAIN)
|
|
if not entries:
|
|
raise HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="no_integration_configured",
|
|
)
|
|
|
|
# Use the first config entry's coordinator
|
|
entry = entries[0]
|
|
if not hasattr(entry, "runtime_data") or entry.runtime_data is None:
|
|
raise HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="integration_not_loaded",
|
|
)
|
|
|
|
return entry.runtime_data # type: ignore[no-any-return]
|
|
|
|
|
|
async def async_setup_services(hass: HomeAssistant) -> None:
|
|
"""Set up Garmin Connect services."""
|
|
|
|
async def handle_add_body_composition(call: ServiceCall) -> None:
|
|
"""Handle add_body_composition service call."""
|
|
coordinator = _get_coordinator(hass)
|
|
|
|
weight = call.data.get("weight")
|
|
timestamp = call.data.get("timestamp")
|
|
percent_fat = call.data.get("percent_fat")
|
|
percent_hydration = call.data.get("percent_hydration")
|
|
visceral_fat_mass = call.data.get("visceral_fat_mass")
|
|
bone_mass = call.data.get("bone_mass")
|
|
muscle_mass = call.data.get("muscle_mass")
|
|
basal_met = call.data.get("basal_met")
|
|
active_met = call.data.get("active_met")
|
|
physique_rating = call.data.get("physique_rating")
|
|
metabolic_age = call.data.get("metabolic_age")
|
|
visceral_fat_rating = call.data.get("visceral_fat_rating")
|
|
bmi = call.data.get("bmi")
|
|
|
|
if not await coordinator.async_login():
|
|
raise HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="login_failed",
|
|
)
|
|
|
|
try:
|
|
await hass.async_add_executor_job(
|
|
coordinator.api.add_body_composition,
|
|
timestamp,
|
|
weight,
|
|
percent_fat,
|
|
percent_hydration,
|
|
visceral_fat_mass,
|
|
bone_mass,
|
|
muscle_mass,
|
|
basal_met,
|
|
active_met,
|
|
physique_rating,
|
|
metabolic_age,
|
|
visceral_fat_rating,
|
|
bmi,
|
|
)
|
|
except Exception as err:
|
|
raise HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="add_body_composition_failed",
|
|
translation_placeholders={"error": str(err)},
|
|
) from err
|
|
|
|
async def handle_add_blood_pressure(call: ServiceCall) -> None:
|
|
"""Handle add_blood_pressure service call."""
|
|
coordinator = _get_coordinator(hass)
|
|
|
|
systolic = call.data.get("systolic")
|
|
diastolic = call.data.get("diastolic")
|
|
pulse = call.data.get("pulse")
|
|
timestamp = call.data.get("timestamp")
|
|
notes = call.data.get("notes")
|
|
|
|
if not await coordinator.async_login():
|
|
raise HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="login_failed",
|
|
)
|
|
|
|
try:
|
|
await hass.async_add_executor_job(
|
|
coordinator.api.set_blood_pressure,
|
|
systolic,
|
|
diastolic,
|
|
pulse,
|
|
timestamp,
|
|
notes,
|
|
)
|
|
except Exception as err:
|
|
raise HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="add_blood_pressure_failed",
|
|
translation_placeholders={"error": str(err)},
|
|
) from err
|
|
|
|
async def handle_create_activity(call: ServiceCall) -> None:
|
|
"""Handle create_activity service call."""
|
|
coordinator = _get_coordinator(hass)
|
|
|
|
activity_name = call.data.get("activity_name")
|
|
activity_type = call.data.get("activity_type")
|
|
start_datetime = call.data.get("start_datetime")
|
|
# API requires milliseconds format: "2023-12-02T10:00:00.000"
|
|
if start_datetime and "." not in start_datetime:
|
|
start_datetime = f"{start_datetime}.000"
|
|
duration_min = call.data.get("duration_min")
|
|
distance_km = call.data.get("distance_km", 0.0)
|
|
# Default to HA's configured timezone
|
|
time_zone = call.data.get("time_zone") or str(hass.config.time_zone)
|
|
|
|
if not await coordinator.async_login():
|
|
raise HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="login_failed",
|
|
)
|
|
|
|
try:
|
|
await hass.async_add_executor_job(
|
|
coordinator.api.create_manual_activity,
|
|
start_datetime,
|
|
time_zone,
|
|
activity_type,
|
|
distance_km,
|
|
duration_min,
|
|
activity_name,
|
|
)
|
|
except Exception as err:
|
|
raise HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="create_activity_failed",
|
|
translation_placeholders={"error": str(err)},
|
|
) from err
|
|
|
|
async def handle_upload_activity(call: ServiceCall) -> None:
|
|
"""Handle upload_activity service call."""
|
|
coordinator = _get_coordinator(hass)
|
|
|
|
file_path = call.data.get("file_path")
|
|
|
|
if not await coordinator.async_login():
|
|
raise HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="login_failed",
|
|
)
|
|
|
|
# Check if file exists
|
|
import os
|
|
if not os.path.isfile(file_path):
|
|
raise HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="file_not_found",
|
|
translation_placeholders={"file_path": file_path},
|
|
)
|
|
|
|
try:
|
|
await hass.async_add_executor_job(
|
|
coordinator.api.upload_activity,
|
|
file_path,
|
|
)
|
|
except Exception as err:
|
|
raise HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="upload_activity_failed",
|
|
translation_placeholders={"error": str(err)},
|
|
) from err
|
|
|
|
async def handle_add_gear_to_activity(call: ServiceCall) -> None:
|
|
"""Handle add_gear_to_activity service call."""
|
|
coordinator = _get_coordinator(hass)
|
|
|
|
activity_id = call.data.get("activity_id")
|
|
gear_uuid = call.data.get("gear_uuid")
|
|
|
|
if not await coordinator.async_login():
|
|
raise HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="login_failed",
|
|
)
|
|
|
|
try:
|
|
await hass.async_add_executor_job(
|
|
coordinator.api.add_gear_to_activity,
|
|
gear_uuid,
|
|
activity_id,
|
|
)
|
|
except Exception as err:
|
|
raise HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="add_gear_to_activity_failed",
|
|
translation_placeholders={"error": str(err)},
|
|
) from err
|
|
|
|
# Register services
|
|
hass.services.async_register(
|
|
DOMAIN,
|
|
SERVICE_ADD_BODY_COMPOSITION,
|
|
handle_add_body_composition,
|
|
schema=ADD_BODY_COMPOSITION_SCHEMA,
|
|
)
|
|
|
|
hass.services.async_register(
|
|
DOMAIN,
|
|
SERVICE_ADD_BLOOD_PRESSURE,
|
|
handle_add_blood_pressure,
|
|
schema=ADD_BLOOD_PRESSURE_SCHEMA,
|
|
)
|
|
|
|
hass.services.async_register(
|
|
DOMAIN,
|
|
SERVICE_CREATE_ACTIVITY,
|
|
handle_create_activity,
|
|
schema=CREATE_ACTIVITY_SCHEMA,
|
|
)
|
|
|
|
hass.services.async_register(
|
|
DOMAIN,
|
|
SERVICE_UPLOAD_ACTIVITY,
|
|
handle_upload_activity,
|
|
schema=UPLOAD_ACTIVITY_SCHEMA,
|
|
)
|
|
|
|
hass.services.async_register(
|
|
DOMAIN,
|
|
SERVICE_ADD_GEAR_TO_ACTIVITY,
|
|
handle_add_gear_to_activity,
|
|
schema=ADD_GEAR_TO_ACTIVITY_SCHEMA,
|
|
)
|
|
|
|
|
|
async def async_unload_services(hass: HomeAssistant) -> None:
|
|
"""Unload Garmin Connect services."""
|
|
hass.services.async_remove(DOMAIN, SERVICE_ADD_BODY_COMPOSITION)
|
|
hass.services.async_remove(DOMAIN, SERVICE_ADD_BLOOD_PRESSURE)
|
|
hass.services.async_remove(DOMAIN, SERVICE_CREATE_ACTIVITY)
|
|
hass.services.async_remove(DOMAIN, SERVICE_UPLOAD_ACTIVITY)
|
|
hass.services.async_remove(DOMAIN, SERVICE_ADD_GEAR_TO_ACTIVITY)
|