Changeset 37403 for branches/eam_branches/ps2-tc3-20130727/ippToPsps/jython/ipptopspsdb.py

Timestamp:

Sep 19, 2014, 4:05:27 PM (12 years ago)

Author:

eugene

Message:

merge changes from trunk

Location:

branches/eam_branches/ps2-tc3-20130727

Files:

• . (modified) (1 prop)
• ippToPsps/jython/ipptopspsdb.py (modified) (16 diffs)

branches/eam_branches/ps2-tc3-20130727/ippToPsps/jython/ipptopspsdb.py

@@ -166 +166 @@
 
         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 +671 @@
             and not self.consistentlyFailed(batchType, stageID)): 
             
-            self.logger.infoPair("heather:","passed logic")
+            # self.logger.infoPair("heather:","passed logic")
 
             sql = "INSERT INTO batch ( \
@@ -803 +688 @@
                        '" + self.skychunk.name + "' \
                        )"
-            self.logger.infoPair("heather:","sql")
+ 
+            # self.logger.infoPair("heather:",sql)
 
             self.execute(sql)
@@ -816 +702 @@
             except:
                 self.logger.exception("Unable to get batch ID")
+                self.logger.infoPair("sql:",sql)
+
                 batchID = -1
 
@@ -1017 +905 @@
        response = raw_input(" * Name for new skychunk, or existing skychunk to edit? ")
        if response == "": return
+       response = response.strip()
        self.skychunk.name = response
 
@@ -1056 +945 @@
            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 + "'")
        
@@ -1116 +1008 @@
         ,queue_ST \
         ,queue_OB \
+        ,queue_DF \
+        ,queue_DO \
+        ,queue_FW \
+        ,queue_FO \
         ,parallel \
         ,P2_smf_version \
@@ -1125 +1021 @@
 
         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")
-
-            self.skychunk.force = True # TODO
-            self.skychunk.parallel = False # TODO
-
-            if rs.getInt(22) == 1: self.skychunk.parallel = True
-            self.skychunk.P2_smf_version = rs.getString(23)
-            self.skychunk.ST_cmf_version = rs.getString(24)
-            self.skychunk.trange_start = rs.getString(25)
-            self.skychunk.trange_end = rs.getString(26) 
-
-            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
@@ -1300 +1252 @@
     def insertPending(self, box_id, batchType, ids):
 
-        # print "starting insert pending"
+        print "starting insert pending"
 
         # first delete old pending items
@@ -1306 +1258 @@
                 WHERE box_id = " + str(box_id) + " \
                 AND batch_type = '" + batchType + "'")
-
-        # print "deleted old items"
 
         for id in ids:
@@ -1315 +1265 @@
                (" + 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
 
     '''
@@ -1340 +1293 @@
                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
@@ -1413 +1373 @@
                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
@@ -1626 +1595 @@
         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
+    '''
+    def storeAllItemsInDvodb(self, rows, dvo_db, batchType):
+
+        #try:
+        #    self.execute("DROP TABLE all_pending")
+        #except:
+        #    pass
+
+        #self.execute("CREATE TEMPORARY TABLE all_pending (stage_id bigint(20), ra_bore float, dec_bore float)")
+        for row in rows:
+
+            try:
+                uniq = dvo_db + "."+ batchType + "." + str(row[0])
+                sql = "INSERT INTO dvodb (dvo_db, batch_type, stage_id, ra_center, dec_center, uniq_key) \
+                       VALUES ( '" + dvo_db + "', '" + batchType + "'," + str(row[0]) + "," + str(row[1]) + " \
+                       , " + str(row[2]) + ", '"+uniq+"' ) " 
+                # print "sql 1: ", sql
+                self.execute(sql)
+            except: continue
+
+        #count = self.getRowCount("all_pending")
+        #self.logger.infoPair("Items written to Db", "%d" % count)
+
+
 
     '''
@@ -1647 +1626 @@
     '''
     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 = []
@@ -1662 +1634 @@
                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)