Function simulate_run

src/codegraphcontext/cli/simulator.py:121–168

Run architectural mutations defined in a simulation scenario JSON file and view the resulting architecture metrics.

(
    config_path: str = typer.Argument(..., help="Path to the simulation JSON configuration file."),
    path: Optional[str] = typer.Argument(None, help="Path to the directory to simulate. Defaults to the current directory."),
    context: Optional[str] = typer.Option(None, "--context", "-c", help="Specific context to use"),
)

@simulate_app.command("run")
def simulate_run(
    config_path: str = typer.Argument(..., help="Path to the simulation JSON configuration file."),
    path: Optional[str] = typer.Argument(None, help="Path to the directory to simulate. Defaults to the current directory."),
    context: Optional[str] = typer.Option(None, "--context", "-c", help="Specific context to use"),
):
    """
    Run architectural mutations defined in a simulation scenario JSON file and view the resulting architecture metrics.
    """
    load_cgc_credentials(context)
    repo_path = get_repo_path(path)
    
    cfg_file = Path(config_path)
    if not cfg_file.exists():
        console.print(f"[bold red]Scenario configuration file not found:[/bold red] {config_path}")
        raise typer.Exit(code=1)

    try:
        with open(cfg_file, "r") as f:
            scenario = json.load(f)
    except Exception as e:
        console.print(f"[bold red]Failed to parse scenario JSON file:[/bold red] {e}")
        raise typer.Exit(code=1)

    services = _initialize_services(context)
    db_manager = services[0]

    try:
        twin = CodeGraphTwin(repo_path).load_from_db(db_manager)
        
        # Apply steps
        steps = scenario.get("steps", [])
        for step in steps:
            stype = step.get("type")
            if stype == "decompose":
                twin.decompose_service(step.get("mapping", {}))
            elif stype == "remove_dependency":
                twin.remove_dependency(step.get("source"), step.get("target"), step.get("rel_type"))
            elif stype == "add_dependency":
                twin.add_dependency(step.get("source"), step.get("target"), step.get("rel_type"))
            elif stype == "remove_node":
                twin.remove_node(step.get("node_id"))
                
        summary = twin.get_metrics_summary()
        print_metrics(summary, f"Simulated Twin: {scenario.get('name', 'Custom Scenario')}")
    except Exception as e:
        console.print(f"[bold red]Simulation Error:[/bold red] {e}")
    finally:
        db_manager.close_driver()


@simulate_app.command("compare")