15b1a9ecea
* Add support for Numato 32 port USB GPIO boards Included are a binary_sensor, sensor and switch component implementations. The binary_sensor interface pushes updates via registered callback functions, so no need to poll here. Unit tests are included to test against a Numato device mockup. * Refactor numato configuration due to PR finding * Resolve minor review findings * Bump numato-gpio requirement * Load numato platforms during domain setup According to review finding * Guard from platform setup without discovery_info According to review finding * Move numato API state into hass.data According to review finding. * Avoid side effects in numato entity constructors According to review finding * Keep only first line of numato module docstrings Removed reference to the documentation. Requested by reviewer. * Minor improvements inspired by review findings * Fix async tests Pytest fixture was returning from the yield too early executing teardown code during test execution. * Improve test coverage * Configure GPIO ports early Review finding * Move read_gpio callback to outside the loop Also continue on failed switch setup, resolve other minor review findings and correct some error messages * Bump numato-gpio requirement This fixes a crash during cleanup. When any device had a communication problem, its cleanup would raise an exception which was not handled, fell through to the caller and prevented the remaining devices from being cleaned up. * Call services directly Define local helper functions for better readability. Resolves a review finding. * Assert something in every test So not only coverage is satisfied but things are actually tested to be in the expected state. Resolves a review finding. * Clarify scope of notification tests Make unit test for hass NumatoAPI independent of Home Assistant (very basic test of notifications). Improve the regular operations test for notifications. * Test for hass.states after operating switches Resolves a review finding. * Check for wrong port directions * WIP: Split numato tests to multiple files test_hass_binary_sensor_notification still fails. * Remove pytest asyncio decorator Apears to be redundant. Resolves a review finding. * Call switch services directly. Resolves a review finding. * Remove obsolete inline pylint config Co-Authored-By: Martin Hjelmare <marhje52@gmail.com> * Improve the numato_gpio module mockup Resolves a review finding. * Remove needless explicit conversions to str Resolves review findings. * Test setup of binary_sensor callbacks * Fix test_hass_binary_sensor_notification * Add forgotten await Review finding. Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
123 lines
3.7 KiB
Python
123 lines
3.7 KiB
Python
"""Sensor platform integration for ADC ports of Numato USB GPIO expanders."""
|
|
import logging
|
|
|
|
from numato_gpio import NumatoGpioError
|
|
|
|
from homeassistant.const import CONF_ID, CONF_NAME, CONF_SENSORS
|
|
from homeassistant.helpers.entity import Entity
|
|
|
|
from . import (
|
|
CONF_DEVICES,
|
|
CONF_DST_RANGE,
|
|
CONF_DST_UNIT,
|
|
CONF_PORTS,
|
|
CONF_SRC_RANGE,
|
|
DATA_API,
|
|
DOMAIN,
|
|
)
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
ICON = "mdi:gauge"
|
|
|
|
|
|
def setup_platform(hass, config, add_entities, discovery_info=None):
|
|
"""Set up the configured Numato USB GPIO ADC sensor ports."""
|
|
if discovery_info is None:
|
|
return
|
|
|
|
api = hass.data[DOMAIN][DATA_API]
|
|
sensors = []
|
|
devices = hass.data[DOMAIN][CONF_DEVICES]
|
|
for device in [d for d in devices if CONF_SENSORS in d]:
|
|
device_id = device[CONF_ID]
|
|
ports = device[CONF_SENSORS][CONF_PORTS]
|
|
for port, adc_def in ports.items():
|
|
try:
|
|
api.setup_input(device_id, port)
|
|
except NumatoGpioError as err:
|
|
_LOGGER.error(
|
|
"Failed to initialize sensor '%s' on Numato device %s port %s: %s",
|
|
adc_def[CONF_NAME],
|
|
device_id,
|
|
port,
|
|
err,
|
|
)
|
|
continue
|
|
sensors.append(
|
|
NumatoGpioAdc(
|
|
adc_def[CONF_NAME],
|
|
device_id,
|
|
port,
|
|
adc_def[CONF_SRC_RANGE],
|
|
adc_def[CONF_DST_RANGE],
|
|
adc_def[CONF_DST_UNIT],
|
|
api,
|
|
)
|
|
)
|
|
add_entities(sensors, True)
|
|
|
|
|
|
class NumatoGpioAdc(Entity):
|
|
"""Represents an ADC port of a Numato USB GPIO expander."""
|
|
|
|
def __init__(self, name, device_id, port, src_range, dst_range, dst_unit, api):
|
|
"""Initialize the sensor."""
|
|
self._name = name
|
|
self._device_id = device_id
|
|
self._port = port
|
|
self._src_range = src_range
|
|
self._dst_range = dst_range
|
|
self._state = None
|
|
self._unit_of_measurement = dst_unit
|
|
self._api = api
|
|
|
|
@property
|
|
def name(self):
|
|
"""Return the name of the sensor."""
|
|
return self._name
|
|
|
|
@property
|
|
def state(self):
|
|
"""Return the state of the sensor."""
|
|
return self._state
|
|
|
|
@property
|
|
def unit_of_measurement(self):
|
|
"""Return the unit the value is expressed in."""
|
|
return self._unit_of_measurement
|
|
|
|
@property
|
|
def icon(self):
|
|
"""Return the icon to use in the frontend, if any."""
|
|
return ICON
|
|
|
|
def update(self):
|
|
"""Get the latest data and updates the state."""
|
|
try:
|
|
adc_val = self._api.read_adc_input(self._device_id, self._port)
|
|
adc_val = self._clamp_to_source_range(adc_val)
|
|
self._state = self._linear_scale_to_dest_range(adc_val)
|
|
except NumatoGpioError as err:
|
|
self._state = None
|
|
_LOGGER.error(
|
|
"Failed to update Numato device %s ADC-port %s: %s",
|
|
self._device_id,
|
|
self._port,
|
|
err,
|
|
)
|
|
|
|
def _clamp_to_source_range(self, val):
|
|
# clamp to source range
|
|
val = max(val, self._src_range[0])
|
|
val = min(val, self._src_range[1])
|
|
return val
|
|
|
|
def _linear_scale_to_dest_range(self, val):
|
|
# linear scale to dest range
|
|
src_len = self._src_range[1] - self._src_range[0]
|
|
adc_val_rel = val - self._src_range[0]
|
|
ratio = float(adc_val_rel) / float(src_len)
|
|
dst_len = self._dst_range[1] - self._dst_range[0]
|
|
dest_val = self._dst_range[0] + ratio * dst_len
|
|
return dest_val
|