starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-25 17:04:04 -04:00
Code Issues Packages Projects Releases Wiki Activity

Move service registration to async_setup in modbus (#181845)

Browse Source
This commit is contained in:
epenet
2026-09-10 20:48:28 +02:00
committed by GitHub
parent 2a2babceab
commit ea1a91ac78
4 changed files with 300 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/modbus/modbus.py
+15 -86
View File
@@ -12,10 +12,8 @@ from pymodbus.client import (
from pymodbus.exceptions import ModbusException
from pymodbus.framer import FramerType
from pymodbus.pdu import ModbusPDU
import voluptuous as vol
from homeassistant.const import (
ATTR_STATE,
CONF_DELAY,
CONF_HOST,
CONF_METHOD,
@@ -25,18 +23,11 @@ from homeassistant.const import (
CONF_TYPE,
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.core import Event, HomeAssistant, ServiceCall
from homeassistant.helpers import config_validation as cv
from homeassistant.core import Event, HomeAssistant
from homeassistant.helpers.discovery import async_load_platform
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.typing import ConfigType
from .const import (
ATTR_ADDRESS,
ATTR_HUB,
ATTR_SLAVE,
ATTR_UNIT,
ATTR_VALUE,
CALL_TYPE_COIL,
CALL_TYPE_DISCRETE,
CALL_TYPE_REGISTER_HOLDING,
@@ -51,17 +42,12 @@ from .const import (
CONF_PARITY,
CONF_STOPBITS,
DATA_MODBUS_HUBS,
DEFAULT_HUB,
DEVICE_ID,
DOMAIN,
LOGGER,
PLATFORMS,
RTUOVERTCP,
SERIAL,
SERVICE_STOP,
SERVICE_WRITE_COIL,
SERVICE_WRITE_REGISTER,
SIGNAL_STOP_ENTITY,
TCP,
UDP,
)
@@ -128,6 +114,20 @@ async def async_modbus_setup(
config: ConfigType,
) -> bool:
"""Set up Modbus component."""
if await _async_modbus_setup(hass, config):
return True
# Hubs are stored as they are created, so a failure part way through leaves
# unusable ones behind. Drop them, so their presence means they are usable.
hass.data.pop(DATA_MODBUS_HUBS, None)
return False
async def _async_modbus_setup(
hass: HomeAssistant,
config: ConfigType,
) -> bool:
"""Set up the Modbus hubs and their platforms."""
if config[DOMAIN]:
config[DOMAIN] = check_config(hass, config[DOMAIN])
@@ -165,77 +165,6 @@ async def async_modbus_setup(
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, async_stop_modbus)
def _get_service_call_details(
service: ServiceCall,
) -> tuple[ModbusHub, int, int]:
"""Return the details required to process the service call."""
device_address = service.data.get(ATTR_SLAVE, service.data.get(ATTR_UNIT, 1))
address = service.data[ATTR_ADDRESS]
hub = hub_collect[service.data[ATTR_HUB]]
return (hub, device_address, address)
async def async_write_register(service: ServiceCall) -> None:
"""Write Modbus registers."""
hub, device_address, address = _get_service_call_details(service)
value = service.data[ATTR_VALUE]
if isinstance(value, list):
await hub.async_pb_call(
device_address, address, value, CALL_TYPE_WRITE_REGISTERS
)
else:
await hub.async_pb_call(
device_address, address, value, CALL_TYPE_WRITE_REGISTER
)
async def async_write_coil(service: ServiceCall) -> None:
"""Write Modbus coil."""
hub, device_address, address = _get_service_call_details(service)
state = service.data[ATTR_STATE]
if isinstance(state, list):
await hub.async_pb_call(
device_address, address, state, CALL_TYPE_WRITE_COILS
)
else:
await hub.async_pb_call(
device_address, address, state, CALL_TYPE_WRITE_COIL
)
for x_write in (
(SERVICE_WRITE_REGISTER, async_write_register, ATTR_VALUE, cv.positive_int),
(SERVICE_WRITE_COIL, async_write_coil, ATTR_STATE, cv.boolean),
):
hass.services.async_register(
DOMAIN,
x_write[0],
x_write[1],
schema=vol.Schema(
{
vol.Optional(ATTR_HUB, default=DEFAULT_HUB): cv.string,
vol.Exclusive(ATTR_SLAVE, "unit"): cv.positive_int,
vol.Exclusive(ATTR_UNIT, "unit"): cv.positive_int,
vol.Required(ATTR_ADDRESS): cv.positive_int,
vol.Required(x_write[2]): vol.Any(
cv.positive_int, vol.All(cv.ensure_list, [x_write[3]])
),
}
),
)
async def async_stop_hub(service: ServiceCall) -> None:
"""Stop Modbus hub."""
async_dispatcher_send(hass, SIGNAL_STOP_ENTITY)
hub = hub_collect[service.data[ATTR_HUB]]
await hub.async_close()
hass.services.async_register(
DOMAIN,
SERVICE_STOP,
async_stop_hub,
schema=vol.Schema({vol.Required(ATTR_HUB): cv.string}),
)
return True
homeassistant/components/modbus/services.py
+116 -3
View File
@@ -1,13 +1,108 @@
"""Support for Modbus services."""
from homeassistant.const import SERVICE_RELOAD
from collections.abc import Callable
from typing import Any
import voluptuous as vol
from homeassistant.const import ATTR_STATE, SERVICE_RELOAD
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.entity_platform import async_get_platforms
from homeassistant.helpers.reload import async_integration_yaml_config
from homeassistant.helpers.service import async_register_admin_service
from .const import DATA_MODBUS_HUBS, DOMAIN, LOGGER
from .modbus import async_modbus_setup
from .const import (
ATTR_ADDRESS,
ATTR_HUB,
ATTR_SLAVE,
ATTR_UNIT,
ATTR_VALUE,
CALL_TYPE_WRITE_COIL,
CALL_TYPE_WRITE_COILS,
CALL_TYPE_WRITE_REGISTER,
CALL_TYPE_WRITE_REGISTERS,
DATA_MODBUS_HUBS,
DEFAULT_HUB,
DOMAIN,
LOGGER,
SERVICE_STOP,
SERVICE_WRITE_COIL,
SERVICE_WRITE_REGISTER,
SIGNAL_STOP_ENTITY,
)
from .modbus import ModbusHub, async_modbus_setup
def _write_service_schema(attr: str, validator: Callable[[Any], Any]) -> vol.Schema:
"""Return the schema shared by the write actions."""
return vol.Schema(
{
vol.Optional(ATTR_HUB, default=DEFAULT_HUB): cv.string,
vol.Exclusive(ATTR_SLAVE, "unit"): cv.positive_int,
vol.Exclusive(ATTR_UNIT, "unit"): cv.positive_int,
vol.Required(ATTR_ADDRESS): cv.positive_int,
vol.Required(attr): vol.Any(
cv.positive_int, vol.All(cv.ensure_list, [validator])
),
}
)
def _get_hubs(hass: HomeAssistant) -> dict[str, ModbusHub]:
"""Return the configured Modbus hubs, raising if Modbus is not set up."""
if not (hubs := hass.data.get(DATA_MODBUS_HUBS)):
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="not_loaded",
)
return hubs
def _get_service_call_details(service: ServiceCall) -> tuple[ModbusHub, int, int]:
"""Return the details required to process the service call."""
device_address = service.data.get(ATTR_SLAVE, service.data.get(ATTR_UNIT, 1))
address = service.data[ATTR_ADDRESS]
hub = _get_hubs(service.hass)[service.data[ATTR_HUB]]
return (hub, device_address, address)
async def _async_write_register(service: ServiceCall) -> None:
"""Write Modbus registers."""
hub, device_address, address = _get_service_call_details(service)
value = service.data[ATTR_VALUE]
if isinstance(value, list):
await hub.async_pb_call(
device_address, address, value, CALL_TYPE_WRITE_REGISTERS
)
else:
await hub.async_pb_call(
device_address, address, value, CALL_TYPE_WRITE_REGISTER
)
async def _async_write_coil(service: ServiceCall) -> None:
"""Write Modbus coil."""
hub, device_address, address = _get_service_call_details(service)
state = service.data[ATTR_STATE]
if isinstance(state, list):
await hub.async_pb_call(device_address, address, state, CALL_TYPE_WRITE_COILS)
else:
await hub.async_pb_call(device_address, address, state, CALL_TYPE_WRITE_COIL)
async def _async_stop_hub(service: ServiceCall) -> None:
"""Stop Modbus hub."""
hass = service.hass
hub = _get_hubs(hass)[service.data[ATTR_HUB]]
async_dispatcher_send(hass, SIGNAL_STOP_ENTITY)
await hub.async_close()
async def _async_reload_config(call: ServiceCall) -> None:
@@ -35,3 +130,21 @@ async def _async_reload_config(call: ServiceCall) -> None:
def async_setup_services(hass: HomeAssistant) -> None:
"""Register the Modbus services."""
async_register_admin_service(hass, DOMAIN, SERVICE_RELOAD, _async_reload_config)
hass.services.async_register(
DOMAIN,
SERVICE_WRITE_REGISTER,
_async_write_register,
schema=_write_service_schema(ATTR_VALUE, cv.positive_int),
)
hass.services.async_register(
DOMAIN,
SERVICE_WRITE_COIL,
_async_write_coil,
schema=_write_service_schema(ATTR_STATE, cv.boolean),
)
hass.services.async_register(
DOMAIN,
SERVICE_STOP,
_async_stop_hub,
schema=vol.Schema({vol.Required(ATTR_HUB): cv.string}),
)
homeassistant/components/modbus/strings.json
+5
View File
@@ -1,4 +1,9 @@
{
"exceptions": {
"not_loaded": {
"message": "Modbus is not loaded, so this action cannot be performed."
}
},
"issues": {
"duplicate_entity_entry": {
"description": "An address can only be associated with one entity. Please correct the entry in your configuration.yaml file and restart Home Assistant to fix this issue.",
tests/components/modbus/test_services.py
+164
View File
@@ -0,0 +1,164 @@
"""Tests for the Modbus services."""
from unittest.mock import patch
import pytest
from homeassistant import config as hass_config
from homeassistant.components.modbus.const import (
ATTR_ADDRESS,
ATTR_HUB,
ATTR_VALUE,
DATA_MODBUS_HUBS,
DEFAULT_HUB,
DOMAIN,
SERVICE_STOP,
SERVICE_WRITE_COIL,
SERVICE_WRITE_REGISTER,
)
from homeassistant.const import (
ATTR_STATE,
CONF_ADDRESS,
CONF_HOST,
CONF_NAME,
CONF_PORT,
CONF_SENSORS,
CONF_TYPE,
SERVICE_RELOAD,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from homeassistant.setup import async_setup_component
from tests.common import get_fixture_path
HUB_CONFIG = {
DOMAIN: [
{
CONF_NAME: DEFAULT_HUB,
CONF_TYPE: "tcp",
CONF_HOST: "modbusHost",
CONF_PORT: 5501,
CONF_SENSORS: [{CONF_NAME: "dummy", CONF_ADDRESS: 9999}],
}
]
}
SERVICES = (
SERVICE_RELOAD,
SERVICE_WRITE_REGISTER,
SERVICE_WRITE_COIL,
SERVICE_STOP,
)
# The actions that need a configured hub, and a minimal valid payload for each.
HUB_SERVICES = [
pytest.param(
SERVICE_WRITE_REGISTER,
{ATTR_HUB: DEFAULT_HUB, ATTR_ADDRESS: 1, ATTR_VALUE: 1},
id="write_register",
),
pytest.param(
SERVICE_WRITE_COIL,
{ATTR_HUB: DEFAULT_HUB, ATTR_ADDRESS: 1, ATTR_STATE: True},
id="write_coil",
),
pytest.param(SERVICE_STOP, {ATTR_HUB: DEFAULT_HUB}, id="stop"),
]
@pytest.mark.parametrize("service", SERVICES)
async def test_services_registered_without_yaml(
hass: HomeAssistant, service: str
) -> None:
"""Test the actions are registered without a Modbus YAML section."""
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
assert hass.services.has_service(DOMAIN, service)
@pytest.mark.parametrize(("service", "data"), HUB_SERVICES)
async def test_service_without_yaml_raises(
hass: HomeAssistant, service: str, data: dict
) -> None:
"""Test the hub actions raise without a Modbus YAML section."""
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
assert DATA_MODBUS_HUBS not in hass.data
with pytest.raises(ServiceValidationError) as err:
await hass.services.async_call(DOMAIN, service, data, blocking=True)
assert err.value.translation_domain == DOMAIN
assert err.value.translation_key == "not_loaded"
assert "Modbus is not loaded" in str(err.value)
@pytest.mark.parametrize(("service", "data"), HUB_SERVICES)
async def test_service_without_hubs_raises(
hass: HomeAssistant, service: str, data: dict
) -> None:
"""Test the hub actions raise when Modbus is configured without any hub."""
assert await async_setup_component(hass, DOMAIN, {DOMAIN: []})
await hass.async_block_till_done()
assert hass.data[DATA_MODBUS_HUBS] == {}
with pytest.raises(ServiceValidationError) as err:
await hass.services.async_call(DOMAIN, service, data, blocking=True)
assert err.value.translation_key == "not_loaded"
@pytest.mark.parametrize(("service", "data"), HUB_SERVICES)
async def test_service_after_failed_setup_raises(
hass: HomeAssistant, service: str, data: dict
) -> None:
"""Test the hub actions raise after a failed hub setup.
Hubs are stored before their setup is awaited, so a failure must drop them
again; stop would otherwise raise AttributeError on the unset _connect_task.
"""
with patch(
"homeassistant.components.modbus.modbus.ModbusHub.async_setup",
return_value=False,
):
assert not await async_setup_component(hass, DOMAIN, HUB_CONFIG)
await hass.async_block_till_done()
assert DATA_MODBUS_HUBS not in hass.data
with pytest.raises(ServiceValidationError) as err:
await hass.services.async_call(DOMAIN, service, data, blocking=True)
assert err.value.translation_key == "not_loaded"
@pytest.mark.parametrize(("service", "data"), HUB_SERVICES)
async def test_service_after_failed_reload_raises(
hass: HomeAssistant, service: str, data: dict
) -> None:
"""Test the hub actions raise after a reload failed to set the hubs up."""
assert await async_setup_component(hass, DOMAIN, HUB_CONFIG)
await hass.async_block_till_done()
assert hass.data[DATA_MODBUS_HUBS]
yaml_path = get_fixture_path("configuration.yaml", DOMAIN)
with (
patch.object(hass_config, "YAML_CONFIG_FILE", yaml_path),
patch(
"homeassistant.components.modbus.modbus.ModbusHub.async_setup",
return_value=False,
),
):
await hass.services.async_call(DOMAIN, SERVICE_RELOAD, {}, blocking=True)
await hass.async_block_till_done()
assert DATA_MODBUS_HUBS not in hass.data
with pytest.raises(ServiceValidationError) as err:
await hass.services.async_call(DOMAIN, service, data, blocking=True)
assert err.value.translation_key == "not_loaded"