Inheritance diagram for Job:

Public Member Functions
	__init__ (self, name, job_dict=None)

	__repr__ (self)

	ready (self)

	update_status (self)

	create_subjob (self, i, input_files=None, args=None)

	status (self)

	status (self, status)

	dump_to_json (self, file_path)

	from_json (cls, file_path)

	job_dict (self)

	dump_input_data (self)

	copy_input_sandbox_files_to_working_dir (self)

	check_input_data_files (self)

	full_command (self)

	append_current_basf2_setup_cmds (self)

Public Attributes
	name = name
	Job object's name.

	splitter = None
	The `SubjobSplitter` used to create subjobs if necessary.

list	input_sandbox_files = []
	Files to be copied directly into the working directory (`pathlib.Path`).

	working_dir = Path()
	Working directory of the job (`pathlib.Path`).

	output_dir = Path()
	Output directory (`pathlib.Path`), where we will download our output_files to.

list	output_patterns = []
	Files that we produce during the job and want to be returned.

list	cmd = []
	Command and arguments as a list that will be run by the job on the backend.

list	args = []
	The arguments that will be applied to the `cmd` (These are ignored by SubJobs as they have their own arguments)

list	input_files = []
	Input files to job (`str`), a list of these is copied to the working directory.

list	setup_cmds = []
	Bash commands to run before the main self.cmd (mainly used for batch system setup)

dict	backend_args = {}
	Config dictionary for the backend to use when submitting the job.

dict	subjobs = {}
	dict of subjobs assigned to this job

	result = None
	The result object of this Job.

	status = job_status
	Not a real attribute, it's a property.

Static Public Attributes
dict	statuses = {"init": 0, "submitted": 1, "running": 2, "failed": 3, "completed": 4}
	Allowed Job status dictionary.

list	exit_statuses = ["failed", "completed"]
	Job statuses that correspond to the Job being finished (successfully or not)

Protected Member Functions
	_get_overall_status_from_subjobs (self)

Protected Attributes
str	_status = "init"
	The actual status of the overall `Job`.

Detailed Description

This generic Job object is used to tell a Backend what to do.
This object basically holds necessary information about a process you want to submit to a `Backend`.
It should *not* do anything that is backend specific, just hold the configuration for a job to be
successfully submitted and monitored using a backend. The result attribute is where backend
specific job monitoring goes.

Parameters:
    name (str): Simply a name to describe the Job, not used for any critical purpose in the CAF

.. warning:: It is recommended to always use absolute paths for files when submitting a `Job`.

Definition at line 336 of file backends.py.

Constructor & Destructor Documentation

◆ init()

__init__	(	self,
		name,
		job_dict = None )

Definition at line 357 of file backends.py.

    def __init__(self, name, job_dict=None):
        """
        """
        
        self.name = name
        
        self.splitter = None
        
        self.input_sandbox_files = []
        
        self.working_dir = Path()
        
        self.output_dir = Path()
        
        self.output_patterns = []
        
        self.cmd = []
        
        self.args = []
        
        self.input_files = []
        
        self.setup_cmds = []
        
        self.backend_args = {}
        
        self.subjobs = {}
 
        if job_dict:
            self.input_sandbox_files = [Path(p) for p in job_dict["input_sandbox_files"]]
            self.working_dir = Path(job_dict["working_dir"])
            self.output_dir = Path(job_dict["output_dir"])
            self.output_patterns = job_dict["output_patterns"]
            self.cmd = job_dict["cmd"]
            self.args = job_dict["args"]
            self.input_files = job_dict["input_files"]
            self.setup_cmds = job_dict["setup_cmds"]
            self.backend_args = job_dict["backend_args"]
            for subjob_dict in job_dict["subjobs"]:
                self.create_subjob(subjob_dict["id"], input_files=subjob_dict["input_files"], args=subjob_dict["args"])
 
        
        self.result = None
        
        self._status = "init"
 

Member Function Documentation

◆ repr()

__repr__ ( self )

