run

run

CLI run commands for Safe Synthesizer.

Functions:

Name	Description
common_run_options	Decorator to add common options for run commands.
run	Run the Safe Synthesizer end-to-end pipeline.
run_train	Run the training stage only.
run_generate	Run the generation stage only.

common_run_options(f)

Decorator to add common options for run commands.

Apply this above @pydantic_options in source order. Python applies decorators bottom-up, so the shared command options are added after the generated parameter options. Environment-variable handling stays in CLISettings rather than Click envvar= declarations, so precedence is centralized in one settings model.

Source code in src/nemo_safe_synthesizer/cli/run.py

def common_run_options(f: Callable[..., object]) -> Callable[..., object]:
    """Decorator to add common options for run commands.

    Apply this above ``@pydantic_options`` in source order. Python applies
    decorators bottom-up, so the shared command options are added after the
    generated parameter options. Environment-variable handling stays in
    ``CLISettings`` rather than Click ``envvar=`` declarations, so precedence is
    centralized in one settings model.
    """
    options = []
    options.append(
        click.option("--config", "config_path", default=None, required=False, help="path to a yaml config file")
    )
    options.append(
        click.option(
            "--data-source",
            type=str,
            default=None,
            required=False,
            help="Dataset name, URL, or path to CSV dataset. "
            "For 'run generate', this is optional if a cached dataset exists in the workdir.",
        )
    )
    options.append(
        click.option(
            "--artifact-path",
            type=click.Path(exists=False, dir_okay=True, file_okay=False, resolve_path=True),
            default=None,
            required=False,
            help="Base directory for all runs. Runs are created as "
            "<artifact-path>/<config>---<dataset>/<timestamp>/. "
            "Can also be set via NSS_ARTIFACTS_PATH env var. "
            "[default: ./safe-synthesizer-artifacts]",
        )
    )
    options.append(
        click.option(
            "--run-path",
            type=click.Path(exists=False, dir_okay=True, file_okay=False, resolve_path=True),
            default=None,
            required=False,
            help="Explicit path for this run's output directory. "
            "When specified, outputs go directly to this path. "
            "Overrides --artifact-path.",
        )
    )
    options.append(
        click.option(
            "--output-file",
            type=click.Path(exists=False),
            default=None,
            required=False,
            help="Path to output CSV file. Overrides the default workdir output location.",
        )
    )
    options.append(
        click.option(
            "--log-format",
            type=click.Choice(["json", "plain"]),
            default=None,
            required=False,
            help="Log format for console output. File logging will always be JSON. "
            "Can also be set via NSS_LOG_FORMAT env var. [default: plain]",
        )
    )
    options.append(
        click.option(
            "--log-color/--no-log-color",
            type=click.BOOL,
            default=None,
            required=False,
            help="Whether to colorize the log output on the console. [default: --log-color]",
        )
    )
    options.append(
        click.option(
            "--log-file",
            type=click.Path(exists=False),
            default=None,
            required=False,
            help="Path to log file. Defaults to a file nested under the run directory. "
            "Can also be set via NSS_LOG_FILE env var.",
        )
    )
    options.append(
        click.option(
            "--wandb-mode",
            type=click.Choice(["online", "offline", "disabled"]),
            default=None,
            required=False,
            help="Wandb mode. 'online' will upload logs to wandb, 'offline' will save logs to a local file, 'disabled' will not upload logs to wandb. Can also be set via WANDB_MODE env var. [default: disabled]",
        )
    )
    options.append(
        click.option(
            "--wandb-project",
            type=str,
            default=None,
            required=False,
            help="Wandb project. Can also be set via WANDB_PROJECT env var.",
        )
    )
    options.append(
        click.option(
            "-v",
            "verbose",
            required=False,
            help="Verbose logging. 'v' shows debug info from main program, 'vv' shows debug from dependencies too",
            count=True,
        )
    )
    options.append(
        click.option(
            "--dataset-registry",
            type=str,
            required=False,
            default=None,
            help="URL or path of a dataset registry YAML file. If provided, "
            "datasets in the registry may be referenced by name in --data-source. "
            "Can also be set via NSS_DATASET_REGISTRY env var. "
            "If both env var and CLI option are provided, the CLI option takes precedence.",
        )
    )
    # Apply each option decorator in reverse order (decorators apply bottom-up)
    for option in reversed(options):
        f = option(f)
    return f

run(ctx, config_path, data_source, artifact_path, run_path, output_file, log_file, log_color, log_format, verbose=0, wandb_mode=None, wandb_project=None, dataset_registry=None, validate=False, **kwargs)

Run the Safe Synthesizer end-to-end pipeline.

Without a subcommand, runs the full end-to-end pipeline. Use 'run train' or 'run generate' for individual stages.

Source code in src/nemo_safe_synthesizer/cli/run.py

