README

Bases: Protocol

              flowchart TD
              liblaf.cherries.core.assets.AssetPluginProtocol[AssetPluginProtocol]

              

              click liblaf.cherries.core.assets.AssetPluginProtocol href "" "liblaf.cherries.core.assets.AssetPluginProtocol"
            

Hook surface for plugins that receive artifact paths.

Methods:

• log_asset –

  Record an existing artifact path.

log_asset(
    path: Path,
    *,
    metadata: Mapping[str, Any] | None = None,
    report: bool = True,
) -> None

 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

def log_asset(
    self,
    path: Path,
    *,
    metadata: Mapping[str, Any] | None = None,
    report: bool = True,
) -> None:
    """Record an existing artifact path.

    Args:
        path: Existing file or directory to record.
        metadata: Optional artifact metadata, usually including `type`.
        report: Whether the path is the primary user-facing artifact.
            Companion files are logged with `report=False`.
    """
    ...

Stage independent inputs and declared outputs inside an active run.

Parameters:

• working_dir (Path) –
• plugins (AssetPluginProtocol) –
• bundles (BundleRegistry, default: BundleRegistry(registry=[BundleLandmarks(suffixes={'.obj', '.vtp', '.ply', '.stl', '.vtr', '.vtkhdf', '.vts', '.vtu', '.vti'}), BundleSeries()]) ) –
• summary (AssetsSummary, default: <dynamic> ) –

  Paths successfully reported during a run.

  The summary contains only primary artifacts. Bundle companions are sent to plugins, but omitted from this user-facing record.

• pending (list[PendingAsset], default: <dynamic> ) –

  Built-in mutable sequence.

  If no argument is given, the constructor creates a new empty list. The argument must be an iterable if specified.

• active (bool, default: False ) –
• store (Store | None, default: None ) –
• run_id (str | None, default: None ) –
• bindings (list[dict[str, Any]], default: <dynamic> ) –

  Built-in mutable sequence.

  If no argument is given, the constructor creates a new empty list. The argument must be an iterable if specified.

• retained_bundles (list[dict[str, Any]], default: <dynamic> ) –

  Built-in mutable sequence.

  If no argument is given, the constructor creates a new empty list. The argument must be an iterable if specified.

Methods:

• end –
• input –
• log_asset –
• log_input –
• log_output –
• log_temp –
• output –
• temp –

Attributes:

• active (bool) –
• bindings (list[dict[str, Any]]) –
• bundles (BundleRegistry) –
• data_dir (Path) –
• pending (list[PendingAsset]) –
• plugins (AssetPluginProtocol) –
• retained_bundles (list[dict[str, Any]]) –
• run_id (str | None) –
• store (Store | None) –
• summary (AssetsSummary) –
• temp_dir (Path) –
• working_dir (Path) –

active: bool = False

bindings: list[dict[str, Any]] = attrs.field(factory=list)

bundles: BundleRegistry = attrs.field(default=bundles)

data_dir: Path

pending: list[PendingAsset] = attrs.field(factory=list)

plugins: AssetPluginProtocol

retained_bundles: list[dict[str, Any]] = attrs.field(
    factory=list
)

run_id: str | None = None

store: Store | None = None

summary: AssetsSummary = attrs.field(factory=AssetsSummary)

temp_dir: Path

working_dir: Path

end() -> None

482
483
484
485
486
487
488
489
490
491

def end(self) -> None:
    self._check()
    self._validate_inputs()
    for pending in self.pending:
        if not pending.path.exists():
            msg = f"Required output was not written: {pending.path}"
            raise FileNotFoundError(msg)
        self.log_output(pending.path, metadata=pending.metadata)
    self.pending.clear()
    self._retain_bundles()

input(
    path: StrPath,
    *,
    name: StrPath | None = None,
    source_run: str | None = None,
    metadata: Mapping[str, Any] | None = None,
) -> Path

135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177

def input(
    self,
    path: StrPath,
    *,
    name: StrPath | None = None,
    source_run: str | None = None,
    metadata: Mapping[str, Any] | None = None,
) -> Path:
    self._check()
    original = os.fspath(path)
    declaration = original
    replay = os.environ.get("CHERRIES_REPLAY_INPUTS")
    if replay:
        declaration = json.loads(Path(replay).read_text()).get(
            original, declaration
        )
    binding: dict[str, Any]
    if declaration.startswith(("sha256:", "sha256-tree:", "run:")):
        target, binding = self._reference_input(declaration, name, source_run)
    else:
        source = Path(declaration).expanduser().resolve(strict=True)
        target = self._target("inputs", name or source.name)
        _copy_verified(source, target)
        binding = {
            "asset_id": "sha256:" + hash_file(target) if target.is_file() else None
        }
        if source.is_file():
            self._companions(source, target)
    binding.update(
        {
            "source": original,
            "resolved_source": declaration,
            "staged_path": target.relative_to(self.working_dir).as_posix(),
        }
    )
    binding["input_snapshot"] = self._input_snapshot(target)
    binding["directory"] = target.is_dir()
    self.bindings.append(binding)
    self.summary.inputs.append(target)
    self.plugins.log_asset(
        target, metadata=_metadata_with_type(metadata, "input"), report=True
    )
    return target

log_asset(
    path: StrPath,
    *,
    metadata: Mapping[str, Any] | None = None,
    name: StrPath | None = None,
) -> Path

446
447
448
449
450
451
452
453

