add departure info feature

myskoda/anonymize.py

@@ -84,6 +84,10 @@ def anonymize_auxiliary_heating(data: dict) -> dict:
     return data
 
 
+def anonymize_departure_timers(data: dict) -> dict:
+    return data
+
+
 def anonymize_positions(data: dict) -> dict:
     if "positions" in data:
         for position in data["positions"]:

myskoda/cli/__init__.py

@@ -23,6 +23,7 @@
     set_ac_at_unlock,
     set_ac_without_external_power,
     set_charge_limit,
+    set_departure_timer,
     set_reduced_current_limit,
     set_seats_heating,
     set_target_temperature,
@@ -41,6 +42,7 @@
     auth,
     auxiliary_heating,
     charging,
+    departure_timers,
     driving_range,
     garage,
     health,
@@ -171,6 +173,9 @@ async def disconnect(  # noqa: PLR0913
 cli.add_command(set_ac_at_unlock)
 cli.add_command(set_windows_heating)
 cli.add_command(set_seats_heating)
+cli.add_command(departure_timers)
+cli.add_command(set_departure_timer)
+
 
 if __name__ == "__main__":
     cli()

myskoda/cli/operations.py

@@ -338,3 +338,38 @@ async def set_seats_heating(
     )
     async with asyncio.timeout(timeout):
         await myskoda.set_seats_heating(vin, settings)
+
+
+@click.command()
+@click.option("timeout", "--timeout", type=float, default=300)
+@click.argument("vin")
+@click.option("timer", "--timer", type=click.Choice(["1", "2", "3"]), required=True)
+@click.option("enabled", "--enabled", type=bool, required=True)
+@click.pass_context
+@mqtt_required
+async def set_departure_timer(
+    ctx: Context,
+    timeout: float,  # noqa: ASYNC109
+    vin: str,
+    timer: str,
+    enabled: bool,
+) -> None:
+    """Enable or disable selected departure timer."""
+    timer_id = int(timer)
+    myskoda: MySkoda = ctx.obj["myskoda"]
+    async with asyncio.timeout(timeout):
+        # Get all timers from vehicle first
+        departure_info = await myskoda.get_departure_timers(vin)
+        if departure_info is not None:
+            selected_timer = (
+                next((t for t in departure_info.timers if t.id == timer_id), None)
+                if departure_info.timers
+                else None
+            )
+            if selected_timer is not None:
+                selected_timer.enabled = enabled
+                await myskoda.set_departure_timer(vin, selected_timer)
+            else:
+                print(f"No timer found with ID {timer_id}.")
+        else:
+            print("No DepartureInfo found for the given VIN.")

myskoda/cli/requests.py

@@ -133,6 +133,15 @@ async def verify_spin(ctx: Context, spin: str, anonymize: bool) -> None:
     await handle_request(ctx, ctx.obj["myskoda"].verify_spin, spin, anonymize)
 
 
+@click.command()
+@click.argument("vin")
+@click.option("anonymize", "--anonymize", help="Strip all personal data.", is_flag=True)
+@click.pass_context
+async def departure_timers(ctx: Context, vin: str, anonymize: bool) -> None:
+    """Get all departure timers."""
+    await handle_request(ctx, ctx.obj["myskoda"].get_departure_timers, vin, anonymize)
+
+
 @click.command()
 @click.pass_context
 async def auth(ctx: Context) -> None:

myskoda/const.py

@@ -28,6 +28,7 @@
     "charging/update-charge-mode",
     "charging/update-charging-profiles",
     "charging/update-charging-current",
+    "departure/update-departure-timers",
     "vehicle-access/honk-and-flash",
     "vehicle-access/lock-vehicle",
     "vehicle-services-backup/apply-backup",

myskoda/models/departure.py

@@ -0,0 +1,109 @@
+"""Models for responses of api/v1/vehicle-automatization/{vin}/departure/ endpoint."""
+
+from dataclasses import dataclass, field
+from datetime import datetime
+from datetime import time as datetime_time
+from typing import Any
+
+from mashumaro import field_options
+from mashumaro.config import BaseConfig
+from mashumaro.mixins.orjson import DataClassORJSONMixin
+
+from .air_conditioning import TemperatureUnit, TimerMode
+from .common import Weekday
+
+
+@dataclass
+class DepartureTemperature(DataClassORJSONMixin):
+    celsius: float | None = field(default=None, metadata=field_options(alias="celsius"))
+    fahrenheit: float | None = field(default=None, metadata=field_options(alias="fahrenheit"))
+    unit_in_car: TemperatureUnit = field(
+        default=TemperatureUnit.CELSIUS, metadata=field_options(alias="unitInCar")
+    )
+
+
+@dataclass
+class ChargingTime(DataClassORJSONMixin):
+    """Information related to DepartureTimer."""
+
+    id: int
+    enabled: bool
+    start_time: datetime_time = field(metadata=field_options(alias="startTime"))
+    end_time: datetime_time = field(metadata=field_options(alias="endTime"))
+
+    class Config(BaseConfig):
+        """Configuration for serialization and deserialization.."""
+
+        serialize_by_alias = True
+
+    def __post_serialize__(self, d: dict[Any, Any]) -> dict[Any, Any]:
+        """Post-process the data before serialization."""
+        if self.start_time:
+            d["startTime"] = self.start_time.strftime("%H:%M")  # Format to hh:mm
+        if self.end_time:
+            d["endTime"] = self.end_time.strftime("%H:%M")  # Format to hh:mm
+        return d
+
+
+@dataclass
+class DepartureTimer(DataClassORJSONMixin):
+    """Information related to DepartureTimer."""
+
+    id: int
+    enabled: bool
+    charging: bool | None = field(default=None, metadata=field_options(alias="charging"))
+    climatisation: bool | None = field(default=None, metadata=field_options(alias="climatisation"))
+    preferred_charging_times: list[ChargingTime] | None = field(
+        default=None, metadata=field_options(alias="preferredChargingTimes")
+    )
+    one_off_day: Weekday | None = field(default=None, metadata=field_options(alias="oneOffDay"))
+    recurring_on: list[Weekday] | None = field(
+        default=None, metadata=field_options(alias="recurringOn")
+    )
+    target_battery_state_of_charge_in_percent: int | None = field(
+        default=None, metadata=field_options(alias="targetBatteryStateOfChargeInPercent")
+    )
+    time: datetime_time | None = field(default=None, metadata=field_options(alias="time"))
+    type: TimerMode | None = field(default=None, metadata=field_options(alias="type"))
+
+    class Config(BaseConfig):
+        """Configuration for serialization and deserialization.."""
+
+        serialize_by_alias = True
+        omit_none = True
+
+    def __post_serialize__(self, d: dict[Any, Any]) -> dict[Any, Any]:
+        """Post-process the data before serialization."""
+        if self.time:
+            d["time"] = self.time.strftime("%H:%M")  # Format to hh:mm
+        return d
+
+
+@dataclass
+class DepartureSettings(DataClassORJSONMixin):
+    """Information related to DepartureSettings."""
+
+    minimum_battery_state_of_charge_in_percent: int | None = field(
+        default=None, metadata=field_options(alias="minimumBatteryStateOfChargeInPercent")
+    )
+
+
+@dataclass
+class DepartureInfo(DataClassORJSONMixin):
+    """Information related to Departure."""
+
+    car_captured_timestamp: datetime | None = field(
+        default=None, metadata=field_options(alias="carCapturedTimestamp")
+    )
+    first_occurring_timer_id: int | None = field(
+        default=None, metadata=field_options(alias="firstOccurringTimerId")
+    )
+    minimum_battery_state_of_charge_in_percent: int | None = field(
+        default=None, metadata=field_options(alias="minimumBatteryStateOfChargeInPercent")
+    )
+    target_temperature: DepartureTemperature | None = field(
+        default=None, metadata=field_options(alias="targetTemperature")
+    )
+    timers: list[DepartureTimer] | None = field(
+        default=None, metadata=field_options(alias="timers")
+    )

myskoda/models/fixtures.py

@@ -25,6 +25,7 @@ class Endpoint(StrEnum):
     MAINTENANCE = "maintenance"
     DRIVING_RANGE = "driving_range"
     TRIP_STATISTICS = "trip_statistics"
+    DEPARTURE_INFO = "departure_info"
     ALL = "all"
 
 

myskoda/myskoda.py

@@ -36,6 +36,7 @@
 )
 from .models.auxiliary_heating import AuxiliaryConfig, AuxiliaryHeating
 from .models.charging import ChargeMode, Charging
