Skip to content

artifacts

artifacts

CLI entry points for artifact management.

Provides CLI commands for inspecting and cleaning up artifact directories produced by Safe Synthesizer runs.

Functions:

Name Description
artifacts

Artifact management commands.
clean

Clean artifacts in a Workdir structure.

artifacts(ctx)

Artifact management commands.

Source code in src/nemo_safe_synthesizer/cli/artifacts.py 
20
21
22
23
24
@click.group(invoke_without_command=True)
@click.pass_context
def artifacts(ctx: click.Context):
    """Artifact management commands."""
    pass

clean(ctx, artifact_path, dry_run, caches_only, force)

Clean artifacts in a Workdir structure.

Source code in src/nemo_safe_synthesizer/cli/artifacts.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@artifacts.command()
@click.option(
    "--artifact-path",
    type=click.Path(exists=True, dir_okay=True, file_okay=False, resolve_path=True),
    help="Path to the artifact directory.",
)
@click.option("--dry-run", is_flag=True, help="Dry run the command.")
@click.option("--caches-only", is_flag=True, help="Only clean caches.")
@click.option("--force", is_flag=True, help="Force clean.")
def clean(ctx: click.Context, artifact_path: PathT | None, dry_run: bool, caches_only: bool, force: bool):
    """Clean artifacts in a Workdir structure."""
    if artifact_path is None:
        artifact_path = Path("safe-synthesizer-artifacts")

    try:
        workdir = Workdir.from_path(Path(artifact_path))
    except ValueError as e:
        click.secho(f"Error: {e}", fg="red", err=True)
        return

    # Determine what to clean
    if caches_only:
        cache_dir = workdir.train.cache
        if not isinstance(cache_dir, BoundDir):
            raise TypeError(f"Expected BoundDir, got {type(cache_dir)}")
        target = cache_dir.path
        item_name = "cache"
    else:
        target = workdir.run_dir
        item_name = "all artifacts (run directory)"

    if not target.exists():
        click.echo(f"Path does not exist, nothing to clean: {target}")
        return

    # Confirmation
    if not force and not dry_run:
        if not click.confirm(f"Are you sure you want to delete {item_name} at {target}?"):
            click.echo("Aborted.")
            return

    if dry_run:
        click.secho(f"[DRY RUN] Would delete: {target}", fg="yellow")
    else:
        click.echo(f"Cleaning {item_name}...")
        try:
            if target.is_dir():
                shutil.rmtree(target)
            else:
                target.unlink()
            click.secho(f"Successfully deleted: {target}", fg="green")
        except Exception as e:
            click.secho(f"Error deleting {target}: {e}", fg="red", err=True)