Skip to content

liblaf.cherries ¤

Modules:

Classes:

Functions:

Attributes:

active_run module-attribute ¤

active_run: Run = Run()

log_asset module-attribute ¤

log_asset = log_asset

log_input module-attribute ¤

log_input = log_input

log_metric module-attribute ¤

log_metric = log_metric

log_metrics module-attribute ¤

log_metrics = log_metrics

log_other module-attribute ¤

log_other = log_other

log_others module-attribute ¤

log_others = log_others

log_output module-attribute ¤

log_output = log_output

log_parameter module-attribute ¤

log_parameter = log_parameter

log_parameters module-attribute ¤

log_parameters = log_parameters

BaseConfig ¤

Bases: BaseSettings

Attributes:

model_config class-attribute instance-attribute ¤

model_config = SettingsConfigDict(cli_parse_args=True)

Plugin ¤

Parameters:

  • _plugin_parent (Self | None, default: None ) –

Methods:

Attributes:

plugin_id property ¤

plugin_id: str

plugin_root property ¤

plugin_root: Self

plugins class-attribute instance-attribute ¤

plugins: dict[str, Plugin] = field(
    factory=dict, kw_only=True
)

delegate ¤

delegate(
    method: MethodName,
    args: Sequence[Any] = (),
    kwargs: Mapping[str, Any] = {},
    *,
    first_result: bool = False,
) -> Any
Source code in src/liblaf/cherries/core/_plugin.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def delegate(
    self,
    method: MethodName,
    args: Sequence[Any] = (),
    kwargs: Mapping[str, Any] = {},
    *,
    first_result: bool = False,
) -> Any:
    plugins: Sequence[Plugin] = self._plugins_sort(method)
    if not plugins:
        if first_result:
            return None
        return []
    results: list[Any] = []
    for plugin in plugins:
        result: Any = getattr(plugin, method)(*args, **kwargs)
        if result is None:
            continue
        if first_result:
            return result
        results.append(result)
    return results

plugin_id_cls classmethod ¤

plugin_id_cls() -> str
Source code in src/liblaf/cherries/core/_plugin.py 
21
22
23
@classmethod
def plugin_id_cls(cls) -> str:
    return cls.__name__

register ¤

register(plugin: Plugin) -> None
Source code in src/liblaf/cherries/core/_plugin.py 
58
59
60
def register(self, plugin: "Plugin") -> None:
    plugin._plugin_parent = self  # noqa: SLF001
    self.plugins[plugin.plugin_id] = plugin

Run ¤

Bases: Plugin

.

Parameters:

  • _plugin_parent (Self | None, default: None ) –
References
  1. Experiment - Comet Docs
  2. Logger | ClearML
  3. MLflow Tracking APIs | MLflow

Methods:

Attributes:

data_dir cached property ¤

data_dir: Path

entrypoint cached property ¤

entrypoint: Path

exp_dir cached property ¤

exp_dir: Path

name cached property ¤

name: str

plugin_id property ¤

plugin_id: str

plugin_root property ¤

plugin_root: Self

plugins class-attribute instance-attribute ¤

plugins: dict[str, Plugin] = field(
    factory=dict, kw_only=True
)

project_name cached property ¤

project_name: str | None

root_dir cached property ¤

root_dir: Path

start_time cached property ¤

start_time: datetime

url cached property ¤

url: str

delegate ¤

delegate(
    method: MethodName,
    args: Sequence[Any] = (),
    kwargs: Mapping[str, Any] = {},
    *,
    first_result: bool = False,
) -> Any
Source code in src/liblaf/cherries/core/_plugin.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def delegate(
    self,
    method: MethodName,
    args: Sequence[Any] = (),
    kwargs: Mapping[str, Any] = {},
    *,
    first_result: bool = False,
) -> Any:
    plugins: Sequence[Plugin] = self._plugins_sort(method)
    if not plugins:
        if first_result:
            return None
        return []
    results: list[Any] = []
    for plugin in plugins:
        result: Any = getattr(plugin, method)(*args, **kwargs)
        if result is None:
            continue
        if first_result:
            return result
        results.append(result)
    return results

