Utils

Stopwatch

class reip.util.stopwatch.Stopwatch(title='', max_samples=500)[source]

Stopwatch for timing and recording execution time of bits of code.

sw = reip.util.Stopwatch()

for _ in range(100):
    with sw('sleeping'):
        time.sleep(0.1)

Statistics

class reip.util.statistics.OnlineStats(history=0)[source]

Online statistics - calculate and update mean and variance as new data comes in.

append(x)[source]: Append a value to the rolling mean.

extend(xs)[source]: Append multiple values.

property sum: Get the rolling sum.

property std: Get the rolling standard deviation.

property var: Get the rolling variance.

Iterators

reip.util.iters.loop(i=0)[source]: Infinite loop

reip.util.iters.timed(it=None, duration=None, error=False)[source]: Run a loop for a predetermined amount of time.

reip.util.iters.throttled(it=None, rate=None, interval=None, delay=1e-05, initial=None)[source]: Throttle a loop to take

reip.util.iters.peakiter(it, n=1)[source]: Check the value first n items of an iterator without unloading them from the iterator queue.

reip.util.iters.npgenarray(it, shape, **kw)[source]: Create a np.ndarray from a generator. Must specify at least the length of the generator or the entire shape of the final array.

Functions

reip.util.func.throttle(func, interval=None)[source]: Don’t call a function more frequently than duration. Caches the return value from the previous call.

reip.util.func.retry(func, n=None, exc=<class 'Exception'>, log=None, **kw)[source]: retry a function on failure.

Shell

class reip.util.shell.ShellResult(out, err, rc, cmd)

cmd: Alias for field number 3

err: Alias for field number 1

out: Alias for field number 0

rc: Alias for field number 2

reip.util.shell.run(cmd, *a, **kw)[source]

Run a shell command. Dictionary arguments passed will be converted to

bash flags. e.g.: dict(x=5, y=True, z=None) -> ‘-x 5 -y’

Arguments:

cmd (str): the command to run. By default, arguments will be quoted.

To pass a value without quoting, use the format pattern {!r}.

*args, **kwargs: arguments to format command with.

If any arg is None, it will insert an empty string. If any argument is a dict, it will attempt to format it as bash flags.

If the key is a single character, only a single preceding dash will be used.

if the value is True, then it will output like a boolean flag. e.g.: dict(asdf=True) => –asdf

if the value is None or False, then it will be omitted.

otherwise, it will be cast to a string.

Examples: >>> shell.run(‘echo 10’) # echo 10 # (‘10

‘, ‘’)

>>> shell.run('echo {}', '10 && echo 15')  # echo '10 && echo 15'
# ('10 && echo 15

‘, ‘’)

>>> shell.run('echo {!r}', '10 && echo 15')  # echo 10 && echo 15
# ('10

15 ‘, ‘’)

>>> shell.run('echo {} {b} {a}', 10, a=11, b=15)  # echo 10 15 11
# ('10 15 11

‘, ‘’)

>>> iface = None  # wlan0
... shell.run('ping {} 8.8.8.8', dict(I=iface, c=3))  # ping -c 3 8.8.8.8
# ('PING 8.8.8.8 ...', '')
# NOTE: notice how because -I is None, it gets filtered out.

class reip.util.shell.ShellArg(value)[source]

Formats a user-specified argument as a bash argument.

If value is a dict, it will be constructed as a series of flags. Otherwise it will just convert to string and potentially quote the value.

reip.util.shell.shmatch(cmd, out=None, err=None, rc=None)[source]

>>> shmatch('piwatcher status', 'OK')
>>> shmatch('ifconfig wlan0', 'inet [.\d]+')
>>> shmatch('docker logs blah --tail 100', err='Error')
>>> shmatch('ls')  # true
>>> shmatch('ls', 'README')  # true
>>> shmatch('ls 1>&2', 'README')  # false
>>> shmatch('ls 1>&2', err='README')  # true
>>> shmatch('true')  # true
>>> shmatch('true', rc=1)  # false
>>> shmatch('false')  # false
>>> shmatch('false', rc=1)  # true

reip.util.shell.git(*cmd, root=None)[source]: Run a git command in the sonycnode repository.

Misc

reip.util.misc.resize_list(lst, length, value=None)[source]

Resize a list to be a specific length.

Example: >>> x = [1, 2] >>> assert resize_list(x, 5) == [1, 2, None, None, None] >>> assert resize_list(x, 5, 10) == [1, 2, 10, 10, 10] >>> assert resize_list(x, 4, lambda: x[0]) == [1, 2, 1, 1] # some callable

reip.util.misc.decorator(__func__=None, **kw)[source]

A convenience wrapper that allows you to define a decorator with optional initialization parameters.

Example: >>> def my_decorator(func, **kw): … @functools.wraps(func) … def inner(*a, **kwi): … return func(*a, **dict(kw, **kw)) … return inner >>> @my_decorator … def asdf(x=10, y=15): … return x+y >>> @my_decorator(y=20) … def asdf2(x=10, y=15): … return x+y >>> assert asdf() == 25 and asdf2() == 30

reip.util.misc.ensure_dir(fname)[source]: Make sure that the directory that this filename is in exists. Does nothing if this file is in the current working directory.

reip.util.misc.fname(file)[source]: Get file name. e.g. path/to/fileA.txt => fileA

reip.util.misc.write(fname, *lines, mode='r')[source]: write to file.

reip.util.misc.as_list(x)[source]

Convert or wrap value as a list.

Examples: >>> assert as_list(5) == [5] >>> assert as_list(‘asdf’) == [‘asdf’] >>> assert as_list((1, 2)) == [1, 2] >>> assert as_list([1, 2]) == [1, 2]

reip.util.misc.squeeze(x)[source]

If the input is a one-element list or tuple, take the first element. (removes an unnecessary container.)

Examples: >>> assert squeeze([5]) == 5 >>> assert squeeze((1,)) == 1 >>> assert squeeze([1, 2]) == [1, 2] >>> assert squeeze(‘asdf’) == ‘asdf’ >>> assert squeeze((‘asdf’,)) == ‘asdf’ >>> assert squeeze((1, 2)) == (1, 2)

reip.util.misc.multicontext(*items)[source]: Use a variable set of context managers as one.

Debugging

reip.util.debug.check_block(block, match='', *a)[source]: Only print out only blocks that match the query text. For debugging specific subclasses.

reip.util.debug.block_stack(message=None, fn=None, offset=0, limit=None)[source]: Format the current stack trace.

reip.util.debug.print_stack(message=None, *a, offset=0, **kw)[source]: Print out the current stack trace.

reip.util.debug.short_stack(match=None, file=False, sep=' << ', n=None)[source]: Print out a compressed view of the stack.

Logging

class reip.util.logging.StrRep(obj, method=None, *a, **kw)[source]: Wrapping an object to provide an alternative string representation.

class reip.util.logging.MaxLevel(level)[source]

Filters (lets through) all messages with level < LEVEL

filter(record)[source]

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.

class reip.util.logging.InjectData(fields, *a, **kw)[source]

filter(record)[source]

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.