Skip to content

liblaf.cherries

Modules:

  • bundle –
  • config –
  • core –
  • meta –
  • plugins –
  • profiles –

Classes:

Functions:

Attributes:

__version__ module-attribute 

__version__: str = '1.1.3.dev4+g7a08421ad'

__version_tuple__ module-attribute 

__version_tuple__: tuple[int | str, ...] = (
    1,
    1,
    3,
    "dev4",
    "g7a08421ad",
)

run module-attribute 

run: Run = Run()

BaseConfig

Bases: BaseSettings


              flowchart TD
              liblaf.cherries.BaseConfig[BaseConfig]

              

              click liblaf.cherries.BaseConfig href "" "liblaf.cherries.BaseConfig"

Run

Bases: PluginManager


              flowchart TD
              liblaf.cherries.Run[Run]
              liblaf.cherries.core._plugin_manager.PluginManager[PluginManager]

                              liblaf.cherries.core._plugin_manager.PluginManager --> liblaf.cherries.Run
                


              click liblaf.cherries.Run href "" "liblaf.cherries.Run"
              click liblaf.cherries.core._plugin_manager.PluginManager href "" "liblaf.cherries.core._plugin_manager.PluginManager"

Parameters:

  • plugins (dict[PluginId, Plugin], 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:

data_dir cached property 

data_dir: Path

entrypoint cached property 

entrypoint: Path

exp_dir cached property 

exp_dir: Path

exp_name cached property 

exp_name: str

fig_dir cached property 

fig_dir: Path

logs_dir cached property 

logs_dir: Path

plugins class-attribute instance-attribute 

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

project_dir cached property 

project_dir: Path

project_name cached property 

project_name: str

start_time cached property 

start_time: datetime

step property writable 

step: int | None

temp_dir cached property 

temp_dir: Path

url property 

url: str

asset 

asset(path: StrPath, *, mkdir: bool = False) -> Path
Source code in src/liblaf/cherries/core/_run.py 
107
108
109
110
111
112
def asset(self, path: StrPath, *, mkdir: bool = False) -> Path:
    absolute: Path = self.exp_dir / path
    if mkdir:
        absolute.parent.mkdir(parents=True, exist_ok=True)
    self._assets_queue.append(absolute)
    return absolute

delegate 

delegate(
    method: MethodName,
    args: Sequence[Any] = (),
    kwargs: Mapping[str, Any] = {},
    *,
    first_result: bool = False,
) -> Any
Source code in src/liblaf/cherries/core/_plugin_manager.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def delegate(
    self,
    method: MethodName,
    args: Sequence[Any] = (),
    kwargs: Mapping[str, Any] = {},
    *,
    first_result: bool = False,
) -> Any:
    __tracebackhide__ = True
    results: list[Any] = []
    for plugin in self._sort_plugins(method):
        try:
            result: Any = getattr(plugin, method)(*args, **kwargs)
        except Exception:
            logger.exception("Plugin %s", plugin.id)
        else:
            if result is None:
                continue
            if first_result:
                return result
            results.append(result)
    if first_result:
        return None
    return results

end 

end(*args, **kwargs) -> None
Source code in src/liblaf/cherries/core/_run.py 
135
136
137
138
139
140
141
142
143
144
145
146
def end(self, *args, **kwargs) -> None:
    __tracebackhide__ = True
    self.log_other("cherries.end_time", datetime.now().astimezone())
    for path in self._assets_queue:
        self.log_asset(path)
    for path in self._inputs_queue:
        self.log_input(path)
    for path in self._outputs_queue:
        self.log_output(path)
    for path in self._temps_queue:
        self.log_temp(path)
    self.delegate("end", args, kwargs)

get_other 

get_other(name: str) -> Any
Source code in src/liblaf/cherries/core/_run.py 
148
149
@delegate(first_result=True)
def get_other(self, name: str) -> Any: ...

get_others 

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

get_param 

get_param(name: str) -> Any
Source code in src/liblaf/cherries/core/_run.py 
154
155
@delegate(first_result=True)
def get_param(self, name: str) -> Any: ...

get_params 

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

get_step 

get_step() -> int | None
Source code in src/liblaf/cherries/core/_run.py 
160
161
@delegate(first_result=True)
def get_step(self) -> int | None: ...

get_url 

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

input 

input(path: StrPath, *, mkdir: bool = False) -> Path
Source code in src/liblaf/cherries/core/_run.py 
114
115
116
117
118
119
def input(self, path: StrPath, *, mkdir: bool = False) -> Path:
    absolute: Path = self.data_dir / path
    if mkdir:
        absolute.parent.mkdir(parents=True, exist_ok=True)
    self._inputs_queue.append(absolute)
    return absolute

log_asset 

log_asset(path: StrPath, **kwargs) -> None
Source code in src/liblaf/cherries/core/_run.py 
166
167
168
def log_asset(self, path: StrPath, **kwargs) -> None:
    __tracebackhide__ = True
    self._log_asset(path, "log_asset", self.exp_dir, **kwargs)

log_input 

log_input(path: StrPath, **kwargs) -> None
Source code in src/liblaf/cherries/core/_run.py 
170
171
172
def log_input(self, path: StrPath, **kwargs) -> None:
    __tracebackhide__ = True
    self._log_asset(path, "log_input", self.data_dir, **kwargs)

log_metric 

log_metric(
    name: str, value: Any, step: int | None = None, **kwargs
) -> None
Source code in src/liblaf/cherries/core/_run.py 
182
183
184
185
@delegate
def log_metric(
    self, name: str, value: Any, step: int | None = None, **kwargs
) -> None: ...

log_metrics 

log_metrics(
    metrics: Mapping[str, Any],
    step: int | None = None,
    **kwargs,
) -> None
Source code in src/liblaf/cherries/core/_run.py 
187
188
189
190
@delegate
def log_metrics(
    self, metrics: Mapping[str, Any], step: int | None = None, **kwargs
) -> None: ...

log_other 

log_other(name: str, value: Any) -> None
Source code in src/liblaf/cherries/core/_run.py 
192
193
@delegate
def log_other(self, name: str, value: Any) -> None: ...

log_others 

log_others(others: Mapping[str, Any]) -> None
Source code in src/liblaf/cherries/core/_run.py 
195
196
@delegate
def log_others(self, others: Mapping[str, Any]) -> None: ...

log_output 

log_output(path: StrPath, **kwargs) -> None
Source code in src/liblaf/cherries/core/_run.py 
174
175
176
def log_output(self, path: StrPath, **kwargs) -> None:
    __tracebackhide__ = True
    self._log_asset(path, "log_output", self.data_dir, **kwargs)

log_param 

log_param(name: str, value: Any) -> None
Source code in src/liblaf/cherries/core/_run.py 
198
199
@delegate
def log_param(self, name: str, value: Any) -> None: ...

log_params 

log_params(params: Mapping[str, Any]) -> None
Source code in src/liblaf/cherries/core/_run.py 
201
202
@delegate
def log_params(self, params: Mapping[str, Any]) -> None: ...

log_temp 

log_temp(path: StrPath, **kwargs) -> None
Source code in src/liblaf/cherries/core/_run.py 
178
179
180
def log_temp(self, path: StrPath, **kwargs) -> None:
    __tracebackhide__ = True
    self._log_asset(path, "log_temp", self.temp_dir, **kwargs)

output 

output(path: StrPath, *, mkdir: bool = False) -> Path
Source code in src/liblaf/cherries/core/_run.py 
121
122
123
124
125
126
def output(self, path: StrPath, *, mkdir: bool = False) -> Path:
    absolute: Path = self.data_dir / path
    if mkdir:
        absolute.parent.mkdir(parents=True, exist_ok=True)
    self._outputs_queue.append(absolute)
    return absolute

register 

register(plugin: Plugin) -> None
Source code in src/liblaf/cherries/core/_plugin_manager.py 
58
59
60
61
62
def register(self, plugin: Plugin) -> None:
    for method_name in collect_impls(plugin):
        self._sort_plugins_cache.pop(method_name, None)
    plugin.manager = self
    self.plugins[plugin.id] = plugin

set_step 

set_step(step: int | None = None) -> None
Source code in src/liblaf/cherries/core/_run.py 
204
205
@delegate
def set_step(self, step: int | None = None) -> None: ...

start 

start(*args, **kwargs) -> None
Source code in src/liblaf/cherries/core/_run.py 
207
208
209
210
211
212
213
214
215
216
217
def start(self, *args, **kwargs) -> None:
    __tracebackhide__ = True
    self.delegate("start", args, kwargs)
    self.log_other(
        "cherries.entrypoint",
        _relative_or_absolute(self.entrypoint, self.project_dir),
    )
    self.log_other(
        "cherries.exp_dir", _relative_or_absolute(self.exp_dir, self.project_dir)
    )
    self.log_other("cherries.start_time", self.start_time)

temp 

temp(path: StrPath, *, mkdir: bool = False) -> Path
Source code in src/liblaf/cherries/core/_run.py 
128
129
130
131
132
133
def temp(self, path: StrPath, *, mkdir: bool = False) -> Path:
    absolute: Path = self.temp_dir / path
    if mkdir:
        absolute.parent.mkdir(parents=True, exist_ok=True)
    self._temps_queue.append(absolute)
    return absolute

asset 

asset(path: StrPath, *, mkdir: bool = False) -> Path

end 

end() -> None
Source code in src/liblaf/cherries/_main/_end.py 
4
5
def end() -> None:
    core.run.end()

get_other 

get_other(name: str) -> Any

get_others 

get_others() -> Mapping[str, Any]

get_param 

get_param(name: str) -> Any

get_params 

get_params() -> Mapping[str, Any]

input 

input(path: StrPath, *, mkdir: bool = False) -> Path

log_asset 

log_asset(path: StrPath, **kwargs) -> None

log_input 

log_input(path: StrPath, **kwargs) -> None

log_metric 

log_metric(
    name: str, value: Any, step: int | None = None, **kwargs
) -> None

log_metrics 

log_metrics(
    metrics: Mapping[str, Any],
    step: int | None = None,
    **kwargs,
) -> None

log_other 

log_other(name: str, value: Any) -> None

log_others 

log_others(others: Mapping[str, Any]) -> None

log_output 

log_output(path: StrPath, **kwargs) -> None

log_param 

log_param(name: str, value: Any) -> None

log_params 

log_params(params: Mapping[str, Any]) -> None

log_temp 

log_temp(path: StrPath, **kwargs) -> None

main 

main[T](
    main: Callable[..., T],
    *,
    profile: ProfileLike | None = None,
) -> T
Source code in src/liblaf/cherries/_main/_main.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def main[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 (*args, *kwargs.values()) if isinstance(arg, pydantic.BaseModel)
    ]
    for config in configs:
        run.log_params(config.model_dump(mode="json"))
    try:
        if inspect.iscoroutinefunction(main):
            return asyncio.run(main(*args, **kwargs))
        return main(*args, **kwargs)
    finally:
        run.end()

output 

output(path: StrPath, *, mkdir: bool = False) -> Path

set_step 

set_step(step: int | None = None) -> None

start 

start(profile: ProfileLike | None = None) -> Run
Source code in src/liblaf/cherries/_main/_start.py 
5
6
7
8
9
def start(profile: ProfileLike | None = None) -> core.Run:
    profile = profiles.factory(profile)
    run: core.Run = profile.init()
    run.start()
    return run

temp 

temp(path: StrPath, *, mkdir: bool = False) -> Path