conversor_vf.py

# -*- coding: utf-8 -*-
import rdflib
from rdflib.namespace import Namespace

# Namespaces
SWRL = Namespace("http://www.w3.org/2003/11/swrl#")
SWRLB = Namespace("http://www.w3.org/2003/11/swrlb#")
SWRLA = Namespace("http://www.w3.org/2003/11/swrl-annotations#")
XSD = rdflib.XSD

# Conjuntos globales para rastrear propiedades de objeto, datavalores, instancias conocidas y variables SWRL
object_props = set()
data_props = set()
known_instances = set()
swrl_vars = set()

def clean_name(name):
    """
    Normaliza un nombre eliminando espacios, ';', '{', '}', y cualquier cosa después de '['.
    """
    return name.strip().replace(";", "").replace("{", "").replace("}", "").split("[")[0].strip()

def create_class(g, base_uri, class_name, parent_class=None):
    """
    Crea una clase OWL con URI base_uri+class_name y opcionalmente la declara como subclase de parent_class.
    """
    class_uri = rdflib.URIRef(base_uri + clean_name(class_name))
    g.add((class_uri, rdflib.RDF.type, rdflib.OWL.Class))
    g.add((class_uri, rdflib.RDFS.label, rdflib.Literal(class_name)))
    if parent_class:
        parent_uri = rdflib.URIRef(base_uri + clean_name(parent_class))
        g.add((class_uri, rdflib.RDFS.subClassOf, parent_uri))

def create_datatype_property(g, base_uri, property_name, property_type, domain=None):
    """
    Crea una DatatypeProperty con nombre property_name y rango XSD según property_type.
    Si se indica domain, agrega rdfs:domain apuntando a esa clase.
    """
    prop_clean = clean_name(property_name)
    prop_uri = rdflib.URIRef(base_uri + prop_clean)
    g.add((prop_uri, rdflib.RDF.type, rdflib.OWL.DatatypeProperty))
    g.add((prop_uri, rdflib.RDFS.label, rdflib.Literal(property_name)))
    if domain:
        domain_uri = rdflib.URIRef(base_uri + clean_name(domain))
        g.add((prop_uri, rdflib.RDFS.domain, domain_uri))
    xsd_type_map = {
        "int": "integer", "integer": "integer", "float": "float", "double": "double",
        "string": "string", "decimal": "decimal", "dateTime": "dateTime", "boolean": "boolean"
    }
    ptype = property_type.lower()
    xsd_name = xsd_type_map.get(ptype, "string")
    range_uri = rdflib.URIRef(XSD[xsd_name])
    g.add((prop_uri, rdflib.RDFS.range, range_uri))
    data_props.add(prop_clean)

def create_object_property(g, base_uri, property_name, domain, range_):
    """
    Crea una ObjectProperty con nombre property_name, indicando su dominio y rango.
    """
    prop_uri = rdflib.URIRef(base_uri + clean_name(property_name))
    domain_uri = rdflib.URIRef(base_uri + clean_name(domain))
    range_uri = rdflib.URIRef(base_uri + clean_name(range_))
    g.add((prop_uri, rdflib.RDF.type, rdflib.OWL.ObjectProperty))
    g.add((prop_uri, rdflib.RDFS.label, rdflib.Literal(property_name)))
    g.add((prop_uri, rdflib.RDFS.domain, domain_uri))
    g.add((prop_uri, rdflib.RDFS.range, range_uri))
    object_props.add(clean_name(property_name))

def create_instance(g, base_uri, instance_name, class_name, attributes):
    """
    Crea un individuo de clase class_name con nombre instance_name y asigna propiedades de objeto o literales según attributes.
    """
    global object_props, known_instances
    # No crear instancia si el nombre empieza con “?”
    if instance_name.strip().startswith("?"):
        return
    instance_clean = clean_name(instance_name)
    instance_uri = rdflib.URIRef(base_uri + instance_clean)
    class_uri = rdflib.URIRef(base_uri + clean_name(class_name))
    known_instances.add(instance_clean)
    g.add((instance_uri, rdflib.RDF.type, rdflib.OWL.NamedIndividual))
    g.add((instance_uri, rdflib.RDF.type, class_uri))
    for attr_name, attr_value in attributes.items():
        # Ignorar atributos cuyo valor inicie en “?”
        if attr_value.strip().startswith("?"):
            continue
        attr_clean = clean_name(attr_name)
        attr_uri = rdflib.URIRef(base_uri + attr_clean)
        value_clean = clean_name(attr_value)
        if attr_clean in object_props and value_clean in known_instances:
            # Si es propiedad de objeto y el valor es instancia conocida
            object_value_uri = rdflib.URIRef(base_uri + value_clean)
            g.add((instance_uri, attr_uri, object_value_uri))
        else:
            # Tratar como literal (booleano, entero, double o string)
            try:
                if value_clean.lower() in ["true", "false"]:
                    literal_value = rdflib.Literal(value_clean.lower() == "true", datatype=XSD.boolean)
                elif value_clean.isdigit():
                    literal_value = rdflib.Literal(int(value_clean), datatype=XSD.integer)
                elif "." in value_clean:
                    literal_value = rdflib.Literal(float(value_clean), datatype=XSD.double)
                else:
                    literal_value = rdflib.Literal(value_clean, datatype=XSD.string)
            except:
                literal_value = rdflib.Literal(value_clean, datatype=XSD.string)
            g.add((instance_uri, attr_uri, literal_value))

