hass-maestro 0.1.0

Strongly-typed Python automation framework for Home Assistant

Maestro

Strongly-typed Python automations for Home Assistant

Maestro is a framework that lets you write Home Assistant automations in Python with full type safety, IDE autocomplete, and a clean decorator-based API. Instead of YAML configurations, write real Python code with access to your entities as typed objects.

Note: A picture repo is worth a thousand words. Check out some real world examples here: https://github.com/papa-marsh/maestro

Why Maestro?

• Type Safety: Full type hints and IDE autocomplete for all entities and attributes
• Python Power: Use the full Python ecosystem—logic, libraries, conditionals, loops
• Event-Driven: Decorator-based triggers for state changes, schedules, and events
• Fast Development: Entity registry with autocomplete for all your Home Assistant devices
• Easy Testing: Write unit tests for your automations like any Python code

Quick Start

Prerequisites

• Python 3.14+
• A Redis instance (backs the state cache, entity locks, and the persistent job scheduler)
• Home Assistant instance with WebSocket API access

Setup

1. Scaffold your project

   uvx --from git+https://github.com/papa-marsh/hass-maestro hass-maestro init my-automations
   

   (Or just hass-maestro init my-automations if the package is already installed.)

   This generates a complete, ready-to-run project:

   my-automations/
     app.py             # Entrypoint: constructs the MaestroApp from env vars
     scripts/           # Your automation modules (with a working example + test)
     registry/          # Generated entity registry (populated from your HA instance)
     custom_domains/    # Optional Entity subclass extensions
     .env               # Runtime configuration
     Dockerfile / docker-compose.yml / justfile    # Docker deployment stack
   

2. Configure it

   Fill in HOME_ASSISTANT_URL and HOME_ASSISTANT_TOKEN in .env.

   To create a long-lived access token, navigate to the <hass_url>/profile/security page in Home Assistant. Pro tip: create the token from a separate account named "Maestro" so entity state history shows which actions Maestro triggered.

3. Run it

   docker compose up -d --build   # app + redis + postgres
   

   Or run bare-metal against your own Redis with gunicorn app:app—MaestroApp subclasses Flask, so it's directly servable by any WSGI server. Run exactly one worker process (gunicorn's default)—the websocket listener and scheduler live in-process, and multiple workers would run your automations multiple times.

   You should see WebSocket authenticated successfully in the logs.

   The generated app.py wires up the full constructor from environment variables, but a minimal maestro app is just:

   import os
   
   from maestro import MaestroApp
   
   app = MaestroApp(
       hass_url=os.environ["HOME_ASSISTANT_URL"],
       hass_token=os.environ["HOME_ASSISTANT_TOKEN"],
       redis_host=os.environ["REDIS_HOST"],
       redis_port=int(os.environ["REDIS_PORT"]),
   )
   

   Everything initializes inside the constructor: configuration, logging, script loading, the job scheduler, and the Home Assistant websocket connection. How you provide settings (env vars, a config file, hardcoded values) is entirely up to your app.py.

Constructor Reference

app = MaestroApp(
    hass_url=...,                     # Required. Home Assistant base URL
    hass_token=...,                   # Required. Long-lived access token
    redis_host=...,                   # Required
    redis_port=...,                   # Required
    db_url=None,                      # Optional SQLAlchemy URL; None disables the DB entirely
    scripts_dir=Path("scripts"),      # Directory of automation modules, imported at startup
    registry_dir=Path("registry"),    # Where generated entity registry modules are written
    custom_domains_dir=None,          # Optional directory of custom Entity subclasses
    redis_key_prefix="maestro",       # Namespace for all maestro Redis keys
    timezone="America/New_York",
    background_services=True,         # False = no websocket/scheduler (REPL and shell use)
    configure_logging=True,           # False = your process owns logging configuration
    autopopulate_registry=False,      # Auto-generate registry entries as state changes arrive
    domain_ignore_list=(),            # HA domains to ignore entirely
    notify_action_mappings={},        # person entity ID -> HA notify action name
    default_notif_sound=...,          # Push notification defaults
    critical_notif_sound=...,
    default_notif_url=...,
)

