From cd0d415e691f46941fc3bd4a132ed3bdacd97931 Mon Sep 17 00:00:00 2001
From: Frederick Muriuki Muriithi
Date: Thu, 17 Oct 2024 12:42:34 -0500
Subject: Extract common functions.

---
 scripts/qc_on_rqtl2_bundle2.py | 44 +-----------------------------------------
 1 file changed, 1 insertion(+), 43 deletions(-)

(limited to 'scripts/qc_on_rqtl2_bundle2.py')

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(
-- 
cgit 1.4.1