sensor.py

Go to the documentation of this file.

"""Support for reading data from a serial port."""
 
from __future__ import annotations
 
import asyncio
import json
import logging
 
from serial import SerialException
import serial_asyncio_fast as serial_asyncio
import voluptuous as vol
 
from homeassistant.components.sensor import (
    PLATFORM_SCHEMA as SENSOR_PLATFORM_SCHEMA,
    SensorEntity,
)
from homeassistant.const import CONF_NAME, CONF_VALUE_TEMPLATE, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HomeAssistant, callback
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
 
_LOGGER = logging.getLogger(__name__)
 
CONF_SERIAL_PORT = "serial_port"
CONF_BAUDRATE = "baudrate"
CONF_BYTESIZE = "bytesize"
CONF_PARITY = "parity"
CONF_STOPBITS = "stopbits"
CONF_XONXOFF = "xonxoff"
CONF_RTSCTS = "rtscts"
CONF_DSRDTR = "dsrdtr"
 
DEFAULT_NAME = "Serial Sensor"
DEFAULT_BAUDRATE = 9600
DEFAULT_BYTESIZE = serial_asyncio.serial.EIGHTBITS
DEFAULT_PARITY = serial_asyncio.serial.PARITY_NONE
DEFAULT_STOPBITS = serial_asyncio.serial.STOPBITS_ONE
DEFAULT_XONXOFF = False
DEFAULT_RTSCTS = False
DEFAULT_DSRDTR = False
 
PLATFORM_SCHEMA = SENSOR_PLATFORM_SCHEMA.extend(
    {
        vol.Required(CONF_SERIAL_PORT): cv.string,
        vol.Optional(CONF_BAUDRATE, default=DEFAULT_BAUDRATE): cv.positive_int,
        vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
        vol.Optional(CONF_VALUE_TEMPLATE): cv.template,
        vol.Optional(CONF_BYTESIZE, default=DEFAULT_BYTESIZE): vol.In(
            [
                serial_asyncio.serial.FIVEBITS,
                serial_asyncio.serial.SIXBITS,
                serial_asyncio.serial.SEVENBITS,
                serial_asyncio.serial.EIGHTBITS,
            ]
        ),
        vol.Optional(CONF_PARITY, default=DEFAULT_PARITY): vol.In(
            [
                serial_asyncio.serial.PARITY_NONE,
                serial_asyncio.serial.PARITY_EVEN,
                serial_asyncio.serial.PARITY_ODD,
                serial_asyncio.serial.PARITY_MARK,
                serial_asyncio.serial.PARITY_SPACE,
            ]
        ),
        vol.Optional(CONF_STOPBITS, default=DEFAULT_STOPBITS): vol.In(
            [
                serial_asyncio.serial.STOPBITS_ONE,
                serial_asyncio.serial.STOPBITS_ONE_POINT_FIVE,
                serial_asyncio.serial.STOPBITS_TWO,
            ]
        ),
        vol.Optional(CONF_XONXOFF, default=DEFAULT_XONXOFF): cv.boolean,
        vol.Optional(CONF_RTSCTS, default=DEFAULT_RTSCTS): cv.boolean,
        vol.Optional(CONF_DSRDTR, default=DEFAULT_DSRDTR): cv.boolean,
    }
)
 
 
async def async_setup_platform(
    hass: HomeAssistant,
    config: ConfigType,
    async_add_entities: AddEntitiesCallback,
    discovery_info: DiscoveryInfoType | None = None,
) -> None:
    """Set up the Serial sensor platform."""
    name = config.get(CONF_NAME)
    port = config.get(CONF_SERIAL_PORT)
    baudrate = config.get(CONF_BAUDRATE)
    bytesize = config.get(CONF_BYTESIZE)
    parity = config.get(CONF_PARITY)
    stopbits = config.get(CONF_STOPBITS)
    xonxoff = config.get(CONF_XONXOFF)
    rtscts = config.get(CONF_RTSCTS)
    dsrdtr = config.get(CONF_DSRDTR)
    value_template = config.get(CONF_VALUE_TEMPLATE)
 
    sensor = SerialSensor(
        name,
        port,
        baudrate,
        bytesize,
        parity,
        stopbits,
        xonxoff,
        rtscts,
        dsrdtr,
        value_template,
    )
 
    hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, sensor.stop_serial_read)
    async_add_entities([sensor], True)
 
 
