diff --git a/docs/api_endpoints.md b/docs/api_endpoints.md index ef0f3945..8a1caed8 100644 --- a/docs/api_endpoints.md +++ b/docs/api_endpoints.md @@ -161,7 +161,7 @@ | EnergyX | POST | api/v2/air-conditioning/{vin}/timers | ✅ | | | EnergyX | POST | api/v2/air-conditioning/{vin}/settings/windows-heating | ✅ | | | EnergyX | POST | api/v2/air-conditioning/{vin}/settings/ac-without-external-power | ✅ | | -| EnergyX | POST | api/v2/air-conditioning/{vin}/auxiliary-heating/timers | | | +| EnergyX | POST | api/v2/air-conditioning/{vin}/auxiliary-heating/timers | ✅ | | | EnergyX | POST | api/v2/air-conditioning/{vin}/active-ventilation/start | | | | EnergyX | POST | api/v2/air-conditioning/{vin}/start | ✅ | | | EnergyX | POST | api/v2/air-conditioning/{vin}/auxiliary-heating/start | ✅ | | diff --git a/myskoda/cli/__init__.py b/myskoda/cli/__init__.py index 36c936cf..1fb6b8f5 100644 --- a/myskoda/cli/__init__.py +++ b/myskoda/cli/__init__.py @@ -24,6 +24,7 @@ set_ac_timer, set_ac_without_external_power, set_auto_unlock_plug, + set_aux_timer, set_charge_limit, set_departure_timer, set_minimum_charge_limit, @@ -181,6 +182,7 @@ async def disconnect( # noqa: PLR0913 cli.add_command(departure_timers) cli.add_command(set_departure_timer) cli.add_command(set_ac_timer) +cli.add_command(set_aux_timer) if __name__ == "__main__": diff --git a/myskoda/cli/operations.py b/myskoda/cli/operations.py index 35edcc9d..bd14bd42 100644 --- a/myskoda/cli/operations.py +++ b/myskoda/cli/operations.py @@ -439,3 +439,40 @@ async def set_ac_timer( print(f"No timer found with ID {timer_id}.") else: print("No AirConditioning found for the given VIN.") + + +@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.option("spin", "--spin", type=str, required=True) +@click.pass_context +@mqtt_required +async def set_aux_timer( # noqa: PLR0913 + ctx: Context, + timeout: float, # noqa: ASYNC109 + vin: str, + timer: str, + enabled: bool, + spin: str, +) -> None: + """Enable or disable selected auxiliary-heating timer.""" + timer_id = int(timer) + myskoda: MySkoda = ctx.obj["myskoda"] + async with asyncio.timeout(timeout): + # Get all timers from vehicle first + auxiliary_heating = await myskoda.get_auxiliary_heating(vin) + if auxiliary_heating is not None: + selected_timer = ( + next((t for t in auxiliary_heating.timers if t.id == timer_id), None) + if auxiliary_heating.timers + else None + ) + if selected_timer is not None: + selected_timer.enabled = enabled + await myskoda.set_auxiliary_heating_timer(vin, selected_timer, spin) + else: + print(f"No timer found with ID {timer_id}.") + else: + print("No AuxiliaryHeating found for the given VIN.") diff --git a/myskoda/models/auxiliary_heating.py b/myskoda/models/auxiliary_heating.py index 8eeca974..ffdee4d5 100644 --- a/myskoda/models/auxiliary_heating.py +++ b/myskoda/models/auxiliary_heating.py @@ -66,11 +66,16 @@ def __pre_serialize__(self) -> "AuxiliaryConfig": return self +@dataclass +class AuxiliaryHeatingTimer(AirConditioningTimer): + """Timer for auxiliary heating.""" + + @dataclass class AuxiliaryHeating(DataClassORJSONMixin): """Information related to auxiliary heating.""" - timers: list[AirConditioningTimer] + timers: list[AuxiliaryHeatingTimer] errors: list[Any] state: AuxiliaryState | None = field(default=None, metadata=field_options(alias="state")) start_mode: AuxiliaryStartMode | None = field( diff --git a/myskoda/myskoda.py b/myskoda/myskoda.py index 7e742b84..62107619 100644 --- a/myskoda/myskoda.py +++ b/myskoda/myskoda.py @@ -35,7 +35,7 @@ SeatHeating, WindowHeating, ) -from .models.auxiliary_heating import AuxiliaryConfig, AuxiliaryHeating +from .models.auxiliary_heating import AuxiliaryConfig, AuxiliaryHeating, AuxiliaryHeatingTimer from .models.charging import ChargeMode, Charging from .models.departure import DepartureInfo, DepartureTimer from .models.driving_range import DrivingRange @@ -289,6 +289,14 @@ async def set_ac_timer(self, vin: str, timer: AirConditioningTimer) -> None: await self.rest_api.set_ac_timer(vin, timer) await future + async def set_auxiliary_heating_timer( + self, vin: str, timer: AuxiliaryHeatingTimer, spin: str + ) -> None: + """Send provided auxiliary heating timer to the vehicle.""" + future = self._wait_for_operation(OperationName.SET_AIR_CONDITIONING_TIMERS) + await self.rest_api.set_auxiliary_heating_timer(vin, timer, spin) + await future + async def lock(self, vin: str, spin: str) -> None: """Lock the car.""" future = self._wait_for_operation(OperationName.LOCK) diff --git a/myskoda/rest_api.py b/myskoda/rest_api.py index 55efd53e..52ad4293 100644 --- a/myskoda/rest_api.py +++ b/myskoda/rest_api.py @@ -40,7 +40,7 @@ SeatHeating, WindowHeating, ) -from .models.auxiliary_heating import AuxiliaryConfig, AuxiliaryHeating +from .models.auxiliary_heating import AuxiliaryConfig, AuxiliaryHeating, AuxiliaryHeatingTimer from .models.charging import Charging from .models.departure import DepartureInfo, DepartureTimer from .models.driving_range import DrivingRange @@ -597,6 +597,20 @@ async def set_ac_timer(self, vin: str, timer: AirConditioningTimer) -> None: json=json_data, ) + async def set_auxiliary_heating_timer( + self, vin: str, timer: AuxiliaryHeatingTimer, spin: str + ) -> None: + """Set auxiliary heating timer.""" + _LOGGER.debug( + "Setting auxiliary heating timer #%i for vehicle %s to %r", timer.id, vin, timer.enabled + ) + + json_data = {"spin": spin, "timers": [timer.to_dict(by_alias=True)]} + await self._make_post_request( + url=f"/v2/air-conditioning/{vin}/auxiliary-heating/timers", + json=json_data, + ) + def _deserialize[T](self, text: str, deserialize: Callable[[str], T]) -> T: try: data = deserialize(text) diff --git a/tests/test_operations.py b/tests/test_operations.py index 271ddd74..d30689e6 100644 --- a/tests/test_operations.py +++ b/tests/test_operations.py @@ -17,7 +17,7 @@ TargetTemperature, WindowHeating, ) -from myskoda.models.auxiliary_heating import AuxiliaryConfig, AuxiliaryStartMode +from myskoda.models.auxiliary_heating import AuxiliaryConfig, AuxiliaryHeating, AuxiliaryStartMode from myskoda.models.charging import ChargeMode from myskoda.models.departure import DepartureInfo from myskoda.myskoda import MySkoda @@ -742,3 +742,43 @@ async def test_set_ac_timer( headers={"authorization": f"Bearer {ACCESS_TOKEN}"}, json=json_data, ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("timer_id", "enabled", "spin"), [(1, True, "1234"), (2, False, "4321")]) +async def test_set_auxiliary_heating_timer( # noqa: PLR0913 + responses: aioresponses, + mqtt_client: MQTTClient, + myskoda: MySkoda, + timer_id: int, + enabled: bool, + spin: str, +) -> None: + url = f"{BASE_URL_SKODA}/api/v2/air-conditioning/{VIN}/auxiliary-heating/timers" + responses.post(url=url) + + aux_info_json = FIXTURES_DIR.joinpath("other/auxiliary-heating-idle.json").read_text() + aux_info = AuxiliaryHeating.from_json(aux_info_json) + + selected_timer = ( + next((timer for timer in aux_info.timers if timer.id == timer_id), None) + if aux_info.timers + else None + ) + assert selected_timer is not None + + selected_timer.enabled = enabled + future = myskoda.set_auxiliary_heating_timer(VIN, selected_timer, spin) + + topic = f"{USER_ID}/{VIN}/operation-request/air-conditioning/set-air-conditioning-timers" + await mqtt_client.publish(topic, create_completed_json("set-air-conditioning-timers"), QOS_2) + + json_data = {"spin": spin, "timers": [selected_timer.to_dict(by_alias=True)]} + + await future + responses.assert_called_with( + url=url, + method="POST", + headers={"authorization": f"Bearer {ACCESS_TOKEN}"}, + json=json_data, + )