README

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`.
    """
    ...

def end(self) -> None:
    """Flush queued output and temporary paths.

    Missing queued paths produce warnings and are left out of the summary.
    """
    for asset in self.pending:
        self.log_asset(asset.path, metadata=asset.metadata)

def input(
    self, name: StrPath, *, metadata: Mapping[str, Any] | None = None
) -> Path:
    """Resolve and immediately log an input path.

    Args:
        name: Path below `data/`.
        metadata: Extra metadata merged with `{"type": "input"}`.

    Returns:
        Resolved input path.
    """
    path: Path = self.data_dir / name
    metadata: dict[str, Any] = _metadata_with_type(metadata, "input")
    self.log_input(path, metadata=metadata)
    return path

def log_asset(
    self, path: StrPath, *, metadata: Mapping[str, Any] | None = None
) -> None:
    """Log an existing path and any companion files from matching bundles.

    Missing primary paths are reported as warnings. Missing optional bundle
    files are ignored, while missing required bundle files are warned.
    """
    path: Path = Path(path)
    if not path.exists():
        logger.warning("No such file or directory: %s", path)
        return
    if metadata is None:
        self.summary.assets.append(path)
    else:
        match metadata.get("type"):
            case "input":
                self.summary.inputs.append(path)
            case "output":
                self.summary.outputs.append(path)
            case "temp":
                self.summary.temps.append(path)
            case _:
                self.summary.assets.append(path)
    self.plugins.log_asset(path, metadata=metadata, report=True)
    for path_, optional in self.bundles.ls_files(path):
        path: Path = Path(path_)
        if not path.exists():
            if not optional:
                logger.warning("No such file or directory: %s", path)
            continue
        self.plugins.log_asset(path, metadata=metadata, report=False)

def log_input(
    self, path: StrPath, *, metadata: Mapping[str, Any] | None = None
) -> None:
    """Log `path` as an input artifact."""
    metadata: dict[str, Any] = _metadata_with_type(metadata, "input")
    self.log_asset(path, metadata=metadata)

def log_output(
    self, path: StrPath, *, metadata: Mapping[str, Any] | None = None
) -> None:
    """Log `path` as an output artifact."""
    metadata: dict[str, Any] = _metadata_with_type(metadata, "output")
    self.log_asset(path, metadata=metadata)

def log_temp(
    self, path: StrPath, *, metadata: Mapping[str, Any] | None = None
) -> None:
    """Log `path` as a temporary artifact."""
    metadata: dict[str, Any] = _metadata_with_type(metadata, "temp")
    self.log_asset(path, metadata=metadata)

def output(
    self,
    name: StrPath,
    *,
    metadata: Mapping[str, Any] | None = None,
    mkdir: bool = True,
) -> Path:
    """Resolve an output path and queue it for end-of-run logging.

    Args:
        name: Path below `data/`.
        metadata: Extra metadata merged with `{"type": "output"}`.
        mkdir: Whether to create the parent directory before returning.

    Returns:
        Resolved output path.
    """
    path: Path = self.data_dir / name
    if mkdir:
        path.parent.mkdir(parents=True, exist_ok=True)
    metadata: dict[str, Any] = _metadata_with_type(metadata, "output")
    self.pending.append(PendingAsset(path=path, metadata=metadata))
    return path

def temp(
    self,
    name: StrPath,
    *,
    metadata: Mapping[str, Any] | None = None,
    mkdir: bool = True,
) -> Path:
    """Resolve a temporary path and queue it for end-of-run logging.

    Args:
        name: Path below `tmp/`.
        metadata: Extra metadata merged with `{"type": "temp"}`.
        mkdir: Whether to create the parent directory before returning.

    Returns:
        Resolved temporary path.
    """
    path: Path = self.temp_dir / name
    if mkdir:
        path.parent.mkdir(parents=True, exist_ok=True)
    metadata: dict[str, Any] = _metadata_with_type(metadata, "temp")
    self.pending.append(PendingAsset(path=path, metadata=metadata))
    return path

@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

@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

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)

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