Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add onewire switches (#42962) 

* Add support for switches

* Fix stale comment
pull/43019/head
epenet 2020-11-08 19:06:41 +01:00 committed by GitHub
parent 2845fca08e
commit 7c397a02b7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 541 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
homeassistant/components/onewire/const.py
View File

@ -1,6 +1,7 @@
"""Constants for 1-Wire component."""
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import (
DEVICE_CLASS_CURRENT,
DEVICE_CLASS_HUMIDITY,
@ -41,6 +42,8 @@ SENSOR_TYPE_SENSED = "sensed"
SENSOR_TYPE_TEMPERATURE = "temperature"
SENSOR_TYPE_VOLTAGE = "voltage"
SENSOR_TYPE_WETNESS = "wetness"
SWITCH_TYPE_LATCH = "latch"
SWITCH_TYPE_PIO = "pio"
SENSOR_TYPES = {
# SensorType: [ Unit, DeviceClass ]
@ -54,9 +57,12 @@ SENSOR_TYPES = {
SENSOR_TYPE_VOLTAGE: [VOLT, DEVICE_CLASS_VOLTAGE],
SENSOR_TYPE_CURRENT: [ELECTRICAL_CURRENT_AMPERE, DEVICE_CLASS_CURRENT],
SENSOR_TYPE_SENSED: [None, None],
SWITCH_TYPE_LATCH: [None, None],
SWITCH_TYPE_PIO: [None, None],
}
SUPPORTED_PLATFORMS = [
BINARY_SENSOR_DOMAIN,
SENSOR_DOMAIN,
SWITCH_DOMAIN,
]

18
homeassistant/components/onewire/onewire_entities.py
View File

@ -7,7 +7,13 @@ from pyownet import protocol
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.typing import StateType
from .const import SENSOR_TYPE_COUNT, SENSOR_TYPE_SENSED, SENSOR_TYPES
from .const import (
SENSOR_TYPE_COUNT,
SENSOR_TYPE_SENSED,
SENSOR_TYPES,
SWITCH_TYPE_LATCH,
SWITCH_TYPE_PIO,
)
_LOGGER = logging.getLogger(__name__)
@ -99,6 +105,10 @@ class OneWireProxy(OneWire):
"""Read a value from the owserver."""
return self._owproxy.read(self._device_file).decode().lstrip()
def _write_value_ownet(self, value: bytes):
"""Write a value to the owserver."""
return self._owproxy.write(self._device_file, value)
def update(self):
"""Get the latest data from the device."""
value = None
@ -109,7 +119,11 @@ class OneWireProxy(OneWire):
else:
if self._sensor_type == SENSOR_TYPE_COUNT:
value = int(self._value_raw)
elif self._sensor_type == SENSOR_TYPE_SENSED:
elif self._sensor_type in [
SENSOR_TYPE_SENSED,
SWITCH_TYPE_LATCH,
SWITCH_TYPE_PIO,
]:
value = int(self._value_raw) == 1
else:
value = round(self._value_raw, 1)

204
homeassistant/components/onewire/switch.py Normal file
View File

@ -0,0 +1,204 @@
"""Support for 1-Wire environment switches."""
import logging
import os
from homeassistant.components.switch import SwitchEntity
from homeassistant.const import CONF_TYPE
from .const import CONF_TYPE_OWSERVER, DOMAIN, SWITCH_TYPE_LATCH, SWITCH_TYPE_PIO
from .onewire_entities import OneWireProxy
from .onewirehub import OneWireHub
DEVICE_SWITCHES = {
# Family : { owfs path }
"12": [
{
"path": "PIO.A",
"name": "PIO A",
"type": SWITCH_TYPE_PIO,
"default_disabled": True,
},
{
"path": "PIO.B",
"name": "PIO B",
"type": SWITCH_TYPE_PIO,
"default_disabled": True,
},
{
"path": "latch.A",
"name": "Latch A",
"type": SWITCH_TYPE_LATCH,
"default_disabled": True,
},
{
"path": "latch.B",
"name": "Latch B",
"type": SWITCH_TYPE_LATCH,
"default_disabled": True,
},
],
"29": [
{
"path": "PIO.0",
"name": "PIO 0",
"type": SWITCH_TYPE_PIO,
"default_disabled": True,
},
{
"path": "PIO.1",
"name": "PIO 1",
"type": SWITCH_TYPE_PIO,
"default_disabled": True,
},
{
"path": "PIO.2",
"name": "PIO 2",
"type": SWITCH_TYPE_PIO,
"default_disabled": True,
},
{
"path": "PIO.3",
"name": "PIO 3",
"type": SWITCH_TYPE_PIO,
"default_disabled": True,
},
{
"path": "PIO.4",
"name": "PIO 4",
"type": SWITCH_TYPE_PIO,
"default_disabled": True,
},
{
"path": "PIO.5",
"name": "PIO 5",
"type": SWITCH_TYPE_PIO,
"default_disabled": True,
},
{
"path": "PIO.6",
"name": "PIO 6",
"type": SWITCH_TYPE_PIO,
"default_disabled": True,
},
{
"path": "PIO.7",
"name": "PIO 7",
"type": SWITCH_TYPE_PIO,
"default_disabled": True,
},
{
"path": "latch.0",
"name": "Latch 0",
"type": SWITCH_TYPE_LATCH,
"default_disabled": True,
},
{
"path": "latch.1",
"name": "Latch 1",
"type": SWITCH_TYPE_LATCH,
"default_disabled": True,
},
{
"path": "latch.2",
"name": "Latch 2",
"type": SWITCH_TYPE_LATCH,
"default_disabled": True,
},
{
"path": "latch.3",
"name": "Latch 3",
"type": SWITCH_TYPE_LATCH,
"default_disabled": True,
},
{
"path": "latch.4",
"name": "Latch 4",
"type": SWITCH_TYPE_LATCH,
"default_disabled": True,
},
{
"path": "latch.5",
"name": "Latch 5",
"type": SWITCH_TYPE_LATCH,
"default_disabled": True,
},
{
"path": "latch.6",
"name": "Latch 6",
"type": SWITCH_TYPE_LATCH,
"default_disabled": True,
},
{
"path": "latch.7",
"name": "Latch 7",
"type": SWITCH_TYPE_LATCH,
"default_disabled": True,
},
],
}
LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up 1-Wire platform."""
# Only OWServer implementation works with switches
if config_entry.data[CONF_TYPE] == CONF_TYPE_OWSERVER:
onewirehub = hass.data[DOMAIN][config_entry.unique_id]
entities = await hass.async_add_executor_job(get_entities, onewirehub)
async_add_entities(entities, True)
def get_entities(onewirehub: OneWireHub):
"""Get a list of entities."""
entities = []
for device in onewirehub.devices:
family = device["family"]
device_type = device["type"]
sensor_id = os.path.split(os.path.split(device["path"])[0])[1]
if family not in DEVICE_SWITCHES:
continue
device_info = {
"identifiers": {(DOMAIN, sensor_id)},
"manufacturer": "Maxim Integrated",
"model": device_type,
"name": sensor_id,
}
for device_switch in DEVICE_SWITCHES[family]:
device_file = os.path.join(
os.path.split(device["path"])[0], device_switch["path"]
)
entities.append(
OneWireSwitch(
sensor_id,
device_file,
device_switch["type"],
device_switch["name"],
device_info,
device_switch.get("default_disabled", False),
onewirehub.owproxy,
)
)
return entities
class OneWireSwitch(SwitchEntity, OneWireProxy):
"""Implementation of a 1-Wire switch."""
@property
def is_on(self):
"""Return true if sensor is on."""
return self._state
def turn_on(self, **kwargs) -> None:
"""Turn the entity on."""
self._write_value_ownet(b"1")
def turn_off(self, **kwargs) -> None:
"""Turn the entity off."""
self._write_value_ownet(b"0")

187
tests/components/onewire/test_entity_owserver.py
View File

@ -9,6 +9,7 @@ from homeassistant.components.onewire.const import (
SUPPORTED_PLATFORMS,
)
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import (
DEVICE_CLASS_CURRENT,
DEVICE_CLASS_HUMIDITY,
@ -110,6 +111,44 @@ MOCK_DEVICE_SENSORS = {
"disabled": True,
},
],
SWITCH_DOMAIN: [
{
"entity_id": "switch.12_111111111111_pio_a",
"unique_id": "/12.111111111111/PIO.A",
"injected_value": b" 1",
"result": STATE_ON,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.12_111111111111_pio_b",
"unique_id": "/12.111111111111/PIO.B",
"injected_value": b" 0",
"result": STATE_OFF,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.12_111111111111_latch_a",
"unique_id": "/12.111111111111/latch.A",
"injected_value": b" 1",
"result": STATE_ON,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.12_111111111111_latch_b",
"unique_id": "/12.111111111111/latch.B",
"injected_value": b" 0",
"result": STATE_OFF,
"unit": None,
"class": None,
"disabled": True,
},
],
},
"1D.111111111111": {
"inject_reads": [
@ -377,6 +416,152 @@ MOCK_DEVICE_SENSORS = {
"disabled": True,
},
],
SWITCH_DOMAIN: [
{
"entity_id": "switch.29_111111111111_pio_0",
"unique_id": "/29.111111111111/PIO.0",
"injected_value": b" 1",
"result": STATE_ON,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.29_111111111111_pio_1",
"unique_id": "/29.111111111111/PIO.1",
"injected_value": b" 0",
"result": STATE_OFF,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.29_111111111111_pio_2",
"unique_id": "/29.111111111111/PIO.2",
"injected_value": b" 1",
"result": STATE_ON,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.29_111111111111_pio_3",
"unique_id": "/29.111111111111/PIO.3",
"injected_value": b" 0",
"result": STATE_OFF,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.29_111111111111_pio_4",
"unique_id": "/29.111111111111/PIO.4",
"injected_value": b" 1",
"result": STATE_ON,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.29_111111111111_pio_5",
"unique_id": "/29.111111111111/PIO.5",
"injected_value": b" 0",
"result": STATE_OFF,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.29_111111111111_pio_6",
"unique_id": "/29.111111111111/PIO.6",
"injected_value": b" 1",
"result": STATE_ON,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.29_111111111111_pio_7",
"unique_id": "/29.111111111111/PIO.7",
"injected_value": b" 0",
"result": STATE_OFF,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.29_111111111111_latch_0",
"unique_id": "/29.111111111111/latch.0",
"injected_value": b" 1",
"result": STATE_ON,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.29_111111111111_latch_1",
"unique_id": "/29.111111111111/latch.1",
"injected_value": b" 0",
"result": STATE_OFF,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.29_111111111111_latch_2",
"unique_id": "/29.111111111111/latch.2",
"injected_value": b" 1",
"result": STATE_ON,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.29_111111111111_latch_3",
"unique_id": "/29.111111111111/latch.3",
"injected_value": b" 0",
"result": STATE_OFF,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.29_111111111111_latch_4",
"unique_id": "/29.111111111111/latch.4",
"injected_value": b" 1",
"result": STATE_ON,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.29_111111111111_latch_5",
"unique_id": "/29.111111111111/latch.5",
"injected_value": b" 0",
"result": STATE_OFF,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.29_111111111111_latch_6",
"unique_id": "/29.111111111111/latch.6",
"injected_value": b" 1",
"result": STATE_ON,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.29_111111111111_latch_7",
"unique_id": "/29.111111111111/latch.7",
"injected_value": b" 0",
"result": STATE_OFF,
"unit": None,
"class": None,
"disabled": True,
},
],
},
"3B.111111111111": {
"inject_reads": [
@ -534,7 +719,7 @@ async def test_owserver_setup_valid_device(owproxy, hass, device_id, platform):
read_side_effect.append(expected_sensor["injected_value"])
# Ensure enough read side effect
read_side_effect.extend([ProtocolError("Missing injected value")] * 10)
read_side_effect.extend([ProtocolError("Missing injected value")] * 20)
owproxy.return_value.dir.return_value = dir_return_value
owproxy.return_value.read.side_effect = read_side_effect

129
tests/components/onewire/test_switch.py Normal file
View File

@ -0,0 +1,129 @@
"""Tests for 1-Wire devices connected on OWServer."""
import copy
from pyownet.protocol import Error as ProtocolError
import pytest
from homeassistant.components.onewire.switch import DEVICE_SWITCHES
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TOGGLE, STATE_OFF, STATE_ON
from homeassistant.setup import async_setup_component
from . import setup_onewire_patched_owserver_integration
from tests.async_mock import patch
from tests.common import mock_registry
MOCK_DEVICE_SENSORS = {
"12.111111111111": {
"inject_reads": [
b"DS2406", # read device type
],
SWITCH_DOMAIN: [
{
"entity_id": "switch.12_111111111111_pio_a",
"unique_id": "/12.111111111111/PIO.A",
"injected_value": b" 1",
"result": STATE_ON,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.12_111111111111_pio_b",
"unique_id": "/12.111111111111/PIO.B",
"injected_value": b" 0",
"result": STATE_OFF,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.12_111111111111_latch_a",
"unique_id": "/12.111111111111/latch.A",
"injected_value": b" 1",
"result": STATE_ON,
"unit": None,
"class": None,
"disabled": True,
},
{
"entity_id": "switch.12_111111111111_latch_b",
"unique_id": "/12.111111111111/latch.B",
"injected_value": b" 0",
"result": STATE_OFF,
"unit": None,
"class": None,
"disabled": True,
},
],
}
}
@pytest.mark.parametrize("device_id", ["12.111111111111"])
@patch("homeassistant.components.onewire.onewirehub.protocol.proxy")
async def test_owserver_switch(owproxy, hass, device_id):
"""Test for 1-Wire switch.
This test forces all entities to be enabled.
"""
await async_setup_component(hass, "persistent_notification", {})
entity_registry = mock_registry(hass)
mock_device_sensor = MOCK_DEVICE_SENSORS[device_id]
device_family = device_id[0:2]
dir_return_value = [f"/{device_id}/"]
read_side_effect = [device_family.encode()]
if "inject_reads" in mock_device_sensor:
read_side_effect += mock_device_sensor["inject_reads"]
expected_sensors = mock_device_sensor[SWITCH_DOMAIN]
for expected_sensor in expected_sensors:
read_side_effect.append(expected_sensor["injected_value"])
# Ensure enough read side effect
read_side_effect.extend([ProtocolError("Missing injected value")] * 10)
owproxy.return_value.dir.return_value = dir_return_value
owproxy.return_value.read.side_effect = read_side_effect
# Force enable switches
patch_device_switches = copy.deepcopy(DEVICE_SWITCHES)
for item in patch_device_switches[device_family]:
item["default_disabled"] = False
with patch(
"homeassistant.components.onewire.SUPPORTED_PLATFORMS", [SWITCH_DOMAIN]
), patch.dict(
"homeassistant.components.onewire.switch.DEVICE_SWITCHES", patch_device_switches
):
await setup_onewire_patched_owserver_integration(hass)
await hass.async_block_till_done()
assert len(entity_registry.entities) == len(expected_sensors)
for expected_sensor in expected_sensors:
entity_id = expected_sensor["entity_id"]
registry_entry = entity_registry.entities.get(entity_id)
assert registry_entry is not None
state = hass.states.get(entity_id)
assert state.state == expected_sensor["result"]
if state.state == STATE_ON:
owproxy.return_value.read.side_effect = [b" 0"]
expected_sensor["result"] = STATE_OFF
elif state.state == STATE_OFF:
owproxy.return_value.read.side_effect = [b" 1"]
expected_sensor["result"] = STATE_ON
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TOGGLE,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == expected_sensor["result"]