Skip to content
Snippets Groups Projects
convert_schema_to_owl.py 2.69 KiB
Newer Older
import yaml
from linkml_runtime.linkml_model.meta import SchemaDefinition, ClassDefinition, SlotDefinition, EnumDefinition
from linkml_runtime.utils.schemaview import SchemaView
from rdflib import Graph, Literal, RDF, RDFS, OWL, URIRef, Namespace

def load_schema(schema_file: str) -> SchemaDefinition:
    with open(schema_file, 'r') as f:
        schema_dict = yaml.safe_load(f)
    return SchemaDefinition(**schema_dict)

def convert_to_owl(schema: SchemaDefinition, output_file: str):
    g = Graph()
    base_uri = URIRef(schema.id)
    ns = Namespace(base_uri)
    g.bind('owl', OWL)
    g.bind('rdfs', RDFS)
    g.bind('ex', ns)

    # Convert classes
    for cls_name, cls in schema.classes.items():
        class_uri = URIRef(base_uri + cls_name)
        g.add((class_uri, RDF.type, OWL.Class))
        g.add((class_uri, RDFS.label, Literal(cls_name)))
        if cls.description:
            g.add((class_uri, RDFS.comment, Literal(cls.description)))
        if cls.is_a:
            parent_class_uri = URIRef(base_uri + cls.is_a)
            g.add((class_uri, RDFS.subClassOf, parent_class_uri))

    # Convert slots (properties)
    for slot_name, slot in schema.slots.items():
        slot_uri = URIRef(base_uri + slot_name)
        g.add((slot_uri, RDF.type, OWL.ObjectProperty if slot.range == 'uriorcurie' else OWL.DatatypeProperty))
        g.add((slot_uri, RDFS.label, Literal(slot_name)))
        if slot.description:
            g.add((slot_uri, RDFS.comment, Literal(slot.description)))
        if slot.domain:
            domain_class_uri = URIRef(base_uri + slot.domain)
            g.add((slot_uri, RDFS.domain, domain_class_uri))
        if slot.range:
            range_class_uri = URIRef(base_uri + slot.range)
            g.add((slot_uri, RDFS.range, range_class_uri))

    # Convert enums
    for enum_name, enum in schema.enums.items():
        enum_uri = URIRef(base_uri + enum_name)
        g.add((enum_uri, RDF.type, OWL.Class))
        g.add((enum_uri, RDFS.label, Literal(enum_name)))
        for pv_name, pv in enum.permissible_values.items():
            pv_uri = URIRef(base_uri + pv_name)
            g.add((pv_uri, RDF.type, OWL.NamedIndividual))
            g.add((pv_uri, RDF.type, enum_uri))
            g.add((pv_uri, RDFS.label, Literal(pv_name)))
            if pv.description:
                g.add((pv_uri, RDFS.comment, Literal(pv.description)))

    # Save the graph to a file
    g.serialize(destination=output_file, format='xml')

# Main function
def main():
    schema_file = 'pizza.yaml'
    output_file = 'pizza_schema.owl'
    schema = load_schema(schema_file)
    convert_to_owl(schema, output_file)
    print(f'Converted {schema_file} to {output_file}')

if __name__ == '__main__':
    main()