Changeset 34041 for branches/meh_branches/ppstack_test/ippToPsps/jython/loader.py

Timestamp:

Jun 19, 2012, 5:24:19 PM (14 years ago)

Author:

mhuber

Message:

merging latest r34040 trunk changes to branch

Location:

branches/meh_branches/ppstack_test

Files:

• . (modified) (1 prop)
• ippToPsps/jython/loader.py (modified) (9 diffs)

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

@@ -16 +16 @@
 from datastore import Datastore
 from batch import Batch
-from dvo import Dvo
+from dvodetections import DvoDetections
 
 from initbatch import InitBatch
 from stackbatch import StackBatch
 from detectionbatch import DetectionBatch
+from objectbatch import ObjectBatch
 
 '''
@@ -31 +32 @@
     '''
     def __init__(self, argv):
-        super(Loader, self).__init__(argv)
+        super(Loader, self).__init__(argv, 1, 1)
 
         # create gpc1 database objects
@@ -44 +45 @@
                 self.scratchDb.disconnect()
                 continue
-            self.logger.infoPair("Using scratch Db", self.scratchDb.dbName)
             break
 
-        if not self.scratchDb.connected: 
-            self.logger.errorPair("Cannot connect to a", "scratch database")
-            raise Exception("No scratch Db")
-        
-        # now pass scratch Db to dvo object
-        self.dvo.setScratchDb(self.scratchDb.dbName)
+        if not self.scratchDb.connected:
+            self.exitProgram("Cannot connect to a scratch database")
+            raise
+       
+        # create Datastore objects
+        self.datastore = Datastore(self.logger, self.config, self.ippToPspsDb)
 
         # if an IN batch is requested, create and quit
         if len(sys.argv) > 2 and sys.argv[2] == "init":
-           batchID = self.ippToPspsDb.createNewBatch("IN", 0, self.config)
+           batchID = self.ippToPspsDb.createNewBatch("IN", 0)
            if batchID > 0:
                batch = InitBatch(self.logger, 
@@ -65 +65 @@
                        batchID)
                batch.run()
-    
+           else:
+               self.logger.errorPair("Could not create batch ID", "%d" % batchID)
+               self.exitProgram("failed to create init batch")
+               
            self.exitProgram("init batch created")
     
 
-        # set a poll time of about 1 minutes
+        # set a poll time of about 1 minute
         self.parsePollTimeArg("0.0166")
 
@@ -75 +78 @@
 
     '''
-    Overrides base-class version
+    Overrides base-class version so we can ensure our scratch Db is using the right DVO Db
     '''
     def refreshConfig(self):
 
         ret = super(Loader, self).refreshConfig()
-        self.dvo = Dvo(self.logger, self.config)
-        try: self.dvo.setScratchDb(self.scratchDb.dbName)
-        except: pass
-        self.datastore = Datastore(self.logger, self.config, self.ippToPspsDb)
+        try: self.scratchDb
+        except: return ret
+           
+        self.dvoDetections = DvoDetections(self.logger, self.config, self.scratchDb.dbName)
 
         return ret
 
     '''
+    Overrides base_class version so we can break out of processing loops if the config changes
+    '''
+    def checkClientStatus(self):
+
+        oldConfigName = self.config.name
+        super(Loader, self).checkClientStatus()
+        if oldConfigName != self.config.name:
+            self.logger.infoPair("Config changed",  "from '" + oldConfigName + "' to '" + self.config.name + "'")
+            self.config.printAll()
+            return False
+
+        return True
+
+    '''
     Main processing loop
     '''
     def run(self):
 
+        if not self.scratchDb.connected: return
+
+        numAttempts = 0
         while True:
 
-            boxIds = self.ippToPspsDb.getBoxIds(self.config.name)
-   
-            self.logger.infoPair("Boxes with unprocessed data", "%d" % len(boxIds))
-
-            for boxId in boxIds:
-      
-                self.checkClientStatus()
-
-                # get box info. if boxes have changed, break and start again
-                try:
-                    boxDim = self.ippToPspsDb.getBoxDimensions(boxId)
-                except:
-                   self.logger.infoPair("Can't get details for this box", "%d" % boxId)
-                   break
-                 
-                for batchType in self.config.batchTypes:
-                    
-                    ids = self.ippToPspsDb.getPendingIds(boxId, batchType)
-            
+            abort = False
+            for batchType in self.config.batchTypes:
+
+                # get a stripe's worth of box IDs
+                boxIds = self.ippToPspsDb.getStripeBoxIds(self.HOST, self.PID, batchType)
+
+                numAttempts += 1
+
+                self.logger.infoPair("Boxes with unprocessed data", "%d" % len(boxIds))
+    
+                if len(boxIds) > 0: numAttempts = 0
+
+                for boxId in boxIds:
+          
+                    if not self.checkClientStatus():
+                        abort = True
+                        break
+    
+                    # get box info. if boxes have changed, break and start again
+                    try:
+                        boxDim = self.ippToPspsDb.getBoxDimensions(boxId)
+                    except:
+                        self.logger.infoPair("Can't get details for this box", "%d" % boxId)
+                        break
+     
+                    ids = self.ippToPspsDb.getPendingIdsForThisBox(boxId, batchType)
+                
                     if len(ids) < 1: 
                         self.logger.debugPair("No " + batchType + " items found in this box", "skipping")
                         continue
-            
+                
                     self.logger.infoSeparator()
+                    self.logger.infoTitle("New box")
                     self.logger.infoPair("Box dimensions", "%.1f / %.1f / %.1f" % (boxDim['RA'], boxDim['DEC'], boxDim['SIDE']))
                     self.logger.infoPair(batchType + " items found in this box", "%d" % len(ids))
                     boxSizeWithBorder = boxDim['SIDE'] + (self.config.BORDER * 2)
