mafw.steering.models

Domain models for steering metadata construction.

Author:: Bulgheroni Antonio
Description:: Simple dataclasses that hold editable steering metadata without execution logic.

Classes

`Condition`(operator, value[, is_implicit, ...])	A single filter condition.
`ConditionalFilterConfig`(name[, enabled, ...])	Conditional filter applied to a model.
`DBConfiguration`(url, pragmas, ...)	Database section content mirrored from the steering file.
`FieldFilterConfig`(name, enabled, model, ...)	Filter applied to a single field of a specific model.
`FieldInfo`(name, type[, help_text])	Metadata about a model field.
`FilterConfig`(name[, enabled, model])	Base configuration for a filter.
`FilterKind`(*values)	The kind of filter configuration.
`GlobalSettings`(processors_to_run, ...)	Top-level globals that appear outside any processor or group table.
`GroupConfig`(name, processors, description, ...)	Configuration for a logical processor group section.
`ModelFilterConfig`(name, enabled, model, ...)	Filter applied to a specific database model.
`ModelInfo`(field_info, ...)	Metadata about a database model.
`ParameterConfig`(name[, value, default, ...])	Configuration and status for a single processor parameter.
`ParameterSchemaStatus`(*values)	Schema reconciliation status for a processor parameter.
`ParameterSource`(*values)	Origin of a processor parameter value in the GUI pipeline.
`ProcessorConfig`(name, parameters, ...)	Configuration store for an individual processor or replica section.
`ProcessorRef`(base_name[, replica])	Reference to a processor entry used in processors_to_run lists.
`ProcessorSchemaStatus`(*values)	Schema reconciliation status for a processor.
`UIConfiguration`([interface])	User-interface section content mirrored from the steering file.

class mafw.steering.models.Condition(operator: str | LogicalOp, value: Any, is_implicit: bool = True, is_valid: bool = True)[source]

Bases: object

A single filter condition.

class mafw.steering.models.ConditionalFilterConfig(name: str, enabled: bool = True, model: str = '', condition: Condition | None = None, then_clause: Condition | None = None, else_clause: Condition | None = None, condition_field: str = '', then_field: str = '', else_field: str | None = None, auto_named: bool = False)[source]

Bases: FilterConfig

Conditional filter applied to a model.

Semantics:: IF condition THEN then_clause ELSE else_clause

fill_model_info() → None[source]: Populate model_info by inspecting the database model.

is_complete() → bool[source]: Return whether every active clause of the conditional filter is valid.

class mafw.steering.models.DBConfiguration(url: str | None = None, pragmas: ~typing.Dict[str, ~typing.Any] = <factory>, attributes: ~typing.Dict[str, ~typing.Any] = <factory>, enabled: bool = True)[source]

Bases: object

Database section content mirrored from the steering file.

Parameters:

url – Optional database URL connection string.
pragmas – SQLite pragma settings to apply on connection.
attributes – Additional arbitrary attributes for the database configuration.
enabled – Whether DBConfiguration should be serialized.

class mafw.steering.models.FieldFilterConfig(name: str, enabled: bool = True, model: str = '', field_name: str = '', conditions: Dict[str, Condition] = <factory>, logic_str_original: Optional[str] = None, logic_buffer: Optional[str] = None, logic_ast: Optional[ExprNode] = None, logic_dirty: bool = False, logic_is_valid: bool = True)[source]

Bases: FilterConfig

Filter applied to a single field of a specific model.

Conditions are AND’ed by default unless a custom logical expression is defined via logic.

property logic: str | None: Backward compatibility for logic string.

class mafw.steering.models.FieldInfo(name: str, type: type, help_text: str | None = None)[source]

Bases: object

Metadata about a model field.

class mafw.steering.models.FilterConfig(name: str, enabled: bool = True, model: str = '')[source]

Bases: ABC

Base configuration for a filter.

class mafw.steering.models.FilterKind(*values)[source]

Bases: StrEnum

The kind of filter configuration.

static _generate_next_value_(name, start, count, last_values): Return the lower-cased version of the member name.

class mafw.steering.models.GlobalSettings(processors_to_run: ~typing.List[str] = <factory>, analysis_name: str | None = None, analysis_description: str | None = None, new_only: bool | None = None, create_standard_tables: bool | None = None)[source]

Bases: object

Top-level globals that appear outside any processor or group table.

Parameters:

processors_to_run – Ordered list of processor or group names to execute.
analysis_name – Optional name for the analysis.
analysis_description – Optional description for the analysis.
new_only – Optional flag to process only new data.
create_standard_tables – Optional flag to create standard output tables.

class mafw.steering.models.GroupConfig(name: str, processors: ~typing.List[str], description: str | None = None, attributes: ~typing.Dict[str, ~typing.Any] = <factory>)[source]

Bases: object

Configuration for a logical processor group section.

Parameters:

name – Section name used to reference the group in processors_to_run.
processors – Ordered list of processor names and/or nested group names.
description – Optional documentation string for the group.
attributes – Additional arbitrary attributes for the group.

