Remove workaround for utility functions on NXF Fusion

CHANGELOG.md

@@ -1,3 +1,9 @@
+# openpipelines x.x.x
+
+# MINOR CHANGES
+
+* Several component (cleanup): remove workaround for using being able to use shared utility functions with Nextflow Fusion (PR #920).
+
 # openpipelines 2.0.0-rc.2
 
 ## BUG FIXES

src/annotate/celltypist/script.py

@@ -27,18 +27,8 @@
 meta = {
 }
 ## VIASH END
-
-# START TEMPORARY WORKAROUND setup_logger
-def setup_logger():
-    logger = logging.getLogger()
-    logger.setLevel(logging.INFO)
-    console_handler = logging.StreamHandler(sys.stdout)
-    logFormatter = logging.Formatter("%(asctime)s %(levelname)-8s %(message)s")
-    console_handler.setFormatter(logFormatter)
-    logger.addHandler(console_handler)
-
-    return logger
-# END TEMPORARY WORKAROUND setup_logger
+sys.path.append(meta["resources_dir"])
+from setup_logger import setup_logger
 
 def check_celltypist_format(indata):
     if np.abs(np.expm1(indata[0]).sum()-10000) > 1:

src/annotate/onclass/script.py

@@ -31,17 +31,7 @@
 ## VIASH END
 
 sys.path.append(meta["resources_dir"])
-# START TEMPORARY WORKAROUND setup_logger
-def setup_logger():
-    logger = logging.getLogger()
-    logger.setLevel(logging.INFO)
-    console_handler = logging.StreamHandler(sys.stdout)
-    logFormatter = logging.Formatter("%(asctime)s %(levelname)-8s %(message)s")
-    console_handler.setFormatter(logFormatter)
-    logger.addHandler(console_handler)
-
-    return logger
-# END TEMPORARY WORKAROUND setup_logger
+from setup_logger import setup_logger
 
 logger = setup_logger()
 

src/annotate/popv/script.py

@@ -52,22 +52,7 @@ def mps_is_available():
 ## VIASH END
 
 sys.path.append(meta["resources_dir"])
-# START TEMPORARY WORKAROUND setup_logger
-# reason: resources aren't available when using Nextflow fusion
-# from setup_logger import setup_logger
-def setup_logger():
-    import logging
-    from sys import stdout
-
-    logger = logging.getLogger()
-    logger.setLevel(logging.INFO)
-    console_handler = logging.StreamHandler(stdout)
-    logFormatter = logging.Formatter("%(asctime)s %(levelname)-8s %(message)s")
-    console_handler.setFormatter(logFormatter)
-    logger.addHandler(console_handler)
-
-    return logger
-# END TEMPORARY WORKAROUND setup_logger
+from setup_logger import setup_logger
 logger = setup_logger()
 
 use_gpu = cuda_is_available()

src/annotate/random_forest_annotation/script.py

@@ -30,17 +30,7 @@
 ## VIASH END
 
 sys.path.append(meta["resources_dir"])
-# START TEMPORARY WORKAROUND setup_logger
-def setup_logger():
-    logger = logging.getLogger()
-    logger.setLevel(logging.INFO)
-    console_handler = logging.StreamHandler(sys.stdout)
-    logFormatter = logging.Formatter("%(asctime)s %(levelname)-8s %(message)s")
-    console_handler.setFormatter(logFormatter)
-    logger.addHandler(console_handler)
-
-    return logger
-# END TEMPORARY WORKAROUND setup_logger
+from setup_logger import setup_logger
 
 logger = setup_logger()
 

src/annotate/scanvi/script.py

@@ -40,22 +40,7 @@
 sys.path.append(meta["resources_dir"])
 from query_reference_allignment import set_var_index, cross_check_genes
 
-# START TEMPORARY WORKAROUND setup_logger
-# reason: resources aren't available when using Nextflow fusion
-# from setup_logger import setup_logger
-def setup_logger():
-    import logging
-    from sys import stdout
-
-    logger = logging.getLogger()
-    logger.setLevel(logging.INFO)
-    console_handler = logging.StreamHandler(stdout)
-    logFormatter = logging.Formatter("%(asctime)s %(levelname)-8s %(message)s")
-    console_handler.setFormatter(logFormatter)
-    logger.addHandler(console_handler)
-
-    return logger
-# END TEMPORARY WORKAROUND setup_logger
+from setup_logger import setup_logger
 logger = setup_logger()
 
 if (not par["scvi_reference_model"]) and not (par["scanvi_reference_model"]) or (par["scvi_reference_model"] and par["scanvi_reference_model"]):

src/annotate/svm_annotation/script.py

@@ -32,17 +32,7 @@
 ## VIASH END
 
 sys.path.append(meta["resources_dir"])
-# START TEMPORARY WORKAROUND setup_logger
-def setup_logger():
-    logger = logging.getLogger()
-    logger.setLevel(logging.INFO)
-    console_handler = logging.StreamHandler(sys.stdout)
-    logFormatter = logging.Formatter("%(asctime)s %(levelname)-8s %(message)s")
-    console_handler.setFormatter(logFormatter)
-    logger.addHandler(console_handler)
-
-    return logger
-# END TEMPORARY WORKAROUND setup_logger
+from setup_logger import setup_logger
 
 logger = setup_logger()
 

src/cluster/leiden/config.vsh.yaml

@@ -71,6 +71,7 @@ resources:
   - type: python_script
     path: script.py
   - path: /src/utils/setup_logger.py
+  - path: /src/utils/compress_h5mu.py
 test_resources:
   - type: python_script
     path: test.py

src/cluster/leiden/script.py

@@ -37,57 +37,7 @@
 
 sys.path.append(meta["resources_dir"])
 
-# START TEMPORARY WORKAROUND compress_h5mu
-# reason: resources aren't available when using Nextflow fusion
-# from compress_h5mu import compress_h5mu
-from h5py import File as H5File
-from h5py import Group, Dataset
-from typing import Union, Literal
-from functools import partial
-
-def compress_h5mu(input_path: Union[str, Path], 
-                output_path: Union[str, Path], 
-                compression: Union[Literal['gzip'], Literal['lzf']]):
-    input_path, output_path = str(input_path), str(output_path)
-
-    def copy_attributes(in_object, out_object):
-        for key, value in in_object.attrs.items():
-            out_object.attrs[key] = value
-
-    def visit_path(output_h5: H5File,
-                   compression: Union[Literal['gzip'], Literal['lzf']], 
-                   name: str, object: Union[Group, Dataset]):
-            if isinstance(object, Group):
-                new_group = output_h5.create_group(name)
-                copy_attributes(object, new_group)
-            elif isinstance(object, Dataset):
-                # Compression only works for non-scalar Dataset objects
-                # Scalar objects dont have a shape defined
-                if not object.compression and object.shape not in [None, ()]: 
-                    new_dataset = output_h5.create_dataset(name, data=object, compression=compression)
-                    copy_attributes(object, new_dataset)
-                else:
-                    output_h5.copy(object, name)
-            else:
-                raise NotImplementedError(f"Could not copy element {name}, "
-                                          f"type has not been implemented yet: {type(object)}")
-
-    with H5File(input_path, 'r') as input_h5, H5File(output_path, 'w', userblock_size=512) as output_h5:
-        copy_attributes(input_h5, output_h5)
-        input_h5.visititems(partial(visit_path, output_h5, compression))
-
-    with open(input_path, "rb") as input_bytes:
-        # Mudata puts metadata like this in the first 512 bytes:
-        # MuData (format-version=0.1.0;creator=muon;creator-version=0.2.0)
-        # See mudata/_core/io.py, read_h5mu() function
-        starting_metadata = input_bytes.read(100)
-        # The metadata is padded with extra null bytes up until 512 bytes
-        truncate_location = starting_metadata.find(b"\x00")
-        starting_metadata = starting_metadata[:truncate_location]
-    with open(output_path, "br+") as f:
-        nbytes = f.write(starting_metadata)
-        f.write(b"\0" * (512 - nbytes)) 
-# END TEMPORARY WORKAROUND compress_h5mu
+from compress_h5mu import compress_h5mu
 
 _shared_logger_name = "leiden"
 

src/compression/compress_h5mu/script.py

@@ -9,59 +9,7 @@
 ### VIASH END
 
 sys.path.append(meta["resources_dir"])
-# START TEMPORARY WORKAROUND compress_h5mu
-# reason: resources aren't available when using Nextflow fusion
-# from compress_h5mu import compress_h5mu
-from h5py import File as H5File
-from h5py import Group, Dataset
-from pathlib import Path
-from typing import Union, Literal
-from functools import partial
-
-
-def compress_h5mu(input_path: Union[str, Path], 
-                output_path: Union[str, Path], 
-                compression: Union[Literal['gzip'], Literal['lzf']]):
-    input_path, output_path = str(input_path), str(output_path)
-
-    def copy_attributes(in_object, out_object):
-        for key, value in in_object.attrs.items():
-            out_object.attrs[key] = value
-
-    def visit_path(output_h5: H5File,
-                   compression: Union[Literal['gzip'], Literal['lzf']], 
-                   name: str, object: Union[Group, Dataset]):
-            if isinstance(object, Group):
-                new_group = output_h5.create_group(name)
-                copy_attributes(object, new_group)
-            elif isinstance(object, Dataset):
-                # Compression only works for non-scalar Dataset objects
-                # Scalar objects dont have a shape defined
-                if not object.compression and object.shape not in [None, ()]: 
-                    new_dataset = output_h5.create_dataset(name, data=object, compression=compression)
-                    copy_attributes(object, new_dataset)
-                else:
-                    output_h5.copy(object, name)
-            else:
-                raise NotImplementedError(f"Could not copy element {name}, "
-                                          f"type has not been implemented yet: {type(object)}")
-
-    with H5File(input_path, 'r') as input_h5, H5File(output_path, 'w', userblock_size=512) as output_h5:
-        copy_attributes(input_h5, output_h5)
-        input_h5.visititems(partial(visit_path, output_h5, compression))
-
-    with open(input_path, "rb") as input_bytes:
-        # Mudata puts metadata like this in the first 512 bytes:
-        # MuData (format-version=0.1.0;creator=muon;creator-version=0.2.0)
-        # See mudata/_core/io.py, read_h5mu() function
-        starting_metadata = input_bytes.read(100)
-        # The metadata is padded with extra null bytes up until 512 bytes
-        truncate_location = starting_metadata.find(b"\x00")
-        starting_metadata = starting_metadata[:truncate_location]
-    with open(output_path, "br+") as f:
-        nbytes = f.write(starting_metadata)
-        f.write(b"\0" * (512 - nbytes)) 
-# END TEMPORARY WORKAROUND compress_h5mu
+from compress_h5mu import compress_h5mu
 
 if __name__ == "__main__":
     compress_h5mu(par["input"], par["output"], compression=par["compression"])

src/convert/from_10xh5_to_h5mu/script.py

@@ -16,22 +16,7 @@
 ## VIASH END
 
 sys.path.append(meta["resources_dir"])
-# START TEMPORARY WORKAROUND setup_logger
-# reason: resources aren't available when using Nextflow fusion
-# from setup_logger import setup_logger
-def setup_logger():
-    import logging
-    from sys import stdout
-
-    logger = logging.getLogger()
-    logger.setLevel(logging.INFO)
-    console_handler = logging.StreamHandler(stdout)
-    logFormatter = logging.Formatter("%(asctime)s %(levelname)-8s %(message)s")
-    console_handler.setFormatter(logFormatter)
-    logger.addHandler(console_handler)
-
-    return logger
-# END TEMPORARY WORKAROUND setup_logger
+from setup_logger import setup_logger
 logger = setup_logger()
 
 logger.info("Reading %s.", par["input"])

src/convert/from_10xmtx_to_h5mu/script.py

@@ -10,22 +10,7 @@
 ## VIASH END
 
 sys.path.append(meta["resources_dir"])
-# START TEMPORARY WORKAROUND setup_logger
-# reason: resources aren't available when using Nextflow fusion
-# from setup_logger import setup_logger
-def setup_logger():
-    import logging
-    from sys import stdout
-
-    logger = logging.getLogger()
-    logger.setLevel(logging.INFO)
-    console_handler = logging.StreamHandler(stdout)
-    logFormatter = logging.Formatter("%(asctime)s %(levelname)-8s %(message)s")
-    console_handler.setFormatter(logFormatter)
-    logger.addHandler(console_handler)
-
-    return logger
-# END TEMPORARY WORKAROUND setup_logger
+from setup_logger import setup_logger
 logger = setup_logger()
 
 logger.info("Reading %s.", par["input"])

src/convert/from_cellranger_multi_to_h5mu/script.py

@@ -25,22 +25,7 @@
 ## VIASH END
 
 sys.path.append(meta["resources_dir"])
-# START TEMPORARY WORKAROUND setup_logger
-# reason: resources aren't available when using Nextflow fusion
-# from setup_logger import setup_logger
-def setup_logger():
-    import logging
-    from sys import stdout
-
-    logger = logging.getLogger()
-    logger.setLevel(logging.INFO)
-    console_handler = logging.StreamHandler(stdout)
-    logFormatter = logging.Formatter("%(asctime)s %(levelname)-8s %(message)s")
-    console_handler.setFormatter(logFormatter)
-    logger.addHandler(console_handler)
-
-    return logger
-# END TEMPORARY WORKAROUND setup_logger
+from setup_logger import setup_logger
 logger = setup_logger()
 
 POSSIBLE_LIBRARY_TYPES = ('vdj_t', 'vdj_b', 'vdj_t_gd', 'count', 'antigen_analysis', 'multiplexing_analysis')

src/convert/from_h5ad_to_h5mu/script.py

@@ -13,22 +13,7 @@
 ## VIASH END
 
 sys.path.append(meta["resources_dir"])
-# START TEMPORARY WORKAROUND setup_logger
-# reason: resources aren't available when using Nextflow fusion
-# from setup_logger import setup_logger
-def setup_logger():
-    import logging
-    from sys import stdout
-
-    logger = logging.getLogger()
-    logger.setLevel(logging.INFO)
-    console_handler = logging.StreamHandler(stdout)
-    logFormatter = logging.Formatter("%(asctime)s %(levelname)-8s %(message)s")
-    console_handler.setFormatter(logFormatter)
-    logger.addHandler(console_handler)
-
-    return logger
-# END TEMPORARY WORKAROUND setup_logger
+from setup_logger import setup_logger
 logger = setup_logger()
 
 assert len(par["input"]) == len(par["modality"]), "Number of input files should be the same length as the number of modalities"

src/convert/from_h5mu_to_h5ad/script.py

@@ -11,22 +11,7 @@
 ## VIASH END
 
 sys.path.append(meta["resources_dir"])
-# START TEMPORARY WORKAROUND setup_logger
-# reason: resources aren't available when using Nextflow fusion
-# from setup_logger import setup_logger
-def setup_logger():
-    import logging
-    from sys import stdout
-
-    logger = logging.getLogger()
-    logger.setLevel(logging.INFO)
-    console_handler = logging.StreamHandler(stdout)
-    logFormatter = logging.Formatter("%(asctime)s %(levelname)-8s %(message)s")
-    console_handler.setFormatter(logFormatter)
-    logger.addHandler(console_handler)
-
-    return logger
-# END TEMPORARY WORKAROUND setup_logger
+from setup_logger import setup_logger
 logger = setup_logger()
 
 # TODO: Merge modalities into one layer