"""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()