Add script automation and visibility options (#176)

Enhance Home Assistant script support with automation triggers and
printer card visibility control.

- Script automation: Run scripts automatically when main plug turns on/off
- Show/hide scripts on printer cards (configurable per script)
- Scripts appear in dedicated row on printer cards with quick-run buttons
- Toast notification when triggering scripts from settings/sidebar

Closes #176
This commit is contained in:
maziggy
2026-01-30 09:46:03 +01:00
parent 48c349bed1
commit 4c2cef64b3
17 changed files with 982 additions and 338 deletions
+3
View File
@@ -376,6 +376,7 @@ async def export_backup(
"schedule_on_time": plug.schedule_on_time,
"schedule_off_time": plug.schedule_off_time,
"show_in_switchbar": plug.show_in_switchbar,
"show_on_printer_card": plug.show_on_printer_card,
}
)
backup["included"].append("smart_plugs")
@@ -1295,6 +1296,7 @@ async def import_backup(
existing.schedule_on_time = plug_data.get("schedule_on_time")
existing.schedule_off_time = plug_data.get("schedule_off_time")
existing.show_in_switchbar = plug_data.get("show_in_switchbar", False)
existing.show_on_printer_card = plug_data.get("show_on_printer_card", True)
restored["smart_plugs"] += 1
else:
skipped["smart_plugs"] += 1
@@ -1324,6 +1326,7 @@ async def import_backup(
schedule_on_time=plug_data.get("schedule_on_time"),
schedule_off_time=plug_data.get("schedule_off_time"),
show_in_switchbar=plug_data.get("show_in_switchbar", False),
show_on_printer_card=plug_data.get("show_on_printer_card", True),
)
db.add(plug)
restored["smart_plugs"] += 1
+115 -14
View File
@@ -56,9 +56,21 @@ async def create_smart_plug(
raise HTTPException(400, "Printer not found")
# Check if printer already has a plug assigned
result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == data.printer_id))
if result.scalar_one_or_none():
raise HTTPException(400, "This printer already has a smart plug assigned")
# Scripts can coexist with other plugs (they're for multi-device control, not power on/off)
is_script = data.plug_type == "homeassistant" and data.ha_entity_id and data.ha_entity_id.startswith("script.")
if not is_script:
# For non-script plugs, check there's no other non-script plug assigned
result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == data.printer_id))
existing = result.scalar_one_or_none()
if existing:
# Allow if existing plug is a script
existing_is_script = (
existing.plug_type == "homeassistant"
and existing.ha_entity_id
and existing.ha_entity_id.startswith("script.")
)
if not existing_is_script:
raise HTTPException(400, "This printer already has a smart plug assigned")
plug = SmartPlug(**data.model_dump())
db.add(plug)
@@ -74,12 +86,48 @@ async def create_smart_plug(
@router.get("/by-printer/{printer_id}", response_model=SmartPlugResponse | None)
async def get_smart_plug_by_printer(printer_id: int, db: AsyncSession = Depends(get_db)):
"""Get the smart plug assigned to a printer."""
"""Get the main smart plug assigned to a printer.
When multiple plugs are assigned (e.g., a regular plug + script),
returns the main (non-script) plug for power control.
"""
result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
plug = result.scalar_one_or_none()
if not plug:
plugs = result.scalars().all()
if not plugs:
return None
return plug
# If multiple plugs, prefer the non-script one (main power plug)
for plug in plugs:
is_script = plug.plug_type == "homeassistant" and plug.ha_entity_id and plug.ha_entity_id.startswith("script.")
if not is_script:
return plug
# All are scripts, return the first one
return plugs[0]
@router.get("/by-printer/{printer_id}/scripts", response_model=list[SmartPlugResponse])
async def get_script_plugs_by_printer(printer_id: int, db: AsyncSession = Depends(get_db)):
"""Get all HA script plugs assigned to a printer.
Returns only script entities (script.*) for the printer that have
show_on_printer_card enabled.
Used to display "Run Script" buttons alongside the main power plug.
"""
result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
plugs = result.scalars().all()
# Filter to only scripts with show_on_printer_card enabled
scripts = [
plug
for plug in plugs
if plug.plug_type == "homeassistant"
and plug.ha_entity_id
and plug.ha_entity_id.startswith("script.")
and plug.show_on_printer_card
]
return scripts
# Tasmota Discovery Endpoints
@@ -287,14 +335,29 @@ async def update_smart_plug(
raise HTTPException(400, "Printer not found")
# Check if that printer already has a different plug assigned
result = await db.execute(
select(SmartPlug).where(
SmartPlug.printer_id == new_printer_id,
SmartPlug.id != plug_id,
# Scripts can coexist with other plugs
# Determine if the plug being updated is/will be a script
new_entity_id = update_data.get("ha_entity_id", plug.ha_entity_id)
new_plug_type = update_data.get("plug_type", plug.plug_type)
is_script = new_plug_type == "homeassistant" and new_entity_id and new_entity_id.startswith("script.")
if not is_script:
result = await db.execute(
select(SmartPlug).where(
SmartPlug.printer_id == new_printer_id,
SmartPlug.id != plug_id,
)
)
)
if result.scalar_one_or_none():
raise HTTPException(400, "This printer already has a smart plug assigned")
existing = result.scalar_one_or_none()
if existing:
# Allow if existing plug is a script
existing_is_script = (
existing.plug_type == "homeassistant"
and existing.ha_entity_id
and existing.ha_entity_id.startswith("script.")
)
if not existing_is_script:
raise HTTPException(400, "This printer already has a smart plug assigned")
for field, value in update_data.items():
setattr(plug, field, value)
@@ -376,6 +439,13 @@ async def control_smart_plug(
plug.last_checked = datetime.utcnow()
await db.commit()
# Trigger associated scripts if this is a main (non-script) plug
is_main_plug = not (
plug.plug_type == "homeassistant" and plug.ha_entity_id and plug.ha_entity_id.startswith("script.")
)
if is_main_plug and plug.printer_id and expected_state:
await trigger_associated_scripts(plug.printer_id, expected_state, db)
# MQTT relay - publish smart plug state change
if expected_state:
try:
@@ -401,6 +471,37 @@ async def control_smart_plug(
return {"success": True, "action": control.action}
async def trigger_associated_scripts(printer_id: int, plug_state: str, db: AsyncSession):
"""Trigger scripts linked to a printer based on main plug state change.
When the main plug turns ON, triggers scripts with auto_on=True.
When the main plug turns OFF, triggers scripts with auto_off=True.
"""
result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
plugs = result.scalars().all()
# Find scripts that should be triggered
for plug in plugs:
is_script = plug.plug_type == "homeassistant" and plug.ha_entity_id and plug.ha_entity_id.startswith("script.")
if not is_script:
continue
should_trigger = False
if plug_state == "ON" and plug.auto_on:
should_trigger = True
logger.info(f"Auto-triggering script '{plug.name}' on printer power-on")
elif plug_state == "OFF" and plug.auto_off:
should_trigger = True
logger.info(f"Auto-triggering script '{plug.name}' on printer power-off")
if should_trigger:
try:
service = await _get_service_for_plug(plug, db)
await service.turn_on(plug) # Scripts are triggered by calling turn_on
except Exception as e:
logger.error(f"Failed to trigger script '{plug.name}': {e}")
@router.get("/{plug_id}/status", response_model=SmartPlugStatus)
async def get_plug_status(plug_id: int, db: AsyncSession = Depends(get_db)):
"""Get current plug status from device including energy data."""
+73
View File
@@ -760,6 +760,79 @@ async def run_migrations(conn):
except Exception:
pass
# Migration: Remove UNIQUE constraint from smart_plugs.printer_id
# This allows HA scripts to coexist with regular plugs (scripts are for multi-device control)
# SQLite requires table recreation to drop constraints
try:
# Check if we need to migrate (if UNIQUE constraint exists)
result = await conn.execute(text("SELECT sql FROM sqlite_master WHERE type='table' AND name='smart_plugs'"))
row = result.fetchone()
if row and "printer_id INTEGER UNIQUE" in (row[0] or ""):
# Create new table without UNIQUE constraint on printer_id
await conn.execute(
text("""
CREATE TABLE smart_plugs_temp (
id INTEGER PRIMARY KEY,
name VARCHAR(100) NOT NULL,
ip_address VARCHAR(45),
plug_type VARCHAR(20) DEFAULT 'tasmota',
ha_entity_id VARCHAR(100),
ha_power_entity VARCHAR(100),
ha_energy_today_entity VARCHAR(100),
ha_energy_total_entity VARCHAR(100),
printer_id INTEGER REFERENCES printers(id) ON DELETE SET NULL,
enabled BOOLEAN NOT NULL DEFAULT 1,
auto_on BOOLEAN NOT NULL DEFAULT 1,
auto_off BOOLEAN NOT NULL DEFAULT 1,
off_delay_mode VARCHAR(20) NOT NULL DEFAULT 'time',
off_delay_minutes INTEGER NOT NULL DEFAULT 5,
off_temp_threshold INTEGER NOT NULL DEFAULT 70,
username VARCHAR(50),
password VARCHAR(100),
power_alert_enabled BOOLEAN NOT NULL DEFAULT 0,
power_alert_high FLOAT,
power_alert_low FLOAT,
power_alert_last_triggered DATETIME,
schedule_enabled BOOLEAN NOT NULL DEFAULT 0,
schedule_on_time VARCHAR(5),
schedule_off_time VARCHAR(5),
show_in_switchbar BOOLEAN DEFAULT 0,
last_state VARCHAR(10),
last_checked DATETIME,
auto_off_executed BOOLEAN NOT NULL DEFAULT 0,
auto_off_pending BOOLEAN DEFAULT 0,
auto_off_pending_since DATETIME,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL
)
""")
)
# Copy data
await conn.execute(
text("""
INSERT INTO smart_plugs_temp
SELECT id, name, ip_address, plug_type, ha_entity_id, ha_power_entity,
ha_energy_today_entity, ha_energy_total_entity, printer_id, enabled,
auto_on, auto_off, off_delay_mode, off_delay_minutes, off_temp_threshold,
username, password, power_alert_enabled, power_alert_high, power_alert_low,
power_alert_last_triggered, schedule_enabled, schedule_on_time, schedule_off_time,
show_in_switchbar, last_state, last_checked, auto_off_executed,
auto_off_pending, auto_off_pending_since, created_at, updated_at
FROM smart_plugs
""")
)
# Drop old table and rename new one
await conn.execute(text("DROP TABLE smart_plugs"))
await conn.execute(text("ALTER TABLE smart_plugs_temp RENAME TO smart_plugs"))
except Exception:
pass
# Migration: Add show_on_printer_card column to smart_plugs
try:
await conn.execute(text("ALTER TABLE smart_plugs ADD COLUMN show_on_printer_card BOOLEAN DEFAULT 1"))
except Exception:
pass
async def seed_notification_templates():
"""Seed default notification templates if they don't exist."""
+4 -5
View File
@@ -24,10 +24,8 @@ class SmartPlug(Base):
ha_energy_today_entity: Mapped[str | None] = mapped_column(String(100), nullable=True) # sensor.xxx_today
ha_energy_total_entity: Mapped[str | None] = mapped_column(String(100), nullable=True) # sensor.xxx_total
# Link to printer (1:1)
printer_id: Mapped[int | None] = mapped_column(
ForeignKey("printers.id", ondelete="SET NULL"), unique=True, nullable=True
)
# Link to printer (scripts can coexist with regular plugs for multi-device control)
printer_id: Mapped[int | None] = mapped_column(ForeignKey("printers.id", ondelete="SET NULL"), nullable=True)
# Automation settings
enabled: Mapped[bool] = mapped_column(Boolean, default=True)
@@ -54,8 +52,9 @@ class SmartPlug(Base):
schedule_on_time: Mapped[str | None] = mapped_column(String(5), nullable=True) # "HH:MM" format
schedule_off_time: Mapped[str | None] = mapped_column(String(5), nullable=True) # "HH:MM" format
# Switchbar visibility
# Visibility options
show_in_switchbar: Mapped[bool] = mapped_column(Boolean, default=False)
show_on_printer_card: Mapped[bool] = mapped_column(Boolean, default=True) # For scripts: show on printer card
# Status tracking
last_state: Mapped[str | None] = mapped_column(String(10), nullable=True) # "ON"/"OFF"
+6 -4
View File
@@ -14,7 +14,7 @@ class SmartPlugBase(BaseModel):
password: str | None = None
# Home Assistant fields (required when plug_type="homeassistant")
ha_entity_id: str | None = Field(default=None, pattern=r"^(switch|light|input_boolean)\.[a-z0-9_]+$")
ha_entity_id: str | None = Field(default=None, pattern=r"^(switch|light|input_boolean|script)\.[a-z0-9_]+$")
# Home Assistant energy sensor entities (optional, for separate energy sensors)
ha_power_entity: str | None = Field(default=None, pattern=r"^sensor\.[a-z0-9_]+$")
ha_energy_today_entity: str | None = Field(default=None, pattern=r"^sensor\.[a-z0-9_]+$")
@@ -35,8 +35,9 @@ class SmartPlugBase(BaseModel):
schedule_enabled: bool = False
schedule_on_time: str | None = Field(default=None, pattern=r"^([01]\d|2[0-3]):[0-5]\d$") # HH:MM format
schedule_off_time: str | None = Field(default=None, pattern=r"^([01]\d|2[0-3]):[0-5]\d$") # HH:MM format
# Switchbar visibility
# Visibility options
show_in_switchbar: bool = False
show_on_printer_card: bool = True # For scripts: show on printer card
@model_validator(mode="after")
def validate_plug_type_fields(self) -> "SmartPlugBase":
@@ -77,8 +78,9 @@ class SmartPlugUpdate(BaseModel):
schedule_enabled: bool | None = None
schedule_on_time: str | None = Field(default=None, pattern=r"^([01]\d|2[0-3]):[0-5]\d$")
schedule_off_time: str | None = Field(default=None, pattern=r"^([01]\d|2[0-3]):[0-5]\d$")
# Switchbar visibility
# Visibility options
show_in_switchbar: bool | None = None
show_on_printer_card: bool | None = None
class SmartPlugResponse(SmartPlugBase):
@@ -147,7 +149,7 @@ class HAEntity(BaseModel):
entity_id: str
friendly_name: str
state: str | None = None
domain: str # "switch", "light", "input_boolean"
domain: str # "switch", "light", "input_boolean", "script"
class HASensorEntity(BaseModel):
+1 -1
View File
@@ -228,7 +228,7 @@ class HomeAssistantService:
- domain: str
"""
# Default domains for smart plug control
default_domains = {"switch", "light", "input_boolean"}
default_domains = {"switch", "light", "input_boolean", "script"}
try:
async with httpx.AsyncClient(timeout=self.timeout) as client:
@@ -195,6 +195,11 @@ class SmartPlugManager:
logger.debug(f"Smart plug '{plug.name}' auto_off is disabled")
return
# Skip auto-off for HA script entities (scripts can only be triggered, not turned off)
if plug.plug_type == "homeassistant" and plug.ha_entity_id and plug.ha_entity_id.startswith("script."):
logger.debug(f"Smart plug '{plug.name}' is a HA script entity, skipping auto-off")
return
# Only auto-off on successful completion, not on failures
# This allows the user to investigate errors before power-off
if status != "completed":
@@ -573,3 +573,200 @@ class TestSmartPlugsAPI:
assert response.status_code == 400
assert "not configured" in response.json()["detail"].lower()
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_homeassistant_script_plug(self, async_client: AsyncClient):
"""Verify Home Assistant script entity can be created as a plug.
Scripts allow users to trigger HA automations that control multiple devices
(e.g., turn on printer + fan together). Scripts can only be triggered (turn_on),
not turned off.
"""
data = {
"name": "Turn On Printer Setup",
"plug_type": "homeassistant",
"ha_entity_id": "script.turn_on_printer_and_fan",
"enabled": True,
"auto_on": True,
"auto_off": False, # Scripts don't support auto_off
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 200
result = response.json()
assert result["name"] == "Turn On Printer Setup"
assert result["plug_type"] == "homeassistant"
assert result["ha_entity_id"] == "script.turn_on_printer_and_fan"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_control_homeassistant_script(
self, async_client: AsyncClient, smart_plug_factory, mock_homeassistant_service, db_session
):
"""Verify HA script entity can be triggered via control endpoint."""
plug = await smart_plug_factory(plug_type="homeassistant", ha_entity_id="script.turn_on_printer")
# Scripts use "on" action to trigger
response = await async_client.post(f"/api/v1/smart-plugs/{plug.id}/control", json={"action": "on"})
assert response.status_code == 200
result = response.json()
assert result["success"] is True
assert result["action"] == "on"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_invalid_ha_entity_domain(self, async_client: AsyncClient):
"""Verify invalid HA entity domains are rejected."""
data = {
"name": "Invalid Entity",
"plug_type": "homeassistant",
"ha_entity_id": "sensor.some_sensor", # sensor domain not allowed
"enabled": True,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 422 # Validation error
@pytest.mark.asyncio
@pytest.mark.integration
async def test_script_can_coexist_with_regular_plug(
self, async_client: AsyncClient, smart_plug_factory, printer_factory, db_session
):
"""Verify HA scripts can be assigned to printers that already have a regular plug.
Scripts are for multi-device control (e.g., turn on printer + fan together),
so they should coexist with the main power plug.
"""
# Create a printer
printer = await printer_factory(name="Test Printer")
# Create a regular Tasmota plug assigned to the printer
main_plug = await smart_plug_factory(
name="Main Power Plug",
plug_type="tasmota",
ip_address="192.168.1.100",
printer_id=printer.id,
)
assert main_plug.printer_id == printer.id
# Now try to create a script also assigned to the same printer
script_data = {
"name": "Turn On Everything",
"plug_type": "homeassistant",
"ha_entity_id": "script.turn_on_printer_setup",
"printer_id": printer.id,
"enabled": True,
}
response = await async_client.post("/api/v1/smart-plugs/", json=script_data)
# Should succeed - scripts can coexist with regular plugs
assert response.status_code == 200
result = response.json()
assert result["printer_id"] == printer.id
assert result["ha_entity_id"] == "script.turn_on_printer_setup"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_regular_plug_blocked_when_another_exists(
self, async_client: AsyncClient, smart_plug_factory, printer_factory, db_session
):
"""Verify regular plugs cannot be assigned if printer already has one."""
# Create a printer
printer = await printer_factory(name="Test Printer")
# Create a regular plug assigned to the printer
await smart_plug_factory(
name="Main Power Plug",
plug_type="tasmota",
ip_address="192.168.1.100",
printer_id=printer.id,
)
# Try to create another regular plug for the same printer
another_plug = {
"name": "Second Plug",
"plug_type": "homeassistant",
"ha_entity_id": "switch.another_plug",
"printer_id": printer.id,
"enabled": True,
}
response = await async_client.post("/api/v1/smart-plugs/", json=another_plug)
# Should fail - only one regular plug per printer
assert response.status_code == 400
assert "already has a smart plug" in response.json()["detail"]
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_scripts_by_printer_filters_by_show_on_printer_card(
self, async_client: AsyncClient, smart_plug_factory, printer_factory, db_session
):
"""Verify scripts endpoint only returns scripts with show_on_printer_card=True."""
printer = await printer_factory(name="Test Printer")
# Create a script with show_on_printer_card=True (default)
visible_script = await smart_plug_factory(
name="Visible Script",
plug_type="homeassistant",
ha_entity_id="script.visible_script",
printer_id=printer.id,
show_on_printer_card=True,
)
# Create a script with show_on_printer_card=False
await smart_plug_factory(
name="Hidden Script",
plug_type="homeassistant",
ha_entity_id="script.hidden_script",
printer_id=printer.id,
show_on_printer_card=False,
)
response = await async_client.get(f"/api/v1/smart-plugs/by-printer/{printer.id}/scripts")
assert response.status_code == 200
scripts = response.json()
# Should only return the visible script
assert len(scripts) == 1
assert scripts[0]["id"] == visible_script.id
assert scripts[0]["name"] == "Visible Script"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_script_auto_on_auto_off_fields(
self, async_client: AsyncClient, smart_plug_factory, printer_factory, db_session
):
"""Verify scripts can have auto_on and auto_off set for automation triggers."""
printer = await printer_factory(name="Test Printer")
# Create a script with custom auto_on/auto_off settings
script_data = {
"name": "Fan Control Script",
"plug_type": "homeassistant",
"ha_entity_id": "script.fan_control",
"printer_id": printer.id,
"auto_on": True,
"auto_off": False,
"show_on_printer_card": True,
}
response = await async_client.post("/api/v1/smart-plugs/", json=script_data)
assert response.status_code == 200
result = response.json()
assert result["auto_on"] is True
assert result["auto_off"] is False
assert result["show_on_printer_card"] is True
# Update the script's auto_off setting
update_response = await async_client.patch(f"/api/v1/smart-plugs/{result['id']}", json={"auto_off": True})
assert update_response.status_code == 200
updated = update_response.json()
assert updated["auto_off"] is True