Changeset 33415 for branches/meh_branches/ppstack_test/ippToPsps/jython/ipptopspsdb.py

Timestamp:

Mar 5, 2012, 5:19:48 PM (14 years ago)

Author:

mhuber

Message:

merging latest r33407 trunk changes to meh_branches/ppstack_test

Location:

branches/meh_branches/ppstack_test

Files:

• . (modified) (1 prop)
• ippToPsps/jython/ipptopspsdb.py (modified) (14 diffs)

branches/meh_branches/ppstack_test/ippToPsps/jython/ipptopspsdb.py

@@ -17 +17 @@
     Constructor
     '''
-    def __init__(self, logger, doc):
-        super(IppToPspsDb, self).__init__(logger, doc, "ipptopspsdatabase")
-
-    '''
-    Creates a new batch
-    '''
-    def createNewBatch(self, batchType, stageID, survey, dvoDb, datastoreProduct):
-
-        sql = "INSERT INTO batch ( \
-               batch_type, \
-               stage_id, \
-               survey, \
-               dvo_db, \
-               datastore_product \
-               ) VALUES ( \
-               '" + batchType + "', \
-               " + str(stageID) + ", \
-               '" + survey + "', \
-               '" + dvoDb + "', \
-               '" + datastoreProduct + "' \
-               )"
-
-        self.execute(sql)
-
-        sql = "SELECT MAX(batch_id) FROM batch"
-
-        batchID = -1;
-
-        try:
-            rs = self.executeQuery(sql)
-            rs.first()
-            batchID = rs.getInt(1)
-        except:
-            self.logger.exception("Unable to get batch ID")
-
-        self.logger.debug("Created new batch in ippToPsps database with batchID = %d" % batchID)
-
-        return batchID;
-
-    '''
-    Returns a list of processed batch IDs that are merged but not yet deleted
-    '''
-    def getMergedButNotDeletedBatchIDs(self, epoch, dvoGpc1Label):
+    def __init__(self, logger, config):
+        super(IppToPspsDb, self).__init__(logger, config, "ipptopspsdatabase")
+
+        self.MAX_FAILS = 5
+
+    '''
+    Returns a list of merged batch IDs that are merged but not yet deleted
+    '''
+    def getMergedButNotDeletedBatchIDs(self, batchType, epoch, dvoGpc1Label, column):
 
         sql = "SELECT DISTINCT batch_id \
                FROM batch \
                WHERE timestamp > '" + epoch + "' \
-               AND dvo_db = '" + dvoGpc1Label + "' \
-               AND merged \
-               AND !deleted"
+               AND batch_type = '" + batchType + "' \
+               AND dvo_db = '" + dvoGpc1Label + "' \
+               AND merged = 1 \
+               AND " + column + " = 0"
+
+        ids = []
+        try:
+            rs = self.executeQuery(sql)
+            while (rs.next()): ids.append(rs.getInt(1))
+            rs.close()
+        except:
+            self.logger.exception("Can't query for merged batch ids in ipptopsps Db")
+
+        self.logger.debug("Found %d merged but un-deleted items" % len(ids))
+
+        return ids
+
+    '''
+    Returns a list of batch IDs marked as 'purged' but not yet deleted
+    '''
+    def getPurgedButNotDeletedBatchIDs(self, batchType, epoch, dvoGpc1Label, column):
+
+        sql = "SELECT DISTINCT batch_id \
+               FROM batch \
+               WHERE timestamp > '" + epoch + "' \
+               AND batch_type = '" + batchType + "' \
+               AND dvo_db = '" + dvoGpc1Label + "' \
+               AND purged = 1 \
+               AND " + column + " = 0"
+
+        ids = []
+        try:
+            rs = self.executeQuery(sql)
+            while (rs.next()): ids.append(rs.getInt(1))
+            rs.close()
+        except:
+            self.logger.exception("Can't query for merged batch ids in ipptopsps Db")
+
+
+        self.logger.debug("Found %d merged but un-deleted items" % len(ids))
+
+        return ids
+
+    '''
+    Returns a list of processed batch IDs that have been loaded to the ODM but not yet deleted
+    '''
+    def getLoadedToODMButNotDeletedBatchIDs(self, batchType, epoch, dvoGpc1Label, column):
+
+        sql = "SELECT DISTINCT batch_id \
+               FROM batch \
+               WHERE timestamp > '" + epoch + "' \
+               AND batch_type = '" + batchType + "' \
+               AND dvo_db = '" + dvoGpc1Label + "' \
+               AND (loaded_to_ODM = -1 OR merge_worthy = 1) \
+               AND " + column + " = 0"
+
+        ids = []
+        try:
+            rs = self.executeQuery(sql)
+            while (rs.next()): ids.append(rs.getInt(1))
+            rs.close()
+        except:
+            self.logger.exception("Can't query for merged batch ids in ipptopsps Db")
+
+        self.logger.debug("Found %d merged but un-deleted items" % len(ids))
+
+        return ids
+
+    '''
+    Returns a list of merged batch IDs that are not deleted from local disk
+    '''
+    def getMergedButNotDeletedFromLocalDisk(self, batchType, epoch, dvoGpc1Label):
+        return self.getMergedButNotDeletedBatchIDs(batchType, epoch, dvoGpc1Label, "deleted_local")
+
+    '''
+    Returns a list of processed batch IDs that are loaded to the ODM, but not deleted from local disk
+    '''
+    def getLoadedToODMButNotDeletedFromLocalDisk(self, batchType, epoch, dvoGpc1Label):
+        return self.getLoadedToODMButNotDeletedBatchIDs(batchType, epoch, dvoGpc1Label, "deleted_local")
+
+    '''
+    Returns a list of processed batch IDs that are loaded to the ODM, but not deleted from datastore
+    '''
+    def getLoadedToODMButNotDeletedFromDatastore(self, batchType, epoch, dvoGpc1Label):
+        return self.getLoadedToODMButNotDeletedBatchIDs(batchType, epoch, dvoGpc1Label, "deleted_datastore")
+
+    '''
+    Returns a list of processed batch IDs that are loaded to the ODM, but not deleted from DXLayer
+    '''
+    def getLoadedToODMButNotDeletedFromDXLayer(self, batchType, epoch, dvoGpc1Label):
+        return self.getLoadedToODMButNotDeletedBatchIDs(batchType, epoch, dvoGpc1Label, "deleted_dxlayer")
+
+    '''
+    Returns a list of purged  batch IDs that not deleted from local disk
+    '''
+    def getPurgedButNotDeletedFromLocalDisk(self, batchType, epoch, dvoGpc1Label):
+        return self.getPurgedButNotDeletedBatchIDs(batchType, epoch, dvoGpc1Label, "deleted_local")
+
+    '''
+    Returns a list of purged batch IDs that are not deleted from datastore
+    '''
+    def getPurgedButNotDeletedFromDatastore(self, batchType, epoch, dvoGpc1Label):
+        return self.getPurgedButNotDeletedBatchIDs(batchType, epoch, dvoGpc1Label, "deleted_datastore")
+
+
+    '''
+    Returns a list of processed batch IDs that are not loaded to the datastore
+    NB we have a date constraint here because, before this time, it is possible that
+    a test batch would be labelled as processed but not loaded to datastore, and we would NOT
+    want to load a test batch to PSPS
+
+    We also check that these batches are more than 4 HOURs old, so that we don't attempt to 
+    publish something simulataneously with the client that it producing it
+    '''
+    def getProcessedButFailedDatastoreBatchIDs(self, epoch, dvoGpc1Label, batchType):
+
+        sql = "SELECT DISTINCT batch_id \
+               FROM batch \
+               WHERE timestamp > '" + epoch + "' \
+               AND batch_type = '" + batchType + "' \
+               AND dvo_db = '" + dvoGpc1Label + "' \
+               AND processed = 1\
+               AND loaded_to_datastore != 1 \
+               AND timestamp > '2011-10-27' \
+               AND timestamp < now() -  INTERVAL 4 HOUR"
 
         ids = []
@@ -74 +162 @@
                 ids.append(rs.getInt(1))
         except:
-            self.logger.exception("Can't query for merged batch ids in ipptopsps Db")
+            self.logger.exception("Can't query for processed batch ids not loaded to datastore ipptopsps Db")
 
         rs.close()
 
-        self.logger.debug("Found %d merged but un-deleted items" % len(ids))
-
-        return ids
-
-    '''
-    Returns a list of processed batch IDs that are not yet merged for this epoch and dvo label
-    '''
-    def getProcessedButUnmergedBatchIDs(self, epoch, dvoGpc1Label):
-
-        sql = "SELECT DISTINCT batch_id \
+        self.logger.debug("Found %d processed but-not-on-datastore items" % len(ids))
+
+        return ids
+
+    '''
+    Returns a list of processed batch IDs that are not yet loaded to the ODM
+    NOTE we exclude batches already deleted from the datastore, as sometimes we delete stuff prior to loading to ODM if there is a known problem with the batch
+    '''
+    def getUnloadedBatchIDs(self, epoch, dvoGpc1Label, batchType):
+
+        sql = "SELECT batch_id \
                FROM batch \
                WHERE timestamp > '" + epoch + "' \
                AND dvo_db = '" + dvoGpc1Label + "' \
-               AND loaded_to_datastore \
-               AND !merged"
+               AND loaded_to_datastore = 1 \
+               AND batch_type = '" + batchType + "' \
+               AND loaded_to_ODM = 0 \
+               AND deleted_datastore = 0"
 
         ids = []
@@ -100 +191 @@
                 ids.append(rs.getInt(1))
         except:
-            self.logger.exception("Can't query for processed batch ids in ipptopsps Db")
+            self.logger.exception("Can't query for unloaded batch ids in ipptopsps Db")
 
         rs.close()
 
-        self.logger.debug("Found %d processed and un-merged items" % len(ids))
-
-        return ids
-
-    '''
-    Returns a list of IDs for this batch type, epoch and dvo label (i.e. either 
-            cam_ids or stack_ids) for which the spefified column is true (=1)
-    '''
-    def getStageIDs(self, batchType, epoch, dvoGpc1Label, column):
-
-        sql = "SELECT DISTINCT stage_id \
+        self.logger.debug("Found %d unloaded items" % len(ids))
+
+        return ids
+
+    '''
+    Returns a list of processed batch IDs that are not yet merge_worthy and that have not failed to load
+    NOTE we exclude batches already deleted from the datastore, as sometimes we delete stuff prior to loading to ODM if there is a known problem with the batch
+    '''
+    def getUnmergeWorthyBatchIDs(self, epoch, dvoGpc1Label, batchType):
+
+        sql = "SELECT batch_id \
+               FROM batch \
+               WHERE timestamp > '" + epoch + "' \
+               AND dvo_db = '" + dvoGpc1Label + "' \
+               AND loaded_to_datastore = 1 \
+               AND batch_type = '" + batchType + "' \
+               AND merge_worthy = 0 \
+               AND loaded_to_ODM != -1 \
+               AND deleted_datastore = 0"
+
+        ids = []
+        try:
+            rs = self.executeQuery(sql)
+            while (rs.next()):
+                ids.append(rs.getInt(1))
+        except:
+            self.logger.exception("Can't query for un-mergeworthy batch ids in ipptopsps Db")
+
+        rs.close()
+
+        self.logger.debug("Found %d unloaded items" % len(ids))
+
+        return ids
+
+    '''
+    Returns a list of processed batch IDs that are merge_worthy, but not yet merged, for this epoch and dvo label
+    '''
+    def getUnmergedBatchIDs(self, epoch, dvoGpc1Label, batchType):
+
+        sql = "SELECT batch_id \
+               FROM batch \
+               WHERE timestamp > '" + epoch + "' \
+               AND dvo_db = '" + dvoGpc1Label + "' \
+               AND merge_worthy = 1 \
+               AND batch_type = '" + batchType + "' \
+               AND merged != 1"
+
+        ids = []
+        try:
+            rs = self.executeQuery(sql)
+            while (rs.next()):
+                ids.append(rs.getInt(1))
+        except:
+            self.logger.exception("Can't query for unmerged batch ids in ipptopsps Db")
+
+        rs.close()
+
+        self.logger.debug("Found %d un-merged items" % len(ids))
+
+        return ids
+
+    '''
+    Returns a list of batch IDs for this epoch and dvo label for which the specified column is set to value
+    '''
+    def getBatchIDs(self, epoch, dvoGpc1Label, column, value):
+
+        sql = "SELECT DISTINCT batch_id \
+               FROM batch \
+               WHERE timestamp > '" + epoch + "' \
+               AND dvo_db = '" + dvoGpc1Label + "' \
+               AND " + column + " = " + str(value)
+
+        ids = []
+        try:
+            rs = self.executeQuery(sql)
+            while (rs.next()):
+                ids.append(rs.getInt(1))
+        except:
+            self.logger.exception("Can't query for batch ids with " + column + " = " + str(value) + " in ipptopsps Db")
+
+        rs.close()
+
+        self.logger.debug("Found %d batches with %s = %d" % (len(ids), column, value))
+
+        return ids
+
+    '''
+    Returns the total detections published to the datastore for this epoch, dvo label and batch type
+    '''
+    def getTotalDetectionsPublished(self, batchType, epoch, dvoGpc1Label):
+
+        sql = "SELECT SUM(total_detections) \
                FROM batch \
                WHERE batch_type = '" + batchType + "' \
                AND timestamp > '" + epoch + "' \
                AND dvo_db = '" + dvoGpc1Label + "' \
-               AND " + column + " = 1"
+               AND processed = 1 \
+               AND loaded_to_datastore = 1"
+
+        total = -1
+        try:
+            rs = self.executeQuery(sql)
+            rs.first()
+            total =  rs.getLong(1)
+        except:
+            self.logger.exception("Can't query for total detections published")
+
+        rs.close()
+
+        self.logger.debug("Found %d detections" % total)
+
+        return total
+
+    '''
+    Returns a list of IDs for this batch type, epoch and dvo label (i.e. either 
+            cam_ids or stack_ids) for which the specified column is set to value
+    '''
+    def getStageIDs(self, batchType, epoch, dvoGpc1Label, column, value):
+
+        sql = "SELECT DISTINCT stage_id \
+               FROM batch \
+               WHERE batch_type = '" + batchType + "' \
+               AND timestamp > '" + epoch + "' \
+               AND dvo_db = '" + dvoGpc1Label + "' \
+               AND " + column + " = " + str(value)
 
         ids = []
@@ -127 +327 @@
                 ids.append(rs.getInt(1))
         except:
-            self.logger.exception("Can't query for ids with " + column + " = 1 in ipptopsps Db")
+            self.logger.exception("Can't query for ids with " + column + " = " + str(value) + " in ipptopsps Db")
 
         rs.close()
@@ -136 +336 @@
 
     '''
