GBSTask.py

Go to the documentation of this file.

import glob
import os
import re

import Ganga.GPI

from   GBSIdCodes       import *
from   GBSConfig        import GetConfigValue
from   GBSLogger        import Log, logger 
from   GBSObject        import GBSObject
from   GBSModelRegistry import GetModelRegistry
from   GBSTimeStamp     import GBSTimeStamp
import GBSUtilities

class GBSTask(GBSObject) :
    """Top level object to manage a single task.

    This object will manages an set of Jobs, each of which looks after
    a single job, submitting and resubmitting until successful or user
    intervention is required.
      
    ProtoJobs
    =========

    Batches of new jobs are introduced into the system initially as "ProtoJobs".
    The following methods are used to manipulate them:-

    AddProtoJob(job_name,args_str="",env_str="")
    PromoteProtoJobs(job_name_pattern = ".*")
    RemoveProtoJobs(job_name_pattern = ".*")
       
    """

    ######  GBSObject inherited responsibilities   ###### 
      

    def __init__(self,name,parent,model,model_args):
        self.__scriptFileName      = ""
        self.__scriptGlobalArgs    = ""
        self.__globalEnvironment   = ""
        self.__globalInputSandbox  = ""
        self.__globalOutputSandbox = ""
        self.__backend             = GetConfigValue("DefaultBackend")
        self.__submitEnabled       = 0
        self.__maxGangaJobs        = int(GetConfigValue("DefaultMaxGangaJobs"))
        self.__maxSubmitJobs       = int(GetConfigValue("DefaultMaxSubmitJobs"))
        self.__mode                = "Test"

        # The remaining members are volatile i.e. not persisted but derived from other state.
        # Note that _jobManagers is not private so that MinosRSMTask can rename jobs.
        self._jobManagers       = {}   # Volatile: Hash name -> Job
        self.__gangaTreeDir     = 'gbs/' + name # Ganga JobTree directory that holds jobs.

        self.__jstatAll         = 0 # Job stats: All
        self.__jstatNReadyAll   = 0 # Job stats: Not ready - all
        self.__jstatNReadyOther = 0 # Job stats: Not ready - other than placed on hold
        self.__jstatReadyAll    = 0 # Job stats: Ready to run - all
        self.__jstatReadyRetry  = 0 # Job stats: Ready to run - retries 
        self.__jstatSubmitAll   = 0 # Job stats: Submitted to Ganga - all
        self.__jstatSubmitNRun  = 0 # Job stats: Submitted to Ganga - but not running
        self.__jstatDoneAll     = 0 # Job stats: Done - all
        self.__jstatDoneFail    = 0 # Job stats: Done - Failed

        GBSObject.__init__(self,name,parent,model)

        # Create Ganga JobTree directory if necessary
        if not Ganga.GPI.jobtree.exists(self.__gangaTreeDir): Ganga.GPI.jobtree.mkdir(self.__gangaTreeDir)

        self.MakeChildDirectory()
        self.__ReloadChildren()
        self.RefreshJobStats()


    def _DoMemberIO(self,ioh):
        self.__scriptFileName       = ioh("Script file name",              "s",self.__scriptFileName)
        self.__scriptGlobalArgs     = ioh("Script Global Args",            "s",self.__scriptGlobalArgs)
        self.__globalEnvironment    = ioh("+Global environment",           "s",self.__globalEnvironment)
        self.__globalInputSandbox   = ioh("+Global Input Sandbox List",    "s",self.__globalInputSandbox)
        self.__globalOutputSandbox  = ioh("+Global Output Sandbox List",   "s",self.__globalOutputSandbox)
        self.__backend              = ioh("Backend",                       "s",self.__backend) 
        self.__submitEnabled        = ioh("Submit Enabled",                "i",self.__submitEnabled) 
        self.__maxGangaJobs         = ioh("Max Ganga Jobs",                "i",self.__maxGangaJobs) 
        self.__maxSubmitJobs        = ioh("Max Submit Jobs",               "i",self.__maxSubmitJobs) 
        self.__mode                 = ioh("+Test/Production Mode",         "s",self.__mode) 
        GBSObject._DoMemberIO(self,ioh)

    def WriteFamily(self):
        self.Write()
        for job_name,job in self._jobManagers.iteritems():job.WriteFamily()

    def GetType(self): return "GBSTask"

    def __repr__(self) : return self.AsString()
    

