gbasf2

`gbasf2`#

The b2luigi.contrib.gbasf2 package provides a batch process for gbasf2, the Belle II submission tool for the LHC Computing Grid. It takes the basf2 path of a Basf2PathTask, submits it as a gbasf2 project, monitors the project, reschedules failed jobs and downloads the output dataset once all jobs have succeeded.

To use it, install the gbasf2 extra and set batch_system = "gbasf2" on the tasks that should run on the grid:

pip3 install "b2luigi[gbasf2]"

gbasf2 is never selected by batch_system = "auto"; it always has to be set explicitly. Parameter grouping (see Parameter Grouping) is not supported for gbasf2 tasks. All gbasf2-specific settings are described in Gbasf2Process below, and a complete example is shown in gbasf2 Analysis Task.

API Reference#

class b2luigi.contrib.gbasf2.process.Gbasf2Process(*args, **kwargs)[source]#

Bases: BatchProcess

Batch process for working with gbasf2 projects on the LHC Computing Grid (LCG).

Features

``gbasf2`` project submission

The gbasf2 batch process takes the basf2 path returned by the create_path method of the task, saves it into a pickle file to the disk and creates a wrapper steering file that executes the saved path. Any basf2 variable aliases added in the Path() or create_path() method are also stored in the pickle file. It then sends both the pickle file and the steering file wrapper to the grid via the Belle II-specific DIRAC-wrapper gbasf2. However, b2luigi supports the submission of custom steering files with the setting gbasf2_custom_steering_file. This conserves the way that the basf2 path os still contained in the create_path() method. In this instance, b2luigi``checks automatically if the corresponding file exists and copies it into the active directory. The ``gbasf2 task is then set to submit the unpickled file to the grid job which allows the utilization of python-based basf2 modules.
Project status monitoring

After the project submission, the gbasf batch process regularly checks the status of all the jobs belonging to a gbasf2 project returns a success if all jobs had been successful, while a single failed job results in a failed project. You can close a running b2luigi process and then start your script again and if a task with the same project name is running, this b2luigi gbasf2 wrapper will recognize that and instead of resubmitting a new project, continue monitoring the running project.

Hint

The outputs of gbasf2 tasks can be a bit overwhelming, so I recommend using the central scheduler which provides a nice overview of all tasks in the browser, including a status/progress indicator how many jobs in a gbasf2 project are already done.
Automatic download of datasets and logs

If all jobs had been successful, it automatically downloads the output dataset and the log files from the job sandboxes and automatically checks if the download was successful before moving the data to the final location. On failure, it only downloads the logs. The dataset download can be optionally disabled.
Automatic rescheduling of failed jobs

Whenever a job fails, gbasf2 reschedules it as long as the number of retries is below the value of the setting gbasf2_max_retries. It keeps track of the number of retries in a local file in the log_file_dir, so that it does not change if you close b2luigi and start it again. Of course it does not persist if you remove that file or move to a different machine.

Note

Despite all the automatization that this gbasf2 wrapper provides, the user is expected to have a basic understanding of how the grid works and know the basics of working with gbasf2 commands manually.

Caveats

The gbasf2 batch process for luigi can only be used for tasks inheriting from Basf2PathTask or other tasks with a create_path() method that returns a basf2 path.
It can be used only for picklable ``basf2`` paths, with only some limited global basf2 state saved (currently aliases and global tags). The batch process stores the path created by create_path in a python pickle file and runs that on the grid. Therefore, python ``basf2`` modules are not yet supported. To see if the path produced by a steering file is picklable, you can try to dump it with basf2 --dump-path and execute it again with basf2 --execute-path. In case the steering file contains content (e.g. modules) that cannot be pickled, the feature setting gbasf2_custom_steering_file can be utilized which has to be set to the path of the steering file the user wishes to be used. This submits the custom steering file to the grid job. The specific use case for this is the usage and interaction with python-based basf2 modules that are not pickable.
Output format: Changing the batch to gbasf2 means you also have to adapt how you handle the output of your gbasf2 task in tasks depending on it, because the output will not be a single root file anymore (e.g. B_ntuple.root), but a collection of root files, one for each file in the input data set, in a directory with the base name of the root files, e.g.:
```
<task output directory>
            ├── B_ntuple.root
            │   └── B_ntuple_0.root
            │   └── B_ntuple_1.root
            │   └── ...
            ├── D_ntuple.root
            │   └── D_ntuple_0.root
            │   └── ...
```

Settings for gbasf2 tasks:

To submit a task with the gbasf2 wrapper, you first you have to add the property batch_system = "gbasf2", which sets the batch_system setting. It is not recommended to set that setting globally, as not all tasks can be submitted to the grid, but only tasks with a create_path method.

For gbasf2 tasks it is further required to set the settings

