Merge remote-tracking branch 'origin/develop' into ontology-save-formats

alkidbaci · alkidbaci · commit 11585bf0dc90 · 2026-04-28T10:33:28.000+02:00
diff --git a/README.md b/README.md
@@ -388,6 +388,41 @@ justifications = reasoner.create_axiom_justifications(axiom)
 </details>
 
 
+### Get Contrastive Explanations
+
+<details><summary> Click me!</summary>
+
+```python
+from owlapy import manchester_to_owl_expression
+from owlapy.iri import IRI
+from owlapy.owl_individual import OWLNamedIndividual
+from owlapy.owl_ontology import SyncOntology
+from owlapy.owl_reasoner import SyncReasoner
+
+# --- Load ontology and reasoner ---
+ontology = SyncOntology("../KGs/Family/family.owl")
+reasoner = SyncReasoner(ontology, reasoner="HermiT")
+
+# --- Define class expression ---
+class_expr = manchester_to_owl_expression(
+    "Sister and (hasSibling some (married some (hasChild some Grandchild)))",
+    "http://www.benchmark.org/family#"
+)
+
+# --- Define individuals ---
+fact = OWLNamedIndividual(IRI.create("http://www.benchmark.org/family#F9F143"))
+foil = OWLNamedIndividual(IRI.create("http://www.benchmark.org/family#F9M161"))
+
+# --- Get contrastive explanation ---
+result = reasoner.get_contrastive_explanation(class_expr, fact, foil)
+
+# --- Print results ---
+for k in ["common", "different", "conflict"]:
+    print(f"{k.capitalize()} axioms: {result[k]}")
+```
+</details>
+
+
 ### Class expression simplification
 
 <details><summary> Click me!</summary>
