fix: refact FutureLen, add FutureRange

Signed-off-by: zjgemi <[email protected]>

Author: zjgemi

.gitignore

@@ -136,6 +136,7 @@ _version.py
 
 # vscode
 .vscode/
+.idea/
 
 # Mac OS
 .DS_store

.idea/.gitignore

@@ -5,7 +5,6 @@
 from typing import Any, Dict, List, Union
 
 import jsonpickle
-from dflow.client.v1alpha1_sequence import V1alpha1Sequence
 
 from .common import S3Artifact
 from .config import config
@@ -527,9 +526,6 @@ def __getattr__(self, key):
                             (self.step.id, self.name)
                 return "outputs.parameters['%s']" % self.name
             return ""
-        if key == "value":
-            res = FutureLen(self)
-            return res
         return super().__getattr__(key)
 
     def __setattr__(self, key, value):
@@ -939,34 +935,3 @@ def convert_to_argo(self):
         for art in self.artifacts.values():
             artifacts.append(art.convert_to_argo())
         return V1alpha1Outputs(parameters=parameters, artifacts=artifacts)
-
-
-class FutureLen:
-    '''
-    to solve problem that Delayed acquisition length about the output parameters
-    used in debug module
-    '''
-
-    def __init__(self, par):
-        self.par = par
-        self.method = []
-
-    def get(self):
-        v = self.par.value
-        # preventing recursive calls
-        if isinstance(v, FutureLen):
-            raise (
-                "no parameters named values, use get_len after the output from pre_train has return"
-            )
-        for i in self.method:
-            if i == "argo_len":
-                self.par = len(v)
-            if "argo_sequence" in i:
-                start = i.find("{{") + 2
-                end = i.find("}}", start)
-                format = i[start:end]
-                self.par = V1alpha1Sequence(count=self.par, format=format)
-        return self.par
-
-    def set_method(self, method):
-        self.method.append(method)

.idea/dflow.iml

@@ -352,8 +352,6 @@ def render_script(self):
 
         script += "import json\n"
         script += "from dflow import config, s3_config\n"
-        if config["mode"] == "debug":
-            script += "config['mode'] = 'debug'\n"
         script += "config.update(json.loads('%s'))\n" % json.dumps(config)
         script += "s3_config.update(json.loads('%s'))\n" % \
             json.dumps(s3_config)

.idea/inspectionProfiles/profiles_settings.xml

@@ -1,5 +1,4 @@
 import logging
-from multiprocessing.sharedctypes import Value
 import re
 from copy import deepcopy
 from typing import Any, Dict, List, Union
@@ -10,7 +9,7 @@
 from .config import config
 from .context_syntax import GLOBAL_CONTEXT
 from .executor import Executor
