Skip to content

Instantly share code, notes, and snippets.

@salt-die
Last active November 9, 2020 19:36
Show Gist options
  • Save salt-die/ac6b7e75df258bdd4cff6e259ee50909 to your computer and use it in GitHub Desktop.
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?
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