# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Deterministic generation of the example steering file.
The generated project ships a runnable steering file in ``conf/``. Rather than
hand-writing TOML, this module reuses MAFw's own
:func:`~mafw.tools.toml_tools.generate_steering_file` so the steering file always
matches the real parameter schema of the example processor (single source of
truth).
Two project-specific adjustments are applied on top of the generic generator:
* the database is a self-contained SQLite file written under a project-relative
``run/`` directory, expressed with environment-variable expansion so the same
steering file works on a laptop and on a headless HPC node;
* the non-deterministic "generated on <timestamp>" comment emitted by the generic
generator is normalized to a fixed line, so that generating the same project
twice yields byte-identical output.
.. versionadded:: 2.4
"""
from __future__ import annotations
import re
import tempfile
from collections.abc import Collection
from pathlib import Path
from typing import Any, cast
from mafw.lazy_import import ProcessorClassProtocol
from mafw.processor import ActiveParameter, Processor
from mafw.tools.toml_tools import generate_steering_file
# The example uses a self-contained in-memory SQLite database so it runs out of
# the box with no directory setup and leaves no files behind. For a real analysis,
# point the URL at a file under a run/ directory, optionally via an environment
# variable, for example: sqlite:///${MAFW_RUN_DIR:-run}/example.db
_SQLITE_URL = 'sqlite:///:memory:'
# Line emitted by the generic generator that carries a timestamp; replaced by a
# fixed comment to keep generation reproducible.
_TIMESTAMP_COMMENT = re.compile(r'^# MAFw steering file generated on .*$', re.MULTILINE)
_FIXED_COMMENT = '# MAFw steering file for the example pipeline'
# The section name the example processor must appear under in the steering file;
# it matches the class name used in the ``example.py`` template.
_EXAMPLE_PROCESSOR_NAME = 'ExampleProcessor'
[docs]
class _ExampleProcessorStandIn(Processor):
"""Example processor that sums the first ``last_value`` integers.
.. note::
This private class is only a *schema stand-in*: it lets the generic
steering generator introspect the example processor's parameters at
generation time, when the real generated module is not yet importable.
Its first docstring line is reused as the steering-file section comment,
so it must read like the real example processor's purpose, and it must
stay in sync with the ``example.py`` template (same name and parameters).
:meta private:
"""
last_value = ActiveParameter(
'last_value', default=10, help_doc='Sum the integers from 0 up to, but excluding, this value.'
)
[docs]
def get_items(self) -> Collection[Any]:
"""Return an empty item set; the stand-in is never executed."""
return []
[docs]
def write_example_steering_file(output_file: Path) -> None:
"""Write the deterministic example steering file to ``output_file``.
:param output_file: Destination path of the steering file (inside ``conf/``).
:type output_file: Path
"""
sqlite_conf: dict[str, Any] = {
'URL': _SQLITE_URL,
'parameters': {
'sqlite': {
'pragmas': {'journal_mode': 'wal', 'cache_size': -64000, 'foreign_keys': 1, 'synchronous': 0},
}
},
}
# Generate into a temporary file first, then normalize the timestamp comment
# before writing the final, reproducible output.
with tempfile.NamedTemporaryFile(mode='w', suffix='.toml', delete=False, encoding='utf-8') as handle:
tmp_path = Path(handle.name)
try:
generate_steering_file(tmp_path, cast(ProcessorClassProtocol, _ExampleProcessorStandIn), sqlite_conf)
content = tmp_path.read_text(encoding='utf-8')
finally:
tmp_path.unlink(missing_ok=True)
content = _TIMESTAMP_COMMENT.sub(_FIXED_COMMENT, content)
# Rename the stand-in to the template processor name, both in the available
# list entry and the section header, so the steering file matches the project.
stand_in = _ExampleProcessorStandIn.__name__
# Replace the stand-in name regardless of the quote style tomlkit emits, and
# in the section header, so no internal class name leaks into the project.
content = content.replace(f'"{stand_in}"', f"'{_EXAMPLE_PROCESSOR_NAME}'")
content = content.replace(f"'{stand_in}'", f"'{_EXAMPLE_PROCESSOR_NAME}'")
content = content.replace(f'[{stand_in}]', f'[{_EXAMPLE_PROCESSOR_NAME}]')
# The generic generator leaves ``processors_to_run`` commented out, expecting
# the user to fill it in. For a ready-to-run example we activate it with the
# example processor so ``mafw run`` works immediately after generation.
content = _activate_processors_to_run(content)
# Give the example a clear, self-explanatory analysis name and description.
content = content.replace('analysis_name = "mafw analysis"', 'analysis_name = "example-analysis"')
content = content.replace(
'analysis_description = """Summing up numbers"""',
'analysis_description = """Minimal MAFw example pipeline."""',
)
output_file.parent.mkdir(parents=True, exist_ok=True)
output_file.write_text(content, encoding='utf-8')
[docs]
def _activate_processors_to_run(content: str) -> str:
"""Replace the commented ``processors_to_run`` placeholder with an active array.
The generic steering generator emits an explanatory comment and a commented-out
``processors_to_run`` line. We replace that block with a valid array listing the
example processor, so the generated steering file is runnable as-is.
:param content: The generated steering-file text.
:type content: str
:return: The steering-file text with an active ``processors_to_run`` array.
:rtype: str
"""
comment_block = (
'# uncomment the line below and insert the processors you want to run '
'from the available processor list\n'
'# processors_to_run = []'
)
active = (
'# The example processor to run. Add or remove processor names here to '
'control the pipeline.\n'
f"processors_to_run = ['{_EXAMPLE_PROCESSOR_NAME}']"
)
return content.replace(comment_block, active, 1)