# Copyright 2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""Reusable steering schema editing operations.
:author: Bulgheroni Antonio
"""
import tomlkit
from mafw.db.db_configurations import db_scheme, default_conf
from mafw.mafw_errors import ValidationIssue
from mafw.models.parameter_schema import ParameterSchema
from mafw.plugin_manager import get_plugin_manager
from mafw.steering.builder import SteeringBuilder, ValidationLevel
from mafw.tools.toml_tools import generate_steering_file
[docs]
class _SteeringOperationError(ValueError):
"""Describe an invalid processor selection or schema default."""
[docs]
def _create_default_steering_file(filename: str, db_engine: str, db_url: str) -> None:
"""Write a steering file using defaults from every discovered processor."""
plugin_manager = get_plugin_manager()
plugins = plugin_manager.load_plugins({'processors'})
database_conf = default_conf[db_engine]
database_conf['URL'] = db_scheme[db_engine] + db_url
generate_steering_file(filename, plugins.processor_list, database_conf)
[docs]
def _validate_steering_file(filename: str, level: str) -> list[ValidationIssue]:
"""Load required plugins and validate a steering file without modifying it."""
try:
get_plugin_manager().load_plugins(('processors', 'db_modules', 'ui'))
except Exception as exc:
raise _SteeringOperationError(f'Could not load plugins needed for steering validation: {exc}') from exc
try:
builder = SteeringBuilder.from_toml(filename)
except Exception as exc:
raise _SteeringOperationError(f'Could not read steering file: {exc}') from exc
try:
return builder.validate(ValidationLevel(level.lower()))
except Exception as exc:
raise _SteeringOperationError(f'Steering validation failed: {exc}') from exc
[docs]
def _load_steering_processors(
names: tuple[str, ...], all_processors: bool = False
) -> dict[str, tuple[list[ParameterSchema], str | None]]:
"""Resolve requested base processor names and schemas before any file is written."""
plugins = get_plugin_manager().load_plugins({'processors'})
if all_processors:
selected_names = list(plugins.processor_dict)
else:
selected_names = list(names)
resolved: dict[str, tuple[list[ParameterSchema], str | None]] = {}
unknown = [name for name in selected_names if name not in plugins.processor_dict]
if unknown:
raise _SteeringOperationError(f'Unknown processor(s): {", ".join(unknown)}')
for name in selected_names:
processor_entry = plugins.processor_dict[name]
# The lazy proxy imports its class on first attribute access; schema access never instantiates it.
processor_schema = processor_entry.processor_schema()
schema: list[ParameterSchema] = processor_schema.parameters
# Validate defaults up front so malformed values cannot trigger a partial write.
try:
for parameter in schema:
tomlkit.item(parameter.default)
except Exception as exc:
raise _SteeringOperationError(f'Unsupported default for {name}.{parameter.name}: {exc}') from exc
resolved[name] = schema, processor_schema.description
return resolved
[docs]
def _edit_steering_file(filename: str, operation: str, names: tuple[str, ...], all_processors: bool = False) -> None:
"""Stage schema changes in the steering model and persist them once validated."""
schemas = _load_steering_processors(names, all_processors)
builder = SteeringBuilder.from_toml(filename)
changed = False
for name, (schema, processor_doc) in schemas.items():
exists = name in builder.processors
added_parameters: set[str] = set()
processor_changed = False
if operation == 'add' and exists:
continue
if not exists and not schema:
builder.ensure_processor_section(name)
changed = True
processor_changed = True
for parameter in schema:
current = builder.processors.get(name)
current_parameter = current.parameters.get(parameter.name) if current else None
should_set = operation == 'reset' or current_parameter is None
if operation == 'reset' and current_parameter is not None:
should_set = current_parameter.value != parameter.default
if should_set:
builder.set_parameter(name, parameter.name, parameter.default)
changed = True
processor_changed = True
added_parameters.add(parameter.name)
if processor_changed:
builder.set_processor_documentation(
name,
processor_doc,
{parameter.name: parameter.help for parameter in schema if parameter.name in added_parameters},
)
if changed:
builder.write(filename)