end ¤

end(*args, **kwargs) -> None
Source code in src/liblaf/cherries/core/_run.py 
79
80
@spec
def end(self, *args, **kwargs) -> None: ...

get_url ¤

get_url() -> str
Source code in src/liblaf/cherries/core/_run.py 
82
83
@spec(first_result=True)
def get_url(self) -> str: ...

log_asset ¤

log_asset(
    path: PathLike,
    name: PathLike | None = None,
    *,
    metadata: Mapping[str, Any] | None = None,
    **kwargs,
) -> None
Source code in src/liblaf/cherries/core/_run.py 
85
86
87
88
89
90
91
92
93
94
95
96
97
98
@spec(delegate=False)
def log_asset(
    self,
    path: PathLike,
    name: PathLike | None = None,
    *,
    metadata: Mapping[str, Any] | None = None,
    **kwargs,
) -> None:
    if name is None:
        path = Path(path)
        with contextlib.suppress(ValueError):
            name = path.relative_to(self.data_dir)
    self.delegate("log_asset", (path, name), {"metadata": metadata, **kwargs})

log_input ¤

log_input(
    path: PathLike,
    name: PathLike | None = None,
    *,
    metadata: Mapping[str, Any] | None = None,
    **kwargs,
) -> None
Source code in src/liblaf/cherries/core/_run.py 
100
101
102
103
104
105
106
107
108
@spec
def log_input(
    self,
    path: PathLike,
    name: PathLike | None = None,
    *,
    metadata: Mapping[str, Any] | None = None,
    **kwargs,
) -> None: ...

log_metric ¤

log_metric(
    name: str,
    value: Any,
    /,
    step: int | None = None,
    epoch: int | None = None,
    **kwargs,
) -> None
Source code in src/liblaf/cherries/core/_run.py 
110
111
112
113
114
115
116
117
118
119
@spec
def log_metric(
    self,
    name: str,
    value: Any,
    /,
    step: int | None = None,
    epoch: int | None = None,
    **kwargs,
) -> None: ...

log_metrics ¤

log_metrics(
    dic: Mapping[str, Any],
    /,
    prefix: str | None = None,
    step: int | None = None,
    epoch: int | None = None,
    **kwargs,
) -> None
Source code in src/liblaf/cherries/core/_run.py 
121
122
123
124
125
126
127
128
129
130
@spec
def log_metrics(
    self,
    dic: Mapping[str, Any],
    /,
    prefix: str | None = None,
    step: int | None = None,
    epoch: int | None = None,
    **kwargs,
) -> None: ...

log_other ¤

log_other(key: Any, value: Any, /, **kwargs) -> None
Source code in src/liblaf/cherries/core/_run.py 
132
133
@spec
def log_other(self, key: Any, value: Any, /, **kwargs) -> None: ...

log_others ¤

log_others(
    dictionary: Mapping[Any, Any], /, **kwargs
) -> None
Source code in src/liblaf/cherries/core/_run.py 
135
136
@spec
def log_others(self, dictionary: Mapping[Any, Any], /, **kwargs) -> None: ...

log_output ¤

log_output(
    path: PathLike,
    name: PathLike | None = None,
    *,
    metadata: Mapping[str, Any] | None = None,
    **kwargs,
) -> None
Source code in src/liblaf/cherries/core/_run.py 
138
139
140
141
142
143
144
145
146
@spec
def log_output(
    self,
    path: PathLike,
    name: PathLike | None = None,
    *,
    metadata: Mapping[str, Any] | None = None,
    **kwargs,
) -> None: ...

log_parameter ¤

log_parameter(
    name: Any,
    value: Any,
    /,
    step: int | None = None,
    **kwargs,
) -> None
Source code in src/liblaf/cherries/core/_run.py 
148
149
150
151
@spec
def log_parameter(
    self, name: Any, value: Any, /, step: int | None = None, **kwargs
) -> None: ...

log_parameters ¤

