mafw.steering_gui.controllers.steering_controller

Application layer controller for steering configuration builds.

Author:

Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)

Description:

Keep GUI and execution logic isolated while managing SteeringBuilder state.

Module Attributes

P	Type variable capturing the positional parameters of the wrapped method.
R	Type variable capturing the return type of the wrapped method.

Functions

build_database_connection_configuration(config)	Return a normalized configuration dict for connection checks.
recorded_mutation(method)	Decorator that records the controller state before mutating operations.

Classes

DirtyStateSignal()	Lightweight notifier for controller dirty-state transitions.
SteeringController([builder, max_history, ...])	Controller orchestrating SteeringBuilder without GUI or execution ties.

Exceptions

SteeringControllerError

Raised when the controller is asked to perform an invalid operation.

exception mafw.steering_gui.controllers.steering_controller.SteeringControllerError

Bases: MAFwException

Raised when the controller is asked to perform an invalid operation.

class mafw.steering_gui.controllers.steering_controller.DirtyStateSignal

Bases: object

Lightweight notifier for controller dirty-state transitions.

connect(callback: Callable[[bool], None]) → None

Subscribe to dirty-state change notifications.

emit(value: bool) → None

Notify every subscriber about the new dirty value.

class mafw.steering_gui.controllers.steering_controller.R

Type variable capturing the return type of the wrapped method.

alias of TypeVar(‘R’)

class mafw.steering_gui.controllers.steering_controller.SteeringController(builder: SteeringBuilder | None = None, *, max_history: int | None = None, plugin_manager: MAFwPluginManager | None = None)

Bases: object

Controller orchestrating SteeringBuilder without GUI or execution ties.

classmethod from_file(path: Path | str) → SteeringController

Instantiate a controller from an existing steering file.

static _normalize_schema_default(value: Any) → Any

Convert schema defaults into serializable values.

_assemble_connection_source(overrides: dict[str, Any] | None = None) → dict[str, Any]

Combine controller-managed attributes with override input.

_mark_clean() → None

Mark the controller as clean and notify subscribers.

_unique_processor_name(base_name: str) → tuple[str, str | None]

Return a unique processor name and replica identifier for the requested base.

_write_to_path(path: Path) → None

Serialize the builder to the provided path.

add_group(name: str, processors: Iterable[str], description: str | None = None) → None

Declare a new group section.

add_group_entry_to_group(target_group: str, entry_name: str) → None

Append a processor or group reference to a group list.

add_group_to_run_list(name: str) → None

Append a group name to the processors_to_run list if missing.

add_processor_to_run_list(base_name: str, replica: str | None = None) → None

Add a processor or replica to the execution list and ensure its section exists.

add_processors_to_group_with_defaults(group_name: str, base_names: Iterable[str]) → None

Append processors to a group and set their parameters to schema defaults.

add_processors_with_defaults(base_names: Iterable[str]) → None

Append processors to the run list and set their parameters to schema defaults.

add_replica(base_name: str, replica: str) → None

Ensure a replica section exists without altering parameters.

available_db_module_count() → int | None

Return how many database modules have been loaded (or None until available).

available_processor_count() → int | None

Return how many processor plugins have been loaded (or None until available).

available_processor_names() → list[str]

Return the available processor names from loaded plugins.

available_ui_count() → int | None

Return how many user interface plugins have been loaded (or None until available).

build_connection_test_task(config: dict[str, Any] | None = None) → Callable[[], bool]

Return a callable that validates the current database settings via Peewee.

Parameters:

config – Optional overrides supplied by the GUI connection tester.

Returns:

Callable that returns True when the connection is verified; it raises on failure.

build_pipeline(*, progress_callback: Callable[[int], None] | None = None) → ProcessorPipeline

Build the in-memory pipeline tree for the steering GUI.

can_redo() → bool

Return whether a previously undone state can be re-applied.

can_undo() → bool

Return whether the controller can revert to a previously recorded state.

clear_processor_parameters(full_name: str) → None

Clear every parameter override for a processor.

current_file() → Path | None

Return the file currently associated with persistence operations.

disable_db_configuration() → None

Disable the DBConfiguration section so it is omitted when written.

enable_db_configuration() → None

Enable the DBConfiguration section for serialization.

ensure_valid(validation_level: ValidationLevel = ValidationLevel.SEMANTIC) → None

Raise the first validation issue or return silently when the builder is valid.

export_to_path(path: Path | str) → None

Serialize the builder to the provided path without validation.

get_analysis_description() → str | None

Return the configured analysis_description safely.

get_analysis_name() → str | None

Return the configured analysis_name without exposing the builder.

get_create_standard_tables() → bool | None

Return the create_standard_tables flag.

get_db_attributes() → dict[str, Any]

Return a copy of the database attributes map.

get_db_pragmas() → dict[str, Any]

Return a copy of the DB pragmas map.

get_db_url() → str | None

Return the configured database URL.

get_globals() → dict[str, Any]

Return the globals metadata used by the UI.

get_group_attributes(name: str) → dict[str, Any]

Return a copy of a group’s attributes.

get_group_snapshot(name: str) → GroupConfig

Return a copy of a group configuration.

get_new_only() → bool | None

Return the current new_only flag.

get_processor_snapshot(full_name: str) → ProcessorConfig

Return a deep copy snapshot of a processor configuration.

get_processors_to_run() → list[str]

Return a copy of the ordered processor references.