gbasf2_input_dataset: String with the logical path of a dataset on the grid to use as an input to the task. You can provide multiple inputs by having multiple paths contained in this string, separated by commas without spaces. An alternative is to just instantiate multiple tasks with different input datasets, if you want to know in retrospect which input dataset had been used for the production of a specific output.
gbasf2_input_dslist: Alternatively to gbasf2_input_dataset, you can use this setting to provide a text file containing the logical grid path names, one per line.
gbasf2_project_name_prefix: A string with which your gbasf2 project names will start. To ensure the project associate with each unique task (i.e. for each of luigi parameters) is unique, the unique task.task_id is hashed and appended to the prefix to create the actual gbasf2 project name. Should be below 22 characters so that the project name with the hash can remain under 32 characters.

The following example shows a minimal class with all required options to run on the gbasf2/grid batch:

class MyTask(Basf2PathTask):
    batch_system = "gbasf2"
    gbasf2_project_name_prefix = b2luigi.Parameter(significant=False)
    gbasf2_input_dataset = b2luigi.Parameter(hashed=True)

Other not required, but noteworthy settings are:

gbasf2_setup_path: Path to gbasf2 environment setup script that needs so be sourced to run gbasf2 commands. Defaults to /cvmfs/belle.kek.jp/grid/gbasf2/pro/bashrc.
gbasf2_release: Defaults to the release of your currently set up basf2 release. Set this if you want the jobs to use another release on the grid.
gbasf2_proxy_lifetime: Defaults to 24. When initializing a proxy, set the lifetime to this number of hours.
gbasf2_min_proxy_lifetime: Defaults to 0. During processing, prompt user to reinitialize proxy if remaining proxy lifetime drops below this number of hours.
gbasf2_print_status_updates: Defaults to True. By setting it to False you can turn off the printing of of the job summaries, that is the number of jobs in different states in a gbasf2 project.
gbasf2_max_retries: Default to 0. Maximum number of times that each job in the project can be automatically rescheduled until the project is declared as failed.
gbasf2_proxy_group: Default to "belle". If provided, the gbasf2 wrapper will work with the custom gbasf2 group, specified in this parameter. No need to specify this parameter in case of usual physics analysis at Belle II. If specified, one has to provide gbasf2_project_lpn_path parameter.
gbasf2_project_lpn_path: Path to the LPN folder for a specified gbasf2 group. The parameter has no effect unless the gbasf2_proxy_group is used with non-default value.
gbasf2_jinja_template_path: This parameter sets a custom basf2 steering template where the user can adapt the default template (e.g. for altering the pdg database, …). Note that this is an expert option that should be treated with care.
gbasf2_additional_download_params: Defaults to "--new". This parameter sets additional parameters that are given to gb2_ds_get. Note that in case you override the parameter, the --new parameter is not automatically set, so you might have to manually add --new if you want this parameter to be used.
gbasf2_download_dataset: Defaults to True. Disable this setting if you don’t want to download the output dataset from the grid on job success. As you can’t use the downloaded dataset as an output target for luigi, you should then use the provided Gbasf2GridProjectTarget, as shown in the following example:
```
from b2luigi.contrib.gbasf2.process import get_unique_project_name, Gbasf2GridProjectTarget

class MyTask(Basf2PathTask):
    # [...]
    def output(self):
        project_name = get_unique_project_name(self)
        return Gbasf2GridProjectTarget(project_name)
```
This is useful when chaining gbasf2 tasks together, as they don’t need the output locally but take the grid datasets as input. Also useful when you just want to produce data on the grid for other people to use.

Caution

For Gbasf2GridProjectTarget``s the global and the task ``gbasf2_proxy_group cannot differ.

Tip

Set only one global gbasf2_proxy_group setting.
gbasf2_download_logs: Whether to automatically download the log output of gbasf2 projects when the task succeeds or fails. Having the logs is important for reproducibility.
gbasf2_custom_steering_file: Optional path to submit a custom steering file to gbasf2. This does not pickle the basf2.Path instance and allows the utilization of python-based basf2 modules. Named modules have to be contained either in the steering file itself or by additional files via the input sandbox.

The following optional settings correspond to the equally named gbasf command line options (without the gbasf_ prefix) that you can set to customize your gbasf2 project:

gbasf2_noscout, gbasf2_additional_files, gbasf2_input_datafiles, gbasf2_n_repition_job, gbasf2_force_submission, gbasf2_cputime, gbasf2_evtpersec, gbasf2_priority, gbasf2_jobtype, gbasf2_basf2opt, gbasf2_lfn_sandboxfiles, gbasf2_input_grouping

It is further possible to append arbitrary command line arguments to the gbasf2 submission command with the gbasf2_additional_params setting. If you want to blacklist a grid site, you can e.g. add

b2luigi.set_setting("gbasf2_additional_params",  "--banned_site LCG.KEK.jp")

