jammy.utils package

Submodules

jammy.utils.argument module

class jammy.utils.argument.UniqueValueGetter(msg='unique value check failed', default=None)

Bases: object

get()

set(v)

jammy.utils.argument.asshape(arr_like)

jammy.utils.argument.astuple(arr_like)

Convert a sequence or a single value to a tuple. This method differ from the system method tuple in that a single value (incl. int, string, bytes) will be converted to a tuple of length 1.

Args:

arr_like: a sequence or a single value.

Returns: a tuple.

jammy.utils.argument.canonize_args_list(args, *, allow_empty=False, cvt=None)

Convert the argument list to a tuple of values. This is useful to make unified interface for shape-related operations. E.g.,

>>> np.zeros(6).reshape(2, 3)
>>> np.zeros(6).reshape((2, 3))

Args:

args: allow_empty: cvt:

Returns:

jammy.utils.argument.get_2dshape(x, default=None, type=<class 'int'>)

Convert a value or a tuple to a tuple of length 2.

Args:

x: a value of type type, or a tuple of length 2. default: default value. type: expected type of the element.

Returns: a tuple of length 2.

jammy.utils.argument.get_3dshape(x, default=None, type=<class 'int'>)

jammy.utils.argument.get_4dshape(x, default=None, type=<class 'int'>)

jammy.utils.cache module

class jammy.utils.cache.cached_property(fget)

Bases: object

jammy.utils.cache.cached_result(func)

jammy.utils.cache.fs_cached_result(filename, force_update=False, verbose=False)

jammy.utils.cfg module

class jammy.utils.cfg.Gcfg(*args, **kwargs)

Bases: object

classmethod reset(cfg=None)

jammy.utils.cfg.get_pub_cfg(cfg)

jammy.utils.cfg.omgcfg_create(obj)

jammy.utils.cfg.set_pub_cfg(cfg)

jammy.utils.cnt module

class jammy.utils.cnt.BufferCnt(thres=inf, reset_over_thres=False)

Bases: object

property cnt

reset()

class jammy.utils.cnt.CBCnt(cb_fn=None, thres=inf, reset_over_thres=False)

Bases: BufferCnt

class jammy.utils.cnt.ExitCnt(msg='Something wrong', thres=inf, reset_over_thres=False)

Bases: CBCnt

jammy.utils.cnt.bufcnt(cond)

jammy.utils.colored_tqdm module

class jammy.utils.colored_tqdm.Coloredtqdm(*_, **__)

Bases: tqdm_asyncio

format_num_to_k(seq, k=4)

set_postfix(ordered_dict=None, refresh=True, color=True, round=4, **kwargs)

Set/modify postfix (additional stats) with automatic formatting based on datatype.

Parameters

ordered_dict : dict or OrderedDict, optional refresh : bool, optional

Forces refresh [default: True].

kwargs : dict, optional

jammy.utils.container module

class jammy.utils.container.G

Bases: dict

format(sep=': ', end='\n')

print(sep=': ', end='\n', file=None)

class jammy.utils.container.GView(dict_=None)

Bases: object

copy()

format(sep=': ', end='\n')

print(sep=': ', end='\n', file=None)

raw()

update(other)

class jammy.utils.container.OrderedSet(initial_list=None)

Bases: object

append(value)

as_list()

remove(value)

class jammy.utils.container.SlotAttrObject(**kwargs)

Bases: object

clone()

update(**kwargs)

jammy.utils.context module

class jammy.utils.context.EmptyContext

Bases: object

jammy.utils.debug module

jammy.utils.debug.exception_hook(enable=True)

jammy.utils.debug.hook_exception_ipdb()

jammy.utils.debug.unhook_exception_ipdb()

jammy.utils.defaults module

jammy.utils.defaults.default_args(func)

jammy.utils.defaults.gen_get_default(cls, default_getter=None)

jammy.utils.defaults.gen_set_default(cls)

jammy.utils.defaults.option_context(name, is_local=True, **kwargs)

jammy.utils.defaults.wrap_custom_as_default(*, is_local=False)

jammy.utils.deprecated module

jammy.utils.deprecated.deprecated(func)

jammy.utils.enum module

class jammy.utils.enum.JamEnum(value)

Bases: Enum

A customized enumeration class, adding helper functions for string-based argument parsing.

classmethod assert_valid(value)

classmethod choice_names()

classmethod choice_objs()

