forked from Rockachopa/Timmy-time-dashboard
317 lines
11 KiB
Python
317 lines
11 KiB
Python
"""Timmy Control Panel CLI — primary control surface for automations.
|
|
|
|
Usage:
|
|
timmyctl daily-run # Run the Daily Run orchestration
|
|
timmyctl log-run # Capture a Daily Run logbook entry
|
|
timmyctl inbox # Show what's "calling Timmy"
|
|
timmyctl config # Display key configuration
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import typer
|
|
import yaml
|
|
from rich.console import Console
|
|
from rich.table import Table
|
|
|
|
# Initialize Rich console for nice output
|
|
console = Console()
|
|
|
|
app = typer.Typer(
|
|
help="Timmy Control Panel — primary control surface for automations",
|
|
rich_markup_mode="rich",
|
|
)
|
|
|
|
# Default config paths
|
|
DEFAULT_CONFIG_DIR = Path("timmy_automations/config")
|
|
AUTOMATIONS_CONFIG = DEFAULT_CONFIG_DIR / "automations.json"
|
|
DAILY_RUN_CONFIG = DEFAULT_CONFIG_DIR / "daily_run.json"
|
|
TRIAGE_RULES_CONFIG = DEFAULT_CONFIG_DIR / "triage_rules.yaml"
|
|
|
|
|
|
def _load_json_config(path: Path) -> dict[str, Any]:
|
|
"""Load a JSON config file, returning empty dict on error."""
|
|
try:
|
|
with open(path, encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except (FileNotFoundError, json.JSONDecodeError) as e:
|
|
console.print(f"[red]Error loading {path}: {e}[/red]")
|
|
return {}
|
|
|
|
|
|
def _load_yaml_config(path: Path) -> dict[str, Any]:
|
|
"""Load a YAML config file, returning empty dict on error."""
|
|
try:
|
|
with open(path, encoding="utf-8") as f:
|
|
return yaml.safe_load(f) or {}
|
|
except (FileNotFoundError, yaml.YAMLError) as e:
|
|
console.print(f"[red]Error loading {path}: {e}[/red]")
|
|
return {}
|
|
|
|
|
|
def _get_config_dir() -> Path:
|
|
"""Return the config directory path."""
|
|
# Allow override via environment variable
|
|
env_dir = os.environ.get("TIMMY_CONFIG_DIR")
|
|
if env_dir:
|
|
return Path(env_dir)
|
|
return DEFAULT_CONFIG_DIR
|
|
|
|
|
|
@app.command()
|
|
def daily_run(
|
|
dry_run: bool = typer.Option(
|
|
False, "--dry-run", "-n", help="Show what would run without executing"
|
|
),
|
|
verbose: bool = typer.Option(False, "--verbose", "-v", help="Show detailed output"),
|
|
):
|
|
"""Run the Daily Run orchestration (agenda + summary).
|
|
|
|
Executes the daily run workflow including:
|
|
- Loop Guard checks
|
|
- Cycle Retrospective
|
|
- Triage scoring (if scheduled)
|
|
- Loop introspection (if scheduled)
|
|
"""
|
|
console.print("[bold green]Timmy Daily Run[/bold green]")
|
|
console.print()
|
|
|
|
config_path = _get_config_dir() / "daily_run.json"
|
|
config = _load_json_config(config_path)
|
|
|
|
if not config:
|
|
console.print("[yellow]No daily run configuration found.[/yellow]")
|
|
raise typer.Exit(1)
|
|
|
|
schedules = config.get("schedules", {})
|
|
triggers = config.get("triggers", {})
|
|
|
|
if verbose:
|
|
console.print(f"[dim]Config loaded from: {config_path}[/dim]")
|
|
console.print()
|
|
|
|
# Show the daily run schedule
|
|
table = Table(title="Daily Run Schedules")
|
|
table.add_column("Schedule", style="cyan")
|
|
table.add_column("Description", style="green")
|
|
table.add_column("Automations", style="yellow")
|
|
|
|
for schedule_name, schedule_data in schedules.items():
|
|
automations = schedule_data.get("automations", [])
|
|
table.add_row(
|
|
schedule_name,
|
|
schedule_data.get("description", ""),
|
|
", ".join(automations) if automations else "—",
|
|
)
|
|
|
|
console.print(table)
|
|
console.print()
|
|
|
|
# Show triggers
|
|
trigger_table = Table(title="Triggers")
|
|
trigger_table.add_column("Trigger", style="cyan")
|
|
trigger_table.add_column("Description", style="green")
|
|
trigger_table.add_column("Automations", style="yellow")
|
|
|
|
for trigger_name, trigger_data in triggers.items():
|
|
automations = trigger_data.get("automations", [])
|
|
trigger_table.add_row(
|
|
trigger_name,
|
|
trigger_data.get("description", ""),
|
|
", ".join(automations) if automations else "—",
|
|
)
|
|
|
|
console.print(trigger_table)
|
|
console.print()
|
|
|
|
if dry_run:
|
|
console.print("[yellow]Dry run mode — no actions executed.[/yellow]")
|
|
else:
|
|
console.print("[green]Executing daily run automations...[/green]")
|
|
# TODO: Implement actual automation execution
|
|
# This would call the appropriate scripts from the automations config
|
|
console.print("[dim]Automation execution not yet implemented.[/dim]")
|
|
|
|
|
|
@app.command()
|
|
def log_run(
|
|
message: str = typer.Argument(..., help="Logbook entry message"),
|
|
category: str = typer.Option(
|
|
"general", "--category", "-c", help="Entry category (e.g., retro, todo, note)"
|
|
),
|
|
):
|
|
"""Capture a quick Daily Run logbook entry.
|
|
|
|
Logs a structured entry to the daily run logbook for later review.
|
|
Entries are timestamped and categorized automatically.
|
|
"""
|
|
from datetime import datetime
|
|
|
|
timestamp = datetime.now().isoformat()
|
|
|
|
console.print("[bold green]Daily Run Log Entry[/bold green]")
|
|
console.print()
|
|
console.print(f"[dim]Timestamp:[/dim] {timestamp}")
|
|
console.print(f"[dim]Category:[/dim] {category}")
|
|
console.print(f"[dim]Message:[/dim] {message}")
|
|
console.print()
|
|
|
|
# TODO: Persist to actual logbook file
|
|
# This would append to a logbook file (e.g., .loop/logbook.jsonl)
|
|
console.print("[green]✓[/green] Entry logged (simulated)")
|
|
|
|
|
|
@app.command()
|
|
def inbox(
|
|
limit: int = typer.Option(10, "--limit", "-l", help="Maximum items to show"),
|
|
include_prs: bool = typer.Option(True, "--prs/--no-prs", help="Show open PRs"),
|
|
include_issues: bool = typer.Option(True, "--issues/--no-issues", help="Show relevant issues"),
|
|
):
|
|
"""Show what's "calling Timmy" — PRs, Daily Run items, alerts.
|
|
|
|
Displays a unified inbox of items requiring attention:
|
|
- Open pull requests awaiting review
|
|
- Daily run queue items
|
|
- Alerts and notifications
|
|
"""
|
|
console.print("[bold green]Timmy Inbox[/bold green]")
|
|
console.print()
|
|
|
|
# Load automations to show what's enabled
|
|
config_path = _get_config_dir() / "automations.json"
|
|
config = _load_json_config(config_path)
|
|
|
|
automations = config.get("automations", [])
|
|
enabled_automations = [a for a in automations if a.get("enabled", False)]
|
|
|
|
# Show automation status
|
|
auto_table = Table(title="Active Automations")
|
|
auto_table.add_column("ID", style="cyan")
|
|
auto_table.add_column("Name", style="green")
|
|
auto_table.add_column("Category", style="yellow")
|
|
auto_table.add_column("Trigger", style="magenta")
|
|
|
|
for auto in enabled_automations[:limit]:
|
|
auto_table.add_row(
|
|
auto.get("id", ""),
|
|
auto.get("name", ""),
|
|
"✓" if auto.get("enabled", False) else "✗",
|
|
auto.get("category", ""),
|
|
)
|
|
|
|
console.print(auto_table)
|
|
console.print()
|
|
|
|
# TODO: Fetch actual PRs from Gitea API
|
|
if include_prs:
|
|
pr_table = Table(title="Open Pull Requests (placeholder)")
|
|
pr_table.add_column("#", style="cyan")
|
|
pr_table.add_column("Title", style="green")
|
|
pr_table.add_column("Author", style="yellow")
|
|
pr_table.add_column("Status", style="magenta")
|
|
pr_table.add_row("—", "[dim]No PRs fetched (Gitea API not configured)[/dim]", "—", "—")
|
|
console.print(pr_table)
|
|
console.print()
|
|
|
|
# TODO: Fetch relevant issues from Gitea API
|
|
if include_issues:
|
|
issue_table = Table(title="Issues Calling for Attention (placeholder)")
|
|
issue_table.add_column("#", style="cyan")
|
|
issue_table.add_column("Title", style="green")
|
|
issue_table.add_column("Type", style="yellow")
|
|
issue_table.add_column("Priority", style="magenta")
|
|
issue_table.add_row(
|
|
"—", "[dim]No issues fetched (Gitea API not configured)[/dim]", "—", "—"
|
|
)
|
|
console.print(issue_table)
|
|
console.print()
|
|
|
|
|
|
@app.command()
|
|
def config(
|
|
key: str | None = typer.Argument(None, help="Show specific config key (e.g., 'automations')"),
|
|
show_rules: bool = typer.Option(False, "--rules", "-r", help="Show triage rules overview"),
|
|
):
|
|
"""Display key configuration — labels, logbook issue ID, token rules overview.
|
|
|
|
Shows the current Timmy automation configuration including:
|
|
- Automation manifest
|
|
- Daily run schedules
|
|
- Triage scoring rules
|
|
"""
|
|
console.print("[bold green]Timmy Configuration[/bold green]")
|
|
console.print()
|
|
|
|
config_dir = _get_config_dir()
|
|
|
|
if key == "automations" or key is None:
|
|
auto_config = _load_json_config(config_dir / "automations.json")
|
|
automations = auto_config.get("automations", [])
|
|
|
|
table = Table(title="Automations Manifest")
|
|
table.add_column("ID", style="cyan")
|
|
table.add_column("Name", style="green")
|
|
table.add_column("Enabled", style="yellow")
|
|
table.add_column("Category", style="magenta")
|
|
|
|
for auto in automations:
|
|
enabled = "✓" if auto.get("enabled", False) else "✗"
|
|
table.add_row(
|
|
auto.get("id", ""),
|
|
auto.get("name", ""),
|
|
enabled,
|
|
auto.get("category", ""),
|
|
)
|
|
|
|
console.print(table)
|
|
console.print()
|
|
|
|
if key == "daily_run" or (key is None and not show_rules):
|
|
daily_config = _load_json_config(config_dir / "daily_run.json")
|
|
|
|
if daily_config:
|
|
console.print("[bold]Daily Run Configuration:[/bold]")
|
|
console.print(f"[dim]Version:[/dim] {daily_config.get('version', 'unknown')}")
|
|
console.print(f"[dim]Description:[/dim] {daily_config.get('description', '')}")
|
|
console.print()
|
|
|
|
if show_rules or key == "triage_rules":
|
|
rules_config = _load_yaml_config(config_dir / "triage_rules.yaml")
|
|
|
|
if rules_config:
|
|
thresholds = rules_config.get("thresholds", {})
|
|
console.print("[bold]Triage Scoring Rules:[/bold]")
|
|
console.print(f" Ready threshold: {thresholds.get('ready', 'N/A')}")
|
|
console.print(f" Excellent threshold: {thresholds.get('excellent', 'N/A')}")
|
|
console.print()
|
|
|
|
scope = rules_config.get("scope", {})
|
|
console.print("[bold]Scope Scoring:[/bold]")
|
|
console.print(f" Meta penalty: {scope.get('meta_penalty', 'N/A')}")
|
|
console.print()
|
|
|
|
alignment = rules_config.get("alignment", {})
|
|
console.print("[bold]Alignment Scoring:[/bold]")
|
|
console.print(f" Bug score: {alignment.get('bug_score', 'N/A')}")
|
|
console.print(f" Refactor score: {alignment.get('refactor_score', 'N/A')}")
|
|
console.print(f" Feature score: {alignment.get('feature_score', 'N/A')}")
|
|
console.print()
|
|
|
|
quarantine = rules_config.get("quarantine", {})
|
|
console.print("[bold]Quarantine Rules:[/bold]")
|
|
console.print(f" Failure threshold: {quarantine.get('failure_threshold', 'N/A')}")
|
|
console.print(f" Lookback cycles: {quarantine.get('lookback_cycles', 'N/A')}")
|
|
console.print()
|
|
|
|
|
|
def main():
|
|
"""Entry point for the timmyctl CLI."""
|
|
app()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|