#-----------------------------------------------------------------------------------------------
#
#  User Callable Methods Getters
#
#-----------------------------------------------------------------------------------------------


    def AsString(self,level = "Full"):

        """Return string description.

         Return string description at the following levels:-
        "Brief"    one line summary suitable for rows in tables
        "Heading"  one line summary suitable as heading for "Brief"
        "Full"     full desciption including value of every data member"""

        if ( level == "Heading"):
            s  = "Name".ljust(20)
            s += "Model".ljust(15)
            s += "!Ready(other)  Ready(retry)    Sub(!run)   Done(fail)".rjust(51)
            s += "Backend".rjust(10) + "  "
            s += "Script file".ljust(40)
            return s
        if ( level == "Brief"):
            s  = self.GetName().ljust(20)
            s += self.GetModel().ljust(15)
            s += str(self.__jstatNReadyAll).rjust(8)   + "("+ str(self.__jstatNReadyOther).rjust(3) + ")" 
            s += str(self.__jstatReadyAll).rjust(9)   + "("+ str(self.__jstatReadyRetry).rjust(3) + ")" 
            s += str(self.__jstatSubmitAll).rjust(8) + "("+ str(self.__jstatSubmitNRun).rjust(3) + ")" 
            s += str(self.__jstatDoneAll).rjust(8)   + "("+ str(self.__jstatDoneFail).rjust(3) + ")" 
            backend = self.__backend
            mo = re.search(r"(.*?):",backend)
            if mo: backend = mo.group(1)
            s += backend.rjust(10) + "  "
            s += self.__scriptFileName.ljust(40)
            return s

        s  = GBSObject.__repr__(self) + "\n\n"
        s += "Model: " + self.GetModel() + "\n\n"
        s += "Job Definition\n"
        s += "  Script file:     "  + self.__scriptFileName  + "\n"
        s += "  Global args:     '" + self.__scriptGlobalArgs + "'\n"
        s += "  Global env:      '" + self.__globalEnvironment + "'\n" 
        s += "  Input Sandbox:   '"  + self.__globalInputSandbox + "'\n" 
        s += "  Output Sandbox:  '"  + self.__globalOutputSandbox + "'\n\n" 
        s += "Job Submission\n"
        s += "  "
        if self.__submitEnabled: s += "Enabled\n"
        else:                    s += "Disabled\n"
        s += "  Backend: " + self.__backend + "\n"
        s += "  Limits:  " + str(self.__maxSubmitJobs) + "(single submit) " + str(self.__maxGangaJobs) + "(maximum)\n"
        s += "  Mode:    " + self.__mode  + "\n\n" 
        s += "Managing "  + str(self.__jstatAll) + " jobs\n"
        s += "  Holding:   " + str(self.__jstatNReadyAll).rjust(5)   + " ("+ str(self.__jstatNReadyOther) + " other)\n" 
        s += "  Waiting:   " + str(self.__jstatReadyAll).rjust(5)   + " ("+ str(self.__jstatReadyRetry) + " retries)\n" 
        s += "  Submitted: " + str(self.__jstatSubmitAll).rjust(5) + " ("+ str(self.__jstatSubmitNRun) + " not running)\n" 
        s += "  Done:      " + str(self.__jstatDoneAll).rjust(5)   + " ("+ str(self.__jstatDoneFail) + " failed)\n" 
        return s
 

    def GetBackend(self):
        """Return backend for future jobs"""
        return self.__backend
    
    def GetGangaTreeDir(self):
        """Return Ganga JobTree Directory that holds all Ganga jobs"""
        return self.__gangaTreeDir

    def GetGlobalEnvironment(self,prettyPrint = False):
        """Return, as a comma separated list string, the environment that is global to all jobs

        If prettyPrint is True print out list one item per line"""
        if not prettyPrint: return self.__globalEnvironment
        print "Global Environment:-"
        GBSUtilities.ListCommaSepList(self.__globalEnvironment,"  ")
    
    def GetGlobalInputSandbox(self):
        """Return, as a comma separated list string, the input sandbox file list that is global to all jobs"""
        return self.__globalInputSandbox
    
    def GetGlobalOutputSandbox(self):
        """Return, as a comma separated list string, the ouput sandbox file list that is global to all jobs"""
        return self.__globalOutputSandbox
    
    def GetMaxGangaJobs(self):
        """Return the maximum number of jobs that can be submitted to Ganga at any one time """
        return self.__maxGangaJobs
    
    def GetMaxSubmitJobs(self):
        """Return the maximum number of jobs that can be submitted by a single call SubmitJobs() """
        return self.__maxSubmitJobs
    
    def GetMode(self):
        """Return the Test/Production Mode """
        return self.__mode
    
    def GetScriptFileName(self):
        """Returns the current user application script file name (or "" if none). """
        return self.__scriptFileName

    def GetScriptFileSpec(self):
        if not self.__scriptFileName : return ""
        return self.GetStoreLocation("child_dir") + "/" + self.__scriptFileName

    def GetJob(self,name,warn=True) :
        """Return Job called job_name e.g. job = task.GetJob("job_00123")"""
        if not self._jobManagers.has_key(name):
            if warn: print "Sorry, there is no Job named " + str(name)
            return None
        else : return self._jobManagers[name]

    def GetJobs(self,job_name_pattern = ".*"):
        """Return a dictionary: dict[name]  job for all jobs matching pattern"""

        job_dict = {}
        for job_name,job in self._jobManagers.iteritems():
            if re.search(job_name_pattern,job_name): job_dict[job_name] = job
        return job_dict

    def GetScriptGlobalArgs(self):
        """Returns (as a string) the comma separated list of application script args that are global to all jobs """
        return self.__scriptGlobalArgs

    def GetTestOnlyMethods(self):
        """Return a list of methods that are only available in 'Test' mode."""
        return ["SetGlobalEnvironment",   \
                "SetGlobalInputSandbox",  \
                "SetGlobalOutputSandbox", \
                "SetScriptFileName",      \
                "SetScriptGlobalArgs"]    \

    def IsAuthorisedToSubmit(self,warn=True):
        """Returns True is authorised to submit jobs.

        Always True except if backend is LCG, then checks proxy"""
        if self.GetBackend()[0:3] != "LCG" : return True
        ok = False
        try:    ok = Ganga.GPI.gridProxy.isValid()
        except: ok = False
        if warn and not ok: print "Cannot submit jobs; backend is LCG but cannot find valid proxy"
        return ok        

    def IsDisabled(self,method):
        """Return True if method is disabled."""

        if self.GetMode() == 'Test': return False
        for disabled_method in self.GetTestOnlyMethods():
            if method == disabled_method:
                print "Method " + method + " is disabled in Production mode"
                return True
        return False
        
    def ListJobs(self,job_name_pattern = ".*"):

        """List existing Jobs matching pattern"""
        
        print self.GetName() + " has the following jobs",
        if job_name_pattern != ".*": print "that match the pattern: '" + str(job_name_pattern) + "'",
        print ":-"
        first = 1
        job_names = self._jobManagers.keys()
        job_names.sort()
        for job_name in job_names:
            if not re.search(job_name_pattern,job_name): continue
            job = self._jobManagers[job_name]
            if first:
                print job.AsString("Heading") + "\n"
                first = 0
            print job.AsString("Brief")
        if first: print " no jobs found"


