Skip to content

liblaf.cherries ¤

Modules:

Classes:

Functions:

Attributes:

__version__ module-attribute ¤

__version__: str = '0.6.3.dev2+g2d7f2f323'

__version_tuple__ module-attribute ¤

__version_tuple__: tuple[int | str, ...] = (
    0,
    6,
    3,
    "dev2",
    "g2d7f2f323",
)

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

version module-attribute ¤

version: str = '0.6.3.dev2+g2d7f2f323'

version_tuple module-attribute ¤

version_tuple: tuple[int | str, ...] = (
    0,
    6,
    3,
    "dev2",
    "g2d7f2f323",
)

BaseConfig ¤

Bases: BaseSettings

Attributes:

model_config class-attribute instance-attribute ¤

model_config = SettingsConfigDict()

Plugin ¤

Parameters:

  • plugins (dict[PluginId, str], default: <class 'dict'> ) –

    dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object’s (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

Methods:

Attributes:

plugin_id property ¤

plugin_id: str

plugin_root property ¤

plugin_root: Self

plugins class-attribute instance-attribute ¤

plugins: dict[PluginId, 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 
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
def delegate(
    self,
    method: MethodName,
    args: Sequence[Any] = (),
    kwargs: Mapping[str, Any] = {},
    *,
    first_result: bool = False,
) -> Any:
    __tracebackhide__ = True
    plugins: Sequence[Plugin] = self._plugins_sort(method)
    if not plugins:
        if first_result:
            return None
        return []
    results: list[Any] = []
    for plugin in plugins:
        try:
            result: Any = getattr(plugin, method)(*args, **kwargs)
        except BaseException as e:
            if isinstance(e, (KeyboardInterrupt, SystemExit)):
                raise
            logger.exception("Plugin {}", plugin.plugin_id)
        else:
            if result is None:
                continue
            if first_result:
                return result
            results.append(result)
    return results

register ¤

register(plugin: Plugin) -> None
Source code in src/liblaf/cherries/core/_plugin.py 
64
65
66
67
68
69
def register(self, plugin: "Plugin") -> None:
    impls: dict[MethodName, ImplInfo] = collect_impls(plugin)
    for name in impls:
        self._cache_sort_plugins.pop(name, None)
    plugin._plugin_parent = self  # noqa: SLF001
    self.plugins[plugin.plugin_id] = plugin

Run ¤

Bases: Plugin

.

References
  1. Experiment - Comet Docs
  2. Logger | ClearML
  3. MLflow Tracking APIs | MLflow

Parameters:

  • plugins (dict[PluginId, str], default: <class 'dict'> ) –

    dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object’s (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

Methods:

Attributes:

plugin_id property ¤

plugin_id: str

plugin_root property ¤

plugin_root: Self

plugins class-attribute instance-attribute ¤

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

data_dir ¤

data_dir() -> Path
Source code in src/liblaf/cherries/core/_run.py 
27
28
29
@plugin_cached_property
def data_dir(self) -> Path:
    return path_utils.data()

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 
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
def delegate(
    self,
    method: MethodName,
    args: Sequence[Any] = (),
    kwargs: Mapping[str, Any] = {},
    *,
    first_result: bool = False,
) -> Any:
    __tracebackhide__ = True
    plugins: Sequence[Plugin] = self._plugins_sort(method)
    if not plugins:
        if first_result:
            return None
        return []
    results: list[Any] = []
    for plugin in plugins:
        try:
            result: Any = getattr(plugin, method)(*args, **kwargs)
        except BaseException as e:
            if isinstance(e, (KeyboardInterrupt, SystemExit)):
                raise
            logger.exception("Plugin {}", plugin.plugin_id)
        else:
            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 
67
68
@spec
def end(self, *args, **kwargs) -> None: ...

entrypoint ¤

entrypoint() -> Path
Source code in src/liblaf/cherries/core/_run.py 
31
32
33
@plugin_cached_property
def entrypoint(self) -> Path:
    return path_utils.entrypoint()

exp_dir ¤

exp_dir() -> Path
Source code in src/liblaf/cherries/core/_run.py 
35
36
37
@plugin_cached_property
def exp_dir(self) -> Path:
    return path_utils.exp_dir()

exp_name ¤

exp_name() -> str
Source code in src/liblaf/cherries/core/_run.py 
39
40
41
42
43
44
45
@plugin_cached_property
def exp_name(self) -> str:
    return (
        self.entrypoint.relative_to(self.project_dir)
        .as_posix()
        .removeprefix("exp/")
    )

get_others ¤

get_others() -> Mapping[str, Any]
Source code in src/liblaf/cherries/core/_run.py 
70
71
@spec(first_result=True)
def get_others(self) -> Mapping[str, Any]: ...

get_params ¤

get_params() -> Mapping[str, Any]
Source code in src/liblaf/cherries/core/_run.py 
73
74
@spec(first_result=True)
def get_params(self) -> Mapping[str, Any]: ...

get_url ¤

get_url() -> str
Source code in src/liblaf/cherries/core/_run.py 
76
77
@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 
79
80
81
82
83
84
85
86
87
@spec
def log_asset(
    self,
    path: PathLike,
    name: PathLike | None = None,
    *,
    metadata: Mapping[str, Any] | None = None,
    **kwargs,
) -> None: ...

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 
89
90
91
92
93
94
95
96
97
@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 
 99
100
101
102
103
104
105
106
107
108
@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 
110
111
112
113
114
115
116
117
118
119
@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 
121
122
@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 
124
125
@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 
127
128
129
130
131
132
133
134
135
@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 
137
138
139
140
@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 
142
143
144
145
146
147
148
149
150
@spec
def log_parameters(
    self,
    parameters: Mapping[Any, Any],
    /,
    prefix: str | None = None,
    step: int | None = None,
    **kwargs,
) -> None: ...

log_temporary ¤

log_temporary(
    path: PathLike,
    name: PathLike | None = None,
    *,
    metadata: Mapping[str, Any] | None = None,
    **kwargs,
) -> None
Source code in src/liblaf/cherries/core/_run.py 
152
153
154
155
156
157
158
159
160
@spec
def log_temporary(
    self,
    path: PathLike,
    name: PathLike | None = None,
    *,
    metadata: Mapping[str, Any] | None = None,
    **kwargs,
) -> None: ...

params ¤

params() -> Mapping[str, Any]
Source code in src/liblaf/cherries/core/_run.py 
47
48
49
@plugin_property
def params(self) -> Mapping[str, Any]:
    return self.plugin_root.get_params()

project_dir ¤

project_dir() -> Path
Source code in src/liblaf/cherries/core/_run.py 
55
56
57
@plugin_cached_property
def project_dir(self) -> Path:
    return path_utils.project_dir()

project_name ¤

project_name() -> str | None
Source code in src/liblaf/cherries/core/_run.py 
51
52
53
@plugin_cached_property
def project_name(self) -> str | None:
    return self.project_dir.name

register ¤

register(plugin: Plugin) -> None
Source code in src/liblaf/cherries/core/_plugin.py 
64
65
66
67
68
69
def register(self, plugin: "Plugin") -> None:
    impls: dict[MethodName, ImplInfo] = collect_impls(plugin)
    for name in impls:
        self._cache_sort_plugins.pop(name, 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 
162
163
164
165
@spec(delegate=False)
def start(self, *args, **kwargs) -> None:
    env.read_env(self.entrypoint.parent / ".env")
    return self.delegate("start", args, kwargs)

start_time ¤

start_time() -> datetime
Source code in src/liblaf/cherries/core/_run.py 
59
60
61
@plugin_cached_property
def start_time(self) -> datetime.datetime:
    return datetime.datetime.now().astimezone()

url ¤

url() -> str
Source code in src/liblaf/cherries/core/_run.py 
63
64
65
@plugin_property
def url(self) -> str:
    return self.plugin_root.get_url()

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/path_utils/_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/path_utils/_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/path_utils/_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/path_utils/_special.py 
14
15
def data(path: PathLike = "", *, mkdir: bool = True, prefix: PathLike = "data") -> Path:
    return _path(path, mkdir=mkdir, prefix=prefix)

end ¤

end(*args, **kwargs) -> None
Source code in src/liblaf/cherries/_entrypoint.py 
13
14
15
def end(*args, **kwargs) -> None:
    core.active_run.log_other("cherries.end_time", datetime.datetime.now().astimezone())
    core.active_run.end(*args, **kwargs)

entrypoint ¤

entrypoint(*, absolute: bool = True) -> Path
Source code in src/liblaf/cherries/path_utils/_path.py 
11
12
13
14
15
16
@utils.cache
def entrypoint(*, absolute: bool = True) -> Path:
    path = Path(sys.argv[0])
    if absolute:
        return path.absolute()
    return path.relative_to(project_dir())

exp_dir ¤

exp_dir(*, absolute: bool = True) -> Path
Source code in src/liblaf/cherries/path_utils/_path.py 
22
23
24
25
26
27
28
29
30
31
@utils.cache
def exp_dir(*, absolute: bool = True) -> Path:
    entrypoint_: Path = entrypoint(absolute=True)
    parent: Path = entrypoint_.parent
    exp_dir: Path
    exp_dir = parent.parent if parent.name in EXP_DIR_NAMES else parent
    if absolute:
        return exp_dir
    exp_dir = exp_dir.relative_to(project_dir())
    return exp_dir

input ¤

input(
    path: PathLike, extra: Extra = None, **kwargs
) -> Path
Source code in src/liblaf/cherries/config/asset/_meta.py 
78
79
def input(path: PathLike, extra: Extra = None, **kwargs) -> Path:  # noqa: A001
    return asset(path, extra=extra, kind=AssetKind.INPUT, **kwargs)

output ¤

output(
    path: PathLike, extra: Extra = None, **kwargs
) -> Path
Source code in src/liblaf/cherries/config/asset/_meta.py 
102
103
def output(path: PathLike, extra: Extra = None, **kwargs) -> Path:
    return asset(path, extra=extra, kind=AssetKind.OUTPUT, **kwargs)

params ¤

params(
    path: PathLike = "",
    *,
    mkdir: bool = True,
    prefix: PathLike = "params",
) -> Path
Source code in src/liblaf/cherries/path_utils/_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/path_utils/_special.py 
24
25
def path(path: PathLike = "", *, mkdir: bool = True, prefix: PathLike = "") -> Path:
    return _path(path, mkdir=mkdir, prefix=prefix)

project_dir ¤

project_dir() -> Path
Source code in src/liblaf/cherries/path_utils/_path.py 
34
35
36
37
38
39
40
41
42
43
@utils.cache
def project_dir() -> Path:
    exp_dir_: Path = exp_dir(absolute=True)
    try:
        repo = git.Repo(exp_dir_, search_parent_directories=True)
    except git.exc.InvalidGitRepositoryError:
        logger.warning("Not in a git repository, using current directory", once=True)
        return exp_dir_
    else:
        return Path(repo.working_dir)

run ¤

run(
    main: Callable[..., T],
    *,
    profile: ProfileLike | None = None,
) -> T
Source code in src/liblaf/cherries/_entrypoint.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
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)
    try:
        return main(*args, **kwargs)
    finally:
        for config in configs:
            for path in _config.get_outputs(config):
                run.log_output(path)
            for path in _config.get_temporaries(config):
                run.log_temporary(path)
        run.end()

src ¤

src(
    path: PathLike = "",
    *,
    mkdir: bool = True,
    prefix: PathLike = "src",
) -> Path
Source code in src/liblaf/cherries/path_utils/_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 
43
44
45
46
47
48
49
50
51
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.project_dir))
    run.log_other("cherries.exp_dir", run.exp_dir.relative_to(run.project_dir))
    run.log_other("cherries.start_time", run.start_time)
    return run

temporary ¤

temporary(
    path: PathLike, extra: Extra = None, **kwargs
) -> Path
Source code in src/liblaf/cherries/config/asset/_meta.py 
106
107
def temporary(path: PathLike, extra: Extra = None, **kwargs) -> Path:
    return asset(path, extra=extra, kind=AssetKind.TEMPORARY, **kwargs)