classmethod choice_values()

classmethod from_string(value)

classmethod is_valid(value)

classmethod type_name()

jammy.utils.env module

jammy.utils.env.jam_getenv(name, default=None, type=None, prefix=None)

jammy.utils.env.jam_is_debug(default='n', prefix=None)

jammy.utils.env.jam_is_verbose(default='n', prefix=None)

jammy.utils.env.jam_setenv(key, value, prefix=None)

jammy.utils.filelock module

jammy.utils.filelock.get_filelock(target_file, timeout=20, is_soft=False)

jammy.utils.filelock.try_filelock(fn, target_file, is_soft=False, timeout=20)

jammy.utils.git module

jammy.utils.git.git_hash(path)

jammy.utils.git.git_rootdir(path='')

jammy.utils.git.is_git(path)

jammy.utils.gpu module

jammy.utils.gpu.get_mem_util()

jammy.utils.gpu.gpu_by_util()

jammy.utils.gpu.gpu_by_weight(mem_prior=1.0)

jammy.utils.gpu.is_gpu_free(ids=0, mem_thres=0.2, util_thres=0.3)

jammy.utils.hash module

jammy.utils.hash.md5_encode_obj(f_obj)

jammy.utils.hash.md5_encodef(f_input)

jammy.utils.hyd module

jammy.utils.hyd.hyd_instantiate(_cfg, *args, **kwargs)

A helper func helps initialize from omegaconf(hydra) _cfg

Args:

_cfg: omegaconf.DictConfig, must contain _target_ key word

if param or params present in the _cfg, they are high priority to initialize instance otherwise, use _cfg

Note:

when use the method pay attention to the parameters order

args and kwargs: other parameters needed to initialize network

jammy.utils.hyd.hydpath(input_path=None)

jammy.utils.hyd.instantiate(_cfg, *args, **kwargs)

A helper func helps instantiate from omegaconf(hydra) _cfg

Args:

_cfg: omegaconf.DictConfig, must contain _target_ key word

if param or params present in the _cfg, they are high priority to initialize instance otherwise, use _cfg

Note:

when use the method pay attention to the parameters order

args and kwargs: other parameters needed to initialize network

jammy.utils.hyd.link_hyd_run(dst_fname='.latest_exp', proj_path=None)

jammy.utils.hyd.update_cfg(cfg, dotlist)

Parameters:

cfg (DictConfig) –

jammy.utils.imp module

jammy.utils.imp.classname_to_tuple(classname)

jammy.utils.imp.load_class(classname, exit_on_error=True)

jammy.utils.imp.load_module(module_name)

jammy.utils.imp.load_module_filename(module_filename)

jammy.utils.imp.load_source(filename, name=None)

jammy.utils.imp.module_vars_as_dict(module)

jammy.utils.imp.tuple_to_classname(t)

jammy.utils.init module

jammy.utils.init.enable_ipdb()

jammy.utils.init.init_main()

jammy.utils.init.main_path()

jammy.utils.init.release_syslim()

jammy.utils.init.tune_opencv()

jammy.utils.matching module

class jammy.utils.matching.IENameMatcher(include=None, exclude=None)

Bases: object

begin()

end()

get_last_stat()

match(k)

class jammy.utils.matching.NameMatcher(rules=None)

Bases: object

append_rule(rule)

begin(*, force_compile=False)

compile()

end()

get_last_stat()

insert_rule(index, rule)

map()

match(k)

pop_rule(index=None)

property rules

jammy.utils.meta module

class jammy.utils.meta.Singleton

Bases: type

https://stackoverflow.com/a/55629949/7672954

classmethod clear()

property ready

jammy.utils.meta.assert_instance(ins, clz, msg=None)

jammy.utils.meta.assert_none(ins, msg=None)

jammy.utils.meta.assert_notnone(ins, msg=None, name='instance')

jammy.utils.meta.cond_with(with_statement, cond)

jammy.utils.meta.cond_with_group(cond, *with_statement)

jammy.utils.meta.decorator_with_optional_args(func=None, *, is_method=False)

jammy.utils.meta.dict_deep_keys(d, sort=True, sep='.', allow_dict=True)

jammy.utils.meta.dict_deep_kv(d, sort=True, sep='.', allow_dict=False)

jammy.utils.meta.dict_deep_update(a, b)

jammy.utils.meta.filter_exec(func, iterable)

