Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Move modbus check_config to setup, to have access to hass (#112828)

pull/113226/head^2
jan iversen 2024-03-14 12:54:01 +01:00 committed by GitHub
parent 77a67191de
commit 7696973932
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 109 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
homeassistant/components/modbus/__init__.py
View File

@ -132,7 +132,6 @@ from .const import ( # noqa: F401
)
from .modbus import ModbusHub, async_modbus_setup
from .validators import (
check_config,
check_hvac_target_temp_registers,
duplicate_fan_mode_validator,
hvac_fixedsize_reglist_validator,
@ -418,7 +417,6 @@ CONFIG_SCHEMA = vol.Schema(
[
vol.Any(SERIAL_SCHEMA, ETHERNET_SCHEMA),
],
check_config,
),
},
extra=vol.ALLOW_EXTRA,

3
homeassistant/components/modbus/modbus.py
View File

@ -69,6 +69,7 @@ from .const import (
TCP,
UDP,
)
from .validators import check_config
_LOGGER = logging.getLogger(__name__)
@ -127,6 +128,8 @@ async def async_modbus_setup(
await async_setup_reload_service(hass, DOMAIN, [DOMAIN])
if config[DOMAIN]:
config[DOMAIN] = check_config(hass, config[DOMAIN])
if DOMAIN in hass.data and config[DOMAIN] == []:
hubs = hass.data[DOMAIN]
for name in hubs:

185
homeassistant/components/modbus/validators.py
View File

@ -24,6 +24,7 @@ from homeassistant.const import (
CONF_TIMEOUT,
CONF_TYPE,
)
from homeassistant.core import HomeAssistant
from .const import (
CONF_DATA_TYPE,
@ -270,7 +271,99 @@ def register_int_list_validator(value: Any) -> Any:
)
def check_config(config: dict) -> dict:
def validate_modbus(
hass: HomeAssistant,
hosts: set[str],
hub_names: set[str],
hub: dict,
hub_name_inx: int,
) -> bool:
"""Validate modbus entries."""
host: str = (
hub[CONF_PORT]
if hub[CONF_TYPE] == SERIAL
else f"{hub[CONF_HOST]}_{hub[CONF_PORT]}"
)
if CONF_NAME not in hub:
hub[CONF_NAME] = (
DEFAULT_HUB if not hub_name_inx else f"{DEFAULT_HUB}_{hub_name_inx}"
)
hub_name_inx += 1
err = f"Modbus host/port {host} is missing name, added {hub[CONF_NAME]}!"
_LOGGER.warning(err)
name = hub[CONF_NAME]
if host in hosts or name in hub_names:
err = f"Modbus {name} host/port {host} is duplicate, not loaded!"
_LOGGER.warning(err)
return False
hosts.add(host)
hub_names.add(name)
return True
def validate_entity(
hass: HomeAssistant,
hub_name: str,
component: str,
entity: dict,
minimum_scan_interval: int,
ent_names: set[str],
ent_addr: set[str],
) -> bool:
"""Validate entity."""
name = f"{component}.{entity[CONF_NAME]}"
addr = f"{hub_name}{entity[CONF_ADDRESS]}"
scan_interval = entity.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
if 0 < scan_interval < 5:
_LOGGER.warning(
(
"%s %s scan_interval(%d) is lower than 5 seconds, "
"which may cause Home Assistant stability issues"
),
hub_name,
name,
scan_interval,
)
entity[CONF_SCAN_INTERVAL] = scan_interval
minimum_scan_interval = min(scan_interval, minimum_scan_interval)
for conf_type in (
CONF_INPUT_TYPE,
CONF_WRITE_TYPE,
CONF_COMMAND_ON,
CONF_COMMAND_OFF,
):
if conf_type in entity:
addr += f"_{entity[conf_type]}"
inx = entity.get(CONF_SLAVE) or entity.get(CONF_DEVICE_ADDRESS, 0)
addr += f"_{inx}"
loc_addr: set[str] = {addr}
if CONF_TARGET_TEMP in entity:
loc_addr.add(f"{hub_name}{entity[CONF_TARGET_TEMP]}_{inx}")
if CONF_HVAC_MODE_REGISTER in entity:
loc_addr.add(f"{hub_name}{entity[CONF_HVAC_MODE_REGISTER][CONF_ADDRESS]}_{inx}")
if CONF_FAN_MODE_REGISTER in entity:
loc_addr.add(f"{hub_name}{entity[CONF_FAN_MODE_REGISTER][CONF_ADDRESS]}_{inx}")
dup_addrs = ent_addr.intersection(loc_addr)
if len(dup_addrs) > 0:
for addr in dup_addrs:
err = (
f"Modbus {hub_name}/{name} address {addr} is duplicate, second"
" entry not loaded!"
)
_LOGGER.warning(err)
return False
if name in ent_names:
err = f"Modbus {hub_name}/{name} is duplicate, second entry not loaded!"
_LOGGER.warning(err)
return False
ent_names.add(name)
ent_addr.update(loc_addr)
return True
def check_config(hass: HomeAssistant, config: dict) -> dict:
"""Do final config check."""
hosts: set[str] = set()
hub_names: set[str] = set()
@ -279,97 +372,10 @@ def check_config(config: dict) -> dict:
ent_names: set[str] = set()
ent_addr: set[str] = set()
def validate_modbus(hub: dict, hub_name_inx: int) -> bool:
"""Validate modbus entries."""
host: str = (
hub[CONF_PORT]
if hub[CONF_TYPE] == SERIAL
else f"{hub[CONF_HOST]}_{hub[CONF_PORT]}"
)
if CONF_NAME not in hub:
hub[CONF_NAME] = (
DEFAULT_HUB if not hub_name_inx else f"{DEFAULT_HUB}_{hub_name_inx}"
)
hub_name_inx += 1
err = f"Modbus host/port {host} is missing name, added {hub[CONF_NAME]}!"
_LOGGER.warning(err)
name = hub[CONF_NAME]
if host in hosts or name in hub_names:
err = f"Modbus {name} host/port {host} is duplicate, not loaded!"
_LOGGER.warning(err)
return False
hosts.add(host)
hub_names.add(name)
return True
def validate_entity(
hub_name: str,
component: str,
entity: dict,
minimum_scan_interval: int,
ent_names: set,
ent_addr: set,
) -> bool:
"""Validate entity."""
name = f"{component}.{entity[CONF_NAME]}"
addr = f"{hub_name}{entity[CONF_ADDRESS]}"
scan_interval = entity.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
if 0 < scan_interval < 5:
_LOGGER.warning(
(
"%s %s scan_interval(%d) is lower than 5 seconds, "
"which may cause Home Assistant stability issues"
),
hub_name,
name,
scan_interval,
)
entity[CONF_SCAN_INTERVAL] = scan_interval
minimum_scan_interval = min(scan_interval, minimum_scan_interval)
for conf_type in (
CONF_INPUT_TYPE,
CONF_WRITE_TYPE,
CONF_COMMAND_ON,
CONF_COMMAND_OFF,
):
if conf_type in entity:
addr += f"_{entity[conf_type]}"
inx = entity.get(CONF_SLAVE) or entity.get(CONF_DEVICE_ADDRESS, 0)
addr += f"_{inx}"
loc_addr: set[str] = {addr}
if CONF_TARGET_TEMP in entity:
loc_addr.add(f"{hub_name}{entity[CONF_TARGET_TEMP]}_{inx}")
if CONF_HVAC_MODE_REGISTER in entity:
loc_addr.add(
f"{hub_name}{entity[CONF_HVAC_MODE_REGISTER][CONF_ADDRESS]}_{inx}"
)
if CONF_FAN_MODE_REGISTER in entity:
loc_addr.add(
f"{hub_name}{entity[CONF_FAN_MODE_REGISTER][CONF_ADDRESS]}_{inx}"
)
dup_addrs = ent_addr.intersection(loc_addr)
if len(dup_addrs) > 0:
for addr in dup_addrs:
err = (
f"Modbus {hub_name}/{name} address {addr} is duplicate, second"
" entry not loaded!"
)
_LOGGER.warning(err)
return False
if name in ent_names:
err = f"Modbus {hub_name}/{name} is duplicate, second entry not loaded!"
_LOGGER.warning(err)
return False
ent_names.add(name)
ent_addr.update(loc_addr)
return True
hub_inx = 0
while hub_inx < len(config):
hub = config[hub_inx]
if not validate_modbus(hub, hub_name_inx):
if not validate_modbus(hass, hosts, hub_names, hub, hub_name_inx):
del config[hub_inx]
continue
minimum_scan_interval = 9999
@ -382,6 +388,7 @@ def check_config(config: dict) -> dict:
entities = hub[conf_key]
while entity_inx < len(entities):
if not validate_entity(
hass,
hub[CONF_NAME],
component,
entities[entity_inx],

20
tests/components/modbus/test_init.py
View File

@ -423,9 +423,9 @@ async def test_exception_struct_validator(do_config) -> None:
],
],
)
async def test_check_config(do_config) -> None:
async def test_check_config(hass: HomeAssistant, do_config) -> None:
"""Test duplicate modbus validator."""
check_config(do_config)
check_config(hass, do_config)
assert len(do_config) == 1
@ -476,9 +476,9 @@ async def test_check_config(do_config) -> None:
],
],
)
async def test_check_config_sensor(do_config) -> None:
async def test_check_config_sensor(hass: HomeAssistant, do_config) -> None:
"""Test duplicate entity validator."""
check_config(do_config)
check_config(hass, do_config)
assert len(do_config[0][CONF_SENSORS]) == 1
@ -691,9 +691,9 @@ async def test_check_config_sensor(do_config) -> None:
],
],
)
async def test_check_config_climate(do_config) -> None:
async def test_check_config_climate(hass: HomeAssistant, do_config) -> None:
"""Test duplicate entity validator."""
check_config(do_config)
check_config(hass, do_config)
assert len(do_config[0][CONF_CLIMATES]) == 1
@ -913,9 +913,9 @@ async def test_duplicate_fan_mode_validator(do_config) -> None:
),
],
)
async def test_duplicate_addresses(do_config, sensor_cnt) -> None:
async def test_duplicate_addresses(hass: HomeAssistant, do_config, sensor_cnt) -> None:
"""Test duplicate entity validator."""
check_config(do_config)
check_config(hass, do_config)
use_inx = len(do_config) - 1
assert len(do_config[use_inx][CONF_SENSORS]) == sensor_cnt
@ -948,9 +948,9 @@ async def test_duplicate_addresses(do_config, sensor_cnt) -> None:
],
],
)
async def test_no_duplicate_names(do_config) -> None:
async def test_no_duplicate_names(hass: HomeAssistant, do_config) -> None:
"""Test duplicate entity validator."""
check_config(do_config)
check_config(hass, do_config)
assert len(do_config[0][CONF_SENSORS]) == 1
assert len(do_config[0][CONF_BINARY_SENSORS]) == 1