Changeset 32337 for branches/eam_branches/ipp-20110710/ippToPsps/jython/ipptopspsdb.py

Timestamp:

Sep 6, 2011, 11:00:22 AM (15 years ago)

Author:

eugene

Message:

merge changes from trunk

Location:

branches/eam_branches/ipp-20110710/ippToPsps/jython

Files:

• . (modified) (1 prop)
• ipptopspsdb.py (modified) (7 diffs)

branches/eam_branches/ipp-20110710/ippToPsps/jython/ipptopspsdb.py

@@ -57 +57 @@
 
     '''
-    TODO
-    '''
-    def getUnprocessedIDsForThisStage(self, dvoDb, batchType, startDate):
-
-        if batchType == "P2": # TODO define these someplace 
-
-            stage = "cam"
-            sql = "SELECT DISTINCT stage_id \
-                   FROM gpc1.addRun \
-                   WHERE stage = '" + stage + "' \
-                   AND dvodb = '" + dvoDb + "' \
-                   AND stage_id NOT IN "
-
-        elif batchType == "ST": 
-
-            stage = "staticsky"
-            sql = "SELECT DISTINCT stack_id \
-                   FROM gpc1.staticskyInput \
-                   JOIN gpc1.addRun ON(gpc1.staticskyInput.sky_id = gpc1.addRun.stage_id) \
-                   WHERE stage = '" + stage + "' \
-                   AND dvodb = '" + dvoDb + "' \
-                   AND stack_id NOT IN "
-
-        sql = sql + "(SELECT stage_id \
-              FROM batch \
-              WHERE batch_type = '" + batchType + "' \
-              AND timestamp > '" + startDate + "' \
-              AND loaded_to_datastore)"
+    Returns a list of processed batch IDs that are merged 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 != 0 \
+               AND " + column + " = 0"
 
         ids = []
@@ -92 +75 @@
                 ids.append(rs.getInt(1))
         except:
-            self.logger.exception("Can't query for ids in DVO")
+            self.logger.exception("Can't query for merged batch ids in ipptopsps Db")
 
         rs.close()
 
