Changeset 33489
- Timestamp:
- Mar 13, 2012, 9:49:54 AM (14 years ago)
- File:
-
- 1 edited
-
trunk/ippToPsps/jython/queue.py (modified) (2 diffs)
Legend:
- Unmodified
- Added
- Removed
-
trunk/ippToPsps/jython/queue.py
r33347 r33489 28 28 # create various objects 29 29 self.gpc1Db = Gpc1Db(self.logger, self.config) 30 self.datastore = Datastore(self.logger, self.config, self.ippToPspsDb) 31 32 self.config.printAll() 30 33 31 34 # set a poll time in hours … … 35 38 ''' 36 39 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 processing40 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 41 by one or more loading clients that may be running on any host 39 42 ''' 40 43 def run(self): 41 44 42 # this outer while loop simply waits a few minutes then starts queuing again as more stuff may appear in DVO over time45 # this outer while loop simply waits a few minutes then starts queuing again (as more stuff may appear in DVO over time) 43 46 while True: 47 48 self.ippToPspsDb.removeAllBoxes() 44 49 45 50 # 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) 51 for batchType in self.config.batchTypes: 52 self.logger.infoTitle("Previous failed datastore loads") 53 self.publishAnyUnpublishedBatches(batchType) 48 54 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)) 55 # get totals for whole area to check if there is anything to do 56 processedIDs = self.ippToPspsDb.getProcessedIDs(batchType, self.config.epoch, self.config.dvoLabel) 57 consistantlyFailedIDs = self.ippToPspsDb.getConsistentlyFailedIDs(batchType, self.config.epoch, self.config.dvoLabel) 58 rows = self.gpc1Db.getItemsInThisDVODbForThisStage( 59 self.config.dvoLabel, 60 batchType, 61 self.config.minRa, 62 self.config.maxRa, 63 self.config.minDec, 64 self.config.maxDec) 55 65 56 # loop through full range of RA/Dec queueing stuff in boxes of size self.config.boxSize 57 self.logger.info("+---------------------------------------------------------------------+") 58 self.logger.info("| All unprocessed items in area: %12d |" % len(ids)) 66 # first report total stuff 67 allIDs = [] 68 for row in rows: 69 try: allIDs.append(row[0]) 70 except: pass 59 71 60 if len(ids) > 0: 72 ids = list(set(allIDs) - set(processedIDs) - set(consistantlyFailedIDs)) 73 self.logger.infoPair("All unprocessed items", "%d" % len(ids)) 61 74 62 self.logger.info("+-------------+-------------+-------------+-------------+-------------+") 63 self.logger.info("| RA | Dec | box size | in DVO | unprocessed |") 64 self.logger.info("+-------------+-------------+-------------+-------------+-------------+") 75 # now trim full list down to only pending stuff 76 pending = [[], [], []] 77 for row in rows: 78 try: 79 findId = row[0] 80 except: continue 81 if findId in ids: pending.append([row[0], row[1], row[2]]) 82 83 # store all items in temp table 84 self.logger.infoPair("Inserting all pending items into", "ippToPsps database") 85 self.ippToPspsDb.storeAllItems(pending) 86 87 # loop through full range of RA/Dec queueing stuff in boxes of size self.config.boxSize 88 if len(ids) > 0: 89 90 self.logger.info("+-------------+-------------+-------------+-------------+") 91 self.logger.info("| RA | Dec | box size | unprocessed |") 92 self.logger.info("+-------------+-------------+-------------+-------------+") 65 93 66 # starting positions67 ra = self.config.minRa + self.config.halfBox68 dec = self.config.minDec + self.config.halfBox94 # starting positions 95 ra = self.config.minRa + self.config.halfBox 96 dec = self.config.minDec + self.config.halfBox 69 97 70 while ra <= self.config.maxRa:71 while dec <= self.config.maxDec:98 while ra <= self.config.maxRa: 99 while dec <= self.config.maxDec: 72 100 73 self.checkClientStatus()74 box_id = self.ippToPspsDb.insertBox(self.config.name, ra, dec, self.config.boxSize)101 self.checkClientStatus() 102 box_id = self.ippToPspsDb.insertBox(ra, dec) 75 103 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) 104 ids = self.ippToPspsDb.getItemsInThisThisBox(ra, dec) 85 105 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 |" % ( 106 self.logger.info("| %5.1f | %5.1f | %9d | %9d |" % ( 97 107 ra, 98 108 dec, 99 109 self.config.boxSize, 100 len(allIDs),101 110 len(ids))) 102 if len(ids) < 1: continue 111 112 if len(ids) > 0: self.ippToPspsDb.insertPending(box_id, batchType, ids) 113 114 dec = dec + self.config.boxSize 115 dec = self.config.minDec + self.config.halfBox 116 ra = ra + self.config.boxSize 103 117 104 #self.logger.infoPair("write IDs to database", "NOW") 105 self.ippToPspsDb.insertPending(box_id, batchType, ids) 106 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("+-------------+-------------+-------------+-------------+-------------+") 118 self.logger.info("+-------------+-------------+-------------+-------------+") 112 119 113 120 if not self.waitForPollTime(): break
Note:
See TracChangeset
for help on using the changeset viewer.