Representation of Job class (what happens when you print a Job() instance).

Definition at line 406 of file backends.py.

    def __repr__(self):
        """
        Representation of Job class (what happens when you print a Job() instance).
        """
        return f"Job({self.name})"
 

◆ _get_overall_status_from_subjobs()

_get_overall_status_from_subjobs ( self )

protected

Definition at line 463 of file backends.py.

    def _get_overall_status_from_subjobs(self):
        """
        """
        subjob_statuses = [subjob.status for subjob in self.subjobs.values()]
        status_level = min([self.statuses[status] for status in subjob_statuses])
        for status, level in self.statuses.items():
            if level == status_level:
                return status
 

◆ append_current_basf2_setup_cmds()

append_current_basf2_setup_cmds ( self )

This adds simple setup commands like ``source /path/to/tools/b2setup`` to your `Job`.
It should detect if you are using a local release or CVMFS and append the correct commands
so that the job will have the same basf2 release environment. It should also detect
if a local release is not compiled with the ``opt`` option.

Note that this *doesn't mean that every environment variable is inherited* from the submitting
process environment.

Definition at line 640 of file backends.py.

    def append_current_basf2_setup_cmds(self):
        """
        This adds simple setup commands like ``source /path/to/tools/b2setup`` to your `Job`.
        It should detect if you are using a local release or CVMFS and append the correct commands
        so that the job will have the same basf2 release environment. It should also detect
        if a local release is not compiled with the ``opt`` option.
 
        Note that this *doesn't mean that every environment variable is inherited* from the submitting
        process environment.
        """
        def append_environment_variable(cmds, envvar):
            """
            Append a command for setting an environment variable.
            """
            if envvar in os.environ:
                cmds.append(f"""if [ -z "${{{envvar}}}" ]; then""")
                cmds.append(f"  export {envvar}={os.environ[envvar]}")
                cmds.append("fi")
 
        if "BELLE2_TOOLS" not in os.environ:
            raise BackendError("No BELLE2_TOOLS found in environment")
        # Export all the environment variables defined via _backend_job_envvars
        for envvar in _backend_job_envvars:
            append_environment_variable(self.setup_cmds, envvar)
        if "BELLE2_RELEASE" in os.environ:
            self.setup_cmds.append(f"source {os.environ['BELLE2_TOOLS']}/b2setup {os.environ['BELLE2_RELEASE']}")
        elif 'BELLE2_LOCAL_DIR' in os.environ:
            self.setup_cmds.append("export BELLE2_NO_TOOLS_CHECK=\"TRUE\"")
            self.setup_cmds.append(f"BACKEND_B2SETUP={os.environ['BELLE2_TOOLS']}/b2setup")
            self.setup_cmds.append(f"BACKEND_BELLE2_RELEASE_LOC={os.environ['BELLE2_LOCAL_DIR']}")
            self.setup_cmds.append(f"BACKEND_BELLE2_OPTION={os.environ['BELLE2_OPTION']}")
            self.setup_cmds.append("pushd $BACKEND_BELLE2_RELEASE_LOC > /dev/null")
            self.setup_cmds.append("source $BACKEND_B2SETUP")
            # b2code-option has to be executed only after the source of the tools.
            self.setup_cmds.append("b2code-option $BACKEND_BELLE2_OPTION")
            self.setup_cmds.append("popd > /dev/null")
 
 

◆ check_input_data_files()

check_input_data_files ( self )

Check the input files and make sure that there aren't any duplicates.
Also check if the files actually exist if possible.