class SerialSensor(SensorEntity):
    """Representation of a Serial sensor."""
 
    _attr_should_poll = False
 
    def __init__(
        self,
        name,
        port,
        baudrate,
        bytesize,
        parity,
        stopbits,
        xonxoff,
        rtscts,
        dsrdtr,
        value_template,
    ):
        """Initialize the Serial sensor."""
        self._name_name = name
        self._state_state = None
        self._port_port = port
        self._baudrate_baudrate = baudrate
        self._bytesize_bytesize = bytesize
        self._parity_parity = parity
        self._stopbits_stopbits = stopbits
        self._xonxoff_xonxoff = xonxoff
        self._rtscts_rtscts = rtscts
        self._dsrdtr_dsrdtr = dsrdtr
        self._serial_loop_task_serial_loop_task = None
        self._template_template = value_template
        self._attributes_attributes = None
 
    async def async_added_to_hass(self) -> None:
        """Handle when an entity is about to be added to Home Assistant."""
        self._serial_loop_task_serial_loop_task = self.hasshass.loop.create_task(
            self.serial_readserial_read(
                self._port_port,
                self._baudrate_baudrate,
                self._bytesize_bytesize,
                self._parity_parity,
                self._stopbits_stopbits,
                self._xonxoff_xonxoff,
                self._rtscts_rtscts,
                self._dsrdtr_dsrdtr,
            )
        )
 
    async def serial_read(
        self,
        device,
        baudrate,
        bytesize,
        parity,
        stopbits,
        xonxoff,
        rtscts,
        dsrdtr,
        **kwargs,
    ):
        """Read the data from the port."""
        logged_error = False
        while True:
            try:
                reader, _ = await serial_asyncio.open_serial_connection(
                    url=device,
                    baudrate=baudrate,
                    bytesize=bytesize,
                    parity=parity,
                    stopbits=stopbits,
                    xonxoff=xonxoff,
                    rtscts=rtscts,
                    dsrdtr=dsrdtr,
                    **kwargs,
                )
 
            except SerialException:
                if not logged_error:
                    _LOGGER.exception(
                        "Unable to connect to the serial device %s. Will retry", device
                    )
                    logged_error = True
                await self._handle_error_handle_error()
            else:
                _LOGGER.debug("Serial device %s connected", device)
                while True:
                    try:
                        line = await reader.readline()
                    except SerialException:
                        _LOGGER.exception(
                            "Error while reading serial device %s", device
                        )
                        await self._handle_error_handle_error()
                        break
                    else:
                        line = line.decode("utf-8").strip()
 
                        try:
                            data = json.loads(line)
                        except ValueError:
                            pass
                        else:
                            if isinstance(data, dict):
                                self._attributes_attributes = data
 
                        if self._template_template is not None:
                            line = self._template_template.async_render_with_possible_json_value(
                                line
                            )
 
                        _LOGGER.debug("Received: %s", line)
                        self._state_state = line
                        self.async_write_ha_stateasync_write_ha_state()
 
    async def _handle_error(self):
        """Handle error for serial connection."""
        self._state_state = None
        self._attributes_attributes = None
        self.async_write_ha_stateasync_write_ha_state()
        await asyncio.sleep(5)
 
    @callback
    def stop_serial_read(self, event):
        """Close resources."""
        if self._serial_loop_task_serial_loop_task:
            self._serial_loop_task_serial_loop_task.cancel()
 
    @property
    def name(self):
        """Return the name of the sensor."""
        return self._name_name
 
    @property
    def extra_state_attributes(self):
        """Return the attributes of the entity (if any JSON present)."""
        return self._attributes_attributes
 
    @property
    def native_value(self):
        """Return the state of the sensor."""
        return self._state_state