revng-revng/python/revng/api/manager.py

#
# This file is distributed under the MIT License. See LICENSE.md for details.
#

from tempfile import TemporaryDirectory
from typing import Dict, Iterable, List, Mapping, Optional, Union

import yaml

from revng.pipeline_description import YamlLoader  # type: ignore

from ._capi import _api, ffi
from .errors import Error, Expected
from .exceptions import RevngException
from .invalidations import Invalidations, ResultWithInvalidations
from .string_map import StringMap
from .target import ContainerToTargetsMap, Target, TargetsList
from .utils import make_c_string, make_python_string


class Manager:
    def __init__(
        self,
        workdir: Optional[str] = None,
        flags: Iterable[str] = (),
    ):
        if workdir is None:
            self.temporary_workdir = TemporaryDirectory(prefix="revng-manager-workdir-")
            self.workdir = self.temporary_workdir.name
        else:
            self.workdir = workdir

        _flags = [make_c_string(s) for s in flags]
        _workdir = make_c_string(self.workdir)

        # Ensures that the _manager property is always defined even if the API call fails
        self._manager = None

        self._manager = _api.rp_manager_create(
            len(_flags),
            _flags,
            _workdir,
        )

        assert self._manager, "Failed to instantiate manager"

        _description = _api.rp_manager_get_pipeline_description(self._manager)
        self._description_str = make_python_string(_description)
        self.description = yaml.load(self._description_str, Loader=YamlLoader)

    @property
    def uid(self) -> int:
        return int(ffi.cast("uintptr_t", self._manager))

    def save(self):
        return _api.rp_manager_save(self._manager)

    # description utilities

    def kind_from_name(self, name: str):
        return next((k for k in self.description.Kinds if k.Name == name), None)

    def rank_from_name(self, name: str):
        return next((r for r in self.description.Ranks if r.Name == name), None)

    def analysis_from_name(self, step: str, name: str):
        step_obj = self.step_from_name(step)
        if step_obj is None:
            return None

        return next((a for a in step_obj.Analyses if a.Name == name), None)

    def analyses_list_from_name(self, name: str):
        return next((al for al in self.description.AnalysesLists if al.Name == name), None)

    def step_from_name(self, name: str):
        return next((s for s in self.description.Steps if s.Name == name), None)

    def container_from_name(self, name: str):
        return next((c for c in self.description.Containers if c.Name == name), None)

    # Pointer getters

    def _get_step_ptr(self, step_name: str):
        if self.step_from_name(step_name) is None:
            raise RevngException(f"Invalid step {step_name}")

        return _api.rp_manager_get_step_from_name(self._manager, make_c_string(step_name))

    def _get_container_identifier_ptr(self, container_name: str):
        if self.container_from_name(container_name) is None:
            raise RevngException(f"Invalid container {container_name}")

        return _api.rp_manager_get_container_identifier_from_name(
            self._manager, make_c_string(container_name)
        )

    def _get_container_ptr_from_step(self, step_name, step_ptr, container_name: str):
        identifier_ptr = self._get_container_identifier_ptr(container_name)

        container_ptr = _api.rp_step_get_container(step_ptr, identifier_ptr)
        if container_ptr == ffi.NULL:
            raise RevngException(f"Step {step_name} does not use container {container_name}")

        return step_ptr, container_ptr

    def _get_step_container_ptr(self, step_name: str, container_name: str):
        step_ptr = self._get_step_ptr(step_name)
        return self._get_container_ptr_from_step(step_name, step_ptr, container_name)

    # Target-related Functions

    def deserialize_target(self, serialized_target: str) -> Optional[Target]:
        step_name, container_name, target = serialized_target.split("/", 2)
        target_path, kind_name = target.split(":", 1)
        targets = self.get_targets(step_name, container_name)
        return next((t for t in targets if t.serialize() == target), None)

    def _produce_target(
        self,
        step_name: str,
        target: Union[Target, List[Target]],
        container_name: str,
    ) -> Dict[str, str | bytes]:
        if isinstance(target, Target):
            targets = [target]
        else:
            targets = target

        _step, _container = self._get_step_container_ptr(step_name, container_name)
        product = _api.rp_manager_produce_targets(
            self._manager, len(targets), [t._target for t in targets], _step, _container
        )
        if not product:
            # TODO: we really should be able to provide a detailed error here
            raise RevngException("Failed to produce targets")

        result = {}
        for target in targets:
            extracted_target = target.extract()
            if extracted_target is None:
                raise RevngException(f"Target {target.serialize()} extraction failed")
            result[target.serialize()] = extracted_target
        return result

    def produce_target(
        self,
        step_name: str,
        target: None | str | List[str],
        container_name: Optional[str] = None,
        only_if_ready=False,
    ) -> Dict[str, str | bytes]:
        step = self.step_from_name(step_name)
        if step is None:
            raise RevngException(f"Invalid step {step_name}")

        if container_name is not None:
            container = container_name
        else:
            container = step.Artifacts.Container
            if container == "":
                raise RevngException(f"Step {step_name} does not have an artifacts container")

        if target is None:
            _targets = [
                "",
            ]
        elif isinstance(target, str):
            _targets = [
                target,
            ]
        else:
            _targets = target

        targets: List[Target] = []
        for _target_elem in _targets:
            targets.append(
                self.create_target(step_name, container, _target_elem, container_name is None)
            )

        if only_if_ready and any(not t.is_ready for t in targets):
            raise RevngException("Requested production of unready targets")

        product = self._produce_target(step_name, targets, container)
        if not product:
            # TODO: we really should be able to provide a detailed error here
            raise RevngException("Failed to produce target")

        return product

    def create_target(
        self, step_name: str, container_name: str, target_path: str, use_artifact_kind: bool
    ) -> Target:
        step_ptr, container_ptr = self._get_step_container_ptr(step_name, container_name)
        step_obj = self.step_from_name(step_name)
        assert step_obj is not None

        if use_artifact_kind:
            if step_obj.Artifacts.Kind == "":
                raise RevngException("Step does not have an artifacts kind")

            kind_name = step_obj.Artifacts.Kind
            kind = self.kind_from_name(kind_name)
            assert kind is not None

            path_components = target_path.split("/") if target_path != "" else []
        else:
            path, kind_name = target_path.rsplit(":", 1)
            kind = self.kind_from_name(kind_name)
            if kind is None:
                raise RevngException(f"Kind {kind_name} does not exist")

            path_components = path.split("/") if path != "" else []

        rank = self.rank_from_name(kind.Rank)  # type: ignore
        if rank is not None and len(path_components) != rank.Depth:
            raise RevngException("Path components need to equal kind rank")

        _kind = _api.rp_manager_get_kind_from_name(self._manager, make_c_string(kind_name))
        assert _kind != ffi.NULL

        target = Target.create(_kind, container_ptr, path_components)
        if target is None:
            raise RevngException("Invalid target")

        return target

    # Utility target functions

    def get_targets(self, step_name: str, container_name: str) -> List[Target]:
        _, container_ptr = self._get_step_container_ptr(step_name, container_name)

        targets_list = _api.rp_manager_get_container_targets_list(self._manager, container_ptr)

        if targets_list != ffi.NULL:
            return list(TargetsList(targets_list, container_ptr))
        else:
            return []

    # Analysis handling

    def run_analysis(
        self,
        step_name: str,
        analysis_name: str,
        target_mapping: Dict[str, List[str]],
        options: Dict[str, str] | None = None,
    ) -> ResultWithInvalidations[Dict[str, str]]:
        step = self.step_from_name(step_name)
        if step is None:
            raise RevngException(f"Invalid step {step_name}")

        analysis = self.analysis_from_name(step_name, analysis_name)
        if analysis is None:
            raise RevngException(f"Invalid Analysis {analysis_name}")

        concrete_target_mapping = {}
        for container_name, target_list in target_mapping.items():
            concrete_targets = [
                self.create_target(step_name, container_name, target, False)
                for target in target_list
            ]

            analysis_ci = next(ci for ci in analysis.ContainerInputs if ci.Name == container_name)
            for target in concrete_targets:
                if target.kind is None or target.kind not in analysis_ci.AcceptableKinds:
                    raise RevngException(
                        f"Wrong kind for analysis: found '{target.kind}', "
                        + f"expected: {[analysis_ci.AcceptableKinds]}"
                    )
            concrete_target_mapping[container_name] = concrete_targets

        options_map = StringMap(options)
        analysis_result = self._run_analysis(
            step_name, analysis_name, concrete_target_mapping, options_map
        )
        if analysis_result.result is None:
            raise RevngException("Failed to run analysis")
        return analysis_result  # type: ignore

    def _run_analysis(
        self,
        step_name: str,
        analysis_name: str,
        target_mapping: Dict[str, List[Target]],
        options: StringMap,
    ) -> ResultWithInvalidations[Optional[Dict[str, str]]]:
        target_map = ContainerToTargetsMap()
        step_ptr = self._get_step_ptr(step_name)

        for container, targets in target_mapping.items():
            _, container_ptr = self._get_container_ptr_from_step(step_name, step_ptr, container)
            for target in targets:
                target_map.add(container_ptr, target)

        invalidations = Invalidations()
        result = _api.rp_manager_run_analysis(
            self._manager,
            make_c_string(step_name),
            make_c_string(analysis_name),
            target_map._map,
            invalidations._invalidations,
            options._string_map,
        )

        if result != ffi.NULL and not _api.rp_diff_map_is_empty(result):
            self.save()
        return ResultWithInvalidations(
            self.parse_diff_map(result) if result != ffi.NULL else None, invalidations
        )

    def run_analyses_list(
        self, analyses_list_name: str, options: Mapping[str, str] | None = None
    ) -> ResultWithInvalidations[Optional[Dict[str, str]]]:
        if self.analyses_list_from_name(analyses_list_name) is None:
            raise RevngException(f"Could not find analyses list {analyses_list_name}")

        options_map = StringMap(options)
        invalidations = Invalidations()
        result = _api.rp_manager_run_analyses_list(
            self._manager,
            make_c_string(analyses_list_name),
            invalidations._invalidations,
            options_map._string_map,
        )

        if result != ffi.NULL and not _api.rp_diff_map_is_empty(result):
            self.save()
        return ResultWithInvalidations(
            self.parse_diff_map(result) if result != ffi.NULL else None, invalidations
        )

    def parse_diff_map(self, diff_map) -> Dict[str, str]:
        result = {}
        for global_name in self.description.Globals:
            _diff_value = _api.rp_diff_map_get_diff(diff_map, make_c_string(global_name))
            if _diff_value != ffi.NULL:
                result[global_name] = make_python_string(_diff_value)
        return result

    # Global Handling & misc.

    def get_global(self, name) -> str:
        _name = make_c_string(name)
        _out = _api.rp_manager_create_global_copy(self._manager, _name)
        return make_python_string(_out)

    def set_global(self, name, content) -> ResultWithInvalidations[Expected[bool]]:
        _name = make_c_string(name)
        _content = make_c_string(content)
        invalidations = Invalidations()
        error = Error()
        res = _api.rp_manager_set_global(
            self._manager, _content, _name, invalidations._invalidations, error._error
        )
        return ResultWithInvalidations(Expected(res, error), invalidations)

    def verify_global(self, name, content) -> Expected[bool]:
        _name = make_c_string(name)
        _content = make_c_string(content)
        error = Error()
        res = _api.rp_manager_verify_global(self._manager, _content, _name, error._error)
        return Expected(res, error)

    def apply_diff(self, name, diff) -> ResultWithInvalidations[Expected[bool]]:
        _name = make_c_string(name)
        _diff = make_c_string(diff)
        invalidations = Invalidations()
        error = Error()
        res = _api.rp_manager_apply_diff(
            self._manager, _diff, _name, invalidations._invalidations, error._error
        )
        return ResultWithInvalidations(Expected(res, error), invalidations)

    def verify_diff(self, name, diff) -> Expected[bool]:
        _name = make_c_string(name)
        _diff = make_c_string(diff)
        error = Error()
        res = _api.rp_manager_verify_diff(self._manager, _diff, _name, error._error)
        return Expected(res, error)

    def set_input(self, container_name: str, content: bytes, _key=None):
        step_ptr = self._get_step_ptr("begin")

        _content = ffi.from_buffer(content)
        success = _api.rp_manager_container_deserialize(
            self._manager, step_ptr, make_c_string(container_name), _content, len(_content)
        )

        if not success:
            raise RevngException(
                f"Failed loading user provided input for container {container_name}"
            )

    def get_pipeline_description(self):
        return self._description_str