Definition at line 598 of file backends.py.

    def check_input_data_files(self):
        """
        Check the input files and make sure that there aren't any duplicates.
        Also check if the files actually exist if possible.
        """
        existing_input_files = []  # We use a list instead of set to avoid losing any ordering of files
        for file_path in self.input_files:
            file_uri = parse_file_uri(file_path)
            if file_uri.scheme == "file":
                p = Path(file_uri.path)
                if p.is_file():
                    if file_uri.geturl() not in existing_input_files:
                        existing_input_files.append(file_uri.geturl())
                    else:
                        B2WARNING(f"Requested input file path {file_path} was already added, skipping it.")
                else:
                    B2WARNING(f"Requested input file path {file_path} does not exist, skipping it.")
            else:
                B2DEBUG(29, f"{file_path} is not a local file URI. Skipping checking if file exists")
                if file_path not in existing_input_files:
                    existing_input_files.append(file_path)
                else:
                    B2WARNING(f"Requested input file path {file_path} was already added, skipping it.")
        if self.input_files and not existing_input_files:
            B2WARNING(f"No valid input file paths found for {self.name}, but some were requested.")
 
        # Replace the Job's input files with the ones that exist + duplicates removed
        self.input_files = existing_input_files
 

◆ copy_input_sandbox_files_to_working_dir()

copy_input_sandbox_files_to_working_dir ( self )

Get all of the requested files for the input sandbox and copy them to the working directory.
Files like the submit.sh or input_data.json are not part of this process.

Definition at line 587 of file backends.py.

    def copy_input_sandbox_files_to_working_dir(self):
        """
        Get all of the requested files for the input sandbox and copy them to the working directory.
        Files like the submit.sh or input_data.json are not part of this process.
        """
        for file_path in self.input_sandbox_files:
            if file_path.is_dir():
                shutil.copytree(file_path, Path(self.working_dir, file_path.name))
            else:
                shutil.copy(file_path, self.working_dir)
 

◆ create_subjob()

create_subjob	(	self,
		i,
		input_files = None,
		args = None )

Creates a subjob Job object that references that parent Job.
Returns the SubJob object at the end.

Definition at line 435 of file backends.py.

    def create_subjob(self, i, input_files=None, args=None):
        """
        Creates a subjob Job object that references that parent Job.
        Returns the SubJob object at the end.
        """
        if i not in self.subjobs:
            B2INFO(f"Creating {self}.Subjob({i})")
            subjob = SubJob(self, i, input_files)
            if args:
                subjob.args = args
            self.subjobs[i] = subjob
            return subjob
        else:
            B2WARNING(f"{self} already contains SubJob({i})! This will not be created.")
 

◆ dump_input_data()

dump_input_data ( self )

Dumps the `Job.input_files` attribute to a JSON file. input_files should be a list of
string URI objects.

Definition at line 579 of file backends.py.

    def dump_input_data(self):
        """
        Dumps the `Job.input_files` attribute to a JSON file. input_files should be a list of
        string URI objects.
        """
        with open(Path(self.working_dir, _input_data_file_path), mode="w") as input_data_file:
            json.dump(self.input_files, input_data_file, indent=2)
 

◆ dump_to_json()

dump_to_json	(		self,
			file_path )

Dumps the Job object configuration to a JSON file so that it can be read in again later.

Parameters:
  file_path(`basf2.Path`): The filepath we'll dump to

Definition at line 538 of file backends.py.

    def dump_to_json(self, file_path):
        """
        Dumps the Job object configuration to a JSON file so that it can be read in again later.
 
        Parameters:
          file_path(`basf2.Path`): The filepath we'll dump to
        """
        # \cond false positive doxygen warning about job_dict
        with open(file_path, mode="w") as job_file:
            json.dump(self.job_dict, job_file, indent=2)
        # \endcond
 

◆ from_json()

from_json	(		cls,
			file_path )

Definition at line 551 of file backends.py.

    def from_json(cls, file_path):
        """
        """
        with open(file_path) as job_file:
            job_dict = json.load(job_file)
        return cls(job_dict["name"], job_dict=job_dict)
 

◆ full_command()

full_command ( self )

Returns:
    str: The full command that this job will run including any arguments.

Definition at line 628 of file backends.py.

    def full_command(self):
        """
        Returns:
            str: The full command that this job will run including any arguments.
        """
        all_components = self.cmd[:]
        all_components.extend(self.args)
        # We do a convert to string just in case arguments were generated as different types
        full_command = " ".join(map(str, all_components))
        B2DEBUG(29, f"Full command of {self} is '{full_command}'")
        return full_command
 

