Begin generalization to multi-folder image parsing (not complete in fractal_cmd.py)

Author: tcompa

fractal/fractal_cmd.py

@@ -391,31 +391,34 @@ def workflow_add_task(project_name, workflow_name, tasks):
 @click.argument("workflow_name", required=True, nargs=1)
 @click.argument("input_dataset", required=True, nargs=1)
 @click.argument("output_dataset", required=True, nargs=1)
-@click.argument("resource_in", required=True, nargs=1)
+@click.argument("resources_in", required=True, nargs=-1)
 @click.argument("resource_out", required=True, nargs=1)
 @click.argument("json_worker_params", required=True, nargs=1)
 def workflow_apply(
     project_name,
     workflow_name,
     input_dataset,
     output_dataset,
-    resource_in,
+    resources_in,
     resource_out,
     json_worker_params,
 ):
 
-    resource_in = add_slash_to_path(resource_in)
+    resources_in = [
+        add_slash_to_path(resource_in) for resource_in in resources_in
+    ]
     resource_out = add_slash_to_path(resource_out)
 
     prj, _ = project_file_load(project_name)
 
-    # Verify that resource_in has been added to the resources of input_dataset
+    # Verify that resources_in has been added to the resources of input_dataset
     dataset_resources = prj["datasets"][input_dataset]["resources"]
-    if resource_in not in dataset_resources:
-        raise Exception(
-            f"Error in workflow_apply, {resource_in} not in"
-            f" {dataset_resources}"
-        )
+    for resource_in in resources_in:
+        if resource_in not in dataset_resources:
+            raise Exception(
+                f"Error in workflow_apply, {resource_in} not in"
+                f" {dataset_resources}"
+            )
 
     # If the resource_out folder is not there, create it
     path_resource_out = Path(resource_out)
@@ -481,8 +484,7 @@ def collect_intermediate_results(inputs=[]):
         "create_zarr_structure_multifov",
     ]:
         kwargs = dict(
-            # FIXME : add support for a list!
-            in_paths=[resource_in],
+            in_paths=resources_in,
             out_path=resource_out,
             ext=ext,
             num_levels=num_levels,
@@ -496,14 +498,15 @@ def app_create_zarr_structure(**kwargs_):
             return dict_tasks[task_names[0]](**kwargs)
 
         future = app_create_zarr_structure(**kwargs)
+
         if task_names[0] == "create_zarr_structure":
-            zarrurls, channels = future.result()
+            zarrurls, chl_list = future.result()
             debug(zarrurls)
-            debug(channels)
+            debug(chl_list)
         elif task_names[0] == "create_zarr_structure_multifov":
-            zarrurls, channels, sites_list = future.result()
+            zarrurls, chl_list, sites_list = future.result()
             debug(zarrurls)
-            debug(channels)
+            debug(chl_list)
             debug(sites_list)
         task_names = task_names[1:]  # FIXME
     else:
@@ -514,25 +517,32 @@ def app_create_zarr_structure(**kwargs_):
     # Tasks 1,2,...
     db = db_load()
     for task in task_names:
+        if len(resources_in) > 1:
+            raise Exception(
+                "ERROR\n"
+                "Support for len(resources_in)>1 is not there.\n"
+                "Hint: we should modify the in_path argument of"
+                "yokogawa_to_zarr."
+            )
 
         if task == "yokogawa_to_zarr":
             kwargs = dict(
-                in_path=resource_in,
+                in_path=resources_in[0],  # FIXME
                 ext=ext,
                 delete_input=delete_input,
                 rows=rows,
                 cols=cols,
-                channels=channels,
+                chl_list=chl_list,
                 num_levels=num_levels,
                 coarsening_xy=coarsening_xy,
                 coarsening_z=coarsening_z,
             )
         if task == "yokogawa_to_zarr_multifov":
             kwargs = dict(
-                in_path=resource_in,
+                in_path=resources_in[0],  # FIXME
                 ext=ext,
                 delete_input=delete_input,
-                channels=channels,
+                chl_list=chl_list,
                 sites_list=sites_list,
                 num_levels=num_levels,
                 coarsening_xy=coarsening_xy,
@@ -541,7 +551,6 @@ def app_create_zarr_structure(**kwargs_):
 
         elif task == "maximum_intensity_projection":
             kwargs = dict(
-                channels=channels,
                 coarsening_xy=coarsening_xy,
             )
         elif task == "replicate_zarr_structure_mip":
@@ -550,7 +559,7 @@ def app_create_zarr_structure(**kwargs_):
             kwargs = dict(newzarrurl="new")
         elif task == "illumination_correction":
             kwargs = dict(
-                channels=channels,
+                chl_list=chl_list,
                 coarsening_xy=coarsening_xy,
                 overwrite=True,
                 # background=background,

fractal/tasks/create_zarr_structure.py

@@ -138,20 +138,19 @@ def create_zarr_structure(
         actual_channels.append(ch)
     print(f"actual_channels: {actual_channels}")
 
-    well = []
+    zarrurls = {"plate": [], "well": []}
+    # zarrurls_in_paths = {}
 
     if not out_path.endswith("/"):
         out_path += "/"
-
-    zarrurls = {"plate": [], "well": []}
-
     for plate in plates:
 
-        print(f"Creating {out_path}{plate}.zarr")
-
         # Define plate zarr
-        group_plate = zarr.group(out_path + f"{plate}.zarr")
-        zarrurls["plate"].append(out_path + f"{plate}.zarr")
+        zarrurl = f"{out_path}{plate}.zarr"
+        print(f"Creating {zarrurl}")
+        group_plate = zarr.group(zarrurl)
+        zarrurls["plate"].append(zarrurl)
+        # zarrurls_in_paths[zarrurl] = dict_plate_paths[plate]
 
         # Identify all wells
         plate_prefix = dict_plate_prefixes[plate]

fractal/tasks/yokogawa_to_zarr.py

@@ -28,7 +28,7 @@ def yokogawa_to_zarr(
     ext=None,
     rows=None,
     cols=None,
-    channels=None,
+    chl_list=None,
     num_levels=5,
     coarsening_xy=2,
     coarsening_z=1,
@@ -49,8 +49,8 @@ def yokogawa_to_zarr(
     :type rows: list
     :param cols: number of columns of the well
     :type cols: list
-    :param chl_list: list of the channels   #FIXME
-    :type chl_list: list                    #FIXME
+    :param chl_list: list of channel names (e.g. A01_C01)
+    :type chl_list: list
     :param num_levels: number of levels in the zarr pyramid
     :type num_levels: int
     :param coarsening_xy: coarsening factor along X and Y
@@ -79,10 +79,10 @@ def yokogawa_to_zarr(
     lazy_imread = delayed(imread)
     fc_list = {level: [] for level in range(num_levels)}
 
-    print(channels)
+    print(f"Channels: {chl_list}")
 
-    for channel in channels:
-        A, C = channel.split("_")
+    for chl in chl_list:
+        A, C = chl.split("_")
 
         l_rows = []
         all_rows = []
@@ -99,7 +99,7 @@ def yokogawa_to_zarr(
                 f"  in_path: {in_path}\n"
                 f"  ext: {ext}\n"
                 f"  well_ID: {well_ID}\n"
-                f"  channel: {channel},\n"
+                f"  channel: {chl},\n"
                 f"  glob_path: {glob_path}"
             )
         max_z = max(
@@ -222,7 +222,7 @@ def yokogawa_to_zarr(
         "-C",
         "--chl_list",
         nargs="+",
-        help="list of channels ",  # FIXME
+        help="list of channel names (e.g. A01_C01)",
     )
 
     parser.add_argument(