Add infrared component; replace remote by infrared in esphome

This commit is contained in:
abmantis
2025-12-22 23:25:26 +00:00
parent c37ca31bec
commit 343c4183f7
18 changed files with 1073 additions and 604 deletions
Generated
+2
View File
@@ -772,6 +772,8 @@ build.json @home-assistant/supervisor
/tests/components/inels/ @epdevlab
/homeassistant/components/influxdb/ @mdegat01
/tests/components/influxdb/ @mdegat01
/homeassistant/components/infrared/ @home-assistant/core
/tests/components/infrared/ @home-assistant/core
/homeassistant/components/inkbird/ @bdraco
/tests/components/inkbird/ @bdraco
/homeassistant/components/input_boolean/ @home-assistant/core
@@ -85,7 +85,7 @@ INFO_TYPE_TO_PLATFORM: dict[type[EntityInfo], Platform] = {
DateTimeInfo: Platform.DATETIME,
EventInfo: Platform.EVENT,
FanInfo: Platform.FAN,
InfraredProxyInfo: Platform.REMOTE,
InfraredProxyInfo: Platform.INFRARED,
LightInfo: Platform.LIGHT,
LockInfo: Platform.LOCK,
MediaPlayerInfo: Platform.MEDIA_PLAYER,
@@ -0,0 +1,199 @@
"""Infrared platform for ESPHome."""
from __future__ import annotations
from functools import partial
import json
import logging
from aioesphomeapi import (
EntityInfo,
EntityState,
InfraredProxyCapability,
InfraredProxyInfo,
InfraredProxyTimingParams,
)
from homeassistant.components.infrared import (
PULSE_WIDTH_COMPAT_PROTOCOLS,
BaseIRCommand,
InfraredEntity,
InfraredEntityFeature,
IRProtocolType,
NECIRCommand,
PulseWidthIRCommand,
SamsungIRCommand,
)
from homeassistant.core import callback
from homeassistant.exceptions import HomeAssistantError
from .const import DOMAIN
from .entity import EsphomeEntity, platform_async_setup_entry
_LOGGER = logging.getLogger(__name__)
PARALLEL_UPDATES = 0
class EsphomeInfraredEntity(
EsphomeEntity[InfraredProxyInfo, EntityState], InfraredEntity
):
"""ESPHome infrared entity using native API."""
@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
features = InfraredEntityFeature(0)
if capabilities & InfraredProxyCapability.TRANSMITTER:
features |= InfraredEntityFeature.TRANSMIT
if capabilities & InfraredProxyCapability.RECEIVER:
features |= InfraredEntityFeature.RECEIVE
self._attr_supported_features = features
if capabilities & InfraredProxyCapability.TRANSMITTER:
self._attr_supported_protocols = {
IRProtocolType.PULSE_WIDTH,
IRProtocolType.NEC,
IRProtocolType.SAMSUNG,
}
else:
self._attr_supported_protocols = set()
@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 entities should go available as soon as the device comes online
self.async_write_ha_state()
async def async_send_command(self, command: BaseIRCommand) -> None:
"""Send an IR command.
Raises:
HomeAssistantError: If transmission fails or not supported.
"""
if not self._static_info.capabilities & InfraredProxyCapability.TRANSMITTER:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="infrared_proxy_transmitter_not_supported",
)
if isinstance(command, (NECIRCommand, SamsungIRCommand)):
await self._async_send_protocol_command(command)
else:
# Fall back to pulse-width transmission if the protocol is compatible
await self._async_send_pulse_width_command(command)
async def _async_send_protocol_command(self, command: BaseIRCommand) -> None:
"""Send command using protocol-specific arguments."""
if isinstance(command, NECIRCommand):
cmd_json = json.dumps(
{
"protocol": "nec",
"address": command.address,
"command": command.command,
"repeat": command.repeat_count,
}
)
elif isinstance(command, SamsungIRCommand):
cmd_json = json.dumps(
{
"protocol": "samsung",
"data": command.code,
"nbits": command.length_in_bits,
"repeat": command.repeat_count,
}
)
else:
raise HomeAssistantError(
f"Unsupported protocol command type: {type(command)}"
)
_LOGGER.debug("Sending command: %s", cmd_json)
try:
self._client.infrared_proxy_transmit_protocol(
self._static_info.key, cmd_json
)
except Exception as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="error_sending_ir_command",
translation_placeholders={
"device_name": self._device_info.name,
"error": str(err),
},
) from err
async def _async_send_pulse_width_command(self, command: BaseIRCommand) -> None:
"""Send command using the pulse-width generic protocol."""
if isinstance(command, PulseWidthIRCommand):
protocol = command.protocol
code = command.code
length_in_bits = command.length_in_bits
elif command.protocol.type in PULSE_WIDTH_COMPAT_PROTOCOLS:
compat_protocol_method = getattr(
command.protocol, "get_pulse_width_compat_protocol", None
)
compat_code_method = getattr(command, "get_pulse_width_compat_code", None)
protocol = compat_protocol_method() # type: ignore[misc]
code = compat_code_method() # type: ignore[misc]
length_in_bits = 32
else:
raise HomeAssistantError(f"Unsupported command type: {type(command)}")
num_bytes = (length_in_bits + 7) // 8
data_bytes = code.to_bytes(
num_bytes, byteorder="big" if protocol.msb_first else "little"
)
timing = InfraredProxyTimingParams(
frequency=protocol.frequency,
length_in_bits=length_in_bits,
header_high_us=protocol.header.high_us,
header_low_us=protocol.header.low_us,
one_high_us=protocol.one.high_us,
one_low_us=protocol.one.low_us,
zero_high_us=protocol.zero.high_us,
zero_low_us=protocol.zero.low_us,
footer_high_us=protocol.footer.high_us,
footer_low_us=protocol.footer.low_us,
repeat_high_us=0,
repeat_low_us=0,
minimum_idle_time_us=protocol.minimum_idle_time_us,
msb_first=protocol.msb_first,
repeat_count=command.repeat_count,
)
_LOGGER.debug(
"Sending pulse-width command via native API: timing=%s, data=%s",
timing,
data_bytes.hex(),
)
try:
self._client.infrared_proxy_transmit(
self._static_info.key, timing, data_bytes
)
except Exception as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="error_sending_ir_command",
translation_placeholders={
"device_name": self._device_info.name,
"error": str(err),
},
) from err
async_setup_entry = partial(
platform_async_setup_entry,
info_type=InfraredProxyInfo,
entity_type=EsphomeInfraredEntity,
state_type=EntityState,
)
-172
View File
@@ -1,172 +0,0 @@
"""Support for ESPHome infrared proxy remote components."""
from __future__ import annotations
from collections.abc import Iterable
from functools import partial
import json
import logging
from typing import Any
from aioesphomeapi import (
EntityInfo,
EntityState,
InfraredProxyCapability,
InfraredProxyInfo,
InfraredProxyTimingParams,
)
from homeassistant.components.remote import RemoteEntity, RemoteEntityFeature
from homeassistant.core import callback
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from .const import DOMAIN
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.
Commands should be JSON strings containing either:
1. Protocol-based format: {"protocol": "NEC", "address": 0x04, "command": 0x08}
2. Pulse-width format: {
"timing": {
"frequency": 38000,
"length_in_bits": 32,
"header_high_us": 9000,
"header_low_us": 4500,
...
},
"data": [0x01, 0x02, 0x03, 0x04]
}
"""
self._check_capabilities()
for cmd in command:
try:
cmd_data = json.loads(cmd)
except json.JSONDecodeError as err:
raise ServiceValidationError(
f"Command must be valid JSON: {err}"
) from err
# Check if this is a protocol-based command
if "protocol" in cmd_data:
self._client.infrared_proxy_transmit_protocol(
self._static_info.key,
cmd, # Pass the original JSON string
)
# Check if this is a pulse-width command
elif "timing" in cmd_data and "data" in cmd_data:
timing_data = cmd_data["timing"]
data_array = cmd_data["data"]
# Convert array of integers to bytes
if not isinstance(data_array, list):
raise ServiceValidationError(
"Data must be an array of integers (0-255)"
)
try:
data_bytes = bytes(data_array)
except (ValueError, TypeError) as err:
raise ServiceValidationError(
f"Invalid data array: {err}. Each element must be an integer between 0 and 255."
) from err
timing = InfraredProxyTimingParams(
frequency=timing_data.get("frequency", 38000),
length_in_bits=timing_data.get("length_in_bits", 32),
header_high_us=timing_data.get("header_high_us", 0),
header_low_us=timing_data.get("header_low_us", 0),
one_high_us=timing_data.get("one_high_us", 0),
one_low_us=timing_data.get("one_low_us", 0),
zero_high_us=timing_data.get("zero_high_us", 0),
zero_low_us=timing_data.get("zero_low_us", 0),
footer_high_us=timing_data.get("footer_high_us", 0),
footer_low_us=timing_data.get("footer_low_us", 0),
repeat_high_us=timing_data.get("repeat_high_us", 0),
repeat_low_us=timing_data.get("repeat_low_us", 0),
minimum_idle_time_us=timing_data.get("minimum_idle_time_us", 0),
msb_first=timing_data.get("msb_first", True),
repeat_count=timing_data.get("repeat_count", 1),
)
self._client.infrared_proxy_transmit(
self._static_info.key,
timing,
data_bytes,
)
else:
raise ServiceValidationError(
"Command must contain either 'protocol' or both 'timing' and 'data' fields"
)
def _check_capabilities(self) -> None:
"""Check if the device supports transmission."""
if not self._static_info.capabilities & InfraredProxyCapability.TRANSMITTER:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="infrared_proxy_transmitter_not_supported",
)
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,
)
@@ -137,6 +137,9 @@
"error_compiling": {
"message": "Error compiling {configuration}. Try again in ESPHome dashboard for more information."
},
"error_sending_ir_command": {
"message": "Error sending IR command to {device_name}: {error}"
},
"error_uploading": {
"message": "Error during OTA (Over-The-Air) update of {configuration}. Try again in ESPHome dashboard for more information."
},
@@ -0,0 +1,143 @@
"""Support for infrared transmitter entities."""
from __future__ import annotations
from abc import abstractmethod
from datetime import timedelta
import logging
from typing import Any
from propcache.api import cached_property
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity import Entity, EntityDescription
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.typing import ConfigType
from homeassistant.util.hass_dict import HassKey
from .const import DOMAIN, InfraredEntityFeature
from .protocols import (
PULSE_WIDTH_COMPAT_PROTOCOLS,
BaseIRCommand,
BaseIRProtocol,
IRProtocolType,
IRTiming,
NECIRCommand,
NECIRProtocol,
PulseWidthIRCommand,
PulseWidthIRProtocol,
SamsungIRCommand,
SamsungIRProtocol,
)
__all__ = [
"DOMAIN",
"PULSE_WIDTH_COMPAT_PROTOCOLS",
"BaseIRCommand",
"BaseIRProtocol",
"IRProtocolType",
"IRTiming",
"InfraredEntity",
"InfraredEntityDescription",
"InfraredEntityFeature",
"NECIRCommand",
"NECIRProtocol",
"PulseWidthIRCommand",
"PulseWidthIRProtocol",
"SamsungIRCommand",
"SamsungIRProtocol",
"async_get_entities",
]
_LOGGER = logging.getLogger(__name__)
DATA_COMPONENT: HassKey[EntityComponent[InfraredEntity]] = HassKey(DOMAIN)
ENTITY_ID_FORMAT = DOMAIN + ".{}"
PLATFORM_SCHEMA = cv.PLATFORM_SCHEMA
PLATFORM_SCHEMA_BASE = cv.PLATFORM_SCHEMA_BASE
SCAN_INTERVAL = timedelta(seconds=30)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the infrared domain."""
component = hass.data[DATA_COMPONENT] = EntityComponent[InfraredEntity](
_LOGGER, DOMAIN, hass, SCAN_INTERVAL
)
await component.async_setup(config)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a config entry."""
return await hass.data[DATA_COMPONENT].async_setup_entry(entry)
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.data[DATA_COMPONENT].async_unload_entry(entry)
@callback
def async_get_entities(
hass: HomeAssistant, protocols: set[str] | None = None
) -> list[InfraredEntity]:
"""Get all infrared entities, optionally filtered by protocol support."""
component = hass.data.get(DATA_COMPONENT)
if component is None:
return []
entities = list(component.entities)
if protocols is not None:
protocol_set = set(protocols)
entities = [e for e in entities if e.supported_protocols & protocol_set]
return entities
class InfraredEntityDescription(EntityDescription, frozen_or_thawed=True):
"""Describes infrared entities."""
CACHED_PROPERTIES_WITH_ATTR_ = {
"supported_features",
"supported_protocols",
}
ATTR_SUPPORTED_PROTOCOLS = "supported_protocols"
class InfraredEntity(Entity, cached_properties=CACHED_PROPERTIES_WITH_ATTR_):
"""Base class for infrared transmitter entities."""
entity_description: InfraredEntityDescription
_attr_supported_features: InfraredEntityFeature = InfraredEntityFeature(0)
_attr_supported_protocols: set[str] = set()
@cached_property
def supported_features(self) -> InfraredEntityFeature:
"""Flag supported features."""
return self._attr_supported_features
@cached_property
def supported_protocols(self) -> set[str]:
"""Return set of supported IR protocol types."""
return self._attr_supported_protocols
@property
def capability_attributes(self) -> dict[str, Any] | None:
"""Return capability attributes."""
return {ATTR_SUPPORTED_PROTOCOLS: sorted(self.supported_protocols)}
@abstractmethod
async def async_send_command(self, command: BaseIRCommand) -> None:
"""Send an IR command.
Args:
command: The IR command to send.
Raises:
HomeAssistantError: If transmission fails.
"""
@@ -0,0 +1,16 @@
"""Constants for the Infrared integration."""
from enum import IntFlag
from typing import Final
DOMAIN: Final = "infrared"
class InfraredEntityFeature(IntFlag):
"""Supported features of infrared entities."""
TRANSMIT = 1
"""Entity can transmit IR signals."""
RECEIVE = 2
"""Entity can receive/learn IR signals."""
@@ -0,0 +1,8 @@
{
"domain": "infrared",
"name": "Infrared",
"codeowners": ["@home-assistant/core"],
"documentation": "https://www.home-assistant.io/integrations/infrared",
"integration_type": "entity",
"quality_scale": "internal"
}
@@ -0,0 +1,146 @@
"""IR protocol definitions for the Infrared integration."""
from __future__ import annotations
from abc import ABC
from dataclasses import dataclass
from enum import StrEnum
class IRProtocolType(StrEnum):
"""IR protocol type identifiers."""
PULSE_WIDTH = "pulse_width"
NEC = "nec"
SAMSUNG = "samsung"
PULSE_WIDTH_COMPAT_PROTOCOLS = {IRProtocolType.NEC, IRProtocolType.SAMSUNG}
@dataclass(frozen=True, slots=True)
class IRTiming:
"""Timing for a signal component."""
high_us: int
low_us: int
class BaseIRProtocol:
"""Base class for IR protocol definitions."""
type: IRProtocolType
@dataclass(frozen=True, slots=True)
class PulseWidthIRProtocol(BaseIRProtocol):
"""Pulse-width modulated IR protocol.
Defines timing for header, one bit, zero bit, and footer.
Used to convert a numeric code into raw timing data.
Attributes:
header: Timing for the header pulse.
one: Timing for a '1' bit.
zero: Timing for a '0' bit.
footer: Timing for the footer pulse.
frequency: Carrier frequency in Hz (e.g., 38000 for 38kHz).
msb_first: If True, send most significant bit first (default for most protocols).
minimum_idle_time_us: Minimum gap between transmissions in microseconds.
"""
type = IRProtocolType.PULSE_WIDTH
header: IRTiming
one: IRTiming
zero: IRTiming
footer: IRTiming
frequency: int = 38000
msb_first: bool = True
minimum_idle_time_us: int = 0
@dataclass(frozen=True, slots=True)
class NECIRProtocol(BaseIRProtocol):
"""NEC IR protocol."""
type = IRProtocolType.NEC
def get_pulse_width_compat_protocol(self) -> PulseWidthIRProtocol:
"""Convert to a PulseWidthIRProtocol for encoding."""
return PulseWidthIRProtocol(
header=IRTiming(high_us=9000, low_us=4500),
one=IRTiming(high_us=560, low_us=1690),
zero=IRTiming(high_us=560, low_us=560),
footer=IRTiming(high_us=560, low_us=0),
msb_first=False,
minimum_idle_time_us=40000,
)
@dataclass(frozen=True, slots=True)
class SamsungIRProtocol(BaseIRProtocol):
"""Samsung 32-bit IR protocol."""
type = IRProtocolType.SAMSUNG
def get_pulse_width_compat_protocol(self) -> PulseWidthIRProtocol:
"""Convert to a PulseWidthIRProtocol for encoding."""
return PulseWidthIRProtocol(
header=IRTiming(high_us=4500, low_us=4500),
one=IRTiming(high_us=560, low_us=1690),
zero=IRTiming(high_us=560, low_us=560),
footer=IRTiming(high_us=560, low_us=0),
msb_first=False,
minimum_idle_time_us=0,
)
@dataclass(frozen=True, slots=True)
class BaseIRCommand[P: BaseIRProtocol](ABC):
"""Base class for IR commands.
Attributes:
protocol: The IR protocol to use for encoding the command.
repeat_count: How many times to send the command.
"""
protocol: P
repeat_count: int
@dataclass(frozen=True, slots=True)
class PulseWidthIRCommand(BaseIRCommand[PulseWidthIRProtocol]):
"""IR command with a numeric code for pulse-width protocols."""
code: int
length_in_bits: int
@dataclass(frozen=True, slots=True)
class NECIRCommand(BaseIRCommand[NECIRProtocol]):
"""NEC IR command."""
address: int
command: int
def get_pulse_width_compat_code(self) -> int:
"""Return the code in pulse-width compatible 32-bit format."""
addr = self.address & 0xFFFF
cmd = self.command & 0xFFFF
return addr | (cmd << 16)
@dataclass(frozen=True, slots=True)
class SamsungIRCommand(BaseIRCommand[SamsungIRProtocol]):
"""Samsung IR command."""
code: int
length_in_bits: int = 32
def get_pulse_width_compat_code(self) -> int:
"""Return the code in pulse-width compatible format.
Samsung codes are already 32-bit integers, so no conversion is needed.
"""
return self.code
@@ -0,0 +1,7 @@
{
"exceptions": {
"send_command_failed": {
"message": "Failed to send IR command: {error}"
}
}
}
+1
View File
@@ -29,6 +29,7 @@ class EntityPlatforms(StrEnum):
HUMIDIFIER = "humidifier"
IMAGE = "image"
IMAGE_PROCESSING = "image_processing"
INFRARED = "infrared"
LAWN_MOWER = "lawn_mower"
LIGHT = "light"
LOCK = "lock"
+1
View File
@@ -2171,6 +2171,7 @@ NO_QUALITY_SCALE = [
"input_text",
"intent_script",
"intent",
"infrared",
"labs",
"logbook",
"logger",
+301
View File
@@ -0,0 +1,301 @@
"""Test ESPHome infrared platform."""
import json
from unittest.mock import patch
from aioesphomeapi import (
APIClient,
InfraredProxyCapability,
InfraredProxyInfo,
InfraredProxyReceiveEvent,
)
import pytest
from homeassistant.components.infrared import (
BaseIRCommand,
InfraredEntityFeature,
IRProtocolType,
IRTiming,
NECIRCommand,
NECIRProtocol,
PulseWidthIRCommand,
PulseWidthIRProtocol,
SamsungIRCommand,
SamsungIRProtocol,
async_get_entities,
)
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr, entity_registry as er
from .conftest import MockESPHomeDeviceType
def _create_infrared_proxy_info(
object_id: str = "myremote",
key: int = 1,
name: str = "my remote",
capabilities: InfraredProxyCapability = InfraredProxyCapability.TRANSMITTER,
) -> InfraredProxyInfo:
"""Create mock InfraredProxyInfo."""
return InfraredProxyInfo(
object_id=object_id, key=key, name=name, capabilities=capabilities
)
@pytest.mark.parametrize(
("capabilities", "expected_features"),
[
(InfraredProxyCapability.TRANSMITTER, InfraredEntityFeature.TRANSMIT),
(
InfraredProxyCapability.RECEIVER,
InfraredEntityFeature.RECEIVE,
),
(
InfraredProxyCapability.TRANSMITTER | InfraredProxyCapability.RECEIVER,
InfraredEntityFeature.TRANSMIT | InfraredEntityFeature.RECEIVE,
),
],
)
async def test_capabilities(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device: MockESPHomeDeviceType,
capabilities: InfraredProxyCapability,
expected_features: InfraredEntityFeature,
) -> None:
"""Test infrared entity capabilities."""
entity_info = [_create_infrared_proxy_info(capabilities=capabilities)]
await mock_esphome_device(mock_client=mock_client, entity_info=entity_info)
await hass.async_block_till_done()
state = hass.states.get("infrared.test_my_remote")
assert state is not None
assert state.attributes.get("supported_features") == expected_features
async def test_supported_protocols(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device: MockESPHomeDeviceType,
) -> None:
"""Test infrared entity supported protocols."""
entity_info = [_create_infrared_proxy_info()]
await mock_esphome_device(mock_client=mock_client, entity_info=entity_info)
await hass.async_block_till_done()
state = hass.states.get("infrared.test_my_remote")
assert state is not None
protocols = state.attributes.get("supported_protocols")
assert protocols is not None
assert IRProtocolType.NEC.value in protocols
assert IRProtocolType.PULSE_WIDTH.value in protocols
assert IRProtocolType.SAMSUNG.value in protocols
async def test_unavailability(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device: MockESPHomeDeviceType,
) -> None:
"""Test infrared entity availability."""
entity_info = [_create_infrared_proxy_info()]
device = await mock_esphome_device(mock_client=mock_client, entity_info=entity_info)
await hass.async_block_till_done()
state = hass.states.get("infrared.test_my_remote")
assert state is not None
assert state.state != STATE_UNAVAILABLE
await device.mock_disconnect(True)
await hass.async_block_till_done()
state = hass.states.get("infrared.test_my_remote")
assert state.state == STATE_UNAVAILABLE
await device.mock_connect()
await hass.async_block_till_done()
state = hass.states.get("infrared.test_my_remote")
assert state.state != STATE_UNAVAILABLE
async def test_receive_event(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device: MockESPHomeDeviceType,
) -> None:
"""Test infrared receive event firing."""
entity_info = [
_create_infrared_proxy_info(capabilities=InfraredProxyCapability.RECEIVER)
]
device = await mock_esphome_device(mock_client=mock_client, entity_info=entity_info)
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],
)
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
@pytest.mark.parametrize(
("command", "expected_json"),
[
(
NECIRCommand(
protocol=NECIRProtocol(),
repeat_count=1,
address=0x10,
command=0x20,
),
{"protocol": "nec", "address": 0x10, "command": 0x20, "repeat": 1},
),
(
SamsungIRCommand(
protocol=SamsungIRProtocol(),
repeat_count=2,
code=0xE0E040BF,
length_in_bits=32,
),
{"protocol": "samsung", "data": 0xE0E040BF, "nbits": 32, "repeat": 2},
),
],
)
async def test_send_nec_command(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device: MockESPHomeDeviceType,
command: BaseIRCommand,
expected_json: dict,
) -> None:
"""Test sending command via native API."""
entity_info = [_create_infrared_proxy_info()]
await mock_esphome_device(mock_client=mock_client, entity_info=entity_info)
await hass.async_block_till_done()
entities = async_get_entities(hass)
assert len(entities) == 1
entity = entities[0]
with patch.object(mock_client, "infrared_proxy_transmit_protocol") as mock_transmit:
await entity.async_send_command(command)
await hass.async_block_till_done()
mock_transmit.assert_called_once()
call_args = mock_transmit.call_args
assert call_args[0][0] == 1 # key
cmd_json = json.loads(call_args[0][1])
assert cmd_json == expected_json
async def test_send_pulse_width_command(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device: MockESPHomeDeviceType,
) -> None:
"""Test sending pulse-width command via native API."""
entity_info = [_create_infrared_proxy_info()]
await mock_esphome_device(
mock_client=mock_client,
entity_info=entity_info,
user_service=[],
states=[],
)
await hass.async_block_till_done()
entities = async_get_entities(hass)
assert len(entities) == 1
entity = entities[0]
protocol = PulseWidthIRProtocol(
header=IRTiming(high_us=9000, low_us=4500),
one=IRTiming(high_us=560, low_us=1690),
zero=IRTiming(high_us=560, low_us=560),
footer=IRTiming(high_us=560, low_us=0),
frequency=38000,
msb_first=False,
)
command = PulseWidthIRCommand(
protocol=protocol, repeat_count=1, code=0x20DF10EF, length_in_bits=32
)
with patch.object(mock_client, "infrared_proxy_transmit") as mock_transmit:
await entity.async_send_command(command)
await hass.async_block_till_done()
mock_transmit.assert_called_once()
call_args = mock_transmit.call_args
assert call_args[0][0] == 1 # key
# Timing params should be second argument
timing = call_args[0][1]
assert timing.frequency == 38000
assert timing.length_in_bits == 32
# Data bytes should be third argument
data_bytes = call_args[0][2]
assert isinstance(data_bytes, bytes)
assert len(data_bytes) == 4
async def test_send_command_no_transmitter(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device: MockESPHomeDeviceType,
) -> None:
"""Test sending command to receiver-only device raises error."""
entity_info = [
_create_infrared_proxy_info(capabilities=InfraredProxyCapability.RECEIVER)
]
await mock_esphome_device(mock_client=mock_client, entity_info=entity_info)
await hass.async_block_till_done()
entities = async_get_entities(hass)
assert len(entities) == 1
entity = entities[0]
command = NECIRCommand(
protocol=NECIRProtocol(), repeat_count=1, address=0x04, command=0x08
)
with pytest.raises(HomeAssistantError):
await entity.async_send_command(command)
async def test_device_association(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device: MockESPHomeDeviceType,
entity_registry: er.EntityRegistry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test infrared entity is associated with ESPHome device."""
entity_info = [_create_infrared_proxy_info()]
await mock_esphome_device(mock_client=mock_client, entity_info=entity_info)
await hass.async_block_till_done()
device = device_registry.async_get_device(
connections={(dr.CONNECTION_NETWORK_MAC, "11:22:33:44:55:aa")}
)
assert device is not None
entry = entity_registry.async_get("infrared.test_my_remote")
assert entry is not None
assert entry.device_id == device.id
-431
View File
@@ -1,431 +0,0 @@
"""Test ESPHome infrared proxy remotes."""
from unittest.mock import patch
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, ServiceValidationError
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_protocol(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device,
) -> None:
"""Test sending protocol-based commands."""
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 protocol-based command
with patch.object(
mock_client, "infrared_proxy_transmit_protocol"
) as mock_transmit_protocol:
await hass.services.async_call(
REMOTE_DOMAIN,
"send_command",
{
"entity_id": "remote.test_my_remote",
"command": ['{"protocol": "NEC", "address": 4, "command": 8}'],
},
blocking=True,
)
await hass.async_block_till_done()
mock_transmit_protocol.assert_called_once_with(
1, '{"protocol": "NEC", "address": 4, "command": 8}'
)
async def test_infrared_proxy_send_command_pulse_width(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device,
) -> None:
"""Test sending pulse-width based commands."""
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 pulse-width command
with patch.object(mock_client, "infrared_proxy_transmit") as mock_transmit:
await hass.services.async_call(
REMOTE_DOMAIN,
"send_command",
{
"entity_id": "remote.test_my_remote",
"command": [
'{"timing": {"frequency": 38000, "length_in_bits": 32}, "data": [1, 2, 3, 4]}'
],
},
blocking=True,
)
await hass.async_block_till_done()
assert mock_transmit.call_count == 1
call_args = mock_transmit.call_args
assert call_args[0][0] == 1 # key
assert call_args[0][2] == b"\x01\x02\x03\x04" # decoded data
async def test_infrared_proxy_send_command_invalid_json(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device,
) -> None:
"""Test sending invalid JSON command."""
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 invalid JSON
with pytest.raises(
ServiceValidationError,
match="Command must be valid JSON",
):
await hass.services.async_call(
REMOTE_DOMAIN,
"send_command",
{"entity_id": "remote.test_my_remote", "command": ["not valid json"]},
blocking=True,
)
async def test_infrared_proxy_send_command_invalid_data_array(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device,
) -> None:
"""Test sending command with invalid data array."""
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 invalid data type (not an array)
with pytest.raises(
ServiceValidationError,
match="Data must be an array of integers",
):
await hass.services.async_call(
REMOTE_DOMAIN,
"send_command",
{
"entity_id": "remote.test_my_remote",
"command": ['{"timing": {"frequency": 38000}, "data": "not_an_array"}'],
},
blocking=True,
)
# Test invalid array values (out of range)
with pytest.raises(
ServiceValidationError,
match="Invalid data array",
):
await hass.services.async_call(
REMOTE_DOMAIN,
"send_command",
{
"entity_id": "remote.test_my_remote",
"command": ['{"timing": {"frequency": 38000}, "data": [1, 2, 300, 4]}'],
},
blocking=True,
)
async def test_infrared_proxy_send_command_no_transmitter(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device,
) -> None:
"""Test sending command to receiver-only device."""
entity_info = [
InfraredProxyInfo(
object_id="myremote",
key=1,
name="my remote",
capabilities=InfraredProxyCapability.RECEIVER, # No 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="does not support infrared transmission",
):
await hass.services.async_call(
REMOTE_DOMAIN,
"send_command",
{
"entity_id": "remote.test_my_remote",
"command": ['{"protocol": "NEC", "address": 4, "command": 8}'],
},
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,
)
+1
View File
@@ -0,0 +1 @@
"""Tests for the Infrared integration."""
+49
View File
@@ -0,0 +1,49 @@
"""Common fixtures for the Infrared tests."""
from __future__ import annotations
import pytest
from homeassistant.components.infrared import (
BaseIRCommand,
InfraredEntity,
InfraredEntityFeature,
IRProtocolType,
)
from homeassistant.components.infrared.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
@pytest.fixture
async def init_integration(hass: HomeAssistant) -> None:
"""Set up the Infrared integration for testing."""
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
class MockInfraredEntity(InfraredEntity):
"""Mock infrared entity for testing."""
_attr_has_entity_name = True
_attr_name = "Test IR transmitter"
def __init__(self, unique_id: str) -> None:
"""Initialize mock entity."""
self._attr_unique_id = unique_id
self._attr_supported_features = InfraredEntityFeature.TRANSMIT
self._attr_supported_protocols = {
IRProtocolType.PULSE_WIDTH,
IRProtocolType.NEC,
}
self.send_command_calls: list[BaseIRCommand] = []
async def async_send_command(self, command: BaseIRCommand) -> None:
"""Mock send command."""
self.send_command_calls.append(command)
@pytest.fixture
def mock_infrared_entity() -> MockInfraredEntity:
"""Return a mock infrared entity."""
return MockInfraredEntity("test_ir_transmitter")
+94
View File
@@ -0,0 +1,94 @@
"""Tests for the Infrared integration setup."""
from homeassistant.components.infrared import (
DATA_COMPONENT,
DOMAIN,
InfraredEntityFeature,
IRProtocolType,
NECIRCommand,
NECIRProtocol,
async_get_entities,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .conftest import MockInfraredEntity
async def test_setup(hass: HomeAssistant) -> None:
"""Test Infrared integration setup."""
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
# Verify the component is loaded
assert DATA_COMPONENT in hass.data
async def test_get_entities_empty(hass: HomeAssistant) -> None:
"""Test getting entities when none are registered."""
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
entities = async_get_entities(hass)
assert entities == []
async def test_get_entities_filter_by_protocol(
hass: HomeAssistant,
init_integration: None,
mock_infrared_entity: MockInfraredEntity,
) -> None:
"""Test filtering entities by protocol support."""
# Add the mock entity to the component
component = hass.data[DATA_COMPONENT]
await component.async_add_entities([mock_infrared_entity])
# Get all entities
all_entities = async_get_entities(hass)
assert len(all_entities) == 1
assert all_entities[0] is mock_infrared_entity
# Filter by NEC protocol (should match)
nec_entities = async_get_entities(hass, protocols=[IRProtocolType.NEC])
assert len(nec_entities) == 1
# Filter by Samsung protocol (should not match since mock only supports NEC and PULSE_WIDTH)
samsung_entities = async_get_entities(hass, protocols=[IRProtocolType.SAMSUNG])
assert len(samsung_entities) == 0
async def test_infrared_entity_send_command(
hass: HomeAssistant,
init_integration: None,
mock_infrared_entity: MockInfraredEntity,
) -> None:
"""Test sending command via infrared entity."""
# Add the mock entity to the component
component = hass.data[DATA_COMPONENT]
await component.async_add_entities([mock_infrared_entity])
# Create a test command
command = NECIRCommand(
protocol=NECIRProtocol(),
repeat_count=1,
address=0x04FB,
command=0x08F7,
)
# Send command
await mock_infrared_entity.async_send_command(command)
# Verify command was recorded
assert len(mock_infrared_entity.send_command_calls) == 1
assert mock_infrared_entity.send_command_calls[0] is command
async def test_infrared_entity_features(
hass: HomeAssistant,
init_integration: None,
mock_infrared_entity: MockInfraredEntity,
) -> None:
"""Test infrared entity features property."""
assert mock_infrared_entity.supported_features == InfraredEntityFeature.TRANSMIT
assert IRProtocolType.NEC in mock_infrared_entity.supported_protocols
assert IRProtocolType.PULSE_WIDTH in mock_infrared_entity.supported_protocols
+101
View File
@@ -0,0 +1,101 @@
"""Tests for the Infrared protocol definitions."""
import pytest
from homeassistant.components.infrared import (
IRProtocolType,
IRTiming,
NECIRCommand,
NECIRProtocol,
SamsungIRCommand,
SamsungIRProtocol,
)
def test_nec_protocol_pulse_width_compat() -> None:
"""Test NEC protocol conversion to pulse-width compatible format."""
protocol = NECIRProtocol()
compat = protocol.get_pulse_width_compat_protocol()
# Verify timing values match NEC standard
assert compat.header.high_us == 9000
assert compat.header.low_us == 4500
assert compat.one.high_us == 560
assert compat.one.low_us == 1690
assert compat.zero.high_us == 560
assert compat.zero.low_us == 560
assert compat.footer.high_us == 560
assert compat.footer.low_us == 0
assert compat.frequency == 38000
assert compat.msb_first is False
assert compat.minimum_idle_time_us == 40000
def test_samsung_protocol_pulse_width_compat() -> None:
"""Test Samsung protocol conversion to pulse-width compatible format."""
protocol = SamsungIRProtocol()
compat = protocol.get_pulse_width_compat_protocol()
# Verify timing values match Samsung standard
assert compat.header.high_us == 4500
assert compat.header.low_us == 4500
assert compat.one.high_us == 560
assert compat.one.low_us == 1690
assert compat.zero.high_us == 560
assert compat.zero.low_us == 560
assert compat.frequency == 38000
def test_nec_command_pulse_width_compat_code() -> None:
"""Test NEC command code conversion to pulse-width format."""
command = NECIRCommand(
protocol=NECIRProtocol(),
repeat_count=1,
address=0x04FB, # 16-bit address
command=0x08F7, # 16-bit command
)
# Code should be: address | (command << 16)
expected_code = 0x04FB | (0x08F7 << 16)
assert command.get_pulse_width_compat_code() == expected_code
def test_samsung_command_pulse_width_compat_code() -> None:
"""Test Samsung command code conversion (should be passthrough)."""
command = SamsungIRCommand(
protocol=SamsungIRProtocol(),
repeat_count=1,
code=0xE0E040BF,
length_in_bits=32,
)
# Samsung code should pass through unchanged
assert command.get_pulse_width_compat_code() == 0xE0E040BF
def test_ir_timing_frozen() -> None:
"""Test that IRTiming is immutable."""
timing = IRTiming(high_us=9000, low_us=4500)
with pytest.raises(AttributeError):
timing.high_us = 1000 # type: ignore[misc]
def test_nec_command_frozen() -> None:
"""Test that NECIRCommand is immutable."""
command = NECIRCommand(
protocol=NECIRProtocol(),
repeat_count=1,
address=0x04FB,
command=0x08F7,
)
with pytest.raises(AttributeError):
command.address = 0x0000 # type: ignore[misc]
def test_protocol_types() -> None:
"""Test protocol type enum values."""
assert IRProtocolType.PULSE_WIDTH == "pulse_width"
assert IRProtocolType.NEC == "nec"
assert IRProtocolType.SAMSUNG == "samsung"