#-----------------------------------------------------------------------------------------------
#
#  User Callable Methods Setters
#
#-----------------------------------------------------------------------------------------------


    def AddJob(self,job_name,args_str = "",env_str = ""):

        """Create a new named Job and optionally assign its application script local args and environment. 

      e.g. job = task.AddJob("job_00034621_0002","F00034621_0002.mdaq.root","mini_flux=no")

      Note that the job_name must begin job_ (if omitted it will be added automatically) and
      will be rejected if it contains anything beyond alphanumeric, '_' and '-'.for as it
      will be used to name a directory."""

        return self._AddJobOrProtoJob(job_name,args_str,env_str,"Job")

    def ClearErrorCountsJobs(self,job_name_pattern= ".*"):
        """After seeking confirmation, clear error counts on all suitable jobs matching pattern"""
        self.__ApplyActionToJobs("CLEAR ERROR COUNTS",job_name_pattern)

    def ClearHistoryJobs(self,job_name_pattern= ".*"):
        """After seeking confirmation, clear histories on all suitable jobs matching pattern"""
        self.__ApplyActionToJobs("CLEAR HISTORY",job_name_pattern)

    def EnableSubmit(self,enable=True):
        """Enable or disable submission of jobs via the SubmitJobs() method.
      The function returns the previous value of the switch."""
        
        old_val = self.__submitEnabled
        self.__submitEnabled  = 0
        if enable: self.__submitEnabled  = 1
        self.Write()
        return old_val

    def HoldJobs(self,job_name_pattern= ".*"):
        """After seeking confirmation, hold all suitable jobs matching pattern"""
        self.__ApplyActionToJobs("HOLD",job_name_pattern)

    def KillJobs(self,job_name_pattern= ".*"):
        """After seeking confirmation, kill all suitable jobs matching pattern"""
        self.__ApplyActionToJobs("KILL",job_name_pattern)

    def RefreshJobStats(self):

        """ Bring job statistics up to date.

        Should not be necessary (but harmless) for user to call this
        as child jobs call it when their state changes.  Note: unlike
        UpdateJobsStatus this simply asks jobs what their current
        state without checking Ganga """
        self.__jstatAll         = 0
        self.__jstatNReadyAll     = 0
        self.__jstatNReadyOther   = 0
        self.__jstatReadyAll     = 0
        self.__jstatReadyRetry   = 0
        self.__jstatSubmitAll   = 0
        self.__jstatSubmitNRun  = 0
        self.__jstatDoneAll     = 0
        self.__jstatDoneFail    = 0
        for job_name,job in self._jobManagers.iteritems():
            self.__jstatAll += 1
            pc = job.GetPhaseCode()
            if pc == GID_JPC_DONE_FAIL   or pc == GID_JPC_DONE_NFAIL:   self.__jstatDoneAll += 1
            if pc == GID_JPC_DONE_FAIL:                                 self.__jstatDoneFail += 1
            if pc == GID_JPC_READY_RETRY or pc == GID_JPC_READY_NRETRY: self.__jstatReadyAll += 1
            if pc == GID_JPC_READY_RETRY:                               self.__jstatReadyRetry += 1
            if pc == GID_JPC_SUBMIT_RUN  or pc == GID_JPC_SUBMIT_NRUN:  self.__jstatSubmitAll += 1
            if pc == GID_JPC_SUBMIT_NRUN:                               self.__jstatSubmitNRun += 1
            if pc == GID_JPC_NREADY_HOLD or pc == GID_JPC_NREADY_NHOLD: self.__jstatNReadyAll += 1
            if pc == GID_JPC_NREADY_NHOLD:                              self.__jstatNReadyOther += 1

    def ReleaseJobs(self,job_name_pattern = ".*"):
        """After seeking confirmation, release all suitable jobs matching pattern"""
        self.__ApplyActionToJobs("RELEASE",job_name_pattern)

    def RemoveJobs(self,job_name_pattern= ".*"):
        """After seeking confirmation, remove all suitable jobs matching pattern"""
        self.__ApplyActionToJobs("REMOVE",job_name_pattern)
        
    def SetBackend(self,backend):

        """Define backend for future jobs:  One of: Local, PBS:queue, LCG:queue"""

        #  Check value before accepting
        if not re.search(r"^(Local|PBS|LCG)(:.+)?$",backend):
            print "Sorry, '" + str(backend) + "' is not an acceptible backend. It should be one of:-\n"\
                  + "  Local, PBS{:queue} or LCG{:queue}"
            return
        self.__backend = backend
        self.Write()

    def SetGlobalEnvironment(self,env_str):

        """Set, as a comma separated list string, the environment that is global to all jobs.

      e.g.  task.SetGlobalEnvironment('var1=123,var2=a string with spaces,var3=456')

      If the first character is + e.g. +var3=567,var4=b, append to existing environment rather than replace it"""

        if self.IsDisabled("SetGlobalEnvironment"): return
        if re.search(r"^\+",env_str):
            s = self.__globalEnvironment
            if s: s += ','
            s += env_str[1:]
            # Force local method to be recalled rather than any inherited one
            # May just be paranoia but see corresponding GBSJob method.
            GBSTask.SetGlobalEnvironment(self,s)
            return
        d =  GBSUtilities.ParseEnvStr(env_str)
        if env_str and not d: print "Cannot parse environment string: '" + env_str + "'"
        else:
            self.__globalEnvironment = GBSUtilities.BuildEnvStr(d)
            self.Write()

    def SetGlobalInputSandbox(self,in_sbox_str):

        """Set, as a comma separated list string, the input sandbox file list that is global to all jobs.

      e.g.  task.SetGlobalInputSandbox('/home/users/west/my_input_data.dat,../my_script.sh')"""

        if self.IsDisabled("SetGlobalInputSandbox"): return
        (ok,str) = GBSUtilities.ProcessSandboxSetup(self,in_sbox_str,"input")
        if ok:
            self.__globalInputSandbox = str
            self.Write()

    def SetGlobalOutputSandbox(self,out_sbox_str):

        """Set, as a comma separated list string, the output sandbox file list that is global to all jobs.

      e.g.  task.SetGlobalOutputSandbox('my_output_data.dat,my_output.log')"""

        if self.IsDisabled("SetGlobalOutputSandbox"): return
        (ok,str) = GBSUtilities.ProcessSandboxSetup(self,out_sbox_str,"output")
        if ok:
            self.__globalOutputSandbox = str
            self.Write()

    def SetMaxGangaJobs(self,n):
        """Set the maximum number of jobs that can be submitted to Ganga at any one time"""
        
        self.__maxGangaJobs = n
        self.Write()

    def SetMaxSubmitJobs(self,n):
        """Set the maximum number of jobs that can be submitted by a single call SubmitJobs()"""
        
        self.__maxSubmitJobs = n
        self.Write()
    
    def SetMode(self,mode):
        """Set the Test/Production Mode """
        if mode != "Test" and mode != "Production":
            print "Illegal mode: '" + mode + "', only 'Test' and 'Production' permitted"
            return
        if self.__mode == mode: return
        if mode == 'Production':
            print "Caution:  Switching to 'Production' mode will disable the following methods:-\n"
            print str(self.GetTestOnlyMethods())
            print "\n but should be done before moving into full production."
        else:
            if not self.GetScriptFileName():
                print "You cannot switch to 'Production' mode; no global script file has been defined."
                return
            print "Caution: Switching back to 'Test' mode is potentially dangerous as it re-enables the following methods:-\n"
            print str(self.GetTestOnlyMethods())
            print "\n and using these methods could produce results that are not homogeneous"
        ans = raw_input("Do you want to proceed ?y[n]")
        if not re.search(r"^y",ans,re.I):
            print "Mode not changed."
            return
        print "Mode changed to " + mode
        self.__mode = mode
        self.Write()

    def SetScriptFileName(self,ext_file_spec):

        """Pass in script file to be used by jobs.  GBS make a local copy
        so no need to keep once passed.

        e.g.  task.SetScriptFileName("my/directory/my_script_file.sh")"""

        if self.IsDisabled("SetScriptFileName"): return
        import os
        if not os.path.isfile(ext_file_spec):
            print "Cannot find script file " + str(ext_file_spec)
            return
        self.__scriptFileName = os.path.basename(ext_file_spec)
        int_file_spec = self.GetScriptFileSpec()
        print "Copying " + str(ext_file_spec) + " -> " + str(int_file_spec)
        if os.system("cp " + str(ext_file_spec) + " " + str(int_file_spec)):
            print "Failed to copy file!"
            self.__scriptFileName = ""
        self.Write()

    def SetScriptGlobalArgs(self,arg_str):

        """Set (as a string) the comma separated list of application script args that are global to all jobs."""

        if self.IsDisabled("SetScriptGlobalArgs"): return
        self.__scriptGlobalArgs = arg_str
        self.Write()

    def SubmitJobs(self,num_req = 0):

        """If enabled (see EnableSubmit()),
        submit up to the limit (see GetMaxSubmitJobs()) jobs,
        but not to exceed the maximum (see GetMaxGangaJobs())
        first choosing ones scheduled for retry and
        then new jobs.  Function returns number submitted."""

        num_sub = 0

        # Begin by making sure that submit is authorised, enabled and that there is a global script.
        if not self.IsAuthorisedToSubmit(): return num_sub
        if not self.__submitEnabled:
            print "Sorry, cannot submit; submit not enabled."
            return num_sub
        if not self.__scriptFileName:
            print "Sorry, cannot submit; no user application script assigned."
            return num_sub
        
        # Make sure all jobs are up to date and that limits are not exceeded.
        self.UpdateJobsStatus()
        if self.__maxGangaJobs <= self.__jstatSubmitAll:
            print "Sorry, cannot submit; already have %d (limit %d)" % (self.__jstatSubmitAll,self.__maxGangaJobs)
            return num_sub
        if num_req <= 0: num_req = self.__maxSubmitJobs
        max_job = min(num_req,self.__maxGangaJobs - self.__jstatSubmitAll)
        if max_job < num_req:  print "Sorry, can only submit %d, close to the limit of %d" % (max_job,self.__maxGangaJobs)

        # First look for retries and then look for new jobs.
        for jsc in [GID_JSC_RETRY,GID_JSC_NEW]:
            job_names = self._jobManagers.keys()
            job_names.sort()
            for job_name in job_names:
                job = self._jobManagers[job_name]
                if job.GetStatusCode() != jsc: continue
                job.Submit()
                num_sub += 1
                if num_sub >= max_job: return num_sub
        return num_sub

    def UpdateJobsStatus(self):

        """Send all jobs their UpdateStatus().  May involve checking Ganga job and retrieving output.

        As a by product, also remove any orphaned Ganga jobs i.e. Ganga jobs in Task's jobtree folder
        but not owned by any Job"""

        # If job polling not enabled, perform a single sweep with a timeout based on the number
        # of Ganga jobs
        if not Ganga.GPI.config['PollThread']['autostart']:
            timeout = 60 + 60*self.__jstatSubmitAll/int(GetConfigValue("GangaRegistryPollRate"))
            Ganga.GPI.runMonitoring(1,timeout)
            
        owned_ganga_ids = {}
        for job_name,job in self._jobManagers.iteritems():
            job.UpdateStatus()
            owned_ganga_ids[job.GetGangaJobId()] = 1
        # Clean up dead links.  This ought not to be necessary but, at least as late as 4.4.8 it is.
        Ganga.GPI.jobtree.cleanlinks()
        for gj in Ganga.GPI.jobtree.getjobs(self.__gangaTreeDir):
            if owned_ganga_ids.has_key(gj.id): continue
            Log(self,logger.WARNING,"Removing orphaned Ganga job %d" % gj.id)
            gj.remove()
            
#-----------------------------------------------------------------------------------------------

    def WarnLowGridProxy(self,email_list,proxy_min_hours=3.,myproxy_min_days=3.):
        """If there is a valid GRID proxy, but it is running low:-

              1)  'proxy_min_hours' defined and Proxy lifetime is less than 'proxy_min_hours'
           or 2)  'myproxy_min_days defined and myproxy lifetime is less that 'myproxy_min_days'
"""
        # No action if there isn't an email list or a GRID Proxy
        if not email_list or os.system("voms-proxy-info --exists"): return

        proxy_warn = ""

        # Test GRID proxy 
        if proxy_min_hours:
            inp  = os.popen("voms-proxy-info --timeleft","r")
            line = inp.readline()
            inp.close()
            try:
                hours_left = int(line)/(60.*60.)
                if hours_left < proxy_min_hours: proxy_warn = "GRID proxy remaining lifetime = %.1f hours " % hours_left
            except: pass

        # Test myproxy
        if myproxy_min_days:
            inp  = os.popen("myproxy-info | grep timeleft","r")
            line = inp.readline()
            inp.close()
            mo = re.search(r"\s+(\d+):(\d+):(\d+)",line)
            if mo:
                days_left = ( (int(mo.group(3))/60.+int(mo.group(2)))/60. + int(mo.group(1)) )/24.
                if days_left < myproxy_min_days: proxy_warn += "myproxy remaining lifetime = %.1f days" % days_left

        if proxy_warn:
            cmd  = 'mail -s"Task:%s WARNING: %s"  %s<< EOD\n' % (self.GetName(),proxy_warn,email_list)
            cmd += "\nPlease log in and renew the proxy/myproxy as appropriate.\nEOD\n"
            os.system(cmd)
 

    def WriteHtmlReport(self,task_dir):
        """Write HTML status report <task_dir>/<task-name>.html.

        Reports of individual jobs will be created in <task_dir>/<task-name>./<job-name>.html."""

        if not os.path.isdir(task_dir):
            print "Unable to find directory: " + str(task_dir)
            return
        task_name = self.GetName()

        #  Create or wipe directory for job reports if necessary
        job_dir = task_dir + "/" + task_name
        if not os.path.isdir(job_dir):
            try:
                os.mkdir(job_dir,0755)
            except OSError:
                print "Unable to create directory: " + str(job_dir)
                return
        else:
            os.system("cd " + job_dir + "; rm -f *")

        #  Make sure that job statistics are right up to date
        self.RefreshJobStats()
        
        #  Create file for task report and record task data
        task_file = task_dir + "/" + task_name + '.html'
        try:
            f_task = open(task_file, 'w')
        except OSError:
                print "Unable to create file: " + str(task_file)
                return
        f_task.write("<html><body>\n")
        f_task.write("<h1>Status report for Task " + task_name + " produced on " +  GBSTimeStamp() + "</h1>\n<pre>\n")
        f_task.write(str(self))
        f_task.write("\n</pre>\n")

        #  Define colour codes
        red        =  "#ff6666"
        green      =  "#66ff66"
        blue       =  "#66aaff"
        purple     = "#ff44ff"
        grey_light = "#999999"
        grey_dark  = "#777777"

        #  Construct the colour table with links to start points in table sorted by job phase.
        f_task.write("<h2>Task Status Table</h2>\n")
        f_task.write("<p><table border=1>\n")

        f_task.write("<tr><td colspan=2 align=center bgcolor=" + grey_light + ">Not ready HELD (")
        num_nready_held =  self.__jstatNReadyAll - self.__jstatNReadyOther
        if num_nready_held:
            f_task.write('<a href="#phase_%d">%d</a>)\n' % (GID_JPC_NREADY_HOLD,num_nready_held))
        else: f_task.write('0)\n')
        
        f_task.write("<tr><td colspan=2 align=center bgcolor=" + grey_dark + ">Not ready Not HELD (")
        if self.__jstatNReadyOther:
            f_task.write('<a href="#phase_%d">%d</a>)\n' % (GID_JPC_NREADY_NHOLD,self.__jstatNReadyOther))
        else: f_task.write('0)\n')

        f_task.write("<tr><td rowspan=2 align=center bgcolor=" + blue + \
                     ">Ready (" + str(self.__jstatReadyAll) + ")\n")

        f_task.write("<td align=center bgcolor=" + green + ">First attempt (")
        num_ready_nretry = self.__jstatReadyAll - self.__jstatReadyRetry
        if num_ready_nretry:
            f_task.write('<a href="#phase_%d">%d</a>)\n' % (GID_JPC_READY_NRETRY,num_ready_nretry))
        else: f_task.write('0)\n')

        f_task.write("<tr><td align=center bgcolor=" + red + ">Retry (")
        if self.__jstatReadyRetry:
            f_task.write('<a href="#phase_%d">%d</a>)\n' % (GID_JPC_READY_RETRY,self.__jstatReadyRetry))
        else: f_task.write('0)\n')
        
        f_task.write("<tr><td rowspan=2 align=center bgcolor=" + purple + \
                     ">Submitted (" + str(self.__jstatSubmitAll) + ")\n")
        
        f_task.write("<td align=center bgcolor=" + green + ">Running (")
        num_submit_run = self.__jstatSubmitAll - self.__jstatSubmitNRun
        if num_submit_run:
            f_task.write('<a href="#phase_%d">%d</a>)\n' % (GID_JPC_SUBMIT_RUN,num_submit_run))
        else: f_task.write('0)\n')
        
        f_task.write("<tr><td align=center bgcolor=" + red + ">Not running (\n")
        if self.__jstatSubmitNRun:
            f_task.write('<a href="#phase_%d">%d</a>)\n' % (GID_JPC_SUBMIT_NRUN,self.__jstatSubmitNRun))
        else: f_task.write('0)\n')
        
        f_task.write("<tr><td colspan=2 align=center bgcolor=" + green + ">Succeeded (")
        num_done_nfail = self.__jstatDoneAll  - self.__jstatDoneFail 
        if num_done_nfail:
            f_task.write('<a href="#phase_%d">%d</a>)\n' % (GID_JPC_DONE_NFAIL,num_done_nfail))
        else: f_task.write('0)\n')
        
        f_task.write("<tr><td colspan=2 align=center bgcolor=" + red + ">Failed (")
        if self.__jstatDoneFail:
            f_task.write('<a href="#phase_%d">%d</a>)\n' % (GID_JPC_DONE_FAIL,self.__jstatDoneFail))
        else: f_task.write('0)\n')
        
        f_task.write("</table>\n")

        # Prepare a hash of lists: job phase -> list of table rows
        phase_rows = {}
        phase_rows[GID_JPC_NREADY_HOLD]  = []
        phase_rows[GID_JPC_NREADY_NHOLD] = []
        phase_rows[GID_JPC_READY_NRETRY] = []
        phase_rows[GID_JPC_READY_RETRY]  = []
        phase_rows[GID_JPC_SUBMIT_RUN]   = []
        phase_rows[GID_JPC_SUBMIT_NRUN]  = []
        phase_rows[GID_JPC_DONE_NFAIL]   = []
        phase_rows[GID_JPC_DONE_FAIL]    = []
        
        ##  Loop over all jobs, recording summary in a table in the parent task file with
        ##  links to a file holding job details and associate Ganga job (if any).
        f_task.write('<h2><a name="table_by_name">Job Table ordered by Name</a></h2\n')
        f_task.write('<p>See also <a href="#table_by_phase">Job Table ordered by Phase</a><p\n')
        table_header = "<p><table border=1>\n<tr><th>Job Name<th>Status<br>Date&nbsp;Time<th>Ganga<br>Job<th>Try<br>Num<th>Status\n"
        f_task.write(table_header)
        job_names = self._jobManagers.keys()
        job_names.sort()
        for job_name in job_names:
            job = self._jobManagers[job_name]
            job_name = job.GetName()
            
            #  Decide colours for table entry
            pc = job.GetPhaseCode()
            colour_1,colour_2 = grey_dark, grey_dark
            if  pc ==  GID_JPC_NREADY_HOLD:  colour_1,colour_2 = grey_light,grey_light
            if  pc ==  GID_JPC_NREADY_NHOLD: colour_1,colour_2 = grey_dark, grey_dark
            if  pc ==  GID_JPC_READY_NRETRY: colour_1,colour_2 = blue,      green
            if  pc ==  GID_JPC_READY_RETRY:  colour_1,colour_2 = blue,      red
            if  pc ==  GID_JPC_SUBMIT_RUN:   colour_1,colour_2 = purple,    green
            if  pc ==  GID_JPC_SUBMIT_NRUN:  colour_1,colour_2 = purple,    red
            if  pc ==  GID_JPC_DONE_NFAIL:   colour_1,colour_2 = green,     green
            if  pc ==  GID_JPC_DONE_FAIL:    colour_1,colour_2 = red,       red

            # Initialise string that will hold a single row
            table_row = ""

            #  Create job file and link table entry to it.
            job_file = job_dir + "/" + job_name + '.html'
            f_job = open(job_file, 'w')
            f_job.write("<html><body>\n")
            f_job.write("<h1>Status report for Job " + job_name + " produced on " +  GBSTimeStamp() + "</h1>\n")
            f_job.write("<p>This job is part of the <a href=\"../" + task_name + ".html\">")
            f_job.write(task_name + "</a> task<p>\n<pre>\n")
            f_job.write(str(job))
            f_job.write("</pre></body></html>\n")
            f_job.close()
            table_row += "<tr><td bgcolor=" + colour_1 + "><a href=\"" + task_name + "/" \
                         + job_name + ".html\">" + job_name + "</a>"
            
            # Chop off year and second to keep date more compact
            table_row += "<td bgcolor=" + colour_1 + ">" + job.GetStatusTime()[6:-4]
            table_row += "<td bgcolor=" + colour_1 + ">"
            gj_id        = "unknown"
            gj_stat_file = ""
            gj_obj       = job.GetGangaJob()
            if gj_obj:
                gj_id =  gj_obj.id
            else :
                gj_stat_file = job._GetTryOutputDir() + "/gbs_ganga.status"
                if os.path.isfile(gj_stat_file):
                    f = os.popen('grep "id " ' + gj_stat_file)
                    line = f.readline()
                    f.close()
                    mo = re.search(r"id = (\d+)",line)
                    if mo: gj_id = mo.group(1)
                    gj_obj = open(gj_stat_file)
            if not gj_obj:
                table_row += "n/a"
            else:
                
                # Create Ganga job file
                gj_file = job_dir + "/"  + job_name + '_ganga_job.html'
                f_gj = open(gj_file, 'w')
                f_gj.write("<html><body>\n")
                f_gj.write("<p><h1>Ganga job for <a href=\"" + job_name + ".html\">" + job_name + "</a></h1>\n")
                f_gj.write("<p>This job is part of the <a href=\"../" + task_name + ".html>")
                f_gj.write(task_name + "</a> task<p>\n<pre>\n")
                if gj_stat_file:
                    f_gj.write("Dump of expired Ganga Job\n\n")
                    for line in gj_obj: f_gj.write(line)
                    gj_obj.close()
                else:
                    f_gj.write("Dump of active Ganga Job\n\n")
                    f_gj.write(str(gj_obj))
                f_gj.write("</pre></body></html>\n")
                f_gj.close()
                table_row += "<a href=\"" +  task_name + "/" + job_name + "_ganga_job.html\">" +  str(gj_id) + "</a>"

            # Create a GLF history file
            hist_file = job_dir + "/"  + job_name + '_history.html'
            f_hist = open(hist_file, 'w')
            f_hist.write("<html><body>\n")
            f_hist.write("<p><h1>Full try history for <a href=\"" + job_name + ".html\">" + job_name + "</a></h1>\n")
            f_hist.write("This file contains all the GBS Log Files (GLF) for "\
                         + job_name + " which is is part of the <a href=\"../" + task_name + ".html\">")
            f_hist.write(task_name + "</a> task<p>\n<pre>\n")
            glf_wildcard = job.GetStoreLocation('child_dir') + "/*/gbs_*job*.log"
            for glf_file in glob.glob(glf_wildcard):
                mo = re.search(r"_(\d+)\.log$",glf_file)
                f_hist.write("</pre><h2>GLF Summary for %s</h2><pre>\n" % mo.group(1))
                f_glf = open(glf_file)
                f_hist.write(f_glf.read())
                f_glf.close()
            f_hist.write("</pre></body></html>\n")
            f_hist.close()

            hist_file_rel =  task_name + "/" + job_name + "_history.html"
            table_row += "<td bgcolor=%s align=center> <a href=\"%s\">%d</a>" % (colour_2,hist_file_rel,job.GetTryNumber())
            table_row += "<td bgcolor=" + colour_2 + ">" + GIDStringForJSC(job.GetStatusCode()).ljust(23) \
                         + "[" + job.GetStatusText() + "]\n"
            # Record row in table ordered by name and and in hash ready forr table ordered by phase.
            f_task.write(table_row)
            phase_rows[pc].append(table_row)

        f_task.write("</table>\n")

        ##  Construct table ordered by phase
        f_task.write('<h2><a name="table_by_phase">Job Table ordered by Phase</a></h2>\n')
        f_task.write('<p>See also <a href="#table_by_name">Job Table ordered by Name</a><p>\n')
        f_task.write(table_header)
        for pc,rows in phase_rows.iteritems():
            f_task.write('<a name="phase_%d"></a>\n' % pc)
            for row in rows: f_task.write(row)
        f_task.write("</table>\n")
        f_task.write("</body></html>\n")
        f_task.close()
        print "HTML Status report written to " + task_file
        
                

#-----------------------------------------------------------------------------------------------
#
#  ProtoJob Methods (user callable)
#
#-----------------------------------------------------------------------------------------------

    
    def AddProtoJob(self,job_name,args_str="",env_str=""):

        """Add a ProtJob with the name job_name and assign its local args the string args_str
      and environment env_str."""

        if env_str:
            d =  GBSUtilities.ParseEnvStr(env_str)
            if not d:
                print "Cannot parse environment string: " + env_str
                return None
        return self._AddJobOrProtoJob(job_name,args_str,env_str,"GBSProtoJob")

    def PromoteProtoJobs(self,job_name_pattern = ".*"):

        """Promote all ProtoJobs whose name matches the supplied pattern. For example ".*" promotes all."""

        pro_list = [] # List of names to be promoted (give user a last chance to abort)
        for job_name,job in self._jobManagers.iteritems():
            if job.GetType() != "GBSProtoJob": continue
            if re.search(job_name_pattern,job_name): pro_list.append(job_name)
        num_pjobs = len(pro_list)
        if not num_pjobs:
            print "No ProtoJobs to promote"
            return
        ans = raw_input(str(num_pjobs) + " ProtoJobs are due for promotion.  Do you want to proceed ?y[n]")
        if not re.search(r"^y",ans,re.I):
            print "Promotion aborted"
            return
        for job_name in pro_list:
            pjob = self._jobManagers[job_name]
            job  = GetModelRegistry().CreateObject(self.GetModel(),"Job",job_name,self)
            job.SetScriptLocalArgs(pjob.GetScriptLocalArgs())
            job.SetLocalEnvironment(pjob.GetLocalEnvironment())
            job.SetLocalInputSandbox(pjob.GetLocalInputSandbox())
            job.SetLocalOutputSandbox(pjob.GetLocalOutputSandbox())
            self._jobManagers[job_name] = job
        print str(len(pro_list)) + " ProtoJobs promoted"
       
    def RemoveProtoJobs(self,job_name_pattern = ".*"):

        """Remove all ProtoJobs whose name matches the supplied pattern. For example ".*" removes all"""

        del_list = [] # List of names to be deleted (cannot delete elements while iterating)
        for job_name,job in self._jobManagers.iteritems():
            if job.GetType() != "GBSProtoJob": continue
            if re.search(job_name_pattern,job_name): del_list.append(job_name)
        for job_name in del_list: del self._jobManagers[job_name]
        print str(len(del_list)) + " ProtoJobs deleted"
       
                