+from .models.departure import DepartureInfo, DepartureTimer
 from .models.driving_range import DrivingRange
 from .models.health import Health
 from .models.info import CapabilityId, Info
@@ -281,6 +282,16 @@ async def unlock(self, vin: str, spin: str) -> None:
         await self.rest_api.unlock(vin, spin)
         await future
 
+    async def set_departure_timer(self, vin: str, timer: DepartureTimer) -> None:
+        """Send provided departure timer to the vehicle."""
+        future = self._wait_for_operation(OperationName.UPDATE_DEPARTURE_TIMERS)
+        await self.rest_api.set_departure_timer(vin, timer)
+        await future
+
+    async def get_departure_timers(self, vin: str, anonymize: bool = False) -> DepartureInfo:
+        """Retrieve departure timers for the specified vehicle."""
+        return (await self.rest_api.get_departure_timers(vin, anonymize=anonymize)).result
+
     async def get_auth_token(self) -> str:
         """Retrieve the main access token for the IDK session."""
         return await self.rest_api.authorization.get_access_token()
@@ -350,6 +361,7 @@ async def get_vehicle(self, vin: str) -> Vehicle:
             CapabilityId.STATE,
             CapabilityId.TRIP_STATISTICS,
             CapabilityId.VEHICLE_HEALTH_INSPECTION,
