Source code for buildingmotif.ingresses.base

import json
from dataclasses import dataclass
from functools import cached_property
from os import PathLike
from pathlib import Path
from typing import List

from rdflib import Graph, Namespace

from buildingmotif import BuildingMOTIF


@dataclass
class Record:
    """Represents a piece of metadata from some metadata ingress"""

    # an arbitrary "type hint"
    rtype: str
    # possibly-nested dictionary of (semi-)structured data from
    # the underlying source
    fields: dict


class IngressHandler:
    """Abstract superclass for Record/Graph ingress handlers"""

    pass


class RecordIngressHandler(IngressHandler):
    """Generates Record instances from an underlying metadata source"""

    def __init__(self, bm: BuildingMOTIF):
        self.bm = bm

    @cached_property
    def records(self) -> List[Record]:
        """
        Generates (then caches) a list of Records from an underlying data source
        """
        raise NotImplementedError("Must be overridden by subclass")

    def dump(self, path: PathLike):
        """
        Takes the contents of the records of this handler and writes them to a JSON file

        :param path: path to write output file to
        :type path: PathLike
        """
        output_string = self.dumps()
        output_file = Path(path)
        with output_file.open("w", encoding="utf-8") as f:
            f.write(output_string)

    def dumps(self) -> str:
        """
        Takes the contents of the records of this handler and writes them to a string
        """
        records = [
            {"rtype": record.rtype, "fields": record.fields} for record in self.records
        ]
        return json.dumps(records)

    @classmethod
    def load(cls, path: PathLike):
        """
        Takes a file generated by 'dump' and creates a new ingress handler with those records
        """
        return cls.loads(Path(path).read_text())

    @classmethod
    def loads(cls, s: str):
        """
        Takes the string output by 'dumps' and creates a new ingress handler with those records
        """
        self = cls.__new__(cls)
        records = []
        for record in json.loads(s):
            records.append(Record(record["rtype"], record["fields"]))
        self.records = records
        return self


class GraphIngressHandler(IngressHandler):
    """Generates a Graph from an underlying metadata source or RecordIngressHandler"""

    def __init__(self, bm: BuildingMOTIF):
        self.bm = bm

    def graph(self, ns: Namespace) -> Graph:
        """Generates an RDF graph with all entities being placed in the given namespace"""
        raise NotImplementedError("Must be overridden by subclass")