Last active
November 9, 2020 19:36
-
-
Save salt-die/ac6b7e75df258bdd4cff6e259ee50909 to your computer and use it in GitHub Desktop.
How best to provide a consistent interface to similar objects across multiple modules?
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from importlib import import_module | |
from inspect import signature | |
modules = 'igraph', 'networkx', 'graph_tool' | |
class interface: | |
"""A decorator for Graph methods which will set the correct interface method on the first call to that method. | |
""" | |
def __init__(self, func): | |
self.name = func.__name__ | |
self.doc = func.__doc__ | |
self.sig = signature(func) # Used to make sure registered interface methods have a matching signature | |
self.impl = {} | |
def register(self, mod): | |
"""Register this interface's implementation for a given module. | |
""" | |
def deco(func): | |
assert signature(func) == self.sig | |
func.__name__ = self.name | |
func.__doc__ = self.doc | |
self.impl[mod] = func | |
return deco | |
def __set_name__(self, cls, name): | |
self.cls_name = cls.__name__ # Used only for repr | |
if hasattr(cls, '_'): | |
delattr(cls, '_') | |
def __get__(self, instance, cls): | |
if instance is None: | |
return self | |
setattr(instance, self.name, self.impl[instance.module].__get__(instance)) | |
return getattr(instance, self.name) | |
def __repr__(self): | |
return f'{self.cls_name}.{self.name}_interface' | |
class Graph: | |
"""A consistent Graph interface for igraph, networkx, and graph_tool Graphs. If a module isn't specified, | |
the first one we find will be used. | |
""" | |
def __init__(self, *, module=None, directed=False): # TODO: params to add: self_loops, parallel_edges | |
if module is None: | |
for module in modules: | |
try: | |
__import__(module) | |
except ModuleNotFoundError: | |
continue | |
else: | |
self.module = module | |
break | |
else: | |
raise ModuleNotFoundError(f"None of {modules} found") | |
elif module not in modules: | |
raise TypeError(f'module must be one of {modules}]') | |
else: | |
self.module = module | |
self._load_module() | |
self._set_G(directed=directed) | |
@interface | |
def _load_module(self): | |
"""Import module needed for graph. | |
""" | |
@_load_module.register('igraph') | |
def _(self): | |
globals()['ig'] = import_module('igraph') | |
@_load_module.register('networkx') | |
def _(self): | |
globals()['nx'] = import_module('networkx') | |
globals()['count'] = import_module('itertools').count # Used to add unique vertices to the nx.Graph | |
@_load_module.register('graph_tool') | |
def _(self): | |
globals()['gt'] = import_module('graph_tool') | |
globals()['generation'] = import_module('graph_tool.generation') | |
globals()['random'] = import_module('random').random | |
@interface | |
def _set_G(self, *, directed=False): | |
"""Initialize a new graph. | |
""" | |
@_set_G.register('igraph') | |
def _(self, *, directed=False): | |
self._G = ig.Graph(directed=directed) | |
@_set_G.register('networkx') | |
def _(self, *, directed=False): | |
self.vertex_id = count() | |
if directed: | |
self._G = nx.DiGraph() | |
else: | |
self._G = nx.Graph() | |
@_set_G.register('graph_tool') | |
def _(self, *, directed=False): | |
self._G = gt.Graph(directed=directed) | |
self._G.set_fast_edge_removal() | |
@classmethod | |
def from_nm(cls, /, n, m, directed=False, **kwargs): | |
"""Return a Graph randomly selected from all graphs with n vertices and m edges. | |
""" | |
g = cls(directed=directed, **kwargs) | |
if g.module == 'igraph': | |
g._G = ig.Graph.Erdos_Renyi(n, m=m, directed=directed, loops=True) | |
elif g.module == 'networkx': | |
g.vertex_id = count(n) | |
g._G = nx.gnm_random_graph(n, m, directed=directed) | |
elif g.module == 'graph_tool': | |
g.add_vertices(n) | |
for _ in range(m): | |
g._G.add_edge(0, 0) # TODO: use interface method in the future | |
generation.random_rewire(g._G, model='erdos', self_loops=True) | |
return g | |
@classmethod | |
def from_np(cls, /, n, p, directed=False, **kwargs): | |
"""Return a Graph with n vertices where each possible edge exists with probability p. | |
""" | |
g = cls(directed=directed, **kwargs) | |
if g.module == 'igraph': | |
g._G = ig.Graph.Erdos_Renyi(n, p=p, directed=directed, loops=True) | |
elif g.module == 'networkx': | |
g.vertex_id = count(n) | |
g._G = nx.gnp_random_graph(n, p, directed=directed) | |
elif g.module == 'graph_tool': | |
# This is O(n**2) in memory and time. TODO: implement a fast gnp generator for sparse graphs. | |
G = generation.complete_graph(n, self_loops=True, directed=directed) | |
G.set_fast_edge_removal() | |
edges_to_delete = [G.edge(*e) for e in G.iter_edges() if random() > p] | |
for e in edges_to_delete: | |
G.remove_edge(e) | |
g._G = G | |
return g | |
@interface | |
def order(self): # TODO: getattr should redirect vcount, num_vertices, or other aliases to here | |
"""Return the number of nodes of the graph. | |
""" | |
@order.register('igraph') | |
def _(self): | |
return self._G.vcount() | |
@order.register('networkx') | |
def _(self): | |
return self._G.order() | |
@order.register('graph_tool') | |
def _(self): | |
return self._G.num_vertices() | |
@interface | |
def size(self): | |
"""Return the number of edges in the graph. | |
""" | |
@size.register('igraph') | |
def _(self): | |
return self._G.ecount() | |
@size.register('networkx') | |
def _(self): | |
return self._G.size() | |
@size.register('graph_tool') | |
def _(self): | |
return self._G.num_edges() | |
@interface | |
def add_vertex(self): | |
"""Add a single vertex to the graph. | |
""" | |
@add_vertex.register('igraph') | |
def _(self): | |
self._G.add_vertex() | |
@add_vertex.register('networkx') | |
def _(self): | |
self._G.add_node(next(self.vertex_id)) | |
@add_vertex.register('graph_tool') | |
def _(self): | |
self._G.add_vertex() | |
@interface | |
def add_vertices(self, n=1): | |
"""Add n new vertices to the graph. | |
""" | |
@add_vertices.register('igraph') | |
def _(self, n=1): | |
self._G.add_vertices(n) | |
@add_vertices.register('networkx') | |
def _(self, n=1): | |
for _ in range(n): | |
self._G.add_node(next(self.vertex_id)) | |
@add_vertices.register('graph_tool') | |
def _(self, n=1): | |
self._G.add_vertex(n) | |
@interface | |
def remove_vertex(self, v): | |
"""Remove vertex v from the graph. Raises a ValueError if v is not in the graph. | |
""" | |
@remove_vertex.register('igraph') | |
def _(self, v): | |
try: | |
self._G.delete_vertices(v) | |
except ig.InternalError as e: | |
raise ValueError(e) from e | |
@remove_vertex.register('networkx') | |
def _(self, v): | |
try: | |
self._G.remove_node(v) | |
except nx.NetworkXError as e: | |
raise ValueError(e) from e | |
@remove_vertex.register('graph_tool') | |
def _(self, v): | |
self._G.remove_vertex(v, fast=True) | |
@interface | |
def add_edge(self, u, v): | |
"""Add an edge from vertex u to vertex v. Raises a ValueError if u or v are not in the graph. | |
""" | |
@add_edge.register('igraph') | |
def _(self, u, v): | |
try: | |
self._G.add_edges([(u, v)]) | |
except ig.InternalError as e: | |
raise ValueError(f"{u} or {v} not in graph") from e | |
@add_edge.register('networkx') | |
def _(self, u, v): | |
if u not in self._G or v not in self._G: | |
raise ValueError(f"{u} or {v} not in graph") | |
self._G.add_edge(u, v) | |
@add_edge.register('graph_tool') | |
def _(self, u, v): | |
self._G.add_edge(u, v, add_missing=False) | |
@interface | |
def remove_edge(self, u, v): | |
"""Remove an edge from vertex u to vertex v. Raises a ValueError if u or v or (u, v) are not in the graph. | |
""" | |
@remove_edge.register('igraph') | |
def _(self, u, v): | |
self._G.delete_edges([(u, v)]) | |
@remove_edge.register('networkx') | |
def _(self, u, v): | |
try: | |
self._G.remove_edge(u, v) | |
except nx.NetworkXError as e: | |
raise ValueError(e) from e | |
@remove_edge.register('graph_tool') | |
def _(self, u, v): | |
e = self._G.edge(u, v) | |
if e is None: | |
raise ValueError(f"edge ({u}, {v}) not in graph") | |
self._G.remove_edge(e) | |
@interface | |
def neighbors(self, v): | |
"""Return a generator over the vertex v's (out-)neighbors. Raises ValueError if v is not in the graph. | |
""" | |
@neighbors.register('igraph') | |
def _(self, v): | |
try: | |
yield from self._G.neighbors(v) | |
except ig.InternalError as e: | |
raise ValueError(e) from e | |
@neighbors.register('networkx') | |
def _(self, v): | |
try: | |
yield from self._G.neighbors(v) | |
except nx.NetworkXError as e: | |
raise ValueError(e) from e | |
@neighbors.register('graph_tool') | |
def _(self, v): | |
yield from self._G.get_out_neighbors(v) | |
# Alternatively: | |
# for v in self._G.vertex(v).out_neighbors(): | |
# yield int(v) | |
# But this creates a vertex descriptor for each neighbor, as opposed to a numpy array of ints. | |
# My wild guess is the array init will be faster than the multiple descriptor inits. | |
@interface | |
def vertices(self): | |
"""Iterator over the vertices of the graph. | |
""" | |
@vertices.register('igraph') | |
def _(self): | |
for v in self._G.vs: | |
yield v.index | |
@vertices.register('networkx') | |
def _(self): | |
yield from self._G.nodes | |
@vertices.register('graph_tool') | |
def _(self): | |
yield from self._G.iter_vertices() | |
@interface | |
def edges(self): | |
"""Iterator over the edges of the graph. | |
""" | |
@edges.register('igraph') | |
def _(self): | |
for e in self._G.es: | |
yield e.target, e.source | |
@edges.register('networkx') | |
def _(self): | |
yield from self._G.edges | |
@edges.register('graph_tool') | |
def _(self): | |
for s, t in self._G.iter_edges(): | |
yield s, t |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment