Move code to functions module

Author: kinow

wdl2cwl/expr.py

@@ -5,7 +5,7 @@
 
 import WDL
 from wdl2cwl.errors import WDLSourceLine, ConversionException
-from wdl2cwl.util import get_input, nice_quote, get_mem_in_bytes, ConversionContext
+from wdl2cwl.util import get_input, nice_quote, ConversionContext
 
 
 def get_expr_ifthenelse(
@@ -140,204 +140,61 @@ def get_expr(wdl_expr: WDL.Expr.Base, ctx: ConversionContext) -> str:
             f"The expression '{wdl_expr}' is not handled yet."
         )
 
+_BINARY_OPS = {
+    "_gt": ">",
+    "_lor": "||",
+    "_neq": "!==",
+    "_lt": "<",
+    "_mul": "*",
+    "_eqeq": "===",
+    "_div": "/",
+    "_sub": "-",
+}
+
+_SINGLE_ARG_FN = {  # implemented elsewhere, just return the argument
+    "read_string",
+    "read_float",
+    "glob",
+    "read_int",
+    "read_boolean",
+    "read_tsv",
+    "read_lines",
+}
+
+
 def get_expr_apply(wdl_apply_expr: WDL.Expr.Apply, ctx: ConversionContext) -> str:
     """Translate WDL Apply Expressions."""
-    binary_ops = {
-        "_gt": ">",
-        "_lor": "||",
-        "_neq": "!==",
-        "_lt": "<",
-        "_mul": "*",
-        "_eqeq": "===",
-        "_div": "/",
-        "_sub": "-",
-    }
-    single_arg_fn = {  # implemented elsewhere, just return the argument
-        "read_string",
-        "read_float",
-        "glob",
-        "read_int",
-        "read_boolean",
-        "read_tsv",
-        "read_lines",
-    }
+    # N.B: This import here avoids circular dependency error when loading the modules.
+    from wdl2cwl import functions
+
     function_name = wdl_apply_expr.function_name
     arguments = wdl_apply_expr.arguments
     if not arguments:
         raise WDLSourceLine(wdl_apply_expr, ConversionException).makeError(
             f"The '{wdl_apply_expr}' expression has no arguments."
         )
     treat_as_optional = wdl_apply_expr.type.optional
-    if function_name == "_add":
-        add_left_operand = arguments[0]
-        add_right_operand = get_expr(arguments[1], ctx)
-        add_left_operand_value = get_expr(add_left_operand, ctx)
-        if getattr(add_left_operand, "function_name", None) == "basename":
-            referer = wdl_apply_expr.parent.name  # type: ignore[attr-defined]
-            treat_as_optional = True if referer in ctx.non_static_values else False
-        return (
-            f"{add_left_operand_value} + {add_right_operand}"
-            if not treat_as_optional
-            else f"{get_input(referer)} === null ? {add_left_operand_value} + {add_right_operand} : {get_input(referer)}"
-        )
-    elif function_name == "basename":
-        if len(arguments) == 1:
-            only_operand = arguments[0]
-            is_file = isinstance(only_operand.type, WDL.Type.File)
-            if isinstance(only_operand, WDL.Expr.Get) and isinstance(
-                only_operand.expr, WDL.Expr.Ident
-            ):
-                only_operand_name = get_expr_name(only_operand.expr)
-            else:
-                only_operand_name = get_expr(only_operand, ctx)
-            return (
-                f"{only_operand_name}.basename"
-                if is_file
-                else f"{only_operand_name}.split('/').reverse()[0]"
-            )
-        else:
-            basename_target, suffix = arguments
-            is_file = isinstance(basename_target.type, WDL.Type.File)
-            if isinstance(basename_target, WDL.Expr.Get):
-                basename_target_name = get_expr_name(basename_target.expr)  # type: ignore[arg-type]
-            elif isinstance(basename_target, WDL.Expr.Apply):
-                basename_target_name = get_expr(basename_target, ctx)
-            suffix_str = str(get_literal_value(suffix))
-            regex_str = re.escape(suffix_str)
-            return (
-                f"{basename_target_name}.basename.replace(/{regex_str}$/, '') "
-                if is_file
-                else f"{basename_target_name}.split('/').reverse()[0].replace(/{regex_str}$/, '')"
-            )
-    elif function_name == "defined":
-        only_operand = arguments[0]
-        assert isinstance(only_operand, WDL.Expr.Get) and isinstance(
-            only_operand.expr, WDL.Expr.Ident
-        )
-        return f"{get_expr_name(only_operand.expr)} !== null"
-    elif function_name == "_interpolation_add":
-        arg_value, arg_name = arguments
-        if isinstance(arg_name, WDL.Expr.String) and isinstance(
-            arg_value, (WDL.Expr.Apply, WDL.Expr.String)
-        ):
-            return f"{get_expr(arg_value, ctx)} + {get_expr(arg_name, ctx)}"
-        if isinstance(arg_name, (WDL.Expr.Placeholder, WDL.Expr.Get)):
-            just_arg_name = get_expr_name(arg_name.expr)  # type: ignore[arg-type]
-            arg_name_with_file_check = get_expr_name_with_is_file_check(
-                arg_name.expr  # type: ignore
-            )
-        elif isinstance(arg_value, (WDL.Expr.Placeholder, WDL.Expr.Get)):
-            just_arg_name = get_expr_name(arg_value.expr)  # type: ignore[arg-type]
-            arg_name_with_file_check = get_expr_name_with_is_file_check(
-                arg_value.expr  # type: ignore
-            )
-            arg_value = arg_name
-        with WDLSourceLine(arg_value, ConversionException):
-            arg_value_str = get_expr(arg_value, ctx)
-            return (
-                f'{just_arg_name} === null ? "" : {arg_value_str} + {arg_name_with_file_check}'
-                if treat_as_optional
-                else f"{arg_value_str} + {arg_name_with_file_check}"
-            )
-    elif function_name == "sub":
-        wdl_apply, arg_string, arg_sub = arguments
-        sub_expr = get_expr(wdl_apply, ctx)
-        arg_string_expr = get_expr(arg_string, ctx)
-        arg_sub_expr = get_expr(arg_sub, ctx)
-        return f"{sub_expr}.replace({arg_string_expr}, {arg_sub_expr}) "
-
-    elif function_name == "_at":
-        iterable_object, index = arguments
-        iterable_object_expr = get_expr(iterable_object, ctx)
-        index_expr = get_expr(index, ctx)
-        return f"{iterable_object_expr}[{index_expr}]"
-    elif function_name in binary_ops:
+
+    if function_name in _BINARY_OPS:
         left_operand, right_operand = arguments
         left_operand_expr = get_expr(left_operand, ctx)
         right_operand_expr = get_expr(right_operand, ctx)
         return (
-            f"{left_operand_expr} {binary_ops[function_name]} {right_operand_expr}"
+            f"{left_operand_expr} {_BINARY_OPS[function_name]} {right_operand_expr}"
         )
