ai2_kit.core.util module#

class ai2_kit.core.util.JoinTag[source]#

Bases: object

a tag to join strings in a list

classmethod from_yaml(constructor, node)[source]#

classmethod register(yaml: YAML)[source]#

classmethod to_yaml(dumper, data)[source]#

yaml_tag = '!join'#

class ai2_kit.core.util.LoadTextTag[source]#

Bases: object

a tag to read string from file

classmethod from_yaml(constructor, node)[source]#

classmethod register(yaml: YAML)[source]#

classmethod to_yaml(dumper, data)[source]#

yaml_tag = '!load_text'#

class ai2_kit.core.util.LoadYamlTag[source]#

Bases: object

a tag to read string from file

classmethod from_yaml(constructor, node)[source]#

classmethod register(yaml: YAML)[source]#

classmethod to_yaml(dumper, data)[source]#

yaml_tag = '!load_yaml'#

ai2_kit.core.util.cmd_with_checkpoint(cmd: str, checkpoint: str, ignore_error: bool = False)[source]#

Add checkpoint to a shell command. If a checkpoint file exists, the command will be skipped, otherwise the command will be executed and a checkpoint file will be created if the command is executed successfully.

Parameters:

cmd – shell command
checkpoint – checkpoint file name
ignore_error – if True, will not raise error if the command failed

ai2_kit.core.util.create_fn(func_str: str, fn_name: str)[source]#

ai2_kit.core.util.dict_nested_get(d: dict, keys: ~typing.List[str], default=<object object>)[source]#: get value from nested dict

ai2_kit.core.util.dict_nested_set(d: dict, keys: List[str], value)[source]#: set value to nested dict

ai2_kit.core.util.dict_remove_dot_keys(d)[source]#

ai2_kit.core.util.dump_json(obj, path: str)[source]#

ai2_kit.core.util.dump_text(text: str, path: str, **kwargs)[source]#

ai2_kit.core.util.ensure_dir(path: str)[source]#

ai2_kit.core.util.expand_globs(patterns: Iterable[str], raise_invalid=False) → List[str][source]#

Expand glob patterns in paths

Parameters:

patterns – list of paths or glob patterns
raise_invalid – if True, will raise error if no file found for a glob pattern

Returns:

list of expanded paths

ai2_kit.core.util.flat_evenly(list_of_lists)[source]#: flat a list of lists and ensure the output result distributed evenly >>> flat_evenly([[1, 2, 3], [4, 5, 6], [7, 8, 9]]) [1, 4, 7, 2, 5, 8, 3, 6, 9] Ref: https://stackoverflow.com/questions/76751171/how-to-flat-a-list-of-lists-and-ensure-the-output-result-distributed-evenly-in-p

ai2_kit.core.util.flatten(l: List[List[T]]) → List[T][source]#

ai2_kit.core.util.flush_stdio()[source]#

ai2_kit.core.util.format_env_string(s: str) → str[source]#

ai2_kit.core.util.get_yaml()[source]#

ai2_kit.core.util.limit(it, size=-1)[source]#: limit the size of an iterable

ai2_kit.core.util.list_even_sample(l, size)[source]#

ai2_kit.core.util.list_random_sample(l, size, seed=None)[source]#

ai2_kit.core.util.list_sample(l, size: int, method: Literal['even', 'random', 'truncate'] = 'even', **kwargs)[source]#

sample list

Parameters:

size – size of sample, if size is larger than data size, return all data
method – method to sample, can be ‘even’, ‘random’, ‘truncate’, default is ‘even’
seed – seed for random sample, only used when method is ‘random’

Note that by default the seed is length of input list, if you want to generate different sample each time, you should set random seed manually

ai2_kit.core.util.list_split(l: List[T], n: int) → List[List[T]][source]#: split list into n chunks

ai2_kit.core.util.load_json(path: Path | str, encoding: str = 'utf-8')[source]#

ai2_kit.core.util.load_text(path: Path | str, encoding: str = 'utf-8')[source]#

ai2_kit.core.util.load_yaml_file(path: Path | str)[source]#

ai2_kit.core.util.load_yaml_files(*paths: Path | str, quiet: bool = False, purge_anonymous=True)[source]#

ai2_kit.core.util.merge_dict(lo: dict, ro: dict, path=None, ignore_none=True, quiet=False)[source]#: Merge two dict, the left dict will be overridden. Note: list will be replaced instead of merged.

ai2_kit.core.util.nested_set(d: dict, keys: List[str], value)[source]#

ai2_kit.core.util.parse_path_list(path_list_str: str | List[str], to_abs: bool = False)[source]#: Parse path list of environment variable style string

ai2_kit.core.util.perf_log(msg)[source]#

ai2_kit.core.util.s_uuid()[source]#: short uuid

ai2_kit.core.util.short_hash(s: str) → str[source]#: short hash string

ai2_kit.core.util.slice_from_str(index: str)[source]#: get slice object from a string expression, for example: “1:10” -> slice(1, 10), “:10” -> slice(None, 10), etc

ai2_kit.core.util.sort_unique_str_list(l: List[str]) → List[str][source]#: remove duplicate str and sort

async ai2_kit.core.util.to_awaitable(value: T) → T[source]#

ai2_kit.core.util.wait_for_change(widget, attribute)[source]#: Wait for attribute change of a Jupyter widget