Changeset 37403 for branches/eam_branches/ps2-tc3-20130727/ippToPsps/jython/loader.py

Timestamp:

Sep 19, 2014, 4:05:27 PM (12 years ago)

Author:

eugene

Message:

merge changes from trunk

Location:

branches/eam_branches/ps2-tc3-20130727

Files:

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

branches/eam_branches/ps2-tc3-20130727/ippToPsps/jython/loader.py

@@ -78 +78 @@
            self.onePassOnly = 1
 
-        # set a poll time of about 1 minute
-        self.parsePollTimeArg("0.0166")
+        if self.config.loopBatch:
+            self.parsePollTimeArg("0.0166")
+        else:
+            self.parsePollTimeArg("0.0")
 
         self.skychunk.printAll()
@@ -160 +162 @@
                     self.logger.infoPair(batchType + " items found in this box", "%d" % len(ids))
 
-                    useFullTables = 0
                     if batchType != "OB":
-
-                        if (self.dvoDetections.useStilts):
-                            # look in DVO for this box (with extra border)
-                            self.logger.infoPair("Querying DVO for this sky area", "")
-
-                            # XXXX EAM : This code should not be used anymore
-                            boxSize = 0.0
-                            self.dvoDetections.setSkyAreaAsBox(boxDim['RA'], boxDim['DEC'], boxSize)
-                            #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", "%7.1e GB" % sizeToBeIngested)
-                            self.logger.infoPair("smfs-per-GB", "%.1f" % smfsPerGB)
-
-                            # should do we pre-ingest stuff from DVO?
-                            # NOTE EAM : this skychunk loads the dvo detections into the mysql db
-                            # XXXX EAM : this should happen for both P2 and Stack detection
-                            # XXXX EAM : in parallel model, this should happen for all P2 & Stack batches
-                            # 
-                            if (batchType == 'P2' or batchType == 'ST') and smfsPerGB > 30:
-                                if not self.dvoDetections.sync():
-                                    self.logger.errorPair("Could not sync DVO with MySQL", "skipping")
-                                    continue
-                                
-                                useFullTables = 1
-                        else:
-                            # XXX EAM : this is not currently optional
-                            # (it is the only path). either remove the
-                            # non-native ingest code or make it an
-                            # option if we are using the native
-                            # loader, always use it need to work out a
-                            # good boundary / region strategy in
-                            # coordination with impact of mysql
-                            # insertion
-                            if not self.ippToPspsDb.isBoxIngested(boxId, self.scratchDb.dbHost):
-                                if not self.dvoDetections.nativeIngestDetections(boxId, boxDim):
-                                    self.logger.error("Unable to ingest detections with nativeIngest")
-                                    return False
-
-                            useFullTables = 1
-                
-                    ## else:
-                    ##     # if we are using the native loader, always use it
-                    ##     # need to work out a good boundary / region strategy in coordination with 
-                    ##     # impact of mysql insertion
-                    ##     self.dvoObjects.nativeIngestObjects(boxDim['RA'], boxDim['DEC'], boxSizeSansBorder)
-                    ##     useFullTables = 1
+                        if self.config.resetDvo or not self.ippToPspsDb.isBoxIngested(boxId, self.scratchDb.dbHost):
+                            if not self.dvoDetections.nativeIngestDetections(boxId, boxDim):
+                                self.logger.error("Unable to ingest detections with nativeIngest")
+                                return False
 
                     '''
                     NOTE EAM : "TheseItems" refers to each of the stack CMFs, camera SMFs, or object batches
                     '''
-                    self.logger.infoBool("Using pre-ingested DVO data?", useFullTables)
-                    if not self.processTheseItems(batchType, ids, useFullTables):
+                    if not self.processTheseItems(batchType, ids):
                         abort = True
                         break
                     self.logger.infoPair("processed","ok")
 
+            self.logger.infoPair("done with processing boxes, numAttempts: ", str(numAttempts))
             if abort or not self.checkClientStatus(): abort = True
             elif numAttempts > 1 and not self.waitForPollTime():  break
