Files
revng-revng/python/revng/pypeline/analysis.py
Giacomo Vercesi 2911fb3e2a pypeline: implement proper config parsing
Overhaul the pipeline configuration logic by collapsing all dynamic
configuration options, both for analyses and pipes into a single
dictionary. Change all the interfaces so that there is no longer
distinction between the configuration of an analysis and of pipes.
Expose these options to the command line via `--{name}-configuration`
options for each pipe/analysis that is applicable to the command-line
invocation.
2026-04-09 17:07:44 +02:00

83 lines
2.3 KiB
Python

#
# This file is distributed under the MIT License. See LICENSE.md for details.
#
from dataclasses import dataclass
from .container import Container
from .model import Model
from .object import ObjectSet
from .utils.cabc import ABC, abstractmethod
class Analysis(ABC):
"""
An analysis makes changes to the model. In order to do this, it might inspect
previously produced results of the pipeline.
An analysis is the way in which users are expected to make changes to the model.
The changes applied by a run of an analysis might lead to invalidate certain
objects in save points.
"""
name: str
def __init__(self):
pass
@classmethod
@abstractmethod
def signature(cls) -> tuple[type[Container], ...]:
"""
The containers required by the analysis, it needs to be a class property
to auto-generate the CLI commands.
"""
raise NotImplementedError()
@abstractmethod
def run(
self,
model: Model,
containers: list[Container],
incoming: list[ObjectSet],
configuration: str,
):
"""
Run the analysis on the model, using the containers and
incoming requests. The analysis will modify inplace the
model so you must make a copy before running it, so you
can compute the diff for invalidation purposes.
"""
raise NotImplementedError()
def is_available(self) -> bool:
"""
Asks if the analysis is available for running. Running an analysis that
is not available for running will result in an error.
"""
return True
@dataclass(frozen=True, slots=True)
class AnalysisList:
"""A named list of analysis to run in sequence."""
name: str
analyses: list[str]
description: str | None = None
def __post_init__(self):
if len(self.analyses) == 0:
raise ValueError("An analysis list must contain at least one analysis.")
if len(set(self.analyses)) != len(self.analyses):
raise ValueError("An analysis list cannot contain duplicate analyses.")
def to_dict(self) -> dict:
"""Convert the data into a dictionary representation."""
return {
"name": self.name,
"analyses": self.analyses,
"description": self.description,
}