def resolve_argument(arg, base_uri):
    """
    Dado un texto arg, si empieza con '?', se supone variable SWRL (pero parse_atom la maneja).
    Si no, intenta convertir a instancia conocida o a literal.
    """
    arg = arg.strip()
    if arg.startswith("?"):
        # En teoría parse_atom debería interceptar variables; aquí devolvemos Variable genérica
        return rdflib.Variable(arg)
    cleaned = clean_name(arg)
    if cleaned in known_instances:
        return rdflib.URIRef(base_uri + cleaned)
    try:
        if cleaned.lower() in ["true", "false"]:
            return rdflib.Literal(cleaned.lower() == "true", datatype=XSD.boolean)
        elif cleaned.isdigit():
            return rdflib.Literal(int(cleaned), datatype=XSD.integer)
        elif "." in cleaned:
            return rdflib.Literal(float(cleaned), datatype=XSD.double)
        else:
            return rdflib.Literal(cleaned, datatype=XSD.string)
    except:
        return rdflib.Literal(cleaned, datatype=XSD.string)

def parse_atom_list(g, base_uri, list_node, atoms):
    """
    Construye un rdf:List en el grafo g a partir de la lista de cadenas 'atoms'.
    Si atoms está vacío, hace que list_node  rdf:nil.
    """
    if not atoms:
        g.add((list_node, rdflib.RDF.rest, rdflib.RDF.nil))
        return

    current = list_node
    for i, atom_str in enumerate(atoms):
        atom_str = atom_str.strip().rstrip(";")
        atom_node = parse_atom(g, base_uri, atom_str)
        g.add((current, rdflib.RDF.first, atom_node))

        if i < len(atoms) - 1:
            next_node = rdflib.BNode()
            g.add((current, rdflib.RDF.rest, next_node))
            current = next_node
        else:
            g.add((current, rdflib.RDF.rest, rdflib.RDF.nil))

