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

Modbus remove unnecessary get calls (#31966)

Browse Source
This commit is contained in:
Vladimír Záhradník 2020-02-19 04:41:12 +01:00 committed by GitHub
parent 0dad2f8e41
commit 32cd58e03d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 66 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -159,10 +159,10 @@ def setup(hass, config):
def write_register(service):
"""Write Modbus registers."""
unit = int(float(service.data.get(ATTR_UNIT)))
address = int(float(service.data.get(ATTR_ADDRESS)))
value = service.data.get(ATTR_VALUE)
client_name = service.data.get(ATTR_HUB)
unit = int(float(service.data[ATTR_UNIT]))
address = int(float(service.data[ATTR_ADDRESS]))
value = service.data[ATTR_VALUE]
client_name = service.data[ATTR_HUB]
if isinstance(value, list):
hub_collect[client_name].write_registers(
unit, address, [int(float(i)) for i in value]
@ -172,10 +172,10 @@ def setup(hass, config):
def write_coil(service):
"""Write Modbus coil."""
unit = service.data.get(ATTR_UNIT)
address = service.data.get(ATTR_ADDRESS)
state = service.data.get(ATTR_STATE)
client_name = service.data.get(ATTR_HUB)
unit = service.data[ATTR_UNIT]
address = service.data[ATTR_ADDRESS]
state = service.data[ATTR_STATE]
client_name = service.data[ATTR_HUB]
hub_collect[client_name].write_coil(unit, address, state)
hass.bus.listen_once(EVENT_HOMEASSISTANT_START, start_modbus)

22
homeassistant/components/modbus/binary_sensor.py
View File

@ -25,8 +25,8 @@ CONF_INPUTS = "inputs"
CONF_INPUT_TYPE = "input_type"
CONF_ADDRESS = "address"
INPUT_TYPE_COIL = "coil"
INPUT_TYPE_DISCRETE = "discrete_input"
DEFAULT_INPUT_TYPE_COIL = "coil"
DEFAULT_INPUT_TYPE_DISCRETE = "discrete_input"
PLATFORM_SCHEMA = vol.All(
cv.deprecated(CONF_DEPRECATED_COILS, CONF_INPUTS),
@ -43,8 +43,10 @@ PLATFORM_SCHEMA = vol.All(
vol.Optional(CONF_HUB, default=DEFAULT_HUB): cv.string,
vol.Optional(CONF_SLAVE): cv.positive_int,
vol.Optional(
CONF_INPUT_TYPE, default=INPUT_TYPE_COIL
): vol.In([INPUT_TYPE_COIL, INPUT_TYPE_DISCRETE]),
CONF_INPUT_TYPE, default=DEFAULT_INPUT_TYPE_COIL
): vol.In(
[DEFAULT_INPUT_TYPE_COIL, DEFAULT_INPUT_TYPE_DISCRETE]
),
}
),
)
@ -57,16 +59,16 @@ PLATFORM_SCHEMA = vol.All(
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Modbus binary sensors."""
sensors = []
for entry in config.get(CONF_INPUTS):
hub = hass.data[MODBUS_DOMAIN][entry.get(CONF_HUB)]
for entry in config[CONF_INPUTS]:
hub = hass.data[MODBUS_DOMAIN][entry[CONF_HUB]]
sensors.append(
ModbusBinarySensor(
hub,
entry.get(CONF_NAME),
entry[CONF_NAME],
entry.get(CONF_SLAVE),
entry.get(CONF_ADDRESS),
entry[CONF_ADDRESS],
entry.get(CONF_DEVICE_CLASS),
entry.get(CONF_INPUT_TYPE),
entry[CONF_INPUT_TYPE],
)
)
@ -110,7 +112,7 @@ class ModbusBinarySensor(BinarySensorDevice):
def update(self):
"""Update the state of the sensor."""
try:
if self._input_type == INPUT_TYPE_COIL:
if self._input_type == DEFAULT_INPUT_TYPE_COIL:
result = self._hub.read_coils(self._slave, self._address, 1)
else:
result = self._hub.read_discrete_inputs(self._slave, self._address, 1)

46
homeassistant/components/modbus/sensor.py
View File

@ -37,8 +37,8 @@ DATA_TYPE_FLOAT = "float"
DATA_TYPE_INT = "int"
DATA_TYPE_UINT = "uint"
REGISTER_TYPE_HOLDING = "holding"
REGISTER_TYPE_INPUT = "input"
DEFAULT_REGISTER_TYPE_HOLDING = "holding"
DEFAULT_REGISTER_TYPE_INPUT = "input"
def number(value: Any) -> Union[int, float]:
@ -74,9 +74,9 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
vol.Optional(CONF_HUB, default=DEFAULT_HUB): cv.string,
vol.Optional(CONF_OFFSET, default=0): number,
vol.Optional(CONF_PRECISION, default=0): cv.positive_int,
vol.Optional(CONF_REGISTER_TYPE, default=REGISTER_TYPE_HOLDING): vol.In(
[REGISTER_TYPE_HOLDING, REGISTER_TYPE_INPUT]
),
vol.Optional(
CONF_REGISTER_TYPE, default=DEFAULT_REGISTER_TYPE_HOLDING
): vol.In([DEFAULT_REGISTER_TYPE_HOLDING, DEFAULT_REGISTER_TYPE_INPUT]),
vol.Optional(CONF_REVERSE_ORDER, default=False): cv.boolean,
vol.Optional(CONF_SCALE, default=1): number,
vol.Optional(CONF_SLAVE): cv.positive_int,
@ -95,17 +95,17 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
data_types[DATA_TYPE_UINT] = {1: "H", 2: "I", 4: "Q"}
data_types[DATA_TYPE_FLOAT] = {1: "e", 2: "f", 4: "d"}
for register in config.get(CONF_REGISTERS):
for register in config[CONF_REGISTERS]:
structure = ">i"
if register.get(CONF_DATA_TYPE) != DATA_TYPE_CUSTOM:
if register[CONF_DATA_TYPE] != DATA_TYPE_CUSTOM:
try:
structure = ">{}".format(
data_types[register.get(CONF_DATA_TYPE)][register.get(CONF_COUNT)]
data_types[register[CONF_DATA_TYPE]][register[CONF_COUNT]]
)
except KeyError:
_LOGGER.error(
"Unable to detect data type for %s sensor, try a custom type",
register.get(CONF_NAME),
register[CONF_NAME],
)
continue
else:
@ -114,35 +114,33 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
try:
size = struct.calcsize(structure)
except struct.error as err:
_LOGGER.error(
"Error in sensor %s structure: %s", register.get(CONF_NAME), err
)
_LOGGER.error("Error in sensor %s structure: %s", register[CONF_NAME], err)
continue
if register.get(CONF_COUNT) * 2 != size:
if register[CONF_COUNT] * 2 != size:
_LOGGER.error(
"Structure size (%d bytes) mismatch registers count (%d words)",
size,
register.get(CONF_COUNT),
register[CONF_COUNT],
)
continue
hub_name = register.get(CONF_HUB)
hub_name = register[CONF_HUB]
hub = hass.data[MODBUS_DOMAIN][hub_name]
sensors.append(
ModbusRegisterSensor(
hub,
register.get(CONF_NAME),
register[CONF_NAME],
register.get(CONF_SLAVE),
register.get(CONF_REGISTER),
register.get(CONF_REGISTER_TYPE),
register[CONF_REGISTER],
register[CONF_REGISTER_TYPE],
register.get(CONF_UNIT_OF_MEASUREMENT),
register.get(CONF_COUNT),
register.get(CONF_REVERSE_ORDER),
register.get(CONF_SCALE),
register.get(CONF_OFFSET),
register[CONF_COUNT],
register[CONF_REVERSE_ORDER],
register[CONF_SCALE],
register[CONF_OFFSET],
structure,
register.get(CONF_PRECISION),
register[CONF_PRECISION],
register.get(CONF_DEVICE_CLASS),
)
)
@ -223,7 +221,7 @@ class ModbusRegisterSensor(RestoreEntity):
def update(self):
"""Update the state of the sensor."""
try:
if self._register_type == REGISTER_TYPE_INPUT:
if self._register_type == DEFAULT_REGISTER_TYPE_INPUT:
result = self._hub.read_input_registers(
self._slave, self._register, self._count
)

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

@ -32,8 +32,8 @@ CONF_STATE_ON = "state_on"
CONF_VERIFY_REGISTER = "verify_register"
CONF_VERIFY_STATE = "verify_state"
REGISTER_TYPE_HOLDING = "holding"
REGISTER_TYPE_INPUT = "input"
DEFAULT_REGISTER_TYPE_HOLDING = "holding"
DEFAULT_REGISTER_TYPE_INPUT = "input"
REGISTERS_SCHEMA = vol.Schema(
{
@ -42,8 +42,8 @@ REGISTERS_SCHEMA = vol.Schema(
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=REGISTER_TYPE_HOLDING): vol.In(
[REGISTER_TYPE_HOLDING, REGISTER_TYPE_INPUT]
vol.Optional(CONF_REGISTER_TYPE, default=DEFAULT_REGISTER_TYPE_HOLDING): vol.In(
[DEFAULT_REGISTER_TYPE_HOLDING, DEFAULT_REGISTER_TYPE_INPUT]
),
vol.Optional(CONF_SLAVE): cv.positive_int,
vol.Optional(CONF_STATE_OFF): cv.positive_int,
@ -77,30 +77,30 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
"""Read configuration and create Modbus devices."""
switches = []
if CONF_COILS in config:
for coil in config.get(CONF_COILS):
hub_name = coil.get(CONF_HUB)
for coil in config[CONF_COILS]:
hub_name = coil[CONF_HUB]
hub = hass.data[MODBUS_DOMAIN][hub_name]
switches.append(
ModbusCoilSwitch(
hub, coil.get(CONF_NAME), coil.get(CONF_SLAVE), coil.get(CONF_COIL)
hub, coil[CONF_NAME], coil[CONF_SLAVE], coil[CONF_COIL]
)
)
if CONF_REGISTERS in config:
for register in config.get(CONF_REGISTERS):
hub_name = register.get(CONF_HUB)
for register in config[CONF_REGISTERS]:
hub_name = register[CONF_HUB]
hub = hass.data[MODBUS_DOMAIN][hub_name]
switches.append(
ModbusRegisterSwitch(
hub,
register.get(CONF_NAME),
register[CONF_NAME],
register.get(CONF_SLAVE),
register.get(CONF_REGISTER),
register.get(CONF_COMMAND_ON),
register.get(CONF_COMMAND_OFF),
register.get(CONF_VERIFY_STATE),
register[CONF_REGISTER],
register[CONF_COMMAND_ON],
register[CONF_COMMAND_OFF],
register[CONF_VERIFY_STATE],
register.get(CONF_VERIFY_REGISTER),
register.get(CONF_REGISTER_TYPE),
register[CONF_REGISTER_TYPE],
register.get(CONF_STATE_ON),
register.get(CONF_STATE_OFF),
)
@ -242,7 +242,7 @@ class ModbusRegisterSwitch(ModbusCoilSwitch):
"""Set switch on."""
# Only holding register is writable
if self._register_type == REGISTER_TYPE_HOLDING:
if self._register_type == DEFAULT_REGISTER_TYPE_HOLDING:
self._write_register(self._command_on)
if not self._verify_state:
self._is_on = True
@ -251,7 +251,7 @@ class ModbusRegisterSwitch(ModbusCoilSwitch):
"""Set switch off."""
# Only holding register is writable
if self._register_type == REGISTER_TYPE_HOLDING:
if self._register_type == DEFAULT_REGISTER_TYPE_HOLDING:
self._write_register(self._command_off)
if not self._verify_state:
self._is_on = False
@ -282,7 +282,7 @@ class ModbusRegisterSwitch(ModbusCoilSwitch):
def _read_register(self) -> Optional[int]:
try:
if self._register_type == REGISTER_TYPE_INPUT:
if self._register_type == DEFAULT_REGISTER_TYPE_INPUT:
result = self._hub.read_input_registers(self._slave, self._register, 1)
else:
result = self._hub.read_holding_registers(

12
tests/components/modbus/test_modbus_sensor.py
View File

@ -17,8 +17,8 @@ from homeassistant.components.modbus.sensor import (
DATA_TYPE_FLOAT,
DATA_TYPE_INT,
DATA_TYPE_UINT,
REGISTER_TYPE_HOLDING,
REGISTER_TYPE_INPUT,
DEFAULT_REGISTER_TYPE_HOLDING,
DEFAULT_REGISTER_TYPE_INPUT,
)
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.const import (
@ -72,7 +72,7 @@ async def run_test(hass, mock_hub, register_config, register_words, expected):
# Setup inputs for the sensor
read_result = ReadResult(register_words)
if register_config.get(CONF_REGISTER_TYPE) == REGISTER_TYPE_INPUT:
if register_config.get(CONF_REGISTER_TYPE) == DEFAULT_REGISTER_TYPE_INPUT:
mock_hub.read_input_registers.return_value = read_result
else:
mock_hub.read_holding_registers.return_value = read_result
@ -310,7 +310,7 @@ async def test_two_word_input_register(hass, mock_hub):
"""Test reaging of input register."""
register_config = {
CONF_COUNT: 2,
CONF_REGISTER_TYPE: REGISTER_TYPE_INPUT,
CONF_REGISTER_TYPE: DEFAULT_REGISTER_TYPE_INPUT,
CONF_DATA_TYPE: DATA_TYPE_UINT,
CONF_SCALE: 1,
CONF_OFFSET: 0,
@ -329,7 +329,7 @@ async def test_two_word_holding_register(hass, mock_hub):
"""Test reaging of holding register."""
register_config = {
CONF_COUNT: 2,
CONF_REGISTER_TYPE: REGISTER_TYPE_HOLDING,
CONF_REGISTER_TYPE: DEFAULT_REGISTER_TYPE_HOLDING,
CONF_DATA_TYPE: DATA_TYPE_UINT,
CONF_SCALE: 1,
CONF_OFFSET: 0,
@ -348,7 +348,7 @@ async def test_float_data_type(hass, mock_hub):
"""Test floating point register data type."""
register_config = {
CONF_COUNT: 2,
CONF_REGISTER_TYPE: REGISTER_TYPE_HOLDING,
CONF_REGISTER_TYPE: DEFAULT_REGISTER_TYPE_HOLDING,
CONF_DATA_TYPE: DATA_TYPE_FLOAT,
CONF_SCALE: 1,
CONF_OFFSET: 0,