Parse sex information from R/qtl bundle.

1 files changed, 28 insertions, 7 deletions

diff --git a/r_qtl/r_qtl2.py b/r_qtl/r_qtl2.py
index 02217ee..2c1e162 100644
--- a/r_qtl/r_qtl2.py
+++ b/r_qtl/r_qtl2.py
@@ -134,6 +134,13 @@ def make_process_data_geno(cdata) -> tuple[
             for items in zip(ids, vals[1:]))
     return (__non_transposed__, __transposed__)
 
+def replace_sex_info(val, cdata: dict):
+    """Replace sex information in files with values in the control data."""
+    sex_info = cdata.get("sex", False)
+    if bool(sex_info):
+        return sex_info.get(val, val)
+    return val
+
 def replace_cross_info(val, cdata: dict):
     """
     Replace cross information in files with the values in the control data.
@@ -148,15 +155,11 @@ def make_process_data_covar(cdata) -> tuple[
         Callable[[str, tuple[str, ...], tuple[str, ...]],
                  tuple[dict, ...]]]:
     """Build functions to process sex and cross information in covar files."""
-    def replace_sex_code(val):
-        sex_info = cdata.get("sex", False)
-        if bool(sex_info):
-            return sex_info.get(val, val)
-        return val
+    rep_sex_info = partial(replace_sex_info, cdata=cdata)
     rep_cross_info = partial(replace_cross_info, cdata=cdata)
     def non_transposed(row: dict) -> dict:
         return {
-            key: thread_op(value, replace_sex_code, rep_cross_info)
+            key: thread_op(value, rep_sex_info, rep_cross_info)
             for key,value in row.items()
         }
     def transposed(id_key: str,
@@ -165,7 +168,7 @@ def make_process_data_covar(cdata) -> tuple[
         return tuple(
             dict(zip(
                 [id_key, vals[0]],
-                (thread_op(item, replace_sex_code, rep_cross_info)
+                (thread_op(item, rep_sex_info, rep_cross_info)
                  for item in items)))
             for items in zip(ids, vals[1:]))
     return (non_transposed, transposed)
@@ -221,3 +224,21 @@ def cross_information(zfile: ZipFile, cdata: dict) -> Iterator[dict]:
         yield {
             key: thread_op(value, partial(replace_cross_info, cdata=cdata))
             for key, value in row.items() if key not in sex_fields}
+
+def sex_information(zfile: ZipFile, cdata: dict) -> Iterator[dict]:
+    """Load cross information where present."""
+    cdata_sex_info = cdata.get("sex", {})
+    sex_info_file_key = "covar"
+    new_cdata = {**cdata}
+    ci_fields = (cdata.get("cross_info",{}).get("covar",""),)
+    if "file" in cdata_sex_info:
+        sex_info_file_key = "gnqc_sex_info_file"
+        new_cdata = {**cdata, "gnqc_sex_info_file": cdata_sex_info["file"]}
+
+    for row in file_data(zfile,
+                         sex_info_file_key,
+                         new_cdata,
+                         *make_process_data_covar(cdata)):
+        yield {
+            key: thread_op(value, partial(replace_sex_info, cdata=cdata))
+            for key, value in row.items() if key not in ci_fields}