Only one MaestroApp may be constructed per process. Constructing it registers the app globally—library internals and your scripts can access it via from maestro import get_app.

Writing Automations

All your automation logic goes in your scripts directory. Modules are auto-imported at startup as a package named after the directory (e.g. scripts.bedroom_lights), which registers your trigger decorators.

A Quick Note on Imports

Everything intended for use in automation scripts is exported from top-level maestro packages. You never need to import from deeper in the package structure. Your generated registry is its own top-level package in your project.

from maestro.domains import ON, OFF, HOME, AWAY        # State constants
from maestro.triggers import state_change_trigger       # Trigger decorators
from maestro.integrations import StateChangeEvent       # Event types
from maestro.utils import Notif, JobScheduler, log      # Utilities
from maestro.testing import MaestroTest                 # Test framework

from registry import switch, light, sensor              # Your generated entity instances
from custom_domains import Thermostat                   # Your custom domain classes (if any)

Basic Example

# scripts/bedroom_lights.py
from maestro.domains import OFF, ON
from maestro.triggers import state_change_trigger
from maestro.integrations import StateChangeEvent

from registry import switch, light

@state_change_trigger(switch.bedroom_motion_sensor, to_state=ON)
def motion_detected(state_change: StateChangeEvent) -> None:
    """Turn on bedroom lights when motion detected"""
    light.bedroom_ceiling.turn_on()

@state_change_trigger(switch.bedroom_motion_sensor, to_state=OFF)
def motion_cleared(state_change: StateChangeEvent) -> None:
    """Turn off bedroom lights when motion clears"""
    light.bedroom_ceiling.turn_off()

Schedule-Based Automation

# scripts/morning_routine.py
from maestro.triggers import cron_trigger

from registry import switch, climate

@cron_trigger(hour=7, minute=0)
def morning_routine() -> None:
    """Run every morning at 7:00 AM"""
    switch.coffee_maker.turn_on()
    climate.bedroom_thermostat.set_temperature(72)

Working with Entity State

from maestro.triggers import state_change_trigger
from maestro.integrations import StateChangeEvent

from registry import sensor, switch

@state_change_trigger(sensor.outdoor_temperature)
def temperature_monitor(state_change: StateChangeEvent) -> None:
    """Control fan based on temperature"""
    # Note: entity state is always returned as a string
    temp = float(state_change.new.state)

    if temp > 75:
        switch.ceiling_fan.turn_on()
    elif temp < 68:
        switch.ceiling_fan.turn_off()

Available Triggers

Trigger decorators automatically register your functions to respond to events. Decorated functions can optionally accept parameters that provide event context—these parameters are optional and the decorator works whether you include them or not.

Note: The script discovery logic ignores any module that starts with either test or an underscore _ (including __init__.py modules)

Note: If a function belongs to a class, then it must be a @staticmethod in order to be decorated.

State Change Trigger

Responds when an entity's state changes. Can trigger on one or multiple entities.

# Single entity with filter
@state_change_trigger(switch.bedroom_motion_sensor, to_state=ON)
def motion_detected(state_change: StateChangeEvent) -> None:
    log.info("Motion detected", new_state=state_change.new.state)

# Multiple entities - same handler for all
@state_change_trigger(
    sensor.living_room_temperature,
    sensor.bedroom_temperature,
    sensor.kitchen_temperature
)
def temperature_changed(state_change: StateChangeEvent) -> None:
    log.info("Temperature changed", entity_id=state_change.entity_id)

# Without parameters - still works!
@state_change_trigger(switch.bedroom_motion_sensor)
def simple_handler() -> None:
    log.info("Entity changed")

Decorator Parameters:

• *entities: One or more Entity objects or entity ID strings
• from_state: Optional - only trigger when transitioning from this state
• to_state: Optional - only trigger when transitioning to this state

