Changeset 33354 for trunk/ippToPsps

Timestamp:

Feb 23, 2012, 4:16:51 PM (14 years ago)

Author:

rhenders

Message:

lots of new methods access new tables for clients, boxes and pending

File:

• trunk/ippToPsps/jython/ipptopspsdb.py (modified) (7 diffs)

trunk/ippToPsps/jython/ipptopspsdb.py

@@ -38 +38 @@
         try:
             rs = self.executeQuery(sql)
-            while (rs.next()):
-                ids.append(rs.getInt(1))
+            while (rs.next()): ids.append(rs.getInt(1))
+            rs.close()
         except:
             self.logger.exception("Can't query for merged batch ids in ipptopsps Db")
-
-        rs.close()
 
         self.logger.debug("Found %d merged but un-deleted items" % len(ids))
@@ -65 +63 @@
         try:
             rs = self.executeQuery(sql)
-            while (rs.next()):
-                ids.append(rs.getInt(1))
+            while (rs.next()): ids.append(rs.getInt(1))
+            rs.close()
         except:
             self.logger.exception("Can't query for merged batch ids in ipptopsps Db")
 
-        rs.close()
 
         self.logger.debug("Found %d merged but un-deleted items" % len(ids))
@@ -92 +89 @@
         try:
             rs = self.executeQuery(sql)
-            while (rs.next()):
-                ids.append(rs.getInt(1))
+            while (rs.next()): ids.append(rs.getInt(1))
+            rs.close()
         except:
             self.logger.exception("Can't query for merged batch ids in ipptopsps Db")
-
-        rs.close()
 
         self.logger.debug("Found %d merged but un-deleted items" % len(ids))
@@ -517 +512 @@
             rs.first()
             if rs.getInt(1) > 0: 
-                self.logger.errorPair(str(stage_id) + " is already being processed", "skipping")
+                self.logger.debugPair(str(stage_id) + " is already being processed", "skipping")
                 return True
             else: 
@@ -560 +555 @@
             rs.first()
             if rs.getInt(1) > 0: 
-                self.logger.errorPair(str(stage_id) + " has already been published", "skipping")
+                self.logger.debugPair(str(stage_id) + " has already been published", "skipping")
                 return True
             else: 
@@ -725 +720 @@
     Insert client
     '''
-    def insertClient(self, host, pid):
-        self.execute("INSERT INTO clients (timestamp, host, pid) VALUES (now(), '" + host + "', " + str(pid) + ")")
+    def insertClient(self, type, host, pid, config):
+        self.execute("INSERT INTO clients (timestamp, type, host, pid, config) VALUES (now(), '" + type + "', '" + host + "', " + str(pid) + ", '" + config + "')")
 
     '''
     Update clients, or inserts it for the first time
     '''
-    def updateClient(self, host, pid):
-
-        try:
-            self.insertClient(host, pid)
+    def updateClient(self, type, host, pid, config):
+
+        try:
+            self.insertClient(type, host, pid, config)
         except:
             self.execute("UPDATE clients SET timestamp = now() WHERE host = '" + host + "' AND pid = " + str(pid))
@@ -771 +766 @@
 
     '''
