diff --git a/tesla_fleet_api/tesla/vehicle/commands.py b/tesla_fleet_api/tesla/vehicle/commands.py index 390a0cf..b956140 100644 --- a/tesla_fleet_api/tesla/vehicle/commands.py +++ b/tesla_fleet_api/tesla/vehicle/commands.py @@ -180,6 +180,15 @@ TeslaAuthResponseAction, SetupCloudProfileWithLocalProfileUuidAction, GetLocalProfilesForVaultUuidAction, + SetRateTariffRequest, + GetRateTariffRequest, + AddManagedChargingSiteRequest, + RemoveManagedChargingSiteRequest, + GetManagedChargingSitesRequest, + SetDischargeLimitAction, + ManagedChargingSite, + ManagerType, + SiteController, ) from google.protobuf.timestamp_pb2 import Timestamp from tesla_protocol.command.vehicle_pb2 import ( @@ -2515,3 +2524,89 @@ async def get_local_profiles_for_vault_uuid( ), mutating=False, ) + + async def set_rate_tariff( + self, + seasons: SetRateTariffRequest.Seasons, + tariff: SetRateTariffRequest.Tariff | None = None, + ) -> dict[str, Any]: + """Sets a time-of-use rate tariff schedule for charge-on-solar-style optimization. + + ``seasons``/``tariff`` are ``tesla_protocol`` message types + (``SetRateTariffRequest.Seasons``/``.Tariff``) - the tariff schedule is + deeply nested (up to 5 named seasons, each with 4 time-of-use period + types), so construct them directly rather than through a parallel + flattened API. + """ + action = SetRateTariffRequest(seasons=seasons) + if tariff is not None: + action.tariff.CopyFrom(tariff) + return await self._sendInfotainment( + Action(vehicleAction=VehicleAction(setRateTariffRequest=action)) + ) + + async def get_rate_tariff(self) -> dict[str, Any]: + """Gets the current time-of-use rate tariff schedule.""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction(getRateTariffRequest=GetRateTariffRequest()) + ), + mutating=False, + ) + + async def add_managed_charging_site( + self, public_key: str, lat: float, lon: float + ) -> dict[str, Any]: + """Registers a managed charging site (utility managed-charging program) for this vehicle.""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + addManagedChargingSiteRequest=AddManagedChargingSiteRequest( + site=ManagedChargingSite( + public_key=public_key, + manager_type=ManagerType(site_controller=SiteController()), + lat_lon=LatLong(latitude=lat, longitude=lon), + ) + ) + ) + ) + ) + + async def remove_managed_charging_site(self, public_key: str) -> dict[str, Any]: + """Removes a previously-registered managed charging site by its public key.""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + removeManagedChargingSiteRequest=RemoveManagedChargingSiteRequest( + public_key=public_key + ) + ) + ) + ) + + async def get_managed_charging_sites(self) -> dict[str, Any]: + """Gets the list of registered managed charging sites.""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + getManagedChargingSitesRequest=GetManagedChargingSitesRequest() + ) + ), + mutating=False, + ) + + async def set_discharge_limit(self, discharge_limit: int) -> dict[str, Any]: + """Sets the vehicle's general discharge limit. + + Not the same feature as ``set_powershare_discharge_limit`` + (``SetPowershareDischargeLimitAction``, a distinct proto message). + """ + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + setDischargeLimitAction=SetDischargeLimitAction( + discharge_limit=discharge_limit + ) + ) + ) + ) diff --git a/tests/test_ble_charging_utility_commands.py b/tests/test_ble_charging_utility_commands.py new file mode 100644 index 0000000..19ecdd4 --- /dev/null +++ b/tests/test_ble_charging_utility_commands.py @@ -0,0 +1,114 @@ +"""Charging/utility commands over the mocked BLE transport. + +``set_rate_tariff``/``add_managed_charging_site`` accept ``tesla_protocol`` +message types directly for their deeply-nested arguments rather than a +parallel flattened API - see ``commands.py`` docstrings. +""" + +from tesla_protocol.command.car_server_pb2 import Action, SetRateTariffRequest +from tesla_protocol.command.universal_message_pb2 import Domain + +from ble_mocked_transport import ( + MockedBleTransportTestCase, + decrypt_sent_command, + infotainment_action_ok_reply, +) + + +def _decode_vehicle_action(vehicle, sent_msg): + plaintext = decrypt_sent_command(vehicle, sent_msg) + action = Action.FromString(plaintext) + assert sent_msg.to_destination.domain == Domain.DOMAIN_INFOTAINMENT + return action.vehicleAction + + +class SetRateTariffTests(MockedBleTransportTestCase): + async def test_sends_seasons_only(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + seasons = SetRateTariffRequest.Seasons( + Summer=SetRateTariffRequest.Season(from_month=6, to_month=8) + ) + await vehicle.set_rate_tariff(seasons) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + action = vehicle_action.setRateTariffRequest + self.assertEqual(action.seasons.Summer.from_month, 6) + self.assertEqual(action.seasons.Summer.to_month, 8) + self.assertFalse(action.HasField("tariff")) + + async def test_sends_tariff_when_given(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + seasons = SetRateTariffRequest.Seasons() + tariff = SetRateTariffRequest.Tariff(seasons=seasons) + await vehicle.set_rate_tariff(seasons, tariff) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + action = vehicle_action.setRateTariffRequest + self.assertTrue(action.HasField("tariff")) + + +class GetRateTariffTests(MockedBleTransportTestCase): + async def test_sends_get_rate_tariff_request(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.get_rate_tariff() + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertTrue(vehicle_action.HasField("getRateTariffRequest")) + + +class AddManagedChargingSiteTests(MockedBleTransportTestCase): + async def test_sends_public_key_and_coordinates(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.add_managed_charging_site("pubkey-bytes", 37.3230, -122.0322) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + site = vehicle_action.addManagedChargingSiteRequest.site + self.assertEqual(site.public_key, "pubkey-bytes") + self.assertTrue(site.manager_type.HasField("site_controller")) + # LatLong lat/lon are 32-bit floats, so compare at reduced precision. + self.assertAlmostEqual(site.lat_lon.latitude, 37.3230, places=4) + self.assertAlmostEqual(site.lat_lon.longitude, -122.0322, places=4) + + +class RemoveManagedChargingSiteTests(MockedBleTransportTestCase): + async def test_sends_public_key(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.remove_managed_charging_site("pubkey-bytes") + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertEqual( + vehicle_action.removeManagedChargingSiteRequest.public_key, + "pubkey-bytes", + ) + + +class GetManagedChargingSitesTests(MockedBleTransportTestCase): + async def test_sends_get_managed_charging_sites_request(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.get_managed_charging_sites() + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertTrue(vehicle_action.HasField("getManagedChargingSitesRequest")) + + +class SetDischargeLimitTests(MockedBleTransportTestCase): + async def test_sends_discharge_limit(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.set_discharge_limit(50) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertEqual(vehicle_action.setDischargeLimitAction.discharge_limit, 50)