Changeset 35235 for branches/eam_branches/ipp-20130207/ippToPsps/jython/ipptopspsdb.py

Timestamp:

Mar 5, 2013, 2:23:09 PM (13 years ago)

Author:

eugene

Message:

merge changes from trunk

Location:

branches/eam_branches/ipp-20130207/ippToPsps

Files:

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

branches/eam_branches/ipp-20130207/ippToPsps/jython/ipptopspsdb.py

@@ -4 +4 @@
 import sys
 import os
+import math
 import logging
 
@@ -997 +998 @@
 
             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.halfBox = self.skychunk.boxSize/2.0
+            # self.skychunk.boxSizeWithBorder = self.skychunk.boxSize + (self.skychunk.BORDER * 2)
 
             self.skychunk.basePath = rs.getString(11)
@@ -1031 +1032 @@
     '''
     Gets id of box with these coords
+    XXX EAM : this code implies we can have multiple boxes at the same location for the same
+    XXX EAM : skychunk distinguished by having different sizes.  that seems quite bad
     '''
     def getBoxId(self, ra, dec):
+
+        # for the float comparison, need to use a small window
+        minRA = ra - 0.01
+        maxRA = ra + 0.01
+        minDEC = dec - 0.01
+        maxDEC = dec + 0.01
 
         sql = "SELECT id FROM box \
                WHERE skychunk  = '" + self.skychunk.name + "' \
-               AND ra_center = " + str(ra) + " \
-               AND dec_center = " + str(dec) + " \
-               AND box_side = " + str(self.skychunk.boxSize)
-
+               AND ra_center BETWEEN " + str(minRA) + " AND " + str(maxRA) + " \
+               AND dec_center BETWEEN " + str(minDEC) + " AND " + str(maxDEC)
+
+               # AND box_side = " + str(self.skychunk.boxSize)
+
+        # print "sql for boxid: ", sql
+       
         id = -1
         try:
@@ -1079 +1091 @@
     Inserts new box into box table. If it already exisits, it returns existing box id 
     '''
-    def isBoxIngested(self, id):
+    def isBoxIngested(self, id, host):
 
         # set the field 'ingested' to 1 for this box
@@ -1088 +1100 @@
                 print "no matching boxes for ", str(id)
                 raise
-            status = rs.getInt(1)
+            value = rs.getString(1)
+        except:
+            print "failed to find box for ", str(id)
+            raise
+
+        if value == host:
+            return 1
+
+        if value == "none":
+            return 0
+
+        if value is None:
+            return 0
+
+        print "box ingested host ", value, " does not match db host ", host
+        raise
+
+    '''
+    Inserts new box into box table. If it already exisits, it returns existing box id 
+    '''
+    def setIngestedBox(self, id, host):
+
+        # set the field 'ingested' to 1 for this box
+        sql = "update box set ingested = '" + host + "' where id = " + str(id)
+        try:
+            self.execute(sql)
         except:
             print "failed to set box ", str(id), "as ingested"
             raise
 
-        return status
-    '''
-    Inserts new box into box table. If it already exisits, it returns existing box id 
-    '''
-    def setIngestedBox(self, id):
+    '''
+    Clear ingested flag for boxes owned by this host
+    '''
+    def clearIngestedBoxes(self, host):
 
         # set the field 'ingested' to 1 for this box
-        sql = "update box set ingested = 1 where id = " + str(id)
+        sql = "update box set ingested = 'none' where ingested = '" + host + "'"
         try:
             self.execute(sql)
         except:
-            print "failed to set box ", str(id), "as ingested"
+            print "failed to clear ingested state for boxes owned by host ", host
             raise
 
     '''