def log_asset(
    self,
    path: StrPath,
    *,
    metadata: Mapping[str, Any] | None = None,
    name: StrPath | None = None,
) -> Path:
    return self._log(path, "artifacts", "asset", metadata, name)

log_input(
    path: StrPath,
    *,
    metadata: Mapping[str, Any] | None = None,
    name: StrPath | None = None,
) -> Path

455
456
457
458
459
460
461
462

def log_input(
    self,
    path: StrPath,
    *,
    metadata: Mapping[str, Any] | None = None,
    name: StrPath | None = None,
) -> Path:
    return self.input(path, name=name, metadata=metadata)

log_output(
    path: StrPath,
    *,
    metadata: Mapping[str, Any] | None = None,
    name: StrPath | None = None,
) -> Path

464
465
466
467
468
469
470
471

def log_output(
    self,
    path: StrPath,
    *,
    metadata: Mapping[str, Any] | None = None,
    name: StrPath | None = None,
) -> Path:
    return self._log(path, "outputs", "output", metadata, name)

log_temp(
    path: StrPath,
    *,
    metadata: Mapping[str, Any] | None = None,
    name: StrPath | None = None,
) -> Path

473
474
475
476
477
478
479
480

def log_temp(
    self,
    path: StrPath,
    *,
    metadata: Mapping[str, Any] | None = None,
    name: StrPath | None = None,
) -> Path:
    return self._log(path, "artifacts", "temp", metadata, name)

output(
    path: StrPath,
    *,
    metadata: Mapping[str, Any] | None = None,
    mkdir: bool = True,
) -> Path

364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380

def output(
    self,
    path: StrPath,
    *,
    metadata: Mapping[str, Any] | None = None,
    mkdir: bool = True,
) -> Path:
    target = self._target("outputs", path)
    if mkdir:
        target.parent.mkdir(parents=True, exist_ok=True)
    if any(item.path == target for item in self.pending):
        msg = f"output already declared: {path}"
        raise ValueError(msg)
    self.pending.append(
        PendingAsset(target, _metadata_with_type(metadata, "output"))
    )
    return target

temp(
    path: StrPath,
    *,
    metadata: Mapping[str, Any] | None = None,
    mkdir: bool = True,
) -> Path

382
383
384
385
386
387
388
389
390
391
392
393
394

def temp(
    self,
    path: StrPath,
    *,
    metadata: Mapping[str, Any] | None = None,
    mkdir: bool = True,
) -> Path:
    # Scratch metadata is accepted for old scripts but is not retained.
    del metadata
    target = self._target("scratch", path)
    if mkdir:
        target.parent.mkdir(parents=True, exist_ok=True)
    return target

Bases: ABC

              flowchart TD
              liblaf.cherries.core.assets.Bundle[Bundle]

              

              click liblaf.cherries.core.assets.Bundle href "" "liblaf.cherries.core.assets.Bundle"
            

Base class for artifact companion-file discovery.

Methods:

• ls_files –

  Yield files that should be logged with path.

• match –

  Return whether this bundle can expand path.

ls_files(path: Path) -> Iterable[BundleItem]

40
41
42
43
44
45
46
47
48
49
50

@abc.abstractmethod
def ls_files(self, path: Path) -> Iterable[BundleItem]:
    """Yield files that should be logged with `path`.

    Args:
        path: Primary artifact path.

    Returns:
        Iterable of companion files and their optional/missing-file policy.
    """
    raise NotImplementedError

match(path: Path) -> bool

27
28
29
30
31
32
33
34
35
36
37
38

@abc.abstractmethod
def match(self, path: Path) -> bool:
    """Return whether this bundle can expand `path`.

    Args:
        path: Primary artifact path.

    Returns:
        `True` when [`ls_files`][liblaf.cherries.core.assets.bundle.Bundle.ls_files]
        can yield companion files for `path`.
    """
    raise NotImplementedError

Bases: NamedTuple

              flowchart TD
              liblaf.cherries.core.assets.BundleItem[BundleItem]

              

              click liblaf.cherries.core.assets.BundleItem href "" "liblaf.cherries.core.assets.BundleItem"
            

Related file discovered for a logged artifact.

Attributes:

Parameters:

• path (ForwardRef, default: None ) –
• optional (ForwardRef, default: None ) –

optional: bool

path: StrPath

Registry that expands logged artifacts through matching bundles.

Examples:

>>> registry = BundleRegistry(registry=[])
>>> list(registry.ls_files(Path("mesh.vtu")))
[]

Parameters:

• registry (list[Bundle], default: [BundleLandmarks(suffixes={'.obj', '.vtp', '.ply', '.stl', '.vtr', '.vtkhdf', '.vts', '.vtu', '.vti'}), BundleSeries()] ) –

Methods:

• ls_files –

  Yield companion files from every bundle that matches path.

• register –

  Append bundle to the registry.

Attributes:

• registry (list[Bundle]) –

registry: list[Bundle] = attrs.field(
    factory=_default_registry
)

ls_files(path: Path) -> Generator[BundleItem]

28
29
30
31
32

def ls_files(self, path: Path) -> Generator[BundleItem]:
    """Yield companion files from every bundle that matches `path`."""
    for bundle in self.registry:
        if bundle.match(path):
            yield from bundle.ls_files(path)

register(bundle: Bundle) -> None

34
35
36

def register(self, bundle: Bundle) -> None:
    """Append `bundle` to the registry."""
    self.registry.append(bundle)