Manage the Peblar autocharge vehicle list (#180606)

This commit is contained in:
Franck Nijhof
2026-08-29 14:42:11 +02:00
committed by GitHub
parent 790cada2d7
commit 0cae37c56b
8 changed files with 311 additions and 12 deletions
+1
View File
@@ -7,6 +7,7 @@ from peblar import ChargeLimiter, CPState
DOMAIN: Final = "peblar"
CONF_EVCC_ID: Final = "evcc_id"
CONF_UID: Final = "uid"
LOGGER = logging.getLogger(__package__)
@@ -85,11 +85,20 @@
"add_rfid_token": {
"service": "mdi:card-plus"
},
"add_vehicle_token": {
"service": "mdi:car-connected"
},
"delete_rfid_token": {
"service": "mdi:card-remove"
},
"delete_vehicle_token": {
"service": "mdi:car-off"
},
"list_rfid_tokens": {
"service": "mdi:card-account-details"
},
"list_vehicle_tokens": {
"service": "mdi:car-multiple"
}
}
}
+86 -10
View File
@@ -6,7 +6,7 @@ from contextlib import asynccontextmanager
from peblar import Peblar, PeblarAuthenticationError, PeblarConnectionError, PeblarError
import voluptuous as vol
from homeassistant.const import ATTR_CONFIG_ENTRY_ID, CONF_DESCRIPTION
from homeassistant.const import ATTR_CONFIG_ENTRY_ID, CONF_ALIAS, CONF_DESCRIPTION
from homeassistant.core import (
HomeAssistant,
ServiceCall,
@@ -20,24 +20,31 @@ from homeassistant.helpers.service import (
async_register_admin_service,
)
from .const import CONF_UID, DOMAIN
from .const import CONF_EVCC_ID, CONF_UID, DOMAIN
from .coordinator import PeblarConfigEntry
SERVICE_ADD_RFID_TOKEN = "add_rfid_token"
SERVICE_ADD_VEHICLE_TOKEN = "add_vehicle_token"
SERVICE_DELETE_RFID_TOKEN = "delete_rfid_token"
SERVICE_DELETE_VEHICLE_TOKEN = "delete_vehicle_token"
SERVICE_LIST_RFID_TOKENS = "list_rfid_tokens"
SERVICE_LIST_VEHICLE_TOKENS = "list_vehicle_tokens"
CHARGER_SCHEMA = vol.Schema({vol.Required(ATTR_CONFIG_ENTRY_ID): str})
TOKEN_SCHEMA = CHARGER_SCHEMA.extend({vol.Required(CONF_UID): str})
ADD_TOKEN_SCHEMA = TOKEN_SCHEMA.extend({vol.Required(CONF_DESCRIPTION): str})
VEHICLE_SCHEMA = CHARGER_SCHEMA.extend({vol.Required(CONF_EVCC_ID): str})
ADD_VEHICLE_SCHEMA = VEHICLE_SCHEMA.extend({vol.Required(CONF_ALIAS): str})
def _get_peblar(hass: HomeAssistant, entry_id: str) -> Peblar:
"""Return the Peblar client for the charger the call targets.
Every action here manages the standalone authorization list, which
lives on the RFID reader, so a charger without one is turned away
here instead of failing somewhere inside the charger.
def _get_rfid_peblar(hass: HomeAssistant, entry_id: str) -> Peblar:
"""Return the client, for a charger that has an RFID reader.
The standalone authorization list lives on the reader, so a charger
without one is turned away here instead of failing somewhere inside
the charger.
"""
entry: PeblarConfigEntry = async_get_config_entry(hass, DOMAIN, entry_id)
@@ -51,6 +58,26 @@ def _get_peblar(hass: HomeAssistant, entry_id: str) -> Peblar:
return entry.runtime_data.user_configuration_coordinator.peblar
def _get_autocharge_peblar(hass: HomeAssistant, entry_id: str) -> Peblar:
"""Return the client, for a charger that can do autocharge.
Autocharge identifies a car by what its own controller presents over
the power line, so it takes the PLC hardware. Without it the charger
still hands out the list, but refuses to change it, so this turns the
charger away rather than let that come back as a failed request.
"""
entry: PeblarConfigEntry = async_get_config_entry(hass, DOMAIN, entry_id)
if not entry.runtime_data.system_information.hardware_has_plc:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="no_autocharge_hardware",
translation_placeholders={"charger": entry.title},
)
return entry.runtime_data.user_configuration_coordinator.peblar
@asynccontextmanager
async def _handle_peblar_errors(
hass: HomeAssistant, entry_id: str
@@ -88,7 +115,7 @@ def async_setup_services(hass: HomeAssistant) -> None:
async def _handle_list_rfid_tokens(call: ServiceCall) -> ServiceResponse:
entry_id = call.data[ATTR_CONFIG_ENTRY_ID]
peblar = _get_peblar(hass, entry_id)
peblar = _get_rfid_peblar(hass, entry_id)
async with _handle_peblar_errors(hass, entry_id):
tokens = await peblar.rfid_tokens()
return {
@@ -103,7 +130,7 @@ def async_setup_services(hass: HomeAssistant) -> None:
async def _handle_add_rfid_token(call: ServiceCall) -> None:
entry_id = call.data[ATTR_CONFIG_ENTRY_ID]
peblar = _get_peblar(hass, entry_id)
peblar = _get_rfid_peblar(hass, entry_id)
async with _handle_peblar_errors(hass, entry_id):
await peblar.add_rfid_token(
rfid_token_uid=call.data[CONF_UID],
@@ -112,10 +139,37 @@ def async_setup_services(hass: HomeAssistant) -> None:
async def _handle_delete_rfid_token(call: ServiceCall) -> None:
entry_id = call.data[ATTR_CONFIG_ENTRY_ID]
peblar = _get_peblar(hass, entry_id)
peblar = _get_rfid_peblar(hass, entry_id)
async with _handle_peblar_errors(hass, entry_id):
await peblar.delete_rfid_token(uid=call.data[CONF_UID])
async def _handle_list_vehicle_tokens(call: ServiceCall) -> ServiceResponse:
entry_id = call.data[ATTR_CONFIG_ENTRY_ID]
peblar = _get_autocharge_peblar(hass, entry_id)
async with _handle_peblar_errors(hass, entry_id):
vehicles = await peblar.vehicle_tokens()
return {
"vehicles": [
{CONF_EVCC_ID: vehicle.evcc_id, CONF_ALIAS: vehicle.alias}
for vehicle in vehicles
]
}
async def _handle_add_vehicle_token(call: ServiceCall) -> None:
entry_id = call.data[ATTR_CONFIG_ENTRY_ID]
peblar = _get_autocharge_peblar(hass, entry_id)
async with _handle_peblar_errors(hass, entry_id):
await peblar.add_vehicle_token(
evcc_id=call.data[CONF_EVCC_ID],
alias=call.data[CONF_ALIAS],
)
async def _handle_delete_vehicle_token(call: ServiceCall) -> None:
entry_id = call.data[ATTR_CONFIG_ENTRY_ID]
peblar = _get_autocharge_peblar(hass, entry_id)
async with _handle_peblar_errors(hass, entry_id):
await peblar.delete_vehicle_token(evcc_id=call.data[CONF_EVCC_ID])
async_register_admin_service(
hass,
DOMAIN,
@@ -138,3 +192,25 @@ def async_setup_services(hass: HomeAssistant) -> None:
_handle_delete_rfid_token,
schema=TOKEN_SCHEMA,
)
async_register_admin_service(
hass,
DOMAIN,
SERVICE_LIST_VEHICLE_TOKENS,
_handle_list_vehicle_tokens,
schema=CHARGER_SCHEMA,
supports_response=SupportsResponse.ONLY,
)
async_register_admin_service(
hass,
DOMAIN,
SERVICE_ADD_VEHICLE_TOKEN,
_handle_add_vehicle_token,
schema=ADD_VEHICLE_SCHEMA,
)
async_register_admin_service(
hass,
DOMAIN,
SERVICE_DELETE_VEHICLE_TOKEN,
_handle_delete_vehicle_token,
schema=VEHICLE_SCHEMA,
)
@@ -33,3 +33,39 @@ delete_rfid_token:
required: true
selector:
text:
list_vehicle_tokens:
fields:
config_entry_id:
required: true
selector:
config_entry:
integration: peblar
add_vehicle_token:
fields:
config_entry_id:
required: true
selector:
config_entry:
integration: peblar
evcc_id:
required: true
selector:
text:
alias:
required: true
selector:
text:
delete_vehicle_token:
fields:
config_entry_id:
required: true
selector:
config_entry:
integration: peblar
evcc_id:
required: true
selector:
text:
@@ -206,6 +206,9 @@
"communication_error": {
"message": "An error occurred while communicating with the Peblar EV charger: {error}"
},
"no_autocharge_hardware": {
"message": "{charger} has no power line communication hardware, so it cannot use autocharge."
},
"no_rfid_hardware": {
"message": "{charger} has no RFID reader, so it has no standalone authorization list."
},
@@ -232,6 +235,24 @@
},
"name": "Add RFID token"
},
"add_vehicle_token": {
"description": "Adds a vehicle to the charger's autocharge list, so it is authorized to charge as soon as it is plugged in.",
"fields": {
"alias": {
"description": "A human-readable label for this vehicle.",
"name": "Alias"
},
"config_entry_id": {
"description": "The Peblar EV charger to add the vehicle to.",
"name": "Peblar EV charger"
},
"evcc_id": {
"description": "The identifier the vehicle presents to the charger.",
"name": "EVCC ID"
}
},
"name": "Add autocharge vehicle"
},
"delete_rfid_token": {
"description": "Deletes an RFID token from the charger's standalone authorization list.",
"fields": {
@@ -246,6 +267,20 @@
},
"name": "Delete RFID token"
},
"delete_vehicle_token": {
"description": "Deletes a vehicle from the charger's autocharge list.",
"fields": {
"config_entry_id": {
"description": "The Peblar EV charger to delete the vehicle from.",
"name": "Peblar EV charger"
},
"evcc_id": {
"description": "The identifier of the vehicle to delete.",
"name": "EVCC ID"
}
},
"name": "Delete autocharge vehicle"
},
"list_rfid_tokens": {
"description": "Returns the RFID tokens configured in the charger's standalone authorization list.",
"fields": {
@@ -255,6 +290,16 @@
}
},
"name": "List RFID tokens"
},
"list_vehicle_tokens": {
"description": "Returns the vehicles configured in the charger's autocharge list.",
"fields": {
"config_entry_id": {
"description": "The Peblar EV charger to list autocharge vehicles for.",
"name": "Peblar EV charger"
}
},
"name": "List autocharge vehicles"
}
}
}
@@ -22,7 +22,7 @@
"HwHasLte": false,
"HwHasMeter": true,
"HwHasMeterDisplay": true,
"HwHasPlc": false,
"HwHasPlc": true,
"HwHasRfid": true,
"HwHasRs485": true,
"HwHasShutter": false,
@@ -62,7 +62,7 @@
'HwHasLte': False,
'HwHasMeter': True,
'HwHasMeterDisplay': True,
'HwHasPlc': False,
'HwHasPlc': True,
'HwHasRfid': True,
'HwHasRs485': True,
'HwHasShutter': False,
+132
View File
@@ -8,14 +8,18 @@ from peblar import (
PeblarConnectionError,
PeblarError,
PeblarRfidToken,
PeblarVehicleToken,
)
import pytest
from homeassistant.components.peblar.const import DOMAIN
from homeassistant.components.peblar.services import (
SERVICE_ADD_RFID_TOKEN,
SERVICE_ADD_VEHICLE_TOKEN,
SERVICE_DELETE_RFID_TOKEN,
SERVICE_DELETE_VEHICLE_TOKEN,
SERVICE_LIST_RFID_TOKENS,
SERVICE_LIST_VEHICLE_TOKENS,
)
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState
from homeassistant.core import HomeAssistant
@@ -277,3 +281,131 @@ async def test_charger_without_rfid_reader(
assert excinfo.value.translation_domain == DOMAIN
assert excinfo.value.translation_key == "no_rfid_hardware"
async def test_list_vehicle_tokens(
hass: HomeAssistant,
mock_peblar: MagicMock,
init_integration: MockConfigEntry,
) -> None:
"""Test list_vehicle_tokens returns the vehicles on the charger."""
mock_peblar.vehicle_tokens.return_value = [
PeblarVehicleToken(evcc_id="EVCC-1234", alias="The blue one"),
PeblarVehicleToken(evcc_id="EVCC-5678", alias="The other one"),
]
result = await hass.services.async_call(
DOMAIN,
SERVICE_LIST_VEHICLE_TOKENS,
{"config_entry_id": init_integration.entry_id},
blocking=True,
return_response=True,
)
assert result == {
"vehicles": [
{"evcc_id": "EVCC-1234", "alias": "The blue one"},
{"evcc_id": "EVCC-5678", "alias": "The other one"},
]
}
mock_peblar.vehicle_tokens.assert_called_once_with()
async def test_add_vehicle_token(
hass: HomeAssistant,
mock_peblar: MagicMock,
init_integration: MockConfigEntry,
) -> None:
"""Test add_vehicle_token calls the library with the right arguments."""
await hass.services.async_call(
DOMAIN,
SERVICE_ADD_VEHICLE_TOKEN,
{
"config_entry_id": init_integration.entry_id,
"evcc_id": "EVCC-1234",
"alias": "The blue one",
},
blocking=True,
)
mock_peblar.add_vehicle_token.assert_called_once_with(
evcc_id="EVCC-1234",
alias="The blue one",
)
async def test_delete_vehicle_token(
hass: HomeAssistant,
mock_peblar: MagicMock,
init_integration: MockConfigEntry,
) -> None:
"""Test delete_vehicle_token calls the library with the right arguments."""
await hass.services.async_call(
DOMAIN,
SERVICE_DELETE_VEHICLE_TOKEN,
{
"config_entry_id": init_integration.entry_id,
"evcc_id": "EVCC-1234",
},
blocking=True,
)
mock_peblar.delete_vehicle_token.assert_called_once_with(evcc_id="EVCC-1234")
AUTOCHARGE_CALLS: list[tuple[str, dict[str, Any]]] = [
(SERVICE_LIST_VEHICLE_TOKENS, {}),
(SERVICE_ADD_VEHICLE_TOKEN, {"evcc_id": "EVCC-1234", "alias": "The blue one"}),
(SERVICE_DELETE_VEHICLE_TOKEN, {"evcc_id": "EVCC-1234"}),
]
@pytest.mark.parametrize("mock_peblar", [{"HwHasRfid": False}], indirect=True)
@pytest.mark.parametrize(("service", "service_data"), AUTOCHARGE_CALLS)
@pytest.mark.usefixtures("mock_peblar")
async def test_autocharge_does_not_need_an_rfid_reader(
hass: HomeAssistant,
init_integration: MockConfigEntry,
service: str,
service_data: dict[str, Any],
) -> None:
"""Test the autocharge list is a separate one from the RFID list.
Autocharge identifies a car by what its own controller presents, so it
has nothing to do with the reader.
"""
await hass.services.async_call(
DOMAIN,
service,
{"config_entry_id": init_integration.entry_id, **service_data},
blocking=True,
return_response=service == SERVICE_LIST_VEHICLE_TOKENS,
)
@pytest.mark.parametrize("mock_peblar", [{"HwHasPlc": False}], indirect=True)
@pytest.mark.parametrize(("service", "service_data"), AUTOCHARGE_CALLS)
@pytest.mark.usefixtures("mock_peblar")
async def test_autocharge_needs_power_line_communication(
hass: HomeAssistant,
init_integration: MockConfigEntry,
service: str,
service_data: dict[str, Any],
) -> None:
"""Test a charger that cannot do autocharge is turned away.
Checked against a charger without the hardware: it answers 200 with
null on the list, and 403 on adding and deleting. Letting that come
back as a failed request tells the user nothing.
"""
with pytest.raises(ServiceValidationError) as excinfo:
await hass.services.async_call(
DOMAIN,
service,
{"config_entry_id": init_integration.entry_id, **service_data},
blocking=True,
return_response=service == SERVICE_LIST_VEHICLE_TOKENS,
)
assert excinfo.value.translation_domain == DOMAIN
assert excinfo.value.translation_key == "no_autocharge_hardware"