@click.group(invoke_without_command=True)
@click.pass_context
@common_run_options
@pydantic_options(SafeSynthesizerParameters, field_separator=CLI_NESTED_FIELD_SEPARATOR)
@click.option(
    "--validate",
    is_flag=True,
    default=False,
    help="Run pre-flight validation only, then exit without training or generating.",
)
def run(
    ctx: click.Context,
    config_path: PathT | None,
    data_source: str | None,
    artifact_path: PathT | None,
    run_path: PathT | None,
    output_file: PathT | None,
    log_file: PathT | None,
    log_color: bool | None,
    log_format: str | None,
    verbose: int = 0,
    wandb_mode: str | None = None,
    wandb_project: str | None = None,
    dataset_registry: str | None = None,
    validate: bool = False,
    **kwargs: object,
) -> None:
    """Run the Safe Synthesizer end-to-end pipeline.

    Without a subcommand, runs the full end-to-end pipeline.
    Use 'run train' or 'run generate' for individual stages.
    """
    # If a subcommand is invoked, skip the default behavior
    if ctx.invoked_subcommand is not None:
        return

    _set_cli_deployment_type_default()

    settings = CLISettings.from_cli_kwargs(
        data_source=data_source,
        config_path=config_path,
        artifact_path=artifact_path,
        run_path=run_path,
        output_file=output_file,
        log_file=log_file,
        log_color=log_color,
        log_format=log_format,
        verbose=verbose,
        wandb_mode=wandb_mode,
        wandb_project=wandb_project,
        synthesis_overrides=_parse_run_overrides(kwargs),
        dataset_registry=dataset_registry,
    )

    if validate:
        os.environ["NSS_PHASE"] = "process_data"
    else:
        os.environ["NSS_PHASE"] = "end_to_end"

    run_logger, config, df, workdir = common_setup(
        settings=settings,
        phase="process_data" if validate else "end_to_end",
        skip_wandb=validate,
        quiet=validate,
        run_name="validate" if validate else None,
    )

    try:
        run_logger.warning("Nemo Safe Synthesizer starting")
        run_logger.debug("running with: ", extra={"config": config.model_dump()})

        with traced_user("SafeSynthesizer"):
            from ..sdk.library_builder import SafeSynthesizer

            assert df is not None
            nss: SafeSynthesizer = SafeSynthesizer(
                config=config,
                workdir=workdir,
                emit_telemetry=config.emit_telemetry,
            ).with_data_source(df)

            if validate:
                _run_validate_and_render(nss, settings=settings, workdir=workdir, config=config, data=df)
                return

            try:
                nss.run(output_file=settings.output_file)
                nss.results.summary.log_summary(run_logger)
                nss.results.summary.timing.log_timing(run_logger)
                nss.results.summary.log_wandb()
            finally:
                if hasattr(nss, "generator") and nss.generator is not None:
                    nss.generator.teardown()
    except UserError as exc:
        click.secho(str(exc), fg="red", err=True)
        raise SystemExit(1)

run_train(config_path, data_source, artifact_path, run_path, output_file, log_format, log_color, log_file, verbose, wandb_mode=None, wandb_project=None, dataset_registry=None, validate=False, **kwargs)

Run the training stage only.

This command processes data and trains the model, saving the adapter to the run directory. Use 'run generate' afterwards to generate synthetic data from the trained adapter.

Source code in src/nemo_safe_synthesizer/cli/run.py

@run.command("train")
@common_run_options
@pydantic_options(SafeSynthesizerParameters, field_separator=CLI_NESTED_FIELD_SEPARATOR)
@click.option(
    "--validate",
    is_flag=True,
    default=False,
    help="Run pre-flight validation only, then exit without training or generating.",
)
def run_train(
    config_path: PathT,
    data_source: str | None,
    artifact_path: PathT | None,
    run_path: PathT | None,
    output_file: PathT | None,
    log_format: str | None,
    log_color: bool | None,
    log_file: PathT | None,
    verbose: int,
    wandb_mode: str | None = None,
    wandb_project: str | None = None,
    dataset_registry: str | None = None,
    validate: bool = False,
    **kwargs: object,
) -> None:
    """Run the training stage only.

    This command processes data and trains the model, saving the adapter to the run directory.
    Use 'run generate' afterwards to generate synthetic data from the trained adapter.
    """
    _set_cli_deployment_type_default()

    settings = CLISettings.from_cli_kwargs(
        data_source=data_source,
        config_path=config_path,
        artifact_path=artifact_path,
        run_path=run_path,
        output_file=output_file,
        log_file=log_file,
        log_color=log_color,
        log_format=log_format,
        verbose=verbose,
        wandb_mode=wandb_mode,
        wandb_project=wandb_project,
        synthesis_overrides=_parse_run_overrides(kwargs),
        dataset_registry=dataset_registry,
    )

    if validate:
        os.environ["NSS_PHASE"] = "process_data"
    else:
        os.environ["NSS_PHASE"] = "train"

    run_logger, config, df, workdir = common_setup(
        settings=settings,
        phase="process_data" if validate else "train",
        skip_wandb=validate,
        quiet=validate,
        run_name="validate" if validate else None,
    )
    from ..sdk.library_builder import SafeSynthesizer

    try:
        with traced_user("SafeSynthesizer"):
            assert df is not None
            nss = SafeSynthesizer(
                config,
                workdir=workdir,
                emit_telemetry=config.emit_telemetry,
            ).with_data_source(df)

            if validate:
                _run_validate_and_render(nss, settings=settings, workdir=workdir, config=config, data=df)
                return

            nss.process_data().train()
            run_logger.info(f"Training complete. Adapter saved to: {workdir.adapter_path}")
    except UserError as exc:
        click.secho(str(exc), fg="red", err=True)
        raise SystemExit(1)

