starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-26 17:31:15 -04:00
Code Issues Packages Projects Releases Wiki Activity

[infrared_proxy] Initial commit

Browse Source
This commit is contained in:
kbx81
2025-12-18 00:51:11 -06:00
parent 52229dc5a8
commit d47d013219
4 changed files with 371 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/esphome/entry_data.py
+24
View File
@@ -29,6 +29,7 @@ from aioesphomeapi import (
Event,
EventInfo,
FanInfo,
InfraredProxyInfo,
LightInfo,
LockInfo,
MediaPlayerInfo,
@@ -84,6 +85,7 @@ INFO_TYPE_TO_PLATFORM: dict[type[EntityInfo], Platform] = {
DateTimeInfo: Platform.DATETIME,
EventInfo: Platform.EVENT,
FanInfo: Platform.FAN,
InfraredProxyInfo: Platform.REMOTE,
LightInfo: Platform.LIGHT,
LockInfo: Platform.LOCK,
MediaPlayerInfo: Platform.MEDIA_PLAYER,
@@ -187,6 +189,7 @@ class RuntimeEntryData:
entity_removal_callbacks: dict[EntityInfoKey, list[CALLBACK_TYPE]] = field(
default_factory=dict
)
infrared_proxy_receive_callbacks: list[CALLBACK_TYPE] = field(default_factory=list)
@property
def name(self) -> str:
@@ -518,6 +521,27 @@ class RuntimeEntryData:
),
)
@callback
def async_on_infrared_proxy_receive(
self, hass: HomeAssistant, receive_event: Any
) -> None:
"""Handle an infrared proxy receive event."""
# Fire a Home Assistant event with the infrared data
device_info = self.device_info
if not device_info:
return
hass.bus.async_fire(
f"{DOMAIN}_infrared_proxy_received",
{
"device_name": device_info.name,
"device_mac": device_info.mac_address,
"entry_id": self.entry_id,
"key": receive_event.key,
"timings": receive_event.timings,
},
)
@callback
def async_register_assist_satellite_config_updated_callback(
self,
homeassistant/components/esphome/manager.py
+9
View File
@@ -692,6 +692,11 @@ class ESPHomeManager:
cli.subscribe_zwave_proxy_request(self._async_zwave_proxy_request)
)
if device_info.infrared_proxy_feature_flags:
entry_data.disconnect_callbacks.add(
cli.subscribe_infrared_proxy_receive(self._async_infrared_proxy_receive)
)
cli.subscribe_home_assistant_states_and_services(
on_state=entry_data.async_update_state,
on_service_call=self.async_on_service_call,
@@ -722,6 +727,10 @@ class ESPHomeManager:
self.hass, self.entry_data.device_info, zwave_home_id
)
def _async_infrared_proxy_receive(self, receive_event: Any) -> None:
"""Handle an infrared proxy receive event."""
self.entry_data.async_on_infrared_proxy_receive(self.hass, receive_event)
async def on_disconnect(self, expected_disconnect: bool) -> None:
"""Run disconnect callbacks on API disconnect."""
entry_data = self.entry_data
homeassistant/components/esphome/remote.py
+93
View File
@@ -0,0 +1,93 @@
"""Support for ESPHome infrared proxy remote components."""
from __future__ import annotations
from collections.abc import Iterable
from functools import partial
import logging
from typing import Any
from aioesphomeapi import (
EntityInfo,
EntityState,
InfraredProxyCapability,
InfraredProxyInfo,
)
from homeassistant.components.remote import RemoteEntity, RemoteEntityFeature
from homeassistant.core import callback
from homeassistant.exceptions import HomeAssistantError
from .entity import EsphomeEntity, platform_async_setup_entry
_LOGGER = logging.getLogger(__name__)
PARALLEL_UPDATES = 0
class EsphomeInfraredProxy(EsphomeEntity[InfraredProxyInfo, EntityState], RemoteEntity):
"""An infrared proxy remote implementation for ESPHome."""
@callback
def _on_static_info_update(self, static_info: EntityInfo) -> None:
"""Set attrs from static info."""
super()._on_static_info_update(static_info)
static_info = self._static_info
capabilities = static_info.capabilities
# Set supported features based on capabilities
features = RemoteEntityFeature(0)
if capabilities & InfraredProxyCapability.RECEIVER:
features |= RemoteEntityFeature.LEARN_COMMAND
self._attr_supported_features = features
@callback
def _on_device_update(self) -> None:
"""Call when device updates or entry data changes."""
super()._on_device_update()
if self._entry_data.available:
# Infrared proxy entities should go available directly
# when the device comes online.
self.async_write_ha_state()
@property
def is_on(self) -> bool:
"""Return true if remote is on."""
# ESPHome infrared proxies are always on when available
return self.available
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the remote on."""
# ESPHome infrared proxies are always on, nothing to do
_LOGGER.debug("Turn on called for %s (no-op)", self.name)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the remote off."""
# ESPHome infrared proxies cannot be turned off
_LOGGER.debug("Turn off called for %s (no-op)", self.name)
async def async_send_command(self, command: Iterable[str], **kwargs: Any) -> None:
"""Send commands to a device."""
# This method would need to parse command data and timing parameters
# For now, we'll raise an error as this requires more complex implementation
raise HomeAssistantError(
"Direct command sending not yet implemented for ESPHome infrared proxy. "
"Use the infrared_proxy_transmit service instead."
)
async def async_learn_command(self, **kwargs: Any) -> None:
"""Learn a command from a device."""
# Learning is handled through the receive event subscription
# which is managed at the entry_data level
raise HomeAssistantError(
"Learning commands is handled automatically through receive events. "
"Listen for esphome_infrared_proxy_received events instead."
)
async_setup_entry = partial(
platform_async_setup_entry,
info_type=InfraredProxyInfo,
entity_type=EsphomeInfraredProxy,
state_type=EntityState,
)
tests/components/esphome/test_remote.py
+245
View File
@@ -0,0 +1,245 @@
"""Test ESPHome infrared proxy remotes."""
from aioesphomeapi import (
APIClient,
InfraredProxyCapability,
InfraredProxyInfo,
InfraredProxyReceiveEvent,
)
import pytest
from homeassistant.components.remote import DOMAIN as REMOTE_DOMAIN, RemoteEntityFeature
from homeassistant.const import STATE_ON, STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
async def test_infrared_proxy_transmitter_only(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device,
) -> None:
"""Test an infrared proxy remote with transmitter capability only."""
entity_info = [
InfraredProxyInfo(
object_id="myremote",
key=1,
name="my remote",
capabilities=InfraredProxyCapability.TRANSMITTER,
)
]
states = []
user_service = []
await mock_esphome_device(
mock_client=mock_client,
entity_info=entity_info,
user_service=user_service,
states=states,
)
await hass.async_block_till_done()
# Test initial state
state = hass.states.get("remote.test_my_remote")
assert state is not None
assert state.state == STATE_ON
# Transmitter-only should not support learn
assert state.attributes["supported_features"] == 0
async def test_infrared_proxy_receiver_capability(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device,
) -> None:
"""Test an infrared proxy remote with receiver capability."""
entity_info = [
InfraredProxyInfo(
object_id="myremote",
key=1,
name="my remote",
capabilities=InfraredProxyCapability.TRANSMITTER
| InfraredProxyCapability.RECEIVER,
)
]
states = []
user_service = []
await mock_esphome_device(
mock_client=mock_client,
entity_info=entity_info,
user_service=user_service,
states=states,
)
await hass.async_block_till_done()
# Test initial state
state = hass.states.get("remote.test_my_remote")
assert state is not None
assert state.state == STATE_ON
# Should support learn command
assert state.attributes["supported_features"] == RemoteEntityFeature.LEARN_COMMAND
async def test_infrared_proxy_unavailability(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device,
) -> None:
"""Test infrared proxy remote availability."""
entity_info = [
InfraredProxyInfo(
object_id="myremote",
key=1,
name="my remote",
capabilities=InfraredProxyCapability.TRANSMITTER,
)
]
states = []
user_service = []
device = await mock_esphome_device(
mock_client=mock_client,
entity_info=entity_info,
user_service=user_service,
states=states,
)
await hass.async_block_till_done()
# Test initial state
state = hass.states.get("remote.test_my_remote")
assert state is not None
assert state.state == STATE_ON
# Test device becomes unavailable
await device.mock_disconnect(True)
await hass.async_block_till_done()
state = hass.states.get("remote.test_my_remote")
assert state.state == STATE_UNAVAILABLE
# Test device becomes available again
await device.mock_connect()
await hass.async_block_till_done()
state = hass.states.get("remote.test_my_remote")
assert state.state == STATE_ON
async def test_infrared_proxy_receive_event(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device,
) -> None:
"""Test infrared proxy receive event firing."""
entity_info = [
InfraredProxyInfo(
object_id="myremote",
key=1,
name="my remote",
capabilities=InfraredProxyCapability.RECEIVER,
)
]
states = []
user_service = []
device = await mock_esphome_device(
mock_client=mock_client,
entity_info=entity_info,
user_service=user_service,
states=states,
)
await hass.async_block_till_done()
events = []
def event_listener(event):
events.append(event)
hass.bus.async_listen("esphome_infrared_proxy_received", event_listener)
# Simulate receiving an infrared signal
receive_event = InfraredProxyReceiveEvent(
key=1,
timings=[1000, 500, 1000, 500, 500, 1000],
)
# Get entry_data from the config entry
entry_data = device.entry.runtime_data
entry_data.async_on_infrared_proxy_receive(hass, receive_event)
await hass.async_block_till_done()
# Verify event was fired
assert len(events) == 1
event_data = events[0].data
assert event_data["key"] == 1
assert event_data["timings"] == [1000, 500, 1000, 500, 500, 1000]
assert event_data["device_name"] == "test"
assert "entry_id" in event_data
async def test_infrared_proxy_send_command_not_implemented(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device,
) -> None:
"""Test that send_command raises appropriate error."""
entity_info = [
InfraredProxyInfo(
object_id="myremote",
key=1,
name="my remote",
capabilities=InfraredProxyCapability.TRANSMITTER,
)
]
states = []
user_service = []
await mock_esphome_device(
mock_client=mock_client,
entity_info=entity_info,
user_service=user_service,
states=states,
)
await hass.async_block_till_done()
# Test send_command raises error
with pytest.raises(
HomeAssistantError,
match="Direct command sending not yet implemented",
):
await hass.services.async_call(
REMOTE_DOMAIN,
"send_command",
{"entity_id": "remote.test_my_remote", "command": ["test"]},
blocking=True,
)
async def test_infrared_proxy_learn_command_not_implemented(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device,
) -> None:
"""Test that learn_command raises appropriate error."""
entity_info = [
InfraredProxyInfo(
object_id="myremote",
key=1,
name="my remote",
capabilities=InfraredProxyCapability.RECEIVER,
)
]
states = []
user_service = []
await mock_esphome_device(
mock_client=mock_client,
entity_info=entity_info,
user_service=user_service,
states=states,
)
await hass.async_block_till_done()
# Test learn_command raises error
with pytest.raises(
HomeAssistantError,
match="Learning commands is handled automatically",
):
await hass.services.async_call(
REMOTE_DOMAIN,
"learn_command",
{"entity_id": "remote.test_my_remote"},
blocking=True,
)