#-----------------------------------------------------------------------------------------------
#
#  Private Methods (not user callable)
#
#-----------------------------------------------------------------------------------------------

                      
    def _AddJobOrProtoJob(self,job_name,args_str,env_str,type):

        """Create a new named Job or ProtoJob and optionally assign its application script local args."""

        #  Check that the name is legal and not already in use.
        if re.search(r"[^a-zA-Z0-9_\-]",job_name):
            print "Sorry, '" + str(job_name) + "' is an illegal job name (characters other than alphanumeric, '_' and '-')"
            return None 
        if not re.search(r"^job_",job_name): job_name = 'job_' + job_name
        if self._jobManagers.has_key(job_name):
            print "Sorry, there already is a Job named '" + str(job_name) + "'"
            return None

        # Name O.K., go create the Job.
        job = None
        if ( type == "GBSProtoJob" ): job = GBSProtoJob(job_name)
        else:                         job = GetModelRegistry().CreateObject(self.GetModel(),"Job",job_name,self)
        if args_str: job.SetScriptLocalArgs(args_str)
        if env_str:  job.SetLocalEnvironment(env_str)
        self._jobManagers[job_name] = job
        return job

    def __ApplyActionToJobs(self,action,job_name_pattern):

        """Apply an action (CLEAR ERROR COUNTS, CLEAR HISTORY, HOLD, KILL, RELEASE) to all jobs matching pattern"""
        job_list = [] # List of job names to which action can be applied
        for job_name,job in self._jobManagers.iteritems():
            if not re.search(job_name_pattern,job_name): continue
            if    ( action == "CLEAR ERROR COUNTS" and job.CanClear() )  \
               or ( action == "CLEAR HISTORY"      and job.CanClear() )  \
               or ( action == "HOLD"               and job.IsReady() )   \
               or ( action == "KILL"               and job.CanKill() )   \
               or ( action == "RELEASE"            and job.IsHeld() )    \
               or ( action == "REMOVE"             and job.CanClear() ):
                job_list.append(job_name)
        num_pjobs = len(job_list)
        if not num_pjobs:
            print "No jobs suitable for action " + action
            return
        print str(num_pjobs) + " jobs suitable for action " + action + ":-"
        for job_name in job_list: print "  " + job_name
        ans = raw_input("Do you want to proceed ?y[n]")
        if not re.search(r"^y",ans,re.I):
            print "Action " + action + " aborted"
            return
        if action == "CLEAR HISTORY" or action == "REMOVE":
            warn = "this wipes all output for the job(s)"
            if action == "REMOVE": warn = "this entirely removes the job(s)"
            ans = raw_input("Are you REALLY sure - " + warn + "! ?y[n]")
            if not re.search(r"^y",ans,re.I):
                print "Action " + action + " aborted"
                return
        for job_name in job_list:
            job = self._jobManagers[job_name]
            if action == "CLEAR ERROR COUNTS": job.ClearErrorCounts()
            if action == "CLEAR HISTORY":      job.ClearHistory(False)
            if action == "HOLD":               job.Hold()
            if action == "KILL":               job.Kill()
            if action == "RELEASE":            job.Release()
            if action == "REMOVE":             job.Remove(False)
        print "Action " + action + " applied to " + str(len(job_list)) + " jobs"


    def __ReloadChildren(self):

        """Reload child Jobs from disk.

        Loop over all job_*.state files and recreate. 
        Make sure any associated Ganga job is in the associated JobTree directory"""

        import os, re

        child_dir         = self.GetStoreLocation("child_dir")
        child_state_files = os.listdir(child_dir)
        for child_state_file in child_state_files:

            #  Only look at job_*.state files
            child_state_file_spec = child_dir + "/" + child_state_file
            if not os.path.isfile(child_state_file_spec): continue
            mo = re.search(r"^(job_.*)\.state$",child_state_file)
            if not mo: continue
            child_name = mo.group(1)
            job = self.AddJob(child_name)
            # Ignore errors while trying to move associated Ganga job into correct JobTree directory
            # There may not be any Ganga job
            try:
                g_job = job.GetGangaJob()
                Ganga.GPI.jobtree.add(g_job,self.__gangaTreeDir)
            except: pass
            