-    Inserts new box into box table. If it already exisits, it returns existing box id 
-    '''
-    def clearIngestedBoxes(self):
-
-        # set the field 'ingested' to 1 for this box
-        sql = "update box set ingested = 0"
-        try:
-            self.execute(sql)
-        except:
-            print "failed to clear boxes as ingested"
-            raise
-
-    '''
     Inserts new box ids into the pending table 
     '''
     def insertPending(self, box_id, batchType, ids):
+
+        # print "starting insert pending"
 
         # first delete old pending items
@@ -1129 +1154 @@
                 WHERE box_id = " + str(box_id) + " \
                 AND batch_type = '" + batchType + "'")
+
+        # print "deleted old items"
 
         for id in ids:
@@ -1136 +1163 @@
                (" + str(box_id) + ", '" + batchType + "', " + str(id) + ")"
 
+            # print "sql: ", sql
             self.execute(sql)
 
@@ -1168 +1196 @@
             rs.close()
         except:
-            self.logger.errorPair("No free stripes", "trying to share with other client...")
+            self.logger.errorPair("No free stripes", "trying to steal boxes from other client...")
             sql = "SELECT ra_center, COUNT(*) AS numPending\
                    FROM box \
@@ -1184 +1212 @@
                 rs.close()
             except:
-                self.logger.errorPair("No available stripes to share", "no stripe obtained")
+                self.logger.errorPair("No outstanding boxes to steal", "no stripe obtained")
+
+        # tests against ra_center need to use a finite box
+        minRA = raCenter - 0.01
+        maxRA = raCenter + 0.01
 
         self.logger.infoPair("Using RA stripe", "%f" % raCenter)
@@ -1190 +1222 @@
                FROM box \
                JOIN pending ON (id = box_id) \
-               WHERE ra_center = " + str(raCenter) + " \
+               WHERE ra_center BETWEEN " + str(minRA) + " AND " + str(maxRA) + " \
                AND batch_type = '" + batchType + "' \
                AND skychunk = '" + self.skychunk.name + "' \
@@ -1203 +1235 @@
 
         self.registerStripe(host, pid, raCenter)
+
+        self.unlockTables()
+
+        return ids
+
+    '''
+    Returns ids for all boxes with pending items in a stripe of RA not already being processed
+    by another client. If none are available, then it uses the stripe already used by another 
+    client that has the most pending items
+    '''
+    def getStripeBoxIdsAnyBatch(self, host, pid):
+
+        self.execute("LOCK TABLES box READ, pending READ, stripe WRITE, clients WRITE")
+
+        ids = []
+        raCenter = -1.0
+        sql = "SELECT ra_center \
+               FROM box \
+               JOIN pending ON (id = box_id) \
+               WHERE skychunk = '" + self.skychunk.name + "' \
+               AND ra_center NOT IN \
+               (SELECT ra_center \
+                FROM stripe \
+                WHERE skychunk = '" + self.skychunk.name + "') \
+               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")
+
+        # tests against ra_center need to use a finite box
+        minRA = raCenter - 0.01
+        maxRA = raCenter + 0.01
+
+        self.logger.infoPair("Using RA stripe", "%f" % raCenter)
+        sql = "SELECT DISTINCT id \
+               FROM box \
+               JOIN pending ON (id = box_id) \
+               WHERE ra_center BETWEEN " + str(minRA) + " AND " + str(maxRA) + " \
+               AND skychunk = '" + self.skychunk.name + "' \
+               ORDER BY dec_center DESC"
+
+        try:
+            rs = self.executeQuery(sql)
+            while (rs.next()): ids.append(rs.getInt(1))
+            rs.close()
+        except:
+            self.logger.errorPair("Can't get stripe boxes", sql)
 
         self.unlockTables()
@@ -1250 +1349 @@
         boxDim['DEC'] = rs.getFloat(2)
         boxDim['SIDE'] = rs.getFloat(3)
+
+        # SIDE is the linear full-width of the box
+
+        # XXX EAM : need to do something special at the poles...
+        if (boxDim['DEC'] > 85): raise
+
+        # a given box centered at RA,DEC with (linear) size of SIDE has the following bounds
+        dR = 0.5*boxDim['SIDE'] / math.cos(math.radians(boxDim['DEC']))
+        dD = 0.5*boxDim['SIDE']
+
+        boxDim['minRA']  = boxDim['RA']  - dR
+        boxDim['maxRA']  = boxDim['RA']  + dR
+        boxDim['minDEC'] = boxDim['DEC'] - dD
+        boxDim['maxDEC'] = boxDim['DEC'] + dD
 
         return boxDim
@@ -1363 +1476 @@
     Gets all items in the all_pending temporary table within the bounds of this box
     '''
-    def getItemsInThisThisBox(self, ra, dec):
-
-        halfSide = self.skychunk.boxSize/2.0
-        minRa =  ra-halfSide
-        maxRa = ra+halfSide
-        minDec = dec-halfSide
-        maxDec = dec+halfSide
+    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 = []
@@ -1375 +1489 @@
         sql = "SELECT DISTINCT stage_id \
                FROM all_pending \
-               WHERE ra_bore BETWEEN " + str(minRa) + " AND " + str(maxRa) + " \
-               AND dec_bore BETWEEN " + str(minDec) + " AND " + str(maxDec)
+               WHERE ra_bore BETWEEN " + str(minRA) + " AND " + str(maxRA) + " \
+               AND dec_bore BETWEEN " + str(minDEC) + " AND " + str(maxDEC)
 
         try: