# 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:'
# Storage volumes have persistent UUID identities in the database. The
# storage-enabled starter project therefore needs a file-backed SQLite DB.
_SQLITE_STORAGE_URL = 'sqlite:///run/analysis.db'
# 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, import_name: str | None = None, *, with_storage: bool = False
) -> 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
:param import_name: The project import name, used to point the
``migration_directory`` entry at the project's ``db/migrations`` folder.
When ``None`` the migration directory is left at its default.
:type import_name: str | None
:param with_storage: Include one example filesystem volume for portable
storage workflows.
:type with_storage: bool
"""
sqlite_conf: dict[str, Any] = {
'URL': _SQLITE_STORAGE_URL if with_storage else _SQLITE_URL,
'parameters': {
'sqlite': {
'pragmas': {'journal_mode': 'wal', 'cache_size': -64000, 'foreign_keys': 1, 'synchronous': 0},
}
},
}
if import_name is not None:
# The steering file lives in ``conf/``; point the migration directory at
# the project's versioned ``db/migrations`` folder relative to it. The
# migration service resolves this relative path against the steering
# file's own directory.
sqlite_conf['migration_directory'] = f'../src/{import_name}/db/migrations'
# 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)
content = content.replace(
_FIXED_COMMENT,
f'{_FIXED_COMMENT}\n'
'# Project configuration and secrets are kept in separate dotenv files.\n'
f"env_files = ['{import_name}.env', 'secrets.env']",
1,
)
if with_storage:
volume_root_variable = f'{import_name.upper()}_STORAGE_ROOT' if import_name else 'ANALYSIS_STORAGE_ROOT'
content += (
'\n# Optional logical storage volume. Its root is configured in the dotenv file above.\n'
'[StorageConfiguration.example]\n'
f'root = "${{{volume_root_variable}}}"\n'
'type = "filesystem"\n'
'access = "full"\n'
)
env_path = output_file.parent / f'{import_name}.env' if import_name else None
if env_path is not None and env_path.exists():
env_path.write_text(
env_path.read_text(encoding='utf-8')
+ f'\n# Root for the example logical storage volume.\n{volume_root_variable}=../run/storage\n',
encoding='utf-8',
)
else:
content += (
'\n# To use portable logical storage, see '
'https://mafw-e8abfd.pages.code.europa.eu/doc/latest/doc_storage.html\n'
)
# 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)
if with_storage:
(output_file.parent.parent / 'run' / 'storage').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)