forked from Rockachopa/Timmy-time-dashboard
271
timmy_automations/__init__.py
Normal file
271
timmy_automations/__init__.py
Normal file
@@ -0,0 +1,271 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user