jammy.utils.meta.first_n(iterable, n=10)

jammy.utils.meta.gofor(v)

jammy.utils.meta.make_dummy_func(message=None)

jammy.utils.meta.map_exec(func, *iterables)

jammy.utils.meta.map_exec_method(method_name, iterable)

jammy.utils.meta.merge_iterable(v1, v2)

jammy.utils.meta.method2func(method_name)

class jammy.utils.meta.notnone_property(fget)

Bases: object

jammy.utils.meta.run_once(func)

jammy.utils.meta.stmap(func, iterable)

jammy.utils.meta.synchronized(mutex=None)

jammy.utils.meta.try_run(lambda_)

jammy.utils.meter module

class jammy.utils.meter.AverageMeter

Bases: object

Computes and stores the average and current value

avg = 0

count = 0

reset()

sum = 0

tot_count = 0

update(val, n=1)

val = 0

class jammy.utils.meter.GroupMeters

Bases: object

property avg

dump(filename, values='avg')

format(caption, values, kv_format, glue)

format_simple(caption, values='avg', compressed=True)

items()

reset()

property sum

update(updates=None, value=None, n=1, **kwargs)

Example:

>>> meters.update(key, value)
>>> meters.update({key1: value1, key2: value2})
>>> meters.update(key1=value1, key2=value2)

property val

jammy.utils.mock module

jammy.utils.mock.any_fn(*args, **kwargs)

jammy.utils.naming module

jammy.utils.naming.class_name(instance_or_class)

jammy.utils.naming.class_name_of_method(method)

jammy.utils.naming.func_name(func)

jammy.utils.naming.method_name(method)

jammy.utils.notifier module

jammy.utils.num_check module

jammy.utils.num_check.is_bool(x)

jammy.utils.num_check.is_int(x)

jammy.utils.num_check.is_nonnegative_int(x)

jammy.utils.num_check.is_positive_int(x)

jammy.utils.num_check.is_power_of_two(n)

jammy.utils.printing module

class jammy.utils.printing.PrintToStringContext(target='STDOUT', stream=None, need_lock=True)

Bases: object

get()

jammy.utils.printing.bubbletext(text, font='cybermedium')

Uses pyfiglet to create bubble text. Args:

font (str): default=cybermedium, other fonts include: cybersmall and

cyberlarge.

References:

http://www.figlet.org/

Example:

>>> bubble_text = bubbletext('TESTING BUBBLE TEXT', font='cybermedium')
>>> print(bubble_text)

jammy.utils.printing.indent_text(text, level=1, indent_format=None, tabsize=None)

jammy.utils.printing.kvformat(data, indent=0, sep=' : ', end='\n', max_key_len=None)

jammy.utils.printing.kvprint(data, indent=0, sep=' : ', end='\n', max_key_len=None, file=None, float_format='{:.6f}', need_lock=True)

jammy.utils.printing.print2format(print_func)

jammy.utils.printing.print_to(print_func, target='STDOUT', rstrip=True)

jammy.utils.printing.print_to_string(target='STDOUT')

jammy.utils.printing.stformat(data, key=None, indent=0, max_depth=100, **kwargs)

jammy.utils.printing.stprint(data, key=None, indent=0, file=None, indent_format='  ', end_format='\n', float_format='{:.6f}', need_lock=True, max_depth=100)

Structure print.

Example:

>>> data = dict(a=np.zeros(shape=(10, 10)), b=3)
>>> stprint(data)
dict{
    a: ndarray(10, 10), dtype=float64
    b: 3
}

Args:

data: data to be print. Currently support Sequnce, Mappings and primitive types. key: for recursion calls. Do not use it if you don’t know how it works. indent: indent level.

jammy.utils.process module

class jammy.utils.process.BashRunner(notifier=None, interval_cpu_check=1.0, cpu_usage_thres=92, seconds_numthread_wait=10, mem_avail_thres=0.9, seconds_resouece_wait=2)

Bases: object

Parameters:

• interval_cpu_check (float) –

• cpu_usage_thres (float) –

• seconds_numthread_wait (int) –

• mem_avail_thres (float) –

• seconds_resouece_wait (int) –

check_num_thread(num_thread)

execute_tasks(batch_cmd, num_thread=0, msg=None)

kill_all()

proc_event(cur_process)

queue_tasks(batch_cmd, msg=None)

reset(batch_cmd)

jammy.utils.process.run_simple_command(cmd)