get_ui_interface() → str

Return the configured UI interface identifier.

is_db_configuration_enabled() → bool

Return whether the DBConfiguration section will appear in serialized output.

list_groups() → list[str]

Return the declared group section names.

list_processors() → list[str]

Return the known processor section names.

load(path: Path | str) → None

Load steering metadata from disk, replace the current builder, and rebase the reset snapshot.

load_plugins() → LoadedPlugins

Load the asynchronous plugin bundles and cache the result for GUI consumers.

Returns:

Plugins container with available processors, database modules, and UIs

Return type:

LoadedPlugins

move_pipeline_entry(entry_name: str, entry_is_group: bool, source_group: str | None, target_group: str | None, position: int) → None

Move a pipeline entry across parents and adjust execution ordering.

move_processor_in_run_list(full_name: str, position: int) → None

Move a scheduled processor to a concrete slot in the execution order.

redo() → None

Re-apply the last undone snapshot.

remove_filter(full_name: str, model_name: str) → None

Delete a filter model definition.

remove_group(name: str) → None

Delete a named group section.

remove_pipeline_entries(names: Iterable[str]) → None

Remove pipeline entries from the top-level processors_to_run list.

remove_pipeline_entry(entry_name: str, source_group: str | None) → None

Remove a pipeline entry from the top-level or a group list.

remove_processor_from_run_list(full_name: str) → None

Remove a processor reference from the execution order.

remove_processor_parameter(full_name: str, key: str) → None

Remove a processor parameter override if present.

rename_group(old_name: str, new_name: str) → None

Rename a group and update all pipeline references.

rename_processor_entry(old_name: str, new_name: str, source_group: str | None) → None

Rename a processor reference and keep its configuration aligned.

replace_builder(builder: SteeringBuilder) → None

Replace the current builder with the provided instance.

reset() → None

Restore the controller to its initial state and clear history.

reset_processor_parameters(full_name: str) → None

Reset the processor parameters to their default values without dropping the overrides.

save(*, skip_validation: bool = False) → None

Persist the current steering state to the associated file path.

save_as(path: Path | str, *, skip_validation: bool = False) → None

Persist the current steering state to a new file path and track it.

set_analysis_description(value: str | None) → None

Update the top-level analysis_description setting.

set_analysis_name(value: str | None) → None

Update the top-level analysis_name setting.

set_create_standard_tables(value: bool | None) → None

Update the create_standard_tables flag.

set_db_attribute(key: str, value: Any | None) → None

Update a single entry stored under DBConfiguration.

set_db_pragmas(pragmas: Mapping[str, Any]) → None

Replace the DB pragmas map with a copy of the provided mapping.

set_db_url(value: str | None) → None

Update the DB URL while keeping builder semantics.

set_filter_conditionals(full_name: str, model_name: str, conditionals: list[Mapping[str, Any]] | None) → None

Assign __conditional__ blocks to a filter model.

set_filter_config(full_name: str, model_name: str, config: Mapping[str, Any]) → None

Replace a filter model configuration with a copy of the provided mapping.

set_filter_field(full_name: str, model_name: str, field: str, value: Any) → None

Set or update a single filter field.

set_filter_logic(full_name: str, logic: str | None) → None

Update the __logic__ string for a processor’s filters.

set_group_attributes(name: str, attributes: Mapping[str, Any]) → None

Replace a group’s attributes with a new copy.

set_group_description(name: str, description: str | None) → None

Set or clear the description for a group.

set_group_processors(name: str, processors: Iterable[str]) → None

Replace the processors list for the given group.

set_new_only(value: bool | None) → None

Update the new_only flag.

set_processor_filters(full_name: str, filters: Mapping[str, Iterable[FilterConfig]], logic: str | None) → None

Replace the whole filter dictionary and logic for a processor.

set_processor_new_only(full_name: str, value: bool | None) → None

Set the __new_only__ override for a processor section.

set_processor_parameter(full_name: str, key: str, value: Any) → None

Update a single processor parameter.

set_processors_to_run(processors: Iterable[str]) → None

Overwrite the processors_to_run list while keeping a copy of the iterable.

set_replica_inheritance(replica_full_name: str, value: bool | None) → None

Toggle replica inheritance.

set_ui_interface(interface: str) → None

Update the selected UI interface (neutral placeholder, no GUI imports).

set_validation_level(level: ValidationLevel | None) → None

Keep track of the validation level preference without affecting builder data.

undo() → None

Revert the builder to the previous snapshot.

validate(validation_level: ValidationLevel = ValidationLevel.SEMANTIC) → list[ValidationIssue]

Run validation and return every detected issue.

validation_level() → ValidationLevel | None

Return the validation level currently configured on the controller.

property dirty: bool

Expose whether the controller currently tracks unsaved changes.

mafw.steering_gui.controllers.steering_controller.build_database_connection_configuration(config: Mapping[str, Any]) → dict[str, Any]

Return a normalized configuration dict for connection checks.

The returned dictionary always contains URL plus backend-specific keys derived from extract_protocol.

mafw.steering_gui.controllers.steering_controller.recorded_mutation(method: Callable[[Concatenate[SteeringController, P]], R]) → Callable[[Concatenate[SteeringController, P]], R]

Decorator that records the controller state before mutating operations.

mafw.steering_gui.controllers.steering_controller.P = ~P

Type variable capturing the positional parameters of the wrapped method.