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

Add mysensors remote platform (#86376)

pull/86404/head
Martin Hjelmare 2023-01-22 21:04:42 +01:00 committed by GitHub
parent 5102d1a5f3
commit 9c76cd1b6a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 387 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
homeassistant/components/mysensors/const.py
View File

@ -91,6 +91,8 @@ LIGHT_TYPES: dict[SensorType, set[ValueType]] = {
NOTIFY_TYPES: dict[SensorType, set[ValueType]] = {"S_INFO": {"V_TEXT"}}
REMOTE_TYPES: dict[SensorType, set[ValueType]] = {"S_IR": {"V_IR_SEND"}}
SENSOR_TYPES: dict[SensorType, set[ValueType]] = {
"S_SOUND": {"V_LEVEL"},
"S_VIBRATION": {"V_LEVEL"},
@ -107,7 +109,7 @@ SENSOR_TYPES: dict[SensorType, set[ValueType]] = {
"S_POWER": {"V_WATT", "V_KWH", "V_VAR", "V_VA", "V_POWER_FACTOR"},
"S_DISTANCE": {"V_DISTANCE"},
"S_LIGHT_LEVEL": {"V_LIGHT_LEVEL", "V_LEVEL"},
"S_IR": {"V_IR_RECEIVE"},
"S_IR": {"V_IR_RECEIVE", "V_IR_RECORD"},
"S_WATER": {"V_FLOW", "V_VOLUME"},
"S_CUSTOM": {"V_VAR1", "V_VAR2", "V_VAR3", "V_VAR4", "V_VAR5", "V_CUSTOM"},
"S_SCENE_CONTROLLER": {"V_SCENE_ON", "V_SCENE_OFF"},
@ -144,6 +146,7 @@ PLATFORM_TYPES: dict[Platform, dict[SensorType, set[ValueType]]] = {
Platform.DEVICE_TRACKER: DEVICE_TRACKER_TYPES,
Platform.LIGHT: LIGHT_TYPES,
Platform.NOTIFY: NOTIFY_TYPES,
Platform.REMOTE: REMOTE_TYPES,
Platform.SENSOR: SENSOR_TYPES,
Platform.SWITCH: SWITCH_TYPES,
Platform.TEXT: TEXT_TYPES,

124
homeassistant/components/mysensors/remote.py Normal file
View File

@ -0,0 +1,124 @@
"""Support MySensors IR transceivers."""
from __future__ import annotations
from collections.abc import Iterable
from typing import Any, Optional, cast
from homeassistant.components.remote import (
ATTR_COMMAND,
RemoteEntity,
RemoteEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import setup_mysensors_platform
from .const import MYSENSORS_DISCOVERY, DiscoveryInfo
from .device import MySensorsEntity
from .helpers import on_unload
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up this platform for a specific ConfigEntry(==Gateway)."""
@callback
def async_discover(discovery_info: DiscoveryInfo) -> None:
"""Discover and add a MySensors remote."""
setup_mysensors_platform(
hass,
Platform.REMOTE,
discovery_info,
MySensorsRemote,
async_add_entities=async_add_entities,
)
on_unload(
hass,
config_entry.entry_id,
async_dispatcher_connect(
hass,
MYSENSORS_DISCOVERY.format(config_entry.entry_id, Platform.REMOTE),
async_discover,
),
)
class MySensorsRemote(MySensorsEntity, RemoteEntity):
"""Representation of a MySensors IR transceiver."""
_current_command: str | None = None
@property
def is_on(self) -> bool | None:
"""Return True if remote is on."""
set_req = self.gateway.const.SetReq
value = cast(Optional[str], self._child.values.get(set_req.V_LIGHT))
if value is None:
return None
return value == "1"
@property
def supported_features(self) -> RemoteEntityFeature:
"""Flag supported features."""
features = RemoteEntityFeature(0)
set_req = self.gateway.const.SetReq
if set_req.V_IR_RECORD in self._values:
features = features | RemoteEntityFeature.LEARN_COMMAND
return features
async def async_send_command(self, command: Iterable[str], **kwargs: Any) -> None:
"""Send commands to a device."""
for cmd in command:
self._current_command = cmd
self.gateway.set_child_value(
self.node_id, self.child_id, self.value_type, cmd, ack=1
)
async def async_learn_command(self, **kwargs: Any) -> None:
"""Learn a command from a device."""
set_req = self.gateway.const.SetReq
commands: list[str] | None = kwargs.get(ATTR_COMMAND)
if commands is None:
raise ValueError("Command not specified for learn_command service")
for command in commands:
self.gateway.set_child_value(
self.node_id, self.child_id, set_req.V_IR_RECORD, command, ack=1
)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the IR transceiver on."""
set_req = self.gateway.const.SetReq
if self._current_command:
self.gateway.set_child_value(
self.node_id,
self.child_id,
self.value_type,
self._current_command,
ack=1,
)
self.gateway.set_child_value(
self.node_id, self.child_id, set_req.V_LIGHT, 1, ack=1
)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the IR transceiver off."""
set_req = self.gateway.const.SetReq
self.gateway.set_child_value(
self.node_id, self.child_id, set_req.V_LIGHT, 0, ack=1
)
@callback
def _async_update(self) -> None:
"""Update the controller with the latest value from a device."""
super()._async_update()
self._current_command = cast(
Optional[str], self._child.values.get(self.value_type)
)

4
homeassistant/components/mysensors/sensor.py
View File

@ -165,6 +165,10 @@ SENSORS: dict[str, SensorEntityDescription] = {
device_class=SensorDeviceClass.CURRENT,
state_class=SensorStateClass.MEASUREMENT,
),
"V_IR_RECORD": SensorEntityDescription(
key="V_IR_RECORD",
icon="mdi:remote",
),
"V_PH": SensorEntityDescription(
key="V_PH",
native_unit_of_measurement="pH",

11
homeassistant/components/mysensors/strings.json
View File

@ -85,6 +85,17 @@
}
},
"issues": {
"deprecated_entity": {
"title": "The {deprecated_entity} entity will be removed",
"fix_flow": {
"step": {
"confirm": {
"title": "The {deprecated_entity} entity will be removed",
"description": "Update any automations or scripts that use this entity in service calls using the `{deprecated_service}` service to instead use the `{alternate_service}` service with a target entity ID of `{alternate_target}`."
}
}
}
},
"deprecated_service": {
"title": "The {deprecated_service} service will be removed",
"fix_flow": {

46
homeassistant/components/mysensors/switch.py
View File

@ -8,10 +8,11 @@ import voluptuous as vol
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON, Platform
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.core import HomeAssistant, ServiceCall, callback, split_entity_id
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from .. import mysensors
from .const import (
@ -151,8 +152,35 @@ class MySensorsIRSwitch(MySensorsSwitch):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the IR switch on."""
set_req = self.gateway.const.SetReq
placeholders = {
"deprecated_entity": self.entity_id,
"alternate_target": f"remote.{split_entity_id(self.entity_id)[1]}",
}
if ATTR_IR_CODE in kwargs:
self._ir_code = kwargs[ATTR_IR_CODE]
placeholders[
"deprecated_service"
] = f"{MYSENSORS_DOMAIN}.{SERVICE_SEND_IR_CODE}"
placeholders["alternate_service"] = "remote.send_command"
else:
placeholders["deprecated_service"] = "switch.turn_on"
placeholders["alternate_service"] = "remote.turn_on"
async_create_issue(
self.hass,
MYSENSORS_DOMAIN,
(
"deprecated_ir_switch_entity_"
f"{self.entity_id}_{placeholders['deprecated_service']}"
),
breaks_in_ha_version="2023.4.0",
is_fixable=True,
is_persistent=True,
severity=IssueSeverity.WARNING,
translation_key="deprecated_entity",
translation_placeholders=placeholders,
)
self.gateway.set_child_value(
self.node_id, self.child_id, self.value_type, self._ir_code
)
@ -169,6 +197,22 @@ class MySensorsIRSwitch(MySensorsSwitch):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the IR switch off."""
async_create_issue(
self.hass,
MYSENSORS_DOMAIN,
f"deprecated_ir_switch_entity_{self.entity_id}_switch.turn_off",
breaks_in_ha_version="2023.4.0",
is_fixable=True,
is_persistent=True,
severity=IssueSeverity.WARNING,
translation_key="deprecated_entity",
translation_placeholders={
"deprecated_entity": self.entity_id,
"deprecated_service": "switch.turn_off",
"alternate_service": "remote.turn_off",
"alternate_target": f"remote.{split_entity_id(self.entity_id)[1]}",
},
)
set_req = self.gateway.const.SetReq
self.gateway.set_child_value(
self.node_id, self.child_id, set_req.V_LIGHT, 0, ack=1

11
homeassistant/components/mysensors/translations/en.json
View File

@ -85,6 +85,17 @@
}
},
"issues": {
"deprecated_entity": {
"fix_flow": {
"step": {
"confirm": {
"description": "Update any automations or scripts that use this entity in service calls using the `{deprecated_service}` service to instead use the `{alternate_service}` service with a target entity ID of `{alternate_target}`.",
"title": "The {deprecated_entity} entity will be removed"
}
}
},
"title": "The {deprecated_entity} entity will be removed"
},
"deprecated_service": {
"fix_flow": {
"step": {

2
tests/components/mysensors/fixtures/ir_transceiver_state.json
View File

@ -9,7 +9,7 @@
"values": {
"2": "0",
"32": "test_code",
"33": "test_code"
"50": "test_code"
}
}
},

2
tests/components/mysensors/test_binary_sensor.py
View File

@ -2,7 +2,6 @@
from __future__ import annotations
from collections.abc import Callable
from unittest.mock import MagicMock
from mysensors.sensor import Sensor
@ -15,7 +14,6 @@ async def test_door_sensor(
hass: HomeAssistant,
door_sensor: Sensor,
receive_message: Callable[[str], None],
transport_write: MagicMock,
) -> None:
"""Test a door sensor."""
entity_id = "binary_sensor.door_sensor_1_1"

165
tests/components/mysensors/test_remote.py Normal file
View File

@ -0,0 +1,165 @@
"""Provide tests for mysensors remote platform."""
from __future__ import annotations
from collections.abc import Callable
from unittest.mock import MagicMock, call
from mysensors.const_14 import SetReq
from mysensors.sensor import Sensor
import pytest
from homeassistant.components.remote import (
ATTR_COMMAND,
DOMAIN as REMOTE_DOMAIN,
SERVICE_LEARN_COMMAND,
SERVICE_SEND_COMMAND,
)
from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TURN_OFF, SERVICE_TURN_ON
from homeassistant.core import HomeAssistant
async def test_ir_transceiver(
hass: HomeAssistant,
ir_transceiver: Sensor,
receive_message: Callable[[str], None],
transport_write: MagicMock,
) -> None:
"""Test an ir transceiver."""
entity_id = "remote.ir_transceiver_1_1"
state = hass.states.get(entity_id)
assert state
assert state.state == "off"
# Test turn on
await hass.services.async_call(
REMOTE_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
assert transport_write.call_count == 2
assert transport_write.call_args_list[0] == call("1;1;1;1;32;test_code\n")
assert transport_write.call_args_list[1] == call("1;1;1;1;2;1\n")
receive_message("1;1;1;0;32;test_code\n")
receive_message("1;1;1;0;2;1\n")
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.state == "on"
transport_write.reset_mock()
# Test send command
await hass.services.async_call(
REMOTE_DOMAIN,
SERVICE_SEND_COMMAND,
{ATTR_ENTITY_ID: entity_id, ATTR_COMMAND: "new_code"},
blocking=True,
)
assert transport_write.call_count == 1
assert transport_write.call_args == call("1;1;1;1;32;new_code\n")
receive_message("1;1;1;0;32;new_code\n")
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.state == "on"
transport_write.reset_mock()
# Test learn command
await hass.services.async_call(
REMOTE_DOMAIN,
SERVICE_LEARN_COMMAND,
{ATTR_ENTITY_ID: entity_id, ATTR_COMMAND: "learn_code"},
blocking=True,
)
assert transport_write.call_count == 1
assert transport_write.call_args == call("1;1;1;1;50;learn_code\n")
receive_message("1;1;1;0;50;learn_code\n")
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.state == "on"
transport_write.reset_mock()
# Test learn command with missing command parameter
with pytest.raises(ValueError):
await hass.services.async_call(
REMOTE_DOMAIN,
SERVICE_LEARN_COMMAND,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
assert transport_write.call_count == 0
transport_write.reset_mock()
# Test turn off
await hass.services.async_call(
REMOTE_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
assert transport_write.call_count == 1
assert transport_write.call_args == call("1;1;1;1;2;0\n")
receive_message("1;1;1;0;2;0\n")
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.state == "off"
transport_write.reset_mock()
# Test turn on with new default code
await hass.services.async_call(
REMOTE_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
assert transport_write.call_count == 2
assert transport_write.call_args_list[0] == call("1;1;1;1;32;new_code\n")
assert transport_write.call_args_list[1] == call("1;1;1;1;2;1\n")
receive_message("1;1;1;0;32;new_code\n")
receive_message("1;1;1;0;2;1\n")
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.state == "on"
# Test unknown state
ir_transceiver.children[1].values.pop(SetReq.V_LIGHT)
# Trigger state update
receive_message("1;1;1;0;32;new_code\n")
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.state == "unknown"

22
tests/components/mysensors/test_sensor.py
View File

@ -55,6 +55,28 @@ async def test_gps_sensor(
assert state.state == f"{new_coords},{altitude}"
async def test_ir_transceiver(
hass: HomeAssistant,
ir_transceiver: Sensor,
receive_message: Callable[[str], None],
) -> None:
"""Test an ir transceiver."""
entity_id = "sensor.ir_transceiver_1_1"
state = hass.states.get(entity_id)
assert state
assert state.state == "test_code"
receive_message("1;1;1;0;50;new_code\n")
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.state == "new_code"
async def test_power_sensor(
hass: HomeAssistant,
power_sensor: Sensor,