+ seg relation table

pipelet/tracker.py

@@ -149,7 +149,7 @@ class SqliteTracker(Tracker,threading.Thread):
     Format is: 
     TODO
     """
-    
+
     def __init__(self, pipe):
         """ Initialize the Sqlite tracker. 
         
@@ -162,6 +162,8 @@ class SqliteTracker(Tracker,threading.Thread):
         ## pipeline object
         self.pipe = pipe
 
+        self.seg_id_cache = {}
+
         ## string, sql filename
         self.sqlfile = path.join(pipe._prefix,'.sqlstatus')
         if self.pipe.sqlfile is not None:
@@ -177,7 +179,7 @@ class SqliteTracker(Tracker,threading.Thread):
                 self.conn.execute('create table segments (seg_id INTEGER PRIMARY KEY, seg TEXT, curr_dir TEXT, tag TEXT, comment TEXT)')
                 self.conn.execute('create table tasks (task_id INTEGER PRIMARY KEY, seg_id INTEGER, status TEXT, input BLOB, output BLOB, str_prod TEXT)')
                 self.conn.execute('create table task_relations (father_id INTEGER, child_id INTEGER)')
-     
+                self.conn.execute('create table segment_relations (father_id INTEGER, child_id INTEGER)')
         self.conn.row_factory = sqlite3.Row
 
         self.segments_registration()
@@ -191,6 +193,7 @@ class SqliteTracker(Tracker,threading.Thread):
     def segments_registration(self):
         """
         """
+        previous = None
         for s in self.pipe.flatten():
             curr_dir = self.pipe.get_curr_dir(s)
             try:
@@ -201,12 +204,20 @@ class SqliteTracker(Tracker,threading.Thread):
                 seg_id = self.conn.execute(
                     'select seg_id from segments where curr_dir = ? limit 1',
                     (curr_dir,)).fetchone()[0]
+                self.seg_id_cache[s] = seg_id
                 logger.info("Segment %s instance (%s) already registered in db."%(s, curr_dir))
             except TypeError:
                 logger.info("Creating segment %s instance (%s)."%(s, curr_dir))
-                self.conn.execute(
+                c = self.conn.execute(
                     'insert into segments (seg, curr_dir, tag, comment) values (?, ?, ?, ?)'
-                        ,(s, curr_dir, "", docline))
+                    ,(s, curr_dir, "", docline))
+                seg_id = c.lastrowid
+                self.seg_id_cache[s] = seg_id
+                logger.info("Storing connectivity for segment %s."%s)
+                for p in self.pipe._parents[s]:
+                    self.conn.execute(
+                        'insert into segment_relations (father_id, child_id) values (?, ?)'
+                        ,(self.seg_id_cache[p], seg_id))
         self.conn.commit()
 
     def _asynchronous_request(self, sqlrequest, args):

pipelet/web.py

@@ -153,7 +153,13 @@ class Web:
         list of segid, for the upstream segment instances.
         """
         ## TO DO : get subpipes somewhere ???
-        
+        conn = sqlite3.connect(self.db_file,check_same_thread=True)
+        with conn:
+            seg, currdir = conn.execute(
+                'select father_id from segments where child_id = ?'
+                ,(segid,)).fetchall()
+            l = conn.execute('select str_prod from tasks where seg_id=? and status=?',(segid, status)).fetchall()
+        conn.close()
         return [int(segid)]
 
     @cherrypy.expose