Skip to content

Labeller

kgm.labeller

Assesses a given Manifest, finds any IRIs in any of the given resources missing labels and tries to patch them from a given source of labels, such as KurrawongAI's Semantic Background (https://github.com/kurrawong/semantic-background) repository.

label(manifest: Path, output_type: LabellerOutputTypes = LabellerOutputTypes.manifest, additional_context: Path | str | Graph = None, http_client: httpx.Client = None) -> set | Graph | None

Main function for labeller module

Source code in kgm/labeller.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def label(
    manifest: Path,
    output_type: LabellerOutputTypes = LabellerOutputTypes.manifest,
    additional_context: Path | str | Graph = None,
    http_client: httpx.Client = None,
) -> set | Graph | None:
    """Main function for labeller module"""
    if not isinstance(output_type, LabellerOutputTypes):
        raise ValueError(
            f"Invalid output_type value, must be one of {', '.join([x for x in LabellerOutputTypes])}"
        )

    # create the target from the Manifest
    manifest_path, manifest_root, manifest_graph = get_manifest_paths_and_graph(
        manifest
    )

    content_graph = Graph()
    context_graph = Graph()

    artifacts = denormalise_artifacts((manifest_path, manifest_root, manifest_graph))

    for k, v in artifacts.items():
            context_graph += load_graph(k)

    # add labels for system IRIs
    context_graph.parse(Path(__file__).parent / "system-labels.ttl")

    if output_type == LabellerOutputTypes.iris:
        combined_graph = manifest_graph + content_graph + context_graph

        iris_missing_labels = find_missing_labels(combined_graph)

        return sorted(iris_missing_labels)

    elif output_type == LabellerOutputTypes.rdf:
        combined_graph = manifest_graph + content_graph + context_graph

        iris_missing_labels = find_missing_labels(combined_graph)

        if additional_context is None:
            additional_context = "https://fuseki.dev.kurrawong.ai/semback/sparql"

        return get_missing_labels(iris_missing_labels, additional_context, "graph", http_client)

    else:  # output_type == LabellerOutputTypes.manifest
        # If this is selected, generate the "rdf" output and create a resource for it in the Manifest
        # If there are no more missing labels then we have a mrr:CompleteCatalogueAndResourceLabels
        # else add mrr:IncompleteCatalogueAndResourceLabels

        # Generate labels for any IRIs missing them, using context given in the Manifest and any
        # Additional Context supplied

        rdf_addition = label(manifest, LabellerOutputTypes.rdf, additional_context)

        if len(rdf_addition) > 0:
            new_artifact = manifest.parent / "labels-additional.ttl"
            rdf_addition.serialize(destination=new_artifact, format="longturtle")
            new_resource = BNode()

            # Add to the Manifest
            manifest_iri = manifest_graph.value(
                predicate=RDF.type, object=PREZ.Manifest
            )
            manifest_graph.add((manifest_iri, PROF.hasResource, new_resource))
            manifest_graph.add(
                (new_resource, PROF.hasRole, MRR.IncompleteCatalogueAndResourceLabels)
            )
            manifest_graph.add(
                (new_resource, PROF.hasArtifact, Literal(new_artifact.name))
            )

            manifest_graph.serialize(destination=manifest, format="longturtle")
        else:
            raise Warning(
                "No new labels have been generated for content in this Manifest. "
                "This could be because none were missing or because no new labels can be found in any "
                "supplied additional context."
            )