Example

Here is an example file to submit an analysis path created by the script in examples/gbasf2/example_mdst_analysis to grid via gbasf2:

File: examples/gbasf2/gbasf2_example.py#

import b2luigi
from b2luigi.contrib.basf2.tasks import Basf2PathTask

import example_mdst_analysis


class AnalysisTask(Basf2PathTask):
    # set the batch_system property to use the gbasf2 wrapper batch process for this task
    batch_system = "gbasf2"
    # Must define a prefix for the gbasf2 project name to submit to the grid.
    # b2luigi will then add a hash derived from the luigi parameters to create a unique project name.
    gbasf2_project_name_prefix = b2luigi.Parameter()
    gbasf2_input_dataset = b2luigi.Parameter(hashed=True)
    # Example luigi cut parameter to facilitate starting multiple projects for different cut values
    mbc_lower_cut = b2luigi.IntParameter()

    def create_path(self):
        mbc_range = (self.mbc_lower_cut, 5.3)
        return example_mdst_analysis.create_analysis_path(
            d_ntuple_filename="D_ntuple.root", b_ntuple_filename="B_ntuple.root", mbc_range=mbc_range
        )

    def output(self):
        yield self.add_to_output("D_ntuple.root")
        yield self.add_to_output("B_ntuple.root")


class AnalysisWrapperTask(b2luigi.WrapperTask):
    """
    We use the AnalysisWrapperTask to be able to require multiple analyse tasks with
    different input datasets and cut values. For each parameter combination, a
    different gbasf2 project will be submitted.
    """

    def requires(self):
        input_dataset = (
            "/belle/MC/release-04-01-04/DB00000774/SkimM13ax1/prod00011778/e1003/4S/r00000/mixed/11180100/udst/sub00/"
            "udst_000006_prod00011778_task10020000006.root"
        )
        # if you want to iterate over different cuts, just add more values to this list
        mbc_lower_cuts = [5.15, 5.2]
        for mbc_lower_cut in mbc_lower_cuts:
            yield AnalysisTask(
                mbc_lower_cut=mbc_lower_cut,
                gbasf2_project_name_prefix="luigiExample",
                gbasf2_input_dataset=input_dataset,
                max_event=100,
            )


if __name__ == "__main__":
    main_task_instance = AnalysisWrapperTask()
    n_gbasf2_tasks = len(list(main_task_instance.requires()))
    b2luigi.process(main_task_instance, workers=n_gbasf2_tasks)

Handling failed jobs

The gbasf2 input wrapper considers the gbasf2 project as failed if any of the jobs in the project failed and reached the maximum number of retries. It then automatically downloads the logs, so please look into them to see what the reason was. For example, it can be that only certain grid sites were affected, so you might want to exclude them by adding the "--banned_site ... to gbasf2_additional_params.

You also always reschedule jobs manually with the gb2_job_reschedule command or delete them with gb2_job_delete so that the gbasf2 batch process doesn’t know they ever existed. Then run just run your luigi task/script again and it will start monitoring the running project again.

pickle_file_path#: file name in which the pickled basf2 path from self.task.create_path() will be stored

wrapper_file_path#: file name for steering file that executes pickled path, which will be send to the grid

max_retries#: Maximum number of times that each job in the project can be rescheduled until the project is declared as failed.

gbasf2_custom_steering_file#: Setting to incorporate custom steering files into b2luigi.

n_retries_by_job#: Store number of times each job had been rescheduled

get_job_status()[source]#

Get the overall status of the gbasf2 project.

This method determines the current status of all jobs in a gbasf2 project (see get_gbasf2_project_job_status_dict). and returns an overall project status. It uses cached job status data if available and updates it if necessary. The method also handles job rescheduling in case of failures and updates task progress and status messages for the central scheduler.

Returns:

The overall status of the gbasf2 project. Possible values are:

JobStatus.running: If the project is still in progress.
JobStatus.successful: If all jobs in the project are completed successfully.
JobStatus.aborted: If any job in the project fails and cannot be rescheduled.

Return type:

JobStatus

Raises:

RuntimeError – If the job status cannot be determined.

Notes

The job status is cached for 5 minutes to optimize performance.
Failed jobs are rescheduled up to a maximum number of retries before the project is marked as aborted.
The project is considered successful only if all jobs are completed successfully.
Progress and status updates are logged if gbasf2_print_status_updates is enabled.

_on_first_success_action()[source]#

Executes actions to be performed after all jobs in the project have successfully completed.

This method checks specific settings to determine whether to download logs and datasets associated with the project. If the respective settings are enabled, it triggers the corresponding download operations.

Note

Downloads logs if the “gbasf2_download_logs” setting is enabled.
Downloads the dataset if the “gbasf2_download_dataset” setting is enabled.

_on_failure_action()[source]#

Handles actions to be performed after a project failure.

