Accessing integration entities

Development
1

I’ve decided to challenge myself to learn how to write an integration for my View Assist project as it is now held together with duct tape (actually it’s configured using template sensors and some helpers).

To be clear, I am not a programmer so this has been most challenging for me but I have made progress. Through help of some kind folks on the VA discord, web searches and some AI help I am at the point of creating services.

At the moment, the challenge is to convert this template into a service:

  - variables:
      target_satellite_device: >-
        {% for sat_id in integration_entities('view_assist')|select('match',
        'sensor\.')|list %}
          {% set sat = states[sat_id] %}
          {% if sat and ((device_id(sat.attributes.mic_device) == trigger.device_id) or (device_id(sat.attributes.display_device) == trigger.device_id)) %}
            {{ sat.entity_id }}
          {% endif %}
        {% endfor %}

So the example above pulls all sensor devices from the view_assist domain then loops through and compares the trigger.device_id with the current iteration’s mic and display device returning the match. Purpose? Home Assistant will return the device_id of the device that made the request in a custom sentence automation. The end goal is to find what VA device has the attribute that has that device id. Once that is found then that VA satellite can be used for finding the appropriate attributes associated with it. Those are used to set media/music player, display device, attribute variables, etc that are used in the custom sentence automation action section.

For better understanding, a VA device could look like this:

template:
  - sensor:
    - name: ViewAssist_livingroom
      state: ""
      attributes:
        type: view_audio
        mic_device: "sensor.streamassist_livingroom_stt" 
        mediaplayer_device: "media_player.browsermod_livingroom"
        musicplayer_device:  "media_player.browsermod_livingroom"  
        display_device: "sensor.browsermod_livingroom_browser_path" 
        browser_id: "ViewAssist-livingroom"

Here’s the working version of my init.py:

from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import (
    HomeAssistant,
    ServiceCall,
    ServiceResponse,
    SupportsResponse,
)
from .const import DOMAIN
import homeassistant.helpers.entity_registry as er
import logging

_LOGGER = logging.getLogger(__name__)

PLATFORMS: list[Platform] = [Platform.SENSOR]

async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
    """Set up View Assist from a config entry."""
    hass.data.setdefault(DOMAIN, {})
    # Request platform setup
    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)


    async def handle_get_id(call: ServiceCall) -> ServiceResponse:
        """Handle a device lookup call."""
        hass: HomeAssistantType = call.hass
        entity_registry = er.async_get(hass)
        entity_id = call.data.get('entity_id')
        
        entity_entry = entity_registry.async_get(entity_id)
        if entity_entry and entity_entry.device_id:
            device_id = entity_entry.device_id
            return {"device_id": device_id}
        
        return {"error":  "Device ID not found"}

        # Examples for later on how to get and return variables
        # entity_id = call.data.get('entity_id')
        # state = hass.states.get(entity_id)
        # mic_device = state.attributes.get('mic_device')
        # display_device = state.attributes.get('display_device')
        # return {"mic_device": mic_device, "display_device": display_device}

    hass.services.async_register(
        DOMAIN,
        "get_id",
        handle_get_id,
        supports_response=SupportsResponse.ONLY,
    )

    async def handle_satellite_lookup(call: ServiceCall) -> ServiceResponse:
        """Handle a satellite lookup call."""
        device_id = call.data.get("device_id")
        return {"device_id_received": device_id}

    hass.services.async_register(
        DOMAIN,
        "get_satellite",
        handle_satellite_lookup,
        supports_response=SupportsResponse.ONLY,
    )

    return True
    
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
    """Unload a config entry."""
    if unloaded := await hass.config_entries.async_forward_entry_unload(entry, PLATFORMS):
        hass.data[DOMAIN].pop(entry.entry_id)
    return unloaded

What I’ve managed to do here is create a service that given an entity_id will look for and return the corresponding device_id. Eventually this will be used to match against the VA device’s attribute mic_device and display_device as seen in the template above.

My question is how do I list all entities associated with my view_assist integration? I think once I can create a loop for each of the entities I can then check the attributes’ device_id for a match for each element in the loop and get the same result the template example provides.

I am finding my way through the integration creation but having a hard time finding easy to follow examples of what I want to do. Again, I am a novice so while this may be easy for most, I am struggling. I am confident that with a few examples my understanding will grow. Any help is most appreciated.

2

So you can get a list of the entities by querying the entity registry. Below are 2 examples. 1 to get the entities by device if you have the device id and 1 to get all entities for your integration instance.

from homeassistant.helpers import entity_registry as er

entity_registry = er.async_get(self.hass)

device_entities=er.async_entries_for_device(
                    entity_registry, device_id, include_disabled_entities=True
 )

integration_entities=er.async_entries_for_config_entry(entity_registry, config_entry.entry_id)

Hope that helps.

3

Thank you for the speedy help. Unfortunately I’m having problems getting the integration to load after making those additions. Can you check what I’ve done and let me know what I am doing wrong?

from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import (
    HomeAssistant,
    ServiceCall,
    ServiceResponse,
    SupportsResponse,
)
from .const import DOMAIN
#import homeassistant.helpers.entity_registry as er
from homeassistant.helpers import entity_registry as er

import logging

_LOGGER = logging.getLogger(__name__)

PLATFORMS: list[Platform] = [Platform.SENSOR]

async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
    """Set up View Assist from a config entry."""
    hass.data.setdefault(DOMAIN, {})
    # Request platform setup
    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)


    async def handle_get_id(call: ServiceCall) -> ServiceResponse:
        """Handle a device lookup call."""
        hass: HomeAssistantType = call.hass
        entity_registry = er.async_get(hass)
        entity_id = call.data.get('entity_id')
        
        entity_entry = entity_registry.async_get(entity_id)
        if entity_entry and entity_entry.device_id:
            device_id = entity_entry.device_id
            return {"device_id": device_id}
        
        return {"error":  "Device ID not found"}

        # entity_id = call.data.get('entity_id')
        # state = hass.states.get(entity_id)
        # mic_device = state.attributes.get('mic_device')
        # display_device = state.attributes.get('display_device')
        # return {"mic_device": mic_device, "display_device": display_device}

    hass.services.async_register(
        DOMAIN,
        "get_id",
        handle_get_id,
        supports_response=SupportsResponse.ONLY,
    )

    async def handle_satellite_lookup(call: ServiceCall) -> ServiceResponse:
        """Handle a satellite lookup call."""
        device_id = call.data.get("device_id")
        return {"device_id_received": device_id}

    hass.services.async_register(
        DOMAIN,
        "get_satellite",
        handle_satellite_lookup,
        supports_response=SupportsResponse.ONLY,
    )
#####

    async def handle_get_members(call: ServiceCall) -> ServiceResponse:
        """Handle a get members lookup call."""
        entity_registry = er.async_get(self.hass)
		integration_entities=er.async_entries_for_config_entry(entity_registry, config_entry.entry_id)

        return {"integration_entities": integration_entities}

    hass.services.async_register(
        DOMAIN,
        "get_members",
        handle_get_members,
        supports_response=SupportsResponse.ONLY,
    )
#####

    return True
    
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
    """Unload a config entry."""
    if unloaded := await hass.config_entries.async_forward_entry_unload(entry, PLATFORMS):
        hass.data[DOMAIN].pop(entry.entry_id)
    return unloaded

Note my logs are showing that the error received points to:

integration_entities=er.async_entries_for_config_entry(entity_registry, config_entry.entry_id)

I’ve also changed my import to match yours though I think what I had previously was just a different way of doing the same?