Add charging and preconditioning actions to Teslemetry

homeassistant/components/teslemetry/icons.json

@@ -758,6 +758,18 @@
     },
     "time_of_use": {
       "service": "mdi:clock-time-eight-outline"
+    },
+    "add_charge_schedule": {
+      "service": "mdi:calendar-plus"
+    },
+    "remove_charge_schedule": {
+      "service": "mdi:calendar-minus"
+    },
+    "add_precondition_schedule": {
+      "service": "mdi:hvac-outline"
+    },
+    "remove_precondition_schedule": {
+      "service": "mdi:hvac-off-outline"
     }
   }
 }

homeassistant/components/teslemetry/services.py

@@ -22,6 +22,7 @@
 ATTR_GPS = "gps"
 ATTR_TYPE = "type"
 ATTR_VALUE = "value"
+ATTR_LOCATION = "location"
 ATTR_LOCALE = "locale"
 ATTR_ORDER = "order"
 ATTR_TIMESTAMP = "timestamp"
@@ -36,6 +37,12 @@
 ATTR_OFF_PEAK_CHARGING_ENABLED = "off_peak_charging_enabled"
 ATTR_OFF_PEAK_CHARGING_WEEKDAYS = "off_peak_charging_weekdays_only"
 ATTR_END_OFF_PEAK_TIME = "end_off_peak_time"
+ATTR_DAYS_OF_WEEK = "days_of_week"
+ATTR_START_TIME = "start_time"
+ATTR_END_TIME = "end_time"
+ATTR_ONE_TIME = "one_time"
+ATTR_NAME = "name"
+ATTR_PRECONDITION_TIME = "precondition_time"
 
 # Services
 SERVICE_NAVIGATE_ATTR_GPS_REQUEST = "navigation_gps_request"
@@ -44,6 +51,10 @@
 SERVICE_VALET_MODE = "valet_mode"
 SERVICE_SPEED_LIMIT = "speed_limit"
 SERVICE_TIME_OF_USE = "time_of_use"
+SERVICE_ADD_CHARGE_SCHEDULE = "add_charge_schedule"
+SERVICE_REMOVE_CHARGE_SCHEDULE = "remove_charge_schedule"
+SERVICE_ADD_PRECONDITION_SCHEDULE = "add_precondition_schedule"
+SERVICE_REMOVE_PRECONDITION_SCHEDULE = "remove_precondition_schedule"
 
 
 def async_get_device_for_service_call(
@@ -137,18 +148,15 @@ async def set_scheduled_charging(call: ServiceCall) -> None:
         config = async_get_config_for_device(hass, device)
         vehicle = async_get_vehicle_for_entry(hass, device, config)
 
-        time: int | None = None
-        # Convert time to minutes since minute
-        if "time" in call.data:
-            (hours, minutes, *seconds) = call.data["time"].split(":")
-            time = int(hours) * 60 + int(minutes)
-        elif call.data["enable"]:
-            raise ServiceValidationError(
-                translation_domain=DOMAIN, translation_key="set_scheduled_charging_time"
-            )
+        charge_time: int | None = None
+        # Convert time to minutes since midnight
+        if time_obj := call.data.get(ATTR_TIME):
+            charge_time = time_obj.hour * 60 + time_obj.minute
 
         await handle_vehicle_command(
-            vehicle.api.set_scheduled_charging(enable=call.data["enable"], time=time)
+            vehicle.api.set_scheduled_charging(
+                enable=call.data[ATTR_ENABLE], time=charge_time
+            )
         )
 
     hass.services.async_register(
@@ -159,7 +167,7 @@ async def set_scheduled_charging(call: ServiceCall) -> None:
             {
                 vol.Required(CONF_DEVICE_ID): cv.string,
                 vol.Required(ATTR_ENABLE): bool,
-                vol.Optional(ATTR_TIME): str,
+                vol.Optional(ATTR_TIME): cv.time,
             }
         ),
     )
@@ -170,37 +178,27 @@ async def set_scheduled_departure(call: ServiceCall) -> None:
         config = async_get_config_for_device(hass, device)
         vehicle = async_get_vehicle_for_entry(hass, device, config)
 
-        enable = call.data.get("enable", True)
+        enable = call.data.get(ATTR_ENABLE, True)
 
         # Preconditioning
         preconditioning_enabled = call.data.get(ATTR_PRECONDITIONING_ENABLED, False)
         preconditioning_weekdays_only = call.data.get(
             ATTR_PRECONDITIONING_WEEKDAYS, False
         )
