Initial commit

Author: relvacode

.gitignore

@@ -0,0 +1,2 @@
+.idea
+__pycache__

README.md

@@ -0,0 +1,43 @@
+# LightCast
+
+Transform your Home Assistant smart bulbs into a media player
+
+When an image is played on a LightCast media player, it uses [colorgram.py](https://github.com/obskyr/colorgram.py/tree/master) 
+to extract a palette of colors equal to the amount of lights configured in the LightCast player and sets those lights to that color palette.
+
+## Configuration
+
+`configuration.yml`
+
+```yaml
+media_player:
+  - platform: lightcast
+    name: Living Room LightCast
+    target: area.living_room
+```
+
+Target can be a light, area, group or device.
+
+You can specify target as a list of entities as well, LightCast will resolve all lights found in all targets. 
+Note that LightCast cannot resolve individual lights if you provide a Zigbee coordinator group.
+
+```yaml
+media_player:
+  - platform: lightcast
+    name: Living Room LightCast
+    target:
+      - entity.living_room_ceiling
+      - entity.living_room_backlight
+```
+
+You can create multiple LightCast media players for each area in your home
+
+```yaml
+media_player:
+  - platform: lightcast
+    name: Living Room LightCast
+    target: area.living_room
+  - platform: lightcast
+    name: Office LightCast
+    target: area.office
+```

__init__.py

@@ -0,0 +1,10 @@
+"""
+LightCast integration
+"""
+from homeassistant.core import HomeAssistant
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
+    await hass.config_entries.async_forward_entry_setups(entry, [Platform.MEDIA_PLAYER])

const.py

@@ -0,0 +1,2 @@
+CONF_TARGET = 'target'
+CONF_NAME = 'name'

entity_resolver.py

@@ -0,0 +1,83 @@
+import typing
+import collections.abc
+from homeassistant.core import HomeAssistant, State
+from homeassistant.helpers.template import area_entities, device_entities
+from homeassistant.helpers.entity import entity_sources
+from homeassistant.const import ATTR_ENTITY_ID
+
+GROUP_DOMAIN_PREFIX = 'group.'
+AREA_DOMAIN_PREFIX = 'area.'
+DEVICE_DOMAIN_PREFIX = 'device.'
+
+EntityRef = str | State
+
+
+def expand_entities(
+        hass: HomeAssistant,
+        expand_id: EntityRef | typing.Iterable[EntityRef],
+        in_domain: str = 'light') -> typing.List[State]:
+    """
+    Expand an entity reference given by expand_id to a list of HomeAssistant states that are a member of that reference.
+    expand_id can be one of:
+        - A string referencing an entity id
+        - A State itself
+        - A device ID
+        - An area ID
+        - A group helper ID
+
+    :param hass:
+        Instance of HASS client
+    :param expand_id:
+        The entity ID or group reference to resolve
+    :param in_domain:
+        Filter returned entities to only those that are of the given domain
+    :return:
+        A list of de-duplicated resolved states
+    """
+    search = [expand_id]
+    found: typing.Dict[str, State] = {}
+
+    while search:
+        entity = search.pop()
+
+        if isinstance(entity, State):
+            found[entity.entity_id] = entity
+            continue
+
+        if (not (is_str := isinstance(entity, str))) and isinstance(entity, collections.abc.Iterable):
+            search += entity
+            continue
+
+        # Must be a string by this point
+        if not is_str:
+            continue  # ignore
+
+        entity_id = entity
+
+        # Entity ref is an area
+        if entity_id.startswith(AREA_DOMAIN_PREFIX):
+            search += area_entities(hass, entity_id.removeprefix(AREA_DOMAIN_PREFIX))
+            continue
+
+        # Entity ref is a device
+        if entity_id.startswith(DEVICE_DOMAIN_PREFIX):
+            search += device_entities(hass, entity_id)
+            continue
+
+        # Entity ref should be an entity (or a group helper)
+        if (entity := hass.states.get(entity_id)) is None:
+            continue  # not defined
+
+        if entity_id.startswith(GROUP_DOMAIN_PREFIX) or (
+                (source := entity_sources(hass).get(entity_id))
+                and source["domain"] == "group"
+        ):
+            if group_entities := entity.attributes.get(ATTR_ENTITY_ID):
+                search += group_entities
+                continue
+
+        # Finally, this should be a single entity in the light domain
+        if entity.domain == in_domain:
+            found[entity.entity_id] = entity
+
+    return list(found.values())

manifest.json

@@ -0,0 +1,11 @@
+{
+  "domain": "lightcast",
+  "name": "Light Cast",
+  "codeowners": ["@relvacode"],
+  "dependencies": [],
+  "requirements": ["aiohttp", "colorgram.py"],
+  "documentation": "https://github.com/relvacode/LightCast#readme",
+  "issue_tracker": "https://github.com/relvacode/LightCast/issues",
+  "version": "0.1.0",
+  "integration_type": "entity"
+}

media_player.py

@@ -0,0 +1,119 @@
+import logging
+import io
+import itertools
+import aiohttp
+import colorgram
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.components.media_player import MediaPlayerEntity, MediaPlayerEntityFeature, \
+    MediaType, MediaPlayerDeviceClass
+from homeassistant.components import media_source
+from homeassistant.components.media_player.browse_media import (
+    BrowseMedia,
+    async_process_play_media_url,
+)
+from homeassistant.const import ATTR_ENTITY_ID
+from homeassistant.components import light
+
+from . import const
+from .entity_resolver import expand_entities
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_platform(
+        hass: HomeAssistant,
+        config: ConfigType,
+        add_entities: AddEntitiesCallback,
+        discovery_info: DiscoveryInfoType | None = None,
+) -> None:
+    _LOGGER.info('LightCast media_player setup: %s', config)
+
+    cast_device = LightCastPlayer(
+        hass,
+        config[const.CONF_NAME],
+        config[const.CONF_TARGET],
+    )
+
+    add_entities([cast_device])
+
+
+class LightCastPlayer(MediaPlayerEntity):
+    _attr_available = True
+    _attr_supported_features = MediaPlayerEntityFeature.PLAY_MEDIA | MediaPlayerEntityFeature.BROWSE_MEDIA
+    _attr_device_class = MediaPlayerDeviceClass.TV
+
+    def __init__(self, hass: HomeAssistant, name: str, device_target: str) -> None:
+        self.hass = hass
+        self.device_target = device_target
+        self._attr_name = name
+
+    async def async_browse_media(self, media_content_type: str | None = None,
+                                 media_content_id: str | None = None) -> BrowseMedia:
+        """
+        Implement MediaPlayerEntityFeature.BROWSE_MEDIA
+        :param media_content_type:
+        :param media_content_id:
+        :return:
+        """
+        return await media_source.async_browse_media(
+            self.hass,
+            media_content_id,
+            content_filter=lambda item: item.media_content_type.startswith('image/')
+        )
+
+    async def process_image(self, media_type: str, media_id: str) -> None:
+        """
+        Download an image found at media_id to an in-memory buffer.
+        Resolve the entity list of lights referenced by device_target which are currently in the on state
+        and extract that many colors from the image using colorgram.
+        For each pair of light and color, call light.turn_on with that color
+        :param media_type:
+        :param media_id:
+        """
+        _LOGGER.info('Processing image %s: %s', media_type, media_id)
+
+        found_entities = expand_entities(self.hass, self.device_target)
+        _LOGGER.info('Found %d entities matching %s', len(found_entities), self.device_target)
+
+        if not found_entities:
+            _LOGGER.warning('No entities were matched')
+            return
+
+        valid_entities = [e for e in found_entities if e.state == 'on']
+        if not valid_entities:
+            _LOGGER.warning('No targets are on')
+            return
+
+        async with aiohttp.ClientSession() as session:
+            async with session.get(media_id) as response:
+                response_data = await response.read()
+
+        n_colors = len(valid_entities)
+        _LOGGER.info('Downloaded image, extracting palette of %d colors', n_colors)
+
+        extracted_colors = colorgram.extract(io.BytesIO(response_data), n_colors)
+        _LOGGER.info('Extracted palette of %d colors from image', len(extracted_colors))
+
+        for e, color in zip(valid_entities, itertools.cycle(extracted_colors)):
+            rgb_color = [color.rgb.r, color.rgb.g, color.rgb.b]
+            _LOGGER.info('Set light %s to %s', e.entity_id, rgb_color)
+            await self.hass.services.async_call(
+                light.DOMAIN,
+                light.SERVICE_TURN_ON,
+                {
+                    ATTR_ENTITY_ID: e.entity_id,
+                    light.ATTR_RGB_COLOR: rgb_color
+                }
+            )
+
+    async def async_play_media(self, media_type: MediaType, media_id: str, **kwargs: any) -> None:
+        if media_source.is_media_source_id(media_id):
+            play_item = await media_source.async_resolve_media(self.hass, media_id, self.entity_id)
+            media_id = async_process_play_media_url(self.hass, play_item.url)
+
+        self.hass.async_create_task(
+            self.process_image(media_type, media_id)
+        )