vsi.tools package

Subpackages

Submodules

vsi.tools.commonpath module

vsi.tools.commonpath.commonpath(uris)[source]

vsi.tools.diff module

vsi.tools.diff.dict_diff(a, b)[source]

vsi.tools.dir_util module

class vsi.tools.dir_util.Chdir(dir, create=False, error_on_exit=False)[source]

Bases: object

Context to change dir and guarantee you get back to your last directory

Example:

These are written in doctest format, and should illustrate how to
use the function.

>>> a=[1,2,3]
>>> print [x + 3 for x in a]
[4, 5, 6]

>>> import os
>>> import tempfile
>>> from vsi.tools.dir_util import Chdir
>>> os.chdir(os.path.abspath(os.sep))
>>> print(os.getcwd())
/
>>> with Chdir(tempfile.tempdir):
...   print(os.getcwd())
/tmp
>>> print(os.getcwd())
/
__exit__(exc_type, exc_value, traceback)[source]

Exits if the previous directory no longer exists.

Parameters:

  • exc_type (str) – The Execption Type

  • exc_value (float) – The Exception Value

  • traceback (str) – The Traceback

error_on_exit

Create Chdir object

Parameters:

  • dir (str) – the directory you want to change directory to

  • create (str) – Optional: Create the directory if it doesn’t exist (else os error 2 will occur) Default: False

  • error_on_exit (bool) – When exiting the with loop, if the return directory does not exist anymore, (OSError 2), if this is true an error is thrown, else it is ignore and you are left in the new directory. Default: False

vsi.tools.dir_util.checksum_dir(checksum, checksum_depth=2, base_dir=None)[source]

Generate checksum directory name

Parameters:

  • checksum (float) – The Checksum

  • checksum_depth (int) – The Checksum Depth

  • base_dir (str) – The base Directory

Returns:

The Path

Return type:

str

Example

If the checksum was 1234567890abcdef, and the depth was 3, you would get 12/34/56/1234567890abcdef

Note

An optional base_dir will be prefixed

vsi.tools.dir_util.copytree(src, dst, symlinks=False, ignore=None)[source]

Recursively copy a directory tree using copy2().

Parameters:

  • src (path-like object) – The Source Tree

  • dst (str) – The Destination Directory may not already exist. If exception(s) occur, an Error is raised with a list of reasons. If the destination directory exists, it will be clobber by the source, including replacing entire directory trees with symlinks, if the symlinks flags is true.

  • symlinks (bool) – Optional. True if the symbolic links in the source tree result in symbolic links in the destination tree. False if the contents of the files pointed to by symbolic links are copied.

  • ignore (function) –

    The optional ignore argument is a callable. If given, it is called with the src parameter, which is the directory being visited by copytree(), and names which is the list of src contents, as returned by os.listdir():

    callable(src, names) -> ignored_names

    Since copytree() is called recursively, the callable will be called once for each directory that is copied. It returns a list of names relative to the src directory that should not be copied.

XXX Consider this example code rather than the ultimate tool.

vsi.tools.dir_util.find_file_in_path(fname, path=None)[source]

Find a file in any of the directories on the PATH. Like distutils.spawn.find_executable, but works for any file.

Parameters:

  • fname (str) – The filename to search the PATH for.

  • path (str) – An optional PATH string. If not provided, os PATH is used.

Returns:

The full path to the found file, or None if not found.

Return type:

str

vsi.tools.dir_util.is_dir_empty(path)[source]

Checks to see if a dir is empty. Much faster than os.listdir and os.walk

vsi.tools.dir_util.is_subdir(path, base_dir, dereference_symlinks=True)[source]

Determines if the path is in the base_dir

Parameters:

  • path (str) – The Path

  • base_dir (str) – The Base Directory

Returns:

  • tuple – containing (True/False, and remainder (relative) of path)

  • str – Remainder, the relative different between the two paths. If on Windows and the paths are on two different drives, the entire path is returned

Note

This will NOT work with Junctions in Windows.

vsi.tools.dir_util.mkdtemp(*args, **kwargs)[source]

Version of tempfile.mkdtemp that is r/w by uid and gid

Parameters:

  • *args – Variable length argument list.

  • **kwargs – Arbitrary keyword arguments.

Returns:

The Temp Directory

Return type:

str

vsi.tools.dir_util.prune_dir(directory, top_dir=None)[source]

Remove directory and ancestor directories if they are empty

Parameters:

  • directory (str) – The Directory

  • top_dir (str) – Optional argument top_dir, prevents pruning below that directory

Raises:

OSError –

vsi.tools.dir_util.root_dir(directory)[source]

OS independent way of getting the root directory of a directory

Parameters:

directory (str) – The Directory

Returns:

The Root Directory

Return type:

str

On Windows:

C:\tmp\blah.txt -> C:\
D:\Program Files\calc.exe -> D:\

On Linux/Darwin:

/tmp/blah.txt -> /
vsi.tools.dir_util.samefile(path1, path2, normpath=True)[source]

OS independent version of os.path.samefile

Parameters:

  • path1 (str) – The File Name

  • path2 (str) – The File Name

  • normpath (bool) – Optional normpath=True. In posix cases when you want symlinks to be followed instead of normalized out, this would be useful to set to false.

Returns:

OS independent version of the path

Return type:

str

Example:

>>> os.symlink('/usr/bin', '/tmp/blah')
>>> samefile('/usr', '/tmp/blah/..')
False
>>> samefile('/usr', '/tmp/blah/..', normpath=False)
True

vsi.tools.docker_token module

vsi.tools.docker_token.get_parser()[source]
vsi.tools.docker_token.main(args=None)[source]

vsi.tools.file_util module

vsi.tools.file_util._path_or_temp(kwarg_key, kwarg_func=<function <lambda>>)[source]

Decorator to replace a missing kwarg (missing, empty, or None) that defines a path with a valid temporary path.

vsi.tools.file_util.file_or_temp(kwarg_key, filename)[source]

Decorator to replace a missing kwarg (missing, empty, or None) that defines a file with a filename in a valid temporary directory.

The temporary directory intended to contain the temporary file will be created on entering the decorated function, and the directory & its contents will be deleted on exiting the decorated function.

