tessl/pypi-homeassistant
Open-source home automation platform running on Python 3.
69
1.18x
core-system.mddocs/
Core System
Central Home Assistant runtime providing the foundation for all automation and integration functionality. The core system includes the main HomeAssistant instance, event system, state management, service registry, and fundamental coordination mechanisms.
Capabilities
HomeAssistant Instance
The central coordinator that manages all Home Assistant components, services, and state. Acts as the main entry point and context for all Home Assistant operations.
class HomeAssistant:
"""Main Home Assistant instance."""
def __init__(self, config_dir: str):
"""Initialize Home Assistant instance.
Args:
config_dir: Path to configuration directory (required)
"""
async def async_start(self) -> None:
"""Start Home Assistant."""
async def async_stop(self, exit_code: int = 0, *, force: bool = False) -> None:
"""Stop Home Assistant.
Args:
exit_code: Exit code for shutdown
force: Force stop regardless of current state (used for testing)
"""
async def async_run(self, *, attach_signals: bool = True) -> int:
"""Run Home Assistant.
Args:
attach_signals: Whether to attach signal handlers
Returns:
Exit code
"""
def async_create_task(self, target: Coroutine, name: str = None, eager_start: bool = True) -> asyncio.Task:
"""Create a task that will be tracked by Home Assistant.
Args:
target: Coroutine to run
name: Optional name for the task
eager_start: Whether to start the task eagerly
Returns:
Created task
"""
def async_add_job(self, target: Callable | Coroutine, *args, eager_start: bool = False) -> asyncio.Future | None:
"""Add job to be executed.
Args:
target: Function or coroutine to execute
*args: Arguments for target
eager_start: Whether to start the job eagerly
Returns:
Future representing the job result
"""
# Core subsystem properties
states: StateMachine
services: ServiceRegistry
bus: EventBus
config: Config
data: dict
loop: asyncio.AbstractEventLoop
@property
def is_running(self) -> bool:
"""Return if Home Assistant is running."""
@property
def is_stopping(self) -> bool:
"""Return if Home Assistant is stopping."""
@property
def state(self) -> CoreState:
"""Return the current state of Home Assistant."""Context
Execution context that tracks the origin and flow of events, state changes, and service calls throughout the system.
class Context:
"""Context of an event, state change, or service call."""
def __init__(self, user_id: str = None, parent_id: str = None, id: str = None):
"""Initialize context.
Args:
user_id: ID of user that initiated the action
parent_id: ID of parent context
id: Context ID (generated if not provided)
"""
@property
def user_id(self) -> str:
"""Return user ID."""
@property
def parent_id(self) -> str:
"""Return parent context ID."""
@property
def origin(self) -> str:
"""Return origin of context."""
@property
def id(self) -> str:
"""Return unique context ID."""Event System
Core event publication and subscription system enabling real-time communication between Home Assistant components.
class Event:
"""Representation of an event within the event bus."""
def __init__(self, event_type: str, data: dict = None, origin: EventOrigin = None,
time_fired: datetime = None, context: Context = None):
"""Initialize event.
Args:
event_type: Type of event
data: Event data
origin: Origin of event (EventOrigin enum)
time_fired: Time event was fired
context: Event context
"""
@property
def event_type(self) -> str:
"""Return event type."""
@property
def data(self) -> dict:
"""Return event data."""
@property
def origin(self) -> str:
"""Return event origin."""
@property
def time_fired(self) -> datetime:
"""Return time event was fired."""
@property
def time_fired_timestamp(self) -> float:
"""Return timestamp when event was fired."""
@property
def context(self) -> Context:
"""Return event context."""
class EventBus:
"""Event bus for firing and listening to events."""
def fire(self, event_type: str, event_data: dict = None, origin: str = None,
context: Context = None) -> None:
"""Fire an event.
Args:
event_type: Type of event to fire
event_data: Event data
origin: Origin of event
context: Event context
"""
async def async_fire(self, event_type: str, event_data: dict = None,
origin: str = None, context: Context = None) -> None:
"""Fire an event asynchronously.
Args:
event_type: Type of event to fire
event_data: Event data
origin: Origin of event
context: Event context
"""
def listen(self, event_type: str, listener: Callable) -> Callable:
"""Listen for events of a specific type.
Args:
event_type: Type of event to listen for
listener: Function to call when event occurs
Returns:
Function to remove listener
"""
def async_listen(self, event_type: str, listener: Callable) -> Callable:
"""Listen for events asynchronously.
Args:
event_type: Type of event to listen for
listener: Async function to call when event occurs
Returns:
Function to remove listener
"""
def listen_once(self, event_type: str, listener: Callable) -> Callable:
"""Listen for first event of a specific type.
Args:
event_type: Type of event to listen for
listener: Function to call when event occurs
Returns:
Function to remove listener
"""State Management
Central state machine that tracks and manages the state of all entities in Home Assistant.
class State:
"""Object to represent a state."""
def __init__(self, entity_id: str, state: str, attributes: dict = None,
last_changed: datetime = None, last_updated: datetime = None,
context: Context = None):
"""Initialize state.
Args:
entity_id: Entity ID
state: State value
attributes: State attributes
last_changed: When state last changed
last_updated: When state was last updated
context: State context
"""
@property
def entity_id(self) -> str:
"""Return entity ID."""
@property
def state(self) -> str:
"""Return state value."""
@property
def attributes(self) -> dict:
"""Return state attributes."""
@property
def last_changed(self) -> datetime:
"""Return when state last changed."""
@property
def last_updated(self) -> datetime:
"""Return when state was last updated."""
@property
def context(self) -> Context:
"""Return state context."""
@property
def domain(self) -> str:
"""Return entity domain."""
@property
def object_id(self) -> str:
"""Return entity object ID."""
class StateMachine:
"""State machine to track states of entities."""
def get(self, entity_id: str) -> State:
"""Retrieve state of entity.
Args:
entity_id: Entity ID to get state for
Returns:
State object or None if not found
"""
def set(self, entity_id: str, new_state: str, attributes: dict = None,
force_update: bool = False, context: Context = None) -> State:
"""Set state of an entity.
Args:
entity_id: Entity ID to set state for
new_state: New state value
attributes: State attributes
force_update: Force update even if state unchanged
context: State context
Returns:
New state object
"""
async def async_set(self, entity_id: str, new_state: str,
attributes: dict = None, force_update: bool = False,
context: Context = None) -> State:
"""Set state of an entity asynchronously.
Args:
entity_id: Entity ID to set state for
new_state: New state value
attributes: State attributes
force_update: Force update even if state unchanged
context: State context
Returns:
New state object
"""
def remove(self, entity_id: str) -> bool:
"""Remove state of entity.
Args:
entity_id: Entity ID to remove
Returns:
True if entity was removed
"""
async def async_remove(self, entity_id: str, context: Context = None) -> bool:
"""Remove state of entity asynchronously.
Args:
entity_id: Entity ID to remove
context: Removal context
Returns:
True if entity was removed
"""
@property
def entity_ids(self) -> list[str]:
"""Return list of entity IDs."""
def all(self, domain_filter: str = None) -> list[State]:
"""Get all states.
Args:
domain_filter: Optional domain to filter by
Returns:
List of state objects
"""Service Registry
System for registering, managing, and calling services that perform actions in Home Assistant.
class ServiceCall:
"""Representation of a service call."""
def __init__(self, domain: str, service: str, data: dict = None,
context: Context = None, hass: HomeAssistant = None):
"""Initialize service call.
Args:
domain: Service domain
service: Service name
data: Service data
context: Service context
hass: Home Assistant instance
"""
@property
def domain(self) -> str:
"""Return service domain."""
@property
def service(self) -> str:
"""Return service name."""
@property
def data(self) -> dict:
"""Return service data."""
@property
def context(self) -> Context:
"""Return service context."""
class ServiceRegistry:
"""Registry of services."""
def register(self, domain: str, service: str, service_func: Callable,
schema: dict = None) -> None:
"""Register a service.
Args:
domain: Service domain
service: Service name
service_func: Function to call for service
schema: Voluptuous schema for service data
"""
async def async_register(self, domain: str, service: str,
service_func: Callable, schema: dict = None) -> None:
"""Register a service asynchronously.
Args:
domain: Service domain
service: Service name
service_func: Async function to call for service
schema: Voluptuous schema for service data
"""
def call(self, domain: str, service: str, service_data: dict = None,
blocking: bool = False, context: Context = None) -> bool:
"""Call a service.
Args:
domain: Service domain
service: Service name
service_data: Service data
blocking: Wait for service to complete
context: Service context
Returns:
True if service was called successfully
"""
async def async_call(self, domain: str, service: str,
service_data: dict = None, blocking: bool = False,
context: Context = None) -> bool:
"""Call a service asynchronously.
Args:
domain: Service domain
service: Service name
service_data: Service data
blocking: Wait for service to complete
context: Service context
Returns:
True if service was called successfully
"""
def has_service(self, domain: str, service: str) -> bool:
"""Check if service exists.
Args:
domain: Service domain
service: Service name
Returns:
True if service exists
"""
@property
def services(self) -> dict:
"""Return dictionary of registered services."""Core Utility Functions
Essential utility functions for working with the core system.
def callback(func: Callable) -> Callable:
"""Decorator to mark function as safe to call from event loop.
Args:
func: Function to decorate
Returns:
Decorated function
"""
def split_entity_id(entity_id: str) -> tuple[str, str]:
"""Split entity ID into domain and object ID.
Args:
entity_id: Entity ID to split
Returns:
Tuple of (domain, object_id)
"""
def valid_entity_id(entity_id: str) -> bool:
"""Validate entity ID format.
Args:
entity_id: Entity ID to validate
Returns:
True if valid entity ID format
"""
async def async_get_hass() -> HomeAssistant:
"""Get current Home Assistant instance.
Returns:
HomeAssistant instance
"""Types
from enum import Enum
from typing import Any, Callable, Dict, Optional, Union
import asyncio
# Event origin enum
class EventOrigin(Enum):
"""Origin of an event."""
local = "LOCAL"
remote = "REMOTE"
# Core state enum
class CoreState(Enum):
"""Core state of Home Assistant."""
not_running = "NOT_RUNNING"
starting = "STARTING"
running = "RUNNING"
stopping = "STOPPING"
final_write = "FINAL_WRITE"
# Type aliases
EventType = str
StateType = Union[None, str, int, float, bool]
ServiceDataType = Dict[str, Any]
ConfigType = Dict[str, Any]
CallbackType = Callable[[], None]
EventListenerType = Callable[[Event], Union[None, Coroutine]]
ServiceHandlerType = Callable[[ServiceCall], Union[None, Coroutine]]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