- Timestamp:
- Jan 12, 2015, 12:24:24 PM (12 years ago)
- File:
-
- 1 edited
Legend:
- Unmodified
- Added
- Removed
-
branches/eam_branches/ipp-pv3-20140717-merge/ippToPsps/jython/ipptopspsdb.py
r36815 r37820 166 166 167 167 return ids 168 169 ## '''170 ## XXX EAM : removed this function and made it more explicit (with an additional block)171 ## Returns a list of merged batch IDs that are merged but not yet deleted172 ## '''173 ## def getMergedButNotDeletedBatchIDs(self, batchType, column):174 ##175 ## sql = "SELECT DISTINCT batch_id \176 ## FROM batch \177 ## WHERE timestamp > '" + self.skychunk.epoch + "' \178 ## AND batch_type = '" + batchType + "' \179 ## AND dvo_db = '" + self.skychunk.dvoLabel + "' \180 ## AND merged = 1 \181 ## AND " + column + " = 0"182 ##183 ## ids = []184 ## try:185 ## rs = self.executeQuery(sql)186 ## while (rs.next()): ids.append(rs.getInt(1))187 ## rs.close()188 ## except:189 ## self.logger.exception("Can't query for merged batch ids in ipptopsps Db")190 ##191 ## self.logger.debug("Found %d merged but un-deleted items" % len(ids))192 ##193 ## return ids194 195 ## '''196 ## Returns a list of batch IDs marked as 'purged' but not yet deleted197 ## XXX EAM : removed this function and made it more explicit (with an additional block)198 ## '''199 ## def getPurgedButNotDeletedBatchIDs(self, batchType, column):200 ##201 ## sql = "SELECT DISTINCT batch_id \202 ## FROM batch \203 ## WHERE timestamp > '" + self.skychunk.epoch + "' \204 ## AND batch_type = '" + batchType + "' \205 ## AND dvo_db = '" + self.skychunk.dvoLabel + "' \206 ## AND purged = 1 \207 ## AND " + column + " = 0"208 ##209 ## ids = []210 ## try:211 ## rs = self.executeQuery(sql)212 ## while (rs.next()): ids.append(rs.getInt(1))213 ## rs.close()214 ## except:215 ## self.logger.exception("Can't query for merged batch ids in ipptopsps Db")216 ##217 ##218 ## self.logger.debug("Found %d merged but un-deleted items" % len(ids))219 ##220 ## return ids221 222 ## '''223 ## Returns a list of processed batch IDs that have been loaded to the ODM but not yet deleted224 ## '''225 ## def getLoadedToODMButNotDeletedBatchIDs(self, batchType, column):226 ##227 ## sql = "SELECT DISTINCT batch_id \228 ## FROM batch \229 ## WHERE timestamp > '" + self.skychunk.epoch + "' \230 ## AND batch_type = '" + batchType + "' \231 ## AND dvo_db = '" + self.skychunk.dvoLabel + "' \232 ## AND (loaded_to_ODM = -1 OR merge_worthy = 1) \233 ## AND " + column + " = 0"234 ##235 ## ids = []236 ## try:237 ## rs = self.executeQuery(sql)238 ## while (rs.next()): ids.append(rs.getInt(1))239 ## rs.close()240 ## except:241 ## self.logger.exception("Can't query for merged batch ids in ipptopsps Db")242 ##243 ## self.logger.debug("Found %d merged but un-deleted items" % len(ids))244 ##245 ## return ids246 247 # '''248 # Returns a list of merged batch IDs that are not deleted from local disk249 # '''250 # def getMergedButNotDeletedFromLocalDisk(self, batchType):251 # return self.getMergedButNotDeletedBatchIDs(batchType, "deleted_local")252 253 ## '''254 ## Returns a list of processed batch IDs that are loaded to the ODM, but not deleted from local disk255 ## '''256 ## def getLoadedToODMButNotDeletedFromLocalDisk(self, batchType):257 ## return self.getLoadedToODMButNotDeletedBatchIDs(batchType, "deleted_local")258 ##259 ## '''260 ## Returns a list of processed batch IDs that are loaded to the ODM, but not deleted from datastore261 ## '''262 ## def getLoadedToODMButNotDeletedFromDatastore(self, batchType):263 ## return self.getLoadedToODMButNotDeletedBatchIDs(batchType, "deleted_datastore")264 ##265 ## '''266 ## Returns a list of processed batch IDs that are loaded to the ODM, but not deleted from DXLayer267 ## '''268 ## def getLoadedToODMButNotDeletedFromDXLayer(self, batchType):269 ## return self.getLoadedToODMButNotDeletedBatchIDs(batchType, "deleted_dxlayer")270 271 ## '''272 ## Returns a list of purged batch IDs that not deleted from local disk273 ## '''274 ## def getPurgedButNotDeletedFromLocalDisk(self, batchType):275 ## return self.getPurgedButNotDeletedBatchIDs(batchType, "deleted_local")276 ##277 ## '''278 ## Returns a list of purged batch IDs that are not deleted from datastore279 ## '''280 ## def getPurgedButNotDeletedFromDatastore(self, batchType):281 ## return self.getPurgedButNotDeletedBatchIDs(batchType, "deleted_datastore")282 283 168 284 169 ''' … … 786 671 and not self.consistentlyFailed(batchType, stageID)): 787 672 788 self.logger.infoPair("heather:","passed logic")673 # self.logger.infoPair("heather:","passed logic") 789 674 790 675 sql = "INSERT INTO batch ( \ … … 803 688 '" + self.skychunk.name + "' \ 804 689 )" 805 self.logger.infoPair("heather:","sql") 690 691 # self.logger.infoPair("heather:",sql) 806 692 807 693 self.execute(sql) … … 816 702 except: 817 703 self.logger.exception("Unable to get batch ID") 704 self.logger.infoPair("sql:",sql) 705 818 706 batchID = -1 819 707 … … 868 756 869 757 self.execute(sql) 758 759 ''' 760 Inserts some forced detection metadata for this batch ID 761 ''' 762 def insertForcedWarpMeta(self, batchID, expID, filter): 763 764 sql = "INSERT INTO forcedwarp ( \ 765 batch_id \ 766 ,exp_id \ 767 ,filter \ 768 ) VALUES ( \ 769 " + str(batchID) + " \ 770 ," + str(expID) + " \ 771 ,'" + filter + "' \ 772 )" 773 774 self.execute(sql) 775 776 777 778 ''' 779 Inserts some forcedobject metadata for this batch ID 780 ''' 781 def insertForcedObjectMeta(self, batchID, region): 782 783 sql = "INSERT INTO forcedobject ( \ 784 batch_id \ 785 ,region \ 786 ) VALUES ( \ 787 " + str(batchID) + " \ 788 ,'" + region + "' \ 789 )" 790 791 self.execute(sql) 792 793 794 795 796 ''' 797 Inserts some diff metadata for this batch ID 798 ''' 799 def insertDiffMeta(self, batchID, filter, diffType): 800 801 sql = "INSERT INTO diff ( \ 802 batch_id \ 803 ,filter \ 804 ,diff_type \ 805 ) VALUES ( \ 806 " + str(batchID) + " \ 807 ,'" + filter + "' \ 808 ,'" + diffType + "')" 809 810 self.execute(sql) 811 812 813 814 815 ''' 816 Inserts some diffobject metadata for this batch ID 817 ''' 818 def insertDiffObjectMeta(self, batchID, region): 819 820 sql = "INSERT INTO diffobject ( \ 821 batch_id \ 822 ,region \ 823 ) VALUES ( \ 824 " + str(batchID) + " \ 825 ,'" + region + "' \ 826 )" 827 828 self.execute(sql) 829 830 831 832 833 834 870 835 ''' 871 836 Resets a batch ready for re-loading to the datastore … … 1017 982 response = raw_input(" * Name for new skychunk, or existing skychunk to edit? ") 1018 983 if response == "": return 984 response = response.strip() 1019 985 self.skychunk.name = response 1020 986 … … 1056 1022 question = question + "? " 1057 1023 response = raw_input(question) 1024 response = response.strip() 1025 print "response: '" + response + "'" 1058 1026 if response != "": 1059 1027 if type.find("varchar") != -1 or type.find("timestamp") != -1: quotes = "'" 1060 1028 else: quotes = "" 1029 # print "UPDATE skychunk SET " + field + " = " + quotes + response + quotes + " WHERE name = '" + self.skychunk.name + "'" 1061 1030 self.execute("UPDATE skychunk SET " + field + " = " + quotes + response + quotes + " WHERE name = '" + self.skychunk.name + "'") 1062 1031 … … 1129 1098 1130 1099 self.skychunk.batchTypes = [] 1131 try: 1132 rs = self.executeQuery(sql) 1133 1134 rs.first() 1135 self.skychunk.datastoreProduct = rs.getString(1) 1136 self.skychunk.datastoreType = rs.getString(2) 1137 if rs.getInt(3) == 1: self.skychunk.datastorePublishing = True 1138 else: self.skychunk.datastorePublishing = False 1139 self.skychunk.dvoLabel = rs.getString(4) 1140 1141 # if dvoLabel is null is can break queries, so set to something 1142 if not self.skychunk.dvoLabel: self.skychunk.dvoLabel = "none" 1143 self.skychunk.dvoLocation = rs.getString(5) 1144 self.skychunk.minRa = rs.getDouble(6) 1145 self.skychunk.maxRa = rs.getDouble(7) 1146 self.skychunk.minDec = rs.getDouble(8) 1147 self.skychunk.maxDec = rs.getDouble(9) 1148 1149 self.skychunk.boxSize = rs.getDouble(10) 1150 # self.skychunk.halfBox = self.skychunk.boxSize/2.0 1151 # self.skychunk.boxSizeWithBorder = self.skychunk.boxSize + (self.skychunk.BORDER * 2) 1152 1153 self.skychunk.basePath = rs.getString(11) 1154 self.skychunk.dataRelease = rs.getInt(12) 1155 1156 self.skychunk.deleteLocal = rs.getInt(13) 1157 self.skychunk.deleteDatastore = rs.getInt(14) 1158 self.skychunk.deleteDxLayer = rs.getInt(15) 1159 1160 self.skychunk.epoch = rs.getString(16) 1161 1162 self.skychunk.survey = rs.getString(17) 1163 self.skychunk.pspsSurvey = rs.getString(18) 1164 1165 if rs.getInt(19) == 1: self.skychunk.batchTypes.append("P2") 1166 if rs.getInt(20) == 1: self.skychunk.batchTypes.append("ST") 1167 if rs.getInt(21) == 1: self.skychunk.batchTypes.append("OB") 1168 if rs.getInt(22) == 1: self.skychunk.batchTypes.append("DF") 1169 if rs.getInt(23) == 1: self.skychunk.batchTypes.append("DO") 1170 if rs.getInt(24) == 1: self.skychunk.batchTypes.append("FW") 1171 if rs.getInt(25) == 1: self.skychunk.batchTypes.append("FO") 1172 1173 self.skychunk.force = True # TODO 1174 self.skychunk.parallel = False # TODO 1175 1176 if rs.getInt(26) == 1: self.skychunk.parallel = True 1177 self.skychunk.P2_smf_version = rs.getString(27) 1178 self.skychunk.ST_cmf_version = rs.getString(28) 1179 self.skychunk.trange_start = rs.getString(29) 1180 self.skychunk.trange_end = rs.getString(30) 1181 1182 if self.skychunk.parallel: print "USING parallel" 1183 self.skychunk.isLoaded = True 1184 except: 1100 1101 try: 1102 rs = self.executeQuery("show tables") 1103 except: 1104 self.logger.errorPair("cannot even show tables?", "Boo hoo") 1105 sys.exit(1) 1106 1107 # while rs.next(): 1108 # print "table row" 1109 # print rs.getString(1) 1110 # 1111 # rs = self.executeQuery("select * from skychunk") 1112 # while rs.next(): 1113 # print "skychunk row" 1114 # print rs.getString(1) 1115 # 1116 # print "SQL (2): " + sql 1117 1118 try: 1119 rs = self.executeQuery(sql) 1120 1121 except: 1122 self.logger.errorPair("problem with sql query for skychunk", sql) 1123 sys.exit(1) 1124 1125 if not rs.next(): 1185 1126 self.logger.errorPair("Could not read skychunk with name", self.skychunk.name) 1186 1127 self.skychunk.isLoaded = False 1128 return False 1129 1130 self.skychunk.datastoreProduct = rs.getString(1) 1131 self.skychunk.datastoreType = rs.getString(2) 1132 if rs.getInt(3) == 1: self.skychunk.datastorePublishing = True 1133 else: self.skychunk.datastorePublishing = False 1134 self.skychunk.dvoLabel = rs.getString(4) 1135 1136 # if dvoLabel is null is can break queries, so set to something 1137 if not self.skychunk.dvoLabel: self.skychunk.dvoLabel = "none" 1138 self.skychunk.dvoLocation = rs.getString(5) 1139 self.skychunk.minRa = rs.getDouble(6) 1140 self.skychunk.maxRa = rs.getDouble(7) 1141 self.skychunk.minDec = rs.getDouble(8) 1142 self.skychunk.maxDec = rs.getDouble(9) 1143 1144 self.skychunk.boxSize = rs.getDouble(10) 1145 # self.skychunk.halfBox = self.skychunk.boxSize/2.0 1146 # self.skychunk.boxSizeWithBorder = self.skychunk.boxSize + (self.skychunk.BORDER * 2) 1147 1148 self.skychunk.basePath = rs.getString(11) 1149 self.skychunk.dataRelease = rs.getInt(12) 1150 1151 self.skychunk.deleteLocal = rs.getInt(13) 1152 self.skychunk.deleteDatastore = rs.getInt(14) 1153 self.skychunk.deleteDxLayer = rs.getInt(15) 1154 1155 self.skychunk.epoch = rs.getString(16) 1156 1157 self.skychunk.survey = rs.getString(17) 1158 self.skychunk.pspsSurvey = rs.getString(18) 1159 1160 if rs.getInt(19) == 1: self.skychunk.batchTypes.append("P2") 1161 if rs.getInt(20) == 1: self.skychunk.batchTypes.append("ST") 1162 if rs.getInt(21) == 1: self.skychunk.batchTypes.append("OB") 1163 if rs.getInt(22) == 1: self.skychunk.batchTypes.append("DF") 1164 if rs.getInt(23) == 1: self.skychunk.batchTypes.append("DO") 1165 if rs.getInt(24) == 1: self.skychunk.batchTypes.append("FW") 1166 if rs.getInt(25) == 1: self.skychunk.batchTypes.append("FO") 1167 1168 self.skychunk.force = True # TODO 1169 self.skychunk.parallel = False # TODO 1170 1171 if rs.getInt(26) == 1: self.skychunk.parallel = True 1172 self.skychunk.P2_smf_version = rs.getString(27) 1173 # options: "use_new", "not_reproc", "use_original" 1174 1175 self.skychunk.ST_cmf_version = rs.getString(28) 1176 # options: not used? 1177 1178 # NOTE : the rs (ResultSet) class is somewhat annoying and finicky esp about timestamp values 1179 # for example, rs.getTimestamp(N) fails silently if the value is not a valid time. this is 1180 # also true to getString() on the same column. Docs for the ResultSet class can be found here: 1181 # http://docs.oracle.com/javase/1.5.0/docs/api/java/sql/ResultSet.html?is-external=true 1182 1183 # rsmd = rs.getMetaData() 1184 # print "got metadata" 1185 # print "number of columns: " + str(rsmd.getColumnCount()) 1186 # print "label: " + rsmd.getColumnLabel(25) 1187 # print "label: " + rsmd.getColumnLabel(29) 1188 # print "name: " + rsmd.getColumnName(25) 1189 # print "name: " + rsmd.getColumnName(29) 1190 # print "typename: " + rsmd.getColumnTypeName(25) 1191 # print "typename: " + rsmd.getColumnTypeName(29) 1192 1193 # epochStamp = rs.getTimestamp(16) 1194 # print "epoch: " + epochStamp.toString() 1195 1196 # print "label: " + rsmd.getColumnLabel(16) 1197 # print "name: " + rsmd.getColumnName(16) 1198 # print "typename: " + rsmd.getColumnTypeName(16) 1199 1200 self.skychunk.trange_start = rs.getString(29) 1201 # options: "YYYY-MM-DD,HH:MM:SS" or "0000-00-00 00:00:00" 1202 1203 self.skychunk.trange_end = rs.getString(30) 1204 # options: "YYYY-MM-DD,HH:MM:SS" or "0000-00-00 00:00:00" 1205 1206 if self.skychunk.parallel: print "USING parallel" 1207 self.skychunk.isLoaded = True 1187 1208 1188 1209 return self.skychunk.isLoaded … … 1308 1329 def insertPending(self, box_id, batchType, ids): 1309 1330 1310 #print "starting insert pending"1331 print "starting insert pending" 1311 1332 1312 1333 # first delete old pending items … … 1314 1335 WHERE box_id = " + str(box_id) + " \ 1315 1336 AND batch_type = '" + batchType + "'") 1316 1317 # print "deleted old items"1318 1337 1319 1338 for id in ids: … … 1323 1342 (" + str(box_id) + ", '" + batchType + "', " + str(id) + ")" 1324 1343 1325 # print "sql: ", sql 1326 self.execute(sql) 1344 try: self.execute(sql) 1345 except: 1346 print "failed to insert into pending" 1347 print "sql: " + sql 1348 raise 1327 1349 1328 1350 ''' … … 1348 1370 GROUP BY ra_center LIMIT 1" 1349 1371 1350 # XXX print "sql: ", sql 1351 1352 try: 1353 rs = self.executeQuery(sql) 1354 rs.first() 1355 raCenter = rs.getFloat(1) 1356 rs.close() 1357 except: 1358 self.logger.errorPair("No free stripes", "trying to steal boxes from other client...") 1359 sql = "SELECT ra_center, COUNT(*) AS numPending\ 1360 FROM box \ 1361 JOIN pending ON (id = box_id) \ 1362 WHERE skychunk = '" + self.skychunk.name + "' \ 1363 AND batch_type = '" + batchType + "' \ 1364 GROUP BY ra_center \ 1365 ORDER BY numPending \ 1366 DESC LIMIT 1" 1367 1368 try: 1369 rs = self.executeQuery(sql) 1370 rs.first() 1371 raCenter = rs.getFloat(1) 1372 rs.close() 1373 except: 1374 self.logger.errorPair("No outstanding boxes to steal", "no stripe obtained") 1372 try: rs = self.executeQuery(sql) 1373 except: 1374 self.logger.errorPair("Problem querying for outstanding boxes", "error in sql?") 1375 print "SQL: " + sql 1376 1377 if not rs.next(): 1378 self.logger.errorPair("No outstanding boxes", "nothing to do here") 1379 rs.close() 1380 self.unlockTables() 1381 return ids 1382 1383 raCenter = rs.getFloat(1) 1384 rs.close() 1385 1386 #except: 1387 # self.logger.errorPair("No free stripes", "trying to steal boxes from other client...") 1388 # sql = "SELECT ra_center, COUNT(*) AS numPending\ 1389 # FROM box \ 1390 # JOIN pending ON (id = box_id) \ 1391 # WHERE skychunk = '" + self.skychunk.name + "' \ 1392 # AND batch_type = '" + batchType + "' \ 1393 # GROUP BY ra_center \ 1394 # ORDER BY numPending \ 1395 # DESC LIMIT 1" 1396 # 1397 # try: 1398 # rs = self.executeQuery(sql) 1399 # rs.first() 1400 # raCenter = rs.getFloat(1) 1401 # rs.close() 1402 # except: 1403 # self.logger.errorPair("No outstanding boxes to steal", "no stripe obtained") 1375 1404 1376 1405 # tests against ra_center need to use a finite box … … 1421 1450 GROUP BY ra_center LIMIT 1" 1422 1451 1423 try: 1424 rs = self.executeQuery(sql) 1425 rs.first() 1426 raCenter = rs.getFloat(1) 1427 rs.close() 1428 except: 1429 self.logger.errorPair("No free stripes", "trying to steal boxes from other client...") 1430 sql = "SELECT ra_center, COUNT(*) AS numPending\ 1431 FROM box \ 1432 JOIN pending ON (id = box_id) \ 1433 WHERE skychunk = '" + self.skychunk.name + "' \ 1434 GROUP BY ra_center \ 1435 ORDER BY numPending \ 1436 DESC LIMIT 1" 1437 1438 try: 1439 rs = self.executeQuery(sql) 1440 rs.first() 1441 raCenter = rs.getFloat(1) 1442 rs.close() 1443 except: 1444 self.logger.errorPair("No outstanding boxes to steal", "no stripe obtained") 1452 try: rs = self.executeQuery(sql) 1453 except: 1454 self.logger.errorPair("Problem querying for outstanding boxes", "error in sql?") 1455 print "SQL: " + sql 1456 1457 if not rs.next(): 1458 self.logger.infoPair("No outstanding boxes", "nothing to do here") 1459 rs.close() 1460 self.unlockTables() 1461 return ids 1462 1463 raCenter = rs.getFloat(1) 1464 rs.close() 1465 1466 # except: 1467 # self.logger.errorPair("No free stripes", "trying to steal boxes from other client...") 1468 # sql = "SELECT ra_center, COUNT(*) AS numPending\ 1469 # FROM box \ 1470 # JOIN pending ON (id = box_id) \ 1471 # WHERE skychunk = '" + self.skychunk.name + "' \ 1472 # GROUP BY ra_center \ 1473 # ORDER BY numPending \ 1474 # DESC LIMIT 1" 1475 # 1476 # try: 1477 # rs = self.executeQuery(sql) 1478 # rs.first() 1479 # raCenter = rs.getFloat(1) 1480 # rs.close() 1481 # except: 1482 # self.logger.errorPair("No outstanding boxes to steal", "no stripe obtained") 1445 1483 1446 1484 # tests against ra_center need to use a finite box … … 1634 1672 self.logger.infoPair("Items written to Db", "%d" % count) 1635 1673 1636 # try:1637 # self.execute("DROP TABLE all_pending_alt")1638 # except:1639 # pass1640 #1641 # self.execute("CREATE TABLE all_pending_alt (stage_id bigint(20), ra_bore float, dec_bore float)")1642 # for row in rows:1643 # try:1644 # sql = "INSERT INTO all_pending_alt (stage_id, ra_bore, dec_bore) \1645 # VALUES (%d, %f, %f)" % (row[0], row[1], row[2])1646 # print "sql 2: ", sql1647 # self.execute(sql)1648 # except: continue1649 #1650 # count = self.getRowCount("all_pending_alt")1651 # self.logger.infoPair("Items written to alt Db", "%d" % count)1652 1653 1654 1674 ''' 1655 1675 Creates a temporary table and shoves it full of stage_ids and ra/dec coords … … 1670 1690 VALUES ( '" + dvo_db + "', '" + batchType + "'," + str(row[0]) + "," + str(row[1]) + " \ 1671 1691 , " + str(row[2]) + ", '"+uniq+"' ) " 1672 print "sql 1: ", sql1692 # print "sql 1: ", sql 1673 1693 self.execute(sql) 1674 1694 except: continue … … 1683 1703 ''' 1684 1704 def getItemsInThisBox(self, minRA, maxRA, minDEC, maxDEC): 1685 1686 # the old linear size is ill-conceived1687 ## XX halfSide = self.skychunk.boxSize/2.01688 ## XX minRa = ra-halfSide1689 ## XX maxRa = ra+halfSide1690 ## XX minDec = dec-halfSide1691 ## XX maxDec = dec+halfSide1692 1705 1693 1706 ids = [] … … 1698 1711 AND dec_bore BETWEEN " + str(minDEC) + " AND " + str(maxDEC) 1699 1712 1700 print "items: ", sql 1701 1702 try: 1703 rs = self.executeQuery(sql) 1704 while (rs.next()): ids.append(rs.getInt(1)) 1713 try: 1714 rs = self.executeQuery(sql) 1715 while (rs.next()): 1716 ids.append(rs.getInt(1)) 1705 1717 except: 1706 1718 self.logger.errorPair("Can't get items in this box", sql)
Note:
See TracChangeset
for help on using the changeset viewer.