◆ job_dict()

job_dict ( self )

Returns:
    dict: A JSON serialisable representation of the `Job` and its `SubJob` objects.
    `Path <basf2.Path>` objects are converted to string via ``Path.as_posix()``.

Reimplemented in SubJob.

Definition at line 559 of file backends.py.

    def job_dict(self):
        """
        Returns:
            dict: A JSON serialisable representation of the `Job` and its `SubJob` objects.
            `Path <basf2.Path>` objects are converted to string via ``Path.as_posix()``.
        """
        job_dict = {}
        job_dict["name"] = self.name
        job_dict["input_sandbox_files"] = [i.as_posix() for i in self.input_sandbox_files]
        job_dict["working_dir"] = self.working_dir.as_posix()
        job_dict["output_dir"] = self.output_dir.as_posix()
        job_dict["output_patterns"] = self.output_patterns
        job_dict["cmd"] = self.cmd
        job_dict["args"] = self.args
        job_dict["input_files"] = self.input_files
        job_dict["setup_cmds"] = self.setup_cmds
        job_dict["backend_args"] = self.backend_args
        job_dict["subjobs"] = [sj.job_dict for sj in self.subjobs.values()]
        return job_dict
 

◆ ready()

ready ( self )

Returns whether or not the Job has finished. If the job has subjobs then it will return true when they are all finished.
It will return False as soon as it hits the first failure. Meaning that you cannot guarantee that all subjobs will have
their status updated when calling this method. Instead use :py:meth:`update_status` to update all statuses if necessary.

Definition at line 412 of file backends.py.

    def ready(self):
        """
        Returns whether or not the Job has finished. If the job has subjobs then it will return true when they are all finished.
        It will return False as soon as it hits the first failure. Meaning that you cannot guarantee that all subjobs will have
        their status updated when calling this method. Instead use :py:meth:`update_status` to update all statuses if necessary.
        """
        if not self.result:
            B2DEBUG(29, f"You requested the ready() status for {self} but there is no result object set, returning False.")
            return False
        else:
            return self.result.ready()
 

◆ status() [1/2]

status ( self )

Returns the status of this Job. If the job has subjobs then it will return the overall status equal to the lowest
subjob status in the hierarchy of statuses in `Job.statuses`.

Reimplemented in SubJob, and SubJob.

Definition at line 451 of file backends.py.

    def status(self):
        """
        Returns the status of this Job. If the job has subjobs then it will return the overall status equal to the lowest
        subjob status in the hierarchy of statuses in `Job.statuses`.
        """
        if self.subjobs:
            job_status = self._get_overall_status_from_subjobs()
            if job_status != self._status:
                
                self.status = job_status
        return self._status
 

◆ status() [2/2]

status	(		self,
			status )

Sets the status of this Job.

Reimplemented in SubJob, and SubJob.

Definition at line 473 of file backends.py.

    def status(self, status):
        """
        Sets the status of this Job.
        """
        # Print an error only if the job failed.
        if status == 'failed':
            B2ERROR(f"Setting {self.name} status to failed")
        else:
            B2INFO(f"Setting {self.name} status to {status}")
        self._status = status
 

◆ update_status()

update_status ( self )

Calls :py:meth:`update_status` on the job's result. The result object should update all of the subjobs (if there are any)
in the best way for the type of result object/backend.

Definition at line 424 of file backends.py.

    def update_status(self):
        """
        Calls :py:meth:`update_status` on the job's result. The result object should update all of the subjobs (if there are any)
        in the best way for the type of result object/backend.
        """
        if not self.result:
            B2DEBUG(29, f"You requested update_status() for {self} but there is no result object set yet. Probably not submitted.")
        else:
            self.result.update_status()
        return self.status
 

Member Data Documentation

◆ _status

str _status = "init"

