post-process.py

import os
import re
import json
import argparse
from tqdm import tqdm
from rdflib import Graph, RDF, RDFS, URIRef, Literal, Namespace

from experiment_paths import experiment_dir, resolve_config_path, resolve_repo_path

BASE = Namespace("https://eurecom.fr/llms4ol/resource/")
RDFFake = Namespace("http://www.w3.org/1999/02/rdf-syntax-ns#")

SOURCE_DIR = experiment_dir(resolve_config_path(), "graphs")
OUTPUT_DIR = experiment_dir(resolve_config_path(), "output")
os.makedirs(OUTPUT_DIR, exist_ok=True)

def to_string(value):
    text = str(value)
    text = re.sub(r'(?<=[a-z0-9])(?=[A-Z])', ' ', text)
    return text.lower().replace(",", "")

def levenshtein_distance(a, b):
    if a == b:
        return 0
    if len(a) < len(b):
        a, b = b, a
    previous = list(range(len(b) + 1))
    for i, ca in enumerate(a, 1):
        current = [i]
        for j, cb in enumerate(b, 1):
            insertions = previous[j] + 1
            deletions = current[j - 1] + 1
            substitutions = previous[j - 1] + (ca != cb)
            current.append(min(insertions, deletions, substitutions))
        previous = current
    return previous[-1]

def run():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--structure-only",
        action="store_true",
        help="Keep only is-a and instance-of triples in output.",
    )

    parser.add_argument(
        "--involve-onto",
        action="store_true",
        help="Load also the OTK ontology.",
    )

    parser.add_argument(
        "--submission",
        action="store_true",
        help="Also output submission JSON with id and primitive-ontology-triples.",
    )
    args = parser.parse_args()

    submission_rows = []

    for x in tqdm(os.listdir(SOURCE_DIR)):
        if not x.endswith(".ttl"):
            continue


        input_file = os.path.join(SOURCE_DIR, x)
        output_file = os.path.join(OUTPUT_DIR, x.replace(".ttl", ".txt"))

        g = Graph()
        g.parse(input_file, format='ttl')
        triples_for_submission = []

        if args.involve_onto:
            onto_path = resolve_repo_path(os.path.join('onto', x.replace(".txt", "")))
            g.parse(onto_path, format='ttl')

        with open(output_file, "w") as f:
            for s, p, o, in g.triples((None, None, None)):
                if p == RDFS.label:
                    continue
                subj = str(s)
                predicate = str(p).replace(str(BASE), "")
                obj = str(o)
                if isinstance(s, URIRef):
                    subj = g.value(s, RDFS.label) or subj.replace(str(BASE), "")
                if isinstance(o, URIRef):
                    obj = g.value(o, RDFS.label) or obj.replace(str(BASE), "")

                if p in [RDFS.subClassOf, BASE.isSubclassOf, BASE.isMemberOf, BASE.isASubtypeOf, BASE.memberOf, BASE.isPartOf]:
                    predicate = "is-a"
                elif p in [RDF.type, RDFFake.type, BASE.type]:
                    predicate = "instance-of"
                elif p == BASE.hasSubtype:
                    predicate = "is-a"
                    temp = subj
                    subj = obj
                    obj = temp

                if args.structure_only and predicate not in {"is-a", "instance-of"}:
                    continue

                subj = to_string(subj)
                obj = to_string(obj)
                if levenshtein_distance(subj, obj) <= 1:
                    continue

                f.write(f'({subj}, {predicate}, {obj})\n')
                triples_for_submission.append([subj, predicate, obj])

        if args.submission:
            submission_rows.append(
                {
                    "id": x.replace(".ttl", "").replace(".txt", ""),
                    "primitive-ontology-triples": triples_for_submission,
                }
            )

    if args.submission:
        submission_path = os.path.join(OUTPUT_DIR, "submission.json")
        with open(submission_path, "w") as f:
            json.dump(submission_rows, f, ensure_ascii=False, indent=2)


if __name__ == "__main__":
    run()