+    Series of getter methods that utilise getBatchIDs() method above
+    '''
+    def getloadedToDatastoreBatchIDs(self, epoch, dvoGpc1Label):
+        return self.getBatchIDs(epoch, dvoGpc1Label, "loaded_to_datastore", 1)
+    def getFailedLoadedToODMBatchIDs(self, epoch, dvoGpc1Label):
+        return self.getBatchIDs(epoch, dvoGpc1Label, "loaded_to_ODM", -1)
+
+    '''
+    TODO remove these
     Series of getter methods that utilise getStageIDs() method above
     '''
     def getProcessedIDs(self, batchType, epoch, dvoGpc1Label):
-        return self.getStageIDs(batchType, epoch, dvoGpc1Label, "processed")
-    def getloadedToDatastoreIDs(self, batchType, epoch, dvoGpc1Label):
-        return self.getStageIDs(batchType, epoch, dvoGpc1Label, "loaded_to_datastore")
+        return self.getStageIDs(batchType, epoch, dvoGpc1Label, "processed", 1)
+    def getFailedProcessedIDs(self, batchType, epoch, dvoGpc1Label):
+        return self.getStageIDs(batchType, epoch, dvoGpc1Label, "processed", -1)
+    def getLoadedToDatastoreIDs(self, batchType, epoch, dvoGpc1Label):
+        return self.getStageIDs(batchType, epoch, dvoGpc1Label, "loaded_to_datastore", 1)
+    def getFailedLoadedToDatastoreIDs(self, batchType, epoch, dvoGpc1Label):
+        return self.getStageIDs(batchType, epoch, dvoGpc1Label, "loaded_to_datastore", -1)
     def getLoadedToODMIDs(self, batchType, epoch, dvoGpc1Label):
-        return self.getStageIDs(batchType, epoch, dvoGpc1Label, "loaded_to_ODM")
-    def getMergeWothyIDs(self, batchType, epoch, dvoGpc1Label):
-        return self.getStageIDs(batchType, epoch, dvoGpc1Label, "merge_worthy")
+        return self.getStageIDs(batchType, epoch, dvoGpc1Label, "loaded_to_ODM", 1)
+    def getFailedLoadedToODMIDs(self, batchType, epoch, dvoGpc1Label):
+        return self.getStageIDs(batchType, epoch, dvoGpc1Label, "loaded_to_ODM", -1)
+    def getMergeWorthyIDs(self, batchType, epoch, dvoGpc1Label):
+        return self.getStageIDs(batchType, epoch, dvoGpc1Label, "merge_worthy", 1)
+    def getFailedMergeWorthyIDs(self, batchType, epoch, dvoGpc1Label):
+        return self.getStageIDs(batchType, epoch, dvoGpc1Label, "merge_worthy", -1)
     def getMergedIDs(self, batchType, epoch, dvoGpc1Label):
-        return self.getStageIDs(batchType, epoch, dvoGpc1Label, "merged")
+        return self.getStageIDs(batchType, epoch, dvoGpc1Label, "merged", 1)
+    def getFailedMergedIDs(self, batchType, epoch, dvoGpc1Label):
+        return self.getStageIDs(batchType, epoch, dvoGpc1Label, "merged", -1)
+
+    '''
+    Returns a list of stages
+    '''
+    def getStages(self):
+        return ['processed', 'loaded_to_datastore', 'loaded_to_ODM', 'merge_worthy', 'deleted_datastore', 'deleted_dxlayer', 'merged', 'deleted_local' ]
+
+    '''
+    Returns a count of batches processed in the specified time period.
+    '''
+    def countBatchesInLastPeriod(self, batchType, epoch, dvoGpc1Label, interval):
+
+        sql = "SELECT COUNT(batch_id) \
+               FROM batch \
+               WHERE batch_type = '" + batchType + "' \
+               AND processed = 1 \
+               AND timestamp > '" + epoch + "' \
+               AND dvo_db = '" + dvoGpc1Label + "' \
+               AND timestamp BETWEEN now() - INTERVAL " + interval + " AND now()"
+
+        try:
+            rs = self.executeQuery(sql)
+            if rs.first(): return rs.getInt(1)
+        except:
+            self.logger.exception("Unable to count batches in interval")
+            return 0
 
     '''
