Merge pull request #6 from encore-ecosystem/dev

Dev

nodeflow/__init__.py

@@ -1,4 +1,3 @@
 from nodeflow.adapters import *
 from nodeflow.converter import *
-from nodeflow.manager import *
 from nodeflow.node import *

nodeflow/adapters/abstract.py

@@ -19,6 +19,11 @@ def get_type_of_source_variable() -> Type[Variable]:
     def get_type_of_target_variable() -> Type[Variable]:
         raise NotImplementedError
 
+    @staticmethod
+    @abstractmethod
+    def is_loose_information() -> bool:
+        raise NotImplementedError
+
 
 __all__ = [
     'Adapter'

nodeflow/adapters/builtin/Bool2Int.py

@@ -16,6 +16,10 @@ def get_type_of_source_variable() -> Type[Bool]:
     def get_type_of_target_variable() -> Type[Int]:
         return Int
 
+    @staticmethod
+    def is_loose_information() -> bool:
+        return False
+
 
 __all__ = [
     "Bool2Int",

nodeflow/adapters/builtin/Float2Int.py

@@ -16,6 +16,10 @@ def get_type_of_source_variable() -> Type[Float]:
     def get_type_of_target_variable() -> Type[Int]:
         return Int
 
+    @staticmethod
+    def is_loose_information() -> bool:
+        return True
+
 
 __all__ = [
     "Float2Int",

nodeflow/adapters/builtin/Int2Float.py

@@ -16,6 +16,11 @@ def get_type_of_source_variable() -> Type[Int]:
     def get_type_of_target_variable() -> Type[Float]:
         return Float
 
+    @staticmethod
+    def is_loose_information() -> bool:
+        return False
+
+
 __all__ = [
     "Int2Float",
 ]

nodeflow/adapters/builtin/Pipeline.py

@@ -0,0 +1,36 @@
+from nodeflow.adapters.abstract import Adapter
+from nodeflow.node.abstract import Variable
+from typing import Type
+
+
+class Pipeline(Adapter):
+    def __init__(self):
+        self._loose_information        = False
+        self._pipeline : list[Adapter] = []
+
+    def add_adapter(self, adapter: Adapter):
+        self._pipeline += [adapter]
+        self._loose_information = self._loose_information or adapter.is_loose_information()
+        if len(self._pipeline) > 1:
+            s1 = self._pipeline[-2].get_type_of_target_variable()
+            s2 = self._pipeline[-1].get_type_of_source_variable()
+            assert s1 == s2, 'Adapter is invalid for current pipeline'
+
+    def convert(self, variable: Variable):
+        assert len(self._pipeline) > 0, "Pipeline is empty"
+        for adapter in self._pipeline:
+            variable = adapter.convert(variable)
+        return variable
+
+    def get_type_of_source_variable(self) -> Type[Variable]:
+        assert len(self._pipeline) > 0, "Pipeline is empty"
+        return self._pipeline[0].get_type_of_source_variable()
+
+    def get_type_of_target_variable(self) -> Type[Variable]:
+        assert len(self._pipeline) > 0, "Pipeline is empty"
+        return self._pipeline[-1].get_type_of_target_variable()
+
+    def is_loose_information(self) -> bool:
+        return self._loose_information
+
+__all__ = ['Pipeline']

nodeflow/adapters/builtin/__init__.py

@@ -1,3 +1,4 @@
 from .Int2Float import *
 from .Float2Int import *
 from .Bool2Int import *
+from .Pipeline import *

nodeflow/converter/__init__.py

@@ -1,59 +1 @@
-from nodeflow.adapters import Adapter
-from typing import Iterable, Type, Optional, List, Dict
-from nodeflow.node import Variable
-from collections import deque
-
-
-class Converter:
-    def __init__(self, adapters: Iterable[Adapter]):
-        self.graph = {}
-        self.add_adapters(adapters)
-
-    def add_adapter(self, adapter: Adapter):
-        # Resolve source type
-        source_type = adapter.get_type_of_source_variable().__name__
-        if source_type not in self.graph:
-            self.graph[source_type] = {}
-
-        # Resolve target type
-        target_type = adapter.get_type_of_target_variable().__name__
-        self.graph[source_type][target_type] = adapter
-
-    def add_adapters(self, adapters: Iterable[Adapter]):
-        for adapter in adapters:
-            self.add_adapter(adapter)
-
-    def is_support_variable(self, variable_type: Type[Variable]) -> bool:
-        return variable_type in self.graph
-
-    def convert(self, variable: Variable, to_type: Type[Variable]) -> Optional[Variable]:
-        pipeline = self._get_converting_pipeline(source=variable.__class__, target=to_type)
-        assert pipeline is not None, "Could not convert variable"
-
-        for i in range(1, len(pipeline)):
-            variable = self.graph[pipeline[i-1]][pipeline[i]].convert(variable)
-
-        return variable
-
-    def _get_converting_pipeline(self, source: Type[Variable], target: Type[Variable]) -> Optional[List[Variable]]:
-        # ---------
-        # BFS
-        # ---------
-        visited = set()
-        queue   = deque()
-        queue.append([source.__name__, [source.__name__]])
-
-        while len(queue) > 0:
-            root_type, type_road = queue.popleft()
-            visited.add(root_type)
-            for child in self.graph[root_type]:
-                if child in visited:
-                    continue
-                if child == target.__name__:
-                    return type_road + [child]
-                queue.append([child, type_road + [child]])
-
-
-__all__ = [
-    "Converter"
-]
+from .converter import *

nodeflow/converter/converter.py

@@ -0,0 +1,68 @@
+from nodeflow.adapters import Adapter, Pipeline
+from typing import Iterable, Type, Optional
+from nodeflow.node import Variable
+from collections import deque
+
+
+
+class Converter:
+    def __init__(self, adapters: Iterable[Adapter]):
+        self.graph = {}
+        self.add_adapters(adapters)
+
+    def add_adapter(self, adapter: Adapter):
+        # Resolve source type
+        source_type = adapter.get_type_of_source_variable().__name__
+        if source_type not in self.graph:
+            self.graph[source_type] = {}
+
+        # Resolve target type
+        target_type = adapter.get_type_of_target_variable().__name__
+        self.graph[source_type][target_type] = adapter
+
+    def add_adapters(self, adapters: Iterable[Adapter]):
+        for adapter in adapters:
+            self.add_adapter(adapter)
+
+    def is_support_variable(self, variable_type: Type[Variable]) -> bool:
+        return variable_type in self.graph
+
+    def convert(self, variable: Variable, to_type: Type[Variable]) -> Optional[Variable]:
+        pipeline = self._get_converting_pipeline(source=variable.__class__, target=to_type)
+        assert pipeline is not None, "Could not convert variable"
+        return pipeline.convert(variable)
+
+    def _get_converting_pipeline(self, source: Type[Variable], target: Type[Variable]) -> Optional[Pipeline]:
+        pipeline_with_loose_information : Optional[Pipeline] = None
+        # ---------
+        # BFS
+        # ---------
+        visited = set()
+        queue   = deque()
+        queue.append([source.__name__, [source.__name__]])
+
+        while len(queue) > 0:
+            root_type, type_road = queue.popleft()
+            visited.add(root_type)
+            for child in self.graph[root_type]:
+                if child in visited:
+                    continue
+                if child == target.__name__:
+                    pipeline = Pipeline()
+                    road = type_road + [child]
+                    for i in range(len(road) - 1):
+                        pipeline.add_adapter(self.graph[road[i]][road[i + 1]])
+
+                    if not pipeline_with_loose_information and pipeline.is_loose_information():
+                        # the shortest pipeline with loosing an information
+                        pipeline_with_loose_information = pipeline
+                    else:
+                        # the shortest pipeline without loosing an information
+                        return pipeline
+                queue.append([child, type_road + [child]])
+
+        return pipeline_with_loose_information
+
+__all__ = [
+    "Converter"
+]

tests/test_converter.py

@@ -27,3 +27,4 @@ def test_bool2float():
         to_type=Float
     ).value == 1.
 
+test_bool2float()