def parse_atom(g, base_uri, atom_str):
    """
    Construye un SWRL Atom (ClassAtom, IndividualPropertyAtom, DatavaluedPropertyAtom o BuiltinAtom).
    Declara variables SWRL con IRI base_uri+var_name para que SWRLTab las reconozca.
    """
    atom_str = atom_str.strip()

    # 1) Caso Builtin (swrlb:función)
    if atom_str.startswith("swrlb:"):
        func_name, args = atom_str.split("(", 1)
        args = args.rstrip(")").split(",")
        atom_node = rdflib.BNode()
        g.add((atom_node, rdflib.RDF.type, SWRL.BuiltinAtom))
        g.add((atom_node, SWRL.builtin, SWRLB[func_name.replace("swrlb:", "")]))

        args_list = rdflib.BNode()
        g.add((atom_node, SWRL.arguments, args_list))
        current = args_list

        for i, a in enumerate(arg.strip() for arg in args):
            if a.startswith("?"):
                var_name = a[1:]
                var_uri = rdflib.URIRef(base_uri + var_name)
                if var_name not in swrl_vars:
                    swrl_vars.add(var_name)
                    g.add((var_uri, rdflib.RDF.type, SWRL.Variable))
                var_or_lit = var_uri
            else:
                # Si es literal numérico o booleano
                var_or_lit = resolve_argument(a, base_uri)

            next_node = rdflib.BNode() if i < len(args) - 1 else rdflib.RDF.nil
            g.add((current, rdflib.RDF.first, var_or_lit))
            g.add((current, rdflib.RDF.rest, next_node))
            current = next_node

        return atom_node

    # 2) Caso ClassAtom, DatavaluedPropertyAtom o IndividualPropertyAtom
    else:
        pred, args = atom_str.split("(", 1)
        args = args.rstrip(")").split(",")
        pred_clean = clean_name(pred)
        args = [arg.strip() for arg in args]

        # 2.1) ClassAtom si sólo hay un argumento
        if len(args) == 1:
            class_uri = rdflib.URIRef(base_uri + pred_clean)
            a = args[0]

            if a.startswith("?"):
                var_name = a[1:]
                var_uri = rdflib.URIRef(base_uri + var_name)
                if var_name not in swrl_vars:
                    swrl_vars.add(var_name)
                    g.add((var_uri, rdflib.RDF.type, SWRL.Variable))
                var = var_uri
            else:
                var = resolve_argument(a, base_uri)

            atom_node = rdflib.BNode()
            g.add((atom_node, rdflib.RDF.type, SWRL.ClassAtom))
            g.add((atom_node, SWRL.classPredicate, class_uri))
            g.add((atom_node, SWRL.argument1, var))
            return atom_node

        # 2.2) DatavaluedPropertyAtom si es data property
        elif len(args) == 2 and pred_clean in data_props:
            subj_arg = args[0]
            obj_arg  = args[1]

            # Argumento sujeto
            if subj_arg.startswith("?"):
                var_name = subj_arg[1:]
                var_uri = rdflib.URIRef(base_uri + var_name)
                if var_name not in swrl_vars:
                    swrl_vars.add(var_name)
                    g.add((var_uri, rdflib.RDF.type, SWRL.Variable))
                subj = var_uri
            else:
                subj = resolve_argument(subj_arg, base_uri)

            # Argumento objeto
            if obj_arg.startswith("?"):
                var_name = obj_arg[1:]
                var_uri = rdflib.URIRef(base_uri + var_name)
                if var_name not in swrl_vars:
                    swrl_vars.add(var_name)
                    g.add((var_uri, rdflib.RDF.type, SWRL.Variable))
                obj = var_uri
            else:
                obj = resolve_argument(obj_arg, base_uri)

            atom_node = rdflib.BNode()
            g.add((atom_node, rdflib.RDF.type, SWRL.DatavaluedPropertyAtom))
            g.add((atom_node, SWRL.propertyPredicate, rdflib.URIRef(base_uri + pred_clean)))
            g.add((atom_node, SWRL.argument1, subj))
            g.add((atom_node, SWRL.argument2, obj))
            return atom_node

        # 2.3) IndividualPropertyAtom si es object property
        elif len(args) == 2:
            subj_arg = args[0]
            obj_arg  = args[1]

            if subj_arg.startswith("?"):
                var_name = subj_arg[1:]
                var_uri = rdflib.URIRef(base_uri + var_name)
                if var_name not in swrl_vars:
                    swrl_vars.add(var_name)
                    g.add((var_uri, rdflib.RDF.type, SWRL.Variable))
                subj = var_uri
            else:
                subj = resolve_argument(subj_arg, base_uri)

            if obj_arg.startswith("?"):
                var_name = obj_arg[1:]
                var_uri = rdflib.URIRef(base_uri + var_name)
                if var_name not in swrl_vars:
                    swrl_vars.add(var_name)
                    g.add((var_uri, rdflib.RDF.type, SWRL.Variable))
                obj = var_uri
            else:
                obj = resolve_argument(obj_arg, base_uri)

            atom_node = rdflib.BNode()
            g.add((atom_node, rdflib.RDF.type, SWRL.IndividualPropertyAtom))
            g.add((atom_node, SWRL.propertyPredicate, rdflib.URIRef(base_uri + pred_clean)))
            g.add((atom_node, SWRL.argument1, subj))
            g.add((atom_node, SWRL.argument2, obj))
            return atom_node

        else:
            print(f"[WARN] Atom no reconocido: {atom_str}")
            return None

def create_swrl_rule(g, base_uri, rule_name, antecedent, consequent):
    """
    Crea una regla SWRL con nombre rule_name:
      - antecedent: cadena con átomos separados por “ and ”
      - consequent: cadena con un átomo único (sin ‘;’ final)
    Declara variables “?x” como URIRef base_uri+x y las marca swrl:Variable.
    """
    rule_uri = rdflib.URIRef(base_uri + clean_name(rule_name))
    g.add((rule_uri, rdflib.RDF.type, SWRL.Imp))
    g.add((rule_uri, rdflib.RDFS.label, rdflib.Literal(rule_name)))

    # Cabeza (head)
    head_list = rdflib.BNode()
    g.add((rule_uri, SWRL.head, head_list))
    parse_atom_list(g, base_uri, head_list, [consequent])

    # Cuerpo (body)
    body_list = rdflib.BNode()
    g.add((rule_uri, SWRL.body, body_list))
    body_atoms = [a.strip() for a in antecedent.split(" and ")]
    parse_atom_list(g, base_uri, body_list, body_atoms)

    # Debug (opcional)
    def format_term(term):
        if isinstance(term, rdflib.URIRef):
            s = str(term)
            if s.startswith(base_uri):
                return "?" + s.split("#")[-1]
            else:
                return s.split("#")[-1]
        elif isinstance(term, rdflib.Literal):
            return f"\"{term}\""
        else:
            return str(term)

    def format_atom(atom_str):
        pred, args = atom_str.split("(", 1)
        args = args.rstrip(")").split(",")
        formatted_args = ", ".join(format_term(resolve_argument(arg.strip(), base_uri)) for arg in args)
        return f"{pred}({formatted_args})"

    body_txt = " ^ ".join(format_atom(a) for a in body_atoms)
    head_txt = format_atom(consequent.strip())
    print(f"[SWRL DEBUG] Regla '{rule_name}': {body_txt} -> {head_txt}")

