Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions src/handlers/command/drivetrain/drivetrain_charging_schedule.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
from saic_ismart_client_ng.api.vehicle_charging import ScheduledChargingMode

from handlers.command.base import (
RESULT_DO_NOTHING,
RESULT_REFRESH_ONLY,
CommandProcessingResult,
PayloadConvertingCommandHandler,
Expand Down Expand Up @@ -54,6 +55,15 @@ def convert_payload(payload: str) -> ChargingScheduleCommandPayload:
async def handle_typed_payload(
self, payload: ChargingScheduleCommandPayload
) -> CommandProcessingResult:
if (
payload.mode == ScheduledChargingMode.UNTIL_CONFIGURED_SOC
and not self.vehicle_state.vehicle.supports_target_soc
):
LOG.warning(
"Ignoring UNTIL_CONFIGURED_SOC charging schedule: "
"vehicle does not support target SoC"
)
return RESULT_DO_NOTHING
LOG.info("Setting charging schedule to %s", str(payload))
await self.saic_api.set_schedule_charging(
self.vin,
Expand Down
6 changes: 6 additions & 0 deletions src/handlers/command/drivetrain/drivetrain_soc_target.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
from saic_ismart_client_ng.api.vehicle_charging import TargetBatteryCode

from handlers.command.base import (
RESULT_DO_NOTHING,
RESULT_REFRESH_ONLY,
CommandProcessingResult,
PayloadConvertingCommandHandler,
Expand All @@ -31,6 +32,11 @@ def convert_payload(payload: str) -> TargetBatteryCode:
async def handle_typed_payload(
self, target_battery_code: TargetBatteryCode
) -> CommandProcessingResult:
if not self.vehicle_state.vehicle.supports_target_soc:
LOG.warning(
"Ignoring target SoC change: vehicle does not support target SoC"
)
return RESULT_DO_NOTHING
LOG.info("Setting SoC target to %s", str(target_battery_code))
await self.saic_api.set_target_battery_soc(
self.vin, target_soc=target_battery_code
Expand Down
35 changes: 21 additions & 14 deletions src/integrations/home_assistant/discovery.py
Original file line number Diff line number Diff line change
Expand Up @@ -98,19 +98,21 @@ def __publish_ha_discovery_messages_real(self) -> None:
self.__publish_doors_sensors()
self.__publish_drivetrain_charging_sensors()

# Target SoC
self._publish_number(
mqtt_topics.DRIVETRAIN_SOC_TARGET,
"Target SoC",
enabled=self.__vin_info.supports_target_soc,
device_class="battery",
unit_of_measurement="%",
icon="mdi:battery-charging-70",
mode="slider",
min_value=40,
max_value=100,
step=10,
)
# Target SoC — only offer the entity if the vehicle actually supports it
if self.__vin_info.supports_target_soc:
self._publish_number(
mqtt_topics.DRIVETRAIN_SOC_TARGET,
"Target SoC",
device_class="battery",
unit_of_measurement="%",
icon="mdi:battery-charging-70",
mode="slider",
min_value=40,
max_value=100,
step=10,
)
else:
self.__unpublish_ha_discovery_message("number", "Target SoC")
options = [
m.limit
for m in ChargeCurrentLimitCode
Expand Down Expand Up @@ -934,7 +936,12 @@ def __publish_scheduled_charging(self) -> None:
"mode": "{{ value }}",
}
)
options = [m.name for m in ScheduledChargingMode]
options = [
m.name
for m in ScheduledChargingMode
if m != ScheduledChargingMode.UNTIL_CONFIGURED_SOC
or self.__vin_info.supports_target_soc
]
self._publish_select(
mqtt_topics.DRIVETRAIN_CHARGING_SCHEDULE,
"Scheduled Charging Mode",
Expand Down
113 changes: 113 additions & 0 deletions tests/handlers/command/test_drivetrain_soc_guards.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,113 @@
from __future__ import annotations

import json
import unittest
from unittest.mock import AsyncMock, MagicMock

from saic_ismart_client_ng.api.vehicle.schema import VehicleModelConfiguration, VinInfo
from saic_ismart_client_ng.api.vehicle_charging import TargetBatteryCode

from handlers.command.base import RESULT_DO_NOTHING, RESULT_REFRESH_ONLY
from handlers.command.drivetrain.drivetrain_charging_schedule import (
DrivetrainChargingScheduleCommand,
)
from handlers.command.drivetrain.drivetrain_soc_target import DrivetrainSoCTargetCommand
from vehicle_info import VehicleInfo


def _make_vehicle_state(*, supports_target_soc: bool) -> MagicMock:
"""Create a minimal mock VehicleState with the desired supports_target_soc flag."""
configurations = []
if supports_target_soc:
configurations.append(
VehicleModelConfiguration(itemCode="BType", itemValue="1")
)
vin_info = VinInfo()
vin_info.vin = "vin_test_000000000"
vin_info.vehicleModelConfiguration = configurations
vehicle_info = VehicleInfo(vin_info, None)

vehicle_state = MagicMock()
vehicle_state.vehicle = vehicle_info
vehicle_state.vin = vehicle_info.vin
return vehicle_state


class TestDrivetrainSoCTargetGuard(unittest.IsolatedAsyncioTestCase):
async def test_reject_target_soc_when_unsupported(self) -> None:
vehicle_state = _make_vehicle_state(supports_target_soc=False)
saic_api = AsyncMock()
handler = DrivetrainSoCTargetCommand(saic_api, vehicle_state)

result = await handler.handle("80")

assert result == RESULT_DO_NOTHING
saic_api.set_target_battery_soc.assert_not_called()

async def test_allow_target_soc_when_supported(self) -> None:
vehicle_state = _make_vehicle_state(supports_target_soc=True)
saic_api = AsyncMock()
handler = DrivetrainSoCTargetCommand(saic_api, vehicle_state)

result = await handler.handle("80")

assert result == RESULT_REFRESH_ONLY
saic_api.set_target_battery_soc.assert_called_once_with(
vehicle_state.vin, target_soc=TargetBatteryCode.P_80
)


class TestDrivetrainChargingScheduleGuard(unittest.IsolatedAsyncioTestCase):
async def test_reject_until_configured_soc_when_unsupported(self) -> None:
vehicle_state = _make_vehicle_state(supports_target_soc=False)
saic_api = AsyncMock()
handler = DrivetrainChargingScheduleCommand(saic_api, vehicle_state)

payload = json.dumps(
{"startTime": "01:00", "endTime": "06:00", "mode": "UNTIL_CONFIGURED_SOC"}
)
result = await handler.handle(payload)

assert result == RESULT_DO_NOTHING
saic_api.set_schedule_charging.assert_not_called()

async def test_allow_until_configured_soc_when_supported(self) -> None:
vehicle_state = _make_vehicle_state(supports_target_soc=True)
saic_api = AsyncMock()
handler = DrivetrainChargingScheduleCommand(saic_api, vehicle_state)

payload = json.dumps(
{"startTime": "01:00", "endTime": "06:00", "mode": "UNTIL_CONFIGURED_SOC"}
)
result = await handler.handle(payload)

assert result == RESULT_REFRESH_ONLY
saic_api.set_schedule_charging.assert_called_once()

async def test_allow_until_configured_time_regardless(self) -> None:
"""UNTIL_CONFIGURED_TIME should work even without SoC support."""
vehicle_state = _make_vehicle_state(supports_target_soc=False)
saic_api = AsyncMock()
handler = DrivetrainChargingScheduleCommand(saic_api, vehicle_state)

payload = json.dumps(
{"startTime": "01:00", "endTime": "06:00", "mode": "UNTIL_CONFIGURED_TIME"}
)
result = await handler.handle(payload)

assert result == RESULT_REFRESH_ONLY
saic_api.set_schedule_charging.assert_called_once()

async def test_allow_disabled_regardless(self) -> None:
"""DISABLED should work even without SoC support."""
vehicle_state = _make_vehicle_state(supports_target_soc=False)
saic_api = AsyncMock()
handler = DrivetrainChargingScheduleCommand(saic_api, vehicle_state)

payload = json.dumps(
{"startTime": "01:00", "endTime": "06:00", "mode": "DISABLED"}
)
result = await handler.handle(payload)

assert result == RESULT_REFRESH_ONLY
saic_api.set_schedule_charging.assert_called_once()