The decorated function is expected to create the file itself.

vsi.tools.file_util.lncp(source, dest)[source]

Symlink or copy if that fails. Should work for Linux and Windows

Parameters:

  • source (str) – The Source

  • dest (str) – The Destination

vsi.tools.iter module

vsi.tools.iter.sub_block(data, block=3, overlap=0, subok=False)[source]

Return an array of windows into the original data array.

Parameters:

  • data (numpy.ndarray) – N-dimensional array

  • block (int) – The block size of the final blocks. Should have length N or be a single number

  • overlap (float) – The amount of overlap between windows. A value of 0 has no overlap, while positive values overlap by that much, and negative values have gaps of that size.

  • subok (bool) – Passed to as_strided

Returns:

  • numpy.ndarray – windows - N + N dimensional array where the first N dimensions are the window indexes, and the next N dimension are the data dimensions. Keep in mind, all the data is linked to the original data array. Changing values in one place changes them all.

  • list – remainders - An N length list of the number of elements not included in the windows: the remainder of the sub-blocks. Another function could use these values to sub_block over them too. In the 2D case, there would be 3 groups of windows, in 3D case 6 groups of windows, etc…

Example:

>> qq = np.arange(25).reshape(5,5)
>> q,r = sub_block(qq, block=3, overlap=(2,1))
>> q

array([[[[ 0,  1,  2],
        [ 5,  6,  7],
        [10, 11, 12]],

        [[ 2,  3,  4],
        [ 7,  8,  9],
        [12, 13, 14]]],


        [[[ 5,  6,  7],
        [10, 11, 12],
        [15, 16, 17]],

        [[ 7,  8,  9],
        [12, 13, 14],
        [17, 18, 19]]],


        [[[10, 11, 12],
        [15, 16, 17],
        [20, 21, 22]],

        [[12, 13, 14],
        [17, 18, 19],
        [22, 23, 24]]]])

REVIEW would be nice if this were a generator…

vsi.tools.logging_helper module

vsi.tools.logging_helper.show_log(k, v)[source]
vsi.tools.logging_helper.show_logs_and_handlers()[source]

vsi.tools.mpl module

class vsi.tools.mpl.SimpleBubblePicker(fig=None, text_function=None, x_offset=10, y_offset=15, offset_units='dots', bbox={'alpha': 0.5, 'boxstyle': 'round', 'facecolor': 'wheat'}, **kwargs)[source]

Bases: object

Simple class to add a picker with your own text function

Currently supports plots (Line2D), scatter (PathCollection), images (AxesImage) and PatchCollection

SimpleBubblePicker comes with a default text function that will display basic information about any point you click.

cleanup(event)[source]
default_text(event)[source]
get_bubble(axes, shift=False)[source]
get_bubbles(axes)[source]
new_bubble(axes)[source]
offset_units

The units of the offset for the current bubble offset

Type:

str

x_offset

The x offset of the currently drawing bubble

Type:

float

y_offset

The y offset of the currently drawing bubble

Type:

float

vsi.tools.mpl.auto_fit_fontsize(text, width=None, height=None, step_size=1, min_font_size=1, max_font_size=100)[source]

Auto-fit the fontsize of a text object.

Args:

text (matplotlib.text.Text) width (float): allowed width in data coordinates height (float): allowed height in data coordinates

vsi.tools.mpl.imshow_chip(axes, img, origin, size, *args, **kwargs)[source]

imshow’s only a chip of an image, and draws the chip at the appropriate coordinates.

Does not support custom transforms

Parameters:
vsi.tools.mpl.imshow_chip_from_raster(axes, raster, origin, size, *args, **kwargs)[source]

Loads and imshow’s a chip of an image, and draws the chip at the appropriate coordinates.

Does not support custom transforms

Parameters:
vsi.tools.mpl.share_xy_custom(ax1, ax2, forward=<function <lambda>>, backward=<function <lambda>>, sharex=True, sharey=True, **kwargs)[source]

Like Axes.sharex/Axes.sharey only it allows you to provide your own function instead of the 1:1 default.

Parameters:

  • ax1 (matplotlib.axes.Axes) – The first Axes

  • ax2 (matplotlib.axes.Axes) – The second Axes

  • forward (Callable) – A function that takes the arguments x0, x1, y0, y1 for ax1 and returns (x0, x1, y0, y1) for ax2. Return None to indicate that no update to ax2 should be performed.

  • backward (Callable) – Same as forward except maps ax2 to ax1

  • sharex (bool, optional) – Should the x axis be shared

  • sharey (bool, optional) – Should the y axis be shared

  • **kwargs – Additional keyword arguments passed to forward and backward

vsi.tools.mpl.surf(z, cmap='jet', ax=None, x=None, y=None, c=None, **kwargs)[source]

Creates an equivalent “surf” plot, like in matlab

Parameters:

By default, mpl_toolkits.mplot3d.axes3d.Axes3D.plot_surface() does not draw the entire mesh, it downsamples it to 50 points instead (for efficiency). To disable downsampling, consider setting rstride and cstride to 1.

Shading is also disabled by default

vsi.tools.mpl_zoom module

vsi.tools.mpl_zoom.figure_with_zoom(*args, **kwargs)[source]
Parameters:

  • *args – Variable length argument list.

  • **kwargs – Arbitrary keyword arguments.

Returns:

The Zoomed Figure

Return type:

matplotlib.figure

vsi.tools.mpl_zoom.zoom_factory(ax=None, base_scale=1.5)[source]
Parameters:
Returns:

The Zoom Function

Return type:

function

vsi.tools.natural_sort module

vsi.tools.natural_sort.natural_sorted(iterable, key=None, *args, **kwargs)[source]

Return sorted list of strings according to (sub)string numerical value.

Parameters:

  • iterable (iterable) – The data to be sorted.

  • key (function) – The sorting function.

  • *args – Variable length argument list.

  • **kwargs – Arbitrary keyword arguments.

Returns:

A sorted list of strings according to (sub)string numerical value.

Return type:

list

>>> natural_sorted(['f10', 'f2', 'f1'])
['f1', 'f2', 'f10']

