Generació de classes Python des d’ontologies RDF i esquemes YAML

Aquest tutorial mostra com generar fitxers Python amb classes entity a partir de dos tipus de font:

Ontologies RDF/OWL (RiC-O) — descarrega, converteix i genera classes
Esquemes YAML (datasets existents) — genera classes des del schema del backend

Pipeline general

RDF/OWL ──→ rdf_to_yaml() ──→ YAML DRM ──→ generate_classes() ──→ .py
YAML  ──────────────────────────────────────→ generate_classes() ──→ .py

1. Generació des d’una ontologia RDF/OWL

Pipeline complet en un pas

La funció download_ontology_and_convert() fa tot el procés automàticament:

[ ]:

from drm.rdf_schema import download_ontology_and_convert

# Descarrega, converteix i genera classes en un sol pas
output_path = download_ontology_and_convert(
    "https://raw.githubusercontent.com/ICA-EGAD/RiC-O/master/ontology/current-version/RiC-O_1-1.rdf",
    "rico",              # nom → genera rico_entities.py
    output_dir="drm/"
)
print(f"Classes generades a: {output_path}")

Pas a pas

Si vols més control, pots fer cada pas per separat:

1. Descarregar l’ontologia

[ ]:

import os
from drm.rdf_schema import download_ontology

ont_path = download_ontology(
    "https://raw.githubusercontent.com/ICA-EGAD/RiC-O/master/ontology/current-version/RiC-O_1-1.rdf",
    output_dir="ontologies/",
    filename="RiC-O_1-1.rdf"
)
print(f"Ontologia descarregada: {ont_path}")
print(f"Mida: {os.path.getsize(ont_path) / 1024 / 1024:.1f} MB")

2. Convertir RDF → YAML DRM

[ ]:

from drm.rdf_schema import rdf_to_yaml

yaml_str = rdf_to_yaml(ont_path, "rico")
print("Primeres línies del YAML generat:")
print(yaml_str[:600])

3. YAML → Classes Python

[ ]:

from drm.schema_gen import generate_classes

py_source = generate_classes(yaml_str)
print(f"Total línies generades: {len(py_source.splitlines())}")
print()
print("Primera classe generada (Thing):")
for line in py_source.splitlines()[:15]:
    print(line)

4. Escriure el fitxer

[ ]:

output_file = "drm/entities_rico.py"
with open(output_file, "w") as f:
    f.write(py_source)
print(f"Fitxer escrit: {output_file}")
print(f"Mida: {os.path.getsize(output_file) / 1024:.0f} KB")

Mapeig RDF/OWL → DRM

L’automata mapeja els constructs OWL de la següent manera:

OWL construct	DRM mapping	Exemple RiC-O
`owl:Class`	Nom de la classe	`Agent`, `Document`
`rdfs:subClassOf`	`WeakNode` (herència)	`Agent` hereta de `Thing`
`owl:DatatypeProperty`	Propietats del node	`Agent.authorizingMandate`
`owl:ObjectProperty`	Relacions entre nodes	`accumulationRelation_role`
`owl:hasKey`	Camps de la PK	(RiC-O no en té)
`rdfs:comment`	Docstring de la classe	Descripcions de RiC-O

Anàlisi de l’ontologia RiC-O

Resum de les classes generades a partir de RiC-O:

[ ]:

import yaml

data = yaml.safe_load(yaml_str)

node_count = sum(1 for v in data["labels"].values() if v["base_class"] == "Node")
weaknode_count = sum(1 for v in data["labels"].values() if v["base_class"] == "WeakNode")
rel_count = len(data.get("relationships", {}))
weakrel_count = len(data.get("weak_relations", {}))

print(f"Node classes: {node_count}")
print(f"WeakNode classes: {weaknode_count}")
print(f"Relacions: {rel_count}")
print(f"WeakRelacions: {weakrel_count}")
print(f"Total classes: {node_count + weaknode_count + rel_count + weakrel_count}")

Jerarquia més profunda de RiC-O

RiC-O té una jerarquia de profunditat 5:

[ ]:

# Construir el mapa de pares
parent_map = {}
for label, info in data["labels"].items():
    if info["base_class"] == "WeakNode" and "parent" in info:
        parent_map[label] = info["parent"]

# Calcular profunditats
def get_depth(label, memo={}):
    if label in memo:
        return memo[label]
    if label not in parent_map:
        memo[label] = 0
        return 0
    depth = 1 + get_depth(parent_map[label], memo)
    memo[label] = depth
    return depth

depths = {l: get_depth(l) for l in parent_map}
max_depth = max(depths.values())

# Trobar la classe més profunda
deepest = [l for l, d in depths.items() if d == max_depth]
label = sorted(deepest)[0]

# Traçar la cadena
path = []
current = label
while current in parent_map:
    path.append(current)
    current = parent_map[current]
path.append(current)
path.reverse()

print(f"Classe més profunda: {label} (profunditat {max_depth})")
print("Jerarquia completa:")
for i, cls in enumerate(path):
    info = data["labels"][cls]
    base = info["base_class"]
    print(f"  {i}. {cls} ({base})")

