LbExec

LbExec.main(function, options, extra_args, export='')

Run a job with lbexec.

Parameters:

• function (callable) – A callable that will return the Gaudi configuration

• options (Options) – An initialised APP.Options object

• extra_args (list of str) – list of strings to add the the call to function

Returns:

The Gaudi process’s return code

Return type:

return_code (int)

CLI Utils

Utilities for parsing the positional arguments to lbexec.

This module provides two callable objects that can be used as types with argparse. The majority of the code is for providing hints to the user about what might be wrong in the case of errors.

FunctionLoader

Wrapper class which takes a function spec of the form module.name:callable. In the event of errors a best effort is made to advise the user of how to correct the error. In the event the module to import or function raises an exception tracebacks are rewritten to hide the implementation details of lbexec.

OptionsLoader

Converts a ‘+’ separated list of YAML file paths into an Application.Options object. The current application is discovered using the GAUDIAPPNAME environment variable. If required OVERRIDE_LBEXEC_APP can be passed to override which application is loaded. This is used by projects created by lb-dev where the value of GAUDIAPPNAME is ${PROJECT_NAME}Dev.

class LbExec.cli_utils.FunctionLoader(spec: str)

Bases: object

Class for parsing the function_spec argument to lbexec

property OptionsClass: type[LbExec.options.OptionsBase]

Return the Options class used by the function

LbExec.cli_utils.OptionsLoader(function: FunctionLoader, options_spec: str) → OptionsBase

Convert a ‘+’ separated list of paths to an Application.Options object.

LbExec.cli_utils.log_error(message)

LbExec.cli_utils.log_info(message)

LbExec.cli_utils.log_warn(message)

Options

class LbExec.options.CompressionAlgs(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

ROOT compression algorithms.

LZ4 = 'LZ4'

LZMA = 'LZMA'

ZLIB = 'ZLIB'

ZSTD = 'ZSTD'

class LbExec.options.CompressionSettings(*, algorithm: CompressionAlgs = CompressionAlgs.ZSTD, level: int = 4, optimise_baskets: bool = True)

Bases: BaseModel

Compression configuration settings.

algorithm: CompressionAlgs

level: int

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

optimise_baskets: bool

class LbExec.options.DataOptions(*, xml_file_catalog: Optional[Path] = None, output_file: str, compression: Optional[CompressionSettings] = None, xml_summary_file: Optional[str] = None, n_threads: int = 1, evt_max: int = -1, first_evt: int = 0, input_files: list[str])

Bases: OptionsBase

Base options class for job which have input files.

classmethod glob_input_files(input_files)

input_files: list[str]

List of input files to process.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True, 'use_enum_values': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod validate_input(values)

LbExec.options.Options

alias of DataOptions

class LbExec.options.OptionsBase(*, xml_file_catalog: Optional[Path] = None, output_file: str, compression: Optional[CompressionSettings] = None, xml_summary_file: Optional[str] = None, n_threads: int = 1, evt_max: int = -1, first_evt: int = 0)

Bases: BaseModel

compression: Optional[CompressionSettings]

Compression settings for the output file.

evt_max: int

Number of events to simulate.

first_evt: int

The first event to process.

get_output_file(stream: str) → str

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True, 'use_enum_values': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

n_threads: int

Number of threads to use for execution.

property output_file: str

output_file_: Annotated[str, Field(alias='output_file')]

Output file name, can contain {stream} to be replaced by the stream name.

xml_file_catalog: Optional[Path]

XML file catalog to use for mapping LFNs to PFNs.

xml_summary_file: Optional[str]

XML summary file to write job information to.

class LbExec.options.SimulationOptions(*, xml_file_catalog: Optional[Path] = None, output_file: str, compression: Optional[CompressionSettings] = None, xml_summary_file: Optional[str] = None, n_threads: int = 1, evt_max: int = -1, first_evt: int = 0, seeds: SimulationSeeds)

Bases: OptionsBase

Base options class for simulation jobs.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True, 'use_enum_values': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

seeds: SimulationSeeds

The seeds to use for the simulation.

classmethod validate_evt_max(evt_max)

class LbExec.options.SimulationSeeds(*, production_id: int, prod_job_id: int)

Bases: BaseModel

Seeds which simulation jobs can use to ensure reproducibility.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

prod_job_id: int

The sequential job number within the transformation.

production_id: int

The transformation ID in LHCbDIRAC.

Utils

Utility functions for writing lbexec payloads.

class LbExec.utils.FileCatalogEntry

Bases: TypedDict

Type for a file catalog entry.

guid: str | None

The GUID of the file, if available.

name: str

The logical file name (LFN) or physical file name (PFN).

pfn: str

The physical file name (PFN) associated with the LFN.

LbExec.utils.add_to_xml_file_catalog(xml_file_catalog_path: Path, entries: Iterable[FileCatalogEntry])

Add entries to the XML file catalog preserving original formatting.

We intentionally avoid xml.etree.ElementTree to keep:

• XML declaration (with standalone attr)

• Comments and DOCTYPE

• Element ordering & indentation expected by tests

The expected format (from tests) places <physical> before an empty <logical/>; no <lfn> element is written (the reader infers LFN from PFN).

LbExec.utils.extract_single_filetype_from_input_file(options)

LbExec.utils.get_output_filename(key, options, extra_opts, lumi_tree_key=None)

LbExec.utils.read_xml_file_catalog(xml_file_catalog)

Lookup the LFN->PFN mapping from the XML file catalog.

LbExec.utils.resolve_input_files(input_files, file_catalog)

Resolve LFNs to PFNs using what was returned from read_xml_file_catalog.

LbExec.utils.write_summary_xml(options: OptionsBase, output_files: Iterable[str], *, n_events: dict[str, int] | None = None)

Write a summary XML file with input and output files.

Workflows

Workflows package for LbExec.

This package contains various workflow functions for processing ROOT files, including skimming and merging operations, MDF file processing, and the process_trees decorator for automated tree processing.

Decorators

ROOT Operations

MDF Utils

MDF file utilities.

LbExec.workflows.mdf_utils.merge_mdf(options: DataOptions)

Merge MDF files into a single compressed output file.