Index: trunk/ippToPsps/jython/ipptopspsdb.py
===================================================================
--- trunk/ippToPsps/jython/ipptopspsdb.py	(revision 33354)
+++ trunk/ippToPsps/jython/ipptopspsdb.py	(revision 33490)
@@ -141,5 +141,5 @@
     want to load a test batch to PSPS
 
-    We also check that these batches are more than 4 HOURs old, so that we don't attempt to 
+    We also check that these batches are more than 6 HOURs old, so that we don't attempt to 
     publish something simulataneously with the client that it producing it
     '''
@@ -154,5 +154,5 @@
                AND loaded_to_datastore != 1 \
                AND timestamp > '2011-10-27' \
-               AND timestamp < now() -  INTERVAL 4 HOUR"
+               AND timestamp < now() -  INTERVAL 6 HOUR"
 
         ids = []
@@ -161,8 +161,7 @@
             while (rs.next()):
                 ids.append(rs.getInt(1))
+            rs.close()
         except:
             self.logger.exception("Can't query for processed batch ids not loaded to datastore ipptopsps Db")
-
-        rs.close()
 
         self.logger.debug("Found %d processed but-not-on-datastore items" % len(ids))
@@ -274,6 +273,6 @@
         except:
             self.logger.exception("Can't query for batch ids with " + column + " = " + str(value) + " in ipptopsps Db")
-
-        rs.close()
+        finally:
+            rs.close()
 
         self.logger.debug("Found %d batches with %s = %d" % (len(ids), column, value))
@@ -284,5 +283,5 @@
     Returns the total detections published to the datastore for this epoch, dvo label and batch type
     '''
-    def getTotalDetectionsPublished(self, batchType, epoch, dvoGpc1Label):
+    def getTotalDetectionsPublished(self, batchType, epoch, dvoGpc1Label, interval="10 YEAR"):
 
         sql = "SELECT SUM(total_detections) \
@@ -292,5 +291,6 @@
                AND dvo_db = '" + dvoGpc1Label + "' \
                AND processed = 1 \
-               AND loaded_to_datastore = 1"
+               AND loaded_to_datastore = 1 \
+               AND timestamp BETWEEN now() - INTERVAL " + interval + " AND now()"
 
         total = -1
@@ -301,6 +301,6 @@
         except:
             self.logger.exception("Can't query for total detections published")
-
-        rs.close()
+        finally:
+            rs.close()
 
         self.logger.debug("Found %d detections" % total)
@@ -372,10 +372,18 @@
     '''
     def getStages(self):
-        return ['processed', 'loaded_to_datastore', 'loaded_to_ODM', 'merge_worthy', 'deleted_datastore', 'deleted_dxlayer', 'merged', 'deleted_local' ]
+        return [
+        'processed', 
+        'loaded_to_datastore', 
+        'loaded_to_ODM', 
+        'merge_worthy', 
+        'deleted_datastore', 
+        'deleted_dxlayer', 
+        'merged', 
+        'deleted_local' ]
 
     '''
     Returns a count of batches processed in the specified time period.
     '''
-    def countBatchesInLastPeriod(self, batchType, epoch, dvoGpc1Label, interval):
+    def countBatches(self, batchType, epoch, dvoGpc1Label, interval="10 YEAR"):
 
         sql = "SELECT COUNT(batch_id) \
@@ -393,4 +401,6 @@
             self.logger.exception("Unable to count batches in interval")
             return 0
+        finally:
+            rs.close()
 
     '''
@@ -497,7 +507,7 @@
 
     '''
-    Is someone processing this item right now? Checks whether this was attempted within the last 2 HOUR
-    '''
-    def processingNow(self, batchType, stage_id, epoch, dvoGpc1Label):
+    Is someone processing this item right now? Checks whether this was attempted within the last 6 HOUR
+    '''
+    def processingNow(self, batchType, stage_id):
 
         sql = "SELECT COUNT(*) \
@@ -505,6 +515,7 @@
                WHERE stage_id = " + str(stage_id) + " \
                AND batch_type = '" + batchType + "' \
-               AND dvo_db = '" + dvoGpc1Label + "' \
-               AND timestamp BETWEEN now() - INTERVAL 4 HOUR AND now()"
+               AND dvo_db = '" + self.config.dvoLabel + "' \
+               AND batch_type != 'IN' \
+               AND timestamp BETWEEN now() - INTERVAL 6 HOUR AND now()"
 
         try:
@@ -541,5 +552,5 @@
     Have we already processed this stage_id?
     '''
-    def alreadyProcessed(self, batchType, stage_id, epoch, dvoGpc1Label):
+    def alreadyProcessed(self, batchType, stage_id):
 
         sql = "SELECT COUNT(*) \
@@ -547,6 +558,7 @@
                WHERE stage_id = " + str(stage_id) + " \
                AND batch_type = '" + batchType + "' \
