hass-core/homeassistant/components/modbus/climate.py

"""Support for Generic Modbus Thermostats."""
import logging
import struct
from typing import Optional

from pymodbus.exceptions import ModbusException
from pymodbus.pdu import ExceptionResponse
import voluptuous as vol

from homeassistant.components.climate import PLATFORM_SCHEMA, ClimateDevice
from homeassistant.components.climate.const import (
    HVAC_MODE_AUTO,
    SUPPORT_TARGET_TEMPERATURE,
)
from homeassistant.const import (
    ATTR_TEMPERATURE,
    CONF_NAME,
    CONF_SLAVE,
    TEMP_CELSIUS,
    TEMP_FAHRENHEIT,
)
import homeassistant.helpers.config_validation as cv

from .const import (
    CALL_TYPE_REGISTER_HOLDING,
    CALL_TYPE_REGISTER_INPUT,
    CONF_CURRENT_TEMP,
    CONF_CURRENT_TEMP_REGISTER_TYPE,
    CONF_DATA_COUNT,
    CONF_DATA_TYPE,
    CONF_HUB,
    CONF_MAX_TEMP,
    CONF_MIN_TEMP,
    CONF_OFFSET,
    CONF_PRECISION,
    CONF_SCALE,
    CONF_STEP,
    CONF_TARGET_TEMP,
    CONF_UNIT,
    DATA_TYPE_FLOAT,
    DATA_TYPE_INT,
    DATA_TYPE_UINT,
    DEFAULT_HUB,
    MODBUS_DOMAIN,
)

_LOGGER = logging.getLogger(__name__)


PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
    {
        vol.Required(CONF_CURRENT_TEMP): cv.positive_int,
        vol.Required(CONF_NAME): cv.string,
        vol.Required(CONF_SLAVE): cv.positive_int,
        vol.Required(CONF_TARGET_TEMP): cv.positive_int,
        vol.Optional(CONF_DATA_COUNT, default=2): cv.positive_int,
        vol.Optional(
            CONF_CURRENT_TEMP_REGISTER_TYPE, default=CALL_TYPE_REGISTER_HOLDING
        ): vol.In([CALL_TYPE_REGISTER_HOLDING, CALL_TYPE_REGISTER_INPUT]),
        vol.Optional(CONF_DATA_TYPE, default=DATA_TYPE_FLOAT): vol.In(
            [DATA_TYPE_INT, DATA_TYPE_UINT, DATA_TYPE_FLOAT]
        ),
        vol.Optional(CONF_HUB, default=DEFAULT_HUB): cv.string,
        vol.Optional(CONF_PRECISION, default=1): cv.positive_int,
        vol.Optional(CONF_SCALE, default=1): vol.Coerce(float),
        vol.Optional(CONF_OFFSET, default=0): vol.Coerce(float),
        vol.Optional(CONF_MAX_TEMP, default=35): cv.positive_int,
        vol.Optional(CONF_MIN_TEMP, default=5): cv.positive_int,
        vol.Optional(CONF_STEP, default=0.5): vol.Coerce(float),
        vol.Optional(CONF_UNIT, default="C"): cv.string,
    }
)


async def async_setup_platform(hass, config, add_entities, discovery_info=None):
    """Set up the Modbus Thermostat Platform."""
    name = config[CONF_NAME]
    modbus_slave = config[CONF_SLAVE]
    target_temp_register = config[CONF_TARGET_TEMP]
    current_temp_register = config[CONF_CURRENT_TEMP]
    current_temp_register_type = config[CONF_CURRENT_TEMP_REGISTER_TYPE]
    data_type = config[CONF_DATA_TYPE]
    count = config[CONF_DATA_COUNT]
    precision = config[CONF_PRECISION]
    scale = config[CONF_SCALE]
    offset = config[CONF_OFFSET]
    unit = config[CONF_UNIT]
    max_temp = config[CONF_MAX_TEMP]
    min_temp = config[CONF_MIN_TEMP]
    temp_step = config[CONF_STEP]
    hub_name = config[CONF_HUB]
    hub = hass.data[MODBUS_DOMAIN][hub_name]

    add_entities(
        [
            ModbusThermostat(
                hub,
                name,
                modbus_slave,
                target_temp_register,
                current_temp_register,
                current_temp_register_type,
                data_type,
                count,
                precision,
                scale,
                offset,
                unit,
                max_temp,
                min_temp,
                temp_step,
            )
        ],
        True,
    )


