From 750b3ea87b8ee8d0affc3b006345bc9c8dbc9db0 Mon Sep 17 00:00:00 2001 From: Giovanni Condello Date: Sat, 17 Oct 2026 15:10:56 +0200 Subject: [PATCH] fix: do not report cable lock state for models that do not provide it --- .../drivetrain_charging_cable_lock.py | 4 ++ src/status_publisher/charge/chrg_mgmt_data.py | 7 ++- src/vehicle.py | 5 ++ .../test_drivetrain_charging_cable_lock.py | 49 +++++++++++++++++++ tests/test_vehicle_state.py | 20 ++++++++ 5 files changed, 84 insertions(+), 1 deletion(-) create mode 100644 tests/handlers/command/test_drivetrain_charging_cable_lock.py diff --git a/src/handlers/command/drivetrain/drivetrain_charging_cable_lock.py b/src/handlers/command/drivetrain/drivetrain_charging_cable_lock.py index ebc7f9a9..554389b2 100644 --- a/src/handlers/command/drivetrain/drivetrain_charging_cable_lock.py +++ b/src/handlers/command/drivetrain/drivetrain_charging_cable_lock.py @@ -23,11 +23,15 @@ def topic(cls) -> str: async def handle_true(self) -> None: LOG.info(f"Vehicle {self.vin} charging cable will be locked") await self.saic_api.control_charging_port_lock(self.vin, unlock=False) + # Some models never report the lock state, so publish the commanded one. + # Models that do report it overwrite it on the forced refresh. + self.vehicle_state.update_charging_cable_lock(locked=True) @override async def handle_false(self) -> None: LOG.info(f"Vehicle {self.vin} charging cable will be unlocked") await self.saic_api.control_charging_port_lock(self.vin, unlock=True) + self.vehicle_state.update_charging_cable_lock(locked=False) @override async def _get_action_result(self, _action_result: None) -> CommandProcessingResult: diff --git a/src/status_publisher/charge/chrg_mgmt_data.py b/src/status_publisher/charge/chrg_mgmt_data.py index 5825590e..b3a24822 100644 --- a/src/status_publisher/charge/chrg_mgmt_data.py +++ b/src/status_publisher/charge/chrg_mgmt_data.py @@ -160,7 +160,12 @@ def publish(self, charge_mgmt_data: ChrgMgmtData) -> ChrgMgmtDataProcessingResul self._transform_and_publish( topic=mqtt_topics.DRIVETRAIN_CHARGING_CABLE_LOCK, - value=charge_mgmt_data.charging_port_locked, + # Some models (e.g. MG ZS EV) do not report the lock state at all + value=( + charge_mgmt_data.charging_port_locked + if charge_mgmt_data.ccuEleccLckCtrlDspCmd is not None + else None + ), transform=int_to_bool, ) diff --git a/src/vehicle.py b/src/vehicle.py index afa1276f..b396e9ac 100644 --- a/src/vehicle.py +++ b/src/vehicle.py @@ -328,6 +328,11 @@ def set_is_charging(self, is_charging: bool) -> None: self.get_topic(mqtt_topics.DRIVETRAIN_CHARGING), self.is_charging ) + def update_charging_cable_lock(self, *, locked: bool) -> None: + self.publisher.publish_bool( + self.get_topic(mqtt_topics.DRIVETRAIN_CHARGING_CABLE_LOCK), locked + ) + def handle_vehicle_status( self, vehicle_status: VehicleStatusResp ) -> VehicleStatusRespProcessingResult: diff --git a/tests/handlers/command/test_drivetrain_charging_cable_lock.py b/tests/handlers/command/test_drivetrain_charging_cable_lock.py new file mode 100644 index 00000000..6dab3334 --- /dev/null +++ b/tests/handlers/command/test_drivetrain_charging_cable_lock.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +import unittest +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from handlers.command.base import RESULT_REFRESH_AND_CLEAR +from handlers.command.drivetrain.drivetrain_charging_cable_lock import ( + DrivetrainChargingCableLockCommand, +) + + +class TestDrivetrainChargingCableLockCommand(unittest.IsolatedAsyncioTestCase): + async def test_lock_publishes_locked_state(self) -> None: + vehicle_state = MagicMock() + saic_api = AsyncMock() + handler = DrivetrainChargingCableLockCommand(saic_api, vehicle_state) + + result = await handler.handle("true") + + assert result == RESULT_REFRESH_AND_CLEAR + saic_api.control_charging_port_lock.assert_called_once_with( + vehicle_state.vin, unlock=False + ) + vehicle_state.update_charging_cable_lock.assert_called_once_with(locked=True) + + async def test_unlock_publishes_unlocked_state(self) -> None: + vehicle_state = MagicMock() + saic_api = AsyncMock() + handler = DrivetrainChargingCableLockCommand(saic_api, vehicle_state) + + await handler.handle("false") + + saic_api.control_charging_port_lock.assert_called_once_with( + vehicle_state.vin, unlock=True + ) + vehicle_state.update_charging_cable_lock.assert_called_once_with(locked=False) + + async def test_failed_command_does_not_publish_state(self) -> None: + vehicle_state = MagicMock() + saic_api = AsyncMock() + saic_api.control_charging_port_lock.side_effect = RuntimeError("boom") + handler = DrivetrainChargingCableLockCommand(saic_api, vehicle_state) + + with pytest.raises(RuntimeError): + await handler.handle("true") + + vehicle_state.update_charging_cable_lock.assert_not_called() diff --git a/tests/test_vehicle_state.py b/tests/test_vehicle_state.py index 2cc89835..662edf10 100644 --- a/tests/test_vehicle_state.py +++ b/tests/test_vehicle_state.py @@ -55,6 +55,26 @@ def _charge_data_availability(self) -> str | None: topic = f"/vehicles/{VIN}/{mqtt_topics.DRIVETRAIN_CHARGING_DATA_AVAILABLE}" return self.publisher.map.get(topic) + def _cable_lock_topic(self) -> str: + return f"/vehicles/{VIN}/{mqtt_topics.DRIVETRAIN_CHARGING_CABLE_LOCK}" + + def test_cable_lock_published_when_reported(self) -> None: + self.vehicle_state.handle_charge_status(get_mock_charge_management_data_resp()) + assert self.publisher.map[self._cable_lock_topic()] is True + + def test_cable_lock_not_published_when_not_reported(self) -> None: + resp = get_mock_charge_management_data_resp() + assert resp.chrgMgmtData is not None + resp.chrgMgmtData.ccuEleccLckCtrlDspCmd = None + self.vehicle_state.handle_charge_status(resp) + assert self._cable_lock_topic() not in self.publisher.map + + def test_update_charging_cable_lock_publishes_commanded_state(self) -> None: + self.vehicle_state.update_charging_cable_lock(locked=True) + assert self.publisher.map[self._cable_lock_topic()] is True + self.vehicle_state.update_charging_cable_lock(locked=False) + assert self.publisher.map[self._cable_lock_topic()] is False + def _vehicle_availability(self) -> str | None: return self.publisher.map.get(f"/vehicles/{VIN}/{mqtt_topics.AVAILABLE}")