-               AND timestamp > '" + epoch + "' \
-               AND dvo_db = '" + dvoGpc1Label + "' \
+               AND timestamp > '" + self.config.epoch + "' \
+               AND dvo_db = '" + self.config.dvoLabel + "' \
+               AND batch_type != 'IN' \
                AND processed = 1"
 
@@ -565,5 +577,5 @@
     Has this stage_id consistently failed to process?
     '''
-    def consistentlyFailed(self, batchType, stage_id, epoch, dvoGpc1Label):
+    def consistentlyFailed(self, batchType, stage_id):
 
         sql = "SELECT COUNT(*) \
@@ -571,6 +583,6 @@
                WHERE stage_id = " + str(stage_id) + " \
                AND batch_type = '" + batchType + "' \
-               AND timestamp > '" + epoch + "' \
-               AND dvo_db = '" + dvoGpc1Label + "' \
+               AND timestamp > '" + self.config.epoch + "' \
+               AND dvo_db = '" + self.config.dvoLabel + "' \
                AND processed = -1"
 
@@ -620,14 +632,14 @@
 
     '''
-    Creates a new batch. This is done with a locked table in order that no two clients can start work on the same item
-    '''
-    def createNewBatch(self, batchType, stageID, config):
+    Creates a new batch. This is done with a locked table in order that no two clients can start work on the same item, i.e. this is a crital section
+    '''
+    def createNewBatch(self, batchType, stageID):
 
         batchID = -1;
 
-        if config.force or \
-            (not self.alreadyProcessed(batchType, stageID, config.epoch, config.dvoLabel) \
-            and not self.processingNow(batchType, stageID, config.epoch, config.dvoLabel) \
-            and not self.consistentlyFailed(batchType, stageID, config.epoch, config.dvoLabel)): 
+        if self.config.force or \
+            (not self.alreadyProcessed(batchType, stageID) \
+            and not self.processingNow(batchType, stageID) \
+            and not self.consistentlyFailed(batchType, stageID)): 
 
             sql = "INSERT INTO batch ( \
@@ -640,7 +652,7 @@
                        '" + batchType + "', \
                        " + str(stageID) + ", \
-                       '" + config.survey + "', \
-                       '" + config.dvoLabel + "', \
-                       '" + config.datastoreProduct + "' \
+                       '" + self.config.survey + "', \
+                       '" + self.config.dvoLabel + "', \
+                       '" + self.config.datastoreProduct + "' \
                        )"
 
@@ -712,4 +724,24 @@
 
     '''
+    Pause all loading clients
+    '''
+    def pauseAllLoaders(self, bool):
+        self.setColForAllLoaders("paused", bool)
+
+    '''
+    Kill all loading clients
+    '''
+    def killAllLoaders(self, bool):
+        self.setColForAllLoaders("killed", bool)
+
+    '''
+    Sets the value for a given column in the clients table
+    '''
+    def setColForAllLoaders(self, col, bool):
+        if bool: value = 1
+        else: value = 0
+        self.execute("UPDATE clients SET " + col + " = " + str(value) + " WHERE type = 'loader.py' AND config = '" + self.config.name + "'")
+
+    '''
     Remove client
     '''
@@ -720,14 +752,14 @@
     Insert client
     '''
-    def insertClient(self, type, host, pid, config):
-        self.execute("INSERT INTO clients (timestamp, type, host, pid, config) VALUES (now(), '" + type + "', '" + host + "', " + str(pid) + ", '" + config + "')")
+    def insertClient(self, type, host, pid):
+        self.execute("INSERT INTO clients (timestamp, type, host, pid, config) VALUES (now(), '" + type + "', '" + host + "', " + str(pid) + ", '" + self.config.name + "')")
 
     '''
     Update clients, or inserts it for the first time
     '''
-    def updateClient(self, type, host, pid, config):
-
-        try:
-            self.insertClient(type, host, pid, config)
+    def updateClient(self, type, host, pid):
+
+        try:
+            self.insertClient(type, host, pid)
         except:
             self.execute("UPDATE clients SET timestamp = now() WHERE host = '" + host + "' AND pid = " + str(pid))
@@ -763,5 +795,4 @@
 
         return False
-
 
     '''
