[ENH]: Introduce junifer.api.generate_yaml

docs/changes/newsfragments/498.feature

@@ -0,0 +1 @@
+Introduce :func:`.generate_yaml` to generate feature YAML from metadata by `Synchon Mandal`_

docs/links.inc

@@ -13,6 +13,7 @@
 .. _`INM-7`: https://www.fz-juelich.de/inm/inm-7/EN/Home/home_node.html
 .. _`julearn`: https://juaml.github.io/julearn
 .. _`junifer-data`: https://github.com/juaml/junifer-data-client
+.. _`julio`: https://github.com/juaml/julio
 
 .. _`pandas`: https://pandas.pydata.org
 .. _`pandas.DataFrame` : https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.DataFrame.html

docs/using/generate_yaml.rst

@@ -0,0 +1,14 @@
+.. include:: ../links.inc
+
+.. _generate_yaml:
+
+Generating YAML from metadata
+=============================
+
+``junifer`` stores the pipeline metadata for a run along with the extracted feature data.
+So, the metadata for all the "elements" processed with a pipeline is unique. The metadata
+contains all the necessary information to recreate the configuration used for the processing.
+
+If one wants to generate the processing YAML, :func:`.generate_yaml` can be used for that.
+The only requirement is providing the metadata which can be extracted by following the initial steps of
+:ref:`analysing results <analysing_extracted_features>`.

docs/using/index.rst

@@ -20,6 +20,7 @@ to interact with HPC and HTC systems.
    queueing
    configuring
    dumping
+   generate_yaml
 
 
 .. _using_components:

junifer/api/__init__.pyi

@@ -1,4 +1,19 @@
-__all__ = ["decorators", "collect", "queue", "run", "reset", "list_elements"]
+__all__ = [
+    "decorators",
+    "collect",
+    "queue",
+    "run",
+    "reset",
+    "list_elements",
+    "generate_yaml",
+]
 
 from . import decorators
-from .functions import collect, list_elements, reset, run, queue
+from .functions import (
+    collect,
+    generate_yaml,
+    list_elements,
+    reset,
+    run,
+    queue,
+)

junifer/api/functions.py

@@ -6,9 +6,12 @@
 # License: AGPL
 
 import atexit
+import datetime as dt
+import io
 import os
 import shutil
 from pathlib import Path
+from typing import TYPE_CHECKING, Any
 
 import structlog
 
@@ -32,7 +35,18 @@
 from ..utils import raise_error, warn_with_log, yaml
 
 
-__all__ = ["collect", "list_elements", "queue", "reset", "run"]
+if TYPE_CHECKING:
+    from ruamel.yaml.comments import CommentedMap
+
+
+__all__ = [
+    "collect",
+    "generate_yaml",
+    "list_elements",
+    "queue",
+    "reset",
+    "run",
+]
 
 _log = structlog.get_logger("junifer")
 logger = _log.bind(pkg="api")
@@ -468,3 +482,147 @@ def list_elements(
         elements_to_list.append(str_element)
 
     return "\n".join(elements_to_list)
+
+
+def _dg_dump_exclude(dg: str) -> set[str]:
+    """Generate datagrabber model dump exclusion set.
+
+    Parameters
+    ----------
+    dg : str
+        The datagrabber kind.
+
+    Returns
+    -------
+    set of str
+
+    """
+    if dg == "PatternDataGrabber":
+        return set()
+    elif dg == "PatternDataladDataGrabber":
+        return {
+            "datadir",
+            "datalad_dirty",
+            "datalad_commit_id",
+            "datalad_id",
+        }
+    else:
+        return {
+            # from PatternDataGrabber
+            "patterns",
+            "replacements",
+            "confounds_format",
+            "partial_pattern_ok",
+            # from DataladDataGrabber
+            "uri",
+            "rootdir",
+            "datadir",
+            "datalad_dirty",
+            "datalad_commit_id",
+            "datalad_id",
+        }
+
+
+def generate_yaml(meta: dict) -> "CommentedMap":
+    """Generate the feature YAML from metadata.
+
+    Parameters
+    ----------
+    meta : dict
+        Feature metadata as dictionary.
+
+    Returns
+    -------
+    ruamel.yaml.comments.CommentedMap
+        Feature YAML.
+
+    """
+    y: dict[str, Any] = {}
+    y["workdir"] = ""
+    # Add "with" section if present
+    if "with" in meta:
+        y["with"] = meta["with"].copy()
+    # Set datagrabber
+    meta_dg = meta["datagrabber"].copy()
+    a = meta_dg.pop("class")
+    dg = PipelineComponentRegistry().get_class(step="datagrabber", name=a)
+    dg_model = dg.model_construct(**meta_dg)
+    y["datagrabber"] = {
+        "kind": a,
+        **dg_model.model_dump(
+            mode="json",
+            exclude=_dg_dump_exclude(a),
+            exclude_defaults=True,
+            exclude_none=True,
+        ),
+    }
+    # Set preprocessor(s)
+    if "preprocess" in meta:
+        y["preprocess"] = []
+        meta_p = meta["preprocess"].copy()
+        if not isinstance(meta_p, list):
+            meta_p = [meta_p]
+        for mp in meta_p:
+            b = mp.pop("class")
+            p = PipelineComponentRegistry().get_class(
+                step="preprocessing", name=b
+            )
+            p_model = p.model_construct(**mp)
+            y["preprocess"].append(
+                {
+                    "kind": b,
+                    **p_model.model_dump(
+                        mode="json",
+                        exclude={"required_data_types"},
+                        exclude_defaults=True,
+                        exclude_none=True,
+                    ),
+                }
+            )
+    # Set marker
+    meta_m = meta["marker"].copy()
+    c = meta_m.pop("class")
+    m = PipelineComponentRegistry().get_class(step="marker", name=c)
+    m_model = m.model_construct(**meta_m)
+    y["markers"] = []
+    y["markers"].append(
+        {
+            "kind": c,
+            **m_model.model_dump(
+                mode="json",
+                exclude_defaults=True,
+                exclude_none=True,
+            ),
+        }
+    )
+    # Set storage
+    y["storage"] = {
+        "kind": "HDF5FeatureStorage",
+        "uri": "",
+    }
+    # Set queue
+    if "queue" in meta:
+        y["queue"] = meta["queue"].copy()
+    else:
+        y["queue"] = {
+            "jobname": meta["name"],
+            "kind": "",
+        }
+    # Dump and load yaml to format
+    f = io.StringIO()
+    yaml.dump(y, stream=f)
+    f.seek(0)
+    d = yaml.load(f)
+    # Add preamble
+    pre = (
+        "Auto-generated by junifer on "
+        f"{dt.datetime.now(tz=dt.UTC).strftime('%Y-%m-%d %H:%M:%S')} UTC\n\n"
+    )
+    if "dependencies" in meta:
+        for k, v in meta["dependencies"].items():
+            pre += f"{k}=={v}\n"
+    d.yaml_set_start_comment(pre)
+    # Add newline between sections
+    for s in d.keys():
+        d.yaml_set_comment_before_after_key(s, before="\n")
+    return d