class mafw.steering.models.ModelFilterConfig(name: str, enabled: bool = True, model: str = '', conditions: Dict[str, Condition] = <factory>, logic_str_original: Optional[str] = None, logic_buffer: Optional[str] = None, logic_ast: Optional[ExprNode] = None, logic_dirty: bool = False, logic_is_valid: bool = True)[source]

Bases: FilterConfig

Filter applied to a specific database model.

Each field condition is AND’ed by default unless a custom logical expression is provided via logic.

fill_model_info() → None[source]: Populate model_info by inspecting the database model.

property logic: str | None: Backward compatibility for logic string.

class mafw.steering.models.ModelInfo(field_info: ~typing.Dict[str, ~mafw.steering.models.FieldInfo] = <factory>)[source]

Bases: object

Metadata about a database model.

class mafw.steering.models.ParameterConfig(name: str, value: Any = None, default: Any = None, source: ParameterSource = ParameterSource.CONFIG, status: ParameterSchemaStatus = ParameterSchemaStatus.OK, help: str | None = None, active_override: bool = False, type: Any = None)[source]

Bases: object

Configuration and status for a single processor parameter.

Parameters:

name – The name of the parameter.
value – The current value of the parameter.
default – The default value of the parameter.
source – The source of the parameter value.
status – The schema reconciliation status of the parameter.
help – Documentation text for the parameter.
type – The type hint for the parameter.

class mafw.steering.models.ParameterSchemaStatus(*values)[source]

Bases: StrEnum

Schema reconciliation status for a processor parameter.

static _generate_next_value_(name, start, count, last_values): Return the lower-cased version of the member name.

class mafw.steering.models.ParameterSource(*values)[source]

Bases: StrEnum

Origin of a processor parameter value in the GUI pipeline.

static _generate_next_value_(name, start, count, last_values): Return the lower-cased version of the member name.

class mafw.steering.models.ProcessorConfig(name: str, parameters: Dict[str, ParameterConfig] = <factory>, processor_status: ProcessorSchemaStatus = ProcessorSchemaStatus.OK, filters: Dict[str, List[FilterConfig]] = <factory>, logic_str_original: Optional[str] = None, logic_buffer: Optional[str] = None, logic_ast: Optional[ExprNode] = None, logic_dirty: bool = False, logic_is_valid: bool = True, new_only: Optional[bool] = None, inheritance: Optional[bool] = None, has_filter_root: bool = True, schema_filter_models: List[str] = <factory>)[source]

Bases: object

Configuration store for an individual processor or replica section.

Parameters:

name – Replica-aware section name (Processor or Processor#Replica).
parameters – Parameter configurations and status.
processor_status – Schema reconciliation status for the processor.
filters – Mapping from model name to list of filter configuration objects.
logic_str_original – Optional __logic__ string stored in the per-filter table.
new_only – Optional __new_only__ override.
inheritance – Optional __inheritance__ flag for replicas.
has_filter_root – Whether the filter table includes a root entry.

property filter_logic: str | None: Backward compatibility for filter_logic string.

class mafw.steering.models.ProcessorRef(base_name: str, replica: str | None = None)[source]

Bases: object

Reference to a processor entry used in processors_to_run lists.

Parameters:

base_name – The canonical name of the processor without replica suffix.
replica – Optional replica identifier appended to the base name with #.

property full_name: str: Return the replica-aware full identifier for this reference.

class mafw.steering.models.ProcessorSchemaStatus(*values)[source]

Bases: StrEnum

Schema reconciliation status for a processor.

static _generate_next_value_(name, start, count, last_values): Return the lower-cased version of the member name.

class mafw.steering.models.UIConfiguration(interface: str = 'rich')[source]

Bases: object

User-interface section content mirrored from the steering file.

Parameters:: interface – The interface type to use (e.g., rich).

mafw.steering.models._field_type_from_peewee(field_obj: Any) → type[source]: Extract a primitive type from the given Peewee field definition.

mafw.steering.models._normalize_field_type(field_type: Any) → type[source]: Normalize a field descriptor to a supported Python primitive.

mafw.steering.models._FIELD_TYPE_STRINGS: dict[str, type] = {'AUTO': <class 'int'>, 'BIGINT': <class 'int'>, 'BOOL': <class 'bool'>, 'BOOLEAN': <class 'bool'>, 'CHAR': <class 'str'>, 'DATE': <class 'str'>, 'DATETIME': <class 'str'>, 'DECIMAL': <class 'float'>, 'DOUBLE': <class 'float'>, 'FLOAT': <class 'float'>, 'INT': <class 'int'>, 'INTEGER': <class 'int'>, 'NUMERIC': <class 'float'>, 'REAL': <class 'float'>, 'SMALLINT': <class 'int'>, 'STRING': <class 'str'>, 'TEXT': <class 'str'>, 'TIME': <class 'str'>, 'TINYINT': <class 'int'>, 'VARCHAR': <class 'str'>}: Mapping from Peewee field_type strings to Python primitives.