@@ -851,11 +882,11 @@
     Gets id of box with these coords
     '''
-    def getBoxId(self, config, ra, dec, boxSide):
+    def getBoxId(self, ra, dec):
 
         sql = "SELECT id FROM box \
-               WHERE config  = '" + config + "' \
+               WHERE config  = '" + self.config.name + "' \
                AND ra_center = " + str(ra) + " \
                AND dec_center = " + str(dec) + " \
-               AND box_side = " + str(boxSide)
+               AND box_side = " + str(self.config.boxSize)
 
         id = -1
@@ -869,27 +900,16 @@
         return id
 
-
     '''
     Remove all boxes and associated ids in pending table
     '''
-    def removeAllBoxes(self, config):
-        self.execute("DELETE FROM box WHERE config = '" + config + "'")
-
-    '''
-    Removes a box and it's associated ids in the pending table
-    '''
-    def removeBox(self, config, ra, dec, boxSide):
-        self.execute("DELETE FROM box \
-                WHERE config = '" + config + "' \
-                AND ra_center = " + str(ra) + " \
-                AND dec_center = " + str(dec) + " \
-                AND box_side = " + str(boxSide))
+    def removeAllBoxes(self):
+        self.execute("DELETE FROM box WHERE config = '" + self.config.name + "'")
 
     '''
     Inserts new box into box table. If it already exisits, it returns existing box id 
     '''
-    def insertBox(self, config, ra, dec, boxSide):
-
-        id = self.getBoxId(config, ra, dec, boxSide)
+    def insertBox(self, ra, dec):
+
+        id = self.getBoxId(ra, dec)
 
         # does not already exists so insert
@@ -899,8 +919,8 @@
                    (config, ra_center, dec_center, box_side) \
                    VALUES \
-                   ('" + config + "', " + str(ra) + ", " + str(dec) + ", " + str(boxSide) + ")"
+                   ('" + self.config.name + "', " + str(ra) + ", " + str(dec) + ", " + str(self.config.boxSize) + ")"
 
             self.execute(sql)
-            id = self.getBoxId(config, ra, dec, boxSide)
+            id = self.getBoxId(ra, dec)
 
         return id
@@ -912,5 +932,7 @@
 
         # first delete old pending items
-        self.execute("DELETE FROM pending WHERE box_id = " + str(box_id))
+        self.execute("DELETE FROM pending \
+                WHERE box_id = " + str(box_id) + " \
+                AND batch_type = '" + batchType + "'")
 
         for id in ids:
@@ -923,15 +945,57 @@
 
     '''
-    Returns ids for all boxes with pending items
-    '''
-    def getBoxIds(self, config):
-
+    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 getStripeBoxIds(self, host, pid, batchType):
+
+        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 config = '" + self.config.name + "' \
+               AND batch_type = '" + batchType + "' \
+               AND ra_center NOT IN \
+               (SELECT ra_center \
+                FROM stripe \
+                WHERE config = '" + self.config.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 available stripes", "sharing with other client")
+            sql = "SELECT ra_center, COUNT(*) AS numPending\
+                   FROM box \
+                   JOIN pending ON (id = box_id) \
+                   WHERE config = '" + self.config.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("Still no available stripes", "")
+
+        self.logger.infoPair("Using RA stripe", "%f" % raCenter)
         sql = "SELECT DISTINCT id \
                FROM box \
-               JOIN pending ON (id = box_id)  \
-               WHERE config = '" + config + "' \
-               ORDER BY ra_center, dec_center"
-
-        ids = []
+               JOIN pending ON (id = box_id) \
+               WHERE ra_center = " + str(raCenter) + " \
+               AND batch_type = '" + batchType + "' \
+               AND config = '" + self.config.name + "'"
+
         try:
             rs = self.executeQuery(sql)
@@ -939,7 +1003,39 @@
             rs.close()
         except:
-            self.logger.errorPair("Can't get boxes", sql)
-
-        return ids
+            self.logger.errorPair("Can't get stripe boxes", sql)
+
+        self.registerStripe(host, pid, raCenter)
+
+        self.unlockTables()
+
+        return ids
+
+    '''
+    Inserts row into strip table
+    '''
+    def registerStripe(self, host, pid, raCenter):
+       
+        # first, delete any existing record
+        try:
+            self.execute("DELETE FROM stripe \
+                    WHERE client_id = \
+                    (SELECT id FROM clients \
+                     WHERE host = '" + host + "' AND pid = " + str(pid) + ")")
+        except:
+            self.logger.errorPair("Can't delete from stripe table", sql)
+
+
+        # now insert new stripe entry
+        sql = "INSERT INTO stripe \
+               (client_id, config, ra_center) \
+               SELECT clients.id, clients.config, " + str(raCenter) + " \
+               FROM clients \
+               WHERE host = '" + host + "' \
+               AND pid = " + str(pid)
+        try:
+            self.execute(sql)
+        except:
+            self.logger.errorPair("Can't insert into stripe", sql)
+
 
     '''
