from collections import defaultdict
from os.path import splitext
from contextlib import contextmanager
from pathlib import Path
import networkx as nx
from networkx.utils import create_random_state, create_py_random_state
import inspect, itertools, collections
import re, gzip, bz2
__all__ = [
"not_implemented_for",
"open_file",
"nodes_or_number",
"preserve_random_state",
"random_state",
"np_random_state",
"py_random_state",
"argmap",
]
[docs]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", or "graph".
Returns
-------
_require : function
The decorated function.
Raises
------
NetworkXNotImplemented
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_implemented_for("directed")
def sp_function(G):
pass
# rule out MultiDiGraph
@not_implemented_for("directed","multigraph")
def sp_np_function(G):
pass
# rule out all except DiGraph
@not_implemented_for("undirected")
@not_implemented_for("multigraph")
def sp_np_function(G):
pass
"""
if ("directed" in graph_types) and ("undirected" in graph_types):
raise ValueError("Function not implemented on directed AND undirected graphs?")
if ("multigraph" in graph_types) and ("graph" in graph_types):
raise ValueError("Function not implemented on graph AND multigraphs?")
if not set(graph_types) < {"directed", "undirected", "multigraph", "graph"}:
raise KeyError(
"use one or more of directed, undirected, multigraph, graph. "
f"You used {graph_types}"
)
# 3-way logic: True if "directed" input, False if "undirected" input, else None
dval = ("directed" in graph_types) or not ("undirected" in graph_types) and None
mval = ("multigraph" in graph_types) or not ("graph" in graph_types) and None
errmsg = f"not implemented for {' '.join(graph_types)} type"
def _not_implemented_for(g):
if (mval is None or mval == g.is_multigraph()) and (
dval is None or dval == g.is_directed()
):
raise nx.NetworkXNotImplemented(errmsg)
return g
return argmap(_not_implemented_for, 0)
# To handle new extensions, define a function accepting a `path` and `mode`.
# Then add the extension to _dispatch_dict.
fopeners = {
".gz": gzip.open,
".gzip": gzip.open,
".bz2": bz2.BZ2File,
}
_dispatch_dict = defaultdict(lambda: open, **fopeners) # type: ignore
[docs]def open_file(path_arg, mode="r"):
"""Decorator to ensure clean opening and closing of files.
Parameters
----------
path_arg : string or int
Name or index of the argument that is a path.
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("pathname","w")
def write_function(G, pathname="graph.dot"):
pass
@open_file("path", "w+")
def another_function(arg, **kwargs):
path = kwargs["path"]
pass
Notes
-----
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 of how to handle this case::
@open_file("path")
def some_function(arg1, arg2, path=None):
if path is None:
fobj = tempfile.NamedTemporaryFile(delete=False)
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
try:
fobj.write("blah")
finally:
if path is None:
fobj.close()
Normally, we'd want to use "with" to ensure that fobj gets closed.
However, the decorator will make `path` a file object for us,
and using "with" would undesirably close that file object.
Instead, we use a try block, as shown above.
When we exit the function, fobj will be closed, if it should be, by the decorator.
"""
def _open_file(path):
# 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 isinstance(path, str):
ext = splitext(path)[1]
elif isinstance(path, Path):
# path is a pathlib reference to a filename
ext = path.suffix
path = str(path)
else:
# could be None, or a file handle, in which case the algorithm will deal with it
return path, lambda: None
fobj = _dispatch_dict[ext](path, mode=mode)
return fobj, lambda: fobj.close()
return argmap(_open_file, path_arg, try_finally=True)
[docs]def nodes_or_number(which_args):
"""Decorator to allow number of nodes or container of nodes.
With this decorator, the specified argument can be either a number or a container
of nodes. If it is a number, the nodes used are `range(n)`.
This allows `nx.complete_graph(50)` in place of `nx.complete_graph(list(range(50)))`.
And it also allows `nx.complete_graph(any_list_of_nodes)`.
Parameters
----------
which_args : string or int or sequence of strings or ints
If string, the name of the argument to be treated.
If int, the index of the argument to be treated.
If more than one node argument is allowed, can be a list of locations.
Returns
-------
_nodes_or_numbers : function
Function which replaces int args with ranges.
Examples
--------
Decorate functions like this::
@nodes_or_number("nodes")
def empty_graph(nodes):
# nodes is converted to a list of nodes
@nodes_or_number(0)
def empty_graph(nodes):
# nodes is converted to a list of nodes
@nodes_or_number(["m1", "m2"])
def grid_2d_graph(m1, m2, periodic=False):
# m1 and m2 are each converted to a list of nodes
@nodes_or_number([0, 1])
def grid_2d_graph(m1, m2, periodic=False):
# m1 and m2 are each converted to a list of nodes
@nodes_or_number(1)
def full_rary_tree(r, n)
# presumably r is a number. It is not handled by this decorator.
# n is converted to a list of nodes
"""
def _nodes_or_number(n):
try:
nodes = list(range(n))
except TypeError:
nodes = tuple(n)
else:
if n < 0:
msg = "Negative number of nodes not valid: {n}"
raise nx.NetworkXError(msg)
return (n, nodes)
try:
iter_wa = iter(which_args)
except TypeError:
iter_wa = (which_args,)
return argmap(_nodes_or_number, *iter_wa)
def preserve_random_state(func):
"""Decorator to preserve the numpy.random state during a function.
.. deprecated:: 2.6
This is deprecated and will be removed in NetworkX v3.0.
Parameters
----------
func : function
function around which to preserve the random state.
Returns
-------
wrapper : function
Function which wraps the input function by saving the state before
calling the function and restoring the function afterward.
Examples
--------
Decorate functions like this::
@preserve_random_state
def do_random_stuff(x, y):
return x + y * numpy.random.random()
Notes
-----
If numpy.random is not importable, the state is not saved or restored.
"""
import warnings
msg = "preserve_random_state is deprecated and will be removed in 3.0."
warnings.warn(msg, DeprecationWarning)
try:
import numpy as np
@contextmanager
def save_random_state():
state = np.random.get_state()
try:
yield
finally:
np.random.set_state(state)
def wrapper(*args, **kwargs):
with save_random_state():
np.random.seed(1234567890)
return func(*args, **kwargs)
wrapper.__name__ = func.__name__
return wrapper
except ImportError:
return func
[docs]def np_random_state(random_state_argument):
"""Decorator to generate a `numpy.random.RandomState` instance.
The decorator processes the argument indicated by `random_state_argument`
using :func:`nx.utils.create_random_state`.
The argument value can be a seed (integer), or a `numpy.random.RandomState`
instance or (`None` or `numpy.random`). The latter options use the glocal
random number generator used by `numpy.random`.
The result is a `numpy.random.RandomState` instance.
Parameters
----------
random_state_argument : string or int
The name or index of the argument to be converted
to a `numpy.random.RandomState` instance.
Returns
-------
_random_state : function
Function whose random_state keyword argument is a RandomState instance.
Examples
--------
Decorate functions like this::
@np_random_state("seed")
def random_float(seed=None):
return seed.rand()
@np_random_state(0)
def random_float(rng=None):
return rng.rand()
@np_random_state(1)
def random_array(dims, random_state=1):
return random_state.rand(*dims)
See Also
--------
py_random_state
"""
return argmap(create_random_state, random_state_argument)
def random_state(random_state_argument):
"""Decorator to generate a `numpy.random.RandomState` instance.
.. deprecated:: 2.7
This function is a deprecated alias for `np_random_state` and will be
removed in version 3.0. Use np_random_state instead.
"""
import warnings
warnings.warn(
(
"`random_state` is a deprecated alias for `np_random_state`\n"
"and will be removed in version 3.0. Use `np_random_state` instead."
),
DeprecationWarning,
stacklevel=2,
)
return np_random_state(random_state_argument)
[docs]def py_random_state(random_state_argument):
"""Decorator to generate a random.Random instance (or equiv).
The decorator processes the argument indicated by `random_state_argument`
using :func:`nx.utils.create_py_random_state`.
The argument value can be a seed (integer), or a random number generator::
If int, return a random.Random instance set with seed=int.
If random.Random instance, return it.
If None or the `random` package, return the global random number
generator used by `random`.
If np.random package, return the global numpy random number
generator wrapped in a PythonRandomInterface class.
If np.random.RandomState instance, return it wrapped in
PythonRandomInterface
If a PythonRandomInterface instance, return it
Parameters
----------
random_state_argument : string or int
The name of the argument or the index of the argument in args that is
to be converted to the random.Random instance or numpy.random.RandomState
instance that mimics basic methods of random.Random.
Returns
-------
_random_state : function
Function whose random_state_argument is converted to a Random instance.
Examples
--------
Decorate functions like this::
@py_random_state("random_state")
def random_float(random_state=None):
return random_state.rand()
@py_random_state(0)
def random_float(rng=None):
return rng.rand()
@py_random_state(1)
def random_array(dims, seed=12345):
return seed.rand(*dims)
See Also
--------
np_random_state
"""
return argmap(create_py_random_state, random_state_argument)
[docs]class argmap:
"""A decorator to apply a map to arguments before calling the function
This class provides a decorator that maps (transforms) arguments of the function
before the function is called. Thus for example, we have similar code
in many functions to determine whether an argument is the number of nodes
to be created, or a list of nodes to be handled. The decorator provides
the code to accept either -- transforming the indicated argument into a
list of nodes before the actual function is called.
This decorator class allows us to process single or multiple arguments.
The arguments to be processed can be specified by string, naming the argument,
or by index, specifying the item in the args list.
Parameters
----------
func : callable
The function to apply to arguments
*args : iterable of (int, str or tuple)
A list of parameters, specified either as strings (their names), ints
(numerical indices) or tuples, which may contain ints, strings, and
(recursively) tuples. Each indicates which parameters the decorator
should map. Tuples indicate that the map function takes (and returns)
multiple parameters in the same order and nested structure as indicated
here.
try_finally : bool (default: False)
When True, wrap the function call in a try-finally block with code
for the finally block created by `func`. This is used when the map
function constructs an object (like a file handle) that requires
post-processing (like closing).
Examples
--------
Most of these examples use `@argmap(...)` to apply the decorator to
the function defined on the next line.
In the NetworkX codebase however, `argmap` is used within a function to
construct a decorator. That is, the decorator defines a mapping function
and then uses `argmap` to build and return a decorated function.
A simple example is a decorator that specifies which currency to report money.
The decorator (named `convert_to`) would be used like::
@convert_to("US_Dollars", "income")
def show_me_the_money(name, income):
print(f"{name} : {income}")
And the code to create the decorator might be::
def convert_to(currency, which_arg):
def _convert(amount):
if amount.currency != currency:
amount = amount.to_currency(currency)
return amount
return argmap(_convert, which_arg)
Despite this common idiom for argmap, most of the following examples
use the `@argmap(...)` idiom to save space.
Here's an example use of argmap to sum the elements of two of the functions
arguments. The decorated function::
@argmap(sum, "xlist", "zlist")
def foo(xlist, y, zlist):
return xlist - y + zlist
is syntactic sugar for::
def foo(xlist, y, zlist):
x = sum(xlist)
z = sum(zlist)
return x - y + z
and is equivalent to (using argument indexes)::
@argmap(sum, "xlist", 2)
def foo(xlist, y, zlist):
return xlist - y + zlist
or::
@argmap(sum, "zlist", 0)
def foo(xlist, y, zlist):
return xlist - y + zlist
Transforming functions can be applied to multiple arguments, such as::
def swap(x, y):
return y, x
# the 2-tuple tells argmap that the map `swap` has 2 inputs/outputs.
@argmap(swap, ("a", "b")):
def foo(a, b, c):
return a / b * c
is equivalent to::
def foo(a, b, c):
a, b = swap(a, b)
return a / b * c
More generally, the applied arguments can be nested tuples of strings or ints.
The syntax `@argmap(some_func, ("a", ("b", "c")))` would expect `some_func` to
accept 2 inputs with the second expected to be a 2-tuple. It should then return
2 outputs with the second a 2-tuple. The returns values would replace input "a"
"b" and "c" respectively. Similarly for `@argmap(some_func, (0, ("b", 2)))`.
Also, note that an index larger than the number of named parameters is allowed
for variadic functions. For example::
def double(a):
return 2 * a
@argmap(double, 3)
def overflow(a, *args):
return a, args
print(overflow(1, 2, 3, 4, 5, 6)) # output is 1, (2, 3, 8, 5, 6)
**Try Finally**
Additionally, this `argmap` class can be used to create a decorator that
initiates a try...finally block. The decorator must be written to return
both the transformed argument and a closing function.
This feature was included to enable the `open_file` decorator which might
need to close the file or not depending on whether it had to open that file.
This feature uses the keyword-only `try_finally` argument to `@argmap`.
For example this map opens a file and then makes sure it is closed::
def open_file(fn):
f = open(fn)
return f, lambda: f.close()
The decorator applies that to the function `foo`::
@argmap(open_file, "file", try_finally=True)
def foo(file):
print(file.read())
is syntactic sugar for::
def foo(file):
file, close_file = open_file(file)
try:
print(file.read())
finally:
close_file()
and is equivalent to (using indexes)::
@argmap(open_file, 0, try_finally=True)
def foo(file):
print(file.read())
Here's an example of the try_finally feature used to create a decorator::
def my_closing_decorator(which_arg):
def _opener(path):
if path is None:
path = open(path)
fclose = path.close
else:
# assume `path` handles the closing
fclose = lambda: None
return path, fclose
return argmap(_opener, which_arg, try_finally=True)
which can then be used as::
@my_closing_decorator("file")
def fancy_reader(file=None):
# this code doesn't need to worry about closing the file
print(file.read())
Notes
-----
An object of this class is callable and intended to be used when
defining a decorator. Generally, a decorator takes a function as input
and constructs a function as output. Specifically, an `argmap` object
returns the input function decorated/wrapped so that specified arguments
are mapped (transformed) to new values before the decorated function is called.
As an overview, the argmap object returns a new function with all the
dunder values of the original function (like `__doc__`, `__name__`, etc).
Code for this decorated function is built based on the original function's
signature. It starts by mapping the input arguments to potentially new
values. Then it calls the decorated function with these new values in place
of the indicated arguments that have been mapped. The return value of the
original function is then returned. This new function is the function that
is actually called by the user.
Three additional features are provided.
1) The code is lazily compiled. That is, the new function is returned
as an object without the code compiled, but with all information
needed so it can be compiled upon it's first invocation. This saves
time on import at the cost of additional time on the first call of
the function. Subsequent calls are then just as fast as normal.
2) If the "try_finally" keyword-only argument is True, a try block
follows each mapped argument, matched on the other side of the wrapped
call, by a finally block closing that mapping. We expect func to return
a 2-tuple: the mapped value and a function to be called in the finally
clause. This feature was included so the `open_file` decorator could
provide a file handle to the decorated function and close the file handle
after the function call. It even keeps track of whether to close the file
handle or not based on whether it had to open the file or the input was
already open. So, the decorated function does not need to include any
code to open or close files.
3) The maps applied can process multiple arguments. For example,
you could swap two arguments using a mapping, or transform
them to their sum and their difference. This was included to allow
a decorator in the `quality.py` module that checks that an input
`partition` is a valid partition of the nodes of the input graph `G`.
In this example, the map has inputs `(G, partition)`. After checking
for a valid partition, the map either raises an exception or leaves
the inputs unchanged. Thus many functions that make this check can
use the decorator rather than copy the checking code into each function.
More complicated nested argument structures are described below.
The remaining notes describe the code structure and methods for this
class in broad terms to aid in understanding how to use it.
Instantiating an `argmap` object simply stores the mapping function and
the input identifiers of which arguments to map. The resulting decorator
is ready to use this map to decorate any function. Calling that object
(`argmap.__call__`, but usually done via `@my_decorator`) a lazily
compiled thin wrapper of the decorated function is constructed,
wrapped with the necessary function dunder attributes like `__doc__`
and `__name__`. That thinly wrapped function is returned as the
decorated function. When that decorated function is called, the thin
wrapper of code calls `argmap._lazy_compile` which compiles the decorated
function (using `argmap.compile`) and replaces the code of the thin
wrapper with the newly compiled code. This saves the compilation step
every import of networkx, at the cost of compiling upon the first call
to the decorated function.
When the decorated function is compiled, the code is recursively assembled
using the `argmap.assemble` method. The recursive nature is needed in
case of nested decorators. The result of the assembly is a number of
useful objects.
sig : the function signature of the original decorated function as
constructed by :func:`argmap.signature`. This is constructed
using `inspect.signature` but enhanced with attribute
strings `sig_def` and `sig_call`, and other information
specific to mapping arguments of this function.
This information is used to construct a string of code defining
the new decorated function.
wrapped_name : a unique internally used name constructed by argmap
for the decorated function.
functions : a dict of the functions used inside the code of this
decorated function, to be used as `globals` in `exec`.
This dict is recursively updated to allow for nested decorating.
mapblock : code (as a list of strings) to map the incoming argument
values to their mapped values.
finallys : code (as a list of strings) to provide the possibly nested
set of finally clauses if needed.
mutable_args : a bool indicating whether the `sig.args` tuple should be
converted to a list so mutation can occur.
After this recursive assembly process, the `argmap.compile` method
constructs code (as strings) to convert the tuple `sig.args` to a list
if needed. It joins the defining code with appropriate indents and
compiles the result. Finally, this code is evaluated and the original
wrapper's implementation is replaced with the compiled version (see
`argmap._lazy_compile` for more details).
Other `argmap` methods include `_name` and `_count` which allow internally
generated names to be unique within a python session.
The methods `_flatten` and `_indent` process the nested lists of strings
into properly indented python code ready to be compiled.
More complicated nested tuples of arguments also allowed though
usually not used. For the simple 2 argument case, the argmap
input ("a", "b") implies the mapping function will take 2 arguments
and return a 2-tuple of mapped values. A more complicated example
with argmap input `("a", ("b", "c"))` requires the mapping function
take 2 inputs, with the second being a 2-tuple. It then must output
the 3 mapped values in the same nested structure `(newa, (newb, newc))`.
This level of generality is not often needed, but was convenient
to implement when handling the multiple arguments.
See Also
--------
not_implemented_for
open_file
nodes_or_number
random_state
py_random_state
networkx.community.quality.require_partition
require_partition
"""
[docs] def __init__(self, func, *args, try_finally=False):
self._func = func
self._args = args
self._finally = try_finally
@staticmethod
def _lazy_compile(func):
"""Compile the source of a wrapped function
Assemble and compile the decorated function, and intrusively replace its
code with the compiled version's. The thinly wrapped function becomes
the decorated function.
Parameters
----------
func : callable
A function returned by argmap.__call__ which is in the process
of being called for the first time.
Returns
-------
func : callable
The same function, with a new __code__ object.
Notes
-----
It was observed in NetworkX issue #4732 [1] that the import time of
NetworkX was significantly bloated by the use of decorators: over half
of the import time was being spent decorating functions. This was
somewhat improved by a change made to the `decorator` library, at the
cost of a relatively heavy-weight call to `inspect.Signature.bind`
for each call to the decorated function.
The workaround we arrived at is to do minimal work at the time of
decoration. When the decorated function is called for the first time,
we compile a function with the same function signature as the wrapped
function. The resulting decorated function is faster than one made by
the `decorator` library, so that the overhead of the first call is
'paid off' after a small number of calls.
References
----------
[1] https://github.com/networkx/networkx/issues/4732
"""
real_func = func.__argmap__.compile(func.__wrapped__)
func.__code__ = real_func.__code__
func.__globals__.update(real_func.__globals__)
func.__dict__.update(real_func.__dict__)
return func
def __call__(self, f):
"""Construct a lazily decorated wrapper of f.
The decorated function will be compiled when it is called for the first time,
and it will replace its own __code__ object so subsequent calls are fast.
Parameters
----------
f : callable
A function to be decorated.
Returns
-------
func : callable
The decorated function.
See Also
--------
argmap._lazy_compile
"""
if inspect.isgeneratorfunction(f):
def func(*args, __wrapper=None, **kwargs):
yield from argmap._lazy_compile(__wrapper)(*args, **kwargs)
else:
def func(*args, __wrapper=None, **kwargs):
return argmap._lazy_compile(__wrapper)(*args, **kwargs)
# standard function-wrapping stuff
func.__name__ = f.__name__
func.__doc__ = f.__doc__
func.__defaults__ = f.__defaults__
func.__kwdefaults__.update(f.__kwdefaults__ or {})
func.__module__ = f.__module__
func.__qualname__ = f.__qualname__
func.__dict__.update(f.__dict__)
func.__wrapped__ = f
# now that we've wrapped f, we may have picked up some __dict__ or
# __kwdefaults__ items that were set by a previous argmap. Thus, we set
# these values after those update() calls.
# If we attempt to access func from within itself, that happens through
# a closure -- which trips an error when we replace func.__code__. The
# standard workaround for functions which can't see themselves is to use
# a Y-combinator, as we do here.
func.__kwdefaults__["_argmap__wrapper"] = func
# this self-reference is here because functools.wraps preserves
# everything in __dict__, and we don't want to mistake a non-argmap
# wrapper for an argmap wrapper
func.__self__ = func
# this is used to variously call self.assemble and self.compile
func.__argmap__ = self
return func
__count = 0
@classmethod
def _count(cls):
"""Maintain a globally-unique identifier for function names and "file" names
Note that this counter is a class method reporting a class variable
so the count is unique within a Python session. It could differ from
session to session for a specific decorator depending on the order
that the decorators are created. But that doesn't disrupt `argmap`.
This is used in two places: to construct unique variable names
in the `_name` method and to construct unique fictitious filenames
in the `_compile` method.
Returns
-------
count : int
An integer unique to this Python session (simply counts from zero)
"""
cls.__count += 1
return cls.__count
_bad_chars = re.compile("[^a-zA-Z0-9_]")
@classmethod
def _name(cls, f):
"""Mangle the name of a function to be unique but somewhat human-readable
The names are unique within a Python session and set using `_count`.
Parameters
----------
f : str or object
Returns
-------
name : str
The mangled version of `f.__name__` (if `f.__name__` exists) or `f`
"""
f = f.__name__ if hasattr(f, "__name__") else f
fname = re.sub(cls._bad_chars, "_", f)
return f"argmap_{fname}_{cls._count()}"
[docs] def compile(self, f):
"""Compile the decorated function.
Called once for a given decorated function -- collects the code from all
argmap decorators in the stack, and compiles the decorated function.
Much of the work done here uses the `assemble` method to allow recursive
treatment of multiple argmap decorators on a single decorated function.
That flattens the argmap decorators, collects the source code to construct
a single decorated function, then compiles/executes/returns that function.
The source code for the decorated function is stored as an attribute
`_code` on the function object itself.
Note that Python's `compile` function requires a filename, but this
code is constructed without a file, so a fictitious filename is used
to describe where the function comes from. The name is something like:
"argmap compilation 4".
Parameters
----------
f : callable
The function to be decorated
Returns
-------
func : callable
The decorated file
"""
sig, wrapped_name, functions, mapblock, finallys, mutable_args = self.assemble(
f
)
call = f"{sig.call_sig.format(wrapped_name)}#"
mut_args = f"{sig.args} = list({sig.args})" if mutable_args else ""
body = argmap._indent(sig.def_sig, mut_args, mapblock, call, finallys)
code = "\n".join(body)
locl = {}
globl = dict(functions.values())
filename = f"{self.__class__} compilation {self._count()}"
compiled = compile(code, filename, "exec")
exec(compiled, globl, locl)
func = locl[sig.name]
func._code = code
return func
[docs] def assemble(self, f):
"""Collects components of the source for the decorated function wrapping f.
If `f` has multiple argmap decorators, we recursively assemble the stack of
decorators into a single flattened function.
This method is part of the `compile` method's process yet separated
from that method to allow recursive processing. The outputs are
strings, dictionaries and lists that collect needed info to
flatten any nested argmap-decoration.
Parameters
----------
f : callable
The function to be decorated. If f is argmapped, we assemble it.
Returns
-------
sig : argmap.Signature
The function signature as an `argmap.Signature` object.
wrapped_name : str
The mangled name used to represent the wrapped function in the code
being assembled.
functions : dict
A dictionary mapping id(g) -> (mangled_name(g), g) for functions g
referred to in the code being assembled. These need to be present
in the ``globals`` scope of ``exec`` when defining the decorated
function.
mapblock : list of lists and/or strings
Code that implements mapping of parameters including any try blocks
if needed. This code will precede the decorated function call.
finallys : list of lists and/or strings
Code that implements the finally blocks to post-process the
arguments (usually close any files if needed) after the
decorated function is called.
mutable_args : bool
True if the decorator needs to modify positional arguments
via their indices. The compile method then turns the argument
tuple into a list so that the arguments can be modified.
"""
# first, we check if f is already argmapped -- if that's the case,
# build up the function recursively.
# > mapblock is generally a list of function calls of the sort
# arg = func(arg)
# in addition to some try-blocks if needed.
# > finallys is a recursive list of finally blocks of the sort
# finally:
# close_func_1()
# finally:
# close_func_2()
# > functions is a dict of functions used in the scope of our decorated
# function. It will be used to construct globals used in compilation.
# We make functions[id(f)] = name_of_f, f to ensure that a given
# function is stored and named exactly once even if called by
# nested decorators.
if hasattr(f, "__argmap__") and f.__self__ is f:
(
sig,
wrapped_name,
functions,
mapblock,
finallys,
mutable_args,
) = f.__argmap__.assemble(f.__wrapped__)
functions = dict(functions) # shallow-copy just in case
else:
sig = self.signature(f)
wrapped_name = self._name(f)
mapblock, finallys = [], []
functions = {id(f): (wrapped_name, f)}
mutable_args = False
if id(self._func) in functions:
fname, _ = functions[id(self._func)]
else:
fname, _ = functions[id(self._func)] = self._name(self._func), self._func
# this is a bit complicated -- we can call functions with a variety of
# nested arguments, so long as their input and output are tuples with
# the same nested structure. e.g. ("a", "b") maps arguments a and b.
# A more complicated nesting like (0, (3, 4)) maps arguments 0, 3, 4
# expecting the mapping to output new values in the same nested shape.
# The ability to argmap multiple arguments was necessary for
# the decorator `nx.algorithms.community.quality.require_partition`, and
# while we're not taking full advantage of the ability to handle
# multiply-nested tuples, it was convenient to implement this in
# generality because the recursive call to `get_name` is necessary in
# any case.
applied = set()
def get_name(arg, first=True):
nonlocal mutable_args
if isinstance(arg, tuple):
name = ", ".join(get_name(x, False) for x in arg)
return name if first else f"({name})"
if arg in applied:
raise nx.NetworkXError(f"argument {arg} is specified multiple times")
applied.add(arg)
if arg in sig.names:
return sig.names[arg]
elif isinstance(arg, str):
if sig.kwargs is None:
raise nx.NetworkXError(
f"name {arg} is not a named parameter and this function doesn't have kwargs"
)
return f"{sig.kwargs}[{arg!r}]"
else:
if sig.args is None:
raise nx.NetworkXError(
f"index {arg} not a parameter index and this function doesn't have args"
)
mutable_args = True
return f"{sig.args}[{arg - sig.n_positional}]"
if self._finally:
# here's where we handle try_finally decorators. Such a decorator
# returns a mapped argument and a function to be called in a
# finally block. This feature was required by the open_file
# decorator. The below generates the code
#
# name, final = func(name) #<--append to mapblock
# try: #<--append to mapblock
# ... more argmapping and try blocks
# return WRAPPED_FUNCTION(...)
# ... more finally blocks
# finally: #<--prepend to finallys
# final() #<--prepend to finallys
#
for a in self._args:
name = get_name(a)
final = self._name(name)
mapblock.append(f"{name}, {final} = {fname}({name})")
mapblock.append("try:")
finallys = ["finally:", f"{final}()#", "#", finallys]
else:
mapblock.extend(
f"{name} = {fname}({name})" for name in map(get_name, self._args)
)
return sig, wrapped_name, functions, mapblock, finallys, mutable_args
[docs] @classmethod
def signature(cls, f):
r"""Construct a Signature object describing `f`
Compute a Signature so that we can write a function wrapping f with
the same signature and call-type.
Parameters
----------
f : callable
A function to be decorated
Returns
-------
sig : argmap.Signature
The Signature of f
Notes
-----
The Signature is a namedtuple with names:
name : a unique version of the name of the decorated function
signature : the inspect.signature of the decorated function
def_sig : a string used as code to define the new function
call_sig : a string used as code to call the decorated function
names : a dict keyed by argument name and index to the argument's name
n_positional : the number of positional arguments in the signature
args : the name of the VAR_POSITIONAL argument if any, i.e. \*theseargs
kwargs : the name of the VAR_KEYWORDS argument if any, i.e. \*\*kwargs
These named attributes of the signature are used in `assemble` and `compile`
to construct a string of source code for the decorated function.
"""
sig = inspect.signature(f, follow_wrapped=False)
def_sig = []
call_sig = []
names = {}
kind = None
args = None
kwargs = None
npos = 0
for i, param in enumerate(sig.parameters.values()):
# parameters can be position-only, keyword-or-position, keyword-only
# in any combination, but only in the order as above. we do edge
# detection to add the appropriate punctuation
prev = kind
kind = param.kind
if prev == param.POSITIONAL_ONLY != kind:
# the last token was position-only, but this one isn't
def_sig.append("/")
if prev != param.KEYWORD_ONLY == kind != param.VAR_POSITIONAL:
# param is the first keyword-only arg and isn't starred
def_sig.append("*")
# star arguments as appropriate
if kind == param.VAR_POSITIONAL:
name = "*" + param.name
args = param.name
count = 0
elif kind == param.VAR_KEYWORD:
name = "**" + param.name
kwargs = param.name
count = 0
else:
names[i] = names[param.name] = param.name
name = param.name
count = 1
# assign to keyword-only args in the function call
if kind == param.KEYWORD_ONLY:
call_sig.append(f"{name} = {name}")
else:
npos += count
call_sig.append(name)
def_sig.append(name)
fname = cls._name(f)
def_sig = f'def {fname}({", ".join(def_sig)}):'
if inspect.isgeneratorfunction(f):
_return = "yield from"
else:
_return = "return"
call_sig = f"{_return} {{}}({', '.join(call_sig)})"
return cls.Signature(fname, sig, def_sig, call_sig, names, npos, args, kwargs)
Signature = collections.namedtuple(
"Signature",
[
"name",
"signature",
"def_sig",
"call_sig",
"names",
"n_positional",
"args",
"kwargs",
],
)
@staticmethod
def _flatten(nestlist, visited):
"""flattens a recursive list of lists that doesn't have cyclic references
Parameters
----------
nestlist : iterable
A recursive list of objects to be flattened into a single iterable
visited : set
A set of object ids which have been walked -- initialize with an
empty set
Yields
------
Non-list objects contained in nestlist
"""
for thing in nestlist:
if isinstance(thing, list):
if id(thing) in visited:
raise ValueError("A cycle was found in nestlist. Be a tree.")
else:
visited.add(id(thing))
yield from argmap._flatten(thing, visited)
else:
yield thing
_tabs = " " * 64
@staticmethod
def _indent(*lines):
"""Indent list of code lines to make executable Python code
Indents a tree-recursive list of strings, following the rule that one
space is added to the tab after a line that ends in a colon, and one is
removed after a line that ends in an hashmark.
Parameters
----------
*lines : lists and/or strings
A recursive list of strings to be assembled into properly indented
code.
Returns
-------
code : str
Examples
--------
argmap._indent(*["try:", "try:", "pass#", "finally:", "pass#", "#",
"finally:", "pass#"])
renders to
'''try:
try:
pass#
finally:
pass#
#
finally:
pass#'''
"""
depth = 0
for line in argmap._flatten(lines, set()):
yield f"{argmap._tabs[:depth]}{line}"
depth += (line[-1:] == ":") - (line[-1:] == "#")