log_parameters(
    parameters: Mapping[Any, Any],
    /,
    prefix: str | None = None,
    step: int | None = None,
    **kwargs,
) -> None
Source code in src/liblaf/cherries/core/_run.py 
153
154
155
156
157
158
159
160
161
@spec
def log_parameters(
    self,
    parameters: Mapping[Any, Any],
    /,
    prefix: str | None = None,
    step: int | None = None,
    **kwargs,
) -> None: ...

plugin_id_cls classmethod ¤

plugin_id_cls() -> str
Source code in src/liblaf/cherries/core/_plugin.py 
21
22
23
@classmethod
def plugin_id_cls(cls) -> str:
    return cls.__name__

register ¤

register(plugin: Plugin) -> None
Source code in src/liblaf/cherries/core/_plugin.py 
58
59
60
def register(self, plugin: "Plugin") -> None:
    plugin._plugin_parent = self  # noqa: SLF001
    self.plugins[plugin.plugin_id] = plugin

start ¤

start(*args, **kwargs) -> None
Source code in src/liblaf/cherries/core/_run.py 
163
164
165
166
@spec(delegate=False)
def start(self, *args, **kwargs) -> None:
    self._plugins_prepare()
    self.delegate("start", args, kwargs)

as_os_path ¤

as_os_path(path: PathLike) -> str

as_os_path(path: None) -> None

as_os_path(path: PathLike | None) -> str | None
Source code in src/liblaf/cherries/paths/_convert.py 
11
12
13
14
def as_os_path(path: PathLike | None) -> str | None:
    if path is None:
        return None
    return str(path)

as_path ¤

as_path(path: PathLike) -> Path

as_path(path: None) -> None

as_path(path: PathLike | None) -> Path | None
Source code in src/liblaf/cherries/paths/_convert.py 
21
22
23
24
def as_path(path: PathLike | None) -> Path | None:
    if path is None:
        return None
    return Path(path)

as_posix ¤

as_posix(path: PathLike) -> str

as_posix(path: None) -> None

as_posix(path: PathLike | None) -> str | None
Source code in src/liblaf/cherries/paths/_convert.py 
31
32
33
34
35
36
def as_posix(path: PathLike | None) -> str | None:
    if path is None:
        return None
    if isinstance(path, str):
        return path
    return Path(path).as_posix()

data ¤

data(
    path: PathLike = "",
    *,
    mkdir: bool = True,
    prefix: PathLike = "data",
) -> Path
Source code in src/liblaf/cherries/paths/_special.py 
14
15
def data(path: PathLike = "", *, mkdir: bool = True, prefix: PathLike = "data") -> Path:
    return _path(path, mkdir=mkdir, prefix=prefix)

end ¤

end() -> None
Source code in src/liblaf/cherries/_entrypoint.py 
12
13
def end() -> None:
    core.active_run.end()

entrypoint ¤

entrypoint(*, absolute: bool = True) -> Path
Source code in src/liblaf/cherries/paths/_path.py 
12
13
14
15
16
@utils.cache
def entrypoint(*, absolute: bool = True) -> Path:
    if absolute:
        return _entrypoint_absolute()
    return _entrypoint_relative()

exp_dir ¤

exp_dir(*, absolute: bool = True) -> Path
Source code in src/liblaf/cherries/paths/_path.py 
35
36
37
38
39
@utils.cache
def exp_dir(*, absolute: bool = True) -> Path:
    if absolute:
        return _exp_dir_absolute()
    return _exp_dir_relative()

git_root ¤

git_root() -> Path
Source code in src/liblaf/cherries/paths/_path.py 
19
20
21
22
23
@utils.cache
def git_root() -> Path:
    entrypoint: Path = _entrypoint_absolute()
    repo = git.Repo(entrypoint, search_parent_directories=True)
    return Path(repo.working_dir)

git_root_safe ¤

git_root_safe() -> Path
Source code in src/liblaf/cherries/paths/_path.py 
26
27
28
29
30
31
32
@utils.cache
def git_root_safe() -> Path:
    try:
        return git_root()
    except git.exc.InvalidGitRepositoryError:
        logger.warning("Not in a git repository, using current directory", once=True)
        return _entrypoint_absolute().parent

