- Timestamp:
- Jun 19, 2012, 5:24:19 PM (14 years ago)
- Location:
- branches/meh_branches/ppstack_test
- Files:
-
- 2 edited
-
. (modified) (1 prop)
-
ippToPsps/jython/queue.py (modified) (4 diffs)
Legend:
- Unmodified
- Added
- Removed
-
branches/meh_branches/ppstack_test
- Property svn:mergeinfo changed
-
branches/meh_branches/ppstack_test/ippToPsps/jython/queue.py
r33415 r34041 13 13 from gpc1db import Gpc1Db 14 14 from datastore import Datastore 15 from dvoobjects import DvoObjects 15 16 from batch import Batch 16 17 … … 28 29 # create various objects 29 30 self.gpc1Db = Gpc1Db(self.logger, self.config) 31 self.datastore = Datastore(self.logger, self.config, self.ippToPspsDb) 32 try: 33 self.dvoObjects = DvoObjects(self.logger, self.config) 34 except: 35 self.exitProgram("Unable to create instance of DvoObject") 36 raise 30 37 31 # set a poll time in hours 32 self.parsePollTimeArg("12") 38 self.config.printAll() 39 40 if len(argv) > 2: self.parsePollTimeArg(sys.argv[2]) 33 41 34 42 35 43 ''' 36 44 Main processing loop. 37 Tile s areadefined in config and looks for available stage_ids for each tile and writes them to ippToPsps database ready for processing45 Tiled area is defined in config and looks for available stage_ids for each tile and writes them to ippToPsps database ready for processing 38 46 by one or more loading clients that may be running on any host 39 47 ''' 40 48 def run(self): 41 49 42 # this outer while loop simply waits a few minutes then starts queuing again as more stuff may appear in DVO over time50 # this outer while loop simply waits a few minutes then starts queuing again (as more stuff may appear in DVO over time) 43 51 while True: 52 53 self.ippToPspsDb.removeAllBoxes() 44 54 45 55 # queue up batches that are processed but not loaded to datastore 46 self.logger.infoTitle("Previous failed datastore loads") 47 for batchType in self.config.batchTypes: self.publishAnyUnpublishedBatches(batchType) 56 for batchType in self.config.batchTypes: 48 57 49 # get totals for whole area to check if there is anything to do 50 processedIDs = self.ippToPspsDb.getProcessedIDs(batchType, self.config.epoch, self.config.dvoLabel) 51 consistantlyFailedIDs = self.ippToPspsDb.getConsistentlyFailedIDs(batchType, self.config.epoch, self.config.dvoLabel) 52 allIDs = self.gpc1Db.getIDsInThisDVODbForThisStage(self.config.dvoLabel, batchType, self.config.minRa, self.config.maxRa, self.config.minDec, self.config.maxDec) 53 #allIDs = self.gpc1Db.getIDsInThisDVODbForThisStage(self.config.dvoLabel, batchType) 54 ids = list(set(allIDs) - set(processedIDs) - set(consistantlyFailedIDs)) 58 self.logger.infoTitle("Previous failed datastore loads") 59 self.publishAnyUnpublishedBatches(batchType) 55 60 56 # loop through full range of RA/Dec queueing stuff in boxes of size self.config.boxSize57 self.logger.info("+---------------------------------------------------------------------+")58 self.logger.info("| All unprocessed items in area: %12d |" % len(ids))61 # get totals for whole area to check if there is anything to do 62 processedIDs = self.ippToPspsDb.getProcessedIDs(batchType) 63 consistantlyFailedIDs = self.ippToPspsDb.getConsistentlyFailedIDs(batchType) 59 64 60 if len(ids) > 0: 65 # for object batches, get full list of stuff from Dvo 66 if batchType == "OB": 67 self.dvoObjects.setSkyArea( 68 self.config.minRa, 69 self.config.maxRa, 70 self.config.minDec, 71 self.config.maxDec) 72 rows = self.dvoObjects.allPopulatedRegionInfo 73 self.dvoObjects.printSummary() 61 74 62 self.logger.info("+-------------+-------------+-------------+-------------+-------------+") 63 self.logger.info("| RA | Dec | box size | in DVO | unprocessed |") 64 self.logger.info("+-------------+-------------+-------------+-------------+-------------+") 75 # for other batches, info comes from gpc1 database 76 else: 77 rows = self.gpc1Db.getItemsInThisDVODbForThisStage( 78 self.config.dvoLabel, 79 batchType, 80 self.config.minRa, 81 self.config.maxRa, 82 self.config.minDec, 83 self.config.maxDec) 84 85 # first report total stuff 86 allIDs = [] 87 for row in rows: 88 try: allIDs.append(row[0]) 89 except: pass 90 91 ids = list(set(allIDs) - set(processedIDs) - set(consistantlyFailedIDs)) 92 self.logger.infoPair("All unprocessed items", "%d" % len(ids)) 93 94 # now trim full list down to only pending stuff 95 pending = [[], [], []] 96 for row in rows: 97 try: 98 findId = row[0] 99 except: continue 100 if findId in ids: pending.append([row[0], row[1], row[2]]) 101 102 # store all items in temp table 103 self.logger.infoPair("Inserting all pending items into", "ippToPsps database") 104 self.ippToPspsDb.storeAllItems(pending) 105 106 # loop through full range of RA/Dec queueing stuff in boxes of size self.config.boxSize 107 if len(ids) > 0: 108 109 self.logger.info("+-------------+-------------+-------------+-------------+") 110 self.logger.info("| RA | Dec | box size | unprocessed |") 111 self.logger.info("+-------------+-------------+-------------+-------------+") 65 112 66 # starting positions67 ra = self.config.minRa + self.config.halfBox68 dec = self.config.minDec + self.config.halfBox113 # starting positions 114 ra = self.config.minRa + self.config.halfBox 115 dec = self.config.minDec + self.config.halfBox 69 116 70 while ra <= self.config.maxRa:71 while dec <= self.config.maxDec:117 while ra <= self.config.maxRa: 118 while dec <= self.config.maxDec: 72 119 73 self.checkClientStatus() 74 box_id = self.ippToPspsDb.insertBox(self.config.name, ra, dec, self.config.boxSize) 120 box_id = self.ippToPspsDb.insertBox(ra, dec) 75 121 76 # for each batch type look for items to queue in this box 77 for batchType in self.config.batchTypes: 78 79 allIDs = self.gpc1Db.getIDsInThisDVODbForThisStageInThisBox( 80 self.config.dvoLabel, 81 batchType, 82 ra, 83 dec, 84 self.config.boxSize) 122 ids = self.ippToPspsDb.getItemsInThisThisBox(ra, dec) 85 123 86 if len(allIDs) < 1: 87 self.logger.info("| %5.1f | %5.1f | %9d | %9d | %9d |" % ( 88 ra, 89 dec, 90 self.config.boxSize, 91 len(allIDs), 92 0)) 93 continue 94 95 ids = list(set(allIDs) - set(processedIDs) - set(consistantlyFailedIDs)) 96 self.logger.info("| %5.1f | %5.1f | %9d | %9d | %9d |" % ( 124 self.logger.info("| %5.1f | %5.1f | %9d | %9d |" % ( 97 125 ra, 98 126 dec, 99 127 self.config.boxSize, 100 len(allIDs),101 128 len(ids))) 102 if len(ids) < 1: continue 129 130 if len(ids) > 0: self.ippToPspsDb.insertPending(box_id, batchType, ids) 131 132 dec = dec + self.config.boxSize 133 dec = self.config.minDec + self.config.halfBox 134 ra = ra + self.config.boxSize 103 135 104 #self.logger.infoPair("write IDs to database", "NOW") 105 self.ippToPspsDb.insertPending(box_id, batchType, ids) 136 self.logger.info("+-------------+-------------+-------------+-------------+") 106 137 107 dec = dec + self.config.boxSize 108 dec = self.config.minDec + self.config.halfBox 109 ra = ra + self.config.boxSize 110 111 self.logger.info("+-------------+-------------+-------------+-------------+-------------+") 112 138 self.checkClientStatus() 113 139 if not self.waitForPollTime(): break 114 140 … … 118 144 def publishAnyUnpublishedBatches(self, batchType): 119 145 120 batchIDs = self.ippToPspsDb.getProcessedButFailedDatastoreBatchIDs( self.config.epoch, self.config.dvoLabel,batchType)146 batchIDs = self.ippToPspsDb.getProcessedButFailedDatastoreBatchIDs(batchType) 121 147 self.logger.infoPair("%s batches" % batchType, "%d" % len(batchIDs)) 122 148 … … 129 155 130 156 Batch.publishToDatastore(self.datastore, batchID, batchName, subDir, tarballFile) 131 157 158 159 ''' 160 Overrides base-class version 161 ''' 162 def printUsage(self): 163 super(Queue, self).printUsage("[<repeat time in hours>]") 164 165 132 166 ''' 133 167 Start of program. 134 168 ''' 135 queue = Queue(sys.argv) 136 queue.run() 137 queue.exitProgram("finished") 169 try: 170 queue = Queue(sys.argv) 171 queue.run() 172 queue.exitProgram("completed") 173 except: pass 174
Note:
See TracChangeset
for help on using the changeset viewer.