-    elif function_name == "length":
-        only_arg_expr = get_expr_get(arguments[0])  # type: ignore
-        return f"{only_arg_expr}.length"
-    elif function_name == "round":
-        only_arg_expr = get_expr(arguments[0], ctx)
-        return f"Math.round({only_arg_expr})"
-    elif function_name in single_arg_fn:
+    elif function_name in _SINGLE_ARG_FN:
         only_arg = arguments[0]
         return get_expr(only_arg, ctx)
-    elif function_name == "select_first":
-        array_obj = cast(WDL.Expr.Array, arguments[0])
-        array_items = [str(get_expr(item, ctx)) for item in array_obj.items]
-        items_str = ", ".join(array_items)
-        return (
-            f"[{items_str}].find(function(element) {{ return element !== null }}) "
+    elif hasattr(functions, function_name):
+        # Call the function if we have it in our wdl2cwl.functions module
+        kwargs = {
+            "treat_as_optional": treat_as_optional,
+            "wdl_apply_expr": wdl_apply_expr,
+        }
+        return cast(
+            str,
+            getattr(functions, function_name)(arguments, ctx, **kwargs),
         )
-    elif function_name == "select_all":
-        array_obj = cast(WDL.Expr.Array, arguments[0])
-        array_items = [str(get_expr(item, ctx)) for item in array_obj.items]
-        items_str = ", ".join(array_items)
-        return f"[{items_str}].filter(function(element) {{ return element !== null }}) "
-    elif function_name == "ceil":
-        only_arg = get_expr(arguments[0], ctx)  # type: ignore
-        return f"Math.ceil({only_arg}) "
-    elif function_name == "size":
-        if len(arguments) == 1:
-            left_operand = arguments[0]
-            unit_value = "1"
-        else:
-            left_operand, right_operand = arguments
-            right_value = str(get_literal_value(right_operand))
-            unit_base, unit_exponent = get_mem_in_bytes(right_value)
-            unit_value = f"{unit_base}^{unit_exponent}"
-        if isinstance(left_operand, WDL.Expr.Array):
-            array_items = [get_expr(item, ctx) for item in left_operand.items]
-            left = ", ".join(array_items)
-            left_str = f"[{left}]"
-        else:
-            left_str = get_expr(left_operand, ctx)
-        return (
-            "(function(size_of=0)"
-            + "{"
-            + f"{left_str}.forEach(function(element)"
-            + "{ if (element) {"
-            + "size_of += element.size"
-            + "}})}"
-            + f") / {unit_value}"
-        )
-    elif function_name == "flatten":
-        flatten_array = arguments[0]
-        with WDLSourceLine(flatten_array, ConversionException):
-            items_str = get_expr(flatten_array, ctx)
-        result = (
-            "(function () {var new_array = []; "
-            + items_str
-            + ".forEach(function(value, index, obj) "
-            "{value.forEach(function(sub_value, sub_index, sub_obj) "
-            "{new_array.push(sub_value);});}); return new_array;})()"
-        )
-        return result
-    elif function_name == "sep":
-        sep, array = arguments
-        if isinstance(array, WDL.Expr.Get) and isinstance(
-            array.type, WDL.Type.Array
-        ):
-            item_type = array.type.item_type
-        else:
-            raise WDLSourceLine(array, ConversionException).makeError(
-                f"Unhandled sep array type: {type(array)}: {array}."
-            )
-        sep_str = get_literal_value(sep) or ""
-        if isinstance(item_type, WDL.Type.File):
-            return (
-                f"{get_expr(array, ctx)}.map("
-                + 'function(el) {return el.path}).join("'
-                + sep_str
-                + '")'
-            )
-        else:
-            return f'{get_expr(array, ctx)}.join("{sep_str}")'
     raise WDLSourceLine(wdl_apply_expr, ConversionException).makeError(
         f"Function name '{function_name}' not yet handled."
     )
@@ -429,7 +286,7 @@ def get_step_input_expr(
                 id_name = wf_expr.expr.name
                 referee = wf_expr.expr.referee
                 if referee and isinstance(referee, WDL.Tree.Scatter):
-                    scatter_name, value_from = get_step_input_expr(referee.expr, ctx)  # type: ignore [arg-type]
+                    scatter_name, value_from = get_step_input_expr(referee.expr, ctx)  # type: ignore[arg-type]
                     ctx.scatter_names.append(scatter_name)
                     return scatter_name, value_from
                 return id_name, None