Spaces:
Running
Running
File size: 2,993 Bytes
409f286 c6c39e9 409f286 c6c39e9 2fe4365 c6c39e9 2fe4365 409f286 c6c39e9 2fe4365 c6c39e9 409f286 c6c39e9 409f286 c6c39e9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 | """
MODULE: VALIDATION ENGINE
=========================
Qualite du graphe RDF au regard de la feuille Ontology, pour TOUTES les classes :
- completude : part des entites de chaque classe qui portent chaque propriete declaree ;
- cardinalite : une propriete declaree "1" doit avoir exactement une valeur par entite.
Score qualite = completude moyenne (en %), sur toutes les paires (classe, propriete).
"""
from rdflib import Namespace
from rdflib.namespace import RDF
VORTEX = Namespace("http://vortex.ai/ontology#")
class ValidationEngine:
def __init__(self, rdf_store, ontology_manager):
self.rdf = rdf_store
self.onto = ontology_manager
self.report = {}
def run_validation(self):
g = self.rdf.g
lignes, violations = [], []
for cls in sorted(self.onto.structure["classes"]):
entites = set(g.subjects(RDF.type, VORTEX[cls]))
if not entites:
continue
attendues = list(dict.fromkeys(self.onto.structure["properties"].get(cls, []) +
self.onto.structure["relations"].get(cls, [])))
for prop in attendues:
nb_valeurs = {e: len(list(g.objects(e, VORTEX[prop]))) for e in entites}
presentes = sum(1 for n in nb_valeurs.values() if n > 0)
lignes.append({"Classe": cls, "Propriété": prop, "Entités": len(entites),
"Renseignées": presentes, "Complétude %": round(100 * presentes / len(entites), 1)})
card = str(self.onto.cardinalities.get((cls, prop), "")).strip()
if card in ("1", "1..1"):
hors = [e for e, n in nb_valeurs.items() if n != 1]
elif card in ("0..1",):
hors = [e for e, n in nb_valeurs.items() if n > 1]
elif card in ("1..*", "1..n"):
hors = [e for e, n in nb_valeurs.items() if n == 0]
else:
hors = []
if hors:
violations.append({"Classe": cls, "Propriété": prop, "Cardinalité": card,
"Entités hors règle": len(hors),
"Exemples": ", ".join(sorted(str(e).split("#")[-1] for e in hors)[:5])})
self.report = {"completude": lignes, "cardinalites": violations,
"volumes": self.rdf.get_full_schema().get("classes", {})}
return self.report
def generate_metrics_dashboard(self):
completude = self.report.get("completude", [])
score = round(sum(l["Complétude %"] for l in completude) / len(completude), 1) if completude else 0.0
return {
"triples": len(self.rdf.g),
"entities": sum(self.report.get("volumes", {}).values()),
"coverage_score": score,
"completude": completude,
"cardinalites": self.report.get("cardinalites", []),
}
|