Changeset 33237 for trunk/ippToPsps

Timestamp:

Feb 10, 2012, 1:43:48 PM (14 years ago)

Author:

rhenders

Message:

big changes to incorporate loading by region of sky

File:

• trunk/ippToPsps/jython/load.py (modified) (10 diffs)

trunk/ippToPsps/jython/load.py

@@ -17 +17 @@
 from datastore import Datastore
 from batch import Batch
+from dvo import Dvo
 
 from initbatch import InitBatch
@@ -40 +41 @@
 
 '''
-Queues up all available stuff for this batch type
-'''
-def queueItems(batchType):
+Queues up all available stuff for this batch type in a box with these dimensions
+'''
+def queueItemsInBox(batchType, ra, dec, boxsize):
 
     logger.infoTitle("Queuing " + batchType + " items")
+    logger.infoPair("Box center and size", "%f/%f/%f" %(ra, dec, boxsize))
 
     allIDs = []
@@ -60 +62 @@
     else:
 
-        allIDs = gpc1Db.getIDsInThisDVODbForThisStage(
+        allIDs = gpc1Db.getIDsInThisDVODbForThisStageInThisBox(
                 DVOGPC1LABEL, 
                 batchType, 
-                MINRA, 
-                MAXRA, 
-                MINDEC, 
-                MAXDEC)
+                ra, 
+                dec, 
+                boxsize)
 
         logger.infoPair("All %s items in DVO" % batchType, "%d" % len(allIDs))
@@ -81 +82 @@
         logger.infoPair("Consistently failing items", "%d" % len(consistaentlyFailedIDs))
         logger.infoPair("Unprocessed items", "%d" % len(ids))
+
+    return ids
+
+
+'''
+Actually loops through items and processes them
+'''
+def processTheseItems(batchType, ids, useFullTables):
 
     logger.infoPair("%s items queued" % batchType, "%d" % len(ids))
@@ -108 +117 @@
                     ippToPspsDb,
                     id,
-                    batchID)
+                    batchID,
+                    useFullTables)
         elif batchType == "ST":
             batch = StackBatch(logger,
@@ -116 +126 @@
                     ippToPspsDb,
                     id,
-                    batchID)
+                    batchID,
+                    useFullTables)
 
         batch.run()
@@ -132 +143 @@
 Start of program.
 '''
-
 if len(sys.argv) > 1: CONFIG = sys.argv[1]
 else:
@@ -156 +166 @@
 
 # if in test mode, print log to screen, otherwise, only to file
-logger.setup(configDoc, "ippToPsps", TEST, not TEST, HOST, PID)
-
-# create database objects
+#logger.setup(configDoc, "ippToPsps", TEST, not TEST, HOST, PID) TODO put back
+logger.setup(configDoc, "ippToPsps")
+
+# create various objects
+dvo = Dvo(logger, configDoc)
 gpc1Db = Gpc1Db(logger, configDoc)
 ippToPspsDb = IppToPspsDb(logger, configDoc)
@@ -174 +186 @@
 PUBLISH = int(configDoc.find("options/publishToDatastore").text)
 BASEPATH = configDoc.find("localOutPath").text
-QUEUE_P2 = int(configDoc.find("options/queueP2").text)
-QUEUE_ST = int(configDoc.find("options/queueST").text)
 SURVEY = configDoc.find("options/survey").text
 
+# get batch types from config
+batchTypes = []
+if int(configDoc.find("options/queueP2").text) == 1: batchTypes.append("P2")
+if int(configDoc.find("options/queueST").text) == 1: batchTypes.append("ST")
+
 # get equatorial coord limits, if any
-try:
-    MINRA = float(configDoc.find("dvo/minRA").text)
-    MAXRA = float(configDoc.find("dvo/maxRA").text)
-    MINDEC = float(configDoc.find("dvo/minDec").text)
-    MAXDEC = float(configDoc.find("dvo/maxDec").text)
-except:
-    MINRA = 0.0
-    MAXRA = 360.0
-    MINDEC = -30.0
-    MAXDEC = 90.0
+try: MINRA = float(configDoc.find("dvo/minRA").text)
+except: MINRA = 0.0
+try: MAXRA = float(configDoc.find("dvo/maxRA").text)
+except: MAXRA = 360.0
+try: MINDEC = float(configDoc.find("dvo/minDec").text)
+except: MINDEC = -30.0 
+try: MAXDEC = float(configDoc.find("dvo/maxDec").text)
+except: MAXDEC = 90.0
+try: BOXSIZE = float(configDoc.find("dvo/boxSize").text)
+except: BOXSIZE = 4.0
 
 # prompt user: FORCE and PUBLISH is a dangerous combination
