home-assistant/utils/validation/validate_devices.py

"""Validate device IDs against Home Assistant device registry via REST API."""

import os
import sys
from pathlib import Path

import requests

# Add parent directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent.parent))

from utils.validation.parse_config import parse_config_files


def get_device_registry(ha_url: str, token: str) -> list[dict]:
    """
    Fetch device registry from Home Assistant REST API.
    
    Args:
        ha_url: Home Assistant base URL (e.g., http://homeassistant:8123)
        token: Long-lived access token
    
    Returns:
        List of device dictionaries from device registry
    
    Raises:
        requests.RequestException: If API call fails
    """
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
    }
    
    response = requests.get(
        f"{ha_url}/api/config/device_registry/list",
        headers=headers,
        timeout=10
    )
    response.raise_for_status()
    
    return response.json()


def validate_devices(config_dir: Path) -> tuple[set[str], set[str], set[str]]:
    """
    Validate device IDs from config files against Home Assistant API.
    
    Args:
        config_dir: Path to Home Assistant config directory
    
    Returns:
        Tuple of (valid_devices, invalid_devices, all_config_devices)
    """
    # Get environment variables
    ha_host = os.environ.get('HA_HOST')
    ha_token = os.environ.get('HA_TOKEN')
    
    if not ha_host or not ha_token:
        raise ValueError(
            "Home Assistant credentials not found. Set HA_HOST and HA_TOKEN "
            "environment variables."
        )
    
    # Construct URL
    ha_url = f"http://{ha_host}:8123" if not ha_host.startswith('http') else ha_host
    
    # Parse config files to get device IDs
    _, config_devices = parse_config_files(config_dir)
    
    # Get device registry from API
    try:
        devices = get_device_registry(ha_url, ha_token)
        valid_device_ids = {device['id'] for device in devices}
    except requests.RequestException as e:
        raise RuntimeError(f"Failed to fetch device registry from {ha_url}: {e}")
    
    # Find valid and invalid devices
    valid = config_devices & valid_device_ids
    invalid = config_devices - valid_device_ids
    
    return valid, invalid, config_devices


def main():
    """Main entry point for device validation."""
    if len(sys.argv) < 2:
        print("Usage: python validate_devices.py <config_directory>")
        sys.exit(1)
    
    config_path = Path(sys.argv[1])
    
    if not config_path.exists():
        print(f"Error: Config directory not found: {config_path}")
        sys.exit(1)
    
    print("Validating device IDs against Home Assistant API...")
    
    try:
        valid, invalid, all_devices = validate_devices(config_path)
        
        print(f"\n✓ Total device IDs in config: {len(all_devices)}")
        print(f"✓ Valid device IDs: {len(valid)}")
        
        if invalid:
            print(f"\n⚠ Warning: {len(invalid)} device IDs not found in device registry:")
            for device_id in sorted(invalid):
                print(f"  - {device_id}")
            print("\nNote: These devices may have been removed or the IDs might be incorrect.")
            # Exit with warning code (not failure)
            sys.exit(0)
        else:
            print("\n✓ All device IDs validated successfully!")
            sys.exit(0)
    
    except Exception as e:
        print(f"\n✗ Error during validation: {e}")
        sys.exit(1)


if __name__ == '__main__':
    main()