Skip to content

Implement Device Automations (Triggers, Conditions, Actions) #5

Description

@adamjacobmuller

Implement Device Automations (Triggers, Conditions, Actions)

🎯 Objective

Implement device automation support for TryFi devices to enable UI-based automation creation without YAML. This is a required feature for Home Assistant Platinum tier.

📋 Requirements

  • Device triggers for pet and base station events
  • Device conditions for state checking
  • Device actions for collar control
  • Full integration with Home Assistant automation UI
  • Support for all device types (pets and bases)

🔧 Implementation Plan

1. Create Device Trigger Support

Create device_trigger.py

"""Provides device triggers for TryFi."""
from __future__ import annotations

import voluptuous as vol

from homeassistant.components.device_automation import DEVICE_TRIGGER_BASE_SCHEMA
from homeassistant.components.homeassistant.triggers import (
    numeric_state,
    state as state_trigger,
)
from homeassistant.const import (
    CONF_DEVICE_ID,
    CONF_DOMAIN,
    CONF_ENTITY_ID,
    CONF_TYPE,
    PERCENTAGE,
)
from homeassistant.core import CALLBACK_TYPE, HomeAssistant
from homeassistant.helpers import config_validation as cv, entity_registry as er
from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo
from homeassistant.helpers.typing import ConfigType

from .const import DOMAIN

TRIGGER_TYPES = {
    "arrived": "Arrived at {location}",
    "left": "Left {location}",
    "battery_low": "Battery dropped below {percentage}%",
    "lost_mode_enabled": "Lost mode was enabled",
    "lost_mode_disabled": "Lost mode was disabled",
    "connected": "Collar connected",
    "disconnected": "Collar disconnected",
    "base_online": "Base station came online",
    "base_offline": "Base station went offline",
}

TRIGGER_SCHEMA = DEVICE_TRIGGER_BASE_SCHEMA.extend(
    {
        vol.Required(CONF_TYPE): vol.In(TRIGGER_TYPES),
        vol.Optional("location"): cv.string,
        vol.Optional("percentage", default=20): vol.All(
            vol.Coerce(int), vol.Range(min=1, max=99)
        ),
    }
)


async def async_get_triggers(
    hass: HomeAssistant, device_id: str
) -> list[dict[str, str]]:
    """List device triggers for TryFi devices."""
    registry = er.async_get(hass)
    triggers = []

    # Get all entities for this device
    entities = er.async_entries_for_device(registry, device_id)
    
    # Find the device type (pet or base)
    device_registry = dr.async_get(hass)
    device = device_registry.async_get(device_id)
    
    if not device:
        return triggers
    
    # Check if it's a pet or base
    is_pet = device.model == "TryFi Collar"
    is_base = device.model == "TryFi Base Station"
    
    if is_pet:
        # Pet-specific triggers
        base_trigger = {
            CONF_DEVICE_ID: device_id,
            CONF_DOMAIN: DOMAIN,
        }
        
        # Location triggers
        triggers.append({**base_trigger, CONF_TYPE: "arrived"})
        triggers.append({**base_trigger, CONF_TYPE: "left"})
        
        # Battery trigger
        triggers.append({**base_trigger, CONF_TYPE: "battery_low"})
        
        # Lost mode triggers
        triggers.append({**base_trigger, CONF_TYPE: "lost_mode_enabled"})
        triggers.append({**base_trigger, CONF_TYPE: "lost_mode_disabled"})
        
        # Connection triggers
        triggers.append({**base_trigger, CONF_TYPE: "connected"})
        triggers.append({**base_trigger, CONF_TYPE: "disconnected"})
    
    elif is_base:
        # Base station triggers
        base_trigger = {
            CONF_DEVICE_ID: device_id,
            CONF_DOMAIN: DOMAIN,
        }
        
        triggers.append({**base_trigger, CONF_TYPE: "base_online"})
        triggers.append({**base_trigger, CONF_TYPE: "base_offline"})
    
    return triggers


async def async_attach_trigger(
    hass: HomeAssistant,
    config: ConfigType,
    action: TriggerActionType,
    trigger_info: TriggerInfo,
) -> CALLBACK_TYPE:
    """Attach a trigger."""
    trigger_type = config[CONF_TYPE]
    
    # Map trigger types to actual entity monitoring
    if trigger_type in ["arrived", "left"]:
        # Monitor device_tracker entity
        entity_id = _get_entity_id(hass, config[CONF_DEVICE_ID], "device_tracker")
        location = config.get("location", "home")
        
        if trigger_type == "arrived":
            state_config = {
                **state_trigger.TRIGGER_SCHEMA(
                    {
                        "platform": "state",
                        "entity_id": entity_id,
                        "to": location,
                    }
                ),
            }
        else:  # left
            state_config = {
                **state_trigger.TRIGGER_SCHEMA(
                    {
                        "platform": "state",
                        "entity_id": entity_id,
                        "from": location,
                    }
                ),
            }
        
        return await state_trigger.async_attach_trigger(
            hass, state_config, action, trigger_info, platform_type="device"
        )
    
    elif trigger_type == "battery_low":
        # Monitor battery sensor
        entity_id = _get_entity_id(hass, config[CONF_DEVICE_ID], "sensor", "battery")
        percentage = config.get("percentage", 20)
        
        numeric_config = {
            **numeric_state.TRIGGER_SCHEMA(
                {
                    "platform": "numeric_state",
                    "entity_id": entity_id,
                    "below": percentage,
                }
            ),
        }
        
        return await numeric_state.async_attach_trigger(
            hass, numeric_config, action, trigger_info, platform_type="device"
        )
    
    # Add more trigger type handlers...
    
    return lambda: None


def _get_entity_id(
    hass: HomeAssistant,
    device_id: str,
    domain: str,
    entity_type: str | None = None,
) -> str | None:
    """Get entity ID for a device."""
    registry = er.async_get(hass)
    entities = er.async_entries_for_device(registry, device_id)
    
    for entry in entities:
        if entry.domain != domain:
            continue
        if entity_type and entity_type not in entry.entity_id:
            continue
        return entry.entity_id
    
    return None

2. Create Device Condition Support

Create device_condition.py

"""Provides device conditions for TryFi."""
from __future__ import annotations

import voluptuous as vol