@@ -155 +400 @@
     def getTimeOfLastBatchPublished(self, batchType, epoch, dvoGpc1Label):
 
-        minutes = None
-
-        sql = "SELECT TIMESTAMPDIFF(MINUTE, timestamp, now()) \
+        seconds = None
+
+        sql = "SELECT TIMESTAMPDIFF(SECOND, timestamp, now()) \
                FROM batch \
                WHERE batch_type = '" + batchType + "' \
-               AND loaded_to_datastore \
+               AND loaded_to_datastore = 1 \
                AND timestamp > '" + epoch + "' \
                AND dvo_db = '" + dvoGpc1Label + "' \
@@ -167 +412 @@
         try:
             rs = self.executeQuery(sql)
-            if rs.first(): minutes = rs.getFloat(1)
+            if rs.first(): seconds = rs.getFloat(1)
         except:
             self.logger.exception("Unable to get last batch published")
 
-        if not minutes: return "Never"
-
+        if not seconds: return "Never"
+
+        minutes = seconds/60.0
         hours = minutes/60.0
-        days = float(hours)/24.0
-        weeks = float(days)/7.0
-
-        if minutes < 60: return "%.1f minutes ago" % minutes
+        days = hours/24.0
+        weeks = days/7.0
+
+        if seconds < 60: return "%.1f secs ago" % seconds
+        elif minutes < 60: return "%.1f mins ago" % minutes
         elif hours < 48: return "%.1f hours ago" % hours
         elif days < 7: return "%.1f days ago" % days