This method retrieves the job status dictionary for the specified gbasf2 project and identifies failed jobs. A job is considered failed if its status is “Failed” or if its status is “Done” but its application status is not “Done”. The number of failed jobs and their details are printed. If the setting gbasf2_download_logs is enabled, it triggers the download of logs for further investigation.

_reschedule_failed_jobs()[source]#

Attempts to reschedule failed jobs in the project if the maximum number of retries (gbasf2_max_retries) has not been reached.

This method evaluates the status of jobs in the project and determines whether they should be rescheduled or if they have hit the maximum retry limit. Jobs that are eligible for rescheduling are added to a rescheduling queue, while jobs that have reached the retry limit are logged with a warning.

Returns:

True if all failed jobs were successfully rescheduled or no jobs needed rescheduling.
False if any jobs reached the maximum retry limit.

Return type:

bool

Raises:

RuntimeWarning – If any jobs have reached the maximum retry limit.

Notes

A job is considered failed if its status is “Failed” or if its status is “Done” but its application status is not “Done”.
The method relies on get_gbasf2_project_job_status_dict to retrieve the current status of jobs and _reschedule_jobs to perform the rescheduling.

_reschedule_jobs(job_ids)[source]#

Reschedule a list of jobs by their IDs.

This method takes a list of job IDs, logs the rescheduling process, and executes a command to reschedule the specified jobs using the gb2_job_reschedule utility. It also includes the number of retries for each job in the log output.

Parameters:: job_ids (list of str) – A list of job IDs to be rescheduled.
Raises:: KeyError – If a job ID in the list is not found in self.n_retries_by_job.

Notes

The gb2_job_reschedule command is executed with the –force flag.
The run_with_gbasf2 function is used to execute the rescheduling command.

start_job()[source]#

Submit a new gbasf2 project to the grid.

This method checks if a project with the specified name already exists on the grid. If it does, the submission is skipped, and a message is printed. Otherwise, it prepares the necessary steering files and submits the project.

Steps:

Check if the project already exists on the grid. (see check_project_exists)
Prepare the steering file:
- If a custom steering file is provided, copy it. (see _copy_custom_steering_script)
- Otherwise, create a wrapper steering file. (see: _create_wrapper_steering_file)
Build the gbasf2 submission command. (see _build_gbasf2_submit_command)
Create a symlink for the pickle file to ensure it is included in the grid input sandbox.
Submit the project using the gbasf2 command. (see run_with_gbasf2)
Clean up the symlink after submission.

Raises:: OSError – If there is an issue creating or removing the symlink.

Note

If the project already exists, the user is advised to change the project name to submit a new project if needed.

terminate_job()[source]#

Terminate a gbasf2 project if it exists.

This method checks if the specified gbasf2 project exists. If it does, it terminates the project using the gb2_job_kill command with the --force option. Terminated jobs are not removed from the job database and can be restarted if needed.

Note

The gb2_job_delete command differs from gb2_job_kill in that deleted jobs are permanently removed from the job database, while terminated jobs remain in the database.

_build_gbasf2_submit_command()[source]#: Calls build_gbasf2_submit_command to create the command to submit the gbasf2 project.

_write_path_to_file()[source]#

Serializes and saves a basf2.Path object and variable aliases to a pickle file.

This method attempts to create a basf2.Path object using the b2luigi.contrib.basf2.tasks.Basf2PathTask.create_path() method of the associated task. The resulting path, along with any variable aliases from the current variable manager instance, is serialized and written to a specified pickle file.

Raises:: Exception – If the associated task does not have a create_path() method, indicating it is not an instance of :class`Basf2PathTask` or a compatible class.

Dependencies:

write_path_and_state_to_file: Used to perform the serialization and file writing.

_create_wrapper_steering_file()[source]#

Generates a steering file for grid execution by processing a Jinja2 template.

This method creates a steering file that executes the pickled basf2 path generated by b2luigi.contrib.basf2.tasks.Basf2PathTask.create_path(). It reads a Jinja2 template, replaces placeholders with appropriate values, and writes the processed template to a new file for grid submission.

The following steps are performed: 1. Reads the Jinja2 template file specified by the gbasf2_jinja_template_path setting. 2. Replaces template variables such as the pickle file path and maximum event count. 3. Writes the processed template to self.wrapper_file_path.

_copy_custom_steering_script()[source]#

Copies a custom gbasf2 steering script to the specified wrapper file path.

This method checks if the custom gbasf2 steering file exists at the specified location. If it exists, the file is copied to the wrapper file path. If the file does not exist, a ValueError is raised.

Raises:: ValueError – If the custom gbasf2 steering file does not exist at the specified path.

_get_gbasf2_dataset_query(output_file_name: str) → str[source]#

Constructs a query string to retrieve a subset of files from a grid project associated with the current task, based on the provided output file name.

