rcActivitiesInterface.py

Go to the documentation of this file.

#!/usr/local/bin/python
#
#                               Copyright 2005
#                                     by
#                        The Board of Trustees of the
#                     Leland Stanford Junior University.
#                            All rights reserved.
#

__facility__ = "Online"
__abstract__ = "Interface to the Run activities table"
__author__   = "S. Tuvi <stuvi@SLAC.Stanford.edu> SLAC - GLAST LAT I&T/Online"
__date__     = "2005/07/23 00:08:27"
__updated__  = "$Date: 2006/04/21 23:39:01 $"
__version__  = "$Revision: 1.17 $"
__release__  = "$Name: HEAD $"
__credits__  = "SLAC"

try:
  import LICOS.copyright_SLAC
except:
  pass

import MySQLdb
import logging as log
import os
import time

try:
  from LICOS.lib.LATconstants import *
except:
  ONLINE_ROOT = os.environ['ONLINE_ROOT']

class ActivitiesInterface(object):
  __stdDbName = "elogbook"
  __stdDbUser = "elogbook"
  __stdDbPass = "elogbook"
  __stdDbPort = 3306

  ## Allowed tables
  TableStates     = "LICOS_states"
  TableActivities = "LICOS_activities"
  TableReport     = "elogreport"

  ## State categories
  CommandingState = "CommandingState_FK"
  DataState       = "DataState_FK"
  AnalysisState   = "AnalysisState_FK"

  ## Time rows
  DataStart       = "DataStart"
  DataStop        = "DataStop"

  ## Suite id
  SuiteIdentifier = "SuiteIdentifier"

  ## LDF file name
  LDFname         = "LdfFileName"

  ## script Output
  ScriptOutput    = "ScriptOutput"

  ## Job complete list
  JobComplete     = "JobCompleteList"

  ## Primary key
  PrimaryKey      = "RunID_FK"

  # Definition of standard processing states in the database
  # used when caching the foreign keys to reduce overhead
  __states    = [ "ActivityCreated",
                  "CommandingStarted",
                  "CommandingComplete",
                  "NoData",
                  "DataStarted",
                  "DataComplete",
                  "NoAnalysis",
                  "AnalysisStarted",
                  "AnalysisComplete", ]

  """!\brief Interface to the activities database
  """
  def __init__(self,
               dbName=__stdDbName,
               dbUser=__stdDbUser,
               dbPass=__stdDbPass,
               dbPort=__stdDbPort):

    self.__db     = None
    self.__dbName = dbName
    self.__dbUser = dbUser
    self.__dbPass = dbPass
    self.__dbPort = dbPort
    self.__stateKeys = {}

    self.initialize()

    pass

  def __del__(self):
    """ Reimplementation of the destructor to make bloody well sure
        that the table ends up in an unlocked state...

    """
    self.unlockRow()
    self.__db = None
    # no need to call super's __del__ as there is no super

  def initialize(self):
    """!\brief Initialize the object.  Do a database connect, etc.
    """
    self.connect()

    # Seed the definitions of states and make them attributes of the class
    for stateName in self.__states:
      constraint = "stateName = '%s'" % (stateName)
      tmp = self.__pull(self.__selectSQL( table=self.TableStates, column = "state_PK", constraint=constraint))
      #print tmp[0], stateName
      setattr(self, stateName, tmp[0])
      self.__stateKeys[tmp[0]] = stateName
    #print dir(self)

  def connect(self):
    """!\brief connect to the database
    """
    if "ELOGBOOK_HOST" in os.environ:
      host = os.environ["ELOGBOOK_HOST"]
    else:
      host = "localhost"

    if 'ELOGBOOK_DB' in os.environ:
      self.__dbName = os.environ['ELOGBOOK_DB']

    try:
      self.__db = MySQLdb.connect(host=host,
                                db     = self.__dbName,
                                user   = self.__dbUser,
                                passwd = self.__dbPass,
                                port   = self.__dbPort)
    except MySQLdb.OperationalError, e:
      msg  = "Error encountered when attempting to connect to MySQL server on host: '%s'\n" % host
      msg += "  Actual exception was: %s" % e
      log.exception(MySQLdb.OperationalError(msg))
      raise

  def newActivity(self, runID, suiteID=None):
    """! Create a new activity in the activities database

    \param runID   run identifier
    \param suiteID Suite linkage identifier  Note:  no check is made to see whether the suite exists.
    """
    if suiteID is None:
      suiteID = "Null"
    #try:
    #  runID_fk = self.getRunID_fk(runID)
    #except Exception, e:
    #  raise RuntimeError("Cannot define activity for runID %s.  Received an exception during query: %s" % (runID, e) )
    #if runID_fk is None:
    #  raise RuntimeError("Cannot define activity for runID %s.  %s not found in table %s" % (runID, runID, self.TableReport))

    runID_fk = runID

    sql  = "insert into %s " % self.TableActivities
    sql += "(RunID_FK, SuiteIdentifier, CommandingState_FK, DataState_FK, AnalysisState_FK) values "
    sql += "(%s, %s, %s, %s, %s)" % \
           (runID_fk, suiteID, self.ActivityCreated, self.NoData, self.NoAnalysis)
    try:
      self.__db.cursor().execute(sql)
      self.__db.commit()
    except Exception, e:
      print "BadQuery: \n  ", sql
      raise

  def setState(self, runID, category, newState):
    """! Set the state of a run in the database

    \param runID     integer Run ID
    \param category  One of the possible state categories
    \param newState  New state
    """
    try:
      runID_fk = self.getRunID_fk(runID)
    except RuntimeError, e:
      raise RuntimeError("Cannot set state for runID %s.  Received an exception during query: %s" % (runID, e) )

    con = "runID_fk = %s" % runID_fk
    sql = self.__updateSQL(table=self.TableActivities,
                           column=category,
                           value=newState,
                           constraint=con)
    self.__push(sql)

  def getState(self, runID, category):
    """!\brief Retrieve the state of a run from the database

    \param runID     integer Run ID
    \param category  One of the possible state categories
    """
    try:
      runID_fk = self.getRunID_fk(runID)
    except RuntimeError, e:
      raise RuntimeError("Cannot retreive state for runID %s.  Received an exception during query: %s" % (runID, e) )

    con = "runID_fk = %s" % runID_fk
    sql = self.__selectSQL( table = self.TableActivities,
                            column = category,
                            constraint = con)
    return self.__pull(sql)[0]

  def getRunsByState(self, category, state):
    """!\brief Get a set of runIDs which are in a particular state

    \param category  One of the State categories
    \param state     One of the possible states
    \return          A collection of runIDs which are in the state requested.
    """
    column = "runID_fk"
    con = "%s = %s" % (category,state)
    sql = self.__selectSQL( table = self.TableActivities,
                            column = column,
                            constraint = con)
    return self.__pull(sql)

  def setTime(self, runID, timeColumn, theTime):
    """!\brief Set a time tag row

    \param runID        integer Run ID
    \param timeColumn   Tag for the column.  e.g., DataStart, DataStop
    \param theTime      Unix timestamp
    """
    try:
      runID_fk = self.getRunID_fk(runID)
    except RuntimeError, e:
      raise RuntimeError("Cannot retreive state for runID %s.  Received an exception during query: %s" % (runID, e) )
    con = "runID_fk = %s" % runID_fk

    timeStr = "FROM_UNIXTIME(%s)" % theTime

    sql = self.__updateSQL( table = self.TableActivities,
                            column = timeColumn,
                            value = timeStr,
                            constraint = con)
    self.__push(sql)

  def getTime(self, runID, timeColumn):

    try:
      runID_fk = self.getRunID_fk(runID)
    except RuntimeError, e:
      raise RuntimeError("Cannot retreive state for runID %s.  Received an exception during query: %s" % (runID, e) )

    con = "runID_fk = %s" % runID_fk

    unixTimeColumn = "UNIX_TIMESTAMP(%s)" % timeColumn
    sql = self.__selectSQL( table = self.TableActivities,
                            column = unixTimeColumn,
                            constraint = con)
    rTime =  self.__pull(sql)[0]
    return rTime



  def setLdfFileName(self, runID, ldfFileName):
    """!\brief set the output LDF/LSF format data file name for a run

    \param runID         integer Run ID
    \param ldfFileName   Fully qualified path name.  Variables will be expanded before insertion.

    """
    try:
      runID_fk = self.getRunID_fk(runID)
    except RuntimeError, e:
      raise RuntimeError("Cannot set LDF file name for runID %s.  Received an exception during query: %s" % (runID, e) )

    fnStr = "\'" + os.path.expandvars(ldfFileName) + "\'"
    con = "runID_fk = %s" % runID_fk
    sql = self.__updateSQL( table = self.TableActivities,
                            column = self.LDFname,
                            value  = fnStr,
                            constraint = con)
    self.__push(sql)

  def getLdfFileName(self, runID):
    """!\brief Retrieve the full path to the LDF file from the db

    \param runID    Integer Run ID
    """
    try:
      runID_fk = self.getRunID_fk(runID)
    except RuntimeError, e:
      raise RuntimeError("Cannot get LDF file name for runID %s.  Received an exception during query: %s" % (runID, e) )

    con = "runID_fk = %s" % runID_fk
    sql = self.__selectSQL( table = self.TableActivities,
                            column = self.LDFname,
                            constraint = con)
    return self.__pull(sql)[0]

  def setScriptOutput(self, runID, blob):
    """!\brief Set the script output blob

    \param runID   Integer run ID
    \param blob    An arbitrary byte stream representation of the script output
    """
    try:
      runID_fk = self.getRunID_fk(runID)
    except RuntimeError, e:
      raise RuntimeError("Cannot set script output for runID %s.  Received an exception during query: %s" % (runID, e) )

    con = "runID_fk = %s" % runID_fk
    blobStr = "\"" + blob + "\""
    sql = self.__updateSQL( table = self.TableActivities,
                            column = self.ScriptOutput,
                            value  = blobStr,
                            constraint = con)
    self.__push(sql)


  def getScriptOutput(self, runID):
    """!\brief Retrieve the script output

    \param runID   Integer run ID
    \return blob  an arbitrary byte stream representation of the script output
    """
    try:
      runID_fk = self.getRunID_fk(runID)
    except RuntimeError, e:
      raise RuntimeError("Cannot get script output for runID %s.  Received an exception during query: %s" % (runID, e) )

    con = "runID_fk = %s" % runID_fk
    sql = self.__selectSQL( table = self.TableActivities,
                            column = self.ScriptOutput,
                            constraint = con)
    return self.__pullBlob(sql)[0]

  def setJobCompleteList(self, runID, jobListString):
    """!\brief Set the job complete list

    \param runID   Integer run ID
    \param jobListString  String representing the list of completed jobs
    """
    try:
      runID_fk = self.getRunID_fk(runID)
    except RuntimeError, e:
      raise RuntimeError("Cannot set jobCompleteList for runID %s.  Received an exception during query: %s" % (runID, e) )

    con = "runID_fk = %s" % runID_fk
    blobStr = "\"" + jobListString + "\""
    sql = self.__updateSQL( table = self.TableActivities,
                            column = self.JobComplete,
                            value  = blobStr,
                            constraint = con)
    self.__push(sql)

  def getJobCompleteList(self, runID):
    """!\brief Retrieve the job complete list

    \param runID   Integer run ID
    \return a string (possibly empty) denoting which jobs are complete.
    """
    try:
      runID_fk = self.getRunID_fk(runID)
    except RuntimeError, e:
      raise RuntimeError("Cannot get jobCompleteList for runID %s.  Received an exception during query: %s" % (runID, e) )

    con = "runID_fk = %s" % runID_fk
    sql = self.__selectSQL( table = self.TableActivities,
                            column = self.JobComplete,
                            constraint = con)
    return self.__pullBlob(sql)[0]

  def lockRow(self, runID_fk):

    con = "runID_fk = %s" % runID_fk
    # The following doesn't work for locking a row because the next
    # query/whatever will unlock.  I thought autocommit=False would
    # prevent this, but no.  So... Lock it down!!!
    #sql = "select * from %s where %s LOCK IN SHARE MODE" %\
    #      ( self.TableActivities, con)
    sql = "lock tables %s write" % ( self.TableActivities )
    self.__db.cursor().execute(sql)



  def unlockRow(self):
    #sql = "commit"
    sql = "unlock tables"
    self.__db.cursor().execute(sql)
    pass

  def stateStr(self, state):
    """! Convert an enumerated state to a string for printing
    \param state  an integer state

    \return  State string
    """
    if state in self.__stateKeys:
      return self.__stateKeys[state]
    else:
      return ""

  def getRunID_fk(self, runID):
    """!Retreive the runID_fk from the database for speed

    \param runID     integer Run ID
    \return the FK (or None if run is nonexistent)
    """
    if runID is None:
      raise RuntimeError("Cannot have a null run identifier.  Input runID = %s" % runID )


    sql = self.__selectSQL( table = self.TableActivities,
                                column = "runid_fk",
                                constraint= "runid_fk=%s" % runID)
    fk = self.__pull(sql)[0]

    return fk

  def __selectSQL(self, table, column, constraint):
    return "select %s from %s where %s" % (column, table, constraint)

  def __updateSQL(self, table, column, value, constraint):
    sql  = "update %s " % table
    sql += "set %s = %s " % (column, value)
    sql += "where %s" % constraint
    return sql

  def __pull(self, sql):
    # JHP:  I have *no* idea why I have to commit *before*
    #    I pull from the db, but without it, getRunID_fk won't return a
    #    recently created runID.
    self.__db.commit()
    cur = self.__db.cursor()
    try:
      cur.execute(sql)
    except Exception, e:
      print "Bad query: \n  ", sql
      raise e
    # LHM change to fetchall
    fetch = cur.fetchall()
    if fetch is None:
      fetch = (None,)
    returnFetch = []
    for item in fetch:
      (column,) = item
      returnFetch.append(column)
    return returnFetch

  def __pullBlob(self, sql):
    fetched = self.__pull(sql)
    try:
      r = tuple([ item.tostring() for item in fetched ])
    except:
      r = fetched
    return r

  def __push(self, sql):
    try:
      self.__db.cursor().execute(sql)
    except Exception, e:
      print "Bad query: \n  ", sql
      raise e
    self.__db.commit()