@@ -183 +430 @@
 
     '''
-    TODO
-    '''
-    def getBatchesPerDay(self, batchType, startTime, endTime=""):
-
-        bph = self.getBatchesPerHour(batchType, startTime, endTime)
-        return bph * 24
-
-    '''
-    TODO
-    '''
-    def getBatchesPerHour(self, batchType, startTime, endTime=""):
-
-        bpm = self.getBatchesPerMinute(batchType, startTime, endTime)
-        return bpm * 60
-
-    '''
-    TODO
-    '''
-    def getAverageTimePerBatch(self, batchType, startTime, endTime=""):
-
-        sql = "SELECT (TIMESTAMPDIFF(MINUTE, min(timestamp), max(timestamp)) / COUNT(*)) \
-               FROM batch \
-               WHERE timestamp > '" + startTime + "' and batch_type = '" + batchType + "'"
-
-        try:
-            rs = self.executeQuery(sql)
-            rs.first()
-            time = rs.getDouble(1)
-        except:
-            self.logger.exception("Unable to get batches per minute")
-
-        return time
-
-    '''
-    TODO
-    '''
-    def getBatchesPerMinute(self, batchType, startTime, endTime=""):
-
-        sql = "SELECT (count(*) / TIMESTAMPDIFF(MINUTE, min(timestamp), max(timestamp)) ) \
-               FROM batch \
-               WHERE timestamp > '" + startTime + "' and batch_type = '" + batchType + "'"
-
-        try:
-            rs = self.executeQuery(sql)
-            rs.first()
-            bpm = rs.getDouble(1)
-        except:
-            self.logger.exception("Unable to get batches per minute")
-
-        return bpm
-
-    '''
     Updates min/max object ID on this table and batch 
     '''