Optional Function Parameters:

• state_change: StateChangeEvent - Contains old and new state information

Cron Trigger

Runs on a schedule using cron syntax.

@cron_trigger(hour=7, minute=30)
def morning_task() -> None:
    pass

# Or use cron pattern
@cron_trigger(pattern="0 */6 * * *")  # Every 6 hours
def periodic_task() -> None:
    pass

Decorator Parameters:

• pattern: Cron pattern string (e.g., "0 0 * * *")
• minute, hour, day_of_month, month, day_of_week: Individual cron fields

Note: You can use either pattern or individual fields, but not both.

Optional Function Parameters:

• None - cron triggers don't provide runtime parameters

Sun Trigger

Runs based on solar events (sunrise, solar noon, dusk, etc.) with optional time offsets. Requires the Home Assistant sun integration (the sun.sun entity).

from datetime import timedelta
from maestro.triggers import sun_trigger, SolarEvent

# At sunrise
@sun_trigger(SolarEvent.SUNRISE)
def sunrise_routine() -> None:
    switch.outdoor_lights.turn_off()

# 30 minutes before sunset
@sun_trigger(SolarEvent.DUSK, offset=timedelta(minutes=-30))
def pre_dusk() -> None:
    switch.patio_lights.turn_on()

Decorator Parameters:

• solar_event: SolarEvent enum value (DAWN, DUSK, SOLAR_MIDNIGHT, SOLAR_NOON, SUNRISE, SUNSET)
• offset: Optional timedelta (negative for before, positive for after). Max ±12 hours.

Optional Function Parameters:

• None - sun triggers don't provide runtime parameters

Event Fired Trigger

Responds to custom Home Assistant events. Supports filtering by user ID and arbitrary event data key-value pairs.

# Basic event trigger
@event_fired_trigger("my_custom_event")
def handle_event(event: FiredEvent) -> None:
    log.info("Event received", data=event.data)

# With event data filtering - only fires when event.data["trigger"] matches
@event_fired_trigger("my_custom_event", trigger="button_press")
def handle_button(event: FiredEvent) -> None:
    log.info("Button pressed")

Decorator Parameters:

• event_type: String matching the Home Assistant event type
• user_id: Optional - filter by user who triggered the event
• **event_data: Optional key-value filters applied to the event's data dict

Optional Function Parameters:

• event: FiredEvent - Contains event data and context

Notification Action Trigger

Responds to notification actions (e.g., buttons pressed on push notifications).

@notif_action_trigger("UNLOCK_DOOR")
def handle_unlock(notif_action: NotifActionEvent) -> None:
    door_id = notif_action.action_data.get("door_id")
    if door_id == "front_door":
        lock.front_door.unlock()

Decorator Parameters:

• action: String matching the notification action ID
• device_id: Optional - filter by specific device

Optional Function Parameters:

• notif_action: NotifActionEvent - Contains action data and device info

Maestro Trigger

Runs on Maestro service lifecycle events.

from maestro.triggers import maestro_trigger, MaestroEvent

@maestro_trigger(MaestroEvent.STARTUP)
def on_startup() -> None:
    """Runs when Maestro service starts"""
    log.info("Maestro started")

@maestro_trigger(MaestroEvent.SHUTDOWN)
def on_shutdown() -> None:
    """Runs when Maestro service shuts down"""
    log.info("Maestro shutting down")

Home Assistant Trigger

Runs on Home Assistant lifecycle events (HA starting up or shutting down).

from maestro.triggers import hass_trigger, HassEvent

@hass_trigger(HassEvent.STARTUP)
def on_ha_start() -> None:
    """Runs when Home Assistant starts"""
    log.info("Home Assistant started")

These are distinct from Maestro triggers—HA can restart independently while Maestro keeps running.

Entity Registry

Maestro generates a typed entity registry from your Home Assistant instance into your project's registry_dir. Access entities with full autocomplete:

from registry import switch, light, climate, sensor

