cli_app

orchard.cli_app

Orchard ML Command-Line Interface.

Provides the orchard entry point with two commands:

• orchard init — generate a starter recipe YAML with all defaults
• orchard run — execute a training pipeline from a YAML recipe

Usage::

orchard init
orchard run recipes/config_mini_cnn.yaml
orchard run recipes/optuna_mini_cnn.yaml --set training.epochs=30

main(_=None)

Orchard ML: type-Safe Deep Learning for Reproducible Research.

Source code in orchard/cli_app.py

@app.callback()
def main(
    _: Annotated[
        bool | None,
        typer.Option(
            "--version",
            "-V",
            callback=_version_callback,
            is_eager=True,
            help="Show version and exit.",
        ),
    ] = None,
) -> None:
    """Orchard ML: type-Safe Deep Learning for Reproducible Research."""
    ...  # pragma: no cover

init(output=Path('recipe.yaml'), force=False)

Generate a starter recipe with all config fields and defaults.

Source code in orchard/cli_app.py

@app.command()
def init(
    output: Annotated[
        Path,
        typer.Argument(help="Output YAML file path."),
    ] = Path("recipe.yaml"),
    force: Annotated[
        bool,
        typer.Option("--force", "-f", help="Overwrite existing file."),
    ] = False,
) -> None:
    """Generate a starter recipe with all config fields and defaults."""
    if output.exists() and not force:
        typer.echo(f"Error: '{output}' already exists. Use --force to overwrite.", err=True)
        raise typer.Exit(code=1)

    data = _build_init_dict()
    yaml_body = _build_commented_yaml(data)
    content = _INIT_HEADER.format(filename=output.name) + yaml_body

    output.write_text(content, encoding="utf-8")
    typer.echo(f"Recipe created: {output}")
    typer.echo(f"Run it with:   orchard run {output}")

run(recipe, set_=None)

Run the ML pipeline from a YAML recipe.

Source code in orchard/cli_app.py

@app.command()
def run(
    recipe: Annotated[
        Path,
        typer.Argument(help="Path to YAML recipe file."),
    ],
    set_: Annotated[
        list[str] | None,
        typer.Option(
            "--set",
            help="Override config value (repeatable): key.path=value",
        ),
    ] = None,
) -> None:
    """Run the ML pipeline from a YAML recipe."""
    from orchard import (
        MLRUNS_DB,
        Config,
        LogStyle,
        RootOrchestrator,
        create_tracker,
        log_pipeline_summary,
        run_export_phase,
        run_optimization_phase,
        run_training_phase,
    )

    if not recipe.exists():
        typer.echo(f"Error: recipe not found: {recipe}", err=True)
        raise typer.Exit(code=1)

    overrides = _parse_overrides(set_ or [])
    cfg = Config.from_recipe(recipe, overrides=overrides or None)

    with RootOrchestrator(cfg) as orchestrator:
        paths = orchestrator.paths
        run_logger = orchestrator.run_logger
        if paths is None or run_logger is None:
            raise RuntimeError("RootOrchestrator failed to initialize paths or logger")

        tracker = create_tracker(cfg)
        tracking_uri = f"sqlite:///{MLRUNS_DB}"
        tracker.start_run(cfg=cfg, run_name=paths.run_id, tracking_uri=tracking_uri)

        orchestrator.log_environment_report()

        training_cfg = cfg

        try:
            # Phase 1: Optimization (if optuna config present)
            if cfg.optuna is not None:
                _, best_config_path = run_optimization_phase(orchestrator, tracker=tracker)
                if best_config_path and best_config_path.exists():
                    training_cfg = Config.from_recipe(best_config_path)
                    run_logger.info(
                        "%s%s %-18s: %s",
                        LogStyle.INDENT,
                        LogStyle.ARROW,
                        "Optimized Config",
                        best_config_path.name,
                    )
            else:
                run_logger.info(
                    "%s%s Skipping optimization (no optuna config)",
                    LogStyle.INDENT,
                    LogStyle.ARROW,
                )

            # Phase 2: Training
            result = run_training_phase(orchestrator, cfg=training_cfg, tracker=tracker)

            # Phase 3: Export (if export config present)
            onnx_path = None
            if cfg.export is not None:
                onnx_path = run_export_phase(
                    orchestrator,
                    checkpoint_path=result.best_model_path,
                    cfg=training_cfg,
                )

            # Log final artifacts
            tracker.log_artifacts_dir(paths.figures)
            tracker.log_artifact(paths.final_report_path)
            tracker.log_artifact(paths.get_config_path())

            log_pipeline_summary(
                test_acc=result.test_acc,
                macro_f1=result.macro_f1,
                test_auc=result.test_auc,
                best_model_path=result.best_model_path,
                run_dir=paths.root,
                duration=orchestrator.time_tracker.elapsed_formatted,
                onnx_path=onnx_path,
                logger_instance=run_logger,
            )

        except KeyboardInterrupt:
            run_logger.warning("%s Interrupted by user.", LogStyle.WARNING)
            raise SystemExit(1)

        except OrchardError as e:
            run_logger.error("%s %s", LogStyle.FAILURE, e)
            raise SystemExit(1)

        except Exception as e:  # top-level catch-all for logging; re-raises
            run_logger.error("%s Pipeline failed: %s", LogStyle.FAILURE, e, exc_info=True)
            raise

        finally:
            tracker.end_run()