Expressions in Janis

janis_core/__init__.py

@@ -55,7 +55,7 @@
     Stdout,
     Stderr,
 )
-from janis_core.types import (
+from janis_core.operators import (
     InputSelector,
     WildcardSelector,
     MemorySelector,

janis_core/code/codetool.py

@@ -85,6 +85,8 @@ def generate_inputs_override(
         hints=None,
         include_defaults=True,
     ):
+        from janis_core.operators.selectors import Selector
+
         d, ad = {}, additional_inputs or {}
         for i in self.inputs():
             if (
@@ -95,8 +97,13 @@ def generate_inputs_override(
                 d[i] = ad.get(i.id(), i.default)
 
         if with_resource_overrides:
-            cpus = self.cpus(hints) or 1
+            cpus = self.cpus(hints)
             mem = self.memory(hints)
+
+            if isinstance(cpus, Selector):
+                cpus = None
+            if isinstance(mem, Selector):
+                mem = None
             d.update(
                 {
                     "runtime_memory": mem,

janis_core/code/pythontool.py

@@ -163,7 +163,7 @@ class PythonTool:
     args = cli.parse_args()
     result = code_block({argkwargs})
     print(json.dumps(result))
-except e:
+except Exception as e:
     print(str(e), file=sys.stderr)
     raise
 """

janis_core/graph/steptaginput.py

@@ -1,5 +1,8 @@
-from typing import Optional, Dict, Any
+from typing import Optional, Dict, Any, List
 
+from janis_core.types import get_instantiated_type
+
+from janis_core.operators import Selector
 from janis_core.graph.node import Node, NodeType
 from janis_core.tool.tool import TInput, TOutput
 from janis_core.types.common_data_types import Array
@@ -21,20 +24,14 @@ def full_dot(node: Node, tag: Optional[str]) -> str:
 
 class Edge:
     def __init__(
-        self,
-        start: Node,
-        stag: Optional[str],
-        finish: Node,
-        ftag: Optional[str],
-        should_scatter,
+        self, source: Selector, finish: Node, ftag: Optional[str], should_scatter
     ):
         Logger.log(
-            f"Creating edge: ({NodeType.to_str(start.node_type)}) '{start.id()}.{stag}' → "
+            f"Creating edge: ({source} → "
             f"({NodeType.to_str(finish.node_type)}) '{finish.id()}.{ftag}'"
         )
 
-        self.start: Node = start
-        self.stag: Optional[str] = stag
+        self.source = source
         self.finish: Node = finish
         self.ftag: Optional[str] = ftag
         self.compatible_types: Optional[bool] = None
@@ -43,20 +40,13 @@ def __init__(
         self.validate_tags()
         self.check_types()
 
-    def source_slashed(self):
-        return full_lbl(self.start, self.stag)
-
-    def source_dotted(self):
-        return full_dot(self.start, self.stag)
+    # def source_slashed(self):
+    #     return str(self.source)
+    #
+    # def source_dotted(self):
+    #     return full_dot(self.start, self.stag)
 
     def validate_tags(self):
-        if (
-            self.start.node_type == NodeType.STEP
-            and self.stag not in self.start.outputs()
-        ):
-            raise Exception(
-                f"Could not find the tag '{self.stag}' in the inputs of '{self.start.id()}'"
-            )
         if (
             self.finish.node_type == NodeType.STEP
             and self.ftag not in self.finish.inputs()
@@ -68,54 +58,39 @@ def validate_tags(self):
     def check_types(self):
         from janis_core.workflow.workflow import InputNode, StepNode
 
-        stoolin: TOutput = self.start.outputs()[
-            self.stag
-        ] if self.stag is not None else first_value(self.start.outputs())
+        # stoolin: TOutput = self.start.outputs()[
+        #     self.stag
+        # ] if self.stag is not None else first_value(self.start.outputs())
         ftoolin: TInput = self.finish.inputs()[
             self.ftag
         ] if self.ftag is not None else first_value(self.finish.inputs())
 
-        stype = stoolin.outtype
-        ftype = ftoolin.intype
-
-        start_is_scattered = (
-            isinstance(self.start, StepNode) and self.start.scatter is not None
-        )
-
-        if start_is_scattered:
-            Logger.log(
-                f"This edge merges the inputs from '{full_dot(self.start, self.stag)}' for "
-                f"'{full_dot(self.finish, self.ftag)}'"
-            )
-            stype = Array(stype)
+        stype = get_instantiated_type(self.source.returntype())
+        ftype = get_instantiated_type(ftoolin.intype)
 
         if self.scatter:
             if not isinstance(stype, Array):
                 raise Exception(
-                    f"Scatter was required for '{self.start.id()}.{self.stag} → '{self.finish.id()}.{self.ftag}' but "
+                    f"Scatter was required for '{self.source} → '{self.finish.id()}.{self.ftag}' but "
                     f"the input type was {type(stype).__name__} and not an array"
                 )
             stype = stype.subtype()
 
-        source_has_default = (
-            isinstance(self.start, InputNode) and self.start.default is not None
-        )
-
         # Scatters are handled automatically by the StepTagInput Array unwrapping
         # Merges are handled automatically by the `start_is_scattered` Array wrap
 
-        self.compatible_types = ftype.can_receive_from(stype, source_has_default)
+        self.compatible_types = ftype.can_receive_from(stype, False)
         if not self.compatible_types:
             if isinstance(ftype, Array) and ftype.subtype().can_receive_from(stype):
                 self.compatible_types = True
 
         if not self.compatible_types:
 
-            s = full_dot(self.start, self.stag)
+            s = str(self.source)
             f = full_dot(self.finish, self.ftag)
             message = (
                 f"Mismatch of types when joining '{s}' to '{f}': "
-                f"{stoolin.outtype.id()} -/→ {ftoolin.intype.id()}"
+                f"{stype.id()} -/→ {ftoolin.intype.id()}"
             )
             if isinstance(stype, Array) and ftype.can_receive_from(stype.subtype()):
                 message += " (did you forget to SCATTER?)"
@@ -133,12 +108,11 @@ def __init__(self, finish: Node, finish_tag: str):
         self.finish: Node = finish
         self.ftag: Optional[str] = finish_tag
 
-        self.default = None
         self.multiple_inputs = False
 
-        self.source_map: Dict[str, Edge] = {}
+        self.source_map: List[Edge] = []
 
-    def add_source(self, start: Node, stag: Optional[str], should_scatter) -> Edge:
+    def add_source(self, operator: Selector, should_scatter) -> Edge:
         """
         Add a connection
         :param start:
@@ -149,33 +123,34 @@ def add_source(self, start: Node, stag: Optional[str], should_scatter) -> Edge:
 
         from janis_core.workflow.workflow import StepNode
 
-        stype = (
-            start.outputs()[stag] if stag is not None else first_value(start.outputs())
-        ).outtype
+        # start: Node, stag: Optional[str]
+
+        # stype = (start.outputs()[stag] if stag is not None else first_value(start.outputs())).outtype
+        stype = operator.returntype()
         ftype = (
             self.finish.inputs()[self.ftag]
             if self.ftag is not None
             else first_value(self.finish.inputs())
         ).intype
 
-        start_is_scattered = isinstance(start, StepNode) and start.scatter is not None
-
-        if start_is_scattered:
-            Logger.log(
-                f"This edge merges the inputs from '{full_dot(start, stag)}' for "
-                f"'{full_dot(self.finish, self.ftag)}'"
-            )
-            stype = Array(stype)
+        # start_is_scattered = isinstance(start, StepNode) and start.scatter is not None
+        #
+        # if start_is_scattered:
+        #     Logger.log(
+        #         f"This edge merges the inputs from '{full_dot(start, stag)}' for "
+        #         f"'{full_dot(self.finish, self.ftag)}'"
+        #     )
+        #     stype = Array(stype)
 
         if should_scatter:
             if not isinstance(stype, Array):
                 raise Exception(
-                    f"Scatter was required for '{start.id()}.{stag} → '{self.finish.id()}.{self.ftag}' but "
+                    f"Scatter was required for '{operator} → '{self.finish.id()}.{self.ftag}' but "
                     f"the input type was {type(stype).__name__} and not an array"
                 )
             stype = stype.subtype()
 
-        if len(self.source_map) == 1 and start.id() not in self.source_map:
+        if len(self.source_map) == 1:  # and start.id() not in self.source_map:
             self.multiple_inputs = True
 
             if not isinstance(ftype, Array):
@@ -187,40 +162,16 @@ def add_source(self, start: Node, stag: Optional[str], should_scatter) -> Edge:
             # https://www.commonwl.org/user_guide/misc/#connect-a-solo-value-to-an-input-that-expects-an-array-of-that-type
             self.multiple_inputs = True
 
-        e = Edge(start, stag, self.finish, self.ftag, should_scatter=should_scatter)
-        self.source_map[start.id()] = e
+        e = Edge(operator, self.finish, self.ftag, should_scatter=should_scatter)
+        # todo: deal with source_map
+        self.source_map.append(e)
         return e
 
-    def set_default(self, default: Any):
-        Logger.log(
-            f"Setting the default of '{self.finish.id()}.{self.ftag}' to be '{str(default)}'"
-        )
-        self.default = default
-
     def source(self):
         n = len(self.source_map)
         if n == 0:
             return None
         elif n == 1:
-            return first_value(self.source_map)
-        else:
-            return list(self.source_map.values())
-
-    def dotted_source(self):
-        n = len(self.source_map)
-
-        if n == 0:
-            return None
-        elif n == 1:
-            return first_value(self.source_map).source_dotted()
-        else:
-            return [e.source_dotted() for e in self.source_map.values()]
-
-    def slashed_source(self):
-        n = len(self.source_map)
-        if n == 0:
-            return None
-        elif n == 1:
-            return first_value(self.source_map).source_slashed()
+            return self.source_map[0]
         else:
-            return [e.source_slashed() for e in self.source_map.values()]
+            return list(self.source_map)

janis_core/operators/__init__.py

@@ -0,0 +1,3 @@
+from .operator import *
+from .selectors import *
+from .stringformatter import StringFormatter