protected

The actual status of the overall Job.

The property handles querying for the subjob status to set this

Definition at line 404 of file backends.py.

◆ args

args = []

The arguments that will be applied to the cmd (These are ignored by SubJobs as they have their own arguments)

Definition at line 376 of file backends.py.

◆ backend_args

dict backend_args = {}

Config dictionary for the backend to use when submitting the job.

Saves us from having multiple attributes that may or may not be used.

Definition at line 383 of file backends.py.

◆ cmd

list cmd = []

Command and arguments as a list that will be run by the job on the backend.

Definition at line 374 of file backends.py.

◆ exit_statuses

list exit_statuses = ["failed", "completed"]

static

Job statuses that correspond to the Job being finished (successfully or not)

Definition at line 355 of file backends.py.

◆ input_files

input_files = []

Input files to job (str), a list of these is copied to the working directory.

Definition at line 378 of file backends.py.

◆ input_sandbox_files

list input_sandbox_files = []

Files to be copied directly into the working directory (pathlib.Path).

Not the input root files, those should be in Job.input_files.

Definition at line 366 of file backends.py.

◆ name

name = name

Job object's name.

Only descriptive, not necessarily unique.

Reimplemented in SubJob.

Definition at line 361 of file backends.py.

◆ output_dir

output_dir = Path()

Output directory (pathlib.Path), where we will download our output_files to.

Default is '.'

Reimplemented in SubJob.

Definition at line 370 of file backends.py.

◆ output_patterns

list output_patterns = []

Files that we produce during the job and want to be returned.

Can use wildcard (*)

Definition at line 372 of file backends.py.

◆ result

result = None

The result object of this Job.

Only filled once the job is submitted to a backend since the backend creates a special result class depending on its type.

Definition at line 402 of file backends.py.

◆ setup_cmds

setup_cmds = []

Bash commands to run before the main self.cmd (mainly used for batch system setup)

Definition at line 380 of file backends.py.

◆ splitter

splitter = None

The SubjobSplitter used to create subjobs if necessary.

Definition at line 363 of file backends.py.

◆ status

status = job_status

Not a real attribute, it's a property.

Reimplemented in SubJob, and SubJob.

Definition at line 460 of file backends.py.

◆ statuses

dict statuses = {"init": 0, "submitted": 1, "running": 2, "failed": 3, "completed": 4}

static

Allowed Job status dictionary.

The key is the status name and the value is its level. The lowest level out of all subjobs is the one that is the overall status of the overall job.

Definition at line 352 of file backends.py.

◆ subjobs

dict subjobs = {}

dict of subjobs assigned to this job

Reimplemented in SubJob.

Definition at line 385 of file backends.py.

◆ working_dir

working_dir = Path()

Working directory of the job (pathlib.Path).

Default is '.', mostly used in Local() backend

Reimplemented in SubJob.

Definition at line 368 of file backends.py.

The documentation for this class was generated from the following file:

calibration/scripts/caf/backends.py

Public Member Functions

Public Attributes

Static Public Attributes

Protected Member Functions

Protected Attributes

Detailed Description

Constructor & Destructor Documentation

◆ __init__()

Member Function Documentation

◆ __repr__()

◆ _get_overall_status_from_subjobs()

◆ append_current_basf2_setup_cmds()

◆ check_input_data_files()

◆ copy_input_sandbox_files_to_working_dir()

◆ create_subjob()

◆ dump_input_data()

◆ dump_to_json()

◆ from_json()

◆ full_command()

◆ job_dict()

◆ ready()

◆ status() [1/2]

◆ status() [2/2]

◆ update_status()

Member Data Documentation

◆ _status

◆ args

◆ backend_args

◆ cmd

◆ exit_statuses

◆ input_files

◆ input_sandbox_files

◆ name

◆ output_dir

◆ output_patterns

◆ result

◆ setup_cmds

◆ splitter

◆ status

◆ statuses

◆ subjobs

◆ working_dir

◆ init()

◆ repr()