-        departure_time: int | None = None
-        if ATTR_DEPARTURE_TIME in call.data:
-            (hours, minutes, *seconds) = call.data[ATTR_DEPARTURE_TIME].split(":")
-            departure_time = int(hours) * 60 + int(minutes)
-        elif preconditioning_enabled:
-            raise ServiceValidationError(
-                translation_domain=DOMAIN,
-                translation_key="set_scheduled_departure_preconditioning",
-            )
+        departure_time: int = 0
+        if departure_time_obj := call.data.get(ATTR_DEPARTURE_TIME):
+            departure_time = departure_time_obj.hour * 60 + departure_time_obj.minute
 
         # Off peak charging
         off_peak_charging_enabled = call.data.get(ATTR_OFF_PEAK_CHARGING_ENABLED, False)
         off_peak_charging_weekdays_only = call.data.get(
             ATTR_OFF_PEAK_CHARGING_WEEKDAYS, False
         )
-        end_off_peak_time: int | None = None
+        end_off_peak_time: int = 0
 
-        if ATTR_END_OFF_PEAK_TIME in call.data:
-            (hours, minutes, *seconds) = call.data[ATTR_END_OFF_PEAK_TIME].split(":")
-            end_off_peak_time = int(hours) * 60 + int(minutes)
-        elif off_peak_charging_enabled:
-            raise ServiceValidationError(
-                translation_domain=DOMAIN,
-                translation_key="set_scheduled_departure_off_peak",
+        if end_off_peak_time_obj := call.data.get(ATTR_END_OFF_PEAK_TIME):
+            end_off_peak_time = (
+                end_off_peak_time_obj.hour * 60 + end_off_peak_time_obj.minute
             )
 
         await handle_vehicle_command(
@@ -225,10 +223,10 @@ async def set_scheduled_departure(call: ServiceCall) -> None:
                 vol.Optional(ATTR_ENABLE): bool,
                 vol.Optional(ATTR_PRECONDITIONING_ENABLED): bool,
                 vol.Optional(ATTR_PRECONDITIONING_WEEKDAYS): bool,
-                vol.Optional(ATTR_DEPARTURE_TIME): str,
+                vol.Optional(ATTR_DEPARTURE_TIME): cv.time,
                 vol.Optional(ATTR_OFF_PEAK_CHARGING_ENABLED): bool,
                 vol.Optional(ATTR_OFF_PEAK_CHARGING_WEEKDAYS): bool,
-                vol.Optional(ATTR_END_OFF_PEAK_TIME): str,
+                vol.Optional(ATTR_END_OFF_PEAK_TIME): cv.time,
             }
         ),
     )
@@ -314,3 +312,195 @@ async def time_of_use(call: ServiceCall) -> None:
             }
         ),
     )
