Open-source home automation platform running on Python 3.
69
System for loading, managing, and configuring integrations that connect Home Assistant to external devices, services, and platforms. Supports dynamic loading, dependency management, and platform-specific setup for over 1,300 built-in integrations.
Core integration discovery, loading, and metadata management system.
class Integration:
"""Integration metadata and loading information."""
def __init__(self, hass: HomeAssistant, pkg_path: str, file_path: str,
manifest: dict):
"""Initialize integration.
Args:
hass: Home Assistant instance
pkg_path: Package path
file_path: File path to integration
manifest: Integration manifest
"""
@property
def domain(self) -> str:
"""Return integration domain."""
@property
def name(self) -> str:
"""Return integration name."""
@property
def documentation(self) -> str:
"""Return integration documentation URL."""
@property
def requirements(self) -> list[str]:
"""Return integration requirements."""
@property
def dependencies(self) -> list[str]:
"""Return integration dependencies."""
@property
def after_dependencies(self) -> list[str]:
"""Return integration after dependencies."""
@property
def version(self) -> str:
"""Return integration version."""
@property
def issue_tracker(self) -> str:
"""Return integration issue tracker URL."""
@property
def quality_scale(self) -> str:
"""Return integration quality scale."""
@property
def iot_class(self) -> str:
"""Return integration IoT class."""
@property
def config_flow(self) -> bool:
"""Return True if integration supports config flow."""
@property
def homekit(self) -> dict:
"""Return HomeKit configuration."""
@property
def ssdp(self) -> list[dict]:
"""Return SSDP configuration."""
@property
def zeroconf(self) -> list[str]:
"""Return Zeroconf configuration."""
@property
def mqtt(self) -> list[str]:
"""Return MQTT configuration."""
@property
def dhcp(self) -> list[dict]:
"""Return DHCP configuration."""
@property
def usb(self) -> list[dict]:
"""Return USB configuration."""
@property
def codeowners(self) -> list[str]:
"""Return integration code owners."""
@property
def disabled(self) -> str:
"""Return disabled reason if integration is disabled."""
@property
def loggers(self) -> list[str]:
"""Return integration loggers."""
def get_component(self) -> ModuleType:
"""Get integration component module.
Returns:
Integration component module
"""
def get_platform(self, platform_name: str) -> ModuleType:
"""Get integration platform module.
Args:
platform_name: Platform name
Returns:
Platform module
"""
async def async_get_integration(hass: HomeAssistant, domain: str) -> Integration:
"""Get integration by domain.
Args:
hass: Home Assistant instance
domain: Integration domain
Returns:
Integration instance
Raises:
IntegrationNotFound: If integration not found
"""
async def async_get_custom_components(hass: HomeAssistant) -> dict[str, Integration]:
"""Get all custom components.
Args:
hass: Home Assistant instance
Returns:
Dictionary of domain -> Integration
"""
async def async_component_dependencies(hass: HomeAssistant, domain: str) -> set[str]:
"""Get component dependencies.
Args:
hass: Home Assistant instance
domain: Component domain
Returns:
Set of dependency domains
"""
def async_get_loaded_integration(hass: HomeAssistant, domain: str) -> Integration:
"""Get loaded integration.
Args:
hass: Home Assistant instance
domain: Integration domain
Returns:
Loaded integration or None
"""Integration setup, initialization, and lifecycle management system.
async def async_setup_component(hass: HomeAssistant, domain: str,
config: dict) -> bool:
"""Set up integration component.
Args:
hass: Home Assistant instance
domain: Component domain
config: Component configuration
Returns:
True if setup successful
"""
async def async_prepare_setup_platform(hass: HomeAssistant, config: dict,
domain: str, platform_name: str) -> ModuleType:
"""Prepare platform setup.
Args:
hass: Home Assistant instance
config: Platform configuration
domain: Platform domain
platform_name: Platform name
Returns:
Platform module
"""
async def async_process_deps_reqs(hass: HomeAssistant, config: dict,
integration: Integration) -> bool:
"""Process integration dependencies and requirements.
Args:
hass: Home Assistant instance
config: Integration configuration
integration: Integration instance
Returns:
True if processing successful
"""
async def async_setup_multi_components(hass: HomeAssistant, domains: set[str],
config: dict) -> dict[str, bool]:
"""Set up multiple components.
Args:
hass: Home Assistant instance
domains: Set of domains to setup
config: Configuration
Returns:
Dictionary of domain -> setup success
"""Platform discovery and loading system for finding and initializing entity platforms.
class EntityPlatform:
"""Platform for managing a group of entities."""
def __init__(self, hass: HomeAssistant, logger: logging.Logger,
domain: str, platform_name: str, platform: ModuleType,
scan_interval: timedelta, entity_namespace: str):
"""Initialize entity platform.
Args:
hass: Home Assistant instance
logger: Platform logger
domain: Entity domain
platform_name: Platform name
platform: Platform module
scan_interval: Entity scan interval
entity_namespace: Entity namespace
"""
@property
def domain(self) -> str:
"""Return platform domain."""
@property
def platform_name(self) -> str:
"""Return platform name."""
@property
def scan_interval(self) -> timedelta:
"""Return scan interval."""
async def async_setup(self, platform_config: dict,
discovery_info: dict = None) -> None:
"""Set up platform.
Args:
platform_config: Platform configuration
discovery_info: Discovery information
"""
async def async_reset(self) -> None:
"""Reset platform and remove all entities."""
async def async_shutdown(self) -> None:
"""Shutdown platform."""
async def async_add_entities(self, new_entities: list[Entity],
update_before_add: bool = False) -> None:
"""Add entities to platform.
Args:
new_entities: List of entities to add
update_before_add: Update entities before adding
"""
async def async_remove_entity(self, entity_id: str) -> None:
"""Remove entity from platform.
Args:
entity_id: Entity ID to remove
"""
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Set up integration from config entry.
Args:
hass: Home Assistant instance
config_entry: Configuration entry
Returns:
True if setup successful
"""
async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Unload integration from config entry.
Args:
hass: Home Assistant instance
config_entry: Configuration entry
Returns:
True if unload successful
"""Integration discovery mechanisms for finding devices and services on the network.
async def async_listen_platform(hass: HomeAssistant, service: str,
callback: Callable) -> None:
"""Listen for platform discoveries.
Args:
hass: Home Assistant instance
service: Service to listen for
callback: Discovery callback
"""
async def async_discover(hass: HomeAssistant, service: str, discovered: dict,
component: str, hass_config: dict) -> None:
"""Fire discovery event.
Args:
hass: Home Assistant instance
service: Discovery service
discovered: Discovery data
component: Target component
hass_config: Home Assistant configuration
"""
def discover(hass: HomeAssistant, service: str, discovered: dict,
component: str, hass_config: dict) -> None:
"""Fire discovery event (synchronous).
Args:
hass: Home Assistant instance
service: Discovery service
discovered: Discovery data
component: Target component
hass_config: Home Assistant configuration
"""
async def async_load_platform(hass: HomeAssistant, component: str,
platform: str, discovered: dict,
hass_config: dict) -> None:
"""Load platform from discovery.
Args:
hass: Home Assistant instance
component: Component name
platform: Platform name
discovered: Discovery data
hass_config: Home Assistant configuration
"""Python package requirement installation and management for integrations.
async def async_process_requirements(hass: HomeAssistant, name: str,
requirements: list[str]) -> bool:
"""Process integration requirements.
Args:
hass: Home Assistant instance
name: Integration name
requirements: List of requirements
Returns:
True if requirements processed successfully
"""
def pip_kwargs(config_dir: str) -> dict:
"""Return pip install arguments.
Args:
config_dir: Configuration directory
Returns:
Pip kwargs dictionary
"""
async def async_get_integration_with_requirements(hass: HomeAssistant,
domain: str) -> Integration:
"""Get integration and install requirements.
Args:
hass: Home Assistant instance
domain: Integration domain
Returns:
Integration with requirements installed
"""Integration manifest file validation and parsing.
def validate_manifest(manifest: dict) -> dict:
"""Validate integration manifest.
Args:
manifest: Manifest dictionary
Returns:
Validated manifest
Raises:
vol.Invalid: If manifest is invalid
"""
MANIFEST_SCHEMA = vol.Schema({
vol.Required("domain"): str,
vol.Required("name"): str,
vol.Optional("config_flow"): bool,
vol.Optional("documentation"): str,
vol.Optional("requirements"): [str],
vol.Optional("dependencies"): [str],
vol.Optional("after_dependencies"): [str],
vol.Optional("codeowners"): [str],
vol.Optional("quality_scale"): vol.In(["gold", "silver", "bronze", "internal"]),
vol.Optional("iot_class"): str,
vol.Optional("homekit"): dict,
vol.Optional("ssdp"): [dict],
vol.Optional("zeroconf"): [str],
vol.Optional("mqtt"): [str],
vol.Optional("dhcp"): [dict],
vol.Optional("usb"): [dict],
vol.Optional("version"): str,
vol.Optional("loggers"): [str],
vol.Optional("issue_tracker"): str,
vol.Optional("disabled"): str,
})# Integration quality scales
GOLD_SCALE = "gold"
SILVER_SCALE = "silver"
BRONZE_SCALE = "bronze"
INTERNAL_SCALE = "internal"
# IoT classes
CLOUD_POLLING = "cloud_polling"
CLOUD_PUSH = "cloud_push"
LOCAL_POLLING = "local_polling"
LOCAL_PUSH = "local_push"
ASSUMED = "assumed"
# Platform types (subset of available platforms)
ALARM_CONTROL_PANEL = "alarm_control_panel"
BINARY_SENSOR = "binary_sensor"
BUTTON = "button"
CAMERA = "camera"
CLIMATE = "climate"
COVER = "cover"
DEVICE_TRACKER = "device_tracker"
FAN = "fan"
LIGHT = "light"
LOCK = "lock"
MEDIA_PLAYER = "media_player"
NOTIFY = "notify"
SCENE = "scene"
SCRIPT = "script"
SENSOR = "sensor"
SWITCH = "switch"
TTS = "tts"
VACUUM = "vacuum"
WATER_HEATER = "water_heater"
# Discovery services
SERVICE_HOMEKIT = "homekit"
SERVICE_ZEROCONF = "zeroconf"
SERVICE_SSDP = "ssdp"
SERVICE_MQTT = "mqtt"
SERVICE_DHCP = "dhcp"
SERVICE_USB = "usb"
# Setup states
SETUP_SUCCESS = True
SETUP_FAILED = False
SETUP_RETRY = "retry"from typing import Any, Callable, Dict, List, Optional, Union, ModuleType
from datetime import timedelta
import logging
# Integration types
IntegrationType = Dict[str, Any]
ManifestType = Dict[str, Any]
PlatformType = ModuleType
ComponentType = ModuleType
# Setup types
SetupCallbackType = Callable[[HomeAssistant, Dict[str, Any]], bool]
AsyncSetupCallbackType = Callable[[HomeAssistant, Dict[str, Any]], Coroutine[Any, Any, bool]]
PlatformSetupType = Callable[[HomeAssistant, Dict[str, Any], Callable], None]
# Discovery types
DiscoveryInfoType = Dict[str, Any]
DiscoveryCallbackType = Callable[[str, Dict[str, Any]], None]
# Platform entity setup
AddEntitiesCallbackType = Callable[[List[Entity], bool], None]Install with Tessl CLI
npx tessl i tessl/pypi-homeassistantdocs
evals
scenario-1
scenario-2
scenario-3
scenario-4
scenario-5
scenario-6
scenario-7
scenario-8
scenario-9
scenario-10