+    Populates Config object
+    '''
+    def readConfig(self, host, pid):
+
+        # first get config defined for this client
+        sql = "SELECT config \
+               FROM clients \
+               WHERE host = '" + host + "' \
+               AND pid = " + str(pid)
+
+        try:
+            rs = self.executeQuery(sql)
+            rs.first()
+            self.config.name = rs.getString(1)
+        except:
+            self.logger.errorPair("No config set for", "%s (pid=%d)" % (host, pid))
+            self.config.isLoaded = False
+            return False
+
+        # now load that config
+        sql = "SELECT \
+        datastore_product, \
+        datastore_type, \
+        datastore_publish, \
+        dvo_label, \
+        dvo_location, \
+        min_ra, \
+        max_ra, \
+        min_dec, \
+        max_dec, \
+        box_size, \
+        base_path, \
+        data_release, \
+        delete_local, \
+        delete_datastore, \
+        delete_dxlayer, \
+        epoch, \
+        survey, \
+        psps_survey, \
+        queue_P2, \
+        queue_ST \
+        FROM config \
+        WHERE name = '" + self.config.name + "'"
+
+        self.config.batchTypes = []
+        try:
+            rs = self.executeQuery(sql)
+            rs.first()
+            self.config.datastoreProduct = rs.getString(1)
+            self.config.datastoreType = rs.getString(2)
+            if rs.getInt(3) == 1: self.config.datastorePublishing = True
+            else: self.config.datastorePublishing = False
+            self.config.dvoLabel = rs.getString(4)
+            self.config.dvoLocation = rs.getString(5)
+            self.config.minRa = rs.getDouble(6)
+            self.config.maxRa = rs.getDouble(7)
+            self.config.minDec = rs.getDouble(8)
+            self.config.maxDec = rs.getDouble(9)
+
+            self.config.boxSize = rs.getDouble(10)
+            self.config.halfBox = self.config.boxSize/2.0
+            self.config.boxSizeWithBorder = self.config.boxSize + (self.config.BORDER * 2)
+
+            self.config.basePath = rs.getString(11)
+            self.config.dataRelease = rs.getInt(12)
+            self.config.deleteLocal = rs.getInt(13)
+            self.config.deleteDatastore = rs.getInt(14)
+            self.config.deleteDxLayer = rs.getInt(15)
+            self.config.epoch = rs.getString(16)
+            self.config.survey = rs.getString(17)
+            self.config.pspsSurvey = rs.getString(18)
+            if rs.getInt(19) == 1: self.config.batchTypes.append("P2")
+            if rs.getInt(20) == 1: self.config.batchTypes.append("ST")
+            self.config.force = False # TODO
+            self.config.test = False # TODO
+            self.config.isLoaded = True
+        except:
+            self.logger.errorPair("Could not read config with name", self.config.name)
+            self.config.isLoaded = False
+
+        return self.config.isLoaded
+
+    '''
+    Gets id of box with these coords
+    '''
+    def getBoxId(self, config, ra, dec, boxSide):
+
+        sql = "SELECT id FROM box \
+               WHERE config  = '" + config + "' \
+               AND ra_center = " + str(ra) + " \
+               AND dec_center = " + str(dec) + " \
+               AND box_side = " + str(boxSide)
+
+        id = -1
+        try:
+            rs = self.executeQuery(sql)
+            rs.first()
+            id = rs.getInt(1)
+        except:
+            pass
+
+        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))
+
+    '''
+    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)
+
+        # does not already exists so insert
+        if id < 0:
+
+            sql = "INSERT INTO box \
+                   (config, ra_center, dec_center, box_side) \
+                   VALUES \
+                   ('" + config + "', " + str(ra) + ", " + str(dec) + ", " + str(boxSide) + ")"
+
+            self.execute(sql)
+            id = self.getBoxId(config, ra, dec, boxSide)
+
+        return id
+
+    '''
+    Inserts new box ids into the pending table 
+    '''
+    def insertPending(self, box_id, batchType, ids):
+
+        # first delete old pending items
+        self.execute("DELETE FROM pending WHERE box_id = " + str(box_id))
+
+        for id in ids:
+            sql = "INSERT INTO pending \
+               (box_id, batch_type, stage_id) \
+               VALUES \
+               (" + str(box_id) + ", '" + batchType + "', " + str(id) + ")"
+
+            self.execute(sql)
+
+    '''
+    Returns ids for all boxes with pending items
+    '''
+    def getBoxIds(self, config):
+
+        sql = "SELECT DISTINCT id \
+               FROM box \
+               JOIN pending ON (id = box_id)  \
+               WHERE config = '" + config + "' \
+               ORDER BY ra_center, dec_center"
+
+        ids = []
+        try:
+            rs = self.executeQuery(sql)
+            while (rs.next()): ids.append(rs.getInt(1))
+            rs.close()
+        except:
+            self.logger.errorPair("Can't get boxes", sql)
+
+        return ids
+
+    '''
+    Returns dimensions of this box
+    '''
+    def getBoxDimensions(self, boxId):
+
+        sql = "SELECT ra_center, dec_center, box_side FROM box WHERE id = " + str(boxId)
+
+        boxDim = {}
+        rs = self.executeQuery(sql)
+        rs.first()
+
+        boxDim['RA'] = rs.getFloat(1)
+        boxDim['DEC'] = rs.getFloat(2)
+        boxDim['SIDE'] = rs.getFloat(3)
+
+        return boxDim
+
+    '''
+    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)
+
+        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)
+
+        return ids
+
+    '''
+    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 \
+                       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)
+            rs.close()
+        except:
+            self.logger.errorPair("Can't get density plot data", sql)
+
+    '''
     Destructor
     '''
     def __del__(self):
 
-        self.logger.debug("IppToPspsDb Desstructor")
+        self.logger.debug("IppToPspsDb Destructor")
         self.stmt.close()
         self.con.close()