Add LHAPDF parser to genpdf

CHANGELOG.md

@@ -15,6 +15,9 @@ Items without prefix refer to a global change.
 
 ## [Unreleased](https://github.com/NNPDF/eko/compare/v0.15.3...HEAD)
 
+### Changed
+- py: Add LHAPDF parser to `genpdf` (soft breaking change) ([#506](https://github.com/NNPDF/eko/pull/506))
+
 ## [0.15.3](https://github.com/NNPDF/eko/compare/v0.15.2...v0.15.3) - 2026-02-05
 
 ### Added

src/ekobox/evol_pdf.py

@@ -9,6 +9,7 @@
 from eko.runner import managed
 
 from . import apply, genpdf, info_file
+from .genpdf.parser import LhapdfDataFile
 from .utils import regroup_evolgrid
 
 DEFAULT_NAME = "eko.tar"
@@ -111,7 +112,9 @@ def evolve_pdfs(
         genpdf.install_pdf(name)
 
 
-def collect_blocks(evolved_PDF: dict, q2block_per_nf: dict, xgrid: list):
+def collect_blocks(
+    evolved_PDF: dict, q2block_per_nf: dict, xgrid: list
+) -> LhapdfDataFile:
     """Collect all LHAPDF blocks for a given replica.
 
     Parameters
@@ -139,4 +142,4 @@ def pdf_xq2(pid, x, Q2):
             pids=np.array(br.flavor_basis_pids),
         )
         all_blocks.append(block)
-    return all_blocks
+    return LhapdfDataFile(header={}, blocks=all_blocks)

src/ekobox/genpdf/__init__.py

@@ -12,6 +12,7 @@
 from eko.io.types import EvolutionPoint as EPoint
 
 from . import export, flavors, load
+from .parser import LhapdfDataBlock, LhapdfDataFile
 
 logger = logging.getLogger(__name__)
 
@@ -21,7 +22,7 @@ def take_data(
     members: bool = False,
     xgrid: Optional[List[float]] = None,
     evolgrid: Optional[List[EPoint]] = None,
-):
+) -> tuple[dict, list[dict[str, str]], list[LhapdfDataFile]]:
     """Auxiliary function for `generate_pdf`.
 
     It provides the info, the heads of the member files and the blocks
@@ -60,35 +61,46 @@ def take_data(
             else:
                 toylh = toy.mkPDF("", 0)
             all_blocks.append(
-                [
-                    generate_block(
-                        toylh.xfxQ2, xgrid, sorted_q2grid, list(br.flavor_basis_pids)
-                    )
-                ]
+                LhapdfDataFile(
+                    header={"PdfType": "central"},
+                    blocks=[
+                        generate_block(
+                            toylh.xfxQ2,
+                            xgrid,
+                            sorted_q2grid,
+                            list(br.flavor_basis_pids),
+                        )
+                    ],
+                )
             )
 
         else:
             info = load.load_info_from_file(parent_pdf_set)
             # iterate on members
             for m in range(int(info["NumMembers"])):
-                head, blocks = load.load_blocks_from_file(parent_pdf_set, m)
-                heads.append(head)
-                all_blocks.append(blocks)
+                f = load.load_blocks_from_file(parent_pdf_set, m)
+                heads.append(f.header)
+                all_blocks.append(f)
                 if not members:
                     break
     elif isinstance(parent_pdf_set, dict):
         info = copy.deepcopy(load.template_info)
         all_blocks.append(
-            [
-                generate_block(
-                    lambda pid, x, Q2: (
-                        0.0 if pid not in parent_pdf_set else parent_pdf_set[pid](x, Q2)
-                    ),
-                    xgrid,
-                    sorted_q2grid,
-                    list(br.flavor_basis_pids),
-                )
-            ]
+            LhapdfDataFile(
+                header={"PdfType": "central"},
+                blocks=[
+                    generate_block(
+                        lambda pid, x, Q2: (
+                            0.0
+                            if pid not in parent_pdf_set
+                            else parent_pdf_set[pid](x, Q2)
+                        ),
+                        xgrid,
+                        sorted_q2grid,
+                        list(br.flavor_basis_pids),
+                    )
+                ],
+            )
         )
     else:
         raise ValueError("Unknown parent pdf type")
@@ -188,7 +200,7 @@ def generate_pdf(
         info["ForcePositive"] = 0
     info["NumMembers"] = len(new_all_blocks)
     # exporting
-    export.dump_set(name, info, new_all_blocks, pdf_type_list=heads)
+    export.dump_set(name, info, new_all_blocks, heads)
 
     # install
     if install:
@@ -220,12 +232,14 @@ def install_pdf(name):
 
 def generate_block(
     xfxQ2: Callable, xgrid: List[float], sorted_q2grid: List[float], pids: List[int]
-) -> dict:
+) -> LhapdfDataBlock:
     """Generate an LHAPDF data block from a callable."""
-    block: dict = dict(mu2grid=sorted_q2grid, pids=pids, xgrid=xgrid)
+    qgrid = np.sqrt(sorted_q2grid)
     data = []
     for x in xgrid:
         for mu2 in sorted_q2grid:
             data.append(np.array([xfxQ2(pid, x, mu2) for pid in pids]))
-    block["data"] = np.array(data)
-    return block
+    data = np.array(data)
+    return LhapdfDataBlock(
+        xgrid=np.array(xgrid), qgrid=qgrid, pids=np.array(pids), data=data
+    )

src/ekobox/genpdf/export.py

@@ -5,89 +5,9 @@
 import re
 from typing import Optional
 
-import numpy as np
 import yaml
 
-
-def list_to_str(ls, fmt="%.6e"):
-    """Convert a list of numbers to a string.
-
-    Parameters
-    ----------
-    ls : list(float)
-        list
-    fmt : str
-        format string
-
-    Returns
-    -------
-    str :
-        final string
-    """
-    return " ".join([fmt % x for x in ls])
-
-
-def array_to_str(ar):
-    """Convert an array of numbers to a string.
-
-    Parameters
-    ----------
-    ar : list(list(float))
-        array
-
-    Returns
-    -------
-    str :
-        final string
-    """
-    table = ""
-    for line in ar:
-        table += f"{line[0]:.8e} " + list_to_str(line[1:], fmt="%.8e") + "\n"
-    return table
-
-
-def dump_blocks(
-    name: str, member: int, blocks: list[dict], pdf_type: Optional[str] = None
-) -> pathlib.Path:
-    """Write LHAPDF data file.
-
-    Parameters
-    ----------
-    name : str or os.PathLike
-        target name or path
-    member : int
-        PDF member
-    blocks : list(dict)
-        pdf blocks of data
-    pdf_type : str
-        PdfType to be copied in the head of member files
-
-    Returns
-    -------
-        pathlib.Path : target file
-    """
-    path_name = pathlib.Path(name)
-    if path_name.suffix == ".dat":
-        target = path_name
-    else:
-        target = path_name / f"{path_name.stem}_{member:04d}.dat"
-    target.parent.mkdir(exist_ok=True)
-    with open(target, "w", encoding="utf-8") as o:
-        if pdf_type is None:
-            if member == 0:
-                o.write("PdfType: central\n")
-            else:
-                o.write("PdfType: replica\n")
-        else:
-            o.write(pdf_type)
-        o.write("Format: lhagrid1\n---\n")
-        for b in blocks:
-            o.write(list_to_str(b["xgrid"]) + "\n")
-            o.write(list_to_str(list(np.sqrt(b["mu2grid"]))) + "\n")
-            o.write(list_to_str(b["pids"], "%d") + "\n")
-            o.write(array_to_str(b["data"]))
-            o.write("---\n")
-    return target
+from .parser import LhapdfDataFile
 
 
 def dump_info(name: str, info: dict) -> pathlib.Path:
@@ -125,23 +45,35 @@ def dump_info(name: str, info: dict) -> pathlib.Path:
     return target
 
 
-def dump_set(name, info, member_blocks, pdf_type_list=None):
+def dump_set(
+    name: str,
+    info: dict,
+    member_files: list[LhapdfDataFile],
+    header_list: Optional[list[dict[str, str]]] = None,
+) -> None:
     """Dump a whole set.
 
     Parameters
     ----------
-    name : str
+    name :
         target name
-    info : dict
+    info :
         info dictionary
-    member_blocks : list(list(dict))
+    member_files :
         blocks for all members
-    pdf_type : list(str)
-        list of strings to be copied in the head of member files
+    header_list :
+        list of optional additional headers to be copied in the head of member files
     """
     dump_info(name, info)
-    for mem, blocks in enumerate(member_blocks):
-        if not isinstance(pdf_type_list, list) or len(pdf_type_list) == 0:
-            dump_blocks(name, mem, blocks)
+    for mem, f in enumerate(member_files):
+        # update header if necessary
+        if header_list is not None and len(header_list) > 0:
+            f.header.update(header_list[mem])
+        # find path
+        path_name = pathlib.Path(name)
+        if path_name.suffix == ".dat":
+            target = path_name
         else:
-            dump_blocks(name, mem, blocks, pdf_type=pdf_type_list[mem])
+            target = path_name / f"{path_name.stem}_{mem:04d}.dat"
+        target.parent.mkdir(exist_ok=True)
+        f.write(target)

src/ekobox/genpdf/flavors.py

@@ -6,6 +6,8 @@
 
 from eko import basis_rotation as br
 
+from .parser import LhapdfDataBlock, LhapdfDataFile
+
 
 def pid_to_flavor(pids):
     """Create flavor representations from PIDs.
@@ -50,26 +52,27 @@ def evol_to_flavor(labels):
     return np.array(ps)
 
 
-def project(blocks, reprs):
+def project(f: LhapdfDataFile, reprs) -> LhapdfDataFile:
     """Project some combination of flavors defined by reprs from the blocks.
 
     Parameters
     ----------
-    blocks : list(dict)
+    blocks :
         PDF blocks
-    reprs : list(int)
+    reprs :
         active distributions in flavor representation
 
     Returns
     -------
-    list(dict) :
+    LhapdfDataFile :
         filtered blocks
     """
-    new_blocks = copy.deepcopy(blocks)
-    for block in new_blocks:
-        current_pids = block["pids"]
-        current_data = block["data"].T
+    new_blocks = []
+    for block in f.blocks:
+        current_pids = block.pids
+        current_data = block.data.T
         if len(current_data) == 0:
+            new_blocks.append(copy.deepcopy(block))
             continue
         # load all flavors
         flavor_data = [np.zeros_like(current_data[0]) for pid in br.flavor_basis_pids]
@@ -81,9 +84,15 @@ def project(blocks, reprs):
         for elem in reprs:
             proj = elem[:, np.newaxis] * elem
             new_data += proj @ flavor_data / (elem @ elem)
-        block["pids"] = br.flavor_basis_pids
-        block["data"] = np.array(new_data).T
-    return new_blocks
+        new_blocks.append(
+            LhapdfDataBlock(
+                xgrid=block.xgrid,
+                qgrid=block.qgrid,
+                pids=br.flavor_basis_pids,
+                data=np.array(new_data).T,
+            )
+        )
+    return LhapdfDataFile(header=copy.deepcopy(f.header), blocks=new_blocks)
 
 
 def is_evolution_labels(labels):