Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Apparently moving to a brand new plugin system.

That does not work as expected.
  • Loading branch information...
commit 6037e4079aa3556c27b0ed5a7ff063ee9e016cdc 1 parent 75f1ebf
@rik0 authored
View
63 pynetsym/graph/__init__.py
@@ -1,8 +1,9 @@
-import warnings
+from pynetsym.util.plugin import conditional_from_import_into_exporting, require
__all__ = ['IGraph', 'GraphError']
import traits.has_traits
+
traits.has_traits.CHECK_INTERFACES = 1
del traits
@@ -10,51 +11,15 @@
from .error import GraphError
-class _RegisteredPlugins:
- _plugins = set()
-
- @classmethod
- def has(cls, what):
- return what in cls._plugins
-
- @classmethod
- def can_test(cls, what):
- return what in cls._plugins, 'Missing module %s' % what
-
- @classmethod
- def register(cls, what):
- cls._plugins.add(what)
-
-try:
- import networkx
- del networkx
-except ImportError:
- warnings.warn("NetworkX not installed.")
-else:
- from nx_impl import NxGraph
- __all__.append('NxGraph')
- _RegisteredPlugins.register('networkx')
-
-try:
- import scipy.sparse
- del scipy
-except ImportError:
- warnings.warn("Scipy not installed.")
-else:
- from scipy_impl import ScipyGraph, DirectedScipyGraph
- __all__.extend(['ScipyGraph', 'DirectedScipyGraph'])
- _RegisteredPlugins.register('scipy')
-
-try:
- import numpy
- del numpy
-except ImportError:
- warnings.warn("Numpy not installed.")
-else:
- _RegisteredPlugins.register('numpy')
-
-
-has = _RegisteredPlugins.has
-can_test = _RegisteredPlugins.can_test
-
-del warnings
+ 'pynetsym.graph.nx_impl',
+ __name__,
+ __all__,
+ 'NxGraph')
+
+conditional_from_import_into_exporting('scipy.sparse',
+ 'pynetsym.graph.scipy_impl',
+ __name__, __all__,
+ 'ScipyGraph', 'DirectedScipyGraph')
+
+require('numpy')
View
72 pynetsym/util/plugin.py
@@ -0,0 +1,72 @@
+import importlib
+import warnings
+from pynetsym.util.iterators import inits
+
+__all__ = ['PluginManager']
+
+class PluginManager(object):
+
+ def __init__(self):
+ self._plugins = set()
+ self._missing = set()
+
+ def has(self, what):
+ module = self.require(what)
+ if module is None:
+ return False
+ else:
+ return True
+
+ def can_test(self, what):
+ return what in self._plugins, 'Missing module %s' % what
+
+ def _register(self, what):
+ self._plugins.add(what)
+
+ def require(self, what):
+ try:
+ module = importlib.import_module(what)
+
+ parts = what.split('.')
+ parts_beginnings = inits(parts)
+ next(parts_beginnings) # throw away empty subsequence
+ for init in parts_beginnings:
+ name = '.'.join(init)
+ return module
+ except ImportError:
+ if what not in self._missing:
+ warnings.warn('Cannot import %s' % what)
+ else:
+ self._missing.add(what)
+ return None
+
+ def conditional_import(self, if_has, then_import):
+ if self.has(if_has):
+ return importlib.import_module(then_import)
+
+ def conditional_from_import(self, if_has, import_from, *names):
+ if self.has(if_has):
+ mod = importlib.import_module(import_from)
+ return [(name, getattr(mod, name)) for name in names if hasattr(mod, name)]
+ else:
+ return []
+
+ def conditional_from_import_into(self, if_has, import_from, into, *names):
+ pairs = self.conditional_from_import(if_has, import_from, *names)
+ into = importlib.import_module(into)
+ for name, object in pairs:
+ setattr(into, name, object)
+
+ def conditional_from_import_into_exporting(self, if_has, import_from, into, namespace, *names):
+ pairs = self.conditional_from_import(if_has, import_from, *names)
+ into = importlib.import_module(into)
+ for name, object in pairs:
+ setattr(into, name, object)
+ namespace.append(name)
+
+
+_general_plugins = PluginManager()
+
+
+globals().update({name: getattr(_general_plugins, name) for name in dir(_general_plugins) if not name.startswith('_')})
+__all__.extend(name for name in dir(_general_plugins) if not name.startswith('_'))
View
13 tests/test_graph.py
@@ -4,7 +4,8 @@
import itertools as it
import networkx as nx
-from pynetsym.graph import NxGraph, ScipyGraph, GraphError, DirectedScipyGraph, has, can_test
+from pynetsym.graph import NxGraph, ScipyGraph, GraphError, DirectedScipyGraph
+from pynetsym.util.plugin import has, can_test
undirected_graph_types = [(ScipyGraph, 100),
@@ -125,8 +126,7 @@ def testToNumpy(self):
A = self.graph.to_numpy()
self.assertTupleEqual((self.size, self.size), A.shape)
- #@unittest.skipUnless(*can_test('networkx'))
- @unittest.skip('not ready')
+ @unittest.skipUnless(*can_test('networkx'))
def testNetworkX(self):
G = self.graph.to_nx()
@@ -278,10 +278,3 @@ def testCopySemantics(self):
self.assertEqual(self.size, self.graph.number_of_nodes())
other_graph.add_edge(self.new_node, 0)
self.assertEqual(self.size - 1, self.graph.number_of_edges())
-
- def testNoCopySemantics(self):
- other_graph = self.graph.to_nx(copy=False)
- other_graph.add_node(self.new_node)
- self.assertEqual(self.size + 1, self.graph.number_of_nodes())
- other_graph.add_edge(self.new_node, 0)
- self.assertEqual(self.size, self.graph.number_of_edges())
Please sign in to comment.
Something went wrong with that request. Please try again.