-from .io import (PVC, ArgoVar, FutureLen, InputArtifact, InputParameter, OutputArtifact,
+from .io import (PVC, ArgoVar, InputArtifact, InputParameter, OutputArtifact,
                  OutputParameter)
 from .op_template import OPTemplate, PythonScriptOPTemplate, ShellOPTemplate
 from .resource import Resource
@@ -30,6 +29,36 @@
 uploaded_python_packages = []
 
 
+class FutureLen:
+    '''
+    To solve the problem of delayed acquisition for length of output
+    parameters in the debug mode
+    '''
+
+    def __init__(self, par):
+        self.par = par
+
+    def get(self):
+        return len(self.par.value)
+
+
+class FutureRange:
+    def __init__(self, *args):
+        self.args = args
+
+    def get(self):
+        args = []
+        for i in self.args:
+            if isinstance(i, FutureLen):
+                args.append(i.get())
+            elif isinstance(i, (InputParameter, OutputParameter)):
+                args.append(i.value)
+            else:
+                args.append(i)
+        args = tuple(args)
+        return list(range(*args))
+
+
 def argo_range(
         *args,
 ) -> ArgoVar:
@@ -40,12 +69,7 @@ def argo_range(
     Each argument can be Argo parameter
     """
     if config["mode"] == "debug":
-        args = tuple(i.value if isinstance(i, (InputParameter, OutputParameter
-                                               )) else i for i in args)
-        for i in range(len(args)):
-            if isinstance(args[i], (InputParameter, OutputParameter)):
-                args[i] = args[i].value
-        return list(range(*args))
+        return FutureRange(*args)
     start = 0
     step = 1
     if len(args) == 1:
@@ -86,19 +110,13 @@ def argo_sequence(
             with count, can be an Argo parameter
         format: a printf format string to format the value in the sequence
     """
-    if isinstance(count,FutureLen):
-        method = argo_sequence.__name__
-        if format != None:
-            #the writing style may not be compliant
-            method = method + ".{{%s}}" % format
-            count.set_method(method=method)
-            return count
-    if isinstance(count, ArgoVar):
-        count = "{{=%s}}" % count.expr
-    if isinstance(start, ArgoVar):
-        start = "{{=%s}}" % start.expr
-    if isinstance(end, ArgoVar):
-        end = "{{=%s}}" % end.expr
+    if config["mode"] != "debug":
+        if isinstance(count, ArgoVar):
+            count = "{{=%s}}" % count.expr
+        if isinstance(start, ArgoVar):
+            start = "{{=%s}}" % start.expr
+        if isinstance(end, ArgoVar):
+            end = "{{=%s}}" % end.expr
     return V1alpha1Sequence(count=count, start=start, end=end, format=format)
 
 
@@ -112,11 +130,7 @@ def argo_len(
         param: the Argo parameter which is a list
     """
     if config["mode"] == "debug":
-        par = param.value
-        if isinstance(par,FutureLen):
-            par.set_method(argo_len.__name__)
-            return par
-        return len(par)
+        return FutureLen(param)
     if isinstance(param, S3Artifact):
         try:
             path_list = catalog_of_artifact(param)
@@ -533,8 +547,7 @@ def __init__(
                 self.inputs.parameters["dflow_with_param"] = InputParameter(
                     value=self.with_param)
                 self.with_param = None
-            if self.with_sequence is not None and \
-            not isinstance(self.with_sequence,FutureLen):
+            if self.with_sequence is not None:
                 if self.with_sequence.start is not None:
                     steps.inputs.parameters["dflow_sequence_start"] = \
                         InputParameter()
@@ -683,6 +696,7 @@ def run(self, context):
 
         import os
         from copy import copy
+
         from .dag import DAG
         from .steps import Steps
 
@@ -750,25 +764,32 @@ def run(self, context):
             return
 
         if self.with_param is not None or self.with_sequence is not None:
-            if isinstance(self.with_param, (InputParameter, OutputParameter)):
+            if isinstance(self.with_param, FutureRange):
+                item_list = self.with_param.get()
+            elif isinstance(self.with_param, (InputParameter,
+                                              OutputParameter)):
                 item_list = self.with_param.value
             elif isinstance(self.with_param, list):
                 item_list = self.with_param
             elif self.with_sequence is not None:
-                if isinstance(self.with_sequence,FutureLen):
-                    self.with_sequence = self.with_sequence.get()
                 start = 0
                 if self.with_sequence.start is not None:
                     start = self.with_sequence.start
+                    if isinstance(start, FutureLen):
+                        start = start.get()
                     if isinstance(start, (InputParameter, OutputParameter)):
                         start = start.value
                 if self.with_sequence.count is not None:
                     count = self.with_sequence.count
+                    if isinstance(count, FutureLen):
+                        count = count.get()
                     if isinstance(count, (InputParameter, OutputParameter)):
                         count = count.value
                     sequence = list(range(start, start + count))
                 if self.with_sequence.end is not None:
                     end = self.with_sequence.end
+                    if isinstance(end, FutureLen):
+                        end = end.get()
                     if isinstance(end, (InputParameter, OutputParameter)):
                         end = end.value
                     sequence = list(range(start, end + 1))