README

def end(self, exc: BaseException | None = None) -> None:
    """Flush artifacts, record shutdown metadata, and end plugins.

    Args:
        exc: Exception raised by the experiment, if any.
    """
    self.log_other("cherries/end_time", datetime.now().astimezone())
    if exc is not None:
        self.log_other(
            "cherries/exception", "\n".join(traceback.format_exception_only(exc))
        )
    self._assets.end()
    self.plugins.delegate("end", exc=exc)

def get_metric(self, name: str) -> pl.DataFrame:
    """Return one metric series."""
    return self._metrics.get_metric(name)

def get_metrics(self, metrics: Iterator[str] | None = None) -> pl.DataFrame:
    """Return selected metric series concatenated into one dataframe."""
    return self._metrics.get_metrics(metrics)

def get_other(self, name: str) -> Any:
    """Return one flattened metadata value."""
    return self._others.get_other(name)

def get_others(self) -> dict[str, Any]:
    """Return logged metadata as a nested dictionary."""
    return self._others.get_others()

def get_param(self, name: str) -> Any:
    """Return one flattened parameter value."""
    return self._params.get_param(name)

def get_params(self) -> dict[str, Any]:
    """Return logged parameters as a nested dictionary."""
    return self._params.get_params()

def get_step(self) -> int:
    """Return the default metric step."""
    return self.step

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

def log_asset(
    self, path: StrPath, metadata: Mapping[str, Any] | None = None
) -> None:
    """Log an existing generic artifact immediately."""
    self._assets.log_asset(path, metadata=metadata)

def log_input(
    self, path: StrPath, metadata: Mapping[str, Any] | None = None
) -> None:
    """Log an existing input artifact immediately."""
    self._assets.log_input(path, metadata=metadata)

def log_metric(
    self,
    name: str,
    value: SupportsFloat,
    *,
    step: int | None = None,
    time: datetime | None = None,
) -> None:
    """Log one scalar metric."""
    self._metrics.log_metric(name, value, step=step, time=time)

def log_metrics(
    self,
    metrics: MetricsLike,
    *,
    step: int | None = None,
    time: datetime | None = None,
) -> None:
    """Log multiple scalar metrics, flattening nested mappings with `/`."""
    self._metrics.log_metrics(metrics, step=step, time=time)

def log_other(self, name: str, value: Any) -> None:
    """Log one metadata value."""
    self._others.log_other(name, value)

def log_others(self, others: Mapping[str, Any]) -> None:
    """Log multiple metadata values."""
    self._others.log_others(others)

def log_output(
    self, path: StrPath, metadata: Mapping[str, Any] | None = None
) -> None:
    """Log an existing output artifact immediately."""
    self._assets.log_output(path, metadata=metadata)

def log_param(self, name: str, value: Any) -> None:
    """Log one parameter value."""
    self._params.log_param(name, value)

def log_params(self, params: Mapping[str, Any]) -> None:
    """Log multiple parameter values."""
    self._params.log_params(params)

def log_temp(
    self, path: StrPath, metadata: Mapping[str, Any] | None = None
) -> None:
    """Log an existing temporary artifact immediately."""
    self._assets.log_temp(path, metadata=metadata)

def output(
    self,
    path: StrPath,
    *,
    metadata: Mapping[str, Any] | None = None,
    mkdir: bool = True,
) -> Path:
    """Resolve an output below `data/` and queue it until run end."""
    return self._assets.output(path, metadata=metadata, mkdir=mkdir)

def set_step(self, step: int) -> None:
    """Set the default metric step."""
    self.step = step

def start(self) -> None:
    """Start plugins and record Cherries run metadata."""
    self.plugins.delegate("start")
    entrypoint: Path = relative_or_absolute(self.entrypoint, self.project_dir)
    exp_dir: Path = relative_or_absolute(self.working_dir, self.project_dir)
    self.log_other("cherries/cmd", shlex.join(sys.orig_argv))
    self.log_other("cherries/entrypoint", entrypoint)
    self.log_other("cherries/exp_dir", exp_dir)
    self.log_other("cherries/start_time", self.start_time)

def summary(self, prefix: StrPath | None = None) -> dict[str, Any]:
    """Build a JSON/YAML-friendly run summary.

    Args:
        prefix: Optional directory to strip from artifact paths.

    Returns:
        Run metadata, parameters, artifact paths, and user metadata.
    """
    summary: dict[str, Any] = {"name": self.run_name}
    if self.tags:
        summary["tags"] = self.tags
    others: dict[str, Any] = self.get_others()
    summary.update(others.pop("cherries"))
    summary["params"] = self.get_params()
    summary.update(self._assets.summary.to_dict(prefix=prefix))
    summary["others"] = others
    return summary

def temp(
    self,
    path: StrPath,
    *,
    metadata: Mapping[str, Any] | None = None,
    mkdir: bool = True,
) -> Path:
    """Resolve a temporary artifact below `tmp/` and queue it until run end."""
    return self._assets.temp(path, metadata=metadata, mkdir=mkdir)