LbProdRun#

LbProdRun.execute(job_spec, *, dry_run=False, gaudi_dry_run=False, interactive=False, prmon=False, perf=False)[source]#

LbProdRun.run_job(spec_file: Path, *, dry_run=False, gaudi_dry_run=False, verbose=False, interactive=False, prmon=False, perf=False)[source]#

Models#

class LbProdRun.models.BaseApplication(*, name: str, version: str, event_timeout: Optional[float] = None, number_of_processors: int = 1)[source]#

Bases: BaseModel

event_timeout: Optional[float]#

property is_lbconda#: Check whether an lb-conda environment is requested, return the name and version else None.

property is_lbexec: bool#: Determine if a given application/version combination should use lbexec

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'validate_assignment': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str#

number_of_processors: int#

property supports_input_run_number: bool#

The application is lbexec and supports the input_run_number

Was added in https://gitlab.cern.ch/lhcb/LHCb/-/merge_requests/4934

property uses_gaudi_mp: bool#

version: str#

class LbProdRun.models.BaseModel[source]#

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'validate_assignment': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class LbProdRun.models.JobSpecV1(*, spec_version: Literal[1], application: Union[ReleaseApplication, LbDevApplication, FullDevApplication], options: Union[LegacyOptions, LbExecOptions], input: Input = Input(files=None, xml_summary_file=None, xml_file_catalog=None, run_number=None, tck=None, n_of_events=-1, first_event_number=None, seeds=None), output: Output, db_tags: DBTags = DBTags(dddb_tag=None, conddb_tag=None, dq_tag=None))[source]#

Bases: BaseModel

class DBTags(*, dddb_tag: Optional[str] = None, conddb_tag: Optional[str] = None, dq_tag: Optional[str] = None)[source]#

Bases: BaseModel

conddb_tag: Optional[str]#

dddb_tag: Optional[str]#

dq_tag: Optional[str]#

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'validate_assignment': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class FullDevApplication(*, name: str, version: str, event_timeout: Optional[float] = None, number_of_processors: int = 1, run_script: Path)[source]#

Bases: BaseApplication

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'validate_assignment': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

run_script: Path#

class Input(*, files: Optional[list[str]] = None, xml_summary_file: Optional[str] = None, xml_file_catalog: Optional[str] = None, run_number: Optional[int] = None, tck: Optional[str] = None, n_of_events: int = -1, first_event_number: Optional[int] = None, seeds: Optional[SimulationSeeds] = None)[source]#

Bases: BaseModel

class SimulationSeeds(*, production_id: int, prod_job_id: int, max_n_events: Optional[int] = None)[source]#

Bases: BaseModel

Seeds which simulation jobs can use to ensure reproducibility.

max_n_events: Optional[int]#: For backwards compatibility, should only be used for Gauss.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'validate_assignment': True}#: 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.

files: Optional[list[str]]#

first_event_number: Optional[int]#

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'validate_assignment': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

n_of_events: int#

run_number: Optional[int]#

seeds: Optional[SimulationSeeds]#

tck: Optional[str]#

xml_file_catalog: Optional[str]#

xml_summary_file: Optional[str]#

class LbDevApplication(*, name: str, version: str, event_timeout: Optional[float] = None, number_of_processors: int = 1, project_base: Path, binary_tag: str)[source]#

Bases: BaseApplication

binary_tag: str#

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'validate_assignment': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

project_base: Path#

class LbExecOptions(*, entrypoint: str, extra_options: dict[str, Any], extra_args: list[str] = [])[source]#

Bases: BaseModel

entrypoint: str#

extra_args: list[str]#

extra_options: dict[str, Any]#

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'validate_assignment': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class LegacyOptions(*, command: list[str] = <factory>, files: list[str], format: ~typing.Optional[str] = None, gaudi_extra_options: ~typing.Optional[str] = None, processing_pass: ~typing.Optional[str] = None)[source]#

Bases: BaseModel

command: list[str]#

files: list[str]#

format: Optional[str]#

gaudi_extra_options: Optional[str]#

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'validate_assignment': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

processing_pass: Optional[str]#

classmethod set_command(command)[source]#

class Output(*, prefix: str, types: list[str], histogram_file: Optional[str] = None, compression: Optional[str] = None)[source]#

Bases: BaseModel

compression: Optional[str]#

histogram_file: Optional[str]#

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'validate_assignment': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

prefix: str#

types: list[str]#

class ReleaseApplication(*, name: str, version: str, event_timeout: Optional[float] = None, number_of_processors: int = 1, data_pkgs: list[str] = [], binary_tag: str = 'best', nightly: Optional[str] = None)[source]#

Bases: BaseApplication

binary_tag: str#

data_pkgs: list[str]#

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'validate_assignment': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

nightly: Optional[str]#

application: Union[ReleaseApplication, LbDevApplication, FullDevApplication]#

db_tags: DBTags#

input: Input#

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'validate_assignment': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

options: Union[LegacyOptions, LbExecOptions]#

output: Output#

spec_version: Literal[1]#

LbProdRun.models.read_jobspec(spec_file: Path)[source]#