272 lines
9.4 KiB
Python
272 lines
9.4 KiB
Python
"""Timmy Automations — Central automation discovery and control module.
|
|
|
|
This module provides:
|
|
- Discovery of all configured automations
|
|
- Enable/disable control
|
|
- Status reporting
|
|
- Configuration management
|
|
|
|
Usage:
|
|
from timmy_automations import AutomationRegistry
|
|
|
|
registry = AutomationRegistry()
|
|
for auto in registry.list_automations():
|
|
print(f"{auto.id}: {auto.name} ({'enabled' if auto.enabled else 'disabled'})")
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
@dataclass
|
|
class Automation:
|
|
"""Represents a single automation configuration."""
|
|
|
|
id: str
|
|
name: str
|
|
description: str
|
|
script: str
|
|
category: str
|
|
enabled: bool
|
|
trigger: str
|
|
executable: str
|
|
config: dict[str, Any]
|
|
outputs: list[str]
|
|
depends_on: list[str]
|
|
schedule: str | None = None
|
|
|
|
@property
|
|
def full_script_path(self) -> Path:
|
|
"""Resolve the script path relative to repo root."""
|
|
repo_root = Path(__file__).parent.parent
|
|
return repo_root / self.script
|
|
|
|
@property
|
|
def is_executable(self) -> bool:
|
|
"""Check if the script file exists and is executable."""
|
|
path = self.full_script_path
|
|
return path.exists() and os.access(path, os.X_OK)
|
|
|
|
@property
|
|
def is_runnable(self) -> bool:
|
|
"""Check if automation can be run (enabled + executable)."""
|
|
return self.enabled and self.is_executable
|
|
|
|
|
|
class AutomationRegistry:
|
|
"""Registry for discovering and managing Timmy automations."""
|
|
|
|
MANIFEST_PATH = Path(__file__).parent / "config" / "automations.json"
|
|
STATE_PATH = Path(__file__).parent.parent / ".loop" / "automation_state.json"
|
|
|
|
def __init__(self, manifest_path: Path | None = None) -> None:
|
|
"""Initialize the registry, loading the manifest.
|
|
|
|
Args:
|
|
manifest_path: Optional override for manifest file location.
|
|
"""
|
|
self._manifest_path = manifest_path or self.MANIFEST_PATH
|
|
self._automations: dict[str, Automation] = {}
|
|
self._load_manifest()
|
|
|
|
def _load_manifest(self) -> None:
|
|
"""Load automations from the manifest file."""
|
|
if not self._manifest_path.exists():
|
|
self._automations = {}
|
|
return
|
|
|
|
try:
|
|
data = json.loads(self._manifest_path.read_text())
|
|
for auto_data in data.get("automations", []):
|
|
auto = Automation(
|
|
id=auto_data["id"],
|
|
name=auto_data["name"],
|
|
description=auto_data["description"],
|
|
script=auto_data["script"],
|
|
category=auto_data["category"],
|
|
enabled=auto_data.get("enabled", True),
|
|
trigger=auto_data["trigger"],
|
|
executable=auto_data.get("executable", "python3"),
|
|
config=auto_data.get("config", {}),
|
|
outputs=auto_data.get("outputs", []),
|
|
depends_on=auto_data.get("depends_on", []),
|
|
schedule=auto_data.get("schedule"),
|
|
)
|
|
self._automations[auto.id] = auto
|
|
except (json.JSONDecodeError, KeyError) as e:
|
|
raise AutomationError(f"Failed to load manifest: {e}")
|
|
|
|
def _save_manifest(self) -> None:
|
|
"""Save current automation states back to manifest."""
|
|
data = {
|
|
"version": "1.0.0",
|
|
"description": "Master manifest of all Timmy automations",
|
|
"last_updated": "2026-03-21",
|
|
"automations": []
|
|
}
|
|
|
|
for auto in self._automations.values():
|
|
auto_dict = {
|
|
"id": auto.id,
|
|
"name": auto.name,
|
|
"description": auto.description,
|
|
"script": auto.script,
|
|
"category": auto.category,
|
|
"enabled": auto.enabled,
|
|
"trigger": auto.trigger,
|
|
"executable": auto.executable,
|
|
"config": auto.config,
|
|
"outputs": auto.outputs,
|
|
"depends_on": auto.depends_on,
|
|
}
|
|
if auto.schedule:
|
|
auto_dict["schedule"] = auto.schedule
|
|
data["automations"].append(auto_dict)
|
|
|
|
self._manifest_path.write_text(json.dumps(data, indent=2) + "\n")
|
|
|
|
def list_automations(
|
|
self,
|
|
category: str | None = None,
|
|
enabled_only: bool = False,
|
|
trigger: str | None = None,
|
|
) -> list[Automation]:
|
|
"""List automations with optional filtering.
|
|
|
|
Args:
|
|
category: Filter by category (daily_run, triage, metrics, workspace)
|
|
enabled_only: Only return enabled automations
|
|
trigger: Filter by trigger type (pre_cycle, post_cycle, scheduled, manual)
|
|
|
|
Returns:
|
|
List of matching Automation objects.
|
|
"""
|
|
results = []
|
|
for auto in self._automations.values():
|
|
if category and auto.category != category:
|
|
continue
|
|
if enabled_only and not auto.enabled:
|
|
continue
|
|
if trigger and auto.trigger != trigger:
|
|
continue
|
|
results.append(auto)
|
|
return sorted(results, key=lambda a: (a.category, a.name))
|
|
|
|
def get_automation(self, automation_id: str) -> Automation | None:
|
|
"""Get a specific automation by ID."""
|
|
return self._automations.get(automation_id)
|
|
|
|
def enable(self, automation_id: str) -> bool:
|
|
"""Enable an automation.
|
|
|
|
Returns:
|
|
True if automation was found and enabled, False otherwise.
|
|
"""
|
|
if automation_id not in self._automations:
|
|
return False
|
|
self._automations[automation_id].enabled = True
|
|
self._save_manifest()
|
|
return True
|
|
|
|
def disable(self, automation_id: str) -> bool:
|
|
"""Disable an automation.
|
|
|
|
Returns:
|
|
True if automation was found and disabled, False otherwise.
|
|
"""
|
|
if automation_id not in self._automations:
|
|
return False
|
|
self._automations[automation_id].enabled = False
|
|
self._save_manifest()
|
|
return True
|
|
|
|
def get_by_trigger(self, trigger: str) -> list[Automation]:
|
|
"""Get all automations for a specific trigger."""
|
|
return [a for a in self._automations.values() if a.trigger == trigger]
|
|
|
|
def get_by_schedule(self, schedule: str) -> list[Automation]:
|
|
"""Get all automations for a specific schedule."""
|
|
return [
|
|
a for a in self._automations.values()
|
|
if a.schedule == schedule
|
|
]
|
|
|
|
def validate_all(self) -> list[tuple[str, str]]:
|
|
"""Validate all automations and return any issues.
|
|
|
|
Returns:
|
|
List of (automation_id, error_message) tuples.
|
|
"""
|
|
issues = []
|
|
for auto in self._automations.values():
|
|
if not auto.full_script_path.exists():
|
|
issues.append((auto.id, f"Script not found: {auto.script}"))
|
|
elif auto.enabled and not auto.is_executable:
|
|
# Check if file is readable even if not executable
|
|
if not os.access(auto.full_script_path, os.R_OK):
|
|
issues.append((auto.id, f"Script not readable: {auto.script}"))
|
|
return issues
|
|
|
|
def get_status(self) -> dict[str, Any]:
|
|
"""Get overall registry status."""
|
|
total = len(self._automations)
|
|
enabled = sum(1 for a in self._automations.values() if a.enabled)
|
|
runnable = sum(1 for a in self._automations.values() if a.is_runnable)
|
|
issues = self.validate_all()
|
|
|
|
return {
|
|
"total_automations": total,
|
|
"enabled": enabled,
|
|
"disabled": total - enabled,
|
|
"runnable": runnable,
|
|
"validation_issues": len(issues),
|
|
"issues": [{"id": i[0], "error": i[1]} for i in issues],
|
|
"categories": sorted(set(a.category for a in self._automations.values())),
|
|
}
|
|
|
|
def save_state(self) -> None:
|
|
"""Save current automation state to .loop directory."""
|
|
state = {
|
|
"automations": {
|
|
id: {
|
|
"enabled": auto.enabled,
|
|
"runnable": auto.is_runnable,
|
|
"script_exists": auto.full_script_path.exists(),
|
|
}
|
|
for id, auto in self._automations.items()
|
|
}
|
|
}
|
|
self.STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
self.STATE_PATH.write_text(json.dumps(state, indent=2) + "\n")
|
|
|
|
|
|
class AutomationError(Exception):
|
|
"""Raised when automation operations fail."""
|
|
pass
|
|
|
|
|
|
# Convenience functions for CLI usage
|
|
def list_automations(category: str | None = None, enabled_only: bool = False) -> list[Automation]:
|
|
"""List automations (convenience function)."""
|
|
return AutomationRegistry().list_automations(category, enabled_only)
|
|
|
|
|
|
def enable_automation(automation_id: str) -> bool:
|
|
"""Enable an automation (convenience function)."""
|
|
return AutomationRegistry().enable(automation_id)
|
|
|
|
|
|
def disable_automation(automation_id: str) -> bool:
|
|
"""Disable an automation (convenience function)."""
|
|
return AutomationRegistry().disable(automation_id)
|
|
|
|
|
|
def get_status() -> dict[str, Any]:
|
|
"""Get registry status (convenience function)."""
|
|
return AutomationRegistry().get_status()
|