# All your entities are available as typed objects
switch.living_room_lamp.turn_on()
climate.bedroom_thermostat.set_temperature(72)
temp = sensor.outdoor_temperature.state

Each entity has typed attributes accessible as properties:

# State is always a string
state = sensor.outdoor_temperature.state

# Attributes are typed via EntityAttribute descriptors
battery = sensor.outdoor_temperature.battery_level   # int
friendly_name = sensor.outdoor_temperature.friendly_name  # str

With autopopulate_registry=True, registry entries are created and refreshed automatically as entity state changes arrive. RegistryManager.prune() removes entries for entities that no longer exist in Home Assistant. The generated modules live in your project—committing or gitignoring them is your call.

Entity Methods

Common methods available on entity objects:

Light

# rgb_color and temperature are required; brightness and transition have defaults
light.bedroom.turn_on(rgb_color=(255, 200, 150), temperature=3000)
light.bedroom.turn_on(rgb_color=(255, 200, 150), temperature=3000, brightness_percent=50)
light.bedroom.turn_off()
light.bedroom.turn_off(transition_seconds=5)
light.bedroom.toggle()

Climate

climate.thermostat.set_temperature(72)
climate.thermostat.set_hvac_mode("heat")
climate.thermostat.set_fan_mode("auto")
climate.thermostat.set_preset_mode("away")
climate.thermostat.turn_on()
climate.thermostat.turn_off()

Switch

switch.coffee_maker.turn_on()
switch.coffee_maker.turn_off()
switch.coffee_maker.toggle()
switch.coffee_maker.is_on  # bool property

Lock

lock.front_door.lock()
lock.front_door.unlock()

Cover

cover.garage_door.open_cover()
cover.garage_door.close_cover()
cover.garage_door.stop_cover()
cover.garage_door.toggle()

Fan

fan.ceiling_fan.turn_on()
fan.ceiling_fan.turn_off()
fan.ceiling_fan.set_speed(Fan.Speed.MEDIUM)  # LOW=33, MEDIUM=66, HIGH=100

Media Player

media_player.living_room.turn_on()
media_player.living_room.turn_off()
media_player.living_room.play(content_id="...", content_type="music")
media_player.living_room.pause()
media_player.living_room.set_volume(0.5)  # 0.0 to 1.0
media_player.living_room.next()
media_player.living_room.previous()

Custom Domain Subclasses

You can extend built-in domain classes with device-specific or integration-specific functionality. Custom domains live in your project's custom_domains_dir (pass it to the MaestroApp constructor); Maestro imports the package at startup, before your registry and scripts load.

Important: Custom domain modules import their parent classes from maestro.domains. Your scripts import your custom classes directly from your package (e.g., from custom_domains import TeslaHVAC).

Example: Attribute Value Enums

Add type safety when passing argument values:

# custom_domains/climate.py
from enum import StrEnum, auto
from typing import override

from maestro.domains import Climate


class TeslaHVAC(Climate):
    class HVACMode(StrEnum):
        OFF = auto()
        HEAT_COOL = auto()

    class PresetMode(StrEnum):
        NORMAL = auto()
        DOG = auto()
        CAMP = auto()

    @override
    def set_hvac_mode(self, mode: HVACMode) -> None:  # type:ignore[override]
        self.perform_action("set_hvac_mode", hvac_mode=mode)

    @override
    def set_preset_mode(self, mode: PresetMode) -> None:  # type:ignore[override]
        self.perform_action("set_preset_mode", preset_mode=mode)

Example: Extended Action Methods

When a HA integration exposes actions under a different domain:

# custom_domains/sonos_speaker.py
from maestro.domains import MediaPlayer
from maestro.integrations import Domain


class SonosSpeaker(MediaPlayer):
    def join(self, members: list["SonosSpeaker"]) -> None:
        speaker_ids = [m.id for m in members]
        self.perform_action("join", group_members=speaker_ids)

    def unjoin(self) -> None:
        self.perform_action("unjoin")

    def snapshot(self, with_group: bool = False) -> None:
        # Sonos snapshot is under the "sonos" domain, not "media_player"
        self.state_manager.hass_client.perform_action(
            domain=Domain.SONOS, action="snapshot",
            entity_id=self.id, with_group=with_group,
        )

