jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-14 08:47:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Decouple service registration in Renault (#143210)

Browse Source
This commit is contained in:
epenet 2025-04-19 12:36:54 +02:00 committed by GitHub
parent 3da77726d0
commit ff1ab1da37
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 73 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

142
homeassistant/components/renault/services.py
View File

@ -2,7 +2,6 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Mapping
from datetime import datetime from datetime import datetime
import logging import logging
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any
@ -105,91 +104,96 @@ SERVICES = [
] ]
def setup_services(hass: HomeAssistant) -> None: async def ac_cancel(service_call: ServiceCall) -> None:
"""Register the Renault services.""" """Cancel A/C."""
proxy = get_vehicle_proxy(service_call)
async def ac_cancel(service_call: ServiceCall) -> None: LOGGER.debug("A/C cancel attempt")
"""Cancel A/C.""" result = await proxy.set_ac_stop()
proxy = get_vehicle_proxy(service_call.data) LOGGER.debug("A/C cancel result: %s", result)
LOGGER.debug("A/C cancel attempt")
result = await proxy.set_ac_stop()
LOGGER.debug("A/C cancel result: %s", result)
async def ac_start(service_call: ServiceCall) -> None: async def ac_start(service_call: ServiceCall) -> None:
"""Start A/C.""" """Start A/C."""
temperature: float = service_call.data[ATTR_TEMPERATURE] temperature: float = service_call.data[ATTR_TEMPERATURE]
when: datetime | None = service_call.data.get(ATTR_WHEN) when: datetime | None = service_call.data.get(ATTR_WHEN)
proxy = get_vehicle_proxy(service_call.data) proxy = get_vehicle_proxy(service_call)
LOGGER.debug("A/C start attempt: %s / %s", temperature, when) LOGGER.debug("A/C start attempt: %s / %s", temperature, when)
result = await proxy.set_ac_start(temperature, when) result = await proxy.set_ac_start(temperature, when)
LOGGER.debug("A/C start result: %s", result.raw_data) LOGGER.debug("A/C start result: %s", result.raw_data)
async def charge_set_schedules(service_call: ServiceCall) -> None:
"""Set charge schedules."""
schedules: list[dict[str, Any]] = service_call.data[ATTR_SCHEDULES]
proxy = get_vehicle_proxy(service_call.data)
charge_schedules = await proxy.get_charging_settings()
for schedule in schedules:
charge_schedules.update(schedule)
if TYPE_CHECKING: async def charge_set_schedules(service_call: ServiceCall) -> None:
assert charge_schedules.schedules is not None """Set charge schedules."""
LOGGER.debug("Charge set schedules attempt: %s", schedules) schedules: list[dict[str, Any]] = service_call.data[ATTR_SCHEDULES]
result = await proxy.set_charge_schedules(charge_schedules.schedules) proxy = get_vehicle_proxy(service_call)
charge_schedules = await proxy.get_charging_settings()
for schedule in schedules:
charge_schedules.update(schedule)
LOGGER.debug("Charge set schedules result: %s", result) if TYPE_CHECKING:
LOGGER.debug( assert charge_schedules.schedules is not None
"It may take some time before these changes are reflected in your vehicle" LOGGER.debug("Charge set schedules attempt: %s", schedules)
) result = await proxy.set_charge_schedules(charge_schedules.schedules)
async def ac_set_schedules(service_call: ServiceCall) -> None: LOGGER.debug("Charge set schedules result: %s", result)
"""Set A/C schedules.""" LOGGER.debug(
schedules: list[dict[str, Any]] = service_call.data[ATTR_SCHEDULES] "It may take some time before these changes are reflected in your vehicle"
proxy = get_vehicle_proxy(service_call.data) )
hvac_schedules = await proxy.get_hvac_settings()
for schedule in schedules:
hvac_schedules.update(schedule)
if TYPE_CHECKING: async def ac_set_schedules(service_call: ServiceCall) -> None:
assert hvac_schedules.schedules is not None """Set A/C schedules."""
LOGGER.debug("HVAC set schedules attempt: %s", schedules) schedules: list[dict[str, Any]] = service_call.data[ATTR_SCHEDULES]
result = await proxy.set_hvac_schedules(hvac_schedules.schedules) proxy = get_vehicle_proxy(service_call)
hvac_schedules = await proxy.get_hvac_settings()
LOGGER.debug("HVAC set schedules result: %s", result) for schedule in schedules:
LOGGER.debug( hvac_schedules.update(schedule)
"It may take some time before these changes are reflected in your vehicle"
)
def get_vehicle_proxy(service_call_data: Mapping) -> RenaultVehicleProxy: if TYPE_CHECKING:
"""Get vehicle from service_call data.""" assert hvac_schedules.schedules is not None
device_registry = dr.async_get(hass) LOGGER.debug("HVAC set schedules attempt: %s", schedules)
device_id = service_call_data[ATTR_VEHICLE] result = await proxy.set_hvac_schedules(hvac_schedules.schedules)
device_entry = device_registry.async_get(device_id)
if device_entry is None:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="invalid_device_id",
translation_placeholders={"device_id": device_id},
)
loaded_entries: list[RenaultConfigEntry] = [ LOGGER.debug("HVAC set schedules result: %s", result)
entry LOGGER.debug(
for entry in hass.config_entries.async_loaded_entries(DOMAIN) "It may take some time before these changes are reflected in your vehicle"
if entry.entry_id in device_entry.config_entries )
]
for entry in loaded_entries:
for vin, vehicle in entry.runtime_data.vehicles.items(): def get_vehicle_proxy(service_call: ServiceCall) -> RenaultVehicleProxy:
if (DOMAIN, vin) in device_entry.identifiers: """Get vehicle from service_call data."""
return vehicle device_registry = dr.async_get(service_call.hass)
device_id = service_call.data[ATTR_VEHICLE]
device_entry = device_registry.async_get(device_id)
if device_entry is None:
raise ServiceValidationError( raise ServiceValidationError(
translation_domain=DOMAIN, translation_domain=DOMAIN,
translation_key="no_config_entry_for_device", translation_key="invalid_device_id",
translation_placeholders={"device_id": device_entry.name or device_id}, translation_placeholders={"device_id": device_id},
) )
loaded_entries: list[RenaultConfigEntry] = [
entry
for entry in service_call.hass.config_entries.async_loaded_entries(DOMAIN)
if entry.entry_id in device_entry.config_entries
]
for entry in loaded_entries:
for vin, vehicle in entry.runtime_data.vehicles.items():
if (DOMAIN, vin) in device_entry.identifiers:
return vehicle
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="no_config_entry_for_device",
translation_placeholders={"device_id": device_entry.name or device_id},
)
def setup_services(hass: HomeAssistant) -> None:
"""Register the Renault services."""
hass.services.async_register( hass.services.async_register(
DOMAIN, DOMAIN,
SERVICE_AC_CANCEL, SERVICE_AC_CANCEL,