Add method to infer minimum compatible SSSOM version.

src/sssom/constants.py

@@ -10,6 +10,7 @@
 import yaml
 from linkml_runtime.utils.schema_as_dict import schema_as_dict
 from linkml_runtime.utils.schemaview import SchemaView
+from sssom_schema.datamodel.sssom_schema import SssomVersionEnum
 
 HERE = pathlib.Path(__file__).parent.resolve()
 
@@ -278,6 +279,27 @@ def propagatable_slots(self) -> List[str]:
                 slots.append(slot_name)
         return slots
 
+    def get_minimum_version(self, slot_name: str, class_name: str = "mapping") -> SssomVersionEnum:
+        """Get the minimum version of SSSOM required for a given slot.
+
+        :param slot_name: The queried slot.
+        :param class_name: The class the slot belongs to. This is needed
+                           because a slot may have been added to a class
+                           in a later version than the version in which
+                           it was first introduced in the schema.
+        :return: A SssomVersionEnum value representing the earliest
+                 version of SSSOM that defines the given slot in the
+                 given class. May be None if the requested slot name
+                 is not a valid slot name.
+        """
+        try:
+            slot = self.view.induced_slot(slot_name, class_name)
+            return SssomVersionEnum(slot.annotations.added_in.value)
+        except AttributeError:  # No added_in annotation, defaults to 1.0
+            return SssomVersionEnum("1.0")
+        except ValueError:  # No such slot
+            return None
+
 
 @lru_cache(1)
 def _get_sssom_schema_object() -> SSSOMSchemaView:

src/sssom/util.py

@@ -393,6 +393,56 @@ def condense(self) -> List[str]:
         self.df.drop(columns=condensed, inplace=True)
         return condensed
 
+    def get_compatible_version(self) -> str:
+        """Get the minimum version of SSSOM this set is compatible with."""
+        schema = SSSOMSchemaView()
+        versions = set()
+
+        # First get the minimum versions required by the slots present
+        # in the set; this is entirely provided by the SSSOM model.
+        for slot in self.metadata.keys():
+            version = schema.get_minimum_version(slot, "mapping set")
+            if version is not None:
+                versions.add(str(version))
+        for slot in self.df.columns:
+            version = schema.get_minimum_version(slot, "mapping")
+            if version is not None:
+                versions.add(str(version))
+
+        # Then take care of enum values; we cannot use the SSSOM model
+        # for that (enum values are not tagged with an "added_in"
+        # annotation the way slots are), so this has to be handled
+        # "manually" based on the informations provided in
+        # <https://mapping-commons.github.io/sssom/spec-model/#model-changes-across-versions>.
+        if (
+            self.metadata.get("subject_type") == "composed entity expression"
+            or self.metadata.get("subject_type") == "composed entity expression"
+            or (
+                "subject_type" in self.df.columns
+                and "composed entity expression" in self.df["subject_type"].values
+            )
+            or (
+                "object_type" in self.df.columns
+                and "composed entity expression" in self.df["object_type"].values
+            )
+        ):
+            versions.add("1.1")
+
+        if (
+            "mapping_cardinality" in self.df.columns
+            and "0:0" in self.df["mapping_cardinality"].values
+        ):
+            versions.add("1.1")
+
+        # Get the highest of the accumulated versions. We do a numerical
+        # sort, so that version 1.10 (if we ever get that far in the 1.x
+        # branch) does not get sorted before version 1.9.
+        def _version_to_compare_key(version):
+            major, minor = [int(s) for s in version.split(".")]
+            return (major * 100) + minor
+
+        return sorted(versions, key=_version_to_compare_key)[-1]
+
 
 def _standardize_curie_or_iri(curie_or_iri: str, *, converter: Converter) -> str:
     """Standardize a CURIE or IRI, returning the original if not possible.

tests/test_utils.py

@@ -595,3 +595,40 @@ def test_propagation_fill_empty_mode(self) -> None:
         self.assertIn("mapping_tool", propagated_slots)
         self.assertNotIn("mapping_tool", msdf.metadata)
         self.assertEqual(2, len(msdf.df["mapping_tool"].unique()))
+
+    def test_inferring_compatible_version(self) -> None:
+        """Test that we can correctly infer the version a set is compatible with."""
+        msdf10 = parse_sssom_table(f"{data_dir}/basic.tsv")
+
+        # Nothing in that set requires 1.1
+        self.assertEqual("1.0", msdf10.get_compatible_version())
+
+        def _clone(msdf):
+            return MappingSetDataFrame(df=msdf.df.copy(), metadata=msdf.metadata.copy())
+
+        # Inject a 1.1-specific mapping set slot
+        msdf11 = _clone(msdf10)
+        msdf11.metadata["cardinality_scope"] = "predicate_id"
+        self.assertEqual("1.1", msdf11.get_compatible_version())
+
+        # Inject a 1.1-specific mapping slot
+        msdf11 = _clone(msdf10)
+        msdf11.df["predicate_type"] = "owl object property"
+        self.assertEqual("1.1", msdf11.get_compatible_version())
+
+        # Inject a 1.1-specific entity_type_enum value
+        msdf11 = _clone(msdf10)
+        msdf11.metadata["subject_type"] = "composed entity expression"
+        self.assertEqual("1.1", msdf11.get_compatible_version())
+
+        # Same, but on a single mapping record
+        msdf11 = _clone(msdf10)
+        msdf11.df["object_type"] = "owl class"
+        msdf11.df.loc[2, "object_type"] = "composed entity expression"
+        self.assertEqual("1.1", msdf11.get_compatible_version())
+
+        # Inject the 1.1-specific "0:0" cardinality value
+        msdf11 = _clone(msdf10)
+        msdf11.df["mapping_cardinality"] = "1:1"
+        msdf11.df.loc[9, "mapping_cardinality"] = "0:0"
+        self.assertEqual("1.1", msdf11.get_compatible_version())