Custom domain classes must be exported from your package's __init__.py:

# custom_domains/__init__.py
from .climate import TeslaHVAC
from .sonos_speaker import SonosSpeaker

__all__ = [
    TeslaHVAC.__name__,
    SonosSpeaker.__name__,
]

To use a custom class as an entity's registry parent, edit the generated registry class's parent (e.g. class MediaPlayerKitchen(SonosSpeaker):). The registry generator preserves custom parents through updates and imports them from your custom domains package.

Sending Push Notifications

Maestro includes a push notification system for iOS devices via Home Assistant. Supports priorities, actionable buttons, custom sounds, and grouping. Notification routing uses the notify_action_mappings constructor parameter to map person entity IDs to HA notify actions (e.g. {"person.john_doe": "mobile_app_johns_iphone"}).

Basic Notification

from maestro.utils import Notif

from registry import person

# Create and send a notification
Notif(
    message="Motion detected in living room",
    title="Security Alert",
    priority=Notif.Priority.TIME_SENSITIVE,
).send(person.john_doe)

# Send to multiple people (variadic, not a list)
Notif(
    message="Good morning!",
    title="Daily Briefing",
).send(person.john_doe, person.jane_doe)

Notification Priorities

# PASSIVE      - Silent, no wake screen
# ACTIVE       - Standard notification (default)
# TIME_SENSITIVE - Bypasses notification summary
# CRITICAL     - Plays sound even on silent mode

Notif(
    message="Critical alert",
    priority=Notif.Priority.CRITICAL,
).send(person.john_doe)

Actionable Notifications

# Build action buttons
unlock_action = Notif.build_action(
    name="UNLOCK_DOOR",
    title="Unlock Door",
    destructive=False,
    require_auth=True,
)
ignore_action = Notif.build_action(
    name="IGNORE",
    title="Ignore",
    destructive=True,
)

# Send notification with actions
Notif(
    message="Someone is at the door",
    title="Doorbell",
    actions=[unlock_action, ignore_action],
    action_data={"door_id": "front_door"},  # Passed back to trigger handler
).send(person.john_doe)

Handling Notification Actions

from maestro.triggers import notif_action_trigger
from maestro.integrations import NotifActionEvent

from registry import lock

@notif_action_trigger("UNLOCK_DOOR")
def handle_unlock(notif_action: NotifActionEvent) -> None:
    door_id = notif_action.action_data.get("door_id")
    if door_id == "front_door":
        lock.front_door.unlock()

Advanced Options

Notif(
    message="Message content",
    title="Title",
    priority=Notif.Priority.ACTIVE,
    group="alarm_system",               # Group related notifications
    tag="front_door",                   # Replace previous notification with same tag
    sound="alarm.caf",                  # Custom sound file (None for silent)
    url="/lovelace-mobile/cameras",     # Open specific view when tapped
)

Scheduling Jobs

Schedule one-off functions to run at a specific time. Jobs persist across restarts via Redis.

from datetime import timedelta
from maestro.utils import JobScheduler, local_now

from registry import light

def delayed_light_off() -> None:
    """Turn off lights after delay"""
    light.bedroom_ceiling.turn_off()

# Schedule job to run 30 minutes from now
scheduler = JobScheduler()
run_time = local_now() + timedelta(minutes=30)
job_id = scheduler.schedule_job(run_time=run_time, func=delayed_light_off)

# Cancel if needed
scheduler.cancel_job(job_id)

Jobs can accept keyword arguments:

def set_temperature(temp: int) -> None:
    climate.thermostat.set_temperature(temp)

scheduler = JobScheduler()
scheduler.schedule_job(
    run_time=local_now() + timedelta(hours=1),
    func=set_temperature,
    func_params={"temp": 72},
)

