Index: trunk/ippToPsps/jython/ipptopspsdb.py =================================================================== --- trunk/ippToPsps/jython/ipptopspsdb.py (revision 36815) +++ trunk/ippToPsps/jython/ipptopspsdb.py (revision 37246) @@ -166,119 +166,4 @@ return ids - - ## ''' - ## XXX EAM : removed this function and made it more explicit (with an additional block) - ## Returns a list of merged batch IDs that are merged but not yet deleted - ## ''' - ## def getMergedButNotDeletedBatchIDs(self, batchType, column): - ## - ## sql = "SELECT DISTINCT batch_id \ - ## FROM batch \ - ## WHERE timestamp > '" + self.skychunk.epoch + "' \ - ## AND batch_type = '" + batchType + "' \ - ## AND dvo_db = '" + self.skychunk.dvoLabel + "' \ - ## AND merged = 1 \ - ## AND " + column + " = 0" - ## - ## ids = [] - ## try: - ## rs = self.executeQuery(sql) - ## while (rs.next()): ids.append(rs.getInt(1)) - ## rs.close() - ## except: - ## self.logger.exception("Can't query for merged batch ids in ipptopsps Db") - ## - ## self.logger.debug("Found %d merged but un-deleted items" % len(ids)) - ## - ## return ids - - ## ''' - ## Returns a list of batch IDs marked as 'purged' but not yet deleted - ## XXX EAM : removed this function and made it more explicit (with an additional block) - ## ''' - ## def getPurgedButNotDeletedBatchIDs(self, batchType, column): - ## - ## sql = "SELECT DISTINCT batch_id \ - ## FROM batch \ - ## WHERE timestamp > '" + self.skychunk.epoch + "' \ - ## AND batch_type = '" + batchType + "' \ - ## AND dvo_db = '" + self.skychunk.dvoLabel + "' \ - ## AND purged = 1 \ - ## AND " + column + " = 0" - ## - ## ids = [] - ## try: - ## rs = self.executeQuery(sql) - ## while (rs.next()): ids.append(rs.getInt(1)) - ## rs.close() - ## except: - ## self.logger.exception("Can't query for merged batch ids in ipptopsps Db") - ## - ## - ## self.logger.debug("Found %d merged but un-deleted items" % len(ids)) - ## - ## return ids - - ## ''' - ## Returns a list of processed batch IDs that have been loaded to the ODM but not yet deleted - ## ''' - ## def getLoadedToODMButNotDeletedBatchIDs(self, batchType, column): - ## - ## sql = "SELECT DISTINCT batch_id \ - ## FROM batch \ - ## WHERE timestamp > '" + self.skychunk.epoch + "' \ - ## AND batch_type = '" + batchType + "' \ - ## AND dvo_db = '" + self.skychunk.dvoLabel + "' \ - ## AND (loaded_to_ODM = -1 OR merge_worthy = 1) \ - ## AND " + column + " = 0" - ## - ## ids = [] - ## try: - ## rs = self.executeQuery(sql) - ## while (rs.next()): ids.append(rs.getInt(1)) - ## rs.close() - ## except: - ## self.logger.exception("Can't query for merged batch ids in ipptopsps Db") - ## - ## self.logger.debug("Found %d merged but un-deleted items" % len(ids)) - ## - ## return ids - - # ''' - # Returns a list of merged batch IDs that are not deleted from local disk - # ''' - # def getMergedButNotDeletedFromLocalDisk(self, batchType): - # return self.getMergedButNotDeletedBatchIDs(batchType, "deleted_local") - - ## ''' - ## Returns a list of processed batch IDs that are loaded to the ODM, but not deleted from local disk - ## ''' - ## def getLoadedToODMButNotDeletedFromLocalDisk(self, batchType): - ## return self.getLoadedToODMButNotDeletedBatchIDs(batchType, "deleted_local") - ## - ## ''' - ## Returns a list of processed batch IDs that are loaded to the ODM, but not deleted from datastore - ## ''' - ## def getLoadedToODMButNotDeletedFromDatastore(self, batchType): - ## return self.getLoadedToODMButNotDeletedBatchIDs(batchType, "deleted_datastore") - ## - ## ''' - ## Returns a list of processed batch IDs that are loaded to the ODM, but not deleted from DXLayer - ## ''' - ## def getLoadedToODMButNotDeletedFromDXLayer(self, batchType): - ## return self.getLoadedToODMButNotDeletedBatchIDs(batchType, "deleted_dxlayer") - - ## ''' - ## Returns a list of purged batch IDs that not deleted from local disk - ## ''' - ## def getPurgedButNotDeletedFromLocalDisk(self, batchType): - ## return self.getPurgedButNotDeletedBatchIDs(batchType, "deleted_local") - ## - ## ''' - ## Returns a list of purged batch IDs that are not deleted from datastore - ## ''' - ## def getPurgedButNotDeletedFromDatastore(self, batchType): - ## return self.getPurgedButNotDeletedBatchIDs(batchType, "deleted_datastore") - ''' @@ -786,5 +671,5 @@ and not self.consistentlyFailed(batchType, stageID)): - self.logger.infoPair("heather:","passed logic") + # self.logger.infoPair("heather:","passed logic") sql = "INSERT INTO batch ( \ @@ -803,5 +688,6 @@ '" + self.skychunk.name + "' \ )" - self.logger.infoPair("heather:","sql") + + # self.logger.infoPair("heather:",sql) self.execute(sql) @@ -816,4 +702,6 @@ except: self.logger.exception("Unable to get batch ID") + self.logger.infoPair("sql:",sql) + batchID = -1 @@ -1017,4 +905,5 @@ response = raw_input(" * Name for new skychunk, or existing skychunk to edit? ") if response == "": return + response = response.strip() self.skychunk.name = response @@ -1056,7 +945,10 @@ question = question + "? " response = raw_input(question) + response = response.strip() + print "response: '" + response + "'" if response != "": if type.find("varchar") != -1 or type.find("timestamp") != -1: quotes = "'" else: quotes = "" + # print "UPDATE skychunk SET " + field + " = " + quotes + response + quotes + " WHERE name = '" + self.skychunk.name + "'" self.execute("UPDATE skychunk SET " + field + " = " + quotes + response + quotes + " WHERE name = '" + self.skychunk.name + "'") @@ -1129,60 +1021,112 @@ self.skychunk.batchTypes = [] - try: - rs = self.executeQuery(sql) - - rs.first() - self.skychunk.datastoreProduct = rs.getString(1) - self.skychunk.datastoreType = rs.getString(2) - if rs.getInt(3) == 1: self.skychunk.datastorePublishing = True - else: self.skychunk.datastorePublishing = False - self.skychunk.dvoLabel = rs.getString(4) - - # if dvoLabel is null is can break queries, so set to something - if not self.skychunk.dvoLabel: self.skychunk.dvoLabel = "none" - self.skychunk.dvoLocation = rs.getString(5) - self.skychunk.minRa = rs.getDouble(6) - self.skychunk.maxRa = rs.getDouble(7) - self.skychunk.minDec = rs.getDouble(8) - self.skychunk.maxDec = rs.getDouble(9) - - self.skychunk.boxSize = rs.getDouble(10) - # self.skychunk.halfBox = self.skychunk.boxSize/2.0 - # self.skychunk.boxSizeWithBorder = self.skychunk.boxSize + (self.skychunk.BORDER * 2) - - self.skychunk.basePath = rs.getString(11) - self.skychunk.dataRelease = rs.getInt(12) - - self.skychunk.deleteLocal = rs.getInt(13) - self.skychunk.deleteDatastore = rs.getInt(14) - self.skychunk.deleteDxLayer = rs.getInt(15) - - self.skychunk.epoch = rs.getString(16) - - self.skychunk.survey = rs.getString(17) - self.skychunk.pspsSurvey = rs.getString(18) - - if rs.getInt(19) == 1: self.skychunk.batchTypes.append("P2") - if rs.getInt(20) == 1: self.skychunk.batchTypes.append("ST") - if rs.getInt(21) == 1: self.skychunk.batchTypes.append("OB") - if rs.getInt(22) == 1: self.skychunk.batchTypes.append("DF") - if rs.getInt(23) == 1: self.skychunk.batchTypes.append("DO") - if rs.getInt(24) == 1: self.skychunk.batchTypes.append("FW") - if rs.getInt(25) == 1: self.skychunk.batchTypes.append("FO") - - self.skychunk.force = True # TODO - self.skychunk.parallel = False # TODO - - if rs.getInt(26) == 1: self.skychunk.parallel = True - self.skychunk.P2_smf_version = rs.getString(27) - self.skychunk.ST_cmf_version = rs.getString(28) - self.skychunk.trange_start = rs.getString(29) - self.skychunk.trange_end = rs.getString(30) - - if self.skychunk.parallel: print "USING parallel" - self.skychunk.isLoaded = True - except: + + try: + rs = self.executeQuery("show tables") + except: + self.logger.errorPair("cannot even show tables?", "Boo hoo") + sys.exit(1) + + # while rs.next(): + # print "table row" + # print rs.getString(1) + # + # rs = self.executeQuery("select * from skychunk") + # while rs.next(): + # print "skychunk row" + # print rs.getString(1) + # + # print "SQL (2): " + sql + + try: + rs = self.executeQuery(sql) + + except: + self.logger.errorPair("problem with sql query for skychunk", sql) + sys.exit(1) + + if not rs.next(): self.logger.errorPair("Could not read skychunk with name", self.skychunk.name) self.skychunk.isLoaded = False + return False + + self.skychunk.datastoreProduct = rs.getString(1) + self.skychunk.datastoreType = rs.getString(2) + if rs.getInt(3) == 1: self.skychunk.datastorePublishing = True + else: self.skychunk.datastorePublishing = False + self.skychunk.dvoLabel = rs.getString(4) + + # if dvoLabel is null is can break queries, so set to something + if not self.skychunk.dvoLabel: self.skychunk.dvoLabel = "none" + self.skychunk.dvoLocation = rs.getString(5) + self.skychunk.minRa = rs.getDouble(6) + self.skychunk.maxRa = rs.getDouble(7) + self.skychunk.minDec = rs.getDouble(8) + self.skychunk.maxDec = rs.getDouble(9) + + self.skychunk.boxSize = rs.getDouble(10) + # self.skychunk.halfBox = self.skychunk.boxSize/2.0 + # self.skychunk.boxSizeWithBorder = self.skychunk.boxSize + (self.skychunk.BORDER * 2) + + self.skychunk.basePath = rs.getString(11) + self.skychunk.dataRelease = rs.getInt(12) + + self.skychunk.deleteLocal = rs.getInt(13) + self.skychunk.deleteDatastore = rs.getInt(14) + self.skychunk.deleteDxLayer = rs.getInt(15) + + self.skychunk.epoch = rs.getString(16) + + self.skychunk.survey = rs.getString(17) + self.skychunk.pspsSurvey = rs.getString(18) + + if rs.getInt(19) == 1: self.skychunk.batchTypes.append("P2") + if rs.getInt(20) == 1: self.skychunk.batchTypes.append("ST") + if rs.getInt(21) == 1: self.skychunk.batchTypes.append("OB") + if rs.getInt(22) == 1: self.skychunk.batchTypes.append("DF") + if rs.getInt(23) == 1: self.skychunk.batchTypes.append("DO") + if rs.getInt(24) == 1: self.skychunk.batchTypes.append("FW") + if rs.getInt(25) == 1: self.skychunk.batchTypes.append("FO") + + self.skychunk.force = True # TODO + self.skychunk.parallel = False # TODO + + if rs.getInt(26) == 1: self.skychunk.parallel = True + self.skychunk.P2_smf_version = rs.getString(27) + # options: "use_new", "not_reproc", "use_original" + + self.skychunk.ST_cmf_version = rs.getString(28) + # options: not used? + + # NOTE : the rs (ResultSet) class is somewhat annoying and finicky esp about timestamp values + # for example, rs.getTimestamp(N) fails silently if the value is not a valid time. this is + # also true to getString() on the same column. Docs for the ResultSet class can be found here: + # http://docs.oracle.com/javase/1.5.0/docs/api/java/sql/ResultSet.html?is-external=true + + # rsmd = rs.getMetaData() + # print "got metadata" + # print "number of columns: " + str(rsmd.getColumnCount()) + # print "label: " + rsmd.getColumnLabel(25) + # print "label: " + rsmd.getColumnLabel(29) + # print "name: " + rsmd.getColumnName(25) + # print "name: " + rsmd.getColumnName(29) + # print "typename: " + rsmd.getColumnTypeName(25) + # print "typename: " + rsmd.getColumnTypeName(29) + + # epochStamp = rs.getTimestamp(16) + # print "epoch: " + epochStamp.toString() + + # print "label: " + rsmd.getColumnLabel(16) + # print "name: " + rsmd.getColumnName(16) + # print "typename: " + rsmd.getColumnTypeName(16) + + self.skychunk.trange_start = rs.getString(29) + # options: "YYYY-MM-DD,HH:MM:SS" or "0000-00-00 00:00:00" + + self.skychunk.trange_end = rs.getString(30) + # options: "YYYY-MM-DD,HH:MM:SS" or "0000-00-00 00:00:00" + + if self.skychunk.parallel: print "USING parallel" + self.skychunk.isLoaded = True return self.skychunk.isLoaded @@ -1308,5 +1252,5 @@ def insertPending(self, box_id, batchType, ids): - # print "starting insert pending" + print "starting insert pending" # first delete old pending items @@ -1314,6 +1258,4 @@ WHERE box_id = " + str(box_id) + " \ AND batch_type = '" + batchType + "'") - - # print "deleted old items" for id in ids: @@ -1323,6 +1265,9 @@ (" + str(box_id) + ", '" + batchType + "', " + str(id) + ")" - # print "sql: ", sql - self.execute(sql) + try: self.execute(sql) + except: + print "failed to insert into pending" + print "sql: " + sql + raise ''' @@ -1348,29 +1293,36 @@ GROUP BY ra_center LIMIT 1" - # XXX print "sql: ", sql - - try: - rs = self.executeQuery(sql) - rs.first() - raCenter = rs.getFloat(1) - rs.close() - except: - self.logger.errorPair("No free stripes", "trying to steal boxes from other client...") - sql = "SELECT ra_center, COUNT(*) AS numPending\ - FROM box \ - JOIN pending ON (id = box_id) \ - WHERE skychunk = '" + self.skychunk.name + "' \ - AND batch_type = '" + batchType + "' \ - GROUP BY ra_center \ - ORDER BY numPending \ - DESC LIMIT 1" - - try: - rs = self.executeQuery(sql) - rs.first() - raCenter = rs.getFloat(1) - rs.close() - except: - self.logger.errorPair("No outstanding boxes to steal", "no stripe obtained") + try: rs = self.executeQuery(sql) + except: + self.logger.errorPair("Problem querying for outstanding boxes", "error in sql?") + print "SQL: " + sql + + if not rs.next(): + self.logger.errorPair("No outstanding boxes", "nothing to do here") + rs.close() + self.unlockTables() + return ids + + raCenter = rs.getFloat(1) + rs.close() + + #except: + # self.logger.errorPair("No free stripes", "trying to steal boxes from other client...") + # sql = "SELECT ra_center, COUNT(*) AS numPending\ + # FROM box \ + # JOIN pending ON (id = box_id) \ + # WHERE skychunk = '" + self.skychunk.name + "' \ + # AND batch_type = '" + batchType + "' \ + # GROUP BY ra_center \ + # ORDER BY numPending \ + # DESC LIMIT 1" + # + # try: + # rs = self.executeQuery(sql) + # rs.first() + # raCenter = rs.getFloat(1) + # rs.close() + # except: + # self.logger.errorPair("No outstanding boxes to steal", "no stripe obtained") # tests against ra_center need to use a finite box @@ -1421,26 +1373,35 @@ GROUP BY ra_center LIMIT 1" - try: - rs = self.executeQuery(sql) - rs.first() - raCenter = rs.getFloat(1) - rs.close() - except: - self.logger.errorPair("No free stripes", "trying to steal boxes from other client...") - sql = "SELECT ra_center, COUNT(*) AS numPending\ - FROM box \ - JOIN pending ON (id = box_id) \ - WHERE skychunk = '" + self.skychunk.name + "' \ - GROUP BY ra_center \ - ORDER BY numPending \ - DESC LIMIT 1" - - try: - rs = self.executeQuery(sql) - rs.first() - raCenter = rs.getFloat(1) - rs.close() - except: - self.logger.errorPair("No outstanding boxes to steal", "no stripe obtained") + try: rs = self.executeQuery(sql) + except: + self.logger.errorPair("Problem querying for outstanding boxes", "error in sql?") + print "SQL: " + sql + + if not rs.next(): + self.logger.infoPair("No outstanding boxes", "nothing to do here") + rs.close() + self.unlockTables() + return ids + + raCenter = rs.getFloat(1) + rs.close() + + # except: + # self.logger.errorPair("No free stripes", "trying to steal boxes from other client...") + # sql = "SELECT ra_center, COUNT(*) AS numPending\ + # FROM box \ + # JOIN pending ON (id = box_id) \ + # WHERE skychunk = '" + self.skychunk.name + "' \ + # GROUP BY ra_center \ + # ORDER BY numPending \ + # DESC LIMIT 1" + # + # try: + # rs = self.executeQuery(sql) + # rs.first() + # raCenter = rs.getFloat(1) + # rs.close() + # except: + # self.logger.errorPair("No outstanding boxes to steal", "no stripe obtained") # tests against ra_center need to use a finite box @@ -1634,22 +1595,4 @@ self.logger.infoPair("Items written to Db", "%d" % count) - # try: - # self.execute("DROP TABLE all_pending_alt") - # except: - # pass - # - # self.execute("CREATE TABLE all_pending_alt (stage_id bigint(20), ra_bore float, dec_bore float)") - # for row in rows: - # try: - # sql = "INSERT INTO all_pending_alt (stage_id, ra_bore, dec_bore) \ - # VALUES (%d, %f, %f)" % (row[0], row[1], row[2]) - # print "sql 2: ", sql - # self.execute(sql) - # except: continue - # - # count = self.getRowCount("all_pending_alt") - # self.logger.infoPair("Items written to alt Db", "%d" % count) - - ''' Creates a temporary table and shoves it full of stage_ids and ra/dec coords @@ -1670,5 +1613,5 @@ VALUES ( '" + dvo_db + "', '" + batchType + "'," + str(row[0]) + "," + str(row[1]) + " \ , " + str(row[2]) + ", '"+uniq+"' ) " - print "sql 1: ", sql + # print "sql 1: ", sql self.execute(sql) except: continue @@ -1683,11 +1626,4 @@ ''' def getItemsInThisBox(self, minRA, maxRA, minDEC, maxDEC): - - # the old linear size is ill-conceived - ## XX halfSide = self.skychunk.boxSize/2.0 - ## XX minRa = ra-halfSide - ## XX maxRa = ra+halfSide - ## XX minDec = dec-halfSide - ## XX maxDec = dec+halfSide ids = [] @@ -1698,9 +1634,8 @@ AND dec_bore BETWEEN " + str(minDEC) + " AND " + str(maxDEC) - print "items: ", sql - - try: - rs = self.executeQuery(sql) - while (rs.next()): ids.append(rs.getInt(1)) + try: + rs = self.executeQuery(sql) + while (rs.next()): + ids.append(rs.getInt(1)) except: self.logger.errorPair("Can't get items in this box", sql)