orchestrata_ultimate.orchestrator.impact.cli

Reference › orchestrata_ultimate › orchestrator › impact › cli

A command-line tool for running impact analysis by hand. main joins the command-line arguments into the intent text (and exits with a usage message if there are none), runs IntentImpactAnalyzer.analyze on it, and prints the report in four parts: the classification, the scope, the parallel plan and the full report as JSON, through _print_classification, _print_scope, _print_plan and _print_json. The analyzer is always closed afterwards. No other module imports it.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/orchestrator/impact/cli.py

#!/usr/bin/env python3
"""CLI tool to analyze intent impact.

Usage:
    python3 -m orchestrata.orchestrator.impact.cli "intent text"
"""

import asyncio
import json
import sys

from orchestrata_ultimate.orchestrator.impact.analyzer import (
    IntentImpactAnalyzer,
)
from orchestrata_ultimate.orchestrator.impact.types import ImpactReport


async def main() -> None:
    """Entry point for intent impact CLI."""
    if len(sys.argv) < 2:
        print("Usage: impact-cli <intent_text>")
        sys.exit(1)

    intent_text = " ".join(sys.argv[1:])
    print(f"Analyzing intent: {intent_text}")
    print("=" * 60)

    analyzer = IntentImpactAnalyzer()
    try:
        report = await analyzer.analyze(intent_text)
        _print_classification(report)
        _print_scope(report)
        _print_plan(report)
        _print_json(report)
    finally:
        await analyzer.close()


def _print_classification(report: ImpactReport) -> None:
    """Print classification section of the report.

    Args:
        report: The impact analysis report.
    """
    cls = report.classification
    print("\n[Classification]")
    print(f"  Type: {cls.type.value}")
    print(f"  Confidence: {cls.confidence:.2f}")
    if cls.target_package:
        print(f"  Target Package: {cls.target_package}")
    if cls.entities:
        print(f"  Entities: {', '.join(cls.entities[:5])}")


def _print_scope(report: ImpactReport) -> None:
    """Print impact scope section of the report.

    Args:
        report: The impact analysis report.
    """
    scope = report.scope
    print("\n[Impact Scope]")
    print(f"  Severity: {scope.severity.value}")
    print(f"  Affected Packages: {len(scope.affected_packages)}")
    print(f"  Affected Files: {len(scope.affected_files)}")
    print(f"  Upstream Dependencies: {len(scope.upstream_dependencies)}")
    print(f"  Downstream Dependencies: {len(scope.downstream_dependencies)}")
    print(f"  Test Files: {len(scope.test_files)}")


def _print_plan(report: ImpactReport) -> None:
    """Print execution plan section of the report.

    Args:
        report: The impact analysis report.
    """
    if not report.parallel_plan:
        return

    plan = report.parallel_plan
    print("\n[Execution Plan]")
    print(f"  Total Tasks: {plan.total_tasks}")
    print(f"  Estimated Speedup: {plan.estimated_speedup:.1f}x")
    print(f"  Parallel Branches: {len(plan.parallel_branches)}")
    for branch in plan.parallel_branches:
        print(f"    - {branch.agent_type}: {len(branch.tasks)} tasks")


def _print_json(report: ImpactReport) -> None:
    """Print full JSON report.

    Args:
        report: The impact analysis report.
    """
    print("\n" + "=" * 60)
    print("[Full Report (JSON)]")
    print(json.dumps(report.to_dict(), indent=2))


if __name__ == "__main__":
    asyncio.run(main())