Testing Your Automations

Maestro includes a testing framework that lets you unit test your automations without a running Home Assistant instance. Tests use in-memory mocks and require no Redis or HA connection. The mt fixture registers automatically as a pytest plugin when hass-maestro is installed—no conftest wiring needed.

pip install hass-maestro[test]   # adds pytest + freezegun

Key Features:

• Automatic mocking - All entities automatically use test mocks, no manual setup required
• In-memory state - State and actions are tracked in memory, no external dependencies
• Trigger simulation - Test state changes, events, and notification actions
• Action assertions - Verify that your automations call the correct Home Assistant actions

Quick Example

# scripts/tests/test_bedroom_lights.py
from maestro.domains import OFF, ON
from maestro.integrations import Domain
from maestro.testing import MaestroTest

from registry import light, switch
from scripts import bedroom_lights  # Import the module under test to register its triggers

def test_motion_turns_on_light(mt: MaestroTest) -> None:
    """Test that motion triggers the bedroom light"""
    mt.set_state(switch.bedroom_motion_sensor, OFF)
    mt.set_state(light.bedroom_ceiling, OFF)

    mt.trigger_state_change(switch.bedroom_motion_sensor, old=OFF, new=ON)

    mt.assert_action_called(Domain.LIGHT, "turn_on", entity_id=light.bedroom_ceiling.id)

Note: You must import the script module(s) you want to test. This registers the trigger decorators so they can be invoked by trigger_state_change() and similar methods.

Common Testing Patterns

Test state changes:

def test_temperature_control(mt: MaestroTest) -> None:
    mt.set_state(sensor.temperature, "70")
    mt.trigger_state_change(sensor.temperature, old="70", new="76")
    mt.assert_action_called(Domain.SWITCH, "turn_on", entity_id=switch.fan.id)

Test custom events:

def test_custom_event(mt: MaestroTest) -> None:
    mt.trigger_event("my_custom_event", data={"duration": 30})
    mt.assert_action_called(Domain.NOTIFY, person.john.notify_action_name)

Test multiple scenarios in one test:

def test_motion_sequence(mt: MaestroTest) -> None:
    mt.set_state(light.bedroom, OFF)

    mt.trigger_state_change(switch.motion, old=OFF, new=ON)
    mt.assert_action_called(Domain.LIGHT, "turn_on")

    mt.clear_action_calls()

    mt.trigger_state_change(switch.motion, old=ON, new=OFF)
    mt.assert_action_called(Domain.LIGHT, "turn_off")

Running Tests

# Run all script tests
pytest scripts

# Run a specific test file
pytest scripts/tests/test_bedroom_lights.py

# Run a single test function
pytest scripts/tests/test_bedroom_lights.py::test_motion_turns_on_light

Development Workflow

Interactive Shell

Construct the app with background_services=False to get a fully-loaded environment (scripts, registry, DB) without the websocket or scheduler—ideal for a REPL or flask shell:

app = MaestroApp(..., background_services=False)

Since MaestroApp is a Flask app, @app.shell_context_processor works as usual for pre-loading imports into flask shell.

Working with the Database

Pass db_url to enable persistent storage via Flask-SQLAlchemy (any SQLAlchemy-supported database; pip install hass-maestro[postgres] for PostgreSQL). Omit it and Maestro runs with no database at all.

Define models:

# scripts/my_automation/models.py
from typing import ClassVar

from maestro import db


class MyModel(db.Model):  # type:ignore[name-defined]
    __tablename__ = "my_table"
    __table_args__: ClassVar = {"extend_existing": True}

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)

Create tables:

with app.app_context():
    db.create_all()  # Creates all tables from imported models

Query data:

from maestro import db
from scripts.my_automation.models import MyModel

# Query
items = db.session.query(MyModel).all()
item = db.session.query(MyModel).filter_by(name="example").first()

# Insert
new_item = MyModel(name="test")
db.session.add(new_item)
db.session.commit()