This method generates a wildcard pattern for use with grid commands such as gb2_ds_list or gb2_ds_get to locate files matching the specified output file name within the project’s directory structure.

Parameters:: output_file_name (str) – The name of the output file, which must be a basename (not a path) and should end with the “.root” extension.
Returns:: A query string containing the wildcard pattern to locate the desired files in the grid project.
Return type:: str
Raises:: ValueError – If the provided output_file_name is not a basename, does not end with “.root”, or if required settings for non-default proxy groups are missing.

static _get_project_download_path(base_output_dir: str)[source]#

Get the location of the .root files of the downloaded project :param base_output_dir: base directory for the gbasf2 download

Returns:: str

_local_gb2_dataset_is_complete(output_file_name: str, check_temp_dir: bool = False) → bool[source]#

Helper method that returns True if the download of the gbasf2 dataset for the output output_file_name is complete.

Parameters:

output_file_name – Output file name, must be a root file, e.g. ntuple.root. Usually defined by the user via b2luigi.Task.add_to_output() in the b2luigi.Task.output() method.
check_temp_dir – Instead of checking the final output path, check whether the download into the temporary (“partial”) directory is complete. This function is usually called with this argument set to True, to check whether the dataset can be moved to its final output path.

_download_dataset()[source]#

Downloads the task outputs from the gbasf2 project dataset.

This method ensures that all files matching the naming pattern name_*.root from the grid dataset corresponding to the project name are downloaded into a specified directory. The download process is handled in a way that prevents marking the task as complete if the download fails.

Steps: 1. Checks if the dataset exists on the grid for the specified project name. 2. For each task output, constructs a query string to locate the dataset files. 3. Skips downloading if the dataset already exists locally and is complete. 4. Downloads the dataset into a temporary directory to ensure atomicity. 5. Handles failed downloads by retrying only the missing files. 6. Verifies the completeness of the downloaded dataset. 7. Moves the successfully downloaded dataset to the final output directory.

Raises:

RuntimeError – If the dataset does not exist on the grid.
RuntimeError – If no output data is found for the specified project.
RuntimeError – If the downloaded dataset is incomplete.

Notes

Temporary directories are used to avoid marking the task as complete prematurely.
Failed downloads are tracked and retried using a monitoring file.
Additional parameters for the gb2_ds_get command can be configured via task settings.

_download_logs()[source]#

Download sandbox files from grid with logs for each job in the gbasf2 project.

It wraps gb2_job_output, which downloads the job sandbox, which has the following structure:

log
└── <project name>
    ├── <first job id>
    │   ├── job.info
    │   ├── Script1_basf2helper.py.log # basf2 outputs
    │   └── std.out
    ├── <second job id>
    │   ├── ...
    ...

These are stored in the task log dir.

class b2luigi.contrib.gbasf2.process.Gbasf2GridProjectTarget(project_name, dirac_user=None, gbasf2_setup_path='/cvmfs/belle.kek.jp/grid/gbasf2/pro/bashrc')[source]#

Bases: Target

Gbasf2GridProjectTarget is a custom Luigi Target that checks the existence and status of a dataset produced by a gbasf2 grid project. It ensures that the dataset exists on the grid and that all jobs associated with the project have completed successfully.

project_name#

Name of the gbasf2 grid project that produced the dataset and under which the dataset is stored.

Type:: str

dirac_user#

Dirac user who produced the output dataset. If None, the current user is used.

Type:: str, optional

gbasf2_setup_path#

Path to the gbasf2 setup file. Defaults to “/cvmfs/belle.kek.jp/grid/gbasf2/pro/bashrc”.

Type:: str

exists()[source]#

Checks the existence and status of a dataset or project on the grid.

This method performs the following checks:

Verifies if a dataset associated with the given project name exists on the grid. (see check_dataset_exists_on_grid)
If a dataset exists, checks whether a project with the same name exists on the grid. (see check_project_exists)
Ensures that no jobs are actively writing to the project by verifying that all jobs associated with the project have a status of “Done” and an application status of “Done”.

Returns:: True if the dataset exists, the project exists, and no jobs are actively writing to it. False otherwise.
Return type:: bool

b2luigi.contrib.gbasf2.process.build_gbasf2_submit_command(task, wrapper_file_path=None, pickle_file_path=None, gbasf2_project_name=None, gbasf2_custom_steering_file=None)[source]#

Constructs the gbasf2 submit command string based on task options and attributes.

This method generates a command string to submit a gbasf2 job, incorporating various settings and parameters. It validates inputs, handles optional parameters, and ensures the command is properly formatted for execution.

Returns:

A list of command-line arguments for the gbasf2 submission command.

Return type:

list

Raises:

