Add support for propagating/condensing slots. (#603)

Add two new methods to the MappingSetDataFrame class: `propagate` and
`condense`.

`propagate` implements propagation of propagatable slots: when such a
slot is present on the MappingSet but _not_ in individual records (i.e.
the data frame has no column for that slot), the value is "propagated"
down to the individual records. That is, a column for that slot is added
to the data frame, with the same value for all records, while the
original slot is removed from the set metadata.

`condense` performs the opposite operation: when a propagatable slot has
only a single value across all records, the value is set once and for
all in the set metadata and removed from the individual records (the
entire column is removed from the data frame).

Author: gouttegd

src/sssom/constants.py

@@ -268,6 +268,16 @@ def double_slots(self) -> Set[str]:
         """Return the slot names for SSSOMSchemaView object."""
         return {k for k, v in self.dict["slots"].items() if v["range"] == "double"}
 
+    @cached_property
+    def propagatable_slots(self) -> List[str]:
+        """Return the names of all propagatable slots."""
+        slots = []
+        for slot_name in self.mapping_set_slots:
+            annotations = self.view.annotation_dict(slot_name)
+            if annotations is not None and "propagated" in annotations:
+                slots.append(slot_name)
+        return slots
+
 
 @lru_cache(1)
 def _get_sssom_schema_object() -> SSSOMSchemaView:

src/sssom/util.py

@@ -58,6 +58,7 @@
     UNKNOWN_IRI,
     MetadataType,
     PathOrIO,
+    SSSOMSchemaView,
     _get_sssom_schema_object,
     get_default_metadata,
 )
@@ -299,6 +300,99 @@ def remove_mappings(self, msdf: "MappingSetDataFrame") -> None:
         self.df = self.df[self.df.columns.drop(list(self.df.filter(regex=r"_2")))]
         self.clean_prefix_map()
 