-   
-                    # look in DVO for this box (with extra border)
-                    self.dvo.setSkyAreaAsBox(boxDim['RA'], boxDim['DEC'], boxSizeWithBorder)
-                    sizeToBeIngested = self.dvo.getDiskSizeOfRegionsToBeIngested()
-                    if sizeToBeIngested == 0.0: smfsPerGB = 999999999
-                    else: smfsPerGB = len(ids)/sizeToBeIngested
-                    self.logger.infoPair("DVO to be ingested", "%.1f GB" % sizeToBeIngested)
-                    self.logger.infoPair("smfs-per-GB", "%.1f" % smfsPerGB)
-            
-                    # should do we pre-ingest stuff from DVO?
-                    if smfsPerGB > 40:
-                        if not self.dvo.sync():
-                            self.logger.errorPair("Could not sync DVO with MySQL", "skipping")
-                            continue
      
-                        useFullTables = 1
-                    else: useFullTables = 0
-            
+                    useFullTables = 0
+                    if batchType != "OB":
+                        # look in DVO for this box (with extra border)
+                        self.logger.infoPair("Querying DVO for this sky area", "")
+                        self.dvoDetections.setSkyAreaAsBox(boxDim['RA'], boxDim['DEC'], boxSizeWithBorder)
+                        #self.dvoDetections.setSkyArea()
+                        sizeToBeIngested = self.dvoDetections.getDiskSizeOfRegionsToBeIngested()
+                        if sizeToBeIngested == 0.0: smfsPerGB = 999999999
+                        else: smfsPerGB = len(ids)/sizeToBeIngested
+                        self.logger.infoPair("DVO to be ingested", "%.1f GB" % sizeToBeIngested)
+                        self.logger.infoPair("smfs-per-GB", "%.1f" % smfsPerGB)
+                        # should do we pre-ingest stuff from DVO?
+                        if batchType == 'P2' and smfsPerGB > 30:
+                            if not self.dvoDetections.sync():
+                                self.logger.errorPair("Could not sync DVO with MySQL", "skipping")
+                                continue
+     
+                            useFullTables = 1
+                
                     self.logger.infoBool("Using pre-ingested DVO data?", useFullTables)
-                    self.processTheseItems(batchType, ids, useFullTables)
-
-            self.checkClientStatus()
-            if not self.waitForPollTime(): break
+                    if not self.processTheseItems(batchType, ids, useFullTables):
+                        abort = True
+                        break
+
+            if abort or not self.checkClientStatus(): abort = True
+            elif numAttempts > 1 and not self.waitForPollTime():  break
 
         
@@ -158 +191 @@
         for id in ids:
     
-            batchID = self.ippToPspsDb.createNewBatch(batchType, id, self.config)
+            batchID = self.ippToPspsDb.createNewBatch(batchType, id)
             
             if batchID < 0: 
@@ -165 +198 @@
     
             self.ippToPspsDb.unlockTables()
-    
-            if batchType == "P2":
-                batch = DetectionBatch(self.logger,
-                        self.config,
-                        self.gpc1Db,
-                        self.ippToPspsDb,
-                        self.scratchDb,
-                        id,
-                        batchID,
-                        useFullTables)
-            elif batchType == "ST":
-                batch = StackBatch(self.logger,
-                        self.config,
-                        self.gpc1Db,
-                        self.ippToPspsDb,
-                        self.scratchDb,
-                        id,
-                        batchID,
-                        useFullTables)
-    
-            batch.run()
-    
-            # check that we should continue
-            self.checkClientStatus()
+            self.ippToPspsDb.deletePendingItem(batchType, id)
+
+            # catch any raised exceptions in batch constructors
+            try:
+                if batchType == "P2":
+                    batch = DetectionBatch(self.logger,
+                            self.config,
+                            self.gpc1Db,
+                            self.ippToPspsDb,
+                            self.scratchDb,
+                            id,
+                            batchID,
+                            useFullTables)
+                elif batchType == "ST":
+                    batch = StackBatch(self.logger,
+                            self.config,
+                            self.gpc1Db,
+                            self.ippToPspsDb,
+                            self.scratchDb,
+                            id,
+                            batchID,
+                            useFullTables)
+    
+                elif batchType == "OB":
+                    batch = ObjectBatch(self.logger,
+                            self.config,
+                            self.gpc1Db,
+                            self.ippToPspsDb,
+                            self.scratchDb,
+                            id,
+                            batchID,
+                            useFullTables)
+    
+                batch.run()
+   
+            # if batch fails, ignore and move on to the next one
+            except: 
+                self.logger.errorPair("Problem with this %s batch (%d)" % (batchType, id), "skipping")
+                pass
+
+            if not self.checkClientStatus(): return False
     
             self.ippToPspsDb.lockBatchTable()
     
             self.logger.infoSeparator()
-    
-            # if in test mode, then quit after one batch
-            if self.config.test: break
     
         self.ippToPspsDb.unlockTables()
         self.logger.infoPair("Unattempted batches", "%d" % unattemptedCount)
+
+        return True
     
     '''
@@ -204 +253 @@
     '''
     def printUsage(self):
-        super(Cleanup, self).printUsage()
-        print " [init]"
+        super(Cleanup, self).printUsage("<[init]>")
 
     
@@ -214 +262 @@
     loader = Loader(sys.argv)
     loader.run()
-    loader.exitProgram("finished")
+    loader.exitProgram("completed")
 except Exception, e:
     print str(e)
-    traceback.print_exc() 
+    traceback.print_exc()
+