run_generate(config_path, data_source, run_path, artifact_path, output_file, log_format, log_color, log_file, verbose, wandb_mode=None, wandb_project=None, auto_discover_adapter=False, wandb_resume_job_id=None, dataset_registry=None, **kwargs)

Run the generation stage only.

This command loads a trained adapter and generates synthetic data. Requires 'run train' to have been executed first.

Use --run-path to specify the exact run directory containing the trained model, or use --auto-discover-adapter with --artifact-path to automatically find the latest trained run.

Source code in src/nemo_safe_synthesizer/cli/run.py

@run.command("generate")
@common_run_options
@click.option(
    "--auto-discover-adapter",
    is_flag=True,
    default=False,
    help="Automatically find the latest trained adapter in --artifacts-path. "
    "Without this flag, --run-path must point to a specific trained run.",
)
@click.option(
    "--wandb-resume-job-id",
    type=str,
    default=None,
    required=False,
    help="Wandb run ID to resume, or path to a file containing the run ID. "
    "Overrides file-based run ID detection from workdir.",
)
@pydantic_options(SafeSynthesizerParameters, field_separator=CLI_NESTED_FIELD_SEPARATOR)
def run_generate(
    config_path: PathT,
    data_source: str | None,
    run_path: PathT | None,
    artifact_path: PathT | None,
    output_file: PathT | None,
    log_format: str | None,
    log_color: bool | None,
    log_file: PathT | None,
    verbose: int,
    wandb_mode: str | None = None,
    wandb_project: str | None = None,
    auto_discover_adapter: bool = False,
    wandb_resume_job_id: str | None = None,
    dataset_registry: str | None = None,
    **kwargs: object,
) -> None:
    """Run the generation stage only.

    This command loads a trained adapter and generates synthetic data.
    Requires 'run train' to have been executed first.

    Use --run-path to specify the exact run directory containing the trained model,
    or use --auto-discover-adapter with --artifact-path to automatically find
    the latest trained run.
    """
    _set_cli_deployment_type_default()

    # Create unified settings from CLI kwargs
    settings = CLISettings.from_cli_kwargs(
        data_source=data_source,
        config_path=config_path,
        artifact_path=artifact_path,
        run_path=run_path,
        output_file=output_file,
        log_file=log_file,
        log_color=log_color,
        log_format=log_format,
        verbose=verbose,
        wandb_mode=wandb_mode,
        wandb_project=wandb_project,
        synthesis_overrides=_parse_run_overrides(kwargs),
        dataset_registry=dataset_registry,
    )

    os.environ["NSS_PHASE"] = "generate"
    # Generation always resumes from an existing workdir with a trained model
    run_logger, config, df, workdir = common_setup(
        settings=settings,
        resume=True,
        phase="generate",
        auto_discover_adapter=auto_discover_adapter,
        wandb_resume_job_id=wandb_resume_job_id,
    )
    from ..sdk.library_builder import SafeSynthesizer, _emit_nss_telemetry

    final_output_file = settings.output_file or workdir.output_file
    with traced_user("SafeSynthesizer"):
        nss = SafeSynthesizer(
            config,
            workdir=workdir,
            emit_telemetry=config.emit_telemetry,
        )

        # Only set data source if provided via --data-source
        # Otherwise, load_from_save_path() will load from cached files
        if df is not None:
            nss = nss.with_data_source(df)

        try:
            nss = (
                nss.load_from_save_path()
                .process_data()
                .generate()
                .evaluate()
                .save_results(output_file=final_output_file)
            )
            _emit_nss_telemetry(nss, TaskStatusEnum.COMPLETED)
            nss.results.summary.log_summary(run_logger)
            nss.results.summary.timing.log_timing(run_logger)
            run_logger.info(f"Generation complete. Results saved to: {final_output_file}")
            nss.results.summary.log_wandb()
        except KeyboardInterrupt:
            _emit_nss_telemetry(nss, TaskStatusEnum.CANCELED)
            raise
        except Exception:
            _emit_nss_telemetry(nss, TaskStatusEnum.ERROR)
            raise
        finally:
            if hasattr(nss, "generator") and nss.generator is not None:
                nss.generator.teardown()