+
+    async def add_charge_schedule(call: ServiceCall) -> None:
+        """Configure charging schedule for a vehicle."""
+        device = async_get_device_for_service_call(hass, call)
+        config = async_get_config_for_device(hass, device)
+        vehicle = async_get_vehicle_for_entry(hass, device, config)
+
+        # Extract parameters from the service call
+        days_of_week = call.data[ATTR_DAYS_OF_WEEK]
+        # If days_of_week is a list (from select with multiple), convert to comma-separated string
+        if isinstance(days_of_week, list):
+            days_of_week = ",".join(days_of_week)
+        enabled = call.data[ATTR_ENABLE]
+
+        # Optional parameters
+        location = call.data.get(
+            ATTR_LOCATION,
+            {
+                CONF_LATITUDE: hass.config.latitude,
+                CONF_LONGITUDE: hass.config.longitude,
+            },
+        )
+
+        # Handle time inputs
+        start_time = None
+        if start_time_obj := call.data.get(ATTR_START_TIME):
+            # Convert time object to minutes since midnight
+            start_time = start_time_obj.hour * 60 + start_time_obj.minute
+
+        end_time = None
+        if end_time_obj := call.data.get(ATTR_END_TIME):
+            # Convert time object to minutes since midnight
+            end_time = end_time_obj.hour * 60 + end_time_obj.minute
+
+        one_time = call.data.get(ATTR_ONE_TIME)
+        schedule_id = call.data.get(ATTR_ID)
+        name = call.data.get(ATTR_NAME)
+
+        await handle_vehicle_command(
+            vehicle.api.add_charge_schedule(
+                days_of_week=days_of_week,
+                enabled=enabled,
+                lat=location[CONF_LATITUDE],
+                lon=location[CONF_LONGITUDE],
+                start_time=start_time,
+                end_time=end_time,
+                one_time=one_time,
+                id=schedule_id,
+                name=name,
+            )
+        )
+
+    hass.services.async_register(
+        DOMAIN,
+        SERVICE_ADD_CHARGE_SCHEDULE,
+        add_charge_schedule,
+        schema=vol.Schema(
+            {
+                vol.Required(CONF_DEVICE_ID): cv.string,
+                vol.Required(ATTR_DAYS_OF_WEEK): cv.ensure_list,
+                vol.Required(ATTR_ENABLE): cv.boolean,
+                vol.Optional(ATTR_LOCATION): {
+                    vol.Required(CONF_LATITUDE): cv.latitude,
+                    vol.Required(CONF_LONGITUDE): cv.longitude,
+                },
+                vol.Optional(ATTR_START_TIME): cv.time,
+                vol.Optional(ATTR_END_TIME): cv.time,
+                vol.Optional(ATTR_ONE_TIME): cv.boolean,
+                vol.Optional(ATTR_ID): cv.positive_int,
+                vol.Optional(ATTR_NAME): cv.string,
+            }
+        ),
+    )
+
+    async def remove_charge_schedule(call: ServiceCall) -> None:
+        """Remove a charging schedule for a vehicle."""
+        device = async_get_device_for_service_call(hass, call)
+        config = async_get_config_for_device(hass, device)
+        vehicle = async_get_vehicle_for_entry(hass, device, config)
+
+        # Extract parameters from the service call
+        schedule_id = call.data[ATTR_ID]
+
+        await handle_vehicle_command(
+            vehicle.api.remove_charge_schedule(
+                id=schedule_id,
+            )
+        )
+
+    hass.services.async_register(
+        DOMAIN,
+        SERVICE_REMOVE_CHARGE_SCHEDULE,
+        remove_charge_schedule,
+        schema=vol.Schema(
+            {
+                vol.Required(CONF_DEVICE_ID): cv.string,
+                vol.Required(ATTR_ID): cv.positive_int,
+            }
+        ),
+    )
+
+    async def add_precondition_schedule(call: ServiceCall) -> None:
+        """Add or modify a precondition schedule for a vehicle."""
+        device = async_get_device_for_service_call(hass, call)
+        config = async_get_config_for_device(hass, device)
+        vehicle = async_get_vehicle_for_entry(hass, device, config)
+
+        # Extract parameters from the service call
+        days_of_week = call.data[ATTR_DAYS_OF_WEEK]
+        # If days_of_week is a list (from select with multiple), convert to comma-separated string
+        if isinstance(days_of_week, list):
+            days_of_week = ",".join(days_of_week)
+        enabled = call.data[ATTR_ENABLE]
+        location = call.data.get(
+            ATTR_LOCATION,
+            {
+                CONF_LATITUDE: hass.config.latitude,
+                CONF_LONGITUDE: hass.config.longitude,
+            },
+        )
+
+        # Convert time object to minutes since midnight
+        precondition_time = (
+            call.data[ATTR_PRECONDITION_TIME].hour * 60
+            + call.data[ATTR_PRECONDITION_TIME].minute
+        )
+
+        # Optional parameters
+        schedule_id = call.data.get(ATTR_ID)
+        one_time = call.data.get(ATTR_ONE_TIME)
+        name = call.data.get(ATTR_NAME)
+
+        await handle_vehicle_command(
+            vehicle.api.add_precondition_schedule(
+                days_of_week=days_of_week,
+                enabled=enabled,
+                lat=location[CONF_LATITUDE],
+                lon=location[CONF_LONGITUDE],
+                precondition_time=precondition_time,
+                id=schedule_id,
+                one_time=one_time,
+                name=name,
+            )
+        )
+
+    hass.services.async_register(
+        DOMAIN,
+        SERVICE_ADD_PRECONDITION_SCHEDULE,
+        add_precondition_schedule,
+        schema=vol.Schema(
+            {
+                vol.Required(CONF_DEVICE_ID): cv.string,
+                vol.Required(ATTR_DAYS_OF_WEEK): cv.ensure_list,
+                vol.Required(ATTR_ENABLE): cv.boolean,
+                vol.Optional(ATTR_LOCATION): {
+                    vol.Required(CONF_LATITUDE): cv.latitude,
+                    vol.Required(CONF_LONGITUDE): cv.longitude,
+                },
+                vol.Required(ATTR_PRECONDITION_TIME): cv.time,
+                vol.Optional(ATTR_ID): cv.positive_int,
+                vol.Optional(ATTR_ONE_TIME): cv.boolean,
+                vol.Optional(ATTR_NAME): cv.string,
+            }
+        ),
+    )
+
+    async def remove_precondition_schedule(call: ServiceCall) -> None:
+        """Remove a preconditioning schedule for a vehicle."""
+        device = async_get_device_for_service_call(hass, call)
+        config = async_get_config_for_device(hass, device)
+        vehicle = async_get_vehicle_for_entry(hass, device, config)
+
+        # Extract parameters from the service call
+        schedule_id = call.data[ATTR_ID]
+
+        await handle_vehicle_command(
+            vehicle.api.remove_precondition_schedule(
+                id=schedule_id,
+            )
+        )
+
+    hass.services.async_register(
+        DOMAIN,
+        SERVICE_REMOVE_PRECONDITION_SCHEDULE,
+        remove_precondition_schedule,
+        schema=vol.Schema(
+            {
+                vol.Required(CONF_DEVICE_ID): cv.string,
+                vol.Required(ATTR_ID): cv.positive_int,
+            }
+        ),
+    )