Parameters:

cmd (str) –

jammy.utils.profiler module

jammy.utils.profiler.profile(func)

dummy profiling func. does nothing

jammy.utils.profiler.profile_now(func)

Eagerly report profile information after each call to func.

Args:

func (Callable): function to profile

Example:

>>> # xdoctest: +SKIP
>>> from xdev.profiler import *  # NOQA
>>> def func_to_profile():
>>>     list(range(10))
>>>     tuple(range(100))
>>>     set(range(1000))
>>> profile_now(func_to_profile)()  # xdoctest: +IGNORE_WANT
Total time: 3.8e-05 s
Function: func_to_profile at line 1
Line #      Hits         Time  Per Hit   % Time  Line Contents
==============================================================
     1                                           def func_to_profile():
     2         1          4.0      4.0     10.5      list(range(10))
     3         1          3.0      3.0      7.9      tuple(range(100))
     4         1         31.0     31.0     81.6      set(range(1000))

jammy.utils.registry module

class jammy.utils.registry.CallbackRegistry

Bases: Registry

A callable manager utils.

If there exists a super callback, it will block all callbacks. A super callback will receive the called name as its first argument.

Then the dispatcher will try to call the callback by name. If such name does not exists, a fallback callback will be called.

The fallback callback will also receive the called name as its first argument.

Examples:

>>> registry = CallbackRegistry()
>>> callback_func = print
>>> registry.register('name', callback_func)  # register a callback.
>>> registry.dispatch('name', 'arg1', 'arg2', kwarg1='kwarg1')  # dispatch.

dispatch(name, *args, **kwargs)

dispatch_direct(name, *args, **kwargs)

Dispatch by name, ignoring the super callback.

property fallback_callback

set_fallback_callback(callback)

set_super_callback(callback)

property super_callback

class jammy.utils.registry.DefaultRegistry

Bases: Registry

lookup(entry, fallback=False, default=None)

class jammy.utils.registry.LockRegistry

Bases: DefaultRegistry

synchronized(entry, activate=True)

class jammy.utils.registry.Registry

Bases: object

property fallback

has(entry)

items()

keys()

lookup(entry, fallback=True, default=None)

register(entry, value)

set_fallback(value)

unregister(entry)

class jammy.utils.registry.RegistryGroup

Bases: object

lookup(registry_name, entry, fallback=True, default=None)

register(registry_name, entry, value, **kwargs)

jammy.utils.retry module

jammy.utils.retry.retry(exceptions=<class 'Exception'>, tries=-1, delay=0, max_delay=None, backoff=1, jitter=0, logger=<loguru.logger handlers=[(id=1, level=20, sink=<stderr>)]>)

Returns a retry decorator. :param exceptions: an exception or a tuple of exceptions to catch. default: Exception. :param tries: the maximum number of attempts. default: -1 (infinite). :param delay: initial delay between attempts. default: 0. :param max_delay: the maximum value of delay. default: None (no limit). :param backoff: multiplier applied to delay between attempts. default: 1 (no backoff). :param jitter: extra seconds added to delay between attempts. default: 0.

fixed if a number, random if a range tuple (min, max)

Parameters:

logger – logger.warning(fmt, error, delay) will be called on failed attempts. default: retry.logging_logger. if None, logging is disabled.

Returns:

a retry decorator.

jammy.utils.retry.retry_call(f, fargs=None, fkwargs=None, exceptions=<class 'Exception'>, tries=-1, delay=0, max_delay=None, backoff=1, jitter=0, logger=<loguru.logger handlers=[(id=1, level=20, sink=<stderr>)]>)

Calls a function and re-executes it if it failed. :param f: the function to execute. :param fargs: the positional arguments of the function to execute. :param fkwargs: the named arguments of the function to execute. :param exceptions: an exception or a tuple of exceptions to catch. default: Exception. :param tries: the maximum number of attempts. default: -1 (infinite). :param delay: initial delay between attempts. default: 0. :param max_delay: the maximum value of delay. default: None (no limit). :param backoff: multiplier applied to delay between attempts. default: 1 (no backoff). :param jitter: extra seconds added to delay between attempts. default: 0.

fixed if a number, random if a range tuple (min, max)

Parameters:

logger – logger.warning(fmt, error, delay) will be called on failed attempts. default: retry.logging_logger. if None, logging is disabled.

Returns:

the result of the f function.

Module contents