@@ -961,9 +1057,9 @@
 
     '''
-    Returns ids for pending items in this box for this batch type
-    '''
-    def getPendingIds(self, boxId, batchType):
-
-        sql = "SELECT stage_id FROM pending WHERE batch_type = '" + batchType + "' AND box_id = " + str(boxId)
+    Returns ids for pending items for this config
+    '''
+    def getPendingIds(self):
+
+        sql = "SELECT DISTINCT stage_id FROM pending JOIN box ON (box_id = id) WHERE config = '" + self.config.name + "'"
 
         ids = []
@@ -973,4 +1069,21 @@
             rs.close()
         except:
+            self.logger.errorPair("Can't get pending ids for this config", sql)
+
+        return ids
+
+    '''
+    Returns ids for pending items in this box for this batch type
+    '''
+    def getPendingIdsForThisBox(self, boxId, batchType):
+
+        sql = "SELECT stage_id FROM pending WHERE batch_type = '" + batchType + "' AND box_id = " + str(boxId)
+
+        ids = []
+        try:
+            rs = self.executeQuery(sql)
+            while (rs.next()): ids.append(rs.getInt(1))
+            rs.close()
+        except:
             self.logger.errorPair("Can't get ids from this box", sql)
 
@@ -978,42 +1091,98 @@
 
     '''
-    Returns max number of pensing items in a box for this config
-    '''
-    def getMaxPendingInBox(self, config):
-
-        sql = "SELECT MAX(pending) \
-               FROM (SELECT COUNT(stage_id) AS pending \
+    Deletes an item from the pending table
+    '''
+    def deletePendingItem(self, batchType, stageID):
+
+        sql = "DELETE pending \
+               FROM pending \
+               JOIN box ON (id = box_id) \
+               WHERE stage_id = " + str(stageID) + " \
+               AND batch_type = '" + batchType + "' \
+               AND config = '" + self.config.name + "'"
+
+        self.execute(sql)
+
+    '''
+    Writes density-plot data to file for pending stuff over sky for given config
+    '''
+    def createPendingDensityPlotData(self, batchType, DATFILE):
+
+        sql = "SELECT id, ra_center, dec_center \
+               FROM box \
+               WHERE config = '" + self.config.name + "' \
+               ORDER BY ra_center, dec_center"
+
+        maxPending = 0
+        try:
+            rs = self.executeQuery(sql)
+            while (rs.next()):
+                
+                boxId = rs.getInt(1)
+
+                sql = "SELECT COUNT(stage_id) \
                        FROM pending \
-                       JOIN box ON (id = box_id) \
-                       WHERE config = '" + config + "' \
-                       GROUP BY box_id) AS t"
-
-        try:
-            rs = self.executeQuery(sql)
-            rs.first()
-            return rs.getInt(1)
-        except:
-            self.logger.errorPair("Could not count max pending", sql)
-            return 100
-
-    '''
-    Writes density-plot data to file for pending stuff over sky for given config
-    '''
-    def createPendingDensityPlotData(self, config, DATFILE):
-
-        sql = "SELECT ra_center, dec_center, COUNT(stage_id) \
-               FROM box \
-               LEFT JOIN pending ON (id = box_id) \
-               WHERE config = '" + config + "' \
-               GROUP BY id \
-               ORDER BY ra_center, dec_center"
-
-        try:
-            rs = self.executeQuery(sql)
-            while (rs.next()):
-                print >> DATFILE, rs.getFloat(1), rs.getFloat(2), rs.getInt(3)
+                       WHERE box_id = " + str(boxId) + " \
+                       AND batch_type = '" + batchType + "'"
+                rs2 = self.executeQuery(sql)
+                rs2.first()
+                max = rs2.getInt(1)
+                rs2.close() 
+                print >> DATFILE, rs.getFloat(2), rs.getFloat(3), max
+                if max > maxPending: maxPending = max
             rs.close()
         except:
             self.logger.errorPair("Can't get density plot data", sql)
+
+        return maxPending
+
+    '''
+    Creates a temporary table and shoves it full of stage_ids and ra/dec coords
+    '''
+    def storeAllItems(self, rows):
+
+        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)")
+        count = 0
+        for row in rows:
+
+            try:
+                sql = "INSERT INTO all_pending (stage_id, ra_bore, dec_bore) \
+                       VALUES (%d, %f, %f)" % (row[0], row[1], row[2]) 
+                self.execute(sql)
+                count += 1
+            except: continue
+
+        self.logger.infoPair("All items written to Db", "%d" % count)
+
+    '''
+    Gets all items in the all_pending temporary table within the bounds of this box
+    '''
+    def getItemsInThisThisBox(self, ra, dec):
+
+        halfSide = self.config.boxSize/2.0
+        minRa =  ra-halfSide
+        maxRa = ra+halfSide
+        minDec = dec-halfSide
+        maxDec = dec+halfSide
+
+        ids = []
+
+        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)
+
+        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)
+
+        return ids
 
     '''