@@ -228 +185 @@
     Actually loops through items and processes them
     '''
-    def processTheseItems(self, batchType, ids, useFullTables):
+    def processTheseItems(self, batchType, ids):
     
         self.logger.infoPair("%s items queued" % batchType, "%d" % len(ids))
@@ -234 +191 @@
         # loop round IDs of all items to be processed
         self.ippToPspsDb.lockBatchTable()
-        self.logger.infoPair("heather:","lockbatchtable")
+        #self.logger.infoPair("heather:","lockbatchtable")
         unattemptedCount = 0
         for id in ids:
-            self.logger.infoPair("heather:","in ids")
-            self.logger.infoPair("heather:id",str(id))
+            #self.logger.infoPair("heather:","in ids")
+            #self.logger.infoPair("heather:id",str(id))
             batchID = self.ippToPspsDb.createNewBatch(batchType, id)
-            self.logger.infoPair("heather:","creatednewbatch")
-            self.logger.infoPair("heather:batchId", str(batchID))
+            #self.logger.infoPair("heather:","creatednewbatch")
+            #self.logger.infoPair("heather:batchId", str(batchID))
             if batchID < 0: 
                 unattemptedCount += 1
                 continue
-            self.logger.infoPair("heather:","passed logic")
+            #self.logger.infoPair("heather:","passed logic")
+
             self.ippToPspsDb.unlockTables()
-            self.logger.infoPair("heather:","unlock table")
+            #self.logger.infoPair("heather:","unlock table")
+
             self.ippToPspsDb.deletePendingItem(batchType, id)
-            self.logger.infoPair("heather:","deleted pending item")
+            #self.logger.infoPair("heather:","deleted pending item")
 
             # catch any raised exceptions in batch constructors
             try:
                 if batchType == "P2":
+                    self.logger.infoPair("defining P2 batch for cam_id:",id)
                     batch = DetectionBatch(self.logger,
                             self.config,
@@ -261 +221 @@
                             self.scratchDb,
                             id,
-                            batchID,
-                            useFullTables)
+                            batchID)
                 elif batchType == "ST":
+                    self.logger.infoPair("defining ST batch for sky_id:",id)
                     batch = StackBatch(self.logger,
                             self.config,
@@ -271 +231 @@
                             self.scratchDb,
                             id,
-                            batchID,
-                            useFullTables)
+                            batchID)
     
                 elif batchType == "OB":
@@ -282 +241 @@
                             self.scratchDb,
                             id,
-                            batchID,
-                            useFullTables)
-    
+                            batchID)
+    
+                elif batchType == "DF":
+                    batch = DiffBatch(self.logger,
+                            self.config,
+                            self.skychunk,
+                            self.gpc1Db,
+                            self.ippToPspsDb,
+                            self.scratchDb,
+                            id,
+                            batchID)
+    
+                elif batchType == "DO":
+                    batch = DiffObjectBatch(self.logger,
+                            self.config,
+                            self.skychunk,
+                            self.gpc1Db,
+                            self.ippToPspsDb,
+                            self.scratchDb,
+                            id,
+                            batchID)
+    
+                elif batchType == "FW":
+                    batch = ForcedWarpBatch(self.logger,
+                            self.config,
+                            self.skychunk,
+                            self.gpc1Db,
+                            self.ippToPspsDb,
+                            self.scratchDb,
+                            id,
+                            batchID)
+    
+                elif batchType == "FO":
+                    batch = ForcedObjectBatch(self.logger,
+                            self.config,
+                            self.skychunk,
+                            self.gpc1Db,
+                            self.ippToPspsDb,
+                            self.scratchDb,
+                            id,
+                            batchID)
+                    
                 batch.run()
    
@@ -291 +289 @@
                 self.logger.error("EXCEPTION: " + str(e))
                 self.logger.errorPair("Problem with this %s batch (%d)" % (batchType, id), "skipping")
+                sys.exit(1)
                 pass