Initial support for HomeKit enabled televisions (#32404)

* Initial support for HomeKit enabled televisions

* Fix nit from review

homeassistant/components/homekit_controller/const.py

@@ -31,4 +31,5 @@ HOMEKIT_ACCESSORY_DISPATCH = {
     "fanv2": "fan",
     "air-quality": "air_quality",
     "occupancy": "binary_sensor",
+    "television": "media_player",
 }

homeassistant/components/homekit_controller/manifest.json

@@ -3,7 +3,7 @@
   "name": "HomeKit Controller",
   "config_flow": true,
   "documentation": "https://www.home-assistant.io/integrations/homekit_controller",
-  "requirements": ["aiohomekit[IP]==0.2.15"],
+  "requirements": ["aiohomekit[IP]==0.2.17"],
   "dependencies": [],
   "zeroconf": ["_hap._tcp.local."],
   "codeowners": ["@Jc2k"]

homeassistant/components/homekit_controller/media_player.py

@@ -0,0 +1,169 @@
+"""Support for HomeKit Controller Televisions."""
+import logging
+
+from aiohomekit.model.characteristics import (
+    CharacteristicsTypes,
+    CurrentMediaStateValues,
+    RemoteKeyValues,
+    TargetMediaStateValues,
+)
+from aiohomekit.utils import clamp_enum_to_char
+
+from homeassistant.components.media_player import DEVICE_CLASS_TV, MediaPlayerDevice
+from homeassistant.components.media_player.const import (
+    SUPPORT_PAUSE,
+    SUPPORT_PLAY,
+    SUPPORT_STOP,
+)
+from homeassistant.const import STATE_IDLE, STATE_PAUSED, STATE_PLAYING
+from homeassistant.core import callback
+
+from . import KNOWN_DEVICES, HomeKitEntity
+
+_LOGGER = logging.getLogger(__name__)
+
+
+HK_TO_HA_STATE = {
+    CurrentMediaStateValues.PLAYING: STATE_PLAYING,
+    CurrentMediaStateValues.PAUSED: STATE_PAUSED,
+    CurrentMediaStateValues.STOPPED: STATE_IDLE,
+}
+
+
+async def async_setup_entry(hass, config_entry, async_add_entities):
+    """Set up Homekit television."""
+    hkid = config_entry.data["AccessoryPairingID"]
+    conn = hass.data[KNOWN_DEVICES][hkid]
+
+    @callback
+    def async_add_service(aid, service):
+        if service["stype"] != "television":
+            return False
+        info = {"aid": aid, "iid": service["iid"]}
+        async_add_entities([HomeKitTelevision(conn, info)], True)
+        return True
+
+    conn.add_listener(async_add_service)
+
+
+class HomeKitTelevision(HomeKitEntity, MediaPlayerDevice):
+    """Representation of a HomeKit Controller Television."""
+
+    def __init__(self, accessory, discovery_info):
+        """Initialise the TV."""
+        self._state = None
+        self._features = 0
+        self._supported_target_media_state = set()
+        self._supported_remote_key = set()
+        super().__init__(accessory, discovery_info)
+
+    def get_characteristic_types(self):
+        """Define the homekit characteristics the entity cares about."""
+        return [
+            CharacteristicsTypes.CURRENT_MEDIA_STATE,
+            CharacteristicsTypes.TARGET_MEDIA_STATE,
+            CharacteristicsTypes.REMOTE_KEY,
+        ]
+
+    def _setup_target_media_state(self, char):
+        self._supported_target_media_state = clamp_enum_to_char(
+            TargetMediaStateValues, char
+        )
+
+        if TargetMediaStateValues.PAUSE in self._supported_target_media_state:
+            self._features |= SUPPORT_PAUSE
+
+        if TargetMediaStateValues.PLAY in self._supported_target_media_state:
+            self._features |= SUPPORT_PLAY
+
+        if TargetMediaStateValues.STOP in self._supported_target_media_state:
+            self._features |= SUPPORT_STOP
+
+    def _setup_remote_key(self, char):
+        self._supported_remote_key = clamp_enum_to_char(RemoteKeyValues, char)
+        if RemoteKeyValues.PLAY_PAUSE in self._supported_remote_key:
+            self._features |= SUPPORT_PAUSE | SUPPORT_PLAY
+
+    @property
+    def device_class(self):
+        """Define the device class for a HomeKit enabled TV."""
+        return DEVICE_CLASS_TV
+
+    @property
+    def supported_features(self):
+        """Flag media player features that are supported."""
+        return self._features
+
+    @property
+    def state(self):
+        """State of the tv."""
+        homekit_state = self.get_hk_char_value(CharacteristicsTypes.CURRENT_MEDIA_STATE)
+        if homekit_state is None:
+            return None
+        return HK_TO_HA_STATE[homekit_state]
+
+    async def async_media_play(self):
+        """Send play command."""
+        if self.state == STATE_PLAYING:
+            _LOGGER.debug("Cannot play while already playing")
+            return
+
+        if TargetMediaStateValues.PLAY in self._supported_target_media_state:
+            characteristics = [
+                {
+                    "aid": self._aid,
+                    "iid": self._chars["target-media-state"],
+                    "value": TargetMediaStateValues.PLAY,
+                }
+            ]
+            await self._accessory.put_characteristics(characteristics)
+        elif RemoteKeyValues.PLAY_PAUSE in self._supported_remote_key:
+            characteristics = [
+                {
+                    "aid": self._aid,
+                    "iid": self._chars["remote-key"],
+                    "value": RemoteKeyValues.PLAY_PAUSE,
+                }
+            ]
+            await self._accessory.put_characteristics(characteristics)
+
+    async def async_media_pause(self):
+        """Send pause command."""
+        if self.state == STATE_PAUSED:
+            _LOGGER.debug("Cannot pause while already paused")
+            return
+
+        if TargetMediaStateValues.PAUSE in self._supported_target_media_state:
+            characteristics = [
+                {
+                    "aid": self._aid,
+                    "iid": self._chars["target-media-state"],
+                    "value": TargetMediaStateValues.PAUSE,
+                }
+            ]
+            await self._accessory.put_characteristics(characteristics)
+        elif RemoteKeyValues.PLAY_PAUSE in self._supported_remote_key:
+            characteristics = [
+                {
+                    "aid": self._aid,
+                    "iid": self._chars["remote-key"],
+                    "value": RemoteKeyValues.PLAY_PAUSE,
+                }
+            ]
+            await self._accessory.put_characteristics(characteristics)
+
+    async def async_media_stop(self):
+        """Send stop command."""
+        if self.state == STATE_IDLE:
+            _LOGGER.debug("Cannot stop when already idle")
+            return
+
+        if TargetMediaStateValues.STOP in self._supported_target_media_state:
+            characteristics = [
+                {
+                    "aid": self._aid,
+                    "iid": self._chars["target-media-state"],
+                    "value": TargetMediaStateValues.STOP,
+                }
+            ]
+            await self._accessory.put_characteristics(characteristics)

requirements_all.txt

@@ -163,7 +163,7 @@ aioftp==0.12.0
 aioharmony==0.1.13
 
 # homeassistant.components.homekit_controller
-aiohomekit[IP]==0.2.15
+aiohomekit[IP]==0.2.17
 
 # homeassistant.components.emulated_hue
 # homeassistant.components.http

requirements_test_all.txt

@@ -62,7 +62,7 @@ aiobotocore==0.11.1
 aioesphomeapi==2.6.1
 
 # homeassistant.components.homekit_controller
-aiohomekit[IP]==0.2.15
+aiohomekit[IP]==0.2.17
 
 # homeassistant.components.emulated_hue
 # homeassistant.components.http

tests/components/homekit_controller/specific_devices/test_lg_tv.py

@@ -0,0 +1,42 @@
+"""Make sure that handling real world LG HomeKit characteristics isn't broken."""
+
+
+from homeassistant.components.media_player.const import SUPPORT_PAUSE, SUPPORT_PLAY
+
+from tests.components.homekit_controller.common import (
+    Helper,
+    setup_accessories_from_file,
+    setup_test_accessories,
+)
+
+
+async def test_lg_tv(hass):
+    """Test that a Koogeek LS1 can be correctly setup in HA."""
+    accessories = await setup_accessories_from_file(hass, "lg_tv.json")
+    config_entry, pairing = await setup_test_accessories(hass, accessories)
+
+    entity_registry = await hass.helpers.entity_registry.async_get_registry()
+
+    # Assert that the entity is correctly added to the entity registry
+    entry = entity_registry.async_get("media_player.lg_webos_tv_af80")
+    assert entry.unique_id == "homekit-999AAAAAA999-48"
+
+    helper = Helper(
+        hass, "media_player.lg_webos_tv_af80", pairing, accessories[0], config_entry
+    )
+    state = await helper.poll_and_get_state()
+
+    # Assert that the friendly name is detected correctly
+    assert state.attributes["friendly_name"] == "LG webOS TV AF80"
+
+    # Assert that all optional features the LS1 supports are detected
+    assert state.attributes["supported_features"] == (SUPPORT_PAUSE | SUPPORT_PLAY)
+
+    device_registry = await hass.helpers.device_registry.async_get_registry()
+
+    device = device_registry.async_get(entry.device_id)
+    assert device.manufacturer == "LG Electronics"
+    assert device.name == "LG webOS TV AF80"
+    assert device.model == "OLED55B9PUA"
+    assert device.sw_version == "04.71.04"
+    assert device.via_device_id is None

tests/components/homekit_controller/test_media_player.py

@@ -0,0 +1,204 @@
+"""Basic checks for HomeKit motion sensors and contact sensors."""
+from aiohomekit.model.characteristics import (
+    CharacteristicPermissions,
+    CharacteristicsTypes,
+)
+from aiohomekit.model.services import ServicesTypes
+
+from tests.components.homekit_controller.common import setup_test_component
+
+CURRENT_MEDIA_STATE = ("television", "current-media-state")
+TARGET_MEDIA_STATE = ("television", "target-media-state")
+REMOTE_KEY = ("television", "remote-key")
+
+
+def create_tv_service(accessory):
+    """
+    Define tv characteristics.
+
+    The TV is not currently documented publicly - this is based on observing really TV's that have HomeKit support.
+    """
+    service = accessory.add_service(ServicesTypes.TELEVISION)
+
+    cur_state = service.add_char(CharacteristicsTypes.CURRENT_MEDIA_STATE)
+    cur_state.value = 0
+
+    remote = service.add_char(CharacteristicsTypes.REMOTE_KEY)
+    remote.value = None
+    remote.perms.append(CharacteristicPermissions.paired_write)
+
+    return service
+
+
+def create_tv_service_with_target_media_state(accessory):
+    """Define a TV service that can play/pause/stop without generate remote events."""
+    service = create_tv_service(accessory)
+
+    tms = service.add_char(CharacteristicsTypes.TARGET_MEDIA_STATE)
+    tms.value = None
+    tms.perms.append(CharacteristicPermissions.paired_write)
+
+    return service
+
+
+async def test_tv_read_state(hass, utcnow):
+    """Test that we can read the state of a HomeKit fan accessory."""
+    helper = await setup_test_component(hass, create_tv_service)
+
+    helper.characteristics[CURRENT_MEDIA_STATE].value = 0
+    state = await helper.poll_and_get_state()
+    assert state.state == "playing"
+
+    helper.characteristics[CURRENT_MEDIA_STATE].value = 1
+    state = await helper.poll_and_get_state()
+    assert state.state == "paused"
+
+    helper.characteristics[CURRENT_MEDIA_STATE].value = 2
+    state = await helper.poll_and_get_state()
+    assert state.state == "idle"
+
+
+async def test_play_remote_key(hass, utcnow):
+    """Test that we can play media on a media player."""
+    helper = await setup_test_component(hass, create_tv_service)
+
+    helper.characteristics[CURRENT_MEDIA_STATE].value = 1
+    await helper.poll_and_get_state()
+
+    await hass.services.async_call(
+        "media_player",
+        "media_play",
+        {"entity_id": "media_player.testdevice"},
+        blocking=True,
+    )
+    assert helper.characteristics[REMOTE_KEY].value == 11
+
+    # Second time should be a no-op
+    helper.characteristics[CURRENT_MEDIA_STATE].value = 0
+    await helper.poll_and_get_state()
+
+    helper.characteristics[REMOTE_KEY].value = None
+    await hass.services.async_call(
+        "media_player",
+        "media_play",
+        {"entity_id": "media_player.testdevice"},
+        blocking=True,
+    )
+    assert helper.characteristics[REMOTE_KEY].value is None
+
+
+async def test_pause_remote_key(hass, utcnow):
+    """Test that we can pause a media player."""
+    helper = await setup_test_component(hass, create_tv_service)
+
+    helper.characteristics[CURRENT_MEDIA_STATE].value = 0
+    await helper.poll_and_get_state()
+
+    await hass.services.async_call(
+        "media_player",
+        "media_pause",
+        {"entity_id": "media_player.testdevice"},
+        blocking=True,
+    )
+    assert helper.characteristics[REMOTE_KEY].value == 11
+
+    # Second time should be a no-op
+    helper.characteristics[CURRENT_MEDIA_STATE].value = 1
+    await helper.poll_and_get_state()
+
+    helper.characteristics[REMOTE_KEY].value = None
+    await hass.services.async_call(
+        "media_player",
+        "media_pause",
+        {"entity_id": "media_player.testdevice"},
+        blocking=True,
+    )
+    assert helper.characteristics[REMOTE_KEY].value is None
+
+
+async def test_play(hass, utcnow):
+    """Test that we can play media on a media player."""
+    helper = await setup_test_component(hass, create_tv_service_with_target_media_state)
+
+    helper.characteristics[CURRENT_MEDIA_STATE].value = 1
+    await helper.poll_and_get_state()
+
+    await hass.services.async_call(
+        "media_player",
+        "media_play",
+        {"entity_id": "media_player.testdevice"},
+        blocking=True,
+    )
+    assert helper.characteristics[REMOTE_KEY].value is None
+    assert helper.characteristics[TARGET_MEDIA_STATE].value == 0
+
+    # Second time should be a no-op
+    helper.characteristics[CURRENT_MEDIA_STATE].value = 0
+    await helper.poll_and_get_state()
+
+    helper.characteristics[TARGET_MEDIA_STATE].value = None
+    await hass.services.async_call(
+        "media_player",
+        "media_play",
+        {"entity_id": "media_player.testdevice"},
+        blocking=True,
+    )
+    assert helper.characteristics[REMOTE_KEY].value is None
+    assert helper.characteristics[TARGET_MEDIA_STATE].value is None
+
+
+async def test_pause(hass, utcnow):
+    """Test that we can turn pause a media player."""
+    helper = await setup_test_component(hass, create_tv_service_with_target_media_state)
+
+    helper.characteristics[CURRENT_MEDIA_STATE].value = 0
+    await helper.poll_and_get_state()
+
+    await hass.services.async_call(
+        "media_player",
+        "media_pause",
+        {"entity_id": "media_player.testdevice"},
+        blocking=True,
+    )
+    assert helper.characteristics[REMOTE_KEY].value is None
+    assert helper.characteristics[TARGET_MEDIA_STATE].value == 1
+
+    # Second time should be a no-op
+    helper.characteristics[CURRENT_MEDIA_STATE].value = 1
+    await helper.poll_and_get_state()
+
+    helper.characteristics[REMOTE_KEY].value = None
+    await hass.services.async_call(
+        "media_player",
+        "media_pause",
+        {"entity_id": "media_player.testdevice"},
+        blocking=True,
+    )
+    assert helper.characteristics[REMOTE_KEY].value is None
+
+
+async def test_stop(hass, utcnow):
+    """Test that we can  stop a media player."""
+    helper = await setup_test_component(hass, create_tv_service_with_target_media_state)
+
+    await hass.services.async_call(
+        "media_player",
+        "media_stop",
+        {"entity_id": "media_player.testdevice"},
+        blocking=True,
+    )
+    assert helper.characteristics[TARGET_MEDIA_STATE].value == 2
+
+    # Second time should be a no-op
+    helper.characteristics[CURRENT_MEDIA_STATE].value = 2
+    await helper.poll_and_get_state()
+
+    helper.characteristics[TARGET_MEDIA_STATE].value = None
+    await hass.services.async_call(
+        "media_player",
+        "media_stop",
+        {"entity_id": "media_player.testdevice"},
+        blocking=True,
+    )
+    assert helper.characteristics[REMOTE_KEY].value is None
+    assert helper.characteristics[TARGET_MEDIA_STATE].value is None