Source code for analyzer.postprocessing.processors

from __future__ import annotations

import abc
from cattrs.strategies import include_subclasses, configure_tagged_union
import functools as ft
from .plots.common import PlotConfiguration
from .style import StyleSet
from analyzer.utils.structure_tools import ItemWithMeta
from analyzer.postprocessing.grouping import GroupBuilder
from analyzer.utils.structure_tools import globWithMeta
from attrs import define, field


[docs]
ResultSet = list[list[ItemWithMeta]]



[docs]
type PostprocessingGroup = (
    dict[str, PostprocessingGroup] | list[PostprocessingGroup] | list[ItemWithMeta]
)



@define

[docs]
class BasePostprocessor(abc.ABC):

[docs]
    inputs: list[list[tuple[str, ...]]]


[docs]
    structure: GroupBuilder


[docs]
    style_set: StyleSet | None = field(default=None, kw_only=True)


[docs]
    plot_configuration: PlotConfiguration | None = field(default=None, kw_only=True)



[docs]
    def run(self, data, prefix=None):
        for i in self.inputs:
            items = [y for x in (globWithMeta(data, l) for l in i) for y in x]
            for x in self.structure.apply(items):
                yield from self.getRunFuncs(x, prefix)



[docs]
    def explain(self, data):
        traces = []
        for i in self.inputs:
            items = [y for x in (globWithMeta(data, l) for l in i) for y in x]
            input_desc = " + ".join("/".join(l) for l in i)
            trace = self.structure.explain(items)
            traces.append((input_desc, trace))
        return traces


    @abc.abstractmethod

[docs]
    def getRunFuncs(self, group: PostprocessingGroup, prefix=None):
        pass





[docs]
def configureConverter(conv):
    union_strategy = ft.partial(configure_tagged_union, tag_name="name")
    include_subclasses(BasePostprocessor, conv, union_strategy=union_strategy)

    base_hook = conv.get_structure_hook(BasePostprocessor)

    def convStr(x):
        if isinstance(x, str):
            return x.split("/")
        return x

    @conv.register_structure_hook
    def _(data, t) -> BasePostprocessor:
        real_inputs = []
        for x in data["inputs"]:
            if isinstance(x, list):
                real_inputs.append([convStr(y) for y in x])
            else:
                real_inputs.append([convStr(x)])
        data["inputs"] = real_inputs
        return base_hook(data, t)