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

Restructure modbus switch, and solve read/write conflicts (#49386)

Browse Source 
Remove old style configuration from switch.
- The old style configuration allowed a number of illegal
  configurations, instead of adding if...log. in setup we
  only support the new configuration style.

Add new/changed configuration switch.

Removed verify_state and cleaned configuration to avoid possible
illegal configurations.
This commit is contained in:
jan iversen 2021-04-30 16:47:18 +02:00 committed by GitHub
parent 39e8a92596
commit 684ea9e49b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 153 additions and 355 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -84,8 +84,8 @@ from .const import (
CONF_SWAP_WORD,
CONF_SWAP_WORD_BYTE,
CONF_TARGET_TEMP,
CONF_VERIFY_REGISTER,
CONF_VERIFY_STATE,
CONF_VERIFY,
CONF_WRITE_TYPE,
DATA_TYPE_CUSTOM,
DATA_TYPE_FLOAT,
DATA_TYPE_INT,
@ -178,15 +178,24 @@ SWITCH_SCHEMA = BASE_COMPONENT_SCHEMA.extend(
{
vol.Required(CONF_ADDRESS): cv.positive_int,
vol.Optional(CONF_DEVICE_CLASS): SWITCH_DEVICE_CLASSES_SCHEMA,
vol.Optional(CONF_INPUT_TYPE, default=CALL_TYPE_REGISTER_HOLDING): vol.In(
[CALL_TYPE_REGISTER_HOLDING, CALL_TYPE_REGISTER_INPUT, CALL_TYPE_COIL]
vol.Optional(CONF_WRITE_TYPE, default=CALL_TYPE_REGISTER_HOLDING): vol.In(
[CALL_TYPE_REGISTER_HOLDING, CALL_TYPE_COIL]
),
vol.Optional(CONF_COMMAND_OFF, default=0x00): cv.positive_int,
vol.Optional(CONF_COMMAND_ON, default=0x01): cv.positive_int,
vol.Optional(CONF_STATE_OFF): cv.positive_int,
vol.Optional(CONF_STATE_ON): cv.positive_int,
vol.Optional(CONF_VERIFY_REGISTER): cv.positive_int,
vol.Optional(CONF_VERIFY_STATE, default=True): cv.boolean,
vol.Optional(CONF_VERIFY): {
vol.Optional(CONF_ADDRESS): cv.positive_int,
vol.Optional(CONF_INPUT_TYPE): vol.In(
[
CALL_TYPE_REGISTER_HOLDING,
CALL_TYPE_DISCRETE,
CALL_TYPE_REGISTER_INPUT,
CALL_TYPE_COIL,
]
),
vol.Optional(CONF_STATE_OFF): cv.positive_int,
vol.Optional(CONF_STATE_ON): cv.positive_int,
},
}
)

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

@ -42,8 +42,10 @@ CONF_SWAP_WORD = "word"
CONF_SWAP_WORD_BYTE = "word_byte"
CONF_SWITCH = "switch"
CONF_TARGET_TEMP = "target_temp_register"
CONF_VERIFY = "verify"
CONF_VERIFY_REGISTER = "verify_register"
CONF_VERIFY_STATE = "verify_state"
CONF_WRITE_TYPE = "write_type"
# service call attributes
ATTR_ADDRESS = "address"

311
homeassistant/components/modbus/switch.py
View File

@ -1,14 +1,10 @@
"""Support for Modbus switches."""
from __future__ import annotations
from abc import ABC, abstractmethod
from datetime import timedelta
import logging
from typing import Any
import voluptuous as vol
from homeassistant.components.switch import PLATFORM_SCHEMA, SwitchEntity
from homeassistant.components.switch import SwitchEntity
from homeassistant.const import (
CONF_ADDRESS,
CONF_COMMAND_OFF,
@ -20,27 +16,20 @@ from homeassistant.const import (
STATE_ON,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import ConfigType
from .const import (
CALL_TYPE_COIL,
CALL_TYPE_DISCRETE,
CALL_TYPE_REGISTER_HOLDING,
CALL_TYPE_REGISTER_INPUT,
CONF_COILS,
CONF_HUB,
CONF_INPUT_TYPE,
CONF_REGISTER,
CONF_REGISTER_TYPE,
CONF_REGISTERS,
CONF_STATE_OFF,
CONF_STATE_ON,
CONF_VERIFY_REGISTER,
CONF_VERIFY_STATE,
DEFAULT_HUB,
DEFAULT_SCAN_INTERVAL,
CONF_VERIFY,
CONF_WRITE_TYPE,
MODBUS_DOMAIN,
)
from .modbus import ModbusHub
@ -48,95 +37,22 @@ from .modbus import ModbusHub
_LOGGER = logging.getLogger(__name__)
REGISTERS_SCHEMA = vol.Schema(
{
vol.Required(CONF_COMMAND_OFF): cv.positive_int,
vol.Required(CONF_COMMAND_ON): cv.positive_int,
vol.Required(CONF_NAME): cv.string,
vol.Required(CONF_REGISTER): cv.positive_int,
vol.Optional(CONF_HUB, default=DEFAULT_HUB): cv.string,
vol.Optional(CONF_REGISTER_TYPE, default=CALL_TYPE_REGISTER_HOLDING): vol.In(
[CALL_TYPE_REGISTER_HOLDING, CALL_TYPE_REGISTER_INPUT]
),
vol.Optional(CONF_SLAVE): cv.positive_int,
vol.Optional(CONF_STATE_OFF): cv.positive_int,
vol.Optional(CONF_STATE_ON): cv.positive_int,
vol.Optional(CONF_VERIFY_REGISTER): cv.positive_int,
vol.Optional(CONF_VERIFY_STATE, default=True): cv.boolean,
}
)
COILS_SCHEMA = vol.Schema(
{
vol.Required(CALL_TYPE_COIL): cv.positive_int,
vol.Required(CONF_NAME): cv.string,
vol.Required(CONF_SLAVE): cv.positive_int,
vol.Optional(CONF_HUB, default=DEFAULT_HUB): cv.string,
}
)
PLATFORM_SCHEMA = vol.All(
cv.has_at_least_one_key(CONF_COILS, CONF_REGISTERS),
PLATFORM_SCHEMA.extend(
{
vol.Optional(CONF_COILS): [COILS_SCHEMA],
vol.Optional(CONF_REGISTERS): [REGISTERS_SCHEMA],
}
),
)
async def async_setup_platform(
hass: HomeAssistant, config: ConfigType, async_add_entities, discovery_info=None
):
"""Read configuration and create Modbus switches."""
switches = []
#  check for old config:
if discovery_info is None:
_LOGGER.warning(
"Switch configuration is deprecated, will be removed in a future release"
)
discovery_info = {
CONF_NAME: "no name",
CONF_SWITCHES: [],
}
if CONF_COILS in config:
discovery_info[CONF_SWITCHES].extend(config[CONF_COILS])
if CONF_REGISTERS in config:
discovery_info[CONF_SWITCHES].extend(config[CONF_REGISTERS])
for entry in discovery_info[CONF_SWITCHES]:
if CALL_TYPE_COIL in entry:
entry[CONF_ADDRESS] = entry[CALL_TYPE_COIL]
entry[CONF_INPUT_TYPE] = CALL_TYPE_COIL
del entry[CALL_TYPE_COIL]
if CONF_REGISTER in entry:
entry[CONF_ADDRESS] = entry[CONF_REGISTER]
del entry[CONF_REGISTER]
if CONF_REGISTER_TYPE in entry:
entry[CONF_INPUT_TYPE] = entry[CONF_REGISTER_TYPE]
del entry[CONF_REGISTER_TYPE]
if CONF_SCAN_INTERVAL not in entry:
entry[CONF_SCAN_INTERVAL] = DEFAULT_SCAN_INTERVAL
config = None
for entry in discovery_info[CONF_SWITCHES]:
if CONF_HUB in entry:
# from old config!
hub: ModbusHub = hass.data[MODBUS_DOMAIN][entry[CONF_HUB]]
else:
hub: ModbusHub = hass.data[MODBUS_DOMAIN][discovery_info[CONF_NAME]]
if entry[CONF_INPUT_TYPE] == CALL_TYPE_COIL:
switches.append(ModbusCoilSwitch(hub, entry))
else:
switches.append(ModbusRegisterSwitch(hub, entry))
hub: ModbusHub = hass.data[MODBUS_DOMAIN][discovery_info[CONF_NAME]]
switches.append(ModbusSwitch(hub, entry))
async_add_entities(switches)
class ModbusBaseSwitch(SwitchEntity, RestoreEntity, ABC):
class ModbusSwitch(SwitchEntity, RestoreEntity):
"""Base class representing a Modbus switch."""
def __init__(self, hub: ModbusHub, config: dict[str, Any]):
def __init__(self, hub: ModbusHub, config: dict):
"""Initialize the switch."""
self._hub: ModbusHub = hub
self._name = config[CONF_NAME]
@ -144,6 +60,36 @@ class ModbusBaseSwitch(SwitchEntity, RestoreEntity, ABC):
self._is_on = None
self._available = True
self._scan_interval = timedelta(seconds=config[CONF_SCAN_INTERVAL])
self._address = config[CONF_ADDRESS]
if config[CONF_WRITE_TYPE] == CALL_TYPE_COIL:
self._write_func = self._hub.write_coil
self._command_on = 0x01
self._command_off = 0x00
else:
self._write_func = self._hub.write_register
self._command_on = config[CONF_COMMAND_ON]
self._command_off = config[CONF_COMMAND_OFF]
if CONF_VERIFY in config:
self._verify_active = True
self._verify_address = config[CONF_VERIFY].get(
CONF_ADDRESS, config[CONF_ADDRESS]
)
self._verify_type = config[CONF_VERIFY].get(
CONF_INPUT_TYPE, config[CONF_WRITE_TYPE]
)
self._state_on = config[CONF_VERIFY].get(CONF_STATE_ON, self._command_on)
self._state_off = config[CONF_VERIFY].get(CONF_STATE_OFF, self._command_off)
if self._verify_type == CALL_TYPE_REGISTER_HOLDING:
self._read_func = self._hub.read_holding_registers
elif self._verify_type == CALL_TYPE_DISCRETE:
self._read_func = self._hub.read_discrete_inputs
elif self._verify_type == CALL_TYPE_REGISTER_INPUT:
self._read_func = self._hub.read_input_registers
else: # self._verify_type == CALL_TYPE_COIL:
self._read_func = self._hub.read_coils
else:
self._verify_active = False
async def async_added_to_hass(self):
"""Handle entity which will be added."""
@ -151,13 +97,10 @@ class ModbusBaseSwitch(SwitchEntity, RestoreEntity, ABC):
if state:
self._is_on = state.state == STATE_ON
async_track_time_interval(
self.hass, lambda arg: self._update(), self._scan_interval
)
@abstractmethod
def _update(self):
"""Update the entity state."""
if self._verify_active:
async_track_time_interval(
self.hass, lambda arg: self._update(), self._scan_interval
)
@property
def is_on(self):
@ -171,12 +114,7 @@ class ModbusBaseSwitch(SwitchEntity, RestoreEntity, ABC):
@property
def should_poll(self):
"""Return True if entity has to be polled for state.
False if entity pushes its state to HA.
"""
# Handle polling directly in this entity
"""Return True if entity has to be polled for state."""
return False
@property
@ -184,126 +122,61 @@ class ModbusBaseSwitch(SwitchEntity, RestoreEntity, ABC):
"""Return True if entity is available."""
return self._available
class ModbusCoilSwitch(ModbusBaseSwitch, SwitchEntity):
"""Representation of a Modbus coil switch."""
def __init__(self, hub: ModbusHub, config: dict[str, Any]):
"""Initialize the coil switch."""
super().__init__(hub, config)
self._coil = config[CONF_ADDRESS]
def turn_on(self, **kwargs):
"""Set switch on."""
self._write_coil(self._coil, True)
self._is_on = True
self.schedule_update_ha_state()
def turn_off(self, **kwargs):
"""Set switch off."""
self._write_coil(self._coil, False)
self._is_on = False
self.schedule_update_ha_state()
def _update(self):
"""Update the state of the switch."""
self._is_on = self._read_coil(self._coil)
self.schedule_update_ha_state()
def _read_coil(self, coil) -> bool:
"""Read coil using the Modbus hub slave."""
result = self._hub.read_coils(self._slave, coil, 1)
if result is None:
result = self._write_func(self._slave, self._address, self._command_on)
if result is False:
self._available = False
return False
self._available = True
# bits[0] select the lowest bit in result,
# is_on for a binary_sensor is true if the bit is 1
# The other bits are not considered.
return bool(result.bits[0] & 1)
def _write_coil(self, coil, value):
"""Write coil using the Modbus hub slave."""
self._available = self._hub.write_coil(self._slave, coil, value)
class ModbusRegisterSwitch(ModbusBaseSwitch, SwitchEntity):
"""Representation of a Modbus register switch."""
def __init__(self, hub: ModbusHub, config: dict[str, Any]):
"""Initialize the register switch."""
super().__init__(hub, config)
self._register = config[CONF_ADDRESS]
self._command_on = config[CONF_COMMAND_ON]
self._command_off = config[CONF_COMMAND_OFF]
self._state_on = config.get(CONF_STATE_ON, self._command_on)
self._state_off = config.get(CONF_STATE_OFF, self._command_off)
self._verify_state = config[CONF_VERIFY_STATE]
self._verify_register = config.get(CONF_VERIFY_REGISTER, self._register)
self._register_type = config[CONF_INPUT_TYPE]
self._available = True
self._is_on = None
def turn_on(self, **kwargs):
"""Set switch on."""
# Only holding register is writable
if self._register_type == CALL_TYPE_REGISTER_HOLDING:
self._write_register(self._command_on)
if not self._verify_state:
self._is_on = True
self.schedule_update_ha_state()
def turn_off(self, **kwargs):
"""Set switch off."""
# Only holding register is writable
if self._register_type == CALL_TYPE_REGISTER_HOLDING:
self._write_register(self._command_off)
if not self._verify_state:
self._is_on = False
self.schedule_update_ha_state()
@property
def available(self) -> bool:
"""Return True if entity is available."""
return self._available
def _update(self):
"""Update the state of the switch."""
if not self._verify_state:
return
value = self._read_register()
if value == self._state_on:
self._is_on = True
elif value == self._state_off:
self._is_on = False
elif value is not None:
_LOGGER.error(
"Unexpected response from hub %s, slave %s register %s, got 0x%2x",
self._hub.name,
self._slave,
self._register,
value,
)
self.schedule_update_ha_state()
def _read_register(self) -> int | None:
if self._register_type == CALL_TYPE_REGISTER_INPUT:
result = self._hub.read_input_registers(
self._slave, self._verify_register, 1
)
self.schedule_update_ha_state()
else:
result = self._hub.read_holding_registers(
self._slave, self._verify_register, 1
)
self._available = True
if self._verify_active:
self._update()
else:
self._is_on = True
self.schedule_update_ha_state()
def turn_off(self, **kwargs):
"""Set switch off."""
result = self._write_func(self._slave, self._address, self._command_off)
if result is False:
self._available = False
self.schedule_update_ha_state()
else:
self._available = True
if self._verify_active:
self._update()
else:
self._is_on = False
self.schedule_update_ha_state()
def _update(self):
"""Update the entity state."""
if not self._verify_active:
return
result = self._read_func(self._slave, self._verify_address, 1)
if result is None:
self._available = False
self.schedule_update_ha_state()
return
self._available = True
return int(result.registers[0])
def _write_register(self, value):
"""Write holding register using the Modbus hub slave."""
self._available = self._hub.write_register(self._slave, self._register, value)
if self._verify_type == CALL_TYPE_COIL:
self._is_on = bool(result.bits[0] & 1)
else:
value = int(result.registers[0])
if value == self._state_on:
self._is_on = True
elif value == self._state_off:
self._is_on = False
elif value is not None:
_LOGGER.error(
"Unexpected response from hub %s, slave %s register %s, got 0x%2x",
self._hub.name,
self._slave,
self._verify_address,
value,
)
self.schedule_update_ha_state()

170
tests/components/modbus/test_modbus_switch.py
View File

@ -7,13 +7,10 @@ from homeassistant.components.modbus.const import (
CALL_TYPE_REGISTER_INPUT,
CONF_COILS,
CONF_INPUT_TYPE,
CONF_REGISTER,
CONF_REGISTER_TYPE,
CONF_REGISTERS,
CONF_STATE_OFF,
CONF_STATE_ON,
CONF_VERIFY_REGISTER,
CONF_VERIFY_STATE,
CONF_VERIFY,
CONF_WRITE_TYPE,
)
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import (
@ -32,126 +29,44 @@ from .conftest import base_config_test, base_test
@pytest.mark.parametrize(
"array_type, do_config",
"do_config",
[
(
None,
{
CONF_ADDRESS: 1234,
},
),
(
None,
{
CONF_ADDRESS: 1234,
CONF_INPUT_TYPE: CALL_TYPE_COIL,
},
),
(
None,
{
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_STATE_OFF: 0,
CONF_STATE_ON: 1,
CONF_VERIFY_REGISTER: 1235,
CONF_VERIFY_STATE: False,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
CONF_DEVICE_CLASS: "switch",
{
CONF_ADDRESS: 1234,
},
{
CONF_ADDRESS: 1234,
CONF_WRITE_TYPE: CALL_TYPE_COIL,
},
{
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
CONF_DEVICE_CLASS: "switch",
CONF_VERIFY: {
CONF_INPUT_TYPE: CALL_TYPE_REGISTER_HOLDING,
},
),
(
None,
{
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_ADDRESS: 1235,
CONF_STATE_OFF: 0,
CONF_STATE_ON: 1,
CONF_VERIFY_REGISTER: 1235,
CONF_VERIFY_STATE: True,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
CONF_DEVICE_CLASS: "switch",
},
},
{
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
CONF_DEVICE_CLASS: "switch",
CONF_VERIFY: {
CONF_INPUT_TYPE: CALL_TYPE_REGISTER_INPUT,
},
),
(
None,
{
CONF_ADDRESS: 1234,
CONF_INPUT_TYPE: CALL_TYPE_COIL,
CONF_SLAVE: 1,
CONF_DEVICE_CLASS: "switch",
CONF_INPUT_TYPE: CALL_TYPE_COIL,
},
),
(
CONF_REGISTERS,
{
CONF_REGISTER: 1234,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
},
),
(
CONF_REGISTERS,
{
CONF_REGISTER: 1234,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
},
),
(
CONF_COILS,
{
CALL_TYPE_COIL: 1234,
CONF_SLAVE: 1,
},
),
(
CONF_REGISTERS,
{
CONF_REGISTER: 1234,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
CONF_SLAVE: 1,
CONF_ADDRESS: 1235,
CONF_STATE_OFF: 0,
CONF_STATE_ON: 1,
CONF_VERIFY_REGISTER: 1235,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
CONF_VERIFY_STATE: True,
CONF_REGISTER_TYPE: CALL_TYPE_REGISTER_INPUT,
},
),
(
CONF_REGISTERS,
{
CONF_REGISTER: 1234,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
CONF_SLAVE: 1,
CONF_STATE_OFF: 0,
CONF_STATE_ON: 1,
CONF_VERIFY_REGISTER: 1235,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
CONF_VERIFY_STATE: True,
CONF_REGISTER_TYPE: CALL_TYPE_REGISTER_HOLDING,
},
),
(
CONF_COILS,
{
CALL_TYPE_COIL: 1234,
CONF_SLAVE: 1,
},
),
},
],
)
async def test_config_switch(hass, array_type, do_config):
async def test_config_switch(hass, do_config):
"""Run test for switch."""
device_name = "test_switch"
@ -166,8 +81,8 @@ async def test_config_switch(hass, array_type, do_config):
device_name,
SWITCH_DOMAIN,
CONF_SWITCHES,
array_type,
method_discovery=(array_type is None),
None,
method_discovery=True,
)
@ -204,7 +119,8 @@ async def test_coil_switch(hass, regs, expected):
{
CONF_NAME: switch_name,
CONF_ADDRESS: 1234,
CONF_INPUT_TYPE: CALL_TYPE_COIL,
CONF_WRITE_TYPE: CALL_TYPE_COIL,
CONF_VERIFY: {},
},
switch_name,
SWITCH_DOMAIN,
@ -250,18 +166,19 @@ async def test_register_switch(hass, regs, expected):
hass,
{
CONF_NAME: switch_name,
CONF_REGISTER: 1234,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
CONF_VERIFY: {},
},
switch_name,
SWITCH_DOMAIN,
CONF_SWITCHES,
CONF_REGISTERS,
None,
regs,
expected,
method_discovery=False,
method_discovery=True,
scan_interval=5,
)
assert state == expected
@ -278,10 +195,6 @@ async def test_register_switch(hass, regs, expected):
[0x04],
STATE_OFF,
),
(
[0xFF],
STATE_OFF,
),
],
)
async def test_register_state_switch(hass, regs, expected):
@ -291,18 +204,19 @@ async def test_register_state_switch(hass, regs, expected):
hass,
{
CONF_NAME: switch_name,
CONF_REGISTER: 1234,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x04,
CONF_COMMAND_ON: 0x40,
CONF_VERIFY: {},
},
switch_name,
SWITCH_DOMAIN,
CONF_SWITCHES,
CONF_REGISTERS,
None,
regs,
expected,
method_discovery=False,
method_discovery=True,
scan_interval=5,
)
assert state == expected