Compiler Package

@dataclass
class Compiler(ABC):
    """
    Base class for all compilers.

    A compiler will compile a transformation specification into
    an alternative representation.

    An example compiler would be a R2RML compiler.

    Note: Compilers and Importers will in general be implemented by providing
    mapping specifications
    """

    source_schemaview: SchemaView = None
    """A view over the schema describing the source."""

    source_python_module: str = None
    """The python module containing the source classes."""

    target_python_module: str = None
    """The python module containing the target classes."""

    def compile(self, specification: TransformationSpecification) -> CompiledSpecification:
        """
        Compile a resolved transformation specification into an alternative representation.

        :param specification: A fully resolved specification (e.g. from
            ``Transformer.derived_specification``).  Must not be ``None``.
        :return: The compiled specification.
        """
        if specification is None:
            raise TypeError("compile() requires a resolved specification")
        s = self._compile_header(specification)
        for chunk in self._compile_iterator(specification):
            s += chunk
        return CompiledSpecification(serialization=s)

    def _compile_header(self, specification: TransformationSpecification) -> str:
        return ""

    def _compile_iterator(self, specification: TransformationSpecification) -> Iterator[str]:
        raise NotImplementedError

    def derived_target_schemaview(self, specification: TransformationSpecification) -> SchemaView:
        """
        Return a view over the target schema, including any derived classes.
        """
        mapper = SchemaMapper(source_schemaview=self.source_schemaview)
        return SchemaView(yaml_dumper.dumps(mapper.derive_schema(specification)))

@dataclass
class PythonCompiler(Compiler):
    """
    Compiles a Transformation Specification to Python code.
    """

    def _compile_header(self, specification: TransformationSpecification) -> str:
        s = ""
        if self.source_python_module:
            s += f"import {self.source_python_module} as src\n"
        if self.target_python_module:
            s += f"import {self.target_python_module} as tgt\n"
        s += "\nNULL = None\n\n"
        return s

    def _compile_iterator(self, specification: TransformationSpecification) -> Iterator[str]:
        for cd in specification.class_derivations:
            yield from self._compiled_class_derivations_iter(cd)

    def _compiled_class_derivations_iter(self, cd: ClassDerivation) -> Iterator[str]:
        sv = self.source_schemaview
        populated_from = cd.populated_from if cd.populated_from else cd.name
        if populated_from not in sv.all_classes():
            return
        induced_slots = {s.name: s for s in sv.class_induced_slots(populated_from)}
        t = Template(CD_TEMPLATE)
        yield t.render(
            cd=cd,
            source_module="src",
            target_module="tgt",
            induced_slots=induced_slots,
            schemaview=sv,
            source_slots=sv.class_induced_slots(populated_from),
        )

@dataclass
class MarkdownCompiler(J2BasedCompiler):
    """
    Compiles a Transformation Specification to Markdown.
    """

    template_name: str = "markdown.j2"

class GraphvizCompiler(Compiler):
    """
    Compile a Transformation Specification to GraphViz.
    """

    def compile(self, specification: TransformationSpecification, elements: list[str] | None = None) -> GraphvizObject:
        dg = Digraph(comment="UML Class Diagram", format="png")
        dg.attr(rankdir="LR")  # Set graph direction from left to right
        target_schemaview = self.derived_target_schemaview(specification)
        source_schemaview = self.source_schemaview

        records = []
        records += self.add_records(source_schemaview, "source")
        records += self.add_records(target_schemaview, "target")

        for record in records:
            dg.node(record.id, str(record), shape="plaintext")

        # Define the class nodes with fields in UML format using HTML-like labels
        # for precise control over the stacking of the fields
        for cd in specification.class_derivations:
            target_cn = cd.name
            source_cn = cd.populated_from
            if source_cn is None:
                source_cn = cd.name
            if elements is not None and target_cn not in elements:
                continue
            source_record = Record(name=source_cn, source="source")
            target_record = Record(name=target_cn, source="target")
            for sd in cd.slot_derivations.values():
                if sd.hide:
                    continue
                target_slot = sd.name
                target_id = f"{target_record.id}:{target_slot}"
                source_slot = sd.populated_from
                if source_slot:
                    source_id = f"{source_record.id}:{source_slot}"
                    dg.edge(source_id, target_id)
                elif sd.expr:
                    # TODO: do this in a less hacky way
                    tokens = re.findall(r"\w+", sd.expr)
                    for token in tokens:
                        if token not in source_schemaview.all_slots():
                            continue
                        dg.edge(f"{source_record.id}:{token}", target_id, style="dashed")
        return GraphvizObject(digraph=dg, serialization=dg.source)

    def add_records(self, schemaview: SchemaView, source: str) -> list[Record]:
        records = []
        for cn in schemaview.all_classes():
            record = Record(name=cn, source=source)
            for induced_slot in schemaview.class_induced_slots(cn):
                record.fields.append((induced_slot.name, induced_slot.range))
            records.append(record)
        return records