from homeassistant.components.device_automation import DEVICE_CONDITION_BASE_SCHEMA
from homeassistant.const import (
    CONF_CONDITION,
    CONF_DEVICE_ID,
    CONF_DOMAIN,
    CONF_TYPE,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import condition, entity_registry as er
from homeassistant.helpers.config_validation import DEVICE_CONDITION_BASE_SCHEMA
from homeassistant.helpers.typing import ConfigType, TemplateVarsType

from .const import DOMAIN

CONDITION_TYPES = {
    "is_at_location": "Is at {location}",
    "is_battery_low": "Battery is below {percentage}%",
    "is_lost": "Is in lost mode",
    "is_connected": "Is connected",
    "is_charging": "Is charging",
    "is_base_online": "Base is online",
}

CONDITION_SCHEMA = DEVICE_CONDITION_BASE_SCHEMA.extend(
    {
        vol.Required(CONF_TYPE): vol.In(CONDITION_TYPES),
        vol.Optional("location"): cv.string,
        vol.Optional("percentage", default=20): vol.All(
            vol.Coerce(int), vol.Range(min=1, max=99)
        ),
    }
)


async def async_get_conditions(
    hass: HomeAssistant, device_id: str
) -> list[dict[str, str]]:
    """List device conditions for TryFi devices."""
    registry = er.async_get(hass)
    conditions = []
    
    # Similar logic to triggers
    device_registry = dr.async_get(hass)
    device = device_registry.async_get(device_id)
    
    if not device:
        return conditions
    
    is_pet = device.model == "TryFi Collar"
    is_base = device.model == "TryFi Base Station"
    
    base_condition = {
        CONF_CONDITION: "device",
        CONF_DEVICE_ID: device_id,
        CONF_DOMAIN: DOMAIN,
    }
    
    if is_pet:
        conditions.extend([
            {**base_condition, CONF_TYPE: "is_at_location"},
            {**base_condition, CONF_TYPE: "is_battery_low"},
            {**base_condition, CONF_TYPE: "is_lost"},
            {**base_condition, CONF_TYPE: "is_connected"},
            {**base_condition, CONF_TYPE: "is_charging"},
        ])
    elif is_base:
        conditions.append({**base_condition, CONF_TYPE: "is_base_online"})
    
    return conditions


@callback
def async_condition_from_config(
    hass: HomeAssistant, config: ConfigType
) -> condition.ConditionCheckerType:
    """Create a function to test a device condition."""
    condition_type = config[CONF_TYPE]
    device_id = config[CONF_DEVICE_ID]
    
    if condition_type == "is_at_location":
        entity_id = _get_entity_id(hass, device_id, "device_tracker")
        location = config.get("location", "home")
        
        def test_is_at_location(
            hass: HomeAssistant, variables: TemplateVarsType
        ) -> bool:
            """Test if pet is at location."""
            state = hass.states.get(entity_id)
            return state is not None and state.state == location
        
        return test_is_at_location
    
    # Add more condition handlers...
    
    return lambda hass, variables: False

3. Create Device Action Support

Create device_action.py

"""Provides device actions for TryFi."""
from __future__ import annotations

import voluptuous as vol

from homeassistant.components.device_automation import DEVICE_ACTION_BASE_SCHEMA
from homeassistant.const import (
    CONF_DEVICE_ID,
    CONF_DOMAIN,
    CONF_TYPE,
)
from homeassistant.core import Context, HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.typing import ConfigType, TemplateVarsType

from . import DOMAIN

ACTION_TYPES = {
    "turn_on_led": "Turn on collar LED",
    "turn_off_led": "Turn off collar LED",
    "set_led_color": "Set LED color to {color}",
    "enable_lost_mode": "Enable lost mode",
    "disable_lost_mode": "Disable lost mode",
}

ACTION_SCHEMA = DEVICE_ACTION_BASE_SCHEMA.extend(
    {
        vol.Required(CONF_TYPE): vol.In(ACTION_TYPES),
        vol.Optional("color"): vol.In(
            ["red", "green", "blue", "yellow", "magenta", "cyan", "white"]
        ),
    }
)


async def async_get_actions(
    hass: HomeAssistant, device_id: str
) -> list[dict[str, str]]:
    """List device actions for TryFi devices."""
    registry = er.async_get(hass)
    actions = []
    
    device_registry = dr.async_get(hass)
    device = device_registry.async_get(device_id)
    
    if not device or device.model != "TryFi Collar":
        return actions
    
    base_action = {
        CONF_DEVICE_ID: device_id,
        CONF_DOMAIN: DOMAIN,
    }
    
    # LED actions
    actions.extend([
        {**base_action, CONF_TYPE: "turn_on_led"},
        {**base_action, CONF_TYPE: "turn_off_led"},
        {**base_action, CONF_TYPE: "set_led_color"},
    ])
    
    # Lost mode actions
    actions.extend([
        {**base_action, CONF_TYPE: "enable_lost_mode"},
        {**base_action, CONF_TYPE: "disable_lost_mode"},
    ])
    
    return actions


async def async_call_action_from_config(
    hass: HomeAssistant,
    config: ConfigType,
    variables: TemplateVarsType,
    context: Context | None,
) -> None:
    """Execute a device action."""
    action_type = config[CONF_TYPE]
    device_id = config[CONF_DEVICE_ID]
    
    if action_type == "turn_on_led":
        entity_id = _get_entity_id(hass, device_id, "light")
        await hass.services.async_call(
            "light",
            "turn_on",
            {"entity_id": entity_id},
            context=context,
        )
    
    elif action_type == "turn_off_led":
        entity_id = _get_entity_id(hass, device_id, "light")
        await hass.services.async_call(
            "light",
            "turn_off",
            {"entity_id": entity_id},
            context=context,
        )
    
    elif action_type == "set_led_color":
        entity_id = _get_entity_id(hass, device_id, "select", "led_color")
        color = config.get("color", "white")
        await hass.services.async_call(
            "select",
            "select_option",
            {"entity_id": entity_id, "option": color},
            context=context,
        )
    
    # Add more action handlers...

4. Update Integration Manifest

Add device automation support to manifest.json:

{
  "device_automation": {
    "trigger": ["device_trigger"],
    "condition": ["device_condition"],
    "action": ["device_action"]
  }
}

5. Add Tests

Create comprehensive tests for each automation type:

  • tests/test_device_trigger.py
  • tests/test_device_condition.py
  • tests/test_device_action.py

📝 Checklist

  • Create device_trigger.py with all trigger types
  • Create device_condition.py with all condition types
  • Create device_action.py with all action types
  • Update manifest.json to register device automation
  • Add helper functions for entity lookup
  • Implement all trigger handlers
  • Implement all condition handlers
  • Implement all action handlers
  • Add comprehensive tests for all automation types
  • Test UI integration in Home Assistant
  • Add documentation for device automations
  • Ensure translations work with device automations

🧪 Testing

  1. Test creating automations from device page
  2. Verify all trigger types work correctly
  3. Test all condition evaluations
  4. Verify all actions execute properly
  5. Test with multiple pets/devices
  6. Verify UI shows correct options
  7. Test automation execution

📚 Resources

⏱️ Estimated Time

  • Implementation: 16 hours
  • Testing: 8 hours
  • Documentation: 2 hours

🏷️ Labels

enhancement, platinum-tier, device-automation, required-for-platinum

Metadata

Metadata

Assignees

No one assigned

    Labels

    enhancementNew feature or requestplatinumTasks required to get this integration to platinum status

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions