Component.py

Go to the documentation of this file.

import os
import subprocess
import time
import datetime
import common_tools as tools
import SAMView as sam_view
import glob

class Component():

    def __init__(self, config_dict, release="development", local=False, verbose=False, user=False):
        self.local    = local
        self.verbose  = verbose
        self.user     = user
        # compulsory augments
        self.name     = config_dict["name"]
        self.version  = int(config_dict["version"])
        self.script   = config_dict["script"]
        self.dataset  = config_dict["dataset"]
        # optional with false defaults
        optional_attributes = ["datasets","local_dataset","local_datasets","sam","directories"]
        for attr in optional_attributes:
            if attr in config_dict.keys():
                setattr(self, attr, config_dict[attr])
            else:
                setattr(self, attr, False)
        # optional with non-false defaults
        if "label" in config_dict.keys():
            self.label = config_dict["label"]
        else:
            self.label = self.dataset

        if "release" in config_dict.keys():
            self.release = config_dict["release"]
        else:
            self.release = release

        string_attributes = ["n_jobs","files_per_job"]
        for attr in string_attributes:
            if attr in config_dict.keys():
                setattr(self, attr, config_dict[attr])
            else:
                setattr(self, attr, "")

        self.status   = "initialised"
        self.sam_merge_status = "not run"

        if (self.local or self.sam):
            self.jobsub_command = "sh %s"%(os.environ["NOVAANAVALID_DIR"])
        else:
            self.jobsub_command = "jobsub_submit -G nova --resource-provides=usage_model=DEDICATED,OPPORTUNISTIC --expected-lifetime=3600s --OS=SL6 file://%s"%(os.environ["NOVAANAVALID_DIR"])

        self.get_dataset()

    def run(self):
        self.status = "run"
        
        self.command = "%s/%s %s %s %s %s %s %s %s"%(self.jobsub_command, self.script, self.user, self.release, self.script_dataset, self.output_dir, self.name, self.n_jobs, self.files_per_job)
        if self.verbose: print "comp :     command: %s"%self.command
        
        process = subprocess.Popen(self.command.split(" "), stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        
        output = process.stdout
        self.lines = output.readlines()
        self.good_submission = False
        for line in self.lines:
            if self.verbose: print line.strip()
            if "JobsubJobId of first job" in line: self.good_submission = True
        
        if (not self.good_submission):
            print "comp : ERROR in submission"
            self.status = "submission error"

        self.run_time = datetime.datetime.now()

    def get_dataset(self):
        if   self.local_datasets: self.script_dataset = " ".join(self.local_datasets)
        elif self.datasets:       self.script_dataset = " ".join(self.datasets)
        elif self.local_dataset:  self.script_dataset = self.local_dataset
        else:                     self.script_dataset = self.dataset

    def setup_sam_view(self):
        sam_url = tools.getSAMURL(self.lines)
        if self.verbose: print "comp : found sam URL: %s"%sam_url
        self.sam_view = sam_view.SamView(sam_url, self.output_dir, verbose=self.verbose)

    def get_sam_status(self, sam):
        if (not hasattr(self, "sam_view")):
            self.setup_sam_view()
        
        self.n_files = self.get_expected_sam_output_files(sam)
        return self.sam_view.get_status(sam, n_files=self.n_files)

    def get_expected_sam_output_files(self, sam):
        # work out the number of files we're expecting
        if (self.files_per_job == ""):
            n_files = False
        elif (self.files_per_job != "") and (self.n_jobs == ""):
            # in this case the number of n_files = files_per_job * n_jobs
            # as n_jobs isn't set here it is determined at run time:
            # the default way to do this is as N_TOTAL_FILES / 100
            total_files = sam.countFiles(dimensions="snapshot_id %s"%self.sam_view.sam_summary["snapshot_id"])
            self.n_jobs = int(total_files / 100)
            n_files = self.n_jobs * self.files_per_job
        else:
            n_files = self.n_jobs * self.files_per_job
        return n_files

    def get_sam_merge_status(self):
        if (not self.sam):
            return "not applicable"
        if (not hasattr(self, "sam_merge_status")):
            self.sam_merge_status = "not run"
        return self.sam_merge_status

    def run_sam_merge(self):
        if hasattr(self, "sam_merge_status") and (self.sam_merge_status != "not run"):
            if (self.sam_merge_status != "complete"):
                print "comp : SAM merge has already run, status: %s"%self.sam_merge_status
            return

        self.sam_merge_status = "running"
        output_files = glob.glob("%s/*.root"%self.output_dir)
        if (not len(output_files)):
            self.sam_merge_status = "no files"
            return
        self.root_file = "%s/hadd_%s_%s.root"%(self.output_dir, self.name, self.version)
        cmd = 'python %s/core/utils/hadd_caf.py -i %s/ -w * -o %s -s'%(os.environ["NOVAANAVALID_DIR"], self.output_dir, self.root_file)
        
        process = subprocess.Popen(cmd.split(" "), stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        lines = []
        while True:
            line = process.stdout.readline()
            if line != "": lines.append(line)
            if line == '' and process.poll() != None: break
            # time.sleep(0.1)
        
        if (process.returncode != 0):
            print "comp : --- Exception caught while executing sam merge, return code: %i"%process.returncode
            print "comp :     command run: %s"%cmd
            for line in lines:
                print line.strip()
            self.sam_merge_status = "failed: return code = %i"%process.returncode

        elif (not os.path.exists(self.root_file)):
            self.sam_merge_status = "failed: no output file"
        else:
            self.sam_merge_status = "complete"