Spaces:
Running
Running
Claude Opus 5.5
feat(ontology): module serieux (vue 360, fraude, coherence, terminal IA fiable)
c6c39e9 Download src/core/validation_engine.py from klydekushy/Vortex-Flux: direct link, hf CLI and curl.
- Browser
- Download file 2.99 kB
-
https://huggingface.co/spaces/klydekushy/Vortex-Flux/resolve/main/src/core/validation_engine.py
- Command line
-
hf download hf://spaces/klydekushy/Vortex-Flux/src/core/validation_engine.py
-
curl -L -o validation_engine.py https://huggingface.co/spaces/klydekushy/Vortex-Flux/resolve/main/src/core/validation_engine.py
2.99 kB
| """ | |
| 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", []), | |
| } | |