engine

class AdjusterRunner(*, phase: Phase, active_sections: Iterable[str] | None, changed_keys: Iterable[tuple[str, ...]] | None, config: RunnerConfig = RunnerConfig(trace=True, trace_paths=('vasculature.binarization', 'vasculature.performance.binarization'), trace_sink=None), schema_registry: Any | None = None)

Bases: object

Orchestrates adjuster selection, ordering, execution, patch aggregation, tracing, and (optionally) authority enforcement.

Design goals:

• keep run() readable

• isolate tracing noise

• isolate policy/enforcement hooks

• keep behavior compatible with the prior run_adjusters() function

run(*, view: Mapping[str, Any], ctx: AdjustmentContext) → Dict[str, Any]

class RunnerConfig(trace: 'bool' = True, trace_paths: 'Tuple[str, ...]' = ('vasculature.binarization', 'vasculature.performance.binarization'), trace_sink: 'Optional[Callable[[str], None]]' = None)

Bases: object

trace: bool = True

trace_paths: Tuple[str, ...] = ('vasculature.binarization', 'vasculature.performance.binarization')

trace_sink: Callable[[str], None] | None = None

inject_sm_view(sm, *, view: dict, section_name: str = 'sample')

Temporarily make sm.config read from view[section_name] instead of the coordinator. We swap the instance’s class to a dynamic subclass that only overrides the config property.

run_adjusters(*, view: Mapping[str, Any], ctx: AdjustmentContext, phase: Phase = Phase.PRE_VALIDATE, active_sections: Iterable[str] | None = None, changed_keys: Iterable[tuple[str, ...]] | None = None, trace: bool = True, trace_paths: Iterable[str] | None = None, trace_sink: Callable[[str], None] | None = None, schema_registry: Any | None = None) → Dict[str, Any]

Wrapper over AdjusterRunner.

sections_from_keys(keys: Iterable[tuple[str, ...]] | None) → Set[str] | None