diff --git a/owlapy/converter.py b/owlapy/converter.py
@@ -132,8 +132,8 @@ def convert(self, root_variable: str,
         self.for_all_de_morgan = for_all_de_morgan
         self.named_individuals = named_individuals
         # # if named_individuals is True, we return only entities that are instances of owl:NamedIndividual
-        # if named_individuals:
-        #     self.append_triple(root_variable, 'a', f"<{OWLRDFVocabulary.OWL_NAMED_INDIVIDUAL.as_str()}>")
+        if named_individuals:
+            self.append_triple(root_variable, 'a', f"<{OWLRDFVocabulary.OWL_NAMED_INDIVIDUAL.as_str()}>")
         with self.stack_variable(root_variable):
             with self.stack_parent(ce):
                 self.process(ce)
@@ -270,11 +270,7 @@ def _(self, ce: OWLObjectComplementOf):
         # the exclusion of "?x ?p ?o" results in the group graph pattern to just return true or false (not bindings)
         # as a result, we need to comment out the if-clause of the following line
         # if not self.in_intersection and self.modal_depth == 1:
-        # if namedIndividual is set to True, do not use variables --> restrict the subject to instances of NamedIndividual
-        if self.named_individuals:
-            self.append_triple(subject, "a", f"<{OWLRDFVocabulary.OWL_NAMED_INDIVIDUAL.as_str()}>")
-        else:
-            self.append_triple(subject, self.mapping.new_individual_variable(), self.mapping.new_individual_variable())
+        self.append_triple(subject, self.mapping.new_individual_variable(), self.mapping.new_individual_variable())
 
         self.append("FILTER NOT EXISTS { ")
         # process the concept after the ¬
@@ -557,7 +553,9 @@ def _(self, ce: OWLDataHasValue):
     @process.register
     def _(self, node: OWLDatatype):
         if node != TopOWLDatatype:
-            self.append(f" FILTER ( DATATYPE ( {self.current_variable} = <{node.to_string_id()}> ) ) ")
+            self.append(f" FILTER ( DATATYPE ( {self.current_variable} ) = <{node.to_string_id()}> ) ")
+        else:
+            self.append(f" FILTER ( isLiteral ( {self.current_variable} ) ")
 
     @process.register
     def _(self, node: OWLDataOneOf):
@@ -627,6 +625,8 @@ def as_query(self,
                 q.append(f"<{x.to_string_id()}>")
             q.append("} . ")
             qs.extend(q)
+        if named_individuals:
+            qs.append(f"{root_variable} a <{OWLRDFVocabulary.OWL_NAMED_INDIVIDUAL.as_str()}> . ")
         qs.extend(tp)
         qs.append(" }")
 
@@ -643,10 +643,12 @@ def as_confusion_matrix_query(self,
                                   for_all_de_morgan: bool = True,
                                   named_individuals: bool = False) -> str:
         # get the graph pattern corresponding to the provided class expression (ce)
-        graph_pattern_str = "".join(self.convert(root_variable,
-                                                 ce,
-                                                 for_all_de_morgan=for_all_de_morgan,
-                                                 named_individuals=named_individuals))
+        tp = self.convert(root_variable, ce, for_all_de_morgan=for_all_de_morgan, named_individuals=named_individuals)
+        if named_individuals:
+            named_individual_triple = f"{root_variable} a <{OWLRDFVocabulary.OWL_NAMED_INDIVIDUAL.as_str()}> . "
+            graph_pattern_str = named_individual_triple + "".join(tp)
+        else:
+            graph_pattern_str = "".join(tp)
         # preparation for the final query
 
         # required to compute false negatives
diff --git a/owlapy/owl_reasoner.py b/owlapy/owl_reasoner.py
@@ -18,7 +18,7 @@
     OWLObjectIntersectionOf, OWLObjectComplementOf, OWLObjectAllValuesFrom, OWLObjectOneOf, OWLObjectHasValue, \
     OWLObjectMinCardinality, OWLObjectMaxCardinality, OWLObjectExactCardinality, OWLObjectCardinalityRestriction, \
     OWLDataSomeValuesFrom, OWLDataOneOf, OWLDatatypeRestriction, OWLFacetRestriction, OWLDataHasValue, \
-    OWLDataAllValuesFrom, OWLNothing, OWLThing
+    OWLDataAllValuesFrom, OWLNothing, OWLThing, OWLDataMinCardinality, OWLDataMaxCardinality, OWLDataExactCardinality
 from owlapy.class_expression import OWLClass
 from owlapy.iri import IRI
 from owlapy.owl_axiom import OWLAxiom, OWLSubClassOfAxiom
@@ -989,6 +989,183 @@ def _lazy_cache_class(self, c: OWLClass) -> None:
         temp = self.get_instances_from_owl_class(c)
         self._cls_to_ind[c] = frozenset(temp)
 
+    @_find_instances.register
+    def _(self, ce: OWLDataMinCardinality):
+        return self._get_instances_data_card_restriction(ce)
+
+    @_find_instances.register
+    def _(self, ce: OWLDataMaxCardinality):
+        all_ = frozenset(self._ontology.individuals_in_signature())
+        min_ind = self._get_instances_data_card_restriction(
+            OWLDataMinCardinality(cardinality=ce.get_cardinality() + 1,
+                                  property=ce.get_property(),
+                                  filler=ce.get_filler()))
+        return all_ ^ min_ind
+
+    @_find_instances.register
+    def _(self, ce: OWLDataExactCardinality):
+        return self._get_instances_data_card_restriction(ce)
+
+    def _get_instances_data_card_restriction(self, ce) -> FrozenSet[OWLNamedIndividual]:
+        """Get instances for OWLDataMinCardinality or OWLDataExactCardinality restrictions."""
+        pe = ce.get_property()
+        filler = ce.get_filler()
+        assert isinstance(pe, OWLDataProperty)
+
+        if isinstance(ce, OWLDataMinCardinality):
+            min_count = ce.get_cardinality()
+            max_count = None
+        elif isinstance(ce, OWLDataExactCardinality):
+            min_count = max_count = ce.get_cardinality()
+        else:
+            raise NotImplementedError
+
+        assert min_count >= 0
+        assert max_count is None or max_count >= 0
+
+        if self._ontology.is_modified and (self.class_cache or self._property_cache):
+            self.reset_and_disable_cache()
+        property_cache = self._property_cache
+
+        if property_cache:
+            self._lazy_cache_data_prop(pe)
+            dps = self._data_prop[pe]
+        else:
+            subs = self._some_values_subject_index(pe)
+
+        ind = set()
+
+        if isinstance(filler, OWLDatatype):
+            if property_cache:
+                for s, literals in dps.items():
+                    count = sum(1 for lit in literals if lit.get_datatype() == filler)
+                    if count >= min_count and (max_count is None or count <= max_count):
+                        ind.add(s)
+            else:
+                for s in subs:
+                    count = sum(1 for lit in self.data_property_values(s, pe) if lit.get_datatype() == filler)
+                    if count >= min_count and (max_count is None or count <= max_count):
+                        ind.add(s)
+        elif isinstance(filler, OWLDataOneOf):
+            values = set(filler.values())
+            if property_cache:
+                for s, literals in dps.items():
+                    count = len(literals & values)
+                    if count >= min_count and (max_count is None or count <= max_count):
+                        ind.add(s)
+            else:
+                for s in subs:
+                    count = sum(1 for lit in self.data_property_values(s, pe) if lit in values)
+                    if count >= min_count and (max_count is None or count <= max_count):
+                        ind.add(s)
+        elif isinstance(filler, OWLDatatypeRestriction):
+            def res_to_callable(res: OWLFacetRestriction):
+                op = res.get_facet().operator
+                v = res.get_facet_value()
+
+                def inner(lv: OWLLiteral):
+                    return op(lv, v)
+
+                return inner
+
+            apply = FunctionType.__call__
+            facet_restrictions = tuple(map(res_to_callable, filler.get_facet_restrictions()))
+
+            def include(lv: OWLLiteral):
+                return lv.get_datatype() == filler.get_datatype() and \
+                    all(map(apply, facet_restrictions, repeat(lv)))
+
+            if property_cache:
+                for s, literals in dps.items():
+                    count = sum(1 for lit in literals if include(lit))
+                    if count >= min_count and (max_count is None or count <= max_count):
+                        ind.add(s)
+            else:
+                for s in subs:
+                    count = sum(1 for lit in self.data_property_values(s, pe) if include(lit))
+                    if count >= min_count and (max_count is None or count <= max_count):
+                        ind.add(s)
+        elif isinstance(filler, OWLDataComplementOf):
+            # Count values NOT matching the inner data range
+            inner_filler = filler.get_data_range()
+            # some_inner = OWLDataSomeValuesFrom(property=pe, filler=inner_filler)
+            # We need to count per-individual, so we iterate
+            if property_cache:
+                for s, literals in dps.items():
+                    # Count literals that match the complement (i.e., do NOT match the inner filler)
+                    match_inner = self._count_data_filler_matches(literals, inner_filler)
+                    count = len(literals) - match_inner
+                    if count >= min_count and (max_count is None or count <= max_count):
+                        ind.add(s)
+            else:
+                for s in subs:
+                    all_lits = set(self.data_property_values(s, pe))
+                    match_inner = self._count_data_filler_matches(all_lits, inner_filler)
+                    count = len(all_lits) - match_inner
+                    if count >= min_count and (max_count is None or count <= max_count):
+                        ind.add(s)
+        elif isinstance(filler, OWLDataUnionOf):
+            if property_cache:
+                for s, literals in dps.items():
+                    count = sum(1 for lit in literals
+                                if any(self._literal_matches_data_range(lit, op)
+                                       for op in filler.operands()))
+                    if count >= min_count and (max_count is None or count <= max_count):
+                        ind.add(s)
+            else:
+                for s in subs:
+                    count = sum(1 for lit in self.data_property_values(s, pe)
+                                if any(self._literal_matches_data_range(lit, op)
+                                       for op in filler.operands()))
+                    if count >= min_count and (max_count is None or count <= max_count):
+                        ind.add(s)
+        elif isinstance(filler, OWLDataIntersectionOf):
+            if property_cache:
+                for s, literals in dps.items():
+                    count = sum(1 for lit in literals
+                                if all(self._literal_matches_data_range(lit, op)
+                                       for op in filler.operands()))
+                    if count >= min_count and (max_count is None or count <= max_count):
+                        ind.add(s)
+            else:
+                for s in subs:
+                    count = sum(1 for lit in self.data_property_values(s, pe)
+                                if all(self._literal_matches_data_range(lit, op)
+                                       for op in filler.operands()))
+                    if count >= min_count and (max_count is None or count <= max_count):
+                        ind.add(s)
+        else:
+            raise ValueError
+
+        return frozenset(ind)
+
+    def _literal_matches_data_range(self, lit: OWLLiteral, dr) -> bool:
+        """Check if a single literal matches a data range."""
+        if isinstance(dr, OWLDatatype):
+            return lit.get_datatype() == dr
+        elif isinstance(dr, OWLDataOneOf):
+            return lit in set(dr.values())
+        elif isinstance(dr, OWLDatatypeRestriction):
+            if lit.get_datatype() != dr.get_datatype():
+                return False
+            for res in dr.get_facet_restrictions():
+                op = res.get_facet().operator
+                if not op(lit, res.get_facet_value()):
+                    return False
+            return True
+        elif isinstance(dr, OWLDataComplementOf):
+            return not self._literal_matches_data_range(lit, dr.get_data_range())
+        elif isinstance(dr, OWLDataUnionOf):
+            return any(self._literal_matches_data_range(lit, op) for op in dr.operands())
+        elif isinstance(dr, OWLDataIntersectionOf):
+            return all(self._literal_matches_data_range(lit, op) for op in dr.operands())
+        else:
+            raise ValueError(f"Unsupported data range type: {type(dr)}")
+
+    def _count_data_filler_matches(self, literals: Set[OWLLiteral], dr) -> int:
+        """Count how many literals in the set match the given data range."""
+        return sum(1 for lit in literals if self._literal_matches_data_range(lit, dr))
+
     def get_instances_from_owl_class(self, c: OWLClass):
         if c.is_owl_thing():
             yield from self._ontology.individuals_in_signature()
diff --git a/owlapy/utils.py b/owlapy/utils.py
@@ -952,28 +952,32 @@ def _merge_card_r_with_same_body(self, restriction, nary_ce = None):
                                  filler=self._simplify(restriction.get_filler()))
 
     def _process_cardinality_restriction(self, restriction, nary_ce = None):
-        # Check for card restrictions that have the share the same cardinality and property.
-        # They can be simplified into a single card restriction with a merged filler.
-        # E.g.:
-        # (> 1 r.A) ⊔ (> 1 r.B) ≡ > 1 r.(A ⊔ B)
-        # (< 2 r.xsd:boolean) ⊔ (< 2 r.xsd:integer) ≡ < 2 r.(xsd:boolean ⊔ xsd:integer)
+        # Check for card restrictions that share the same cardinality and property.
+        # They can be simplified into a single card restriction with a merged filler,
+        # but only for scenarios where the equivalence actually holds:
+        #
+        # For OWLObjectMinCardinality / OWLDataMinCardinality with cardinality = 1 in a union:
+        #   (≥ 1 r.A) ⊔ (≥ 1 r.B) ≡ ≥ 1 r.(A ⊔ B)   ← valid (equivalent to ∃r)
         if isinstance(nary_ce, OWLObjectUnionOf):
-            operands = set(nary_ce.operands())
-            same_root = []
-            for op in operands:
-                if (isinstance(op, type(restriction))
-                        and op.get_cardinality() == restriction.get_cardinality()
-                        and op.get_property() == restriction.get_property()):
-                    same_root.append(op)
-            if not len(same_root) == 1:
-                fillers = _sort_by_ordered_owl_object([p.get_filler() for p in same_root])
-                if isinstance(list(fillers)[0],OWLDataRange):
+            # Only apply filler merging for min-cardinality restrictions with cardinality == 1
+            if (isinstance(restriction, (OWLObjectMinCardinality, OWLDataMinCardinality))
+                    and restriction.get_cardinality() == 1):
+                operands = set(nary_ce.operands())
+                same_root = []
+                for op in operands:
+                    if (isinstance(op, type(restriction))
+                            and op.get_cardinality() == restriction.get_cardinality()
+                            and op.get_property() == restriction.get_property()):
+                        same_root.append(op)
+                if not len(same_root) == 1:
+                    fillers = _sort_by_ordered_owl_object([p.get_filler() for p in same_root])
+                    if isinstance(list(fillers)[0], OWLDataRange):
+                        return type(restriction)(cardinality=restriction.get_cardinality(),
+                                                 property=restriction.get_property(),
+                                                 filler=self._simplify(OWLDataUnionOf(fillers)))
                     return type(restriction)(cardinality=restriction.get_cardinality(),
                                              property=restriction.get_property(),
-                                             filler=self._simplify(OWLDataUnionOf(fillers)))
-                return type(restriction)(cardinality=restriction.get_cardinality(),
-                                           property=restriction.get_property(),
-                                           filler=self._simplify(OWLObjectUnionOf(fillers)))
+                                             filler=self._simplify(OWLObjectUnionOf(fillers)))
         return type(restriction)(cardinality=restriction.get_cardinality(),
                                  property=restriction.get_property(),
                                  filler=self._simplify(restriction.get_filler(), None))
diff --git a/tests/test_data_cardinality_restrictions.py b/tests/test_data_cardinality_restrictions.py