# Delete
db.session.delete(item)
db.session.commit()

Note: Maestro doesn't manage migrations. Schema changes require manual SQL or db.drop_all() + db.create_all() (loses data).

Example: Complete Automation

# scripts/bedtime_routine/models.py
from typing import ClassVar

from maestro import db


class SnoozeHistory(db.Model):  # type:ignore[name-defined]
    __tablename__ = "snooze_history"
    __table_args__: ClassVar = {"extend_existing": True}

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    timestamp = db.Column(db.DateTime, nullable=False)
    duration_minutes = db.Column(db.Integer, nullable=False)

# scripts/bedtime_routine/bedtime.py
from maestro import db
from maestro.domains import HOME
from maestro.integrations import StateChangeEvent, NotifActionEvent
from maestro.triggers import state_change_trigger, cron_trigger, notif_action_trigger
from maestro.utils import Notif, JobScheduler, local_now

from registry import switch, light, climate, person
from .models import SnoozeHistory

SNOOZE_MINUTES = 15
SNOOZE_JOB_ID = "bedtime_snooze"


@cron_trigger(hour=22, minute=0)
def bedtime_warning() -> None:
    """Notify 30 min before lights out"""
    snooze_action = Notif.build_action(name="SNOOZE_BEDTIME", title="Snooze 15 min")
    dismiss_action = Notif.build_action(name="DISMISS_BEDTIME", title="Dismiss")

    Notif(
        title="Bedtime Soon",
        message="Lights will turn off in 30 minutes",
        actions=[snooze_action, dismiss_action],
    ).send(person.john)


@cron_trigger(hour=22, minute=30)
def bedtime_routine() -> None:
    """Execute bedtime routine"""
    light.bedroom.turn_on(brightness_percent=20)
    light.living_room.turn_off()
    climate.bedroom.set_temperature(68)
    switch.sound_machine.turn_on()


@notif_action_trigger("SNOOZE_BEDTIME")
def snooze_bedtime_routine(notif_action: NotifActionEvent) -> None:
    """Delay bedtime routine and track snooze history"""
    snooze = SnoozeHistory(timestamp=local_now(), duration_minutes=SNOOZE_MINUTES)
    db.session.add(snooze)
    db.session.commit()

    Notif(
        title="Bedtime Snoozed",
        message=f"Routine delayed by {SNOOZE_MINUTES} minutes",
    ).send(person.john)


@state_change_trigger(person.john, to_state=HOME)
def welcome_home(state_change: StateChangeEvent) -> None:
    """Turn on lights if arriving home late"""
    hour = local_now().hour
    if 22 <= hour <= 23:
        light.living_room.turn_on()

Troubleshooting

WebSocket Connection Issues

Problem: Maestro can't connect to Home Assistant WebSocket

Solutions:

1. Verify hass_url is correct (e.g., http://192.168.1.100:8123)
2. Verify hass_token is a valid long-lived access token
3. Check Home Assistant is accessible: curl http://<your-ha-url>/api/
4. Review your service logs

Problem: WebSocket keeps reconnecting

Solutions:

1. Check network stability between Maestro and Home Assistant
2. Verify Home Assistant isn't restarting frequently
3. Check Home Assistant logs for WebSocket errors

Problem: Events seem to be missing

After reconnection, Maestro automatically syncs all entity states if it was disconnected for more than 30 minutes. If you still see issues:

1. Check logs for "State sync completed" message after reconnection
2. Verify the entity's domain is not in domain_ignore_list

Contributing

Maestro is actively developed but doesn't yet cover all Home Assistant domains, actions, and features. If you find something missing:

• Entity domains: Add new domain classes in maestro/domains/
• Entity methods: Extend existing domain classes with additional actions
• Trigger types: Implement new trigger decorators in maestro/triggers/
• Utilities: Add helpful utilities in maestro/utils/

Pull requests are welcome!

License

Creative Commons Attribution-NonCommercial 4.0 - see LICENSE

Support

For issues and questions, see the GitHub Issues.