Extract common functions.

2 files changed, 45 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(
diff --git a/scripts/rqtl2/bundleutils.py b/scripts/rqtl2/bundleutils.py
new file mode 100644
index 0000000..17faa7c
--- /dev/null
+++ b/scripts/rqtl2/bundleutils.py
@@ -0,0 +1,44 @@
+"""Common utilities to operate in R/qtl2 bundles."""
+from typing import Union, Callable
+
+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__