class GBSProtoJob:

    """A potential job made actual through the Task.PromoteProtoJob() method.

    Protojobs allow users to create large numbers of jobs
    automatically using scripts or data files and review and correct
    them before "promoting" them into jobs.  As such they are easy
    to create and destroy, are not persisted and are consequently lost at
    program termination, unless promoted.

    GBSProtoJob emulate the following small subset of a real job:-

    GetType()
      Return object type.

    GetScriptLocalArgs()
      Return (as a string) the list of application script args that are local to this job

    SetScriptLocalArgs(args_str)
      Set (as a string) the list of application script args that are local to this job

      e.g.  job.SetScriptLocalArgs('123 "a string with spaces" 456')

    """

    def __init__(self,name):
        self.__name               = name
        self.__scriptLocalArgs    = ""
        self.__localEnvironment   = ""
        self.__localInputSandbox  = ""
        self.__localOutputSandbox = ""
        

    def GetType(self): return "GBSProtoJob"

    def __repr__(self): return self.AsString()
    
    
    def AsString(self,level = "Brief"):

        """Return string description.

         Return string description at the following levels:-
        "Brief"    one line summary suitable for rows in tables
        "Heading"  one line summary suitable as heading for "Brief" """

        if ( level == "Heading"):
            s  = "Name".ljust(20)
            s += "Status".ljust(23)
            s += "Input".ljust(40)
            s += "Status Details"
            return s
        s  = self.__name.ljust(20)
        s += "(protojob)".ljust(23)
        s += (self.__scriptLocalArgs + ";" + self.__localEnvironment).ljust(40)
        return s

    def GetLocalEnvironment(self):
        """Return, as a comma separated list string, the environment that is local to this job"""
        return self.__localEnvironment 

    def GetScriptLocalArgs(self):
        """Return (as a string) the list of application script args that are local to this job"""
        return self.__scriptLocalArgs
    
    def GetLocalInputSandbox(self):
        """Return, as a comma separated list string, the input sandbox file list that is local to this job."""
        return self.__localInputSandbox
    
    def GetLocalOutputSandbox(self):
        """Return, as a comma separated list string, the ouput sandbox file list that is local to this job."""
        return self.__localOutputSandbox

    def SetLocalEnvironment(self,env_str):
        """Set, as a comma separated list string, the environment that local to this job.."""
        self.__localEnvironment = env_str

    def SetScriptLocalArgs(self,arg_str):
        """Set (as a string) the comma separated list of application script args that are local to this job."""
        self.__scriptLocalArgs  = arg_str

    def SetLocalInputSandbox(self,in_sbox_str):
        """Set, as a comma separated list string, the input sandbox file list that is local to this job"""
        (ok,str) = GBSUtilities.ProcessSandboxSetup(self,in_sbox_str,"input")
        if ok:
            # Store the raw input sandbox file list; files won't get stored until promotion.
            self.__localInputSandbox = in_sbox_str

    def SetLocalOutputSandbox(self,out_sbox_str):
        """Set, as a comma separated list string, the output sandbox file list that is local to this job."""

        (ok,str) = GBSUtilities.ProcessSandboxSetup(self,out_sbox_str,"output")
        if ok:
            self.__localOutputSandbox = str

---