+    def propagate(self, fill_empty=False) -> List[str]:
+        """Propagate slot values from the set level down to individual records.
+
+        Propagation, as defined by the SSSOM specification, is the process by
+        which the values of so-called "propagatable slots" in the set metadata
+        are moved to the corresponding slots in each individual mapping
+        records.
+
+        Propagation of a slot is only allowed iff no individual records
+        already have a value for that slot.
+
+        :param fill_empty: If True, propagation of a slot is allowed even if
+                           some individual records already have a value for
+                           that slot. The set-level value will be propagated to
+                           all the records for which the slot is empty. Note
+                           that (1) this is not spec-compliant behaviour, and
+                           (2) this makes the operation non-reversible by a
+                           subsequent condensation.
+        :return: The list of slots that were effectively propagated.
+        """
+        schema = SSSOMSchemaView()
+        propagated = []
+
+        for slot in schema.propagatable_slots:
+            if slot not in self.metadata:  # Nothing to propagate
+                continue
+            is_present = slot in self.df.columns
+            if is_present and not fill_empty:
+                logging.warning(
+                    f"Not propagating value for '{slot}' because the slot is already set on individual records."
+                )
+                continue
+
+            if schema.view.get_slot(slot).multivalued:
+                value = "|".join(self.metadata.pop(slot))
+            else:
+                value = self.metadata.pop(slot)
+
+            if is_present:
+                self.df.loc[self.df[slot].eq("") | self.df[slot].isna(), slot] = value
+            else:
+                self.df[slot] = value
+            propagated.append(slot)
+
+        return propagated
+
+    def condense(self) -> List[str]:
+        """Condense record-level slot values to the set whenever possible.
+
+        Condensation is the opposite of propagation. It is the process by
+        which the values of so-called "propagatable" slots found in individual
+        mapping records are moved to the corresponding slots in the set
+        metadata.
+
+        Condensation of a slot is only allowed iff (1) all records have the
+        same value for that slot and (2) the slot does not already have a
+        different value in the set metadata.
+
+        :return: The list of slots that were effectively condensed.
+        """
+        schema = SSSOMSchemaView()
+        condensed = []
+
+        for slot in schema.propagatable_slots:
+            if slot not in self.df.columns:  # Nothing to condense
+                continue
+            values = self.df[slot].unique()
+            if len(values) > 1:
+                # Different values across the records, cannot condense
+                continue
+
+            if schema.view.get_slot(slot).multivalued:
+                value = values[0].split("|")
+            else:
+                value = values[0]
+
+            if slot in self.metadata:
+                if self.metadata[slot] != value:
+                    logging.warning(
+                        f"Not condensing slot '{slot}' because it already has a different value in the set metadata."
+                    )
+                    continue
+                # No need to set the condensed value in the set metadata as it
+                # is already there, but we must still remove the column from
+                # the dataframe
+                condensed.append(slot)
+            else:
+                self.metadata[slot] = value
+                condensed.append(slot)
+
+        self.df.drop(columns=condensed, inplace=True)
+        return condensed
+
 
 def _standardize_curie_or_iri(curie_or_iri: str, *, converter: Converter) -> str:
     """Standardize a CURIE or IRI, returning the original if not possible.

tests/data/propagatable.tsv

@@ -0,0 +1,22 @@
+#curie_map:
+#  c: http://example.org/c/
+#  orcid: https://orcid.org/
+#  x: http://example.org/x/
+#  y: http://example.org/y/
+#  z: http://example.org/z/
+#mapping_set_id: https://w3id.org/sssom/mapping/tests/data/propagatable.tsv
+#creator_id:
+#  - orcid:1234
+#  - orcid:5678
+#license: https://creativecommons.org/publicdomain/zero/1.0/
+#mapping_provider: https://example.org/mappings
+#mapping_tool: https://github.com/cmungall/rdf_matcher
+#subject_preprocessing:
+#  - c:rule1
+#  - c:rule2
+subject_id	predicate_id	object_id	mapping_justification	mapping_tool
+x:appendage	owl:equivalentClass	y:appendage	semapv:ManualMappingCuration	
+x:appendage	owl:equivalentClass	z:appendage	semapv:LexicalMatching	
+x:appendage	owl:equivalentClass	z:appendage	semapv:ManualMappingCuration	foo matcher
+x:bone_element	owl:equivalentClass	y:bone	semapv:LexicalMatching	
+x:bone_element	owl:equivalentClass	y:bone	semapv:ManualMappingCuration	

tests/test_utils.py

@@ -522,3 +522,76 @@ def test_get_file_extension(self) -> None:
             with self.subTest(path=path, mode="file"), tempfile.TemporaryDirectory() as d:
                 with Path(d).joinpath(part).open("w") as file:
                     self.assertEqual(value, get_file_extension(file))
+
+    def test_propagation_and_condensation(self) -> None:
+        """Test propagating/condensing values of propagatable slots."""
+        msdf = parse_sssom_table(f"{data_dir}/propagatable.tsv")
+
+        propagated_slots = msdf.propagate()
+        # creator_id is not a propagatable slot
+        self.assertNotIn("creator_id", propagated_slots)
+        self.assertNotIn("creator_id", msdf.df.columns)
+        # mapping_tool has values for some records and should not be propagated
+        self.assertNotIn("mapping_tool", propagated_slots)
+        # mapping_provider should be propagated
+        self.assertIn("mapping_provider", propagated_slots)
+        self.assertIn("mapping_provider", msdf.df.columns)
+        self.assertNotIn("mapping_provider", msdf.metadata)
+        # Ditto for subject_preprocessing
+        self.assertIn("subject_preprocessing", propagated_slots)
+        self.assertIn("subject_preprocessing", msdf.df.columns)
+        self.assertNotIn("subject_preprocessing", msdf.metadata)
+
+        propagated_slots = msdf.propagate()
+        # Set has been propagated already, no further propagation possible
+        self.assertEqual(0, len(propagated_slots))
+
+        condensed_slots = msdf.condense()
+        # mapping_tool has not a unique value and should not be condensed
+        self.assertNotIn("mapping_tool", condensed_slots)
+        self.assertIn("mapping_tool", msdf.df.columns)
+        # mapping_provider should be condensed back
+        self.assertIn("mapping_provider", condensed_slots)
+        self.assertNotIn("mapping_provider", msdf.df.columns)
+        self.assertIn("mapping_provider", msdf.metadata)
+        # Ditto for subject_preprocessing
+        self.assertIn("subject_preprocessing", condensed_slots)
+        self.assertNotIn("subject_preprocessing", msdf.df.columns)
+        self.assertIn("subject_preprocessing", msdf.metadata)
+
+        condensed_slots = msdf.condense()
+        # Set has been condensed already, no further condensation possible
+        self.assertEqual(0, len(condensed_slots))
+
+    def test_condensation_with_existing_set_values(self) -> None:
+        """Test that condensation works as expected with the mapping set already contains values for the to-be-condensed slots."""
+        msdf = parse_sssom_table(f"{data_dir}/propagatable.tsv")
+        msdf.propagate()
+        # Following propagation, all records in msdf have the same
+        # mapping_provider ("https://example.org/mappings/)"
+
+        # Inject a different mapping_provider value in the set metadata;
+        # this should prevent that slot from being condensed back
+        msdf.metadata["mapping_provider"] = "https://example.org/mappings/2"
+        condensed_slots = msdf.condense()
+        self.assertNotIn("mapping_provider", condensed_slots)
+        self.assertIn("mapping_provider", msdf.df.columns)
+        self.assertEqual("https://example.org/mappings/2", msdf.metadata["mapping_provider"])
+
+        # Inject the same mapping_provider value as the one contained in
+        # the records; this should allow the slot to be condensed
+        msdf.metadata["mapping_provider"] = "https://example.org/mappings"
+        condensed_slots = msdf.condense()
+        self.assertIn("mapping_provider", condensed_slots)
+        self.assertNotIn("mapping_provider", msdf.df.columns)
+        self.assertEqual("https://example.org/mappings", msdf.metadata["mapping_provider"])
+
+    def test_propagation_fill_empty_mode(self) -> None:
+        """Test propagate with fill_empty=True."""
+        msdf = parse_sssom_table(f"{data_dir}/propagatable.tsv")
+
+        propagated_slots = msdf.propagate(fill_empty=True)
+        # mapping_tool should have been propagated
+        self.assertIn("mapping_tool", propagated_slots)
+        self.assertNotIn("mapping_tool", msdf.metadata)
+        self.assertEqual(2, len(msdf.df["mapping_tool"].unique()))