+            CapabilityId.DEPARTURE_TIMERS,
         ]
 
         return await self.get_partial_vehicle(vin, all_capabilities)
@@ -363,25 +375,31 @@ async def get_partial_vehicle(self, vin: str, capabilities: list[CapabilityId])
 
         for capa in capabilities:
             if info.is_capability_available(capa):
-                match capa:
-                    case CapabilityId.AIR_CONDITIONING:
-                        vehicle.air_conditioning = await self.get_air_conditioning(vin)
-                    case CapabilityId.AUXILIARY_HEATING:
-                        vehicle.auxiliary_heating = await self.get_auxiliary_heating(vin)
-                    case CapabilityId.CHARGING:
-                        vehicle.charging = await self.get_charging(vin)
-                    case CapabilityId.PARKING_POSITION:
-                        vehicle.positions = await self.get_positions(vin)
-                    case CapabilityId.STATE:
-                        vehicle.status = await self.get_status(vin)
-                        vehicle.driving_range = await self.get_driving_range(vin)
-                    case CapabilityId.TRIP_STATISTICS:
-                        vehicle.trip_statistics = await self.get_trip_statistics(vin)
-                    case CapabilityId.VEHICLE_HEALTH_INSPECTION:
-                        vehicle.health = await self.get_health(vin)
+                await self._load_capability(vehicle, vin, capa)
 
         return vehicle
 
+    async def _load_capability(self, vehicle: Vehicle, vin: str, capa: CapabilityId) -> None:
+        """Load specific capability into the vehicle."""
+        match capa:
+            case CapabilityId.AIR_CONDITIONING:
+                vehicle.air_conditioning = await self.get_air_conditioning(vin)
+            case CapabilityId.AUXILIARY_HEATING:
+                vehicle.auxiliary_heating = await self.get_auxiliary_heating(vin)
+            case CapabilityId.CHARGING:
+                vehicle.charging = await self.get_charging(vin)
+            case CapabilityId.PARKING_POSITION:
+                vehicle.positions = await self.get_positions(vin)
+            case CapabilityId.STATE:
+                vehicle.status = await self.get_status(vin)
+                vehicle.driving_range = await self.get_driving_range(vin)
+            case CapabilityId.TRIP_STATISTICS:
+                vehicle.trip_statistics = await self.get_trip_statistics(vin)
+            case CapabilityId.VEHICLE_HEALTH_INSPECTION:
+                vehicle.health = await self.get_health(vin)
+            case CapabilityId.DEPARTURE_TIMERS:
+                vehicle.departure_info = await self.get_departure_timers(vin)
+
     async def get_all_vehicles(self) -> list[Vehicle]:
         """Load all vehicles based on their capabilities."""
         vins = await self.list_vehicle_vins()
