Add sdk.utils.compare_metadata

CHANGELOG.rst

@@ -14,6 +14,7 @@ Changelog
 * Add new method :meth:`sdk.utils.Archive.check_metadata`.
 * Add new error ``sdk.utils.IncorrectMetadataError``.
 * Update :command:`run-ngs-pipeline` command to check input files more vigorously.
+* Add new method :meth:`sdk.utils.compare_metadata`.
 
 0.8.0 (2021-11-20)
 ------------------

pypgx/api/utils.py

@@ -1161,8 +1161,6 @@ def train_cnv_caller(copy_number, cnv_calls, confusion_matrix=None):
 
     cnv_calls.check_type('SampleTable[CNVCalls]')
 
-    copy_number = _process_copy_number(copy_number)
-
     gene1 = copy_number.metadata['Gene']
     gene2 = cnv_calls.metadata['Gene']
 
@@ -1177,6 +1175,8 @@ def train_cnv_caller(copy_number, cnv_calls, confusion_matrix=None):
         raise ValueError("Different genes detected, CovFrame[CopyNumber] "
             f"has '{assembly1}' and SampleTable[CNVCalls] has '{assembly2}'")
 
+    copy_number = _process_copy_number(copy_number)
+
     cnv_table = core.load_cnv_table()
     cnv_table = cnv_table[cnv_table.Gene == copy_number.metadata['Gene']]
     name2code = dict(zip(cnv_table.Name, cnv_table.Code))

pypgx/sdk/__init__.py

@@ -1,3 +1,3 @@
-from .utils import (Archive, parse_input_bams)
+from .utils import (Archive, parse_input_bams, compare_metadata)
 
-__all__ = ['Archive', 'parse_input_bams']
+__all__ = ['Archive', 'parse_input_bams', 'compare_metadata']

pypgx/sdk/utils.py

@@ -129,6 +129,20 @@ def check_metadata(self, key, value):
                 f"Expected '{key}={value}' but found '{key}={actual_value}' "
                 f"for semantic type '{semantic_type}'")
 
+def compare_metadata(a, b, key):
+    """
+    Raise IncorrectMetadataError if two archives have different values for
+    specified metadata key.
+    """
+    type1 = a.metadata['SemanticType']
+    type2 = a.metadata['SemanticType']
+    value1 = a.metadata[key]
+    value2 = b.metadata[key]
+    if value1 != value2:
+        raise IncorrectMetadataError(
+            f"First archive '{type1}' has '{key}={value1}' and second "
+            f"archive '{type2}' has '{key}={value2}'")
+
 def zipdir(dir, output):
     """
     Create a ZIP archive of a directory.