def sysml_to_owl(sysml_file, output_file):
    """
    Lee un archivo SysML, crea clases, propiedades, instancias y reglas SWRL,
    y serializa el grafo a OWL/XML en output_file, incluyendo prefijos para SWRLTab.
    """
    with open(sysml_file, 'r', encoding='utf-8') as file:
        lines = file.readlines()

    g = rdflib.Graph()
    base_uri = "http://www.MySysml.org/ontology#"

    # --- BIND de prefijos para que SWRLTab los reconozca directamente ---
    g.bind("", Namespace(base_uri))           # Prefijo ‘:’ → base_uri
    g.bind("swrl", SWRL)                       # swrl:
    g.bind("swrlb", SWRLB)                     # swrlb:
    g.bind("swrla", SWRLA)                     # swrla:
    g.bind("rdf", rdflib.RDF)                  # rdf:
    g.bind("rdfs", rdflib.RDFS)                # rdfs:
    g.bind("owl", rdflib.OWL)                  # owl:

    current_connection = None
    domain = None
    range_ = None
    current_instance = None
    instance_type = None
    instance_attributes = {}
    current_class = None
    capturing_rule = False
    rule_body = ""
    rule_name = ""

    for line in lines:
        line = line.strip()
        if line.startswith("private import"):
            continue

        # ------------ Definición de clases (part def) ------------
        if line.startswith("part def"):
            parts = line.split(":>")
            part_name = parts[0].split()[2]
            parent_class = parts[1].strip() if len(parts) > 1 else None
            create_class(g, base_uri, part_name, parent_class)
            current_class = part_name

        # --------- Definición de atributos (attribute def) -------
        elif line.startswith("attribute def"):
            parts = line.split(":>")
            attribute_name = clean_name(parts[0].split()[2])
            attribute_type = clean_name(parts[1])
            if current_class:
                create_datatype_property(g, base_uri, attribute_name, attribute_type, domain=current_class)

        # -------- Definición de conexiones (connection def) -------
        elif line.startswith("connection def"):
            parts = line.split()
            current_connection = clean_name(parts[2])
            domain = None
            range_ = None

        elif line.startswith("end domain") and current_connection:
            domain = clean_name(line.split(":")[-1])

        elif line.startswith("end range") and current_connection:
            range_ = clean_name(line.split(":")[-1])

        if current_connection and domain and range_:
            create_object_property(g, base_uri, current_connection, domain, range_)
            current_connection = None
            domain = None
            range_ = None

        # ------------ Definición de instancias (instance) -------------
        elif line.startswith("instance "):
            if current_instance:
                create_instance(g, base_uri, current_instance, instance_type, instance_attributes)
            parts = line.split(":")
            if len(parts) < 2:
                continue
            instance_name_parts = parts[0].split()
            if len(instance_name_parts) < 2:
                continue
            current_instance = clean_name(instance_name_parts[1])
            instance_type = clean_name(parts[1])
            instance_attributes = {}

        elif "=" in line and current_instance:
            attr_parts = line.split("=")
            attr_name = clean_name(attr_parts[0])
            attr_value = clean_name(attr_parts[1])
            instance_attributes[attr_name] = attr_value

        # -------------- Definición de reglas SWRL (rule def) -------------
        elif line.startswith("rule def"):
            rule_name = line.split()[2]
            rule_body = ""
            capturing_rule = True

        elif line.startswith("}") and capturing_rule:
            capturing_rule = False
            if 'if' in rule_body and 'then' in rule_body:
                rb = rule_body.replace("\n", " ")
                cond = rb.split("if")[1].split("then")[0].strip()
                cons = rb.split("then")[1].strip().rstrip(";")
                create_swrl_rule(g, base_uri, rule_name, cond, cons)
            rule_body = ""

        elif capturing_rule:
            rule_body += " " + line

    # Crear la última instancia si quedaba pendiente
    if current_instance:
        create_instance(g, base_uri, current_instance, instance_type, instance_attributes)

    # Serializar el grafo a OWL/XML
    g.serialize(destination=output_file, format='xml')
    print(f"Archivo OWL guardado como: {output_file}")

# Ejecuta la conversión cuando se invoque el script
if __name__ == "__main__":
    sysml_to_owl('ONTOLOGIA_vf.sysml', 'ONTOLOGIA_vf.owl')