Changeset 33098 for branches/meh_branches/ppsub_test/ippToPsps/jython/ipptopspsdb.py

Timestamp:

Jan 11, 2012, 11:19:21 PM (15 years ago)

Author:

mhuber

Message:

merged latest r33093 trunk changes to meh_branches/ppsub_test

Location:

branches/meh_branches/ppsub_test

Files:

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

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

@@ -20 +20 @@
         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):
+        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 = []
@@ -83 +50 @@
 
     '''
-    Returns a list of processed batch IDs that are not yet merged for this epoch and dvo label
-    '''
-    def getProcessedButUnmergedBatchIDs(self, epoch, dvoGpc1Label):
+    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 dvo_db = '" + dvoGpc1Label + "' \
-               AND loaded_to_datastore \
-               AND !merged"
-
-        ids = []
-        try:
-            rs = self.executeQuery(sql)
-            while (rs.next()):
-                ids.append(rs.getInt(1))
-        except:
-            self.logger.exception("Can't query for processed 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 \
+               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))
+        except:
+            self.logger.exception("Can't query for merged batch ids in ipptopsps Db")
+
+        rs.close()
+
+        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 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 = []
+        try:
+            rs = self.executeQuery(sql)
+            while (rs.next()):
+                ids.append(rs.getInt(1))
+        except:
+            self.logger.exception("Can't query for processed batch ids not loaded to datastore ipptopsps Db")
+
+        rs.close()
+
+        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 = 1 \
+               AND batch_type = '" + batchType + "' \
+               AND loaded_to_ODM = 0 \
+               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 unloaded 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 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"
-
-        ids = []
-        try:
-            rs = self.executeQuery(sql)
-            while (rs.next()):
-                ids.append(rs.getInt(1))
-        except:
-            self.logger.exception("Can't query for ids with " + column + " = 1 in ipptopsps Db")
+               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 = []
+        try:
+            rs = self.executeQuery(sql)
+            while (rs.next()):
+                ids.append(rs.getInt(1))
+        except:
+            self.logger.exception("Can't query for ids with " + column + " = " + str(value) + " in ipptopsps Db")
 
         rs.close()
@@ -136 +302 @@
 
     '''
+    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 +366 @@
     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 +378 @@
         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 +396 @@
 
     '''
-    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 +419 @@
         self.execute(sql)
 
-
     '''
     Updates batch processed field 
@@ -265 +426 @@
 
     '''
-    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)
 
     '''
@@ -281 +454 @@
     def updateOdmStatus(self, batchID, odmStatus):
 
+        if odmStatus['LOADFAILED'] == 1: loadedToODM = -1
+        else: loadedToODM = odmStatus['LOADEDTOODM']
         sql = "UPDATE batch \
-               SET loaded_to_ODM = " + str(odmStatus['LOADEDTOODM']) + ", \
+               SET loaded_to_ODM = " + str(loadedToODM) + ", \
                merge_worthy = " + str(odmStatus['MERGEWORTHY']) + ", \
                merged = " + str(odmStatus['MERGED']) + " \
@@ -290 +465 @@
 
     '''
-    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 +480 @@
             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.errorPair(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.errorPair(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, survey, epoch, dvoDb, datastoreProduct, force):
+
+        batchID = -1;
+
+        if force or \
+            (not self.alreadyProcessed(batchType, stageID, epoch, dvoDb) \
+            and not self.processingNow(batchType, stageID, epoch, dvoDb) \
+            and not self.consistentlyFailed(batchType, stageID, epoch, dvoDb)): 
+
+            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"
+
+            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 +662 @@
         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)
 
     '''