Python Home Assistant Automation — Deep Dive

Home Assistant Architecture

Home Assistant Core is a Python application built on asyncio. Its architecture consists of several key components:

Event Bus — Every state change, service call, and time event generates an event on the central event bus. Automations and integrations subscribe to events they care about.

State Machine — Tracks the current state and attributes of every entity. State changes fire state_changed events.

Service Registry — Maps service calls (like light.turn_on) to the integration code that handles them.

Config Entries — Manages integration configuration, credentials, and lifecycle.

The entire system runs in a single Python process using asyncio for concurrency. Long-running or blocking operations use the thread pool executor.

Building Custom Integrations

Custom integrations live in config/custom_components/<domain>/:

custom_components/
└── my_sensor/
    ├── __init__.py
    ├── manifest.json
    ├── config_flow.py
    ├── sensor.py
    └── const.py

manifest.json

{
  "domain": "my_sensor",
  "name": "My Custom Sensor",
  "version": "1.0.0",
  "documentation": "https://github.com/user/my_sensor",
  "dependencies": [],
  "codeowners": ["@username"],
  "requirements": ["some-pypi-package==1.2.3"],
  "iot_class": "local_polling"
}

Sensor Platform

# sensor.py
from homeassistant.components.sensor import (
    SensorEntity,
    SensorDeviceClass,
    SensorStateClass,
)
from homeassistant.const import UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.entity_platform import AddEntitiesCallback

async def async_setup_entry(
    hass: HomeAssistant,
    config_entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    coordinator = hass.data[DOMAIN][config_entry.entry_id]
    async_add_entities([
        MyTemperatureSensor(coordinator),
        MyHumiditySensor(coordinator),
    ])

class MyTemperatureSensor(SensorEntity):
    _attr_device_class = SensorDeviceClass.TEMPERATURE
    _attr_state_class = SensorStateClass.MEASUREMENT
    _attr_native_unit_of_measurement = UnitOfTemperature.CELSIUS
    _attr_name = "My Temperature"
    _attr_unique_id = "my_sensor_temperature"
    
    def __init__(self, coordinator):
        self.coordinator = coordinator
    
    @property
    def native_value(self):
        return self.coordinator.data.get("temperature")
    
    @property
    def available(self):
        return self.coordinator.last_update_success

Data Update Coordinator

The coordinator pattern centralizes API calls and distributes data to entities:

# __init__.py
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from datetime import timedelta
import logging

_LOGGER = logging.getLogger(__name__)

class MySensorCoordinator(DataUpdateCoordinator):
    def __init__(self, hass, api_client):
        super().__init__(
            hass,
            _LOGGER,
            name="My Sensor",
            update_interval=timedelta(seconds=30),
        )
        self.api = api_client
    
    async def _async_update_data(self):
        try:
            return await self.api.get_readings()
        except ConnectionError as err:
            raise UpdateFailed(f"Error communicating with sensor: {err}")

async def async_setup_entry(hass, entry):
    api = MySensorAPI(entry.data["host"])
    coordinator = MySensorCoordinator(hass, api)
    await coordinator.async_config_entry_first_refresh()
    
    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
    await hass.config_entries.async_forward_entry_setups(entry, ["sensor"])
    return True

Config Flow for UI Setup

# config_flow.py
from homeassistant import config_entries
import voluptuous as vol

class MyConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
    VERSION = 1
    
    async def async_step_user(self, user_input=None):
        errors = {}
        
        if user_input is not None:
            # Validate the connection
            try:
                api = MySensorAPI(user_input["host"])
                await api.test_connection()
            except ConnectionError:
                errors["base"] = "cannot_connect"
            else:
                return self.async_create_entry(
                    title=f"Sensor at {user_input['host']}",
                    data=user_input,
                )
        
        return self.async_show_form(
            step_id="user",
            data_schema=vol.Schema({
                vol.Required("host"): str,
                vol.Optional("port", default=8080): int,
            }),
            errors=errors,
        )

Advanced Automation Templates

Home Assistant’s template engine (Jinja2) enables dynamic automations:

Adaptive Lighting Based on Sun Position

- alias: "Adaptive Living Room Brightness"
  trigger:
    - platform: time_pattern
      minutes: "/5"
  condition:
    - condition: state
      entity_id: light.living_room
      state: "on"
  action:
    - service: light.turn_on
      target:
        entity_id: light.living_room
      data:
        brightness_pct: >
          {% set elevation = state_attr('sun.sun', 'elevation') %}
          {% if elevation > 30 %}
            100
          {% elif elevation > 10 %}
            {{ (elevation * 3) | int }}
          {% elif elevation > 0 %}
            30
          {% else %}
            {{ [80 - (elevation | abs * 2), 20] | max | int }}
          {% endif %}
        color_temp_kelvin: >
          {% set elevation = state_attr('sun.sun', 'elevation') %}
          {% if elevation > 20 %}
            5000
          {% elif elevation > 0 %}
            {{ 3000 + (elevation * 100) | int }}
          {% else %}
            2700
          {% endif %}

Template Sensors

Create computed sensors from raw data:

# configuration.yaml
template:
  - sensor:
      - name: "House Energy Cost Today"
        unit_of_measurement: "€"
        state: >
          {% set kwh = states('sensor.energy_daily') | float(0) %}
          {% set hour = now().hour %}
          {% if 7 <= hour < 23 %}
            {{ (kwh * 0.28) | round(2) }}
          {% else %}
            {{ (kwh * 0.15) | round(2) }}
          {% endif %}
      
      - name: "Rooms Occupied"
        state: >
          {% set rooms = [
            'binary_sensor.motion_living_room',
            'binary_sensor.motion_kitchen',
            'binary_sensor.motion_bedroom',
            'binary_sensor.motion_office'
          ] %}
          {{ rooms | select('is_state', 'on') | list | count }}

AppDaemon: Python-First Automations

AppDaemon is an execution engine that runs Python apps alongside Home Assistant. It provides a more powerful programming model than YAML automations:

# apps/climate_control.py
import appdaemon.plugins.hass.hassapi as hass
from datetime import datetime, time

class ClimateController(hass.Hass):
    def initialize(self):
        self.comfort_temp = self.args.get("comfort_temp", 22)
        self.away_temp = self.args.get("away_temp", 18)
        self.sleep_temp = self.args.get("sleep_temp", 19)
        self.thermostat = self.args["thermostat"]
        
        # React to presence changes
        self.listen_state(self.presence_changed, "group.family")
        
        # Schedule temperature adjustments
        self.run_daily(self.morning_routine, time(6, 30))
        self.run_daily(self.night_routine, time(22, 30))
        
        # React to window opening
        for window in self.args.get("windows", []):
            self.listen_state(self.window_changed, window)
    
    def presence_changed(self, entity, attribute, old, new, kwargs):
        if new == "not_home":
            self.set_temperature(self.away_temp)
            self.log(f"Everyone left, setting {self.away_temp}°C")
        elif old == "not_home" and new == "home":
            self.set_temperature(self.comfort_temp)
            self.log(f"Welcome home, setting {self.comfort_temp}°C")
    
    def window_changed(self, entity, attribute, old, new, kwargs):
        if new == "on":  # window opened
            self.call_service("climate/turn_off",
                            entity_id=self.thermostat)
            self.log(f"Window opened ({entity}), HVAC off")
        elif new == "off":  # window closed
            self.call_service("climate/turn_on",
                            entity_id=self.thermostat)
            self.set_temperature(self.comfort_temp)
    
    def morning_routine(self, kwargs):
        if self.get_state("group.family") == "home":
            self.set_temperature(self.comfort_temp)
    
    def night_routine(self, kwargs):
        self.set_temperature(self.sleep_temp)
    
    def set_temperature(self, temp):
        self.call_service("climate/set_temperature",
                         entity_id=self.thermostat,
                         temperature=temp)

# apps/apps.yaml
climate_controller:
  module: climate_control
  class: ClimateController
  thermostat: climate.living_room
  comfort_temp: 22
  away_temp: 17
  sleep_temp: 19
  windows:
    - binary_sensor.window_living_room
    - binary_sensor.window_bedroom

REST API Integration

Home Assistant exposes a REST API for external control:

import requests

HA_URL = "http://homeassistant.local:8123"
TOKEN = "your_long_lived_access_token"

headers = {
    "Authorization": f"Bearer {TOKEN}",
    "Content-Type": "application/json",
}

# Get entity state
response = requests.get(
    f"{HA_URL}/api/states/sensor.outdoor_temperature",
    headers=headers
)
state = response.json()
print(f"Temperature: {state['state']}°C")

# Call a service
requests.post(
    f"{HA_URL}/api/services/light/turn_on",
    headers=headers,
    json={
        "entity_id": "light.living_room",
        "brightness_pct": 75
    }
)

# Fire an event
requests.post(
    f"{HA_URL}/api/events/custom_alert",
    headers=headers,
    json={"source": "external_script", "severity": "warning"}
)

WebSocket API for Real-Time Updates

import asyncio
import aiohttp
import json

async def listen_state_changes():
    async with aiohttp.ClientSession() as session:
        async with session.ws_connect(
            f"ws://homeassistant.local:8123/api/websocket"
        ) as ws:
            # Authentication
            auth_msg = await ws.receive_json()
            await ws.send_json({"type": "auth", "access_token": TOKEN})
            auth_result = await ws.receive_json()
            
            # Subscribe to state changes
            await ws.send_json({
                "id": 1,
                "type": "subscribe_events",
                "event_type": "state_changed"
            })
            
            async for msg in ws:
                if msg.type == aiohttp.WSMsgType.TEXT:
                    data = json.loads(msg.data)
                    if data.get("type") == "event":
                        event = data["event"]["data"]
                        entity = event["entity_id"]
                        new_state = event["new_state"]["state"]
                        print(f"{entity}: {new_state}")

asyncio.run(listen_state_changes())

Production Deployment Patterns

Backup and Version Control

# automations, scripts, and scenes in version control
# .gitignore
*.db
*.log
secrets.yaml
.storage/
tts/

Keep secrets.yaml separate and use it for all credentials:

# secrets.yaml (NOT in version control)
mqtt_password: "secret123"
latitude: 52.5200
longitude: 13.4050

# configuration.yaml
mqtt:
  password: !secret mqtt_password

Monitoring Home Assistant

# Monitor HA itself
sensor:
  - platform: systemmonitor
    resources:
      - type: processor_use
      - type: memory_use_percent
      - type: disk_use_percent
        arg: /

automation:
  - alias: "Alert on High CPU"
    trigger:
      - platform: numeric_state
        entity_id: sensor.processor_use
        above: 90
        for: "00:05:00"
    action:
      - service: notify.mobile_app
        data:
          message: "Home Assistant CPU at {{ states('sensor.processor_use') }}%"

ESPHome for Custom Sensors

ESPHome compiles YAML into firmware for ESP32/ESP8266 boards that integrate natively with Home Assistant:

# esphome/garage_sensor.yaml
esphome:
  name: garage-sensor

esp32:
  board: esp32dev

wifi:
  ssid: !secret wifi_ssid
  password: !secret wifi_password

api:
  encryption:
    key: !secret api_key

sensor:
  - platform: dht
    pin: GPIO4
    temperature:
      name: "Garage Temperature"
    humidity:
      name: "Garage Humidity"
    update_interval: 60s

binary_sensor:
  - platform: gpio
    pin: GPIO5
    name: "Garage Door"
    device_class: garage_door

This creates a sensor device that Home Assistant automatically discovers and integrates, with no custom integration code needed.

One thing to remember: Home Assistant’s true power is the combination of its universal Python integration framework, local-first architecture, and thriving ecosystem — whether you use YAML automations, AppDaemon Python apps, or build custom integrations, you have full control over your smart home without surrendering your data to the cloud.