New [Inputs] : input_settings are now list of input maps, handles both dicts/lists for user

Author: Guillaume V.

src/vip_client/classes/VipLauncher.py

@@ -130,34 +130,39 @@ def pipeline_id(self) -> None:
 
     # Input settings
     @property
-    def input_settings(self) -> dict:
+    def input_settings(self) -> list[dict] | None:
         """All parameters needed to run the pipeline 
         Run show_pipeline() for more information"""
         # Return None if the private attribute is unset
         return self._get_input_settings() if self._is_defined("_input_settings") else None
 
     @input_settings.setter
-    def input_settings(self, input_settings: dict):
+    def input_settings(self, input_settings: dict | list[dict]):
         # Call deleter if agument is None
         if input_settings is None: 
             del self.input_settings
             return
         # Display
         self._print("Input Settings --> ", end="", flush=True)
         # Check type
-        if not isinstance(input_settings, dict):
-            raise TypeError("`input_settings` should be a dictionary")
-        
+        is_input_dict_list = isinstance(input_settings, list)
+        if not isinstance(input_settings, dict) and not is_input_dict_list:
+            raise TypeError("`input_settings` should be a dictionary or a list of dictionnary")
+
         # Check if each input can be converted to a string with valid characters and no empty strings
-        self._check_invalid_input(input_settings)
-        
-        # Parse the input settings
-        new_settings = self._parse_input_settings(input_settings)
-        self._print("parsed")
-        # Check conflicts with private attribute
-        self._check_value("_input_settings", new_settings)
-        # Update
-        self._input_settings = new_settings
+        inputs = input_settings if is_input_dict_list else [input_settings]
+        new_input_settings = []
+        for input_dict in inputs:
+            self._check_invalid_input(input_dict)
+            # Parse the input settings
+            new_settings = self._parse_input_settings(input_dict)
+            self._print("parsed")
+            # Check conflicts with private attribute
+            self._check_value("_input_settings", new_settings)
+            # Update
+            new_input_settings.append(new_settings)
+
+        self._input_settings = new_input_settings
 
     @input_settings.deleter
     def input_settings(self) -> None:
@@ -556,7 +561,7 @@ def monitor_workflows(self, refresh_time=30) -> VipLauncher:
             self._print("Run launch_pipeline() to launch workflows on VIP.")
             return self
         # Update existing workflows
-        self._print("Updating worflow inventory ... ", end="", flush=True)
+        self._print("Updating workflow inventory ... ", end="", flush=True)
         self._update_workflows()
         self._print("Done.")
         # Check if workflows are still running
@@ -565,7 +570,7 @@ def monitor_workflows(self, refresh_time=30) -> VipLauncher:
             self._execution_report()
             # Display standby
             self._print("\n-------------------------------------------------------------")
-            self._print("The current proccess will wait until all executions are over.")
+            self._print("The current process will wait until all executions are over.")
             self._print("Their progress can be monitored on VIP portal:")
             self._print(f"\t{self._VIP_PORTAL}")
             self._print("-------------------------------------------------------------")
@@ -1538,21 +1543,21 @@ def parse_value(input):
         }
 
     # Get the input settings after files are parsed as PathLib objects
-    def _get_input_settings(self, location="vip") -> dict:
+    def _get_input_settings(self, location="vip") -> list[dict]:
         """
         Returns the input settings with their orignal values in string format.
         `location` is destined to subclasses.
         """
         if location != "vip":
             raise NotImplementedError(f"Unknown location: {location}")
-        return {
-            key: [str(v) for v in value] if isinstance(value, list) else str(value)
-            for key, value in self._input_settings.items()
-        }
+        return [
+            {key: [str(v) for v in value] if isinstance(value, list) else str(value) for key, value in input_dict.items()}
+            for input_dict in self._input_settings
+        ]
     # ------------------------------------------------
 
     # Check the input settings based on the pipeline descriptor
-    def _check_input_settings(self, input_settings: dict=None, location: str=None) -> None:
+    def _check_input_settings(self, input_settings: list[dict]=None, location: str=None) -> None:
         """
         Checks `input_settings` with respect to pipeline descriptor. If not provided, checks the instance property.
         Prerequisite: input_settings contains only strings or lists of strings.
@@ -1580,10 +1585,12 @@ def _check_input_settings(self, input_settings: dict=None, location: str=None) -
         # Check the pipeline identifier
         if not self._is_defined("_pipeline_id"): 
             raise AttributeError("Input settings could not be checked without a pipeline identifier.")
-        # Parameter names
-        self._check_input_keys(input_settings)
-        # Parameter values
-        self._check_input_values(input_settings, location=location)
+
+        for input_dict in input_settings:
+            # Parameter names
+            self._check_input_keys(input_dict)
+            # Parameter values
+            self._check_input_values(input_dict, location=location)
         # Return True when all checks are complete
         return True
     # ------------------------------------------------      
@@ -1677,7 +1684,9 @@ def _check_input_values(self, input_settings: dict, location: str) -> None:
                     missing_files.extend(missing_files_found)
                     continue
             if param["type"] == "Boolean":
-                if value not in ["true", "false"]:
+                # Handle boolean lists
+                values = value if isinstance(value, list) else [value]
+                if not all(v in ["true", "false"] for v in values):
                     wrong_type_inputs.append(name)
                     continue
             # Check other input formats ?

src/vip_client/classes/VipSession.py

@@ -1140,7 +1140,7 @@ def parse_value(input):
     # ------------------------------------------------
 
     # Get the input settings after they are parsed
-    def _get_input_settings(self, location="vip") -> dict:
+    def _get_input_settings(self, location="vip") -> list[dict]:
         """
         Fits `self._input_settings` to `location`, i.e. write the input paths relatively to `location`.
         Returns the modified settings.
@@ -1173,10 +1173,10 @@ def get_input(value, location) -> str:
         if location not in ("vip", "local"):
             raise NotImplementedError(f"Unknown location: {location}")
         # Browse input settings
-        return {
-            key: get_input(value, location)
-            for key, value in self._input_settings.items()
-        }
+        return [
+            {key: get_input(value, location) for key, value in input_dict.items()}
+            for input_dict in self._input_settings
+        ]
     # ------------------------------------------------
 
     def _update_input_settings(self) -> None:
@@ -1185,7 +1185,7 @@ def _update_input_settings(self) -> None:
         This method does nothing if `input_settings` is unset.
         """
         if self._is_defined('_input_settings'):
-            self._input_settings = self._parse_input_settings(self._input_settings)
+            self._input_settings = [self._parse_input_settings(input_dict) for input_dict in self._input_settings]
     # ------------------------------------------------
 
     # Function to convert a VIP path to local output directory

src/vip_client/utils/vip.py

@@ -350,7 +350,7 @@ def count_executions()->int:
     return int(rq.text)
 
 # -----------------------------------------------------------------------------
-def init_exec(pipeline, name="default", inputValues={}, resultsLocation="/vip/Home") -> str:
+def init_exec(pipeline, name="default", inputValues=[], resultsLocation="/vip/Home") -> str:
     url = __PREFIX + 'executions'
     headers = {
                 'apikey': __apikey,
@@ -359,7 +359,7 @@ def init_exec(pipeline, name="default", inputValues={}, resultsLocation="/vip/Ho
     data_ = {
             "name": name, 
             'pipelineIdentifier': pipeline,
-            "inputValues": inputValues,
+            "inputValues": [inputValues] if isinstance(inputValues, dict) else inputValues,
             "resultsLocation": resultsLocation
            }
     rq = SESSION.post(url, headers=headers, json=data_)