ValueError – If gbasf2_additional_files is not an iterable or is a string.
RuntimeError – If both gbasf2_input_dataset and gbasf2_input_dslist are set, or if neither is set.
FileNotFoundError – If the file specified in gbasf2_input_dslist does not exist.
ValueError – If priority is not an integer between 0 and 10.
ValueError – If gbasf2_proxy_group is non-default and gbasf2_project_lpn_path is not provided.

Notes

The method uses various task settings to construct the command, such as:
gbasf2_release, gbasf2_additional_files, gbasf2_input_dataset, gbasf2_input_dslist, gbasf2_n_repition_job, gbasf2_input_datafiles, gbasf2_force_submission, gbasf2_cputime, gbasf2_evtpersec, gbasf2_priority, gbasf2_jobtype, gbasf2_basf2opt, and gbasf2_additional_params.
If the proxy group is not "belle", an output dataset path must be specified.

b2luigi.contrib.gbasf2.process.check_dataset_exists_on_grid(gbasf2_project_name, dirac_user=None, gbasf2_setup_path='/cvmfs/belle.kek.jp/grid/gbasf2/pro/bashrc', task=None)[source]#

Check if an output dataset exists for the specified gbasf2 project on the grid.

Parameters:

gbasf2_project_name (str) – The name of the gbasf2 project to check.
dirac_user (str, optional) – The DIRAC user to use for querying. Defaults to None.
gbasf2_setup_path (str, optional) – The path to the gbasf2 setup script. Defaults to “/cvmfs/belle.kek.jp/grid/gbasf2/pro/bashrc”.
task (object, optional) – An optional task object for retrieving settings. Defaults to None.

Returns:

True if the dataset exists on the grid, False otherwise.

Return type:

bool

Raises:

ValueError – If gbasf2_proxy_group is set to a non-default value (“belle”) and gbasf2_project_lpn_path is not provided in the settings.

b2luigi.contrib.gbasf2.process.get_gbasf2_project_job_status_dict(gbasf2_project_name, dirac_user=None, gbasf2_setup_path='/cvmfs/belle.kek.jp/grid/gbasf2/pro/bashrc', task=None)[source]#

Returns a dictionary for all jobs in the project with a structure like the following, which I have taken and adapted from an example output:

{
    "<JobID>": {
        "SubmissionTime": "2020-03-27 13:08:49",
        "Owner": "<dirac username>",
        "JobGroup": "<ProjectName>",
        "ApplicationStatus": "Done",
        "HeartBeatTime": "2020-03-27 16:01:39",
        "Site": "LCG.KEK.jp",
        "MinorStatus": "Execution Complete",
        "LastUpdateTime": "2020-03-27 16:01:40",
        "Status": "<Job Status>"
    }
...
}

For that purpose, the script in gbasf2_job_status.py is called. That script directly interfaces with Dirac via its API, but it only works with the gbasf2 environment and python2, which is why it is called as a subprocess. The job status dictionary is passed to this function via json.

b2luigi.contrib.gbasf2.process.check_project_exists(gbasf2_project_name, dirac_user=None, gbasf2_setup_path='/cvmfs/belle.kek.jp/grid/gbasf2/pro/bashrc', task=None)[source]#

Check if a gbasf2 project exists on the grid.

This function verifies the existence of a gbasf2 project by attempting to retrieve its job status using the get_gbasf2_project_job_status_dict function. If the retrieval fails due to a RuntimeError, the function returns False.

Parameters:

gbasf2_project_name (str) – The name of the gbasf2 project to check.
dirac_user (str, optional) – The DIRAC user associated with the project. Defaults to None.
gbasf2_setup_path (str, optional) – The path to the gbasf2 setup script. Defaults to “/cvmfs/belle.kek.jp/grid/gbasf2/pro/bashrc”.
task (optional) – Additional task information to pass to the status retrieval function. Defaults to None.

Returns:

True if the project exists on the grid, False otherwise.

Return type:

bool

b2luigi.contrib.gbasf2.process.run_with_gbasf2(cmd, *args, ensure_proxy_initialized=True, check=True, encoding='utf-8', capture_output=False, gbasf2_setup_path='/cvmfs/belle.kek.jp/grid/gbasf2/pro/bashrc', task=None, **kwargs)[source]#

Call a command in a subprocess with the gbasf2 environment.

This function wraps the execution of a command in a subprocess, ensuring that the gbasf2 environment is properly set up and optionally verifying that the DIRAC proxy is initialized. It provides additional features such as capturing output and specifying encoding.

Parameters:

