Compare commits

...

2 Commits

Author SHA1 Message Date
abmantis
6c610dfe73 Add infrared platform to ESPHome 2026-02-05 20:23:19 +00:00
abmantis
90bacbb98e Add infrared entity integration 2026-02-05 20:06:33 +00:00
19 changed files with 937 additions and 0 deletions

View File

@@ -282,6 +282,7 @@ homeassistant.components.imgw_pib.*
homeassistant.components.immich.*
homeassistant.components.incomfort.*
homeassistant.components.inels.*
homeassistant.components.infrared.*
homeassistant.components.input_button.*
homeassistant.components.input_select.*
homeassistant.components.input_text.*

2
CODEOWNERS generated
View File

@@ -782,6 +782,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

View File

@@ -29,6 +29,7 @@ from aioesphomeapi import (
Event,
EventInfo,
FanInfo,
InfraredInfo,
LightInfo,
LockInfo,
MediaPlayerInfo,
@@ -85,6 +86,7 @@ INFO_TYPE_TO_PLATFORM: dict[type[EntityInfo], Platform] = {
DateTimeInfo: Platform.DATETIME,
EventInfo: Platform.EVENT,
FanInfo: Platform.FAN,
InfraredInfo: Platform.INFRARED,
LightInfo: Platform.LIGHT,
LockInfo: Platform.LOCK,
MediaPlayerInfo: Platform.MEDIA_PLAYER,
@@ -189,6 +191,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:
@@ -520,6 +523,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,

View File

@@ -0,0 +1,98 @@
"""Infrared platform for ESPHome."""
from __future__ import annotations
import logging
from aioesphomeapi import EntityInfo, EntityState, InfraredCapability, InfraredInfo
from homeassistant.components.infrared import InfraredCommand, InfraredEntity
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_platform
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN
from .entity import EsphomeEntity, async_static_info_updated
from .entry_data import ESPHomeConfigEntry
_LOGGER = logging.getLogger(__name__)
PARALLEL_UPDATES = 0
class EsphomeInfraredEntity(EsphomeEntity[InfraredInfo, EntityState], InfraredEntity):
"""ESPHome infrared entity using native API."""
_attr_has_entity_name = True
_attr_name = "Infrared Transmitter"
@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: InfraredCommand) -> None:
"""Send an IR command.
Raises:
HomeAssistantError: If transmission fails.
"""
timings = [
interval
for timing in command.get_raw_timings()
for interval in (timing.high_us, -timing.low_us)
]
_LOGGER.debug("Sending command: %s", timings)
try:
self._client.infrared_rf_transmit_raw_timings(
self._static_info.key, carrier_frequency=38000, timings=timings
)
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_setup_entry(
hass: HomeAssistant,
entry: ESPHomeConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up ESPHome infrared entities, filtering out receiver-only devices."""
entry_data = entry.runtime_data
entry_data.info[InfraredInfo] = {}
platform = entity_platform.async_get_current_platform()
def filtered_static_info_update(infos: list[EntityInfo]) -> None:
transmiter_infos: list[EntityInfo] = [
info
for info in infos
if isinstance(info, InfraredInfo)
and info.capabilities & InfraredCapability.TRANSMITTER
]
async_static_info_updated(
hass,
entry_data,
platform,
async_add_entities,
InfraredInfo,
EsphomeInfraredEntity,
EntityState,
transmiter_infos,
)
entry_data.cleanup_callbacks.append(
entry_data.async_register_static_info_callback(
InfraredInfo, filtered_static_info_update
)
)

View File

@@ -17,6 +17,8 @@ from aioesphomeapi import (
EncryptionPlaintextAPIError,
ExecuteServiceResponse,
HomeassistantServiceCall,
InfraredCapability,
InfraredInfo,
InvalidAuthAPIError,
InvalidEncryptionKeyAPIError,
LogLevel,
@@ -692,6 +694,15 @@ class ESPHomeManager:
cli.subscribe_zwave_proxy_request(self._async_zwave_proxy_request)
)
if any(
isinstance(info, InfraredInfo)
and info.capabilities & InfraredCapability.RECEIVER
for info in entity_infos
):
entry_data.disconnect_callbacks.add(
cli.subscribe_infrared_rf_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 +733,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

View File

@@ -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."
},

View File

@@ -0,0 +1,156 @@
"""Provides functionality to interact with infrared devices."""
from __future__ import annotations
from abc import abstractmethod
from datetime import datetime, timedelta
import logging
from typing import final
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import Context, HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv, entity_registry as er
from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import ConfigType
from homeassistant.util import dt as dt_util
from homeassistant.util.hass_dict import HassKey
from .const import DOMAIN
from .protocols import InfraredCommand, NECInfraredCommand, Timing
__all__ = [
"DOMAIN",
"InfraredCommand",
"InfraredEntity",
"InfraredEntityDescription",
"NECInfraredCommand",
"Timing",
"async_get_emitters",
"async_send_command",
]
_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_emitters(hass: HomeAssistant) -> list[InfraredEntity]:
"""Get all infrared emitters."""
component = hass.data.get(DATA_COMPONENT)
if component is None:
return []
return list(component.entities)
async def async_send_command(
hass: HomeAssistant,
entity_uuid: str,
command: InfraredCommand,
context: Context | None = None,
) -> None:
"""Send an IR command to the specified infrared entity.
Raises:
HomeAssistantError: If the infrared entity is not found.
"""
component = hass.data.get(DATA_COMPONENT)
if component is None:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="component_not_loaded",
)
ent_reg = er.async_get(hass)
entity_id = er.async_validate_entity_id(ent_reg, entity_uuid)
entity = component.get_entity(entity_id)
if entity is None:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="entity_not_found",
translation_placeholders={"entity_id": entity_id},
)
if context is not None:
entity.async_set_context(context)
await entity.async_send_command_internal(command)
class InfraredEntityDescription(EntityDescription, frozen_or_thawed=True):
"""Describes infrared entities."""
class InfraredEntity(RestoreEntity):
"""Base class for infrared transmitter entities."""
entity_description: InfraredEntityDescription
_attr_should_poll = False
_attr_state: None
__last_command_sent: datetime | None = None
@property
@final
def state(self) -> str | None:
"""Return the entity state."""
if (last_command := self.__last_command_sent) is None:
return None
return last_command.isoformat(timespec="milliseconds")
@final
async def async_send_command_internal(self, command: InfraredCommand) -> None:
"""Send an IR command and update state.
Should not be overridden, handles setting last sent timestamp.
"""
await self.async_send_command(command)
self.__last_command_sent = dt_util.utcnow()
self.async_write_ha_state()
@final
async def async_internal_added_to_hass(self) -> None:
"""Call when the infrared entity is added to hass."""
await super().async_internal_added_to_hass()
state = await self.async_get_last_state()
if state is not None and state.state is not None:
self.__last_command_sent = dt_util.parse_datetime(state.state)
@abstractmethod
async def async_send_command(self, command: InfraredCommand) -> None:
"""Send an IR command.
Args:
command: The IR command to send.
Raises:
HomeAssistantError: If transmission fails.
"""

View File

@@ -0,0 +1,5 @@
"""Constants for the Infrared integration."""
from typing import Final
DOMAIN: Final = "infrared"

View File

@@ -0,0 +1,7 @@
{
"entity_component": {
"_": {
"default": "mdi:led-on"
}
}
}

View File

@@ -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"
}

View File

@@ -0,0 +1,119 @@
"""IR protocol definitions for the Infrared integration."""
import abc
from dataclasses import dataclass
from typing import override
@dataclass(frozen=True, slots=True)
class Timing:
"""High/low signal timing."""
high_us: int
low_us: int
class InfraredCommand(abc.ABC):
"""Base class for IR commands."""
repeat_count: int
modulation: int
def __init__(self, *, modulation: int, repeat_count: int = 0) -> None:
"""Initialize the IR command."""
self.modulation = modulation
self.repeat_count = repeat_count
@abc.abstractmethod
def get_raw_timings(self) -> list[Timing]:
"""Get raw timings for the command."""
class NECInfraredCommand(InfraredCommand):
"""NEC IR command."""
address: int
command: int
def __init__(
self, *, address: int, command: int, modulation: int, repeat_count: int = 0
) -> None:
"""Initialize the NEC IR command."""
super().__init__(modulation=modulation, repeat_count=repeat_count)
self.address = address
self.command = command
@override
def get_raw_timings(self) -> list[Timing]:
"""Get raw timings for the NEC command.
NEC protocol timing (in microseconds):
- Leader pulse: 9000µs high, 4500µs low
- Logical '0': 562µs high, 562µs low
- Logical '1': 562µs high, 1687µs low
- End pulse: 562µs high
- Repeat code: 9000µs high, 2250µs low, 562µs end pulse
- Frame gap: ~96ms between end pulse and next frame (total frame ~108ms)
Data format (32 bits, LSB first):
- Standard NEC: address (8-bit) + ~address (8-bit) + command (8-bit) + ~command (8-bit)
- Extended NEC: address_low (8-bit) + address_high (8-bit) + command (8-bit) + ~command (8-bit)
"""
# NEC timing constants (microseconds)
leader_high = 9000
leader_low = 4500
bit_high = 562
zero_low = 562
one_low = 1687
repeat_low = 2250
frame_gap = 96000 # Gap to make total frame ~108ms
timings: list[Timing] = [Timing(high_us=leader_high, low_us=leader_low)]
# Determine if standard (8-bit) or extended (16-bit) address
if self.address <= 0xFF:
# Standard NEC: address + inverted address
address_low = self.address & 0xFF
address_high = (~self.address) & 0xFF
else:
# Extended NEC: 16-bit address (no inversion)
address_low = self.address & 0xFF
address_high = (self.address >> 8) & 0xFF
command_byte = self.command & 0xFF
command_inverted = (~self.command) & 0xFF
# Build 32-bit command data (LSB first in transmission)
data = (
address_low
| (address_high << 8)
| (command_byte << 16)
| (command_inverted << 24)
)
for _ in range(32):
bit = data & 1
if bit:
timings.append(Timing(high_us=bit_high, low_us=one_low))
else:
timings.append(Timing(high_us=bit_high, low_us=zero_low))
data >>= 1
# End pulse
timings.append(Timing(high_us=bit_high, low_us=0))
# Add repeat codes if requested
for _ in range(self.repeat_count):
# Replace the last timing's low_us with the frame gap
last_timing = timings[-1]
timings[-1] = Timing(high_us=last_timing.high_us, low_us=frame_gap)
# Repeat code: leader burst + shorter space + end pulse
timings.extend(
[
Timing(high_us=leader_high, low_us=repeat_low),
Timing(high_us=bit_high, low_us=0),
]
)
return timings

View File

@@ -0,0 +1,10 @@
{
"exceptions": {
"component_not_loaded": {
"message": "Infrared component not loaded"
},
"entity_not_found": {
"message": "Infrared entity `{entity_id}` not found"
}
}
}

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"

10
mypy.ini generated
View File

@@ -2576,6 +2576,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.infrared.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.input_button.*]
check_untyped_defs = true
disallow_incomplete_defs = true

View File

@@ -0,0 +1,166 @@
"""Test ESPHome infrared platform."""
from aioesphomeapi import APIClient, InfraredCapability, InfraredInfo
import pytest
from homeassistant.components import infrared
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from .conftest import MockESPHomeDeviceType
ENTITY_ID = "infrared.test_ir"
async def _mock_ir_device(
mock_esphome_device: MockESPHomeDeviceType,
mock_client: APIClient,
capabilities: InfraredCapability = InfraredCapability.TRANSMITTER,
) -> MockESPHomeDeviceType:
entity_info = [
InfraredInfo(object_id="ir", key=1, name="IR", capabilities=capabilities)
]
return await mock_esphome_device(
mock_client=mock_client, entity_info=entity_info, states=[]
)
@pytest.mark.parametrize(
("capabilities", "entity_created"),
[
(InfraredCapability.TRANSMITTER, True),
(InfraredCapability.RECEIVER, False),
(InfraredCapability.TRANSMITTER | InfraredCapability.RECEIVER, True),
(0, False),
],
)
async def test_infrared_entity_transmitter(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device: MockESPHomeDeviceType,
capabilities: InfraredCapability,
entity_created: bool,
) -> None:
"""Test infrared entity with transmitter capability is created."""
await _mock_ir_device(mock_esphome_device, mock_client, capabilities)
state = hass.states.get(ENTITY_ID)
assert (state is not None) == entity_created
emitters = infrared.async_get_emitters(hass)
assert (len(emitters) == 1) == entity_created
async def test_infrared_multiple_entities_mixed_capabilities(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device: MockESPHomeDeviceType,
) -> None:
"""Test multiple infrared entities with mixed capabilities."""
entity_info = [
InfraredInfo(
object_id="ir_transmitter",
key=1,
name="IR Transmitter",
capabilities=InfraredCapability.TRANSMITTER,
),
InfraredInfo(
object_id="ir_receiver",
key=2,
name="IR Receiver",
capabilities=InfraredCapability.RECEIVER,
),
InfraredInfo(
object_id="ir_transceiver",
key=3,
name="IR Transceiver",
capabilities=InfraredCapability.TRANSMITTER | InfraredCapability.RECEIVER,
),
]
await mock_esphome_device(
mock_client=mock_client,
entity_info=entity_info,
states=[],
)
# Only transmitter and transceiver should be created
assert hass.states.get("infrared.test_ir_transmitter") is not None
assert hass.states.get("infrared.test_ir_receiver") is None
assert hass.states.get("infrared.test_ir_transceiver") is not None
emitters = infrared.async_get_emitters(hass)
assert len(emitters) == 2
async def test_infrared_send_command_success(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device: MockESPHomeDeviceType,
) -> None:
"""Test sending IR command successfully."""
await _mock_ir_device(mock_esphome_device, mock_client)
command = infrared.NECInfraredCommand(address=0x04, command=0x08, modulation=38000)
await infrared.async_send_command(hass, ENTITY_ID, command)
# Verify the command was sent to the ESPHome client
mock_client.infrared_rf_transmit_raw_timings.assert_called_once()
call_args = mock_client.infrared_rf_transmit_raw_timings.call_args
assert call_args[0][0] == 1 # key
assert call_args[1]["carrier_frequency"] == 38000
# Verify timings (alternating positive/negative values)
timings = call_args[1]["timings"]
assert len(timings) > 0
for i in range(0, len(timings), 2):
assert timings[i] >= 0
for i in range(1, len(timings), 2):
assert timings[i] <= 0
async def test_infrared_send_command_failure(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device: MockESPHomeDeviceType,
) -> None:
"""Test sending IR command with failure raises HomeAssistantError."""
await _mock_ir_device(mock_esphome_device, mock_client)
mock_client.infrared_rf_transmit_raw_timings.side_effect = Exception(
"Connection lost"
)
command = infrared.NECInfraredCommand(address=0x04, command=0x08, modulation=38000)
with pytest.raises(HomeAssistantError) as exc_info:
await infrared.async_send_command(hass, ENTITY_ID, command)
assert exc_info.value.translation_domain == "esphome"
assert exc_info.value.translation_key == "error_sending_ir_command"
async def test_infrared_entity_availability(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device: MockESPHomeDeviceType,
) -> None:
"""Test infrared entity becomes available after device reconnects."""
mock_device = await _mock_ir_device(mock_esphome_device, mock_client)
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state != STATE_UNAVAILABLE
await mock_device.mock_disconnect(False)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state == STATE_UNAVAILABLE
await mock_device.mock_connect()
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state != STATE_UNAVAILABLE

View File

@@ -0,0 +1 @@
"""Tests for the Infrared integration."""

View File

@@ -0,0 +1,37 @@
"""Common fixtures for the Infrared tests."""
import pytest
from homeassistant.components.infrared import InfraredCommand, InfraredEntity
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.send_command_calls: list[InfraredCommand] = []
async def async_send_command(self, command: InfraredCommand) -> 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")

View File

@@ -0,0 +1,146 @@
"""Tests for the Infrared integration setup."""
from unittest.mock import AsyncMock
from freezegun.api import FrozenDateTimeFactory
import pytest
from homeassistant.components.infrared import (
DATA_COMPONENT,
DOMAIN,
NECInfraredCommand,
async_get_emitters,
async_send_command,
)
from homeassistant.const import STATE_UNKNOWN
from homeassistant.core import HomeAssistant, State
from homeassistant.exceptions import HomeAssistantError
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from .conftest import MockInfraredEntity
from tests.common import mock_restore_cache
async def test_get_entities_integration_setup(hass: HomeAssistant) -> None:
"""Test getting entities when the integration is not setup."""
assert async_get_emitters(hass) == []
@pytest.mark.usefixtures("init_integration")
async def test_get_entities_empty(hass: HomeAssistant) -> None:
"""Test getting entities when none are registered."""
assert async_get_emitters(hass) == []
@pytest.mark.usefixtures("init_integration")
async def test_infrared_entity_initial_state(
hass: HomeAssistant, mock_infrared_entity: MockInfraredEntity
) -> None:
"""Test infrared entity has no state before any command is sent."""
component = hass.data[DATA_COMPONENT]
await component.async_add_entities([mock_infrared_entity])
state = hass.states.get("infrared.test_ir_transmitter")
assert state is not None
assert state.state == STATE_UNKNOWN
@pytest.mark.usefixtures("init_integration")
async def test_async_send_command_success(
hass: HomeAssistant,
mock_infrared_entity: MockInfraredEntity,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test sending command via async_send_command helper."""
# Add the mock entity to the component
component = hass.data[DATA_COMPONENT]
await component.async_add_entities([mock_infrared_entity])
# Freeze time so we can verify the state update
now = dt_util.utcnow()
freezer.move_to(now)
command = NECInfraredCommand(address=0x04FB, command=0x08F7, modulation=38000)
await async_send_command(hass, mock_infrared_entity.entity_id, command)
assert len(mock_infrared_entity.send_command_calls) == 1
assert mock_infrared_entity.send_command_calls[0] is command
state = hass.states.get("infrared.test_ir_transmitter")
assert state is not None
assert state.state == now.isoformat(timespec="milliseconds")
@pytest.mark.usefixtures("init_integration")
async def test_async_send_command_error_does_not_update_state(
hass: HomeAssistant,
mock_infrared_entity: MockInfraredEntity,
) -> None:
"""Test that state is not updated when async_send_command raises an error."""
component = hass.data[DATA_COMPONENT]
await component.async_add_entities([mock_infrared_entity])
state = hass.states.get("infrared.test_ir_transmitter")
assert state is not None
assert state.state == STATE_UNKNOWN
command = NECInfraredCommand(address=0x04FB, command=0x08F7, modulation=38000)
mock_infrared_entity.async_send_command = AsyncMock(
side_effect=HomeAssistantError("Transmission failed")
)
with pytest.raises(HomeAssistantError, match="Transmission failed"):
await async_send_command(hass, mock_infrared_entity.entity_id, command)
# Verify state was not updated after the error
state = hass.states.get("infrared.test_ir_transmitter")
assert state is not None
assert state.state == STATE_UNKNOWN
@pytest.mark.usefixtures("init_integration")
async def test_async_send_command_entity_not_found(hass: HomeAssistant) -> None:
"""Test async_send_command raises error when entity not found."""
command = NECInfraredCommand(
address=0x04FB, command=0x08F7, modulation=38000, repeat_count=1
)
with pytest.raises(
HomeAssistantError,
match="Infrared entity `infrared.nonexistent_entity` not found",
):
await async_send_command(hass, "infrared.nonexistent_entity", command)
async def test_async_send_command_component_not_loaded(hass: HomeAssistant) -> None:
"""Test async_send_command raises error when component not loaded."""
command = NECInfraredCommand(
address=0x04FB, command=0x08F7, modulation=38000, repeat_count=1
)
with pytest.raises(HomeAssistantError, match="component_not_loaded"):
await async_send_command(hass, "infrared.some_entity", command)
async def test_infrared_entity_state_restore(
hass: HomeAssistant,
mock_infrared_entity: MockInfraredEntity,
) -> None:
"""Test infrared entity restores state from previous session."""
previous_timestamp = "2026-01-01T12:00:00.000+00:00"
mock_restore_cache(
hass, [State("infrared.test_ir_transmitter", previous_timestamp)]
)
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
component = hass.data[DATA_COMPONENT]
await component.async_add_entities([mock_infrared_entity])
state = hass.states.get("infrared.test_ir_transmitter")
assert state is not None
assert state.state == previous_timestamp

View File

@@ -0,0 +1,128 @@
"""Tests for the Infrared protocol definitions."""
from homeassistant.components.infrared import NECInfraredCommand, Timing
def test_nec_command_get_raw_timings_standard() -> None:
"""Test NEC command raw timings generation for standard 8-bit address."""
expected_raw_timings = [
Timing(high_us=9000, low_us=4500),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=0),
]
command = NECInfraredCommand(
address=0x04, command=0x08, modulation=38000, repeat_count=0
)
timings = command.get_raw_timings()
assert timings == expected_raw_timings
# Same command now with 2 repeats
command_with_repeats = NECInfraredCommand(
address=command.address,
command=command.command,
modulation=command.modulation,
repeat_count=2,
)
timings_with_repeats = command_with_repeats.get_raw_timings()
assert timings_with_repeats == [
*expected_raw_timings[:-1],
Timing(high_us=562, low_us=96000),
Timing(high_us=9000, low_us=2250),
Timing(high_us=562, low_us=96000),
Timing(high_us=9000, low_us=2250),
Timing(high_us=562, low_us=0),
]
def test_nec_command_get_raw_timings_extended() -> None:
"""Test NEC command raw timings generation for extended 16-bit address."""
expected_raw_timings = [
Timing(high_us=9000, low_us=4500),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=562),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=1687),
Timing(high_us=562, low_us=0),
]
command = NECInfraredCommand(
address=0x04FB, command=0x08, modulation=38000, repeat_count=0
)
timings = command.get_raw_timings()
assert timings == expected_raw_timings
# Same command now with 2 repeats
command_with_repeats = NECInfraredCommand(
address=command.address,
command=command.command,
modulation=command.modulation,
repeat_count=2,
)
timings_with_repeats = command_with_repeats.get_raw_timings()
assert timings_with_repeats == [
*expected_raw_timings[:-1],
Timing(high_us=562, low_us=96000),
Timing(high_us=9000, low_us=2250),
Timing(high_us=562, low_us=96000),
Timing(high_us=9000, low_us=2250),
Timing(high_us=562, low_us=0),
]