>>> natural_sorted([('f10',2), ('f2',5), ('f1',1), ('f1a',3)], key=lambda x: x[0])
[('f1', 1), ('f1a', 3), ('f2', 5), ('f10', 2)]

vsi.tools.patch_wheel module

vsi.tools.patch_wheel.get_parser()[source]
vsi.tools.patch_wheel.main(filename, name=None, version=None)[source]

vsi.tools.python module

class vsi.tools.python.ARGS[source]

Bases: object

class vsi.tools.python.ArgvContext(*args)[source]

Bases: object

Context to temporarily change the sys.argv variable

Parameters:

*args (str) – Arguments to replace sys.argv, starting with argv[0]

class vsi.tools.python.KWARGS[source]

Bases: object

class vsi.tools.python.OptionalArgumentDecorator(*args, **kwargs)[source]

Bases: object

Decorator for easily defining a decorator class that may take arguments

Write a decorator class as normal, that would always take arguments, and make sure they all have default values. Then decorate that decorator with this decorator and both @decorator and @decorator() notations will work.

See also

BasicDecorator

Simple decorator with optional arguments

class vsi.tools.python.Try(default_ignore=<class 'Exception'>, *other_ignore)[source]

Bases: object

Try catch helper for cases when you want to ignore certain exceptions

Parameters:

*ignore_exceptions (Exception) – Exception classes to be ignored. Default is all.

class vsi.tools.python._BasicArgumentDecorator[source]

Bases: object

A basic decorator class that takes arguments

It’s best to define __init__ with a proper signature when inheriting

__call__(fun)[source]

No need to rewrite this

Parameters:

fun (function) – The Function

__inner_call__(*args, **kwargs)[source]

re-write THIS. No need for super().__inner_call__ :param *args: Variable length argument list. :param **kwargs: Arbitrary keyword arguments.

class vsi.tools.python._BasicDecorator(*args, **kwargs)[source]

Bases: object

A basic decorator class that does not take arguments

Parameters:

fun (function) – The function that gets wrapped

Variables:

fun (function) – The function that is being wrapped

Examples

Usage:

class MyDecorAdd1(_BasicDecorator):
  def __call__(self, *args, **kwargs):
    result = self.fun(*args, **kwargs)
    res

@MyDecorAdd1
def fun(a, b=2):
  return a+b
__call__(*args, **kwargs)[source]

Re-write this. Do need to call super().__call__

The general idea of this class is you re-write the __call__ method to do what you want, and call self.fun and return the result. This can be accomplished by return super().__call__(*args, **kwargs), but more often then not, you will want the result of self.fun, and will call result = self.fun(*args, **kwargs) yourself, and then return result

fun = None
vsi.tools.python._meta_generate_class(cls, *args, **kwargs)[source]

Determine class to use for decorators

Parameters:

  • cls (class) – The class of the decorator type

  • args (tuple) – The arguments passed to the decorate, when decorating a class

  • kwargs (dict) – The keyword arguments passed to the decorate, when decorating a class

Helper function to parse the arguments from a class’s __new__ or __init__ to handle both the normal case, and when the class is being inherited by another class:

class A:
  def __new__(cls, *args):
    return super(A, cls).__new__(_meta_generate_class(A, *args))
@A
class B():
  pass
class C(B):
  pass

When B is decorated by A, A.__new__/__init__ is called with 1 argument, B. When C inherits from B, A.__new__/__init__ is called with 3 arguments instead of one, the 3 arguments to a type() call.

This helper will run all that logic for you, and just always return the class you need.

vsi.tools.python.args_to_kwargs(function, args=(), kwargs={})[source]

returns a single dict of all the args and kwargs

Should handle: functions, classes (their __init__), bound and unbound versions of methods, class methods, and static methods. Furthermore, if a class instance has a __call__ method, this is used.

It does not call the function.