Importar les classes generades

[ ]:

from drm.rico_entities import Thing, Agent, Document, AuthorshipRelation

# Thing és un Node (arrel de la jerarquia)
print(f"Thing: {Thing.__name__} → base: {Thing.__bases__[0].__name__}")

# Agent és un WeakNode (fill de Thing)
print(f"Agent: {Agent.__name__} → base: {Agent.__bases__[0].__name__}")

# AuthorshipRelation és un WeakNode (profunditat 5)
print(f"AuthorshipRelation: {AuthorshipRelation.__name__} → base: {AuthorshipRelation.__bases__[0].__name__}")

2. Generació des d’un esquema YAML (datasets existents)

Els datasets existents tenen un schema YAML que es pot generar amb GraphStore.schema_yaml(). Aquest schema es pot utilitzar per generar classes Python.

Exemple: Karate Club

Primer carreguem el dataset i generem el schema YAML:

[ ]:

from drm import NetworkXGraph
from drm.exemples import load_karate_club

# Carregar el dataset
g = NetworkXGraph()
stats = load_karate_club(g)
print("Estadístiques:", stats)

# Generar el schema YAML
yaml_str = g.schema_yaml("karate")
print()
print("Schema YAML generat:")
print(yaml_str)

Analitzem el schema:

[ ]:

import yaml

data = yaml.safe_load(yaml_str)
print("Labels:")
for label, info in data["labels"].items():
    print(f"  {label}:")
    print(f"    base_class: {info['base_class']}")
    print(f"    primary_key: {info.get('primary_key', [])}")
    print(f"    properties: {info.get('properties', {})}")
    if 'parent' in info:
        print(f"    parent: {info['parent']}")
print()
print("Relacions:", list(data.get("relationships", {}).keys()))
print("WeakRelacions:", list(data.get("weak_relations", {}).keys()))

Generem les classes Python:

[ ]:

from drm.schema_gen import generate_classes

py_source = generate_classes(yaml_str)
print(py_source)

Exemple: Bibliografia (OpenAlex)

El dataset bibliografia té classes Paper i Author amb relacions AUTHORED i CITES:

[ ]:

from drm import NetworkXGraph
from drm.exemples import load_bibliografia_openalex

g = NetworkXGraph()
stats = load_bibliografia_openalex(g, query="graph database", per_page=5)
print("Estadístiques:", stats)

yaml_str = g.schema_yaml("bibliografia")
data = yaml.safe_load(yaml_str)

print("\nLabels:")
for label, info in data["labels"].items():
    print(f"  {label}: base={info['base_class']}, pk={info.get('primary_key', [])}")
print("\nRelacions:", list(data.get("relationships", {}).keys()))

py_source = generate_classes(yaml_str)
print("\nClasses generades:")
for line in py_source.splitlines():
    if line.startswith("class ") or "def __init__" in line or "super().__init__" in line:
        print(line)

g.close()

3. Generació des d’un YAML manual

També pots crear un YAML manual i generar les classes:

[ ]:

from drm.schema_gen import generate_classes

yaml_str = """
labels:
  Document:
    class_name: Document
    base_class: Node
    properties:
      title: string
      year: integer
    primary_key: ["id"]
    doc: "A cultural document."
  Page:
    class_name: Page
    base_class: WeakNode
    properties: {}
    primary_key: []
    parent: Document
    parent_relation: HAS_PAGE
    doc: "A page within a document."
relationships:
  references:
    class_name: References
    src: Document
    dst: Document
    properties: {}
weak_relations:
  HAS_PAGE:
    class_name: HasPage
    base_class: WeakRelation
    propagate: true
    src: Document
    dst: Page
"""

py_source = generate_classes(yaml_str)
print(py_source)

4. PK optional vs mandatory

El generador detecta automàticament si una classe té primary_key definit:

YAML `primary_key`	Codi generat
`[]` (buit)	`pk: Optional[Dict[str, Any]] = None`
`["id"]`	`pk: Dict[str, Any]` (mandatory)

Això és important per a ontologies RDF que no fan servir owl:hasKey: el backend assigna un ID intern (neo4j_id) que es converteix en la PK.

[ ]:

# RiC-O no té owl:hasKey → totes les classes tenen primary_key: []
no_pk = [l for l, i in data["labels"].items() if not i.get("primary_key")]
with_pk = [l for l, i in data["labels"].items() if i.get("primary_key")]

print(f"Classes sense PK definida: {len(no_pk)}")
print(f"Classes amb PK definida: {len(with_pk)}")
print()
print("Exemples de classes sense PK (RiC-O):")
for cls in sorted(no_pk)[:5]:
    info = data["labels"][cls]
    parent = info.get('parent', 'arrel')
    print(f"  - {cls} (parent: {parent})")

Resum

Font	Funció	Sortida
RDF/OWL URL	`download_ontology_and_convert()`	`drm/{db_name}_entities.py`
RDF/OWL fitxer	`rdf_to_yaml()` + `generate_classes()`	Codi Python (string)
Schema YAML	`generate_classes()`	Codi Python (string)
YAML manual	`generate_classes()`	Codi Python (string)