input ¤

input(
    path: PathLike,
    extra: PathGenerator | None = None,
    **kwargs,
) -> Path
Source code in src/liblaf/cherries/config/_asset.py 
63
64
65
66
def input(path: PathLike, extra: PathGenerator | None = None, **kwargs) -> Path:  # noqa: A001
    field_info: pydantic.fields.FieldInfo = pydantic.Field(paths.data(path), **kwargs)  # pyright: ignore[reportAssignmentType]
    field_info.metadata.append(MetaAsset(kind=AssetKind.INPUT, extra=extra))
    return field_info  # pyright: ignore[reportReturnType]

output ¤

output(
    path: PathLike,
    extra: PathGenerator | None = None,
    **kwargs,
) -> Path
Source code in src/liblaf/cherries/config/_asset.py 
88
89
90
91
def output(path: PathLike, extra: PathGenerator | None = None, **kwargs) -> Path:
    field_info: pydantic.fields.FieldInfo = pydantic.Field(paths.data(path), **kwargs)  # pyright: ignore[reportAssignmentType]
    field_info.metadata.append(MetaAsset(kind=AssetKind.OUTPUT, extra=extra))
    return field_info  # pyright: ignore[reportReturnType]

params ¤

params(
    path: PathLike = "",
    *,
    mkdir: bool = True,
    prefix: PathLike = "params",
) -> Path
Source code in src/liblaf/cherries/paths/_special.py 
18
19
20
21
def params(
    path: PathLike = "", *, mkdir: bool = True, prefix: PathLike = "params"
) -> Path:
    return _path(path, mkdir=mkdir, prefix=prefix)

path ¤

path(
    path: PathLike = "",
    *,
    mkdir: bool = True,
    prefix: PathLike = "",
) -> Path
Source code in src/liblaf/cherries/paths/_special.py 
24
25
def path(path: PathLike = "", *, mkdir: bool = True, prefix: PathLike = "") -> Path:
    return _path(path, mkdir=mkdir, prefix=prefix)

run ¤

run(
    main: Callable[..., T],
    *,
    profile: ProfileLike | None = None,
) -> T
Source code in src/liblaf/cherries/_entrypoint.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def run[T](main: Callable[..., T], *, profile: profiles.ProfileLike | None = None) -> T:
    run: core.Run = start(profile=profile)
    args: Sequence[Any]
    kwargs: Mapping[str, Any]
    args, kwargs = _make_args(main)
    configs: list[pydantic.BaseModel] = [
        arg
        for arg in itertools.chain(args, *kwargs.values())
        if isinstance(arg, pydantic.BaseModel)
    ]
    for config in configs:
        run.log_parameters(_config.model_dump_without_assets(config, mode="json"))
        for path in _config.get_inputs(config):
            run.log_input(path)
    result: T = main(*args, **kwargs)
    for config in configs:
        for path in _config.get_outputs(config):
            run.log_output(path)
    run.end()
    return result

src ¤

src(
    path: PathLike = "",
    *,
    mkdir: bool = True,
    prefix: PathLike = "src",
) -> Path
Source code in src/liblaf/cherries/paths/_special.py 
28
29
def src(path: PathLike = "", *, mkdir: bool = True, prefix: PathLike = "src") -> Path:
    return _path(path, mkdir=mkdir, prefix=prefix)

start ¤

start(profile: ProfileLike | None = None) -> Run
Source code in src/liblaf/cherries/_entrypoint.py 
38
39
40
41
42
43
44
45
46
def start(
    profile: profiles.ProfileLike | None = None,
) -> core.Run:
    run: core.Run = profiles.factory(profile).init()
    run.start()
    run.log_other("cherries.entrypoint", run.entrypoint.relative_to(run.root_dir))
    run.log_other("cherries.exp_dir", run.exp_dir.relative_to(run.root_dir))
    run.log_other("cherries.start_time", run.start_time)
    return run