class ModbusThermostat(ClimateDevice):
    """Representation of a Modbus Thermostat."""

    def __init__(
        self,
        hub,
        name,
        modbus_slave,
        target_temp_register,
        current_temp_register,
        current_temp_register_type,
        data_type,
        count,
        precision,
        scale,
        offset,
        unit,
        max_temp,
        min_temp,
        temp_step,
    ):
        """Initialize the unit."""
        self._hub = hub
        self._name = name
        self._slave = modbus_slave
        self._target_temperature_register = target_temp_register
        self._current_temperature_register = current_temp_register
        self._current_temperature_register_type = current_temp_register_type
        self._target_temperature = None
        self._current_temperature = None
        self._data_type = data_type
        self._count = int(count)
        self._precision = precision
        self._scale = scale
        self._offset = offset
        self._unit = unit
        self._max_temp = max_temp
        self._min_temp = min_temp
        self._temp_step = temp_step
        self._structure = ">f"
        self._available = True

        data_types = {
            DATA_TYPE_INT: {1: "h", 2: "i", 4: "q"},
            DATA_TYPE_UINT: {1: "H", 2: "I", 4: "Q"},
            DATA_TYPE_FLOAT: {1: "e", 2: "f", 4: "d"},
        }

        self._structure = f">{data_types[self._data_type][self._count]}"

    @property
    def supported_features(self):
        """Return the list of supported features."""
        return SUPPORT_TARGET_TEMPERATURE

    async def async_update(self):
        """Update Target & Current Temperature."""
        self._target_temperature = await self._read_register(
            CALL_TYPE_REGISTER_HOLDING, self._target_temperature_register
        )
        self._current_temperature = await self._read_register(
            self._current_temperature_register_type, self._current_temperature_register
        )

    @property
    def hvac_mode(self):
        """Return the current HVAC mode."""
        return HVAC_MODE_AUTO

    @property
    def hvac_modes(self):
        """Return the possible HVAC modes."""
        return [HVAC_MODE_AUTO]

    @property
    def name(self):
        """Return the name of the climate device."""
        return self._name

    @property
    def current_temperature(self):
        """Return the current temperature."""
        return self._current_temperature

    @property
    def target_temperature(self):
        """Return the target temperature."""
        return self._target_temperature

    @property
    def temperature_unit(self):
        """Return the unit of measurement."""
        return TEMP_FAHRENHEIT if self._unit == "F" else TEMP_CELSIUS

    @property
    def min_temp(self):
        """Return the minimum temperature."""
        return self._min_temp

    @property
    def max_temp(self):
        """Return the maximum temperature."""
        return self._max_temp

    @property
    def target_temperature_step(self):
        """Return the supported step of target temperature."""
        return self._temp_step

    async def set_temperature(self, **kwargs):
        """Set new target temperature."""
        target_temperature = int(
            (kwargs.get(ATTR_TEMPERATURE) - self._offset) / self._scale
        )
        if target_temperature is None:
            return
        byte_string = struct.pack(self._structure, target_temperature)
        register_value = struct.unpack(">h", byte_string[0:2])[0]
        await self._write_register(self._target_temperature_register, register_value)

    @property
    def available(self) -> bool:
        """Return True if entity is available."""
        return self._available

    async def _read_register(self, register_type, register) -> Optional[float]:
        """Read register using the Modbus hub slave."""
        if register_type == CALL_TYPE_REGISTER_INPUT:
            result = await self._hub.read_input_registers(
                self._slave, register, self._count
            )
        else:
            result = await self._hub.read_holding_registers(
                self._slave, register, self._count
            )
        if result is None:
            self._available = False
            return
        if isinstance(result, (ModbusException, ExceptionResponse)):
            self._available = False
            return

        byte_string = b"".join(
            [x.to_bytes(2, byteorder="big") for x in result.registers]
        )
        val = struct.unpack(self._structure, byte_string)[0]
        register_value = format(
            (self._scale * val) + self._offset, f".{self._precision}f"
        )
        register_value = float(register_value)
        self._available = True

        return register_value

    async def _write_register(self, register, value):
        """Write holding register using the Modbus hub slave."""
        await self._hub.write_registers(self._slave, register, [value, 0])
        self._available = True