@@ -206 +221 @@
 logger.infoPair("Loading epoch", EPOCH)
 logger.infoPair("DVO gpc1 label", DVOGPC1LABEL)
-logger.infoBool("Queuing P2 batches?", QUEUE_P2)
-logger.infoBool("Queuing ST batches?", QUEUE_ST)
+for batchType in batchTypes: logger.infoPair("Queuing batch type", batchType)
 logger.infoBool("Forcing?", FORCE)
 logger.infoBool("Test mode?", TEST)
 logger.infoBool("Publishing?", PUBLISH)
 
+# if an IN batch is requested, create and quit
+if QUEUE_IN:
+   batchID = ippToPspsDb.createNewBatch(
+             "IN",
+             0,
+             SURVEY,
+             EPOCH,
+             DVOGPC1LABEL,
+             datastore.product,
+             1)
+   if batchID > 0:
+       batch = InitBatch(logger, CONFIG, configDoc, gpc1Db, ippToPspsDb, batchID)
+       batch.run()
+
+   sys.exit(1)
+
 '''
 Main processing loop:
 
- 1) creates an IN batch if requested
- 2) queues available P2 batches (if requested)
- 3) queues available ST batches (if requested)
- 4) if in test mode, quits
- 5) otherwise, waits then repeats from 2
-
-'''
+ 1) queues available P2 batches (if requested)
+ 2) queues available ST batches (if requested)
+ 3) if in test mode, quits
+ 4) else waits then repeats from 2
+
+'''
+BORDER = 1.60
+HALFBOX = BOXSIZE/2.0
+BOXSIZEWITHBORDER = BOXSIZE + (BORDER * 2)
+
+# this outer while loop simply waits a few minutes then starts queuing againas more stuff may appear in DVO over time
 while True:
 
-    # queue all items for all batch types, sequentially
-    if QUEUE_IN:
-       batchID = ippToPspsDb.createNewBatch(
-                 "IN",
-                 0,
-                 SURVEY,
-                 EPOCH,
-                 DVOGPC1LABEL,
-                 datastore.product,
-                 1)
-       if batchID > 0:
-           batch = InitBatch(logger, CONFIG, configDoc, gpc1Db, ippToPspsDb, batchID)
-           batch.run()
-       break;
+    # starting positions
+    ra = MINRA + HALFBOX
+    dec = MINDEC + HALFBOX
 
     # queue up batches that are processed but not loaded to datastore
     logger.infoTitle("Previous failed datastore loads")
-    publishAnyUnpublishedBatches("P2")
-    publishAnyUnpublishedBatches("ST")
-
-    if QUEUE_P2: queueItems("P2")
-    if QUEUE_ST: queueItems("ST")
-
-    # in the TEST mode, we quit after submitting one batch
-    if TEST: break
+    for batchType in batchTypes: publishAnyUnpublishedBatches(batchType)
+
+    # loop through full range of RA/Dec queueing stuff in boxes of size BOXSIZE
+    while ra <= MAXRA:
+
+       while dec <= MAXDEC:
+
+           # for each batch type
+           # - queue items in this box
+           # - check if we should pre-load DVO region
+           # - process the items
+           for batchType in batchTypes: 
+               ids = queueItemsInBox(batchType, ra, dec, BOXSIZE)
+
+               if len(ids) < 1: 
+                   logger.infoPair("No items found in this region", "skipping")
+                   continue
+
+               dvo.setSkyAreaAsBox(ra, dec, BOXSIZEWITHBORDER)
+               dvo.printSummary()
+
+               sizeToBeIngested = dvo.getDiskSizeOfRegionsToBeIngested()
+               if sizeToBeIngested == 0.0: smfsPerGB = 999999999
+               else: smfsPerGB = len(ids)/dvo.getDiskSizeOfRegionsToBeIngested()
+               logger.infoPair("smfs-per-GB", "%f" % smfsPerGB)
+
+               # do we pre-ingest stuff from DVO?
+               if smfsPerGB > 40:
+                   if not dvo.sync():
+                       logger.errorPair("Could not sync DVO with MySQL", "skipping")
+                       continue
+
+                   useFullTables = 1
+               else: useFullTables = 0
+
+               logger.infoBool("Using pre-ingested DVO data?", useFullTables)
+               processTheseItems(batchType, ids, useFullTables)
+
+               # in the TEST mode, we quit after submitting one batch
+               if TEST: break
+
+           dec = dec + BOXSIZE
+           if TEST: break
+
+       dec = MINDEC + HALFBOX
+       ra = ra + BOXSIZE
+       if TEST: break
 
     # wait for the POLLPERIOD before checking for new ids