-    def updateMinMaxObjID(self, batchID, minObjID, maxObjID):
+    def updateDetectionStats(self, batchID, minObjID, maxObjID, totalDetections):
 
         sql = "UPDATE batch SET \
                min_obj_id = " + str(minObjID) + ", \
-               max_obj_id = " + str(maxObjID) + " \
+               max_obj_id = " + str(maxObjID) + ", \
+               total_detections = " + str(totalDetections) + " \
                WHERE batch_id = " + str(batchID)
 
@@ -257 +453 @@
         self.execute(sql)
 
-
     '''
     Updates batch processed field 
@@ -265 +460 @@
 
     '''
-    Updates batch deleted field 
-    '''
-    def updateDeleted(self, batchID, deleted):
-        self.updateColumn(batchID, "deleted", deleted)
+    Updates batch deleted local field 
+    '''
+    def updateDeletedLocal(self, batchID, value):
+        self.updateColumn(batchID, "deleted_local", value)
+
+    '''
+    Updates batch deleted datastore field 
+    '''
+    def updateDeletedDatastore(self, batchID, value):
+        self.updateColumn(batchID, "deleted_datastore", value)
+
+    '''
+    Updates batch deleted dxlayer field 
+    '''
+    def updateDeletedDxlayer(self, batchID, value):
+        self.updateColumn(batchID, "deleted_dxlayer", value)
 
     '''
@@ -290 +497 @@
 
     '''
-    Have we already processed and published this batch?
-    '''
-    def alreadyProcessed(self, stage_id):
+    Is someone processing this item right now? Checks whether this was attempted within the last 2 HOUR
+    '''
+    def processingNow(self, batchType, stage_id, epoch, dvoGpc1Label):
 
         sql = "SELECT COUNT(*) \
                FROM batch \
                WHERE stage_id = " + str(stage_id) + " \
-               AND processed \
-               AND loaded_to_datastore"
+               AND batch_type = '" + batchType + "' \
+               AND dvo_db = '" + dvoGpc1Label + "' \
+               AND timestamp BETWEEN now() - INTERVAL 4 HOUR AND now()"
 
         try:
@@ -304 +512 @@
             rs.first()
             if rs.getInt(1) > 0: 
-                self.logger.error("Batch with stage_id = "+str(stage_id) + " has already been processed and published to datastore")
+                self.logger.debugPair(str(stage_id) + " is already being processed", "skipping")
                 return True
             else: 
                 return False
         except:
+            self.logger.exception("Unable to check whether this batch is being processed")
+
+    '''
+    Gets the batch type of this batch
+    '''
+    def getBatchType(self, batch_id):
+
+        sql = "SELECT batch_type \
+               FROM batch \
+               WHERE batch_id = " + str(batch_id)
+
+        batch_type = "NA"
+        try:
+            rs = self.executeQuery(sql)
+            rs.first()
+            batch_type = rs.getString(1) 
+        except:
             self.logger.exception("Unable to check whether this batch has already been processed")
-
+ 
+        return batch_type
+
+    '''
+    Have we already processed this stage_id?
+    '''
+    def alreadyProcessed(self, batchType, stage_id, epoch, dvoGpc1Label):
+
+        sql = "SELECT COUNT(*) \
+               FROM batch \
+               WHERE stage_id = " + str(stage_id) + " \
+               AND batch_type = '" + batchType + "' \
+               AND timestamp > '" + epoch + "' \
+               AND dvo_db = '" + dvoGpc1Label + "' \
+               AND processed = 1"
+
+        try:
+            rs = self.executeQuery(sql)
+            rs.first()
+            if rs.getInt(1) > 0: 
+                self.logger.debugPair(str(stage_id) + " has already been published", "skipping")
+                return True
+            else: 
+                return False
+        except:
+            self.logger.exception("Unable to check whether this batch has already been processed")
+
+    '''
+    Has this stage_id consistently failed to process?
+    '''
+    def consistentlyFailed(self, batchType, stage_id, epoch, dvoGpc1Label):
+
+        sql = "SELECT COUNT(*) \
+               FROM batch \
+               WHERE stage_id = " + str(stage_id) + " \
+               AND batch_type = '" + batchType + "' \
+               AND timestamp > '" + epoch + "' \
+               AND dvo_db = '" + dvoGpc1Label + "' \
+               AND processed = -1"
+
+        try:
+            rs = self.executeQuery(sql)
+            rs.first()
+            if rs.getInt(1) >= self.MAX_FAILS: 
+                self.logger.errorPair(str(stage_id) + " has failed %d times" % self.MAX_FAILS, "skipping")
+                return True
+            else: 
+                return False
+        except:
+            self.logger.exception("Unable to check whether this batch has consistently failed")
+
+    '''
+    Returns a list of stage_ids that have failed the max number of times
+    '''
+    def getConsistentlyFailedIDs(self, batchType, epoch, dvoGpc1Label):
+
+        sql = "SELECT DISTINCT stage_id, COUNT(stage_id) AS numoccur \
+               FROM batch \
+               WHERE batch_type = '" + batchType + "' \
+               AND timestamp > '" + epoch + "' \
+               AND dvo_db = '" + dvoGpc1Label + "' \
+               AND processed = -1 \
+               GROUP BY stage_id HAVING numoccur >= " + str(self.MAX_FAILS)
+
+        ids = []
+        try:
+            rs = self.executeQuery(sql)
+            while (rs.next()):
+                ids.append(rs.getInt(1))
+        except:
+            self.logger.exception("Can't query for consistantly failed ids in ipptopsps Db")
+
+        rs.close()
+
+        self.logger.debug("Found %d consistantly failed stage_ids for batch type='%s'" % (len(ids), batchType))
+
+        return ids
+
+    '''
+    Locks the batch table. This will wait if the lock is held by someone else
+    '''
+    def lockBatchTable(self):
+        self.lockTable("batch")
+
+    '''
+    Creates a new batch. This is done with a locked table in order that no two clients can start work on the same item
+    '''
+    def createNewBatch(self, batchType, stageID, config):
+
+        batchID = -1;
+
+        if config.force or \
+            (not self.alreadyProcessed(batchType, stageID, config.epoch, config.dvoLabel) \
+            and not self.processingNow(batchType, stageID, config.epoch, config.dvoLabel) \
+            and not self.consistentlyFailed(batchType, stageID, config.epoch, config.dvoLabel)): 
+
+            sql = "INSERT INTO batch ( \
+                   batch_type, \
+                   stage_id, \
+                   survey, \
+                   dvo_db, \
+                   datastore_product \
+                   ) VALUES ( \
+                       '" + batchType + "', \
+                       " + str(stageID) + ", \
+                       '" + config.survey + "', \
+                       '" + config.dvoLabel + "', \
+                       '" + config.datastoreProduct + "' \
+                       )"
+
+            self.execute(sql)
+
+            sql = "SELECT MAX(batch_id) FROM batch"
+
+            try:
+                rs = self.executeQuery(sql)
+                rs.first()
+                batchID = rs.getInt(1)
+                self.logger.debug("Created new batch in ippToPsps database with batchID = %d" % batchID)
+            except:
+                self.logger.exception("Unable to get batch ID")
+                batchID = -1
+
+        return batchID;
      
 
@@ -346 +694 @@
         self.execute(sql)
 
