diff --git a/CODEOWNERS b/CODEOWNERS index 629bed98fa49..e92435935e7d 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -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 diff --git a/homeassistant/components/esphome/entry_data.py b/homeassistant/components/esphome/entry_data.py index 776fa0817b4b..f2bf956ec96a 100644 --- a/homeassistant/components/esphome/entry_data.py +++ b/homeassistant/components/esphome/entry_data.py @@ -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, diff --git a/homeassistant/components/esphome/infrared.py b/homeassistant/components/esphome/infrared.py new file mode 100644 index 000000000000..c29c1705de2d --- /dev/null +++ b/homeassistant/components/esphome/infrared.py @@ -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, +) diff --git a/homeassistant/components/esphome/remote.py b/homeassistant/components/esphome/remote.py deleted file mode 100644 index adeb6a257042..000000000000 --- a/homeassistant/components/esphome/remote.py +++ /dev/null @@ -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, -) diff --git a/homeassistant/components/esphome/strings.json b/homeassistant/components/esphome/strings.json index 8755db8e840f..5356e0826a03 100644 --- a/homeassistant/components/esphome/strings.json +++ b/homeassistant/components/esphome/strings.json @@ -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." }, diff --git a/homeassistant/components/infrared/__init__.py b/homeassistant/components/infrared/__init__.py new file mode 100644 index 000000000000..82f1af9cfba8 --- /dev/null +++ b/homeassistant/components/infrared/__init__.py @@ -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. + """ diff --git a/homeassistant/components/infrared/const.py b/homeassistant/components/infrared/const.py new file mode 100644 index 000000000000..ea3da88d4ff7 --- /dev/null +++ b/homeassistant/components/infrared/const.py @@ -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.""" diff --git a/homeassistant/components/infrared/manifest.json b/homeassistant/components/infrared/manifest.json new file mode 100644 index 000000000000..bebda5e5ec75 --- /dev/null +++ b/homeassistant/components/infrared/manifest.json @@ -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" +} diff --git a/homeassistant/components/infrared/protocols.py b/homeassistant/components/infrared/protocols.py new file mode 100644 index 000000000000..d97ed7c25c34 --- /dev/null +++ b/homeassistant/components/infrared/protocols.py @@ -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 diff --git a/homeassistant/components/infrared/strings.json b/homeassistant/components/infrared/strings.json new file mode 100644 index 000000000000..c7d4ec1a337f --- /dev/null +++ b/homeassistant/components/infrared/strings.json @@ -0,0 +1,7 @@ +{ + "exceptions": { + "send_command_failed": { + "message": "Failed to send IR command: {error}" + } + } +} diff --git a/homeassistant/generated/entity_platforms.py b/homeassistant/generated/entity_platforms.py index 7010ffc9be73..718c3745be89 100644 --- a/homeassistant/generated/entity_platforms.py +++ b/homeassistant/generated/entity_platforms.py @@ -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" diff --git a/script/hassfest/quality_scale.py b/script/hassfest/quality_scale.py index ebcbb798f6d3..bc3bcab0d222 100644 --- a/script/hassfest/quality_scale.py +++ b/script/hassfest/quality_scale.py @@ -2171,6 +2171,7 @@ NO_QUALITY_SCALE = [ "input_text", "intent_script", "intent", + "infrared", "labs", "logbook", "logger", diff --git a/tests/components/esphome/test_infrared.py b/tests/components/esphome/test_infrared.py new file mode 100644 index 000000000000..fa6424faeff3 --- /dev/null +++ b/tests/components/esphome/test_infrared.py @@ -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 diff --git a/tests/components/esphome/test_remote.py b/tests/components/esphome/test_remote.py deleted file mode 100644 index be49e506b851..000000000000 --- a/tests/components/esphome/test_remote.py +++ /dev/null @@ -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, - ) diff --git a/tests/components/infrared/__init__.py b/tests/components/infrared/__init__.py new file mode 100644 index 000000000000..f5712a639f4b --- /dev/null +++ b/tests/components/infrared/__init__.py @@ -0,0 +1 @@ +"""Tests for the Infrared integration.""" diff --git a/tests/components/infrared/conftest.py b/tests/components/infrared/conftest.py new file mode 100644 index 000000000000..4cd181401a16 --- /dev/null +++ b/tests/components/infrared/conftest.py @@ -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") diff --git a/tests/components/infrared/test_init.py b/tests/components/infrared/test_init.py new file mode 100644 index 000000000000..009eeedbc26d --- /dev/null +++ b/tests/components/infrared/test_init.py @@ -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 diff --git a/tests/components/infrared/test_protocols.py b/tests/components/infrared/test_protocols.py new file mode 100644 index 000000000000..65254b560dd7 --- /dev/null +++ b/tests/components/infrared/test_protocols.py @@ -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"