Source code for validphys.fkparser

This module implements parsers for FKtable  and CFactor files into useful
datastructures, contained in the :py:mod:`validphys.coredata` module, which can
be easily pickled and interfaced with common Python libraries.

Most users will be interested in using the high level interface
:py:func:`load_fktable`.  Given a :py:class:`validphys.core.FKTableSpec`
object, it returns an instance of :py:class:`validphys.coredata.FKTableData`,
an object with the required information to compute a convolution, with the
CFactors applied.

.. code-block:: python

    from validphys.fkparser import load_fktable
    from validphys.loader import Loader
    l = Loader()
    fk = l.check_fktable(setname="ATLASTTBARTOT", theoryID=53, cfac=('QCD',))
    res = load_fktable(fk)
import dataclasses
import functools
import io
import tarfile

import numpy as np
import pandas as pd

from validphys.coredata import CFactorData, FKTableData
from validphys.pineparser import pineappl_reader

[docs]class BadCFactorError(Exception): """Exception raised when an CFactor cannot be parsed correctly"""
[docs]class BadFKTableError(Exception): """Exception raised when an FKTable cannot be parsed correctly"""
[docs]@dataclasses.dataclass(frozen=True) class GridInfo: """Class containing the basic properties of an FKTable grid.""" setname: str hadronic: bool ndata: int nx: int
[docs]@functools.lru_cache() def load_fktable(spec): """Load the data corresponding to a FKSpec object. The cfactors will be applied to the grid. If we have a new-type fktable, call directly `load()`, otherwise fallback to the old parser """ if spec.legacy: with open_fkpath(spec.fkpath) as handle: tabledata = parse_fktable(handle) else: tabledata = pineappl_reader(spec) # In the new theories, the cfactor get applied as the fktables are loaded if not spec.cfactors or not spec.legacy: return tabledata cfprod = 1.0 for cf in spec.cfactors: with open(cf, "rb") as f: cfdata = parse_cfactor(f) cfprod *= cfdata.central_value return tabledata.with_cfactor(cfprod)
def _get_compressed_buffer(path): archive = members = archive.getmembers() l = len(members) if l != 1: raise BadFKTableError(f"Archive {path} should contain one file, but it contains {l}.") return archive.extractfile(members[0])
[docs]def open_fkpath(path): """Return a file-like object from the fktable path, regardless of whether it is compressed Parameters .......... path: Path or str Path like file containing a valid FKTable. It can be either inside a tarball or in plain text. Returns ------- f: file A file like object for further processing. """ if tarfile.is_tarfile(path): return _get_compressed_buffer(path) return open(path, 'rb')
def _is_header_line(line): return line.startswith((b'_', b'{')) def _bytes_to_bool(x): return bool(int(x)) def _parse_fk_options(line_and_stream, value_parsers=None): """Parse a sequence of lines of the form *OPTION: VALUE into a dictionary. """ res = {} if value_parsers is None: value_parsers = {} for lineno, next_line in line_and_stream: if _is_header_line(next_line): return res, lineno, next_line if not next_line.startswith(b'*'): raise BadFKTableError(f"Error on line {lineno}: Expecting an option starting with '*'") try: keybytes, valuebytes = next_line.split(b':', maxsplit=1) except ValueError: raise BadFKTableError(f"Error on line {lineno}: Expecting an option containing ':'") key = keybytes[1:].strip().decode() if key in value_parsers: try: value = value_parsers[key](valuebytes) except Exception as e: raise BadFKTableError(f"Could not parse key {key} on line {lineno}") from e else: value = valuebytes.strip().decode() res[key] = value raise BadFKTableError("FKTable should end with FastKernel spec, not with a set of options") def _segment_parser(f): @functools.wraps(f) def f_(line_and_stream): buf = io.BytesIO() for lineno, next_line in line_and_stream: if _is_header_line(next_line): processed = f(buf) return processed, lineno, next_line buf.write(next_line) raise BadFKTableError("FKTable should end with FastKernel spec, not with a segment string") return f_ @_segment_parser def _parse_string(buf): return buf.getvalue().decode() @_segment_parser def _parse_flavour_map(buf): return np.loadtxt(buf, dtype=bool) @_segment_parser def _parse_xgrid(buf): return np.fromstring(buf.getvalue(), sep='\n') # This used a different interface from segment parser because we want it to # be fast. # We assume it is going to be the last section. def _parse_hadronic_fast_kernel(f): """Parse the FastKernel secrion of an hadronic FKTable into a DataFrame. ``f`` should be a stream containing only the section""" # Note that we need the slower whitespace here because it turns out # that there are fktables where space and tab are used as separators # within the same table. df = pd.read_csv(f, sep=r'\s+', header=None, index_col=(0, 1, 2)) df.columns = list(range(14 * 14)) df.index.names = ['data', 'x1', 'x2'] return df def _parse_dis_fast_kernel(f): """Parse the FastKernel section of a DIS FKTable into a DataFrame. ``f`` should be a stream containing only the section""" df = pd.read_csv(f, sep=r'\s+', header=None, index_col=(0, 1)) df.columns = list(range(14)) df.index.names = ['data', 'x'] return df def _parse_gridinfo(line_and_stream): dict_result, line_number, next_line = _parse_fk_options( line_and_stream, value_parsers={"HADRONIC": _bytes_to_bool, "NDATA": int, "NX": int} ) gi = GridInfo(**{k.lower(): v for k, v in dict_result.items()}) return gi, line_number, next_line def _parse_header(lineno, header): if not _is_header_line(header): raise BadFKTableError( f"Bad header at line {lineno}: First character should be either '_' or '{{'" ) try: endname = header.index(b'_', 1) except ValueError: raise BadFKTableError(f"Bad header at line {lineno}: Expected '_' after name") from None header_name = header[1:endname] # Note: This is not the same as header[0]. Bytes iterate as ints. return header[0:1], header_name.decode() def _build_sigma(f, res): gi = res["GridInfo"] fm = res["FlavourMap"] table = _parse_hadronic_fast_kernel(f) if gi.hadronic else _parse_dis_fast_kernel(f) # Filter out empty flavour indices table = table.loc[:, fm.ravel()] return table _KNOWN_SEGMENTS = { "GridDesc": _parse_string, "VersionInfo": _parse_fk_options, "GridInfo": _parse_gridinfo, "FlavourMap": _parse_flavour_map, "xGrid": _parse_xgrid, "TheoryInfo": functools.partial( _parse_fk_options, value_parsers={ "ID": int, "PTO": int, "DAMP": _bytes_to_bool, "IC": _bytes_to_bool, "XIR": float, "XIF": float, "NfFF": int, "MaxNfAs": int, "MaxNfPdf": int, "Q0": float, "alphas": float, "Qref": float, "QED": _bytes_to_bool, "alphaqed": float, "Qedref": float, "SxRes": _bytes_to_bool, "mc": float, "Qmc": float, "kcThr": float, "mb": float, "Qmb": float, "kbThr": float, "mt": float, "Qmt": float, "ktThr": float, "MZ": float, "MW": float, "GF": float, "SIN2TW": float, "TMC": _bytes_to_bool, "MP": float, "global_nx": int, "EScaleVar": _bytes_to_bool, }, ), } def _check_required_sections(res, lineno): """Check that we have found all the required sections by the time we reach 'FastKernel'""" for section in _KNOWN_SEGMENTS: if section not in res: raise BadFKTableError(f"{section} must come before 'FastKernel' section at {lineno}")
[docs]def parse_fktable(f): """Parse an open byte stream into an FKTableData. Raise a BadFKTableError if problems are encountered. Parameters ---------- f : file Open file-like object. See :func:`open_fkpath`to obtain it. Returns ------- fktable : FKTableData An object containing the FKTable data and information. Notes ----- This function operates at the level of a single file, and therefore it does not apply CFactors (see :py:func:`load_fktable` for that) or handle operations within COMPOUND ensembles. """ line_and_stream = enumerate(f, start=1) res = {} lineno, header = next(line_and_stream) while True: marker, header_name = _parse_header(lineno, header) if header_name == "FastKernel": _check_required_sections(res, lineno) Q0 = res['TheoryInfo']['Q0'] sigma = _build_sigma(f, res) hadronic = res['GridInfo'].hadronic ndata = res['GridInfo'].ndata xgrid = res.pop('xGrid') return FKTableData( sigma=sigma, ndata=ndata, Q0=Q0, metadata=res, hadronic=hadronic, xgrid=xgrid, ) elif header_name in _KNOWN_SEGMENTS: parser = _KNOWN_SEGMENTS[header_name] elif marker == b'{': parser = _parse_string elif marker == b'_': parser = _parse_fk_options else: raise RuntimeError("Should not be here") try: out, lineno, header = parser(line_and_stream) except Exception as e: # Note that the old lineno is the one we want raise BadFKTableError(f"Failed processing header {header_name} on line {lineno}") from e res[header_name] = out
[docs]def parse_cfactor(f): """Parse an open byte stream into a :py:class`CFactorData`. Raise a BadCFactorError if problems are encountered. Parameters ---------- f : file Binary file-like object Returns ------- cfac : CFactorData An object containing the data on the cfactor for each point. """ stars = f.readline() if not stars.startswith(b'*'): raise BadCFactorError("First line should start with '*'.") descbytes = io.BytesIO() for line in f: if line.startswith(b'*'): break descbytes.write(line) description = descbytes.getvalue().decode() try: data = np.loadtxt(f) except Exception as e: raise BadCFactorError(e) from e data = data.reshape(-1, 2) central_value = data[:, 0] uncertainty = data[:, 1] return CFactorData( description=description, central_value=central_value, uncertainty=uncertainty )