cmd (str or list) – The command to execute. Can be a string or a list of arguments.
*args – Additional positional arguments to pass to subprocess.run.
ensure_proxy_initialized (bool, optional) – If True, ensures that the DIRAC proxy is initialized and alive. Defaults to True.
check (bool, optional) – If True, raises a subprocess.CalledProcessError if the command exits with a non-zero status. Defaults to True.
encoding (str, optional) – The encoding to use for interpreting the command output. Defaults to “utf-8”.
capture_output (bool, optional) – If True, captures the stdout and stderr of the command. Equivalent to setting them to subprocess.PIPE. Defaults to False.
gbasf2_setup_path (str, optional) – Path to the gbas`f2 setup file. Defaults to “/cvmfs/belle.kek.jp/grid/gbasf2/pro/bashrc”.
task (optional) – Task-specific information to pass to the environment setup. Defaults to None.
**kwargs – Additional keyword arguments to pass to subprocess.run.

Returns:

An instance representing the completed process.

Return type:

subprocess.CompletedProcess

Raises:

ValueError – If stdout or stderr are specified in kwargs while capture_output is True.

Notes

If capture_output is True, the stdout and stderr of the command are captured and returned as strings (decoded using the specified encoding).
The gbasf2 environment is set up using the get_gbasf2_env function.
If ensure_proxy_initialized is True, the DIRAC proxy is initialized using the setup_dirac_proxy function.

b2luigi.contrib.gbasf2.process.get_gbasf2_env(gbasf2_setup_path, task=None)[source]#

Retrieve the gbasf2 environment as a dictionary, which can be used to run gbasf2 commands.

This function sets up the gbasf2 environment by sourcing the specified setup file in a fresh shell and capturing the resulting environment variables. It ensures that the environment is isolated, except for the HOME variable, which is required for the setup process.

Parameters:

gbasf2_setup_path (str) – Path to the gbasf2 setup file.
task (optional) – Task parameter used to retrieve the gbasf2_proxy_group setting. Defaults to "belle" if not provided.

Returns:

A dictionary containing the environment variables set up by the gbasf2 setup file.

Return type:

dict

Raises:

FileNotFoundError – If the specified gbasf2 setup file does not exist.

b2luigi.contrib.gbasf2.process.get_proxy_info(gbasf2_setup_path='/cvmfs/belle.kek.jp/grid/gbasf2/pro/bashrc', task=None)[source]#

Retrieve a dictionary containing the proxy status by running the gbasf2_proxy_info.py script.

Parameters:

gbasf2_setup_path (str, optional) – The path to the gbasf2 setup script. Defaults to “/cvmfs/belle.kek.jp/grid/gbasf2/pro/bashrc”.
task (optional) – An optional task parameter to pass to the subprocess.

Returns:

A dictionary containing the proxy status information.

Return type:

dict

Notes

The function uses the run_with_gbasf2 utility to execute the gbasf2_proxy_info.py script.

b2luigi.contrib.gbasf2.process.get_dirac_user(gbasf2_setup_path='/cvmfs/belle.kek.jp/grid/gbasf2/pro/bashrc', task=None)[source]#

Retrieve the DIRAC username associated with the initialized proxy.

Parameters:

gbasf2_setup_path (str) – The file path to the gbasf2 setup script. Defaults to “/cvmfs/belle.kek.jp/grid/gbasf2/pro/bashrc”.
task (optional) – An optional task object that may be used during proxy setup.

Returns:

The DIRAC username extracted from the proxy information.

Return type:

str

Raises:

RuntimeError – If the username cannot be obtained from the proxy information.

b2luigi.contrib.gbasf2.process.setup_dirac_proxy(gbasf2_setup_path='/cvmfs/belle.kek.jp/grid/gbasf2/pro/bashrc', task=None)[source]#

Ensures that a valid DIRAC proxy is initialized for the Belle II grid system. If a proxy is already active and has sufficient remaining lifetime, no action is taken. Otherwise, the function initializes a new proxy using the gb2_proxy_init command.

Parameters:

gbasf2_setup_path (str) – Path to the gbasf2 setup script. Defaults to “/cvmfs/belle.kek.jp/grid/gbasf2/pro/bashrc”.
task (optional) – Task-specific context or configuration, if applicable.

Behavior:

Checks if an active proxy exists and has sufficient lifetime remaining.
If no valid proxy exists, initializes a new proxy using the gb2_proxy_init command.
Prompts the user for the certificate password during proxy initialization.
Validates the proxy initialization process and handles errors such as incorrect passwords or other initialization issues.

Raises:: RuntimeWarning – If the gbasf2_proxy_lifetime setting is not a positive integer.

Notes

The proxy lifetime and group name can be configured using the gbasf2_proxy_lifetime and gbasf2_proxy_group settings, respectively.
The function ensures sensitive information like passwords is securely handled and deleted after use.

b2luigi.contrib.gbasf2.process.query_lpns(ds_query: str, dirac_user: str | None = None, gbasf2_setup_path: str = '/cvmfs/belle.kek.jp/grid/gbasf2/pro/bashrc', task=None) → List[str][source]#

Query DIRAC for Logical Physical Names (LPNs) matching the given dataset query.

This function interacts with the DIRAC system to retrieve a list of LPNs based on the provided dataset query string. It uses the gbasf2_ds_list.py script to perform the query and parses the output to return the results as a list of strings.

Parameters:

ds_query (str) – The dataset query string to search for matching LPNs.
dirac_user (Optional[str]) – The DIRAC user to perform the query as. If not provided, it will be determined using the get_dirac_user function.
gbasf2_setup_path (str) – The path to the gbasf2 setup script. Defaults to “/cvmfs/belle.kek.jp/grid/gbasf2/pro/bashrc”.
task – An optional task to associate with the query.

Returns:

A list of LPNs matching the dataset query.

Return type:

List[str]

Raises:

TypeError – If the output of the query is not a list.

b2luigi.contrib.gbasf2.process.get_unique_project_name(task)[source]#

Combine the gbasf2_project_name_prefix setting and the task_id hash to a unique project name.

This is done to make sure that different instances of a task with different luigi parameters result in different gbasf2 project names. When trying to redoing a task on the grid with identical parameters, rename the project name prefix, to ensure that you get a new project.

Parameters:

task – A task. The task must have a gbasf2_project_name_prefix setting, parameter, or attribute.

Returns:

A unique project name combining the prefix and a hashed task ID.

Return type:

str

Raises:

Exception – If the task does not have a gbasf2_project_name_prefix setting, parameter, or attribute.
ValueError – If the generated project name exceeds the maximum allowed length (32 characters) or contains invalid characters (non-alphanumeric, excluding _ and -).

Notes

The task ID hash is generated using hashlib.md5 for a consistent and compact representation.
The maximum length of the project name is 32 characters.
Only alphanumeric characters, underscores (_), and hyphens (-) are allowed in the project name.

b2luigi.contrib.gbasf2.process.lfn_follows_gb2v5_convention(lfn: str) → bool[source]#

Check if the LFN follows the convention of gbasf2 release 5, i.e.: <name>_<gbasf2param>_<jobID>_<rescheduleNum>.root

The expected naming convention is:

lfn (str): Logical file name, typically a file path on the grid.

Returns:: True if the LFN adheres to the gbasf2 release 5 naming convention, False otherwise.
Return type:: bool

b2luigi.contrib.gbasf2.process._get_lfn_upto_reschedule_number(lfn: str) → str[source]#

Get a string of the gbasf2 v5 LFN upto the reschule number.

E.g. if the LFN is <name>_<gbasf2param>_<jobID>_<rescheduleNum>.root return ``<name>_<gbasf2param>_<jobID>.

b2luigi.contrib.gbasf2.process.get_unique_lfns(lfns: Iterable[str]) → Set[str][source]#

From a list of gbasf2 Logical File Names (LFNs) that may include duplicates due to rescheduled jobs, return a filtered set containing only the LFNs corresponding to the jobs with the highest reschedule number.

Gbasf2 v5 outputs have the format: <name>_<gbasf2param>_<jobID>_<rescheduleNum>.root. When using gb2_ds_list, duplicates may appear where all parts are identical except for the rescheduleNum. This function ensures that only the LFNs with the highest rescheduleNum are retained.

If the dataset does not follow the gbasf2 v5 naming convention, it is assumed to be produced with an older release and does not contain duplicates.

Parameters:: lfns (Iterable[str]) – A collection of LFNs to process.
Returns:: A set of LFNs with duplicates removed, retaining only the ones with the highest reschedule number for each job.
Return type:: Set[str]

b2luigi.contrib.gbasf2.process._move_downloaded_dataset_to_output_dir(project_download_path: str, output_path: str) → None[source]#

Move files downloaded downloaded to the grid to their final output location.

In the Gbasf2Process, the outputs are usually downloaded with gb2_ds_get to a temporary directory (project_download_path) with a structure like

<result_dir>/B.root.partial/<project_name> ├── sub00/job_name*B.root ├── sub01/job_name*B.root ├── …

This function moves those files to their final output_path directory which has the same name as the original root file (e.g. B.root) to fulfill the luigi output definition. This output directory has the structure

<result_dir>/B.root/job_name*B.root

Parameters:

project_download_path – Directory into which gb2_ds_get downloaded the grid dataset. The contents should be sub<xy> data blocks containing root files.
output_path – Final output directory into which the ROOT files should be copied.

b2luigi.contrib.gbasf2.process._split_all_extensions(path: str) → Tuple[str, str][source]#

Split all extensions from a string pathname.

Similar to os.path.splitext, but with the difference that the extensions-part is considered everything from the first non-leading dot to to the end. Leading dots at the beginning of path are considered part of the stem and not considered an extensions separator.

Returns (stem, extensions).

extensions may be empty if path does not contain non-leading dots.

gbasf2

Contents

gbasf2#

API Reference#

`gbasf2`#