From c650deef98a50c6423e7b65a39d3ea8d52014046 Mon Sep 17 00:00:00 2001 From: jan iversen Date: Thu, 20 May 2021 16:56:11 +0200 Subject: [PATCH] Add base class for all modbus platforms (#50878) * Add base for all platforms. * Please pylint. --- .../components/modbus/base_platform.py | 67 +++++++++++++++++++ .../components/modbus/binary_sensor.py | 49 ++------------ homeassistant/components/modbus/climate.py | 38 +++-------- homeassistant/components/modbus/cover.py | 47 +++---------- homeassistant/components/modbus/sensor.py | 46 ++----------- homeassistant/components/modbus/switch.py | 33 ++------- 6 files changed, 98 insertions(+), 182 deletions(-) create mode 100644 homeassistant/components/modbus/base_platform.py diff --git a/homeassistant/components/modbus/base_platform.py b/homeassistant/components/modbus/base_platform.py new file mode 100644 index 00000000000..ddfe11717de --- /dev/null +++ b/homeassistant/components/modbus/base_platform.py @@ -0,0 +1,67 @@ +"""Base implementation for all modbus platforms.""" +from __future__ import annotations + +from abc import abstractmethod +from datetime import timedelta +import logging +from typing import Any + +from homeassistant.const import ( + CONF_ADDRESS, + CONF_DEVICE_CLASS, + CONF_NAME, + CONF_SCAN_INTERVAL, + CONF_SLAVE, +) +from homeassistant.helpers.entity import Entity +from homeassistant.helpers.event import async_track_time_interval + +from .const import CONF_INPUT_TYPE +from .modbus import ModbusHub + +PARALLEL_UPDATES = 1 +_LOGGER = logging.getLogger(__name__) + + +class BasePlatform(Entity): + """Base for readonly platforms.""" + + def __init__(self, hub: ModbusHub, entry: dict[str, Any]) -> None: + """Initialize the Modbus binary sensor.""" + self._hub = hub + self._name = entry[CONF_NAME] + self._slave = entry.get(CONF_SLAVE) + self._address = int(entry[CONF_ADDRESS]) + self._device_class = entry.get(CONF_DEVICE_CLASS) + self._input_type = entry[CONF_INPUT_TYPE] + self._value = None + self._available = True + self._scan_interval = timedelta(seconds=entry[CONF_SCAN_INTERVAL]) + + @abstractmethod + async def async_update(self, now=None): + """Virtual function to be overwritten.""" + + async def async_base_added_to_hass(self): + """Handle entity which will be added.""" + async_track_time_interval(self.hass, self.async_update, self._scan_interval) + + @property + def name(self): + """Return the name of the sensor.""" + return self._name + + @property + def should_poll(self): + """Return True if entity has to be polled for state.""" + return False + + @property + def device_class(self) -> str | None: + """Return the device class of the sensor.""" + return self._device_class + + @property + def available(self) -> bool: + """Return True if entity is available.""" + return self._available diff --git a/homeassistant/components/modbus/binary_sensor.py b/homeassistant/components/modbus/binary_sensor.py index 3605b623c3c..045447f7246 100644 --- a/homeassistant/components/modbus/binary_sensor.py +++ b/homeassistant/components/modbus/binary_sensor.py @@ -1,7 +1,6 @@ """Support for Modbus Coil and Discrete Input sensors.""" from __future__ import annotations -from datetime import timedelta import logging import voluptuous as vol @@ -21,9 +20,9 @@ from homeassistant.const import ( ) 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.typing import ConfigType, DiscoveryInfoType +from .base_platform import BasePlatform from .const import ( CALL_TYPE_COIL, CALL_TYPE_DISCRETE, @@ -92,65 +91,25 @@ async def async_setup_platform( hub = hass.data[MODBUS_DOMAIN][discovery_info[CONF_NAME]] if CONF_SCAN_INTERVAL not in entry: entry[CONF_SCAN_INTERVAL] = DEFAULT_SCAN_INTERVAL - sensors.append(ModbusBinarySensor(hub, hass, entry)) + sensors.append(ModbusBinarySensor(hub, entry)) async_add_entities(sensors) -class ModbusBinarySensor(BinarySensorEntity): +class ModbusBinarySensor(BasePlatform, BinarySensorEntity): """Modbus binary sensor.""" - def __init__(self, hub, hass, entry): - """Initialize the Modbus binary sensor.""" - self._hub = hub - self._hass = hass - self._name = entry[CONF_NAME] - self._slave = entry.get(CONF_SLAVE) - self._address = int(entry[CONF_ADDRESS]) - self._device_class = entry.get(CONF_DEVICE_CLASS) - self._input_type = entry[CONF_INPUT_TYPE] - self._value = None - self._available = True - self._scan_interval = timedelta(seconds=entry[CONF_SCAN_INTERVAL]) - async def async_added_to_hass(self): """Handle entity which will be added.""" - async_track_time_interval(self._hass, self.async_update, self._scan_interval) - - @property - def name(self): - """Return the name of the sensor.""" - return self._name + await self.async_base_added_to_hass() @property def is_on(self): """Return the state of the sensor.""" return self._value - @property - def device_class(self) -> str | None: - """Return the device class of the sensor.""" - return self._device_class - - @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 False - - @property - def available(self) -> bool: - """Return True if entity is available.""" - return self._available - async def async_update(self, now=None): """Update the state of the sensor.""" - # remark "now" is a dummy parameter to avoid problems with - # async_track_time_interval result = await self._hub.async_pymodbus_call( self._slave, self._address, 1, self._input_type ) diff --git a/homeassistant/components/modbus/climate.py b/homeassistant/components/modbus/climate.py index a8a3f2a8557..d23146bd2ba 100644 --- a/homeassistant/components/modbus/climate.py +++ b/homeassistant/components/modbus/climate.py @@ -1,7 +1,6 @@ """Support for Generic Modbus Thermostats.""" from __future__ import annotations -from datetime import timedelta import logging import struct from typing import Any @@ -12,19 +11,18 @@ from homeassistant.components.climate.const import ( SUPPORT_TARGET_TEMPERATURE, ) from homeassistant.const import ( + CONF_ADDRESS, CONF_NAME, CONF_OFFSET, - CONF_SCAN_INTERVAL, - CONF_SLAVE, CONF_STRUCTURE, CONF_TEMPERATURE_UNIT, TEMP_CELSIUS, TEMP_FAHRENHEIT, ) from homeassistant.core import HomeAssistant -from homeassistant.helpers.event import async_track_time_interval from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType +from .base_platform import BasePlatform from .const import ( ATTR_TEMPERATURE, CALL_TYPE_REGISTER_HOLDING, @@ -34,6 +32,7 @@ from .const import ( CONF_CURRENT_TEMP_REGISTER_TYPE, CONF_DATA_COUNT, CONF_DATA_TYPE, + CONF_INPUT_TYPE, CONF_MAX_TEMP, CONF_MIN_TEMP, CONF_PRECISION, @@ -99,7 +98,7 @@ async def async_setup_platform( async_add_entities(entities) -class ModbusThermostat(ClimateEntity): +class ModbusThermostat(BasePlatform, ClimateEntity): """Representation of a Modbus Thermostat.""" def __init__( @@ -108,9 +107,9 @@ class ModbusThermostat(ClimateEntity): config: dict[str, Any], ) -> None: """Initialize the modbus thermostat.""" - self._hub: ModbusHub = hub - self._name = config[CONF_NAME] - self._slave = config.get(CONF_SLAVE) + config[CONF_ADDRESS] = "0" + config[CONF_INPUT_TYPE] = "" + super().__init__(hub, config) self._target_temperature_register = config[CONF_TARGET_TEMP] self._current_temperature_register = config[CONF_CURRENT_TEMP] self._current_temperature_register_type = config[ @@ -123,26 +122,15 @@ class ModbusThermostat(ClimateEntity): self._count = config[CONF_DATA_COUNT] self._precision = config[CONF_PRECISION] self._scale = config[CONF_SCALE] - self._scan_interval = timedelta(seconds=config[CONF_SCAN_INTERVAL]) self._offset = config[CONF_OFFSET] self._unit = config[CONF_TEMPERATURE_UNIT] self._max_temp = config[CONF_MAX_TEMP] self._min_temp = config[CONF_MIN_TEMP] self._temp_step = config[CONF_STEP] - self._available = True async def async_added_to_hass(self): """Handle entity which will be added.""" - async_track_time_interval(self.hass, self.async_update, self._scan_interval) - - @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 False + await self.async_base_added_to_hass() @property def supported_features(self): @@ -164,11 +152,6 @@ class ModbusThermostat(ClimateEntity): # Home Assistant expects this method. # We'll keep it here to avoid getting exceptions. - @property - def name(self): - """Return the name of the climate device.""" - return self._name - @property def current_temperature(self): """Return the current temperature.""" @@ -217,11 +200,6 @@ class ModbusThermostat(ClimateEntity): self._available = result is not None await self.async_update() - @property - def available(self) -> bool: - """Return True if entity is available.""" - return self._available - async def async_update(self, now=None): """Update Target & Current Temperature.""" # remark "now" is a dummy parameter to avoid problems with diff --git a/homeassistant/components/modbus/cover.py b/homeassistant/components/modbus/cover.py index e04d4bd8ce3..f6e1b21b0cf 100644 --- a/homeassistant/components/modbus/cover.py +++ b/homeassistant/components/modbus/cover.py @@ -1,17 +1,14 @@ """Support for Modbus covers.""" from __future__ import annotations -from datetime import timedelta import logging from typing import Any from homeassistant.components.cover import SUPPORT_CLOSE, SUPPORT_OPEN, CoverEntity from homeassistant.const import ( + CONF_ADDRESS, CONF_COVERS, - CONF_DEVICE_CLASS, CONF_NAME, - CONF_SCAN_INTERVAL, - CONF_SLAVE, STATE_CLOSED, STATE_CLOSING, STATE_OPEN, @@ -20,15 +17,16 @@ from homeassistant.const import ( STATE_UNKNOWN, ) from homeassistant.core import HomeAssistant -from homeassistant.helpers.event import async_track_time_interval from homeassistant.helpers.restore_state import RestoreEntity from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType +from .base_platform import BasePlatform from .const import ( CALL_TYPE_COIL, CALL_TYPE_REGISTER_HOLDING, CALL_TYPE_WRITE_COIL, CALL_TYPE_WRITE_REGISTER, + CONF_INPUT_TYPE, CONF_REGISTER, CONF_STATE_CLOSED, CONF_STATE_CLOSING, @@ -67,7 +65,7 @@ async def async_setup_platform( async_add_entities(covers) -class ModbusCover(CoverEntity, RestoreEntity): +class ModbusCover(BasePlatform, CoverEntity, RestoreEntity): """Representation of a Modbus cover.""" def __init__( @@ -76,21 +74,17 @@ class ModbusCover(CoverEntity, RestoreEntity): config: dict[str, Any], ) -> None: """Initialize the modbus cover.""" - self._hub: ModbusHub = hub + config[CONF_ADDRESS] = "0" + config[CONF_INPUT_TYPE] = "" + super().__init__(hub, config) self._coil = config.get(CALL_TYPE_COIL) - self._device_class = config.get(CONF_DEVICE_CLASS) - self._name = config[CONF_NAME] self._register = config.get(CONF_REGISTER) - self._slave = config.get(CONF_SLAVE) self._state_closed = config[CONF_STATE_CLOSED] self._state_closing = config[CONF_STATE_CLOSING] self._state_open = config[CONF_STATE_OPEN] self._state_opening = config[CONF_STATE_OPENING] self._status_register = config.get(CONF_STATUS_REGISTER) self._status_register_type = config[CONF_STATUS_REGISTER_TYPE] - self._scan_interval = timedelta(seconds=config[CONF_SCAN_INTERVAL]) - self._value = None - self._available = True # If we read cover status from coil, and not from optional status register, # we interpret boolean value False as closed cover, and value True as open cover. @@ -114,6 +108,7 @@ class ModbusCover(CoverEntity, RestoreEntity): async def async_added_to_hass(self): """Handle entity which will be added.""" + await self.async_base_added_to_hass() state = await self.async_get_last_state() if state: convert = { @@ -126,28 +121,11 @@ class ModbusCover(CoverEntity, RestoreEntity): } self._value = convert[state.state] - async_track_time_interval(self.hass, self.async_update, self._scan_interval) - - @property - def device_class(self) -> str | None: - """Return the device class of the sensor.""" - return self._device_class - - @property - def name(self): - """Return the name of the switch.""" - return self._name - @property def supported_features(self): """Flag supported features.""" return SUPPORT_OPEN | SUPPORT_CLOSE - @property - def available(self) -> bool: - """Return True if entity is available.""" - return self._available - @property def is_opening(self): """Return if the cover is opening or not.""" @@ -163,15 +141,6 @@ class ModbusCover(CoverEntity, RestoreEntity): """Return if the cover is closed or not.""" return self._value == self._state_closed - @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 False - async def async_open_cover(self, **kwargs: Any) -> None: """Open cover.""" result = await self._hub.async_pymodbus_call( diff --git a/homeassistant/components/modbus/sensor.py b/homeassistant/components/modbus/sensor.py index 614925b79a6..85bb591711c 100644 --- a/homeassistant/components/modbus/sensor.py +++ b/homeassistant/components/modbus/sensor.py @@ -1,7 +1,6 @@ """Support for Modbus Register sensors.""" from __future__ import annotations -from datetime import timedelta import logging import struct @@ -26,11 +25,11 @@ from homeassistant.const import ( ) 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, DiscoveryInfoType from . import number +from .base_platform import BasePlatform from .const import ( CALL_TYPE_REGISTER_HOLDING, CALL_TYPE_REGISTER_INPUT, @@ -194,7 +193,7 @@ async def async_setup_platform( async_add_entities(sensors) -class ModbusRegisterSensor(RestoreEntity, SensorEntity): +class ModbusRegisterSensor(BasePlatform, RestoreEntity, SensorEntity): """Modbus register sensor.""" def __init__( @@ -204,12 +203,9 @@ class ModbusRegisterSensor(RestoreEntity, SensorEntity): structure, ): """Initialize the modbus register sensor.""" - self._hub = hub - self._name = entry[CONF_NAME] - slave = entry.get(CONF_SLAVE) - self._slave = int(slave) if slave else None - self._register = int(entry[CONF_ADDRESS]) - self._register_type = entry[CONF_INPUT_TYPE] + super().__init__(hub, entry) + self._register = self._address + self._register_type = self._input_type self._unit_of_measurement = entry.get(CONF_UNIT_OF_MEASUREMENT) self._count = int(entry[CONF_COUNT]) self._swap = entry[CONF_SWAP] @@ -218,54 +214,24 @@ class ModbusRegisterSensor(RestoreEntity, SensorEntity): self._precision = entry[CONF_PRECISION] self._structure = structure self._data_type = entry[CONF_DATA_TYPE] - self._device_class = entry.get(CONF_DEVICE_CLASS) - self._value = None - self._available = True - self._scan_interval = timedelta(seconds=entry.get(CONF_SCAN_INTERVAL)) async def async_added_to_hass(self): """Handle entity which will be added.""" + await self.async_base_added_to_hass() state = await self.async_get_last_state() if state: self._value = state.state - async_track_time_interval(self.hass, self.async_update, self._scan_interval) - @property def state(self): """Return the state of the sensor.""" return self._value - @property - def name(self): - """Return the name of the sensor.""" - return self._name - - @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 False - @property def unit_of_measurement(self): """Return the unit of measurement.""" return self._unit_of_measurement - @property - def device_class(self) -> str | None: - """Return the device class of the sensor.""" - return self._device_class - - @property - def available(self) -> bool: - """Return True if entity is available.""" - return self._available - def _swap_registers(self, registers): """Do swap as needed.""" if self._swap in [CONF_SWAP_BYTE, CONF_SWAP_WORD_BYTE]: diff --git a/homeassistant/components/modbus/switch.py b/homeassistant/components/modbus/switch.py index 1b0cd37eb87..ef068d7bd18 100644 --- a/homeassistant/components/modbus/switch.py +++ b/homeassistant/components/modbus/switch.py @@ -1,7 +1,6 @@ """Support for Modbus switches.""" from __future__ import annotations -from datetime import timedelta import logging from homeassistant.components.switch import SwitchEntity @@ -10,16 +9,14 @@ from homeassistant.const import ( CONF_COMMAND_OFF, CONF_COMMAND_ON, CONF_NAME, - CONF_SCAN_INTERVAL, - CONF_SLAVE, CONF_SWITCHES, STATE_ON, ) from homeassistant.core import HomeAssistant -from homeassistant.helpers.event import async_track_time_interval from homeassistant.helpers.restore_state import RestoreEntity from homeassistant.helpers.typing import ConfigType +from .base_platform import BasePlatform from .const import ( CALL_TYPE_COIL, CALL_TYPE_WRITE_COIL, @@ -49,18 +46,14 @@ async def async_setup_platform( async_add_entities(switches) -class ModbusSwitch(SwitchEntity, RestoreEntity): +class ModbusSwitch(BasePlatform, SwitchEntity, RestoreEntity): """Base class representing a Modbus switch.""" def __init__(self, hub: ModbusHub, config: dict) -> None: """Initialize the switch.""" - self._hub: ModbusHub = hub - self._name = config[CONF_NAME] - self._slave = config.get(CONF_SLAVE) + config[CONF_INPUT_TYPE] = "" + super().__init__(hub, config) 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_type = CALL_TYPE_WRITE_COIL else: @@ -84,32 +77,16 @@ class ModbusSwitch(SwitchEntity, RestoreEntity): async def async_added_to_hass(self): """Handle entity which will be added.""" + await self.async_base_added_to_hass() state = await self.async_get_last_state() if state: self._is_on = state.state == STATE_ON - async_track_time_interval(self.hass, self.async_update, self._scan_interval) - @property def is_on(self): """Return true if switch is on.""" return self._is_on - @property - def name(self): - """Return the name of the switch.""" - return self._name - - @property - def should_poll(self): - """Return True if entity has to be polled for state.""" - return False - - @property - def available(self) -> bool: - """Return True if entity is available.""" - return self._available - async def async_turn_on(self, **kwargs): """Set switch on."""