feat(vehicle): add charging/utility signed commands

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
+                    )
+                )
+            )
+        )

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)