Changeset 35097 for trunk/ippToPsps/jython/loader.py

Timestamp:

Feb 6, 2013, 3:16:35 PM (13 years ago)

Author:

eugene

Message:

upgrade to ippToPsps (see doc/upgrade.txt): adds native dvo to mysql ingest operations, adds autogen configuration & installation, splits out global config information from new "skychunk" information (current region on the sky being processed), adds test suites

Location:

trunk/ippToPsps

Files:

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

trunk/ippToPsps/jython

@@ -1 +1 @@
 *.class
+Makefile
+Makefile.in

@@ -1 +1 @@
 *.class
+Makefile
+Makefile.in

trunk/ippToPsps/jython/loader.py

@@ -34 +34 @@
         super(Loader, self).__init__(argv, 1, 1)
 
-        # create gpc1 database objects
+        if self.skychunk.parallel:
+            print "PARALLEL dvo"
+        else: 
+            print "SERIAL dvo"
+
+        # connect to the gpc1 database
         self.gpc1Db = Gpc1Db(self.logger, self.config)
 
         # connect to scratch database
-        scratchDbs = ['ipptopsps_scratch', 'ipptopsps_scratch2', 'ipptopsps_scratch3']
-        for dbName in scratchDbs:
-            self.scratchDb = ScratchDb(self.logger, self.config, dbName)
-            if self.scratchDb.anyOtherConnections():
-                self.logger.errorPair("This scratch Db is already in use", dbName)
+        scratchDbs = ['1', '2', '3']
+        for dbVersion in scratchDbs:
+            self.scratchDb = ScratchDb(self.logger, self.config, dbVersion)
+            if not self.config.test and self.scratchDb.anyOtherConnections():
+                self.logger.errorPair("This scratch Db is already in use", self.scratchDb.dbName)
                 self.scratchDb.disconnect()
                 continue
@@ -52 +57 @@
        
         # create Datastore objects
-        self.datastore = Datastore(self.logger, self.config, self.ippToPspsDb)
+        self.datastore = Datastore(self.logger, self.skychunk, self.ippToPspsDb)
 
         # if an IN batch is requested, create and quit
@@ -60 +65 @@
                batch = InitBatch(self.logger, 
                        self.config, 
+                       self.skychunk, 
                        self.gpc1Db, 
                        self.ippToPspsDb, 
@@ -72 +78 @@
     
 
+        # if the 'once' option is passed, we do not loop multiple times in 'run'
+        self.onePassOnly = 0
+        if len(sys.argv) > 2 and sys.argv[2] == "once":
+           self.onePassOnly = 1
+
         # set a poll time of about 1 minute
         self.parsePollTimeArg("0.0166")
 
-        self.config.printAll()
+        self.skychunk.printAll()
 
     '''
     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()
+    def refreshSkychunk(self):
+
+        ret = super(Loader, self).refreshSkychunk()
         try: self.scratchDb
         except: return ret
            
-        self.dvoDetections = DvoDetections(self.logger, self.config, self.scratchDb.dbName)
+        self.dvoDetections = DvoDetections(self.logger, self.config, self.skychunk, self.ippToPspsDb, self.scratchDb.dbName)
 
         return ret
 
     '''
-    Overrides base_class version so we can break out of processing loops if the config changes
+    Overrides base_class version so we can break out of processing loops if the skychunk changes
     '''
     def checkClientStatus(self):
 
-        oldConfigName = self.config.name
+        oldSkychunkName = self.skychunk.name
         super(Loader, self).checkClientStatus()
-        if oldConfigName != self.config.name:
-            self.logger.infoPair("Config changed",  "from '" + oldConfigName + "' to '" + self.config.name + "'")
-            self.config.printAll()
+        if oldSkychunkName != self.skychunk.name:
+            self.logger.infoPair("Skychunk changed",  "from '" + oldSkychunkName + "' to '" + self.skychunk.name + "'")
+            self.skychunk.printAll()
             return False
 
@@ -115 +126 @@
 
             abort = False
-            for batchType in self.config.batchTypes:
+            for batchType in self.skychunk.batchTypes:
 
                 # get a stripe's worth of box IDs
@@ -141 +152 @@
                     ids = self.ippToPspsDb.getPendingIdsForThisBox(boxId, batchType)
                 
+                    # ids are the stage_ids for items to be processed (eg, stack_id for a stack CMF, cam_id for P2 smf)
+
                     if len(ids) < 1: 
                         self.logger.debugPair("No " + batchType + " items found in this box", "skipping")
@@ -149 +162 @@
                     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)
+                    boxSizeWithBorder = boxDim['SIDE'] + (self.skychunk.BORDER * 2)
+                    boxSizeSansBorder = boxDim['SIDE']
                     self.logger.infoPair("got here", "ok")
                     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
-     
+
+                        if (self.dvoDetections.useStilts):
+                            # 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", "%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 an
+                            # option. 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.dvoDetections.nativeIngestDetections(boxId, boxDim['RA'], boxDim['DEC'], boxSizeSansBorder)
+
                             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
+
+                    '''
+                    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):
@@ -175 +217 @@
                         break
                     self.logger.infoPair("processed","ok")
+
             if abort or not self.checkClientStatus(): abort = True
             elif numAttempts > 1 and not self.waitForPollTime():  break
 
+            if self.onePassOnly: self.exitProgram("one pass completed")
         
     '''
@@ -210 +254 @@
                     batch = DetectionBatch(self.logger,
                             self.config,
+                            self.skychunk,
                             self.gpc1Db,
                             self.ippToPspsDb,
@@ -219 +264 @@
                     batch = StackBatch(self.logger,
                             self.config,
+                            self.skychunk,
                             self.gpc1Db,
                             self.ippToPspsDb,
@@ -229 +275 @@
                     batch = ObjectBatch(self.logger,
                             self.config,
+                            self.skychunk,
                             self.gpc1Db,
                             self.ippToPspsDb,
@@ -260 +307 @@
     '''
     def printUsage(self):
-        super(Cleanup, self).printUsage("<[init]>")
+        super(Loader, self).printUsage("<[init]>")
 
     

@@ -1 +1 @@
 *.class
+Makefile
+Makefile.in