Index: trunk/ippToPsps/jython/load.py =================================================================== --- trunk/ippToPsps/jython/load.py (revision 33345) +++ (revision ) @@ -1,300 +1,0 @@ -#!/usr/bin/env jython - -# -# The main loading program for ippToPsps. See usage below. -# - -import signal -import time -import sys -import os -import logging.config - -from config import Config -from pslogger import PSLogger -from gpc1db import Gpc1Db -from ipptopspsdb import IppToPspsDb -from datastore import Datastore -from batch import Batch -from dvo import Dvo - -from initbatch import InitBatch -from stackbatch import StackBatch -from detectionbatch import DetectionBatch - -''' -Finds batches that processed ok, but did not get loaded to the datastore, then tries to load them -''' -def publishAnyUnpublishedBatches(batchType): - - batchIDs = ippToPspsDb.getProcessedButFailedDatastoreBatchIDs(config.epoch, config.dvoLabel, batchType) - logger.infoPair("%s batches" % batchType, "%d" % len(batchIDs)) - - for batchID in batchIDs: - - batchName = Batch.getNameFromID(batchID) - subDir = Batch.getSubDir(config.basePath, batchType, config.dvoLabel) - tarballFile = Batch.getTarballFile(batchID) - logger.infoPair("Batch name", batchName) - - Batch.publishToDatastore(datastore, batchID, batchName, subDir, tarballFile) - -''' -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 = config.ids - - if len(allIDs) > 0: - logger.infoPair("Using config IDs (not DVO)", "%d" % len(allIDs)) - # no config dictated IDs, so get from DVO instead - else: - - allIDs = gpc1Db.getIDsInThisDVODbForThisStageInThisBox( - config.dvoLabel, - batchType, - ra, - dec, - boxsize) - - logger.infoPair("All %s items in DVO" % batchType, "%d" % len(allIDs)) - - # if in force mode, then queue full list - if config.force: ids = allIDs - - # if not in test mode, then only queue un-processed items - else: - processedIDs = ippToPspsDb.getProcessedIDs(batchType, config.epoch, config.dvoLabel) - consistaentlyFailedIDs = ippToPspsDb.getConsistentlyFailedIDs(batchType, config.epoch, config.dvoLabel) - ids = list(set(allIDs) - set(processedIDs) - set(consistaentlyFailedIDs)) - logger.infoPair("Processed items", "%d" % len(processedIDs)) - 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)) - - # loop round IDs of all items to be processed - ippToPspsDb.lockBatchTable() - for id in ids: - - batchID = ippToPspsDb.createNewBatch(batchType, id, config) - - if batchID < 0: continue; - - ippToPspsDb.unlockTables() - - if batchType == "P2": - batch = DetectionBatch(logger, - config, - gpc1Db, - ippToPspsDb, - id, - batchID, - useFullTables) - elif batchType == "ST": - batch = StackBatch(logger, - config, - gpc1Db, - ippToPspsDb, - id, - batchID, - useFullTables) - - batch.run() - - # check that we should continue - checkClientStatus() - - ippToPspsDb.lockBatchTable() - - logger.infoSeparator() - - # if in test mode, then quit after one batch - if config.test: break - - ippToPspsDb.unlockTables() - -''' -Handler for Ctrl-C signal -''' -def signal_handler(signal, frame): - - exitProgram("Ctrl-C") - -''' -Cleanly exits program, prints to log the way it exited -''' -def exitProgram(exitType): - - logger.infoPair("Program exited", exitType) - ippToPspsDb.removeClient(logger.host, logger.pid) - sys.exit(0) - - -''' -Checks if we should pause or kill loading, otherwise updates the clients table with current time -''' -def checkClientStatus(): - - ippToPspsDb.updateClient(logger.host, logger.pid) - - if ippToPspsDb.isKilled(logger.host, logger.pid): exitProgram("killed") - - PAUSEPERIOD = 60 - while ippToPspsDb.isPaused(logger.host, logger.pid): - - ippToPspsDb.updateClient(logger.host, logger.pid) - if ippToPspsDb.isKilled(logger.host, logger.pid): exitProgram("killed while paused") - logger.infoPair("Program paused for", "%d secs" % PAUSEPERIOD) - time.sleep(PAUSEPERIOD) - - -''' -Start of program. -''' -if len(sys.argv) > 1: CONFIGPATH = sys.argv[1] -else: - print "** Usage: " + sys.argv[0] + " [init]" - sys.exit(1) - -if len(sys.argv) > 2 and sys.argv[2] == "init": - QUEUE_IN = 1 -else: - QUEUE_IN = 0 - -# open config file -config = Config(CONFIGPATH) -logger = config.getLogger(sys.argv[0], 1, 1) - -# create various objects -dvo = Dvo(logger, config) -gpc1Db = Gpc1Db(logger, config) -ippToPspsDb = IppToPspsDb(logger, config) -datastore = Datastore(logger, config, ippToPspsDb) - -checkClientStatus() - -# catch Ctrl-C signal -signal.signal(signal.SIGINT, signal_handler) - -# check we connected ok -if not gpc1Db.everythingOK: sys.exit(1) -if not ippToPspsDb.everythingOK: sys.exit(1) - -POLLPERIOD = 600 - -# prompt user: force and publishing is a dangerous combination -if config.force and config.datastorePublishing and not QUEUE_IN: - response = raw_input("\n*** Are you sure you want to publish data with the 'force' option enabled (y/n)? ") - if response != "y": sys.exit(1) - -# prompt user: test and publishing is a dangerous combination -if config.test and config.datastorePublishing and not QUEUE_IN: - response = raw_input("\n*** Are you sure you want to publish data with the 'testMode' option enabled (y/n)? ") - if response != "y": sys.exit(1) - -# print info to log -logger.infoSeparator() -logger.infoTitle("ippToPsps loading started") - -# if an IN batch is requested, create and quit -if QUEUE_IN: - batchID = ippToPspsDb.createNewBatch("IN", 0, config) - if batchID > 0: - batch = InitBatch(logger, config, gpc1Db, ippToPspsDb, batchID) - batch.run() - - sys.exit(1) - - -''' -Main processing loop: - - 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 - -''' - -# this outer while loop simply waits a few minutes then starts queuing againas more stuff may appear in DVO over time -while True: - - config.refresh() - config.printAll() - - # starting positions - ra = config.minRa + config.halfBox - dec = config.minDec + config.halfBox - - # queue up batches that are processed but not loaded to datastore - logger.infoTitle("Previous failed datastore loads") - for batchType in config.batchTypes: publishAnyUnpublishedBatches(batchType) - - # loop through full range of RA/Dec queueing stuff in boxes of size config.boxSize - while ra <= config.maxRa: - - while dec <= config.maxDec: - - checkClientStatus() - - # for each batch type - # - queue items in this box - # - check if we should pre-load DVO region - # - process the items - for batchType in config.batchTypes: - - ids = queueItemsInBox(batchType, ra, dec, config.boxSize) - - if len(ids) < 1: - logger.infoPair("No items found in this region", "skipping") - continue - - dvo.setSkyAreaAsBox(ra, dec, config.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 config.test: break - - dec = dec + config.boxSize - if config.test: break - - dec = config.minDec + config.halfBox - ra = ra + config.boxSize - if config.test: break - - # wait for the POLLPERIOD before checking for new ids - logger.infoPair("Finished. Waiting for", "%.1f minutes before checking DVO for new items" % (POLLPERIOD/60.0)) - time.sleep(POLLPERIOD) - -exitProgram("end of program") - Index: trunk/ippToPsps/jython/loader.py =================================================================== --- trunk/ippToPsps/jython/loader.py (revision 33346) +++ trunk/ippToPsps/jython/loader.py (revision 33346) @@ -0,0 +1,219 @@ +#!/usr/bin/env jython + +# +# The main loading program for ippToPsps. See usage below. +# + +import time +import sys +import os +import traceback +import logging.config + +from ipptopsps import IppToPsps +from scratchdb import ScratchDb +from gpc1db import Gpc1Db +from datastore import Datastore +from batch import Batch +from dvo import Dvo + +from initbatch import InitBatch +from stackbatch import StackBatch +from detectionbatch import DetectionBatch + +''' +Loader class +''' +class Loader(IppToPsps): + + ''' + Constructor + ''' + def __init__(self, argv): + super(Loader, self).__init__(argv) + + # create gpc1 database objects + 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) + self.scratchDb.disconnect() + continue + self.logger.infoPair("Using scratch Db", self.scratchDb.dbName) + break + + if not self.scratchDb.connected: + self.logger.errorPair("Cannot connect to a", "scratch database") + raise Exception("No scratch Db") + + # now pass scratch Db to dvo object + self.dvo.setScratchDb(self.scratchDb.dbName) + + # if an IN batch is requested, create and quit + if len(sys.argv) > 2 and sys.argv[2] == "init": + batchID = self.ippToPspsDb.createNewBatch("IN", 0, self.config) + if batchID > 0: + batch = InitBatch(self.logger, + self.config, + self.gpc1Db, + self.ippToPspsDb, + self.scratchDb, + batchID) + batch.run() + + self.exitProgram("init batch created") + + + # set a poll time of about 1 minutes + self.parsePollTimeArg("0.0166") + + self.config.printAll() + + ''' + Overrides base-class version + ''' + def refreshConfig(self): + + ret = super(Loader, self).refreshConfig() + self.dvo = Dvo(self.logger, self.config) + try: self.dvo.setScratchDb(self.scratchDb.dbName) + except: pass + self.datastore = Datastore(self.logger, self.config, self.ippToPspsDb) + + return ret + + ''' + Main processing loop + ''' + def run(self): + + while True: + + boxIds = self.ippToPspsDb.getBoxIds(self.config.name) + + self.logger.infoPair("Boxes with unprocessed data", "%d" % len(boxIds)) + + for boxId in boxIds: + + self.checkClientStatus() + + # get box info. if boxes have changed, break and start again + try: + boxDim = self.ippToPspsDb.getBoxDimensions(boxId) + except: + self.logger.infoPair("Can't get details for this box", "%d" % boxId) + break + + for batchType in self.config.batchTypes: + + ids = self.ippToPspsDb.getPendingIds(boxId, batchType) + + if len(ids) < 1: + self.logger.debugPair("No " + batchType + " items found in this box", "skipping") + continue + + self.logger.infoSeparator() + 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) + + # look in DVO for this box (with extra border) + self.dvo.setSkyAreaAsBox(boxDim['RA'], boxDim['DEC'], boxSizeWithBorder) + sizeToBeIngested = self.dvo.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 smfsPerGB > 40: + if not self.dvo.sync(): + self.logger.errorPair("Could not sync DVO with MySQL", "skipping") + continue + + useFullTables = 1 + else: useFullTables = 0 + + self.logger.infoBool("Using pre-ingested DVO data?", useFullTables) + self.processTheseItems(batchType, ids, useFullTables) + + self.checkClientStatus() + if not self.waitForPollTime(): break + + + ''' + Actually loops through items and processes them + ''' + def processTheseItems(self, batchType, ids, useFullTables): + + self.logger.infoPair("%s items queued" % batchType, "%d" % len(ids)) + + # loop round IDs of all items to be processed + self.ippToPspsDb.lockBatchTable() + unattemptedCount = 0 + for id in ids: + + batchID = self.ippToPspsDb.createNewBatch(batchType, id, self.config) + + if batchID < 0: + unattemptedCount += 1 + continue + + self.ippToPspsDb.unlockTables() + + if batchType == "P2": + batch = DetectionBatch(self.logger, + self.config, + self.gpc1Db, + self.ippToPspsDb, + self.scratchDb, + id, + batchID, + useFullTables) + elif batchType == "ST": + batch = StackBatch(self.logger, + self.config, + self.gpc1Db, + self.ippToPspsDb, + self.scratchDb, + id, + batchID, + useFullTables) + + batch.run() + + # check that we should continue + self.checkClientStatus() + + self.ippToPspsDb.lockBatchTable() + + self.logger.infoSeparator() + + # if in test mode, then quit after one batch + if self.config.test: break + + self.ippToPspsDb.unlockTables() + self.logger.infoPair("Unattempted batches", "%d" % unattemptedCount) + + ''' + Overrides base-class version + ''' + def printUsage(self): + super(Cleanup, self).printUsage() + print " [init]" + + +''' +Start of program. +''' +try: + loader = Loader(sys.argv) + loader.run() + loader.exitProgram("finished") +except Exception, e: + print str(e) + traceback.print_exc()