Parameters:
Returns:

  • dict – The returned dictionary has the keywords that would be received in a real function call. Leftover args are put into the key ARGS, and leftover KWARGS are placed in the key KWARGS. While everything should behave exactly as python would, certain failure situations are not reproduced, for exampled it does not raise exception if you declare the same parameter in both //args and /*/kwargs)

  • On python3, args_to_kwargs_unbound must be used for unbound class

  • methods

  • Based on

  • https (//github.com/merriam/dectools/blob/master/dectools/dectools.py)

vsi.tools.python.args_to_kwargs_easy(*args, **kwargs)[source]
Parameters:

  • *args (tuple) – Variable length argument list.

  • **kwargs (dict) – Arbitrary keyword arguments.

Return type:

dict

vsi.tools.python.args_to_kwargs_unbound(function, attribute=None, args=(), kwargs={})[source]
vsi.tools.python.args_to_kwargs_unbound_easy(*args, **kwargs)[source]
Parameters:

  • *args (tuple) – Variable length argument list.

  • **kwargs (dict) – Arbitrary keyword arguments.

Return type:

dict

vsi.tools.python.command_list_to_string(cmd)[source]
Parameters:

cmd (list) – The Command List

Returns:

The Command List as a string

Return type:

str

vsi.tools.python.get_file(fid, mode='rb')[source]

Helper function to take either a filename or fid

Parameters:

  • fid (str) – File object or filename

  • mode (str, optional) – Optional, file mode to open file if filename supplied Default is ‘rb’

Returns:

The opened file object

Return type:

file-like object

vsi.tools.python.is_class_method(cls, attribute)[source]
vsi.tools.python.is_static_method(cls, attribute)[source]

Returns whether the attribute refers to a staticmethod or not

Parameters:

  • cls (object) – The class/instance being checked

  • attribute (str) – The name of the function to be checked

Returns:

True if the attribute is a static function, else false if anything else

Return type:

bool

vsi.tools.python.is_string_like(obj)[source]

Check whether obj behaves like a string.

Copied from numpy

Parameters:

obj (object) – Object being tested for string like behavior

Returns:

True if object behaves like a string. False otherwise.

Return type:

bool

vsi.tools.python.nested_in_dict(dict1, dict2)[source]

Checks to see if dict1 is in dict2

Parameters:

  • dict1 (dict) – Subset dictionary

  • dict2 (dict) – Superset dictionary

vsi.tools.python.nested_patch(obj, condition, patch, _spare_key=None)[source]

Patch strings in a nested python dict

Will patch values in mapping and iterable containers recursively. This includes (but is not limited to) set, list, dict, tuple, etc… Only iterates through values, not keys.

When the condition is met for a given key,value pair, then the patch function is used to replace the value.

Parameters:

  • obj (mapping or iterable or object) – The python object to be patched. Typically a dict, but can be a list, etc… or even a normal object, but that kind of defeats the purpose

  • condition (function) – The condition function to decide if each value should be patched. condition takes two arguments, (key, value)

  • patch (function) – Callable that should return a patched version of the value. patch takes two arguments, (key, value)

Returns:

Returns a patched version of the object. This should not be though of as a deep-copy of the original object, as unpatched values will still be the same python objects, not copies.

Return type:

object

Example

patterns = ['_file', '_dir', '_path',
      '_files', '_dirs', '_paths']
condition = lambda key, value: isinstance(key, str) and \
            any(key.endswith(pattern) for pattern in patterns)

def patch_value(value, volume_map):
  for vol_from, vol_to in volume_map:
    if isinstance(value, str) and value.startswith(vol_from):
      return value.replace(vol_from, vol_to, 1)
  return value

volume_map = [['/tmp', '/temp'],
              ['/tmp/home', '/nope'],
              ['/home', '/Home']]
patch = lambda key, value: patch_value(value, reversed(volume_map))

z = {'test': '/tmp',
    'foo_file': '/tmp',
    'foo': 15,
    17: 'bar',
    'foo_dir': ['/tmp', '/home'],
    'foo_files': 15,
    'stuff': {
      'this_path': '/home',
      'a': {
        'b': {
          'c': {
            'e': [{'b_path': '/home'}, {'c_dir': '/tmp'}],
            'd': {
              'q_path': (('/home', '/opt'), ('/tmp', '/tmp/home/foo/bar')),
              'q_orig': (('/home', '/opt'), ('/tmp', '/tmp/home/foo/bar')),
              'a_path': ('/home', '/opt', '/tmp', '/tmp/home/foo/bar')
            }
          }
        }
      }
    }
    }

z2 = nested_patch(z, condition, patch)
vsi.tools.python.nested_patch_inplace(obj, condition, patch, _spare_key=None)[source]

Destructive inplace version of vsi.tools.python.nested_patch()

vsi.tools.python.nested_update(dict_, *args, **kwargs)[source]

Updated a dictionary in a nested fashion

Parameters:

  • dict (dict) – The dict to be updated

  • *args (tuple) – Same arguments as dict.update

  • **kwargs (dict) – Same arguments as dict.update

vsi.tools.python.reloadModules(pattern='.*', skipPattern='^IPython')[source]

Reload modules that match pattern regular expression (string or re)

Parameters:

  • pattern (str or re.Pattern) – The regular expression pattern of modules that will be reloaded.

  • skipPattern (str or re.Pattern) – The regular expression pattern of modules that will not be reloaded.

vsi.tools.python.static(**kwargs)[source]

Decorator for easily defining static variables

Parameters:

**kwargs – Arbitrary keyword arguments.

Example:

@static(count=0)
def test(a, b):
  test.count += 1
  print(a+b, test.count)
vsi.tools.python.unwrap_wraps(func)[source]

Unwraps a wrapped function

Finds the originally wrapped function, using the functools.wraps() pattern of storing in __wrapped__

vsi.tools.python.update_wrapper_class(wrapper, wrapped)[source]

functools.update_wrapper for classes

Version of functools.update_wrapper that works when the wrapper is a class

Parameters:

  • wrapper (class) – The class to be updated

  • wrapped (class) – The original function/class

Returns:

A subclass of wrapper that has the updated attributes. If wrapped was a function, wrapper is still a class.

Return type:

class

vsi.tools.redirect module

class vsi.tools.redirect.Capture(stdout_c=1, stderr_c=2, stdout_py=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, stderr_py=<_io.TextIOWrapper name='<stderr>' mode='w' encoding='utf-8'>, group=True, group_outerr=True, group_out=True, group_err=True)[source]

Bases: RedirectBase

A class to easily redirect stdout and stderr to a sting buffer

There are times in python when using “those kind of libraries” where you have to capture either stdout or stderr, and many situations are tricky. This class will help you, by using a with statement.

Example:

with Redirect() as rid:
  some_library.call157()

print(rid.stdout)

There are 4 output pipe of concern, none of which are synced with each other (flushed together).

stdout and stderr from c (referred to as stdout_c, stderr_c) and stdout and stderr from python (referred to as stdout_py, stderr_py).

Most python functions use the python stdout/stderr, controlled by sys.stdout and sys.stderr. However c embedded calls are used the c stdout/stderr, usually fd 1 and 2. Because of this, there are 4 streams to consider

Once Redirect is done, the results are stored in

>>> self.stdout_c - Standard out output for c
>>> self.stderr_c - Standard error output for c
>>> self.stdout_py - Standard out output for python
>>> self.stderr_py - Standard error output for python

When the streams are grouped, they both contain the same data. In the group_out and group_err case, an addition attribute is defined

>>> self.stdout - Standard out output
>>> self.stderr - Standard error output

Just to make it easier

If you want a more complicated scenario, say you want stdout_c and stderr_py grouped, and stderr_c and stdout_py groups because you’ve been drinking too much. Well, this is easy to do by calling Redirect twice:

Example:

with Redirect(stdout_c=None, stderr_py=None) as rid1:
  with Redirect(stderr_c=None, stdout_py=None) as rid2:
    stuff()

Now rid1’s buffer has just stderr_c and stdout_py as one group stream and rid2 has stfout_c and stderr_py as one grouped stream

__enter__()[source]

enter function for with statement.

Switched out stderr and stdout, and starts pipe threads

__exit__(exc_type=None, exc_value=None, traceback=None)[source]

exit function for with statement.

Restores stdout and sterr, closes write pipe and joins with the threads

Parameters:

  • exc_type (str) – The Execption Type

  • exc_value (float) – The Exception Value

  • traceback (str) – The Traceback

group_err

Initialize the Redirect object

Parameters:

  • stdout_c (int, optional) – The fd to be replace, usually 1 will work, but change it in case this is not right in your case (default: 1) None means to not redirect

  • stderr_c (int, optional) – The fd to be replaced, usually 2 will work, but change it in case this is not right in your case (default: 2) None means to not redirect

  • stdout_py (file-like object, optional) – The file object to be replaced (default: sys.stdout) None means to not redirect

  • stderr_py (file-like object, optional) – The file object to be replaced (default: sys.stderr) None means to not redirect

  • group (bool, optional) – Should ANY of the stream be joined together. This overrides ALL of the following group options

  • group_outerr (bool, optional) – Should stdout and stderr use the a group stream or else it will have separate streams (default: True)

  • group_out (bool, optional) – Should stdout_c and stdout_py use the a group stream or else it will have separate streams (default: True)

  • group_err (bool, optional) – Should stderr_c and stderr_py use the a group stream or else it will have separate streams (default: True)

startMonitor(readPipe, bufferIndex)[source]

Start a read pipe monitoring thread

Runs __bleed in a background thread to capture all the redirected output and stores the information in the self.buffers[bufferIndex].

Appends the thread object to self.tids

Parameters:

  • readPipe (str) – File descriptor number of the read pipe (from by os.pipe)

  • bufferIndex (int) – The xth buffer to store the string in.

class vsi.tools.redirect.FileRedirect(outputs=[])[source]

Bases: object

Redirect a real file object to any object with .write

Some function need a REAL file object with file number and all. This class wraps a pair of pipes per output and so what everytime the pipe is written to, output.write is called.

Only supports one way communication. You can write to the file object via any valid call, and the other end of the pipe performs “readlines” and calls output.write() It would be possible to expand this to two way, but not currently needed.

Primarily used in a with expression:

>>> from StringIO import StringIO
>>> s = StringIO()
>>> with FileRedirect([s]) as f:
>>>   f.wids[0].write('hiya')
>>> s.seek(0); print(s.read())

IF you call __enter__ manually, you only need to close the wids file objects, and the rest of the threads and file objects are cleaned up, of course it would be better to call __exit__ in that case.

__enter__()[source]

Create the pipes and monitoring threads

__exit__(exc_type=None, exc_value=None, traceback=None)[source]

Closes the write ends of the pipes, and join with monitoring threads

The read pipes are automatically closed by the monitors

Parameters:

  • exc_type (str) – The Exception Type

  • exc_value (float) – The Exception Value

  • traceback (str) – The Traceback

outputs

Create a FileRedirect

Parameters:

outputs (list) – list of outputs objects to write to. For every output in outputs, a rid in self.rids and wid in self.wids is created when call with with

Yields:

list – list of outputs objects.

startMonitor(stream_index)[source]

Start a new thread to monitor the stream

Should only be called once per stream_index

Parameters:

stream_index (int) – The Stream Index

class vsi.tools.redirect.Logger(logger, lvl=20)[source]

Bases: object

File object wrapper for logger.

Careful when using this, if the logging out goes to a stream that is redireted, you have an infinite capture loop and does not go well

Use PopenRedirect instead of Redirect in that case

write(str)[source]

Write method

Parameters:

str (str) – The String

class vsi.tools.redirect.PopenRedirect(stdout=<vsi.tools.redirect.Foo object>, stderr=<vsi.tools.redirect.Bar object>)[source]

Bases: FileRedirect

Version FileRedirect object geared towards redirecting for Popen commands

>>> from StringIO import StringIO
>>> from subprocess import Popen
>>> out = StringIO(); err = StringIO()
>>> with PopenRedirect(out, err) as f:
>>>   Popen(['whoami'], stdout=f.stdout, stderr=f.stderr).wait()
>>>   Popen(['whoami', 'non user'], stdout=f.stdout, stderr=f.stderr).wait()
>>> out.seek(0); err.seek(0)
>>> print(out.read(), err.read())
property stderr
Returns:

Second object in wireless intrusion prevention system list.

Return type:

str

property stdout
Returns:

First object in wireless intrusion prevention system list.

Return type:

str

class vsi.tools.redirect.Redirect(all=None, stdout=None, stderr=None, c=None, py=None, stdout_c=None, stderr_c=None, stdout_py=None, stderr_py=None, stdout_c_fd=1, stderr_c_fd=2, stdout_py_module=<module 'sys' (built-in)>, stderr_py_module=<module 'sys' (built-in)>, stdout_py_name='stdout', stderr_py_name='stderr')[source]

Bases: RedirectBase

FAILED EXPERIMENT! You can use this, but I don’t recommend it for anything that matters. There are too many situations where buffered output or multiple stdouts in windows cause this to not behave 100% right. Best to just not use it. PopenRedirect is good

MORE FAILURES! It looks like when extending python with C, too many cout/flushes can result in a deadlock, I’m guess the GIL does something weird

Similar to Capture class, except it redirect to file like objects

There are times in python when using “those kind of libraries” where you have to capture either stdout or stderr, and many situations are tricky. This class will help you, by using a with statement.

Example:

from StringIO import StringIO
stdout = StringIO()
with Redirect(stdout=stdout):
  some_library.call157()

stdout.seek(0,0)
print(stdout.read())

There are 4 output pipe of concern, none of which are synced with each other (flushed together).

stdout and stderr from c (referred to as stdout_c, stderr_c) and stdout and stderr from python (referred to as stdout_py, stderr_py).

Most python functions use the python stdout/stderr, controlled by sys.stdout and sys.stderr. However c embedded calls are used the c stdout/stderr, usually fd 1 and 2. Because of this, there are 4 streams to consider

Known bugs: In windows, after stdout_c is done being redirected, stdout buffering appears to no longer be disabled in interactive mode. What this means is typing

>>> print(123)

No longer returns 123 right away, without a

>>> sys.stdout.flush()

Why? I’m guessing os.dup2 breaks the buffered mode. The fix is to

>>> sys.stdout = os.fdopen(sys.__stdout__.fileno(), 'w', 0)

However this can only be done once or twice, and then starts failing.

Work around is to redirect stdout when calling the command. For example ‘python | find /v “”’ Mostly works.

Basically don’t do this in Windows Command line

__enter__()[source]

enter function for with statement.

Switched out stderr and stdout, and starts pipe threads

__exit__(exc_type=None, exc_value=None, traceback=None)[source]

exit function for with statement.

Restores stdout and stderr, closes write pipe and joins with the threads

Parameters:

  • exc_type (str) – The Exception Type

  • exc_value (float) – The Exception Value

  • traceback (str) – The Traceback

startMonitor(readPipe, output)[source]
Parameters:

  • readPipe (int) – File descriptor number of the read pipe (from by os.pipe)

  • output (int) – The xth buffer to store the string in.

Yields:

array_like – Appends the thread object to self.tids

stderr_py_out

Create the Redirect object

Parameters:

  • file_like (file-like object, optional) – File Output Argument. All File output arguments should use File like Python objects that have a .write call. Many of the arguments override the other argument for ease of use

  • all (str, optional) – Output stdout_c, stderr_c, stdout_py and stderr_py to the all file.

  • stdout (str, optional) – Output stdout_c and stdout_py to the stdout file.

  • stderr (str, optional) – Output stderr_c and stderr_py to the stderr file.

  • c (str, optional) – Output stdout_c and stderr_c to the c file.

  • py (str, optional) – Output stdout_py and stderr_py to the py file.

  • stdout_c (str, optional) –

  • stderr_c (str, optional) –

  • stdout_py (file-like object, optional) –

  • stderr_py (file-like object, optional) – Output to each individual stream for maximum customization.

  • stdout_c_fd (int, optional) –

  • stderr_c_fd (int, optional) – The default file number used for stdout (1) and stderr (2). There should be no reason to override this

  • stdout_py_module (module, optional) –

  • stderr_py_module (module, optional) –

  • stdout_py_name (str, optional) –

  • stderr_py_name (str, optional) – Because of the nature of python, in order to replace and restore the python object, the module and name of attribute must be passed through, where name is a string and module and the actual module. The default is sys module and “stdout” or “stderr” (Including the quotes). Again, there should be no real reason to override these, unless you are doing some IPython/colorama redirecting, or any other library that messes with sys.stdout/sys.stderr

class vsi.tools.redirect.RedirectBase[source]

Bases: object

flush()[source]

Flushes stdout and stderr

class vsi.tools.redirect.StdRedirect(stdout=None, stderr=None)[source]

Bases: object

Redirect stdout and stderr to a file object (must have a working fileno)

This is very stable and safe to use. The only caveat to remember is that C can have unflushed data in the buffer, and it will not be captured after the StdRedirect with statement. This is not vulnerable to the deadlock that the GIL can introduce with the pipe methods that need extra threads.

Primarily used in a with expression:

>>> test = open('test.txt', 'w')
>>> with StdRedirect(test):
...   boxm2_batch.print_db()

Supports stdout and stderr, and stderr can be set to StdRedirect. STDOUT to use the same output as stdout

STDOUT = -1
__exit__(exc_type=None, exc_value=None, traceback=None)[source]
Parameters:

  • exc_type (str) – The Execption Type

  • exc_value (float) – The Exception Value

  • traceback (str) – The Traceback

new_stderr
Parameters:

  • stdout (str) – Standard Output

  • stderr (str) – Standard Error

vsi.tools.stdout_profile module

vsi.tools.stdout_profile.argumet_parser()[source]
Returns:

The Parser

Return type:

str

vsi.tools.stdout_profile.main(args=['-M', 'html', '.', '_build', '-W', '--keep-going', '-n', '-E', '-a'])[source]
Parameters:

args (list) – An array of arguments

vsi.tools.subprocess_util module

class vsi.tools.subprocess_util.PopenBg(*args, **kwargs)[source]

Bases: Popen

vsi.tools.time_utils module

class vsi.tools.time_utils.GenericTimer(msg, logging_func=<built-in function print>)[source]

Bases: object

class vsi.tools.time_utils.TaskTimer(task_name, num_tasks=1, logging_func=<built-in function print>)[source]

Bases: object

vsi.tools.time_utils.format_time_string(seconds)[source]

Format a time string the way I want it. You would think I wouldn’t have to write this myself!

Parameters:

seconds (float) –

Returns:

E.g. “1 hour, 42 minutes, and 7 seconds”

Return type:

str

vsi.tools.time_utils.timeThisFunc(msg=None)[source]

Generic decorator to time function execution

vsi.tools.vdb module

class vsi.tools.vdb.DbStopIfErrorGeneric(threading_support=False, *args, **kwargs)[source]

Bases: object

With statement for local dbstop situations

__exit__(exc_type, exc_value, tb)[source]
Parameters:

  • exc_type (str) – The Execption Type

  • exc_value (float) – The Exception Value

  • tb (str) – The Traceback

get_post_mortem()[source]
Returns:

Should return a function that takes a traceback as the first argument and any additional args/kwargs sent to __init__ after that

Return type:

function

Raises:

Exception – For a virtual function

ignore_exception = False
kwargs
Parameters:

  • threading_support (bool) – Optional. Support the threading module and patch a bug preventing catching exceptions in other threads. See add_threading_excepthook for more info. Only neccesary if you want to catch exceptions not on the main thread. This is only patched after __enter__ unpatched at __exit__

  • *args – Variable length argument list.

  • **kwargs – Arbitrary keyword arguments.

All other args from db_stop_if_error()

classmethod set_continue_exception()[source]

Continue running code after exception

Parameters:

cls (bool) – True to continue to run code after the exception. Default: False.

After the with statement scope fails, if this is called, python will continue running as if there was no error. Can be useful, can also be dangerous. So don’t abuse it!

class vsi.tools.vdb.PostMortemHook[source]

Bases: object

static dbclear_if_error()[source]
classmethod dbstop_if_error(interactive=False, *args, **kwargs)[source]
Parameters:

  • interactive (bool) – True if interactive. False if not.

  • *args – Variable length argument list.

  • **kwargs – Arbitrary keyword arguments.

original_excepthook = None
static set_post_mortem(interactive=False)[source]

Overrite this function for each debugger

Parameters:

interactive (bool) – True if interactive. False if not.

Raises:

Exception – Makes users aware that this is purely a virtual function

class vsi.tools.vdb.RunningTrace[source]

Bases: object

Mixin to give any bdb a running settrace

Example:

from pdb import Pdb as OldPdb
from vsi.tools.vdb import RunningTrace
class Pdb(OldPdb, RunningTrace):
  pass

p = Pdb()
p.set_running_trace()
p.onecmd('b my_function')
classmethod get_db(debugger_cls=<class 'pdb.Pdb'>)[source]

Helper function to mixin and instantiate in simple cases

Parameters:

debugger_cls (class) – Bdb based Debugger class to be mixed in

Returns:

An instance of the debugger class with set_runnable_trace added

Return type:

object

set_running_trace(frame=None)[source]

Start debugging from frame, but don’t enter interactive mode

If frame is not specified, debugging starts from caller’s frame.

vsi.tools.vdb.add_threading_excepthook()[source]

Workaround for sys.excepthook thread bug From http://spyced.blogspot.com/2007/06/workaround-for-sysexcepthook-bug.html

(https://sourceforge.net/tracker/?func=detail&atid=105470&aid=1230540&group_id=5470). Call once from __main__ before creating any threads. If using psyco, call psyco.cannotcompile(threading.Thread.run) since this replaces a new-style class method.

Raises:
vsi.tools.vdb.attach(pid, signal=Signals.SIGUSR1)[source]

Trigger a python pid that’s been already run set_attach

Parameters:

  • pid (str) – The Process ID

  • signal (int) – The Attach Signal

This is the second part of attaching to a python process. Once set_attach is run, on another prompt running attach will trigger the interrupt thing attaching or triggering whatever db_cmd was

vsi.tools.vdb.break_pool_worker()[source]

Setup the ThreadPool to break when an exception occurs (so that it can be debugged)

The ThreadPool class (and the Pool class too, but not useful here) always catches any exception and raises it in the main thread. This is nice for normal behavior, but for debugging, it makes it impossible to do post mortem debugging. In order to automatically attach a post mortem debugger, the exception has to be thrown. An exception being thrown WILL BREAK the pool call, and not allow your main function to continue, however you can now attach a debugger post_mortem. Useful with dbstop_if_error

Threading has a “bug” where exceptions are also automatically caught. http://bugs.python.org/issue1230540 In order to make THIS work, call add_threading_excepthook too

Example:

>>> from multiprocessing.pool import ThreadPool
>>> import vsi.tools.vdb as vdb
>>> def a(b):
...   print(b)
...   if b==3:
...     does_not_exist()
>>> vdb.dbstop_if_error()
>>> vdb.break_pool_worker()
>>> vdb.add_threading_excepthook()
>>> tp = ThreadPool(3)
>>> tp.map(a, range(10))
vsi.tools.vdb.dbstop_exception_hook(type, value, tb, post_mortem, interactive=False)[source]
Parameters:

  • type (class) – sys.excepthook variable

  • value (object) – sys.excepthook variable

  • tb (object) – sys.excepthook variable

  • post_mortem (function) – The Post Mortem Handler Function

  • Interactive (bool) – True if interactive. False if not.

vsi.tools.vdb.find_frame(frame, depth=0)[source]
Parameters:

  • frame (str) – The Frame

  • depth (int) – The Depth

vsi.tools.vdb.handle_db(sig, frame, db_cmd=None, signal=Signals.SIGUSR1)[source]

signal handler part of attach/set_attach

Parameters:

  • sig –

  • frame –

  • db_cmd (str) – The Debugger Command

  • signal (int) – The Attach Signal

vsi.tools.vdb.main()[source]
vsi.tools.vdb.pipe_server()[source]

Part of attach/set_attach for Windows

vsi.tools.vdb.set_attach(db_cmd=None, signal=Signals.SIGUSR1)[source]

Set up this process to be “debugger attachable”

Parameters:

  • db_cmd (str) – The Debugger Command

  • signal (int) – The Attach Signal

Just like gdb can attach to a running process, if you execute this on a process, now you can “attach” to the running python using the attach command

This works pretty well, and allows you to resume the code UNLESS you are running in windows and happen to interrupt a sleep command.

vsi.tools.vdb_ipdb module

class vsi.tools.vdb_ipdb.DbStopIfError(threading_support=False, *args, **kwargs)[source]

Bases: DbStopIfErrorGeneric

With statement for local dbstop situations

get_post_mortem()[source]
Returns:

Should return a function that takes a traceback as the first argument and any additional args/kwargs sent to __init__ after that

Return type:

function

Raises:

Exception – For a virtual function

get_post_mortem_class()[source]

Get post mortem class for Vdb

class vsi.tools.vdb_ipdb.Vdb(*args: Any, **kwargs: Any)[source]

Bases: TerminalPdb

VSI Debugger, based off of IPython’s 5.x or newer debugger

user_return(frame, return_value)[source]
class vsi.tools.vdb_ipdb.VdbPostMortemHook[source]

Bases: PostMortemHook

static set_post_mortem(interactive=False)[source]

Overrite this function for each debugger

Parameters:

interactive (bool) – True if interactive. False if not.

Raises:

Exception – Makes users aware that this is purely a virtual function

vsi.tools.vdb_ipdb.dbclear_if_error()[source]
vsi.tools.vdb_ipdb.dbstop_if_error(interactive=False)[source]

Run this to auto start the vdb debugger on an exception.

Parameters:

interactive (bool, optional) – Default False. dbstop if console is interactive. You are still able to print and run commands in the debugger, just listing code declared interactively will not work. Does not appear to work in ipython. Use %debug instead. This will not help in the multithread case in ipython… ipython does too much, just don’t try that. Unless someone adds a way to override ipython’s override.

vsi.tools.vdb_ipdb.post_mortem(tb=None)[source]

Helper function, like pdb.post_mortem

vsi.tools.vdb_ipdb.runpdb(lines, debugger=None)[source]

Executes a list of vdb command

Parameters:

lines (str or list or tuple) – Collection of strings to be executed as if you were already in the debugger. Useful for setting breakpoints programatically.

Returns:

Returns the debugger object, since this can only be executed on the debugger object, you can optionally pass it in as the second argument if you want to call runpdb multiple times. If you do not, a new debugger object is created, and all the “memory” of the last debugger is lost, such as breakpoints, etc…

Return type:

object

vsi.tools.vdb_ipdb.set_trace(frame=None, depth=None)[source]

Helper function, like pdb.set_trace

vsi.tools.vdb_rpdb module

class vsi.tools.vdb_rpdb.DbStopIfError(threading_support=False, *args, **kwargs)[source]

Bases: DbStopIfErrorGeneric

get_post_mortem()[source]
Returns:

Should return a function that takes a traceback as the first argument and any additional args/kwargs sent to __init__ after that

Return type:

function

Raises:

Exception – For a virtual function

get_post_mortem_class()[source]
class vsi.tools.vdb_rpdb.RpdbPostMortemHook[source]

Bases: PostMortemHook

static set_post_mortem(interactive=False, ip='127.0.0.1', port=4444)[source]

Overrite this function for each debugger

Parameters:

interactive (bool) – True if interactive. False if not.

Raises:

Exception – Makes users aware that this is purely a virtual function

vsi.tools.vdb_rpdb.attach(pid, ip='127.0.0.1', port=4444)[source]

NOT IMPLEMENTED! Needs a telnet client

Parameters:

  • pid (str) – The Process ID

  • ip (str) – The IP Address

  • port (str) – The Port

vsi.tools.vdb_rpdb.dbclear_if_error()[source]
vsi.tools.vdb_rpdb.dbstop_if_error(interactive=False, ip='127.0.0.1', port=4444)[source]

Run this to auto start the debugger on an exception.

Optional arguments

Parameters:

  • interactive (bool) – see vsi.tools.vdb.dbstop_if_error

  • ip (str) – Default 127.0.0.1 - Ip to bind to for remote debugger

  • port (int) – Default 4444 - Port to bind to for remote debugger

vsi.tools.vdb_rpdb.post_mortem(tb=None, ip='127.0.0.1', port=4444)[source]
Parameters:

  • tb (str) – The Traceback

  • ip (str) – The IP Address

  • port (int) – The Port

Raises:

ValueError – Passes a valied traceback

vsi.tools.vdb_rpdb.set_attach(ip='127.0.0.1', port=4444)[source]
Parameters:

  • ip (str) – The IP Address

  • port (str) – The Port

vsi.tools.vdb_rpdb.set_trace(frame=None, depth=None, ip='127.0.0.1', port=4444)[source]

Wrapper function to keep the same import x; x.set_trace() interface.

We catch all the possible exceptions from pdb and cleanup.

Parameters:

  • frame (str) – The Frame

  • depth (int) – The Depth

  • ip (str) – The IP Address

  • port (int) – The Port

vsi.tools.vdb_rpdb2 module

class vsi.tools.vdb_rpdb2.CDebuggerCoreThread2(*args: Any, **kwargs: Any)[source]

Bases: CDebuggerCoreThread

I just wanted to add some output on exception!!!

profile(frame, event, arg)[source]
vsi.tools.vdb_rpdb2.attach(pid, ip='127.0.0.1', password='vsi', gui=False, break_exit=False)[source]
Parameters:

  • pid (int) – The Process ID

  • ip (str) – The IP Address

  • password (str) – The Password

  • gui (bool) – If true use winpbd GUI

  • break_exit (bool) – If break_exit is true, it set’s a breakpoint when the program extis.

vsi.tools.vdb_rpdb2.dbstop_if_error(_rpdb2_pwd='vsi', fAllowUnencrypted=True, fAllowRemote=False, timeout=0, source_provider=None, fDebug=False, depth=0)[source]

Run this to auto start the debugger on an exception.

Parameters:

  • _rpdb2_pwd (str) – The Remote Python Debugger Password

  • fAllowUnencrypted (bool) – True if unencrypted is allowed. False if not.

  • fAllowRemote (bool) – True if a remote is allowed. False if not. Default: False

  • timeout (int) – The timeout period in seconds.

  • source_provider (str) – The Source Provider

  • fDebug (bool) – The Debug Output

  • depth (int) – The Depth of the frame. Default: 0

I THINK rpdb2 does not use the same traceback or the frame/stack objects as pdb. So there is no way to just hand the debugger a traceback (yet). So by starting the debugger, all exceptions will be cause, and just attach to the debugger and type “analyze” to start debugging the last exception.

Of course, this means that if the debugger slows down execution, it will have to slow down all of the execution, instead of being loaded “just in time”

vsi.tools.vdb_rpdb2.set_attach(_rpdb2_pwd='vsi', *args, **kwargs)[source]
Parameters:

  • _rpdb2_pwd (str) – The Remote Python Debugger Password

  • *args – Variable length argument list.

  • **kwargs – Arbitrary keyword arguments.

vsi.tools.vdb_rpdb2.set_trace(_rpdb2_pwd='vsi', fAllowUnencrypted=True, fAllowRemote=False, timeout=300, source_provider=None, fDebug=False, depth=1)[source]

Works, but without the other parts, it’s far from auto

Parameters:

  • _rpdb2_pwd (str) – The Remote Python Debugger Password

  • fAllowUnencrypted (bool) – True if unencrypted is allowed. False if not.

  • fAllowRemote (bool) – True if a remote is allowed. False if not. Default: False

  • timeout (int) – The timeout period in seconds.

  • source_provider (str) – The Source Provider

  • fDebug (bool) – The Debug Output

  • depth (int) – The Depth of the frame. Default: 0

vsi.tools.watch_dog module

class vsi.tools.watch_dog.WatchDog(timeout=20)[source]

Bases: object

reset(signum, frame)[source]
Parameters:

  • signum (int) – The Signal Number

  • frame (frame) – The Frame

start(poll_interval=1)[source]
Parameters:

poll_interval (int) – The Poll Interval

stop(signum, frame)[source]
Parameters:

  • signum (int) – The Signal Number

  • frame (frame) – The Frame

timeout
Parameters:

timeout (int) – The Timeout

Raises:

Exception – When there is no usual signal found.

Module contents