@@ -428,6 +446,7 @@ async def get_endpoint(
             Endpoint.MAINTENANCE: self.rest_api.get_maintenance,
             Endpoint.DRIVING_RANGE: self.rest_api.get_driving_range,
             Endpoint.TRIP_STATISTICS: self.rest_api.get_trip_statistics,
+            Endpoint.DEPARTURE_INFO: self.rest_api.get_departure_timers,
         }
 
         # Look up the method, or raise an error if unsupported

myskoda/rest_api.py

@@ -5,13 +5,15 @@
 import logging
 from collections.abc import Callable
 from dataclasses import dataclass
+from datetime import UTC, datetime
 
 from aiohttp import ClientResponseError, ClientSession
 
 from myskoda.anonymize import (
     anonymize_air_conditioning,
     anonymize_auxiliary_heating,
     anonymize_charging,
+    anonymize_departure_timers,
     anonymize_driving_range,
     anonymize_garage,
     anonymize_health,
@@ -38,6 +40,7 @@
 )
 from .models.auxiliary_heating import AuxiliaryConfig, AuxiliaryHeating
 from .models.charging import Charging
+from .models.departure import DepartureInfo, DepartureTimer
 from .models.driving_range import DrivingRange
 from .models.health import Health
 from .models.info import Info
@@ -276,6 +279,19 @@ async def get_garage(self, anonymize: bool = False) -> GetEndpointResult[Garage]
         result = self._deserialize(raw, Garage.from_json)
         return GetEndpointResult(url=url, raw=raw, result=result)
 
+    async def get_departure_timers(
+        self, vin: str, anonymize: bool = False
+    ) -> GetEndpointResult[DepartureInfo]:
+        """Retrieve departure timers for the vehicle."""
+        url = f"/v1/vehicle-automatization/{vin}/departure/timers"
+        raw = self.process_json(
+            data=await self._make_get_request(url),
+            anonymize=anonymize,
+            anonymization_fn=anonymize_departure_timers,
+        )
+        result = self._deserialize(raw, DepartureInfo.from_json)
+        return GetEndpointResult(url=url, raw=raw, result=result)
+
     async def _headers(self) -> dict[str, str]:
         return {"authorization": f"Bearer {await self.authorization.get_access_token()}"}
 
@@ -514,6 +530,21 @@ async def flash(
             url=f"/v1/vehicle-access/{vin}/honk-and-flash", json=json_data
         )
 
+    async def set_departure_timer(self, vin: str, timer: DepartureTimer) -> None:
+        """Set departure timer."""
+        _LOGGER.debug(
+            "Setting departure timer #%i for vehicle %s to %r", timer.id, vin, timer.enabled
+        )
+
+        now = datetime.now(UTC)
+        datetime_str = now.isoformat()
+
+        json_data = {"deviceDateTime": datetime_str, "timers": [timer.to_dict()]}
+        await self._make_post_request(
+            url=f"/v1/vehicle-automatization/{vin}/departure/timers",
+            json=json_data,
+        )
+
     def _deserialize[T](self, text: str, deserialize: Callable[[str], T]) -> T:
         try:
             data = deserialize(text)

myskoda/vehicle.py

@@ -3,6 +3,7 @@
 from .models.air_conditioning import AirConditioning
 from .models.auxiliary_heating import AuxiliaryHeating
 from .models.charging import Charging
+from .models.departure import DepartureInfo
 from .models.driving_range import DrivingRange
 from .models.health import Health
 from .models.info import CapabilityId, Info
@@ -25,6 +26,7 @@ class Vehicle:
     trip_statistics: TripStatistics | None = None
     maintenance: Maintenance
     health: Health | None = None
+    departure_info: DepartureInfo | None = None
 
     def __init__(self, info: Info, maintenance: Maintenance) -> None:  # noqa: D107
         self.info = info