-        self.logger.debug("Found %d unprocessed items in DVO database '%s' for stage='%s'" % (len(ids), dvoDb, stage))
+        self.logger.debug("Found %d merged but un-deleted items" % len(ids))
 
         return ids
 
     '''
-    TODO
-    '''
-    def getLastBatchPublished(self, batchType):
-
-        sql = "SELECT TIMESTAMPDIFF(MINUTE, timestamp, now())/60.0 \
+    Returns a list of processed batch IDs that are merged 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 merged 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 merged 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
+    '''
+    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"
+
+        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 have not yet been loaded to the ODM
+    '''
+    def getBatchIDsUnloadedToODM(self, epoch, dvoGpc1Label):
+
+        sql = "SELECT DISTINCT batch_id \
+               FROM batch \
+               WHERE timestamp > '" + epoch + "' \
+               AND dvo_db = '" + dvoGpc1Label + "' \
+               AND loaded_to_datastore = 1 \
+               AND loaded_to_ODM = 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 have not failed to load and are not yet merged, for this epoch and dvo label
+    '''
+    def getUnfinishedBatchIDs(self, epoch, dvoGpc1Label):
+
+        sql = "SELECT DISTINCT batch_id \
+               FROM batch \
+               WHERE timestamp > '" + epoch + "' \
+               AND dvo_db = '" + dvoGpc1Label + "' \
+               AND loaded_to_datastore = 1 \
+               AND merged != 1 \
+               AND loaded_to_ODM != -1"
+
+
+        ids = []
+        try:
+            rs = self.executeQuery(sql)
+            while (rs.next()):
+                ids.append(rs.getInt(1))
+        except:
+            self.logger.exception("Can't query for unfinished batch ids in ipptopsps Db")
+
+        rs.close()
+
+        self.logger.debug("Found %d unfinished 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 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 loaded_to_datastore \
+               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()
+
+        self.logger.debug("Found %d batches with %s = 1 for batch type='%s'" % (len(ids), column, batchType))
+
+        return ids
+
+    '''
+    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)
+
+    '''
+    Series of getter methods that utilise getStageIDs() method above
+    '''
+    def getProcessedIDs(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 getLoadedToODMIDs(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 getMergedIDs(self, batchType, epoch, dvoGpc1Label):
+        return self.getStageIDs(batchType, epoch, dvoGpc1Label, "merged", 1)
+
+    '''
+    Returns the time (as a string) that the last batch was published to the datastore for
+    this epoch, dvo label and batch type
+    '''
+    def getTimeOfLastBatchPublished(self, batchType, epoch, dvoGpc1Label):
+
+        minutes = None
+
+        sql = "SELECT TIMESTAMPDIFF(MINUTE, timestamp, now()) \
+               FROM batch \
+               WHERE batch_type = '" + batchType + "' \
+               AND loaded_to_datastore = 1 \
+               AND timestamp > '" + epoch + "' \
+               AND dvo_db = '" + dvoGpc1Label + "' \
                ORDER BY timestamp DESC LIMIT 1"
 
         try:
             rs = self.executeQuery(sql)
-            rs.first()
-            hours = rs.getFloat(1)
+            if rs.first(): minutes = rs.getFloat(1)
         except:
             self.logger.exception("Unable to get last batch published")
 
-        return hours
+        if not minutes: return "Never"
+
+        hours = minutes/60.0
+        days = hours/24.0
+        weeks = days/7.0
+
+        if minutes < 60: return "%.1f minutes ago" % minutes
+        elif hours < 48: return "%.1f hours ago" % hours
+        elif days < 7: return "%.1f days ago" % days
+        return "%.1f weeks ago" % weeks
 
     '''
@@ -172 +343 @@
         return bpm
 
-
-    '''
-    TODO
-    '''
-    def getTotalFailedBatches(self, batchType, startTime, endTime=""):
-
-        sql = "SELECT COUNT(DISTINCT stage_id) \
-               FROM batch \
-               WHERE timestamp > '" + startTime + "' \
-               AND batch_type = '" + batchType + "' \
-               AND !processed \
-               AND stage_id NOT IN \
-               (SELECT DISTINCT stage_id \
-                FROM batch \
-                WHERE timestamp > '" + startTime + "'  \
-                AND batch_type = '" + batchType + "' \
-                AND loaded_to_datastore)"
-
-        try:
-            rs = self.executeQuery(sql)
-            rs.first()
-            total = rs.getInt(1)
-        except:
-            self.logger.exception("Unable to count failed batches")
-
-        return total
-    '''
-    TODO
-    '''
-    def getTotalBatchesPublished(self, batchType, startTime, endTime=""):
-
-        sql = "SELECT COUNT(*) \
-               FROM batch \
-               WHERE timestamp > '" + startTime + "' \
-               AND batch_type = '" + batchType + "' \
-               AND loaded_to_datastore"
-
-        try:
-            rs = self.executeQuery(sql)
-            rs.first()
-            total = rs.getInt(1)
-        except:
-            self.logger.exception("Unable to count batches")
-
-        return total
-
     '''
     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)
 
@@ -231 +357 @@
 
     '''
+    Updates a batch-table column
+    '''
+    def updateColumn(self, batchID, column, value):
+
+        sql = "UPDATE batch \
+               SET " + column + " = " + str(value) + " \
+               WHERE batch_id = " + str(batchID)
+
+        self.execute(sql)
+
+    '''
     Updates batch processed field 
     '''
     def updateProcessed(self, batchID, processed):
-
+        self.updateColumn(batchID, "processed", processed)
+
+    '''
+    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)
+
+    '''
+    Updates batch loaded_to_datastore field 
+    '''
+    def updateLoadedToDatastore(self, batchID, loadedToDatastore):
+        self.updateColumn(batchID, "loaded_to_datastore", loadedToDatastore)
+
+    '''
+    Updates ODM status for this batch 
+    '''
+    def updateOdmStatus(self, batchID, odmStatus):
+
+        if odmStatus['LOADFAILED'] == 1: loadedToODM = -1
+        else: loadedToODM = odmStatus['LOADEDTOODM']
         sql = "UPDATE batch \
-               SET processed = " + str(processed) + " \
+               SET loaded_to_ODM = " + str(loadedToODM) + ", \
+               merge_worthy = " + str(odmStatus['MERGEWORTHY']) + ", \
+               merged = " + str(odmStatus['MERGED']) + " \
                WHERE batch_id = " + str(batchID)
 
@@ -242 +413 @@
 
     '''
-    Updates batch loaded_to_datastore field 
-    '''
-    def updateLoadedToDatastore(self, batchID, loadedToDatastore):
-
-        sql = "UPDATE batch \
-               SET loaded_to_datastore = " + str(loadedToDatastore) + " \
-               WHERE batch_id = " + str(batchID)
-
-        self.execute(sql)
-
-    '''
     Have we already processed and published this batch?
     '''
-    def alreadyProcessed(self, stage_id):
+    def alreadyProcessed(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 timestamp > '" + epoch + "' \
+               AND dvo_db = '" + dvoGpc1Label + "' \
+               AND processed = 1 \
+               AND loaded_to_datastore != 0"
 
         try:
@@ -267 +430 @@
             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) + " has already been published", "skipping")
                 return True
             else: 
@@ -273 +436 @@
         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, count):
+
+        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) >= count: 
+                self.logger.errorPair(str(stage_id) + " has failed %d times" % count, "skipping")
+                return True
+            else: 
+                return False
+        except:
+            self.logger.exception("Unable to check whether this batch has consistently failed")