if __name__ == '__main__':
  # regression test for activities database.
  #   Also a short usage and syntax guide.


  import sys
  try:
    db = ActivitiesInterface()
  except Exception, e:
    log.exception(e)
    sys.exit()

  runID       = 999000999  # Can also use '999000999'
  suiteID     = 999      # Can also use '000999'
  LDFname     = os.path.join(ONLINE_ROOT, "Work/scratch/foo.ldf")

  # Delete this runID from the database so we can test it
  try:
    sql = "delete from %s where runID_fk = %s" % (db.TableActivities, runID)
    db._ActivitiesInterface__db.cursor().execute(sql)
  except Exception, e:
    # ignore exceptions
    pass

  try:
    print "attempting to create run id %s in elogbook.%s..." %\
          ( runID, db.TableActivities )
    db.newActivity(runID, suiteID)
    print "...Created it\n"
  except:
    raise

  try:
    print "attempting to retreive run id %s from elogbook.%s..." %\
          ( runID, db.TableReport )
    runID_fk = db.getRunID_fk(runID)
    print "...got it: %s\n" % runID_fk
  except Exception, e:
    raise

  try:
    print "attempting to set the commanding state to CommandingStarted..."
    db.setState(runID, db.CommandingState, db.CommandingStarted)
    print "...and read it back..."
    state = db.getState(runID, db.CommandingState)
    print "...read: state: %d == %s\n" %\
          ( state, db.stateStr(state))
  except:
    raise

  try:
    print "attempting to set the timestamp to now (%s)..." % time.time()
    db.setTime(runID, db.DataStart, time.time())
    print "...and read it back..."
    t = db.getTime(runID, db.DataStart)
    print "...read: time=%s\n" % t
  except:
    raise

  try:
    print "attempting to set the LDFfile name to %s" %\
          ( LDFname )
    db.setLdfFileName(runID, LDFname)
    print "...and read it back, noticing variables were expanded..."
    readStr = db.getLdfFileName(runID)
    print "...LDFfile = %s\n" % readStr
  except:
    raise

  try:
    print "attempting to set script output to range(10)..."
    db.setScriptOutput(runID, str(range(10)))
    print "...and read it back: %s" %\
          (db.getScriptOutput(runID))
    q = db.getScriptOutput(runID)
    k = eval(q)
    print "...and eval it:", k, "is of type", type(k), "\n"
  except:
    raise

  try:
    print "attempting to set the jobComplete list to something with double quotes..."
    tmp = [ " a b c" , "d e f", 'zqxkjshdfadsy' ]

    db.setJobCompleteList(runID, str( tmp ) )
    print "...and read it back: %s" %\
          (db.getJobCompleteList(runID))
    q = db.getJobCompleteList(runID)
    k = eval(q)
    print "...and eval it:", k, "is of type", type(k), "\n"
  except:
    raise

  try:
    import copy
    print "attempting to set the jobComplete list to db.__dict__..."
    z = copy.deepcopy(db.__dict__)
    # gotta remove an object (<mysql...>) from the dict, or it won't stringify
    # reversibly
    z['_ActivitiesInterface__db'] = None
    db.setJobCompleteList(runID, str( z ) )
    print "...and read it back: %s" %\
          (db.getJobCompleteList(runID))
    q = db.getJobCompleteList(runID)
    k = eval(q)
    print "...and eval it:", k, "is of type", type(k), "\n"
  except:
    raise

  try:
    db2 = ActivitiesInterface()

    import threading
    t = threading.Timer(5, db.unlockRow)
    t.start()
    db.lockRow(runID)

    print "attempting to test a lock on runID %s..." % runID
    start = time.time()
    print "...by setting the LDF file name to 'foo' with a second database connection"
    db2.setLdfFileName(runID, "foo")
    end = time.time()

    print "...table locked for %.1f seconds\n" % (end - start)
  except Exception, e:
    db.unlockRow()
    raise

  try:
    print "attempting to update a row from the current db connection during a lock..."
    db.lockRow(runID)
    db.setLdfFileName(runID, LDFname)
    db.unlockRow()
    print "...Read back LDFfile name %s\n" % db.getLdfFileName(runID)
  except:
    db.unlockRow()
    raise

---