+    '''
+    Resets a batch ready for re-loading to the datastore
+    '''
+    def resetBatch(self, batchID):
+
+        sql = "UPDATE batch SET \
+               loaded_to_datastore = 0, \
+               loaded_to_ODM = 0, \
+               merge_worthy = 0, \
+               merged = 0, \
+               deleted_datastore = 0, \
+               deleted_dxlayer = 0, \
+               comment = 'This batch was reset, ready for re-load to datastore' \
+               WHERE batch_id = " + str(batchID)
+
+        self.execute(sql)
+
+    '''
+    Remove client
+    '''
+    def removeClient(self, host, pid):
+        self.execute("DELETE FROM clients WHERE host = '" + host + "' AND pid = " + str(pid))
+
+    '''
+    Insert client
+    '''
+    def insertClient(self, type, host, pid, config):
+        self.execute("INSERT INTO clients (timestamp, type, host, pid, config) VALUES (now(), '" + type + "', '" + host + "', " + str(pid) + ", '" + config + "')")
+
+    '''
+    Update clients, or inserts it for the first time
+    '''
+    def updateClient(self, type, host, pid, config):
+
+        try:
+            self.insertClient(type, host, pid, config)
+        except:
+            self.execute("UPDATE clients SET timestamp = now() WHERE host = '" + host + "' AND pid = " + str(pid))
+
+    '''
+    Is this client paused?
+    '''
+    def isPaused(self, host, pid):
+        return self.getClientsColumnBoolean("paused", host, pid)
+
+    '''
+    Is this client killed?
+    '''
+    def isKilled(self, host, pid):
+        return self.getClientsColumnBoolean("killed", host, pid)
+
+    '''
+    Returns boolean value for this column in clients table
+    '''
+    def getClientsColumnBoolean(self, col, host, pid):
+
+        sql = "SELECT " + col + " \
+               FROM clients \
+               WHERE host = '" + host + "' AND pid = " + str(pid)
+
+        try:
+            rs = self.executeQuery(sql)
+            rs.first()
+            if rs.getInt(1) == 1: return True
+            else: return False
+        except:
+            self.logger.exception("Unable to check whether " + col + " is true or false in clients table")
+
+        return False
+
+
+    '''
+    Populates Config object
+    '''
+    def readConfig(self, host, pid):
+
+        # first get config defined for this client
+        sql = "SELECT config \
+               FROM clients \
+               WHERE host = '" + host + "' \
+               AND pid = " + str(pid)
+
+        try:
+            rs = self.executeQuery(sql)
+            rs.first()
+            self.config.name = rs.getString(1)
+        except:
+            self.logger.errorPair("No config set for", "%s (pid=%d)" % (host, pid))
+            self.config.isLoaded = False
+            return False
+
+        # now load that config
+        sql = "SELECT \
+        datastore_product, \
+        datastore_type, \
+        datastore_publish, \
+        dvo_label, \
+        dvo_location, \
+        min_ra, \
+        max_ra, \
+        min_dec, \
+        max_dec, \
+        box_size, \
+        base_path, \
+        data_release, \
+        delete_local, \
+        delete_datastore, \
+        delete_dxlayer, \
+        epoch, \
+        survey, \
+        psps_survey, \
+        queue_P2, \
+        queue_ST \
+        FROM config \
+        WHERE name = '" + self.config.name + "'"
+
+        self.config.batchTypes = []
+        try:
+            rs = self.executeQuery(sql)
+            rs.first()
+            self.config.datastoreProduct = rs.getString(1)
+            self.config.datastoreType = rs.getString(2)
+            if rs.getInt(3) == 1: self.config.datastorePublishing = True
+            else: self.config.datastorePublishing = False
+            self.config.dvoLabel = rs.getString(4)
+            self.config.dvoLocation = rs.getString(5)
+            self.config.minRa = rs.getDouble(6)
+            self.config.maxRa = rs.getDouble(7)
+            self.config.minDec = rs.getDouble(8)
+            self.config.maxDec = rs.getDouble(9)
+
+            self.config.boxSize = rs.getDouble(10)
+            self.config.halfBox = self.config.boxSize/2.0
+            self.config.boxSizeWithBorder = self.config.boxSize + (self.config.BORDER * 2)
+
+            self.config.basePath = rs.getString(11)
+            self.config.dataRelease = rs.getInt(12)
+            self.config.deleteLocal = rs.getInt(13)
+            self.config.deleteDatastore = rs.getInt(14)
+            self.config.deleteDxLayer = rs.getInt(15)
+            self.config.epoch = rs.getString(16)
+            self.config.survey = rs.getString(17)
+            self.config.pspsSurvey = rs.getString(18)
+            if rs.getInt(19) == 1: self.config.batchTypes.append("P2")
+            if rs.getInt(20) == 1: self.config.batchTypes.append("ST")
+            self.config.force = False # TODO
+            self.config.test = False # TODO
+            self.config.isLoaded = True
+        except:
+            self.logger.errorPair("Could not read config with name", self.config.name)
+            self.config.isLoaded = False
+
+        return self.config.isLoaded
+
+    '''
+    Gets id of box with these coords
+    '''
+    def getBoxId(self, config, ra, dec, boxSide):
+
+        sql = "SELECT id FROM box \
+               WHERE config  = '" + config + "' \
+               AND ra_center = " + str(ra) + " \
+               AND dec_center = " + str(dec) + " \
+               AND box_side = " + str(boxSide)
+
+        id = -1
+        try:
+            rs = self.executeQuery(sql)
+            rs.first()
+            id = rs.getInt(1)
+        except:
+            pass
+
+        return id
+
+
+    '''
+    Remove all boxes and associated ids in pending table
+    '''
+    def removeAllBoxes(self, config):
+        self.execute("DELETE FROM box WHERE config = '" + config + "'")
+
+    '''
+    Removes a box and it's associated ids in the pending table
+    '''
+    def removeBox(self, config, ra, dec, boxSide):
+        self.execute("DELETE FROM box \
+                WHERE config = '" + config + "' \
+                AND ra_center = " + str(ra) + " \
+                AND dec_center = " + str(dec) + " \
+                AND box_side = " + str(boxSide))
+
+    '''
+    Inserts new box into box table. If it already exisits, it returns existing box id 
+    '''
+    def insertBox(self, config, ra, dec, boxSide):
+
+        id = self.getBoxId(config, ra, dec, boxSide)
+
+        # does not already exists so insert
+        if id < 0:
+
+            sql = "INSERT INTO box \
+                   (config, ra_center, dec_center, box_side) \
+                   VALUES \
+                   ('" + config + "', " + str(ra) + ", " + str(dec) + ", " + str(boxSide) + ")"
+
+            self.execute(sql)
+            id = self.getBoxId(config, ra, dec, boxSide)
+
+        return id
+
+    '''
+    Inserts new box ids into the pending table 
+    '''
+    def insertPending(self, box_id, batchType, ids):
+
+        # first delete old pending items
+        self.execute("DELETE FROM pending WHERE box_id = " + str(box_id))
+
+        for id in ids:
+            sql = "INSERT INTO pending \
+               (box_id, batch_type, stage_id) \
+               VALUES \
+               (" + str(box_id) + ", '" + batchType + "', " + str(id) + ")"
+
+            self.execute(sql)
+
+    '''
+    Returns ids for all boxes with pending items
+    '''
+    def getBoxIds(self, config):
+
+        sql = "SELECT DISTINCT id \
+               FROM box \
+               JOIN pending ON (id = box_id)  \
+               WHERE config = '" + config + "' \
+               ORDER BY ra_center, dec_center"
+
+        ids = []
+        try:
+            rs = self.executeQuery(sql)
+            while (rs.next()): ids.append(rs.getInt(1))
+            rs.close()
+        except:
+            self.logger.errorPair("Can't get boxes", sql)
+
+        return ids
+
+    '''
+    Returns dimensions of this box
+    '''
+    def getBoxDimensions(self, boxId):
+
+        sql = "SELECT ra_center, dec_center, box_side FROM box WHERE id = " + str(boxId)
+
+        boxDim = {}
+        rs = self.executeQuery(sql)
+        rs.first()
+
+        boxDim['RA'] = rs.getFloat(1)
+        boxDim['DEC'] = rs.getFloat(2)
+        boxDim['SIDE'] = rs.getFloat(3)
+
+        return boxDim
+
+    '''
+    Returns ids for pending items in this box for this batch type
+    '''
+    def getPendingIds(self, boxId, batchType):
+
+        sql = "SELECT stage_id FROM pending WHERE batch_type = '" + batchType + "' AND box_id = " + str(boxId)
+
+        ids = []
+        try:
+            rs = self.executeQuery(sql)
+            while (rs.next()): ids.append(rs.getInt(1))
+            rs.close()
+        except:
+            self.logger.errorPair("Can't get ids from this box", sql)
+
+        return ids
+
+    '''
+    Returns max number of pensing items in a box for this config
+    '''
+    def getMaxPendingInBox(self, config):
+
+        sql = "SELECT MAX(pending) \
+               FROM (SELECT COUNT(stage_id) AS pending \
+                       FROM pending \
+                       JOIN box ON (id = box_id) \
+                       WHERE config = '" + config + "' \
+                       GROUP BY box_id) AS t"
+
+        try:
+            rs = self.executeQuery(sql)
+            rs.first()
+            return rs.getInt(1)
+        except:
+            self.logger.errorPair("Could not count max pending", sql)
+            return 100
+
+    '''
+    Writes density-plot data to file for pending stuff over sky for given config
+    '''
+    def createPendingDensityPlotData(self, config, DATFILE):
+
+        sql = "SELECT ra_center, dec_center, COUNT(stage_id) \
+               FROM box \
+               LEFT JOIN pending ON (id = box_id) \
+               WHERE config = '" + config + "' \
+               GROUP BY id \
+               ORDER BY ra_center, dec_center"
+
+        try:
+            rs = self.executeQuery(sql)
+            while (rs.next()):
+                print >> DATFILE, rs.getFloat(1), rs.getFloat(2), rs.getInt(3)
+            rs.close()
+        except:
+            self.logger.errorPair("Can't get density plot data", sql)
 
     '''
@@ -352 +1022 @@
     def __del__(self):
 
-        self.logger.debug("IppToPspsDb Desstructor")
+        self.logger.debug("IppToPspsDb Destructor")
         self.stmt.close()
         self.con.close()