How to use the b2luigi.basf2_helper.utils.get_basf2_git_hash function in b2luigi

To help you get started, we’ve selected a few b2luigi examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github nils-braun / b2luigi / b2luigi / batch / processes / gbasf2.py View on Github external
def _build_gbasf2_submit_command(self):
        """
        Function to create the gbasf2 submit command to pass to run_with_gbasf2
        from the task options and attributes.
        """
        gbasf2_release = get_setting("gbasf2_release", default=get_basf2_git_hash(), task=self.task)
        gbasf2_additional_files = get_setting("gbasf2_additional_files", default=[], task=self.task)
        assert not isinstance(gbasf2_additional_files, str), "gbasf2_additional_files should be a list or tuple, not a string."
        gbasf2_input_sandbox_files = [os.path.basename(self.pickle_file_path)] + gbasf2_additional_files
        gbasf2_command_str = (f"gbasf2 {self.wrapper_file_path} -f {' '.join(gbasf2_input_sandbox_files)} " +
                              f"-p {self.gbasf2_project_name} -s {gbasf2_release} ")

        gbasf2_input_dataset = get_setting("gbasf2_input_dataset", default=False, task=self.task)
        if gbasf2_input_dataset is not False:
            gbasf2_command_str += f" -i {gbasf2_input_dataset} "

        gbasf2_n_repition_jobs = get_setting("gbasf2_n_repition_job", default=False, task=self.task)
        if gbasf2_n_repition_jobs is not False:
            gbasf2_command_str += f" --repetition {gbasf2_n_repition_jobs} "

        # now add some additional optional options to the gbasf2 job submission string
github nils-braun / b2luigi / b2luigi / basf2_helper / tasks.py View on Github external
import collections
import os
import shutil

import b2luigi
from b2luigi.basf2_helper.targets import ROOTLocalTarget
from b2luigi.basf2_helper.utils import get_basf2_git_hash

import subprocess

from b2luigi.core.utils import create_output_dirs, get_serialized_parameters


class Basf2Task(b2luigi.DispatchableTask):
    git_hash = b2luigi.Parameter(default=get_basf2_git_hash())

    def get_output_file_target(self, *args, **kwargs):
        file_name = self.get_output_file_name(*args, **kwargs)
        if os.path.splitext(file_name)[-1] == ".root":
            return ROOTLocalTarget(file_name)
        else:
            return super().get_output_file_target(*args, **kwargs)

    def get_serialized_parameters(self):
        serialized_parameters = get_serialized_parameters(self)

        # Git hash should go to the front
        return_dict = collections.OrderedDict()
        return_dict["git_hash"] = serialized_parameters["git_hash"]

        for key, value in serialized_parameters.items():
github nils-braun / b2luigi / b2luigi / basf2_helper / tasks.py View on Github external
def process(self):
        assert get_basf2_git_hash() == self.git_hash

        try:
            import basf2
            import ROOT
        except ImportError:
            raise ImportError("Can not find ROOT or basf2. Can not use the basf2 task.")

        if self.num_processes:
            basf2.set_nprocesses(self.num_processes)

        if self.max_event:
            ROOT.Belle2.Environment.Instance().setNumberEventsOverride(self.max_event)

        path = self.create_path()

        path.add_module("Progress")