From c37ca31becae25e3bcb7a499f3cd4b9963cd02ca Mon Sep 17 00:00:00 2001 From: kbx81 Date: Mon, 15 Dec 2025 23:31:21 -0600 Subject: [PATCH] json --- homeassistant/components/esphome/remote.py | 95 ++++++++- homeassistant/components/esphome/strings.json | 3 + tests/components/esphome/test_remote.py | 200 +++++++++++++++++- 3 files changed, 283 insertions(+), 15 deletions(-) diff --git a/homeassistant/components/esphome/remote.py b/homeassistant/components/esphome/remote.py index 5ad9cbc7ea9..adeb6a25704 100644 --- a/homeassistant/components/esphome/remote.py +++ b/homeassistant/components/esphome/remote.py @@ -4,6 +4,7 @@ from __future__ import annotations from collections.abc import Iterable from functools import partial +import json import logging from typing import Any @@ -12,12 +13,14 @@ from aioesphomeapi import ( EntityState, InfraredProxyCapability, InfraredProxyInfo, + InfraredProxyTimingParams, ) from homeassistant.components.remote import RemoteEntity, RemoteEntityFeature from homeassistant.core import callback -from homeassistant.exceptions import HomeAssistantError +from homeassistant.exceptions import HomeAssistantError, ServiceValidationError +from .const import DOMAIN from .entity import EsphomeEntity, platform_async_setup_entry _LOGGER = logging.getLogger(__name__) @@ -67,13 +70,89 @@ class EsphomeInfraredProxy(EsphomeEntity[InfraredProxyInfo, EntityState], Remote _LOGGER.debug("Turn off called for %s (no-op)", self.name) async def async_send_command(self, command: Iterable[str], **kwargs: Any) -> None: - """Send commands to a device.""" - # This method would need to parse command data and timing parameters - # For now, we'll raise an error as this requires more complex implementation - raise HomeAssistantError( - "Direct command sending not yet implemented for ESPHome infrared proxy. " - "Use the infrared_proxy_transmit service instead." - ) + """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.""" diff --git a/homeassistant/components/esphome/strings.json b/homeassistant/components/esphome/strings.json index 9abcec6df96..8755db8e840 100644 --- a/homeassistant/components/esphome/strings.json +++ b/homeassistant/components/esphome/strings.json @@ -140,6 +140,9 @@ "error_uploading": { "message": "Error during OTA (Over-The-Air) update of {configuration}. Try again in ESPHome dashboard for more information." }, + "infrared_proxy_transmitter_not_supported": { + "message": "Device does not support infrared transmission" + }, "ota_in_progress": { "message": "An OTA (Over-The-Air) update is already in progress for {configuration}." } diff --git a/tests/components/esphome/test_remote.py b/tests/components/esphome/test_remote.py index 7bad3734d9c..be49e506b85 100644 --- a/tests/components/esphome/test_remote.py +++ b/tests/components/esphome/test_remote.py @@ -1,5 +1,7 @@ """Test ESPHome infrared proxy remotes.""" +from unittest.mock import patch + from aioesphomeapi import ( APIClient, InfraredProxyCapability, @@ -11,7 +13,7 @@ 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 +from homeassistant.exceptions import HomeAssistantError, ServiceValidationError async def test_infrared_proxy_transmitter_only( @@ -171,12 +173,12 @@ async def test_infrared_proxy_receive_event( assert "entry_id" in event_data -async def test_infrared_proxy_send_command_not_implemented( +async def test_infrared_proxy_send_command_protocol( hass: HomeAssistant, mock_client: APIClient, mock_esphome_device, ) -> None: - """Test that send_command raises appropriate error.""" + """Test sending protocol-based commands.""" entity_info = [ InfraredProxyInfo( object_id="myremote", @@ -195,15 +197,199 @@ async def test_infrared_proxy_send_command_not_implemented( ) await hass.async_block_till_done() - # Test send_command raises error + # 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( - HomeAssistantError, - match="Direct command sending not yet implemented", + ServiceValidationError, + match="Command must be valid JSON", ): await hass.services.async_call( REMOTE_DOMAIN, "send_command", - {"entity_id": "remote.test_my_remote", "command": ["test"]}, + {"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, )