add support for writing --split and --combined

Author: nebfield

pgscatalog.matchapp/src/pgscatalog/matchapp/_write.py

@@ -3,6 +3,9 @@
 It expects Config class attributes to be set up before being called
 """
 import gzip
+import itertools
+
+from pgscatalog.matchlib.plinkscorefiles import PlinkScoreFiles
 
 from ._config import Config
 
@@ -11,11 +14,19 @@ def write_matches(matchresults, score_df):
     """Write matchresults out to scoring files and logs"""
     match (Config.SPLIT, Config.COMBINED):
         case (True, True):
-            # requires extra work: first write split, then re-combine without recomputing matches
-            raise NotImplementedError
+            # requires extra work: first write split
+            outfs = matchresults.write_scorefiles(
+                directory=Config.OUTDIR,
+                split=True,
+                score_df=score_df,
+                min_overlap=Config.MIN_OVERLAP,
+                **Config.MATCH_PARAMS,
+            )
+            # now re-combine without recomputing matches
+            PlinkScoreFiles(*list(itertools.chain(*outfs))).merge(Config.OUTDIR)
         case (True, False) | (False, True):
             # split parameter can handle this case OK
-            matchresults.write_scorefiles(
+            _ = matchresults.write_scorefiles(
                 directory=Config.OUTDIR,
                 split=Config.SPLIT,
                 score_df=score_df,

pgscatalog.matchapp/tests/test_merge_cli.py

@@ -87,6 +87,38 @@ def test_split_output(tmp_path_factory, good_scorefile, match_ipc):
     assert sum("recessive" in x for x in splitf) == 1
 
 
+def test_splitcombine_output(tmp_path_factory, good_scorefile, match_ipc):
+    """Test merging runs without errors with good data outputting both combined and split scorefiles"""
+    outdir = tmp_path_factory.mktemp("outdir")
+
+    args = [
+        (
+            "pgscatalog-matchmerge",
+            "-d",
+            "test",
+            "-s",
+            str(good_scorefile),
+            "--matches",
+            str(match_ipc),
+            "--outdir",
+            str(outdir),
+            "--min_overlap",
+            str(0.75),
+            "--split",
+            "--combined",
+        )
+    ]
+    flargs = list(itertools.chain(*args))
+
+    with patch("sys.argv", flargs):
+        run_merge()
+    splitf = glob(str(outdir / "*scorefile.gz"))
+    assert len(splitf) == 21 + 3  # split + combined
+    assert sum("additive" in x for x in splitf) == 19 + 1
+    assert sum("dominant" in x for x in splitf) == 1 + 1
+    assert sum("recessive" in x for x in splitf) == 1 + 1
+
+
 def test_strict_merge(tmp_path_factory, good_scorefile, match_ipc):
     """Test merging with extremely strict overlap to trigger a MatchRateError"""
     outdir = tmp_path_factory.mktemp("outdir")

pgscatalog.matchlib/src/pgscatalog/matchlib/_plinkframe.py

@@ -37,6 +37,7 @@ def pivot_wide(self):
 
     def write(self, directory, dataset, split=False):
         """Write a plink2 --score compatible file, optionally splitting"""
+        fouts = []
         if split:
             dfs = self.split_pivot()
             for chrom, df in dfs.items():
@@ -46,6 +47,7 @@ def write(self, directory, dataset, split=False):
                 )
                 with gzip.open(fout, "wb") as f:
                     df.write_csv(f, separator="\t")
+                fouts.append(fout)
         else:
             chrom = "ALL"
             fout = (
@@ -55,6 +57,8 @@ def write(self, directory, dataset, split=False):
             df = self.pivot_wide()
             with gzip.open(fout, "wb") as f:
                 df.write_csv(f, separator="\t")
+            fouts.append(fout)
+        return fouts
 
 
 class PlinkFrames(collections.abc.Sequence):

pgscatalog.matchlib/src/pgscatalog/matchlib/matchresult.py

@@ -122,8 +122,8 @@ class MatchResults(collections.abc.Sequence):
 
     >>> with scorefile as score_df:
     ...     x = MatchResult.from_ipc(fout.name, dataset="goodmatch")
-    ...     MatchResults(x).write_scorefiles(directory=foutdir, score_df=score_df)
-    ...     MatchResults(x).write_scorefiles(directory=splitfoutdir, split=True, score_df=score_df)
+    ...     _ = MatchResults(x).write_scorefiles(directory=foutdir, score_df=score_df)
+    ...     _ = MatchResults(x).write_scorefiles(directory=splitfoutdir, split=True, score_df=score_df)
     >>> MatchResults(x)  # doctest: +ELLIPSIS
     MatchResults([MatchResult(dataset=goodmatch, matchresult=None, ipc_path=...])
 
@@ -275,12 +275,14 @@ def write_scorefiles(
         self._log_OK = check_log_count(scorefile=score_df, summary_log=self.summary_log)
 
         plink = PlinkFrames.from_matchresult(self.df)
-
+        outfs = []
         for frame in plink:
-            frame.write(directory=directory, split=split, dataset=self.dataset)
+            f = frame.write(directory=directory, split=split, dataset=self.dataset)
+            outfs.append(f)
 
         # collect after joining in check_log_count (can't join df and lazy df)
         self.summary_log = self.summary_log.collect()
+        return outfs
 
     def full_variant_log(self, score_df, **kwargs):
         """Generate a log for each variant in a scoring file

pgscatalog.matchlib/src/pgscatalog/matchlib/plinkscorefiles.py

@@ -45,7 +45,7 @@ def merge(self, directory):
         >>> x = MatchResult.from_ipc(fout.name, dataset="goodmatch")
         >>> foutdir = tempfile.mkdtemp()
         >>> with scorefile as score_df:
-        ...     MatchResults(x).write_scorefiles(directory=foutdir, split=True, score_df=score_df)  # doctest: +ELLIPSIS
+        ...     _ = MatchResults(x).write_scorefiles(directory=foutdir, split=True, score_df=score_df)  # doctest: +ELLIPSIS
         >>> plink_files = (pathlib.Path(foutdir) / x for x in os.listdir(foutdir))
         >>> psf = PlinkScoreFiles(*plink_files)
         >>> psf  # doctest: +ELLIPSIS