about summary refs log tree commit diff
path: root/scripts/qc_on_rqtl2_bundle2.py
diff options
context:
space:
mode:
Diffstat (limited to 'scripts/qc_on_rqtl2_bundle2.py')
-rw-r--r--scripts/qc_on_rqtl2_bundle2.py44
1 files changed, 1 insertions, 43 deletions
diff --git a/scripts/qc_on_rqtl2_bundle2.py b/scripts/qc_on_rqtl2_bundle2.py
index 9136243..7e5d253 100644
--- a/scripts/qc_on_rqtl2_bundle2.py
+++ b/scripts/qc_on_rqtl2_bundle2.py
@@ -30,49 +30,7 @@ from r_qtl import fileerrors as rqfe
 from scripts.process_rqtl2_bundle import parse_job
 from scripts.redis_logger import setup_redis_logger
 from scripts.cli_parser import init_cli_parser, add_global_data_arguments
-
-
-def build_line_splitter(cdata: dict) -> Callable[[str], tuple[Union[str, None], ...]]:
-    """Build and return a function to use to split data in the files.
-
-    Parameters
-    ----------
-    cdata: A dict holding the control information included with the R/qtl2
-        bundle.
-
-    Returns
-    -------
-    A function that takes a string and return a tuple of strings.
-    """
-    separator = cdata["sep"]
-    na_strings = cdata["na.strings"]
-    def __splitter__(line: str) -> tuple[Union[str, None], ...]:
-        return tuple(
-            item if item not in na_strings else None
-            for item in
-            (field.strip() for field in line.strip().split(separator)))
-    return __splitter__
-
-
-def build_line_joiner(cdata: dict) -> Callable[[tuple[Union[str, None], ...]], str]:
-    """Build and return a function to use to split data in the files.
-
-    Parameters
-    ----------
-    cdata: A dict holding the control information included with the R/qtl2
-        bundle.
-
-    Returns
-    -------
-    A function that takes a string and return a tuple of strings.
-    """
-    separator = cdata["sep"]
-    na_strings = cdata["na.strings"]
-    def __joiner__(row: tuple[Union[str, None], ...]) -> str:
-        return separator.join(
-            (na_strings[0] if item is None else item)
-            for item in row)
-    return __joiner__
+from scripts.rqtl2.bundleutils import build_line_joiner, build_line_splitter
 
 
 def check_for_missing_files(