diff options
Diffstat (limited to 'lib/python2.7/site-packages/setoolsgui/networkx/utils/decorators.py')
| -rw-r--r-- | lib/python2.7/site-packages/setoolsgui/networkx/utils/decorators.py | 270 |
1 files changed, 270 insertions, 0 deletions
diff --git a/lib/python2.7/site-packages/setoolsgui/networkx/utils/decorators.py b/lib/python2.7/site-packages/setoolsgui/networkx/utils/decorators.py new file mode 100644 index 0000000..def1548 --- /dev/null +++ b/lib/python2.7/site-packages/setoolsgui/networkx/utils/decorators.py @@ -0,0 +1,270 @@ +import sys + +from collections import defaultdict +from os.path import splitext + +import networkx as nx +from networkx.external.decorator import decorator +from networkx.utils import is_string_like + +def not_implemented_for(*graph_types): + """Decorator to mark algorithms as not implemented + + Parameters + ---------- + graph_types : container of strings + Entries must be one of 'directed','undirected', 'multigraph', 'graph'. + + Returns + ------- + _require : function + The decorated function. + + Raises + ------ + NetworkXNotImplemnted + If any of the packages cannot be imported + + Notes + ----- + Multiple types are joined logically with "and". + For "or" use multiple @not_implemented_for() lines. + + Examples + -------- + Decorate functions like this:: + + @not_implemnted_for('directed') + def sp_function(): + pass + + @not_implemnted_for('directed','multigraph') + def sp_np_function(): + pass + """ + @decorator + def _not_implemented_for(f,*args,**kwargs): + graph = args[0] + terms= {'directed':graph.is_directed(), + 'undirected':not graph.is_directed(), + 'multigraph':graph.is_multigraph(), + 'graph':not graph.is_multigraph()} + match = True + try: + for t in graph_types: + match = match and terms[t] + except KeyError: + raise KeyError('use one or more of ', + 'directed, undirected, multigraph, graph') + if match: + raise nx.NetworkXNotImplemented('not implemented for %s type'% + ' '.join(graph_types)) + else: + return f(*args,**kwargs) + return _not_implemented_for + + +def require(*packages): + """Decorator to check whether specific packages can be imported. + + If a package cannot be imported, then NetworkXError is raised. + If all packages can be imported, then the original function is called. + + Parameters + ---------- + packages : container of strings + Container of module names that will be imported. + + Returns + ------- + _require : function + The decorated function. + + Raises + ------ + NetworkXError + If any of the packages cannot be imported + + Examples + -------- + Decorate functions like this:: + + @require('scipy') + def sp_function(): + import scipy + pass + + @require('numpy','scipy') + def sp_np_function(): + import numpy + import scipy + pass + """ + @decorator + def _require(f,*args,**kwargs): + for package in reversed(packages): + try: + __import__(package) + except: + msg = "{0} requires {1}" + raise nx.NetworkXError( msg.format(f.__name__, package) ) + return f(*args,**kwargs) + return _require + + +def _open_gz(path, mode): + import gzip + return gzip.open(path,mode=mode) + +def _open_bz2(path, mode): + import bz2 + return bz2.BZ2File(path,mode=mode) + +# To handle new extensions, define a function accepting a `path` and `mode`. +# Then add the extension to _dispatch_dict. +_dispatch_dict = defaultdict(lambda : open) +_dispatch_dict['.gz'] = _open_gz +_dispatch_dict['.bz2'] = _open_bz2 +_dispatch_dict['.gzip'] = _open_gz + + +def open_file(path_arg, mode='r'): + """Decorator to ensure clean opening and closing of files. + + Parameters + ---------- + path_arg : int + Location of the path argument in args. Even if the argument is a + named positional argument (with a default value), you must specify its + index as a positional argument. + mode : str + String for opening mode. + + Returns + ------- + _open_file : function + Function which cleanly executes the io. + + Examples + -------- + Decorate functions like this:: + + @open_file(0,'r') + def read_function(pathname): + pass + + @open_file(1,'w') + def write_function(G,pathname): + pass + + @open_file(1,'w') + def write_function(G, pathname='graph.dot') + pass + + @open_file('path', 'w+') + def another_function(arg, **kwargs): + path = kwargs['path'] + pass + """ + # Note that this decorator solves the problem when a path argument is + # specified as a string, but it does not handle the situation when the + # function wants to accept a default of None (and then handle it). + # Here is an example: + # + # @open_file('path') + # def some_function(arg1, arg2, path=None): + # if path is None: + # fobj = tempfile.NamedTemporaryFile(delete=False) + # close_fobj = True + # else: + # # `path` could have been a string or file object or something + # # similar. In any event, the decorator has given us a file object + # # and it will close it for us, if it should. + # fobj = path + # close_fobj = False + # + # try: + # fobj.write('blah') + # finally: + # if close_fobj: + # fobj.close() + # + # Normally, we'd want to use "with" to ensure that fobj gets closed. + # However, recall that the decorator will make `path` a file object for + # us, and using "with" would undesirably close that file object. Instead, + # you use a try block, as shown above. When we exit the function, fobj will + # be closed, if it should be, by the decorator. + + @decorator + def _open_file(func, *args, **kwargs): + + # Note that since we have used @decorator, *args, and **kwargs have + # already been resolved to match the function signature of func. This + # means default values have been propagated. For example, the function + # func(x, y, a=1, b=2, **kwargs) if called as func(0,1,b=5,c=10) would + # have args=(0,1,1,5) and kwargs={'c':10}. + + # First we parse the arguments of the decorator. The path_arg could + # be an positional argument or a keyword argument. Even if it is + try: + # path_arg is a required positional argument + # This works precisely because we are using @decorator + path = args[path_arg] + except TypeError: + # path_arg is a keyword argument. It is "required" in the sense + # that it must exist, according to the decorator specification, + # It can exist in `kwargs` by a developer specified default value + # or it could have been explicitly set by the user. + try: + path = kwargs[path_arg] + except KeyError: + # Could not find the keyword. Thus, no default was specified + # in the function signature and the user did not provide it. + msg = 'Missing required keyword argument: {0}' + raise nx.NetworkXError(msg.format(path_arg)) + else: + is_kwarg = True + except IndexError: + # A "required" argument was missing. This can only happen if + # the decorator of the function was incorrectly specified. + # So this probably is not a user error, but a developer error. + msg = "path_arg of open_file decorator is incorrect" + raise nx.NetworkXError(msg) + else: + is_kwarg = False + + # Now we have the path_arg. There are two types of input to consider: + # 1) string representing a path that should be opened + # 2) an already opened file object + if is_string_like(path): + ext = splitext(path)[1] + fobj = _dispatch_dict[ext](path, mode=mode) + close_fobj = True + elif hasattr(path, 'read'): + # path is already a file-like object + fobj = path + close_fobj = False + else: + # could be None, in which case the algorithm will deal with it + fobj = path + close_fobj = False + + # Insert file object into args or kwargs. + if is_kwarg: + new_args = args + kwargs[path_arg] = fobj + else: + # args is a tuple, so we must convert to list before modifying it. + new_args = list(args) + new_args[path_arg] = fobj + + # Finally, we call the original function, making sure to close the fobj. + try: + result = func(*new_args, **kwargs) + finally: + if close_fobj: + fobj.close() + + return result + + return _open_file |