project: move some methods to Binary

Move some methods from the `Project` class to `model.Binary`.
This commit is contained in:
Giacomo Vercesi
2025-07-11 15:13:29 +02:00
committed by Alessandro Di Federico
parent d22d9201f5
commit 460ffbf965
6 changed files with 103 additions and 45 deletions
+62 -4
View File
@@ -2,7 +2,12 @@
# This file is distributed under the MIT License. See LICENSE.md for details.
#
from typing import Mapping
from typing import Dict, Iterable, List, Mapping, Optional, TypeAlias, Union
from revng.support.artifacts import Artifact
from revng.tupletree import StructBase
ArtifactResult: TypeAlias = Union[Artifact, Mapping[str, Artifact]]
class AllMixin:
@@ -21,9 +26,9 @@ class _ArtifactMixin(AllMixin):
def get_artifact(self, artifact_name: str):
"""
Fetch the artifacts from the `Binary`.
Fetch the artifacts that belong to the current model entity
"""
result = self._project.get_artifact(artifact_name, [self]) # type: ignore[attr-defined]
result = self._project._get_artifact(artifact_name, [self]) # type: ignore[attr-defined]
if isinstance(result, Mapping):
keys = list(result.keys())
assert len(keys) == 1
@@ -32,6 +37,59 @@ class _ArtifactMixin(AllMixin):
return result
BinaryMixin = _ArtifactMixin
class BinaryMixin(_ArtifactMixin):
def get_artifact(self, artifact_name: str, targets=None):
"""
Fetch the artifacts from the `Binary`.
"""
if targets is None:
return self._project._get_artifact(artifact_name, [self]) # type: ignore[attr-defined]
else:
return self._project._get_artifact(artifact_name, targets) # type: ignore[attr-defined]
def commit(self):
"""
Persist the changes to the backend.
"""
self._project._commit() # type: ignore[attr-defined]
def revert(self):
"""
Revert changes made since the last call to `commit()`.
"""
self._project._revert() # type: ignore[attr-defined]
def get_artifacts(
self, params: Dict[str, Optional[Iterable[Union[str, StructBase]]]]
) -> Dict[str, ArtifactResult]:
"""
Allows fetching multiple artifacts at once. The `params` is a dict
containing the name of the artifact and a list of targets (it can be
empty to fetch all the targets).
Example `params`:
params = {
"disassemble": [Function_1, Function_2],
"decompile": []
}
"""
return self._project._get_artifacts(params) # type: ignore[attr-defined]
def analyze(
self, analysis_name: str, targets: Dict[str, List[str]] = {}, options: Dict[str, str] = {}
):
"""
Run a single analysis. In addition to the `analysis_name` you need to
specify a dict of targets, some analysis require you to also pass an
`options` dict.
"""
return self._project._analyze(analysis_name, targets, options) # type: ignore[attr-defined]
def analyses_list(self, analysis_name: str):
"""
Run analysis list, these are predefined list of analysis that run sequentially.
"""
return self._project._analyses_list(analysis_name) # type: ignore[attr-defined]
FunctionMixin = _ArtifactMixin
TypeDefinitionMixin = _ArtifactMixin
+10 -10
View File
@@ -34,7 +34,7 @@ class CLIProject(Project, CLIProjectMixin, ResumeProjectMixin):
assert os.path.isfile(binary_path)
self.input_binary_path = binary_path
def _get_artifact(self, artifact_name: str, targets: Set[str]) -> bytes:
def _get_artifact_impl(self, artifact_name: str, targets: Set[str]) -> bytes:
assert self.input_binary_path is not None
args = [
"artifact",
@@ -46,15 +46,15 @@ class CLIProject(Project, CLIProjectMixin, ResumeProjectMixin):
return self._run_revng_cli(args, capture_output=True).stdout
def analyze(self, analysis_name: str, targets={}, options={}):
assert analysis_name in self._analysis_names
self._analyze(analysis_name, targets, options)
def analyses_list(self, analysis_list_name: str):
assert analysis_list_name in self._analyses_list_names
return self._analyze(analysis_list_name)
def _analyze(self, analysis_name: str, targets={}, options={}):
assert analysis_name in self._analysis_names
self._analyze_impl(analysis_name, targets, options)
def _analyses_list(self, analysis_list_name: str):
assert analysis_list_name in self._analyses_list_names
return self._analyze_impl(analysis_list_name)
def _analyze_impl(self, analysis_name: str, targets={}, options={}):
assert self.input_binary_path is not None
# TODO: add `targets` when `revng` allows it
assert len(targets) == 0
@@ -69,7 +69,7 @@ class CLIProject(Project, CLIProjectMixin, ResumeProjectMixin):
model = Binary.deserialize(result.decode("utf-8"))
self._set_model(model)
def commit(self):
def _commit(self):
diff = DiffSet.make(self.last_saved_model, self.model)
if len(diff.Changes) == 0:
return
+4 -4
View File
@@ -49,7 +49,7 @@ class DaemonProject(Project):
if not result["uploadFile"]:
raise RuntimeError("File upload failed")
def _get_artifact(self, artifact_name: str, targets: Set[str]) -> Dict[str, bytes]:
def _get_artifact_impl(self, artifact_name: str, targets: Set[str]) -> Dict[str, bytes]:
query = gql(
"""
query($step: String!, $paths: String!, $index: BigInt!) {
@@ -83,7 +83,7 @@ class DaemonProject(Project):
return container_mime.removesuffix("+tar+gz")
raise ValueError
def analyze(
def _analyze(
self, analysis_name: str, targets: Dict[str, List[str]] = {}, options: Dict[str, str] = {}
):
step = self._get_step_name(analysis_name)
@@ -114,7 +114,7 @@ class DaemonProject(Project):
assert response["runAnalysis"]["__typename"] == "Diff"
self._set_model(self._get_model())
def analyses_list(self, analysis_name: str):
def _analyses_list(self, analysis_name: str):
query = gql(
"""
mutation($analysis: String!, $index: BigInt!) {
@@ -130,7 +130,7 @@ class DaemonProject(Project):
assert response["runAnalysesList"]["__typename"] == "Diff"
self._set_model(self._get_model())
def commit(self):
def _commit(self):
diff = DiffSet.make(self.last_saved_model, self.model)
if len(diff.Changes) == 0:
return
+12 -12
View File
@@ -13,11 +13,11 @@ from subprocess import CalledProcessError, run
from tempfile import TemporaryDirectory
from typing import Dict, Iterable, List, Mapping, Optional, Set, TypeAlias, TypeVar, Union
from revng.model import iterate_fields # type: ignore[attr-defined]
from revng.model import Binary, StructBase # type: ignore[attr-defined]
from revng.model import Binary, iterate_fields # type: ignore[attr-defined]
from revng.model.mixins import AllMixin
from revng.pipeline_description import Container, PipelineDescription # type: ignore[attr-defined]
from revng.support.artifacts import Artifact
from revng.tupletree import StructBase
StructBaseT = TypeVar("StructBaseT", bound=StructBase)
ArtifactResult: TypeAlias = Union[Artifact, Mapping[str, Artifact]]
@@ -58,7 +58,7 @@ class Project(ABC):
rank_name = self.pipeline_description.Kinds[kind_name].Rank
return self._rank_targets[rank_name]
def get_artifact(
def _get_artifact(
self, artifact_name: str, targets: Optional[Iterable[Union[str, StructBaseT]]] = None
) -> ArtifactResult:
"""
@@ -88,7 +88,7 @@ class Project(ABC):
assert len(actual_targets) > 0
assert actual_targets.issubset(all_targets)
results = self._get_artifact(artifact_name, actual_targets)
results = self._get_artifact_impl(artifact_name, actual_targets)
container_mime = self._get_artifact_container(artifact_name).MIMEType
kind_name = self.pipeline_description.Steps[artifact_name].Artifacts.Kind
@@ -107,7 +107,7 @@ class Project(ABC):
return Artifact.make(results, container_mime)
@abstractmethod
def _get_artifact(
def _get_artifact_impl(
self, artifact_name: str, targets: Set[str]
) -> Union[bytes, Dict[str, bytes]]:
"""
@@ -123,7 +123,7 @@ class Project(ABC):
"""
raise ValueError
def get_artifacts(
def _get_artifacts(
self, params: Dict[str, Optional[Iterable[Union[str, StructBase]]]]
) -> Dict[str, ArtifactResult]:
"""
@@ -138,12 +138,12 @@ class Project(ABC):
"""
result: Dict[str, ArtifactResult] = {}
for artifact_name, targets in params.items():
result[artifact_name] = self.get_artifact(artifact_name, targets)
result[artifact_name] = self._get_artifact(artifact_name, targets)
return result
@abstractmethod
def analyze(
def _analyze(
self, analysis_name: str, targets: Dict[str, List[str]] = {}, options: Dict[str, str] = {}
):
"""
@@ -154,7 +154,7 @@ class Project(ABC):
pass
@abstractmethod
def analyses_list(self, analysis_name: str):
def _analyses_list(self, analysis_name: str):
"""
Run analysis list, these are predefined list of analysis that run sequentially.
"""
@@ -175,13 +175,13 @@ class Project(ABC):
return inputs
@abstractmethod
def commit(self):
def _commit(self):
"""
Persist the changes from `self.model` to the backend.
"""
pass
def revert(self):
def _revert(self):
"""
Revert changes made to the `self.model` since the last call to `commit()`.
"""
@@ -201,7 +201,7 @@ class Project(ABC):
the binary and runs the `revng` initial auto analysis.
"""
self.set_binary_path(binary_path)
self.analyses_list("revng-initial-auto-analysis")
self._analyses_list("revng-initial-auto-analysis")
def _set_model(self, model: Binary):
self.model = model