## Copyright (C) 2008, 2009, 2010 APC LPNHE CNRS Universite Paris Diderot <lejeune@apc.univ-paris7.fr>  <betoule@apc.univ-paris7.fr>
##
## This program is free software; you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation; either version 3 of the License, or
## (at your option) any later version.
##
## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
## GNU General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with this program; if not, see http://www.gnu.org/licenses/gpl.html

from __builtin__ import zip

def cross_prod(*args):
    """ Return the cartesian product of the input sets.

    >>> print(len(cross_prod(['a','b'])))
    2
    >>> print(len(cross_prod(['a','b'],[1,2,3])))
    6
    >>> print(len(cross_prod(['a','b'],[1,2,3],[0.1,0.2,0.3,0.4])))
    24
    """
    res = []
    if len(args) == 1:
        return zip(*args)
    l1 = args[0]
    l2 = cross_prod(*(args[1:]))
    if l1 and l2:
        for a in l1:
            res.extend([(a,) + b for b in l2])
    elif l1:
        res.extend(zip(l1))
    elif l2:
        res.extend(l2)
    return res

def union(*args):
    """ Return the union of the input sets.

    >>> print(len(union(['a','b'])))
    2
    >>> print(len(union(['a','b'],[1,2,3])))
    5
    >>> print(len(union(['a','b'],[1,2,3],[0.1,0.2,0.3,0.4])))
    9
    """
    """
    """

    l = []
    for l_ in args:
        l.extend(zip(l_))
    return l

def _group(l, code):
    """Group entries of l (prod, id, parent name) matching expression code.
    
    Returns list of tuple (prod, list of parent id)
    """
    c = compile(code, '<string>', 'eval')
    ## l is a list of tuple (prod, id , parent name)
    ## apply group by directive using a dictionnary of name:prod as local
    values = [eval(c, {}, dict([(e[2], e[0]) for e in t])) for t in l]
    classes = set(values)  # reduce to unique list of product
    ## build a dictionnary prod:list of id
    d = {}
    for e in classes:  # for each class init dict
        d[e] = []
    # l and values have same size. zip match input with eval result
    for t, v in zip(l, values):
        # add new contrib to class
        d[v].extend([e[1] for e in t if e[1] is not None])
    ## how id can be None ?
    return d.items()  # return dict as a list of tuple (prod, list of id)

def _where(code):
    """ Return a selection function for entries matching expression code.
    """

    def condition(x):
        ## x is a list of tuple (prod, id, parent name)
        for p in x:
            vars()[p[2]] = p[0]  # build a local dict {parent name:prod}
        return eval(c)  # apply selection condition return True or False
    return condition

## 
##
##

import threading
from utils import str_date, make_dict
from contextlib import closing
import pickle
import multiplex

class Task:
    """ A segment code associated with its input/output product(s). 
    
    A task is the association of a segment with its input product, its
    execution status and its output product(s).
    """

    def __init__(self, seg, task_input=[], status=None, task_output=[], id=None, queued_on=None, parents=[], str_input=None):
        """ Initialize a task object. 
        
        """

        ## string, a segment name
        self.seg = seg
        ## pythonn object, an input product
        self.task_input = task_input
        ## string, task status (done, queued, failed, running)
        self.status = status
        ## list, task output product(s)
        self.task_output = task_output
        ## integer, a uniq id for the task attributed by the tracker
        self.id = id
        ## String, store the result of the computation
        self.tag = None
        ## Dict of module:version to save on disk at the segment level
        self.version = {}
        ## date_time object
        self.queued_on = queued_on
        self.begun_on = None
        self.ended_on = None
        ## List of the task id whose output become the input of this task
        self.parents = parents
        ## Dict of str_input of the parent tasks
        self.str_parents = {}
        ## Dict of parameters to save on disk at the segment level
        self.args = {}
        ## store the disk path
        self.str_input = str_input

    def __str__(self):
        """ Convert task object to string. 
        """
        return 'seg: %s, status: %s, id: %s, input: %s, output: %s' \
            % (self.seg, self.status, self.id, str(self.task_input), str(self.task_output)) 

    def update_status(self, status):
        """ Update task status.

        """
        str_d = str_date()
        if status == 'done':
            self.ended_on = str_d
        elif status == 'queued':
            self.queued_on = str_d
        else:
            self.task_output = None
            if status == 'running':
                self.begun_on = str_d

    def store_meta(self, fn):
        """ Store meta information for task. 
        
        This is used to rebuild db. 
        
        """
        with closing(file(fn, 'w')) as f:
            r = pickle.dump(dict({'id':self.id, 'parents':self.str_parents, 'queued_on':self.queued_on, 'ended_on':self.ended_on, 'begun_on':self.begun_on, 'input':pickle.dumps(self.task_input), 'output':pickle.dumps(self.task_output), 'status':self.status}),f)
            
class TaskList:
    """ List of accomplished tasks per segment.
    """

    def __init__(self):
        """ Initialize a task list. 
        """
        ## dict, task list
        self._list={}
        ## lock on task list
        self._void_task_lock = threading.Lock()

    def pop(self, seg):
        """ Remove and return stored products of a segment. 
        
        """
        with self._void_task_lock:
            try:
                return self._list[seg]
            except KeyError:
                return None

    def push(self, t):
        """ Add a task to the task list. 
        
        Parameters
        ----------
        t : task object. 
        """
        if not isinstance(t, Task):
            raise TypeError
        if not self._list.has_key(t.seg):
            self._list[t.seg] = []
        self._list[t.seg].append(t)

    def multiplex (self, seg, parent_list, directive):
        """ Compute the result of a multiplex directive

        Multiplex method is one of 'cross_prod', 'union', 'zip'.
   
        """
        method = eval("multiplex.%s"%directive.method)
        
        a = [] ## as many as parent_list
        for p in parent_list:
            b = []
            try:
                lst_p = self._list[p]
            ## no input for orphan segment
                lst_p = []
            b = [(o, t.id, p) for t in lst_p for o in t.task_output]
            #for t in lst_p:
            #    b = b+ [(o, t.id, p) for o in t.task_output]
            a.append(b)
        output_set = method (*a) ## output set is a tuple (prod, list of parent id, segname)

        if directive.where:
            f = multiplex._where(directive.where) ## get selection function
            output_set = filter(f, output_set) ## apply selection

        if directive.group_by:
            output_set = multiplex._group(output_set, directive.group_by)
            lst_task = [Task(seg, e[0], status='queued', parents=list(set(e[1]))) for e in output_set]
        ## make the list of task to push, input are dictionnaries
        else:
            lst_task = [Task(seg, make_dict([(r[2],r[0]) for r in e]), status='queued', parents=list(set([r[1] for r in e if r[1] is not None]))) for e in output_set]

        return lst_task

try:
    import sqlite3
except ImportError:
    import pysqlite2 as sqlite3
from glob import glob
import os
import shutil
def old_to_new (old_file, new_file):
    """ Convert an old database to the new database format."""
    conn1 = sqlite3.connect(old_file)
    conn2 = sqlite3.connect(new_file)
    # Creates tables in the new database
    with conn2:
        conn2.execute('create table segments (seg_id INTEGER PRIMARY KEY, seg TEXT, curr_dir TEXT, tag TEXT, comment TEXT)')
        conn2.execute('create table tasks (task_id INTEGER PRIMARY KEY, seg_id INTEGER, status TEXT, input BLOB, output BLOB, str_prod TEXT)')
        conn2.execute('create table task_relations (father_id INTEGER, child_id INTEGER)')
    with conn1:
        l = conn1.execute('select distinct seg, curr_dir from tasks').fetchall()
        for e in l:
            conn2.execute('insert into segments (seg, curr_dir) values (?, ?)', e)
    ls = conn2.execute('select seg_id, curr_dir from segments').fetchall()
        for seg_id, curr_dir in ls:
            l = conn1.execute('select status, prod, res, str_prod from tasks where curr_dir==?',(curr_dir,))
            for t in l:
                conn2.execute('insert into tasks (seg_id, status, input, output, str_prod) values (?, ?, ?, ?, ?)',(seg_id,)+t)

def change_root(sql_file,old_prefix, new_prefix):
    """ Update the data root path in the database."""
    conn1 = sqlite3.connect(sql_file)
        conn1.execute('update segments set curr_dir = replace(curr_dir, ?,?)', (old_prefix, new_prefix))
        conn1.execute('update tasks set str_input = replace(str_input, ?,?)', (old_prefix, new_prefix))

def deltask(db_file, lst_task, report_only=False):
    """ Delete a tasks instances.

    Delete all products directories of a tasks instance. 
    
    """
    import shutil
    # remove everyboby
    conn = sqlite3.connect(db_file,check_same_thread=True)
    while lst_task: 
        taskid = lst_task[0]
        with conn:
            print "removing task %s from db"%taskid
            str_input  = conn.execute('select str_input from tasks where task_id = ?',(int(taskid),)).fetchone()[0]
        
            lst_task.remove(taskid)
            
            # delete from tasks_relations
            if not report_only:
                l = conn.execute('delete from task_relations where child_id = ? ',(int(taskid),))
            # mark child tasks for deletion
            children = conn.execute('select child_id from task_relations where father_id = ? ',(int(taskid),)).fetchall()
            # delete from tasks
                l = conn.execute('delete from tasks where task_id = ?',(int(taskid),))
                print 'Task %s removed from db'%taskid
            else:
                print 'Task %s would be removed from db'%taskid
        #conn.close()
        try:
            print "Removing directory %s"%str_input
                shutil.rmtree(str_input)
                print "%s removed"%str_input
            else:
                print "%s would be removed"%str_input
        except OSError:
            print "Failed to remove %s"%str_input
        if children:
            print "Adding children of %s"%taskid
            lst_task += [c[0] for c in children]
            lst_task = list(set(lst_task))
            lst_task.sort()
    conn.close()
    #if lst_task:
    #    deltask(db_file, lst_task, report_only=report_only)

def get_lst_tag (db_file):
    """ Return the list of existing tags
    
    Tags are ; separated in the db. 
    
    """
    conn.text_factory=str
    lst = []
    with conn:
        l = conn.execute('select tag from segments').fetchall()
        for s in l:
            if s[0] is not None:
                str_tag = s[0].split(";")
                for e in str_tag:
                    if e:
                        lst.append(e)
    return lst

def get_lst_seg (db_file):
    """ Return the list of segment name
    
    """
    lst = []
    with conn:
        l = conn.execute('select distinct seg from segments').fetchall()
        lst = [s[0] for s in l]
    return lst

def get_lst_date (db_file):
        """ Return the list of existing dates

        Date strings are picked from queued_on field corresponding to the
        first task of each segment.
        
        """
        conn = sqlite3.connect(db_file,check_same_thread=True)
        conn.text_factory=str
        with conn:
            l = conn.execute('select queued_on from tasks group by queued_on').fetchall()
        return l[0]

def add_tag (db_file, segid, tag):
    """ Add new tag to the database
    
    Check that the tag do not exist yet for this id
    
    """
    seglst = segid.split('-')
    for segid in seglst:
        if segid:
            with conn:
                l = conn.execute('select tag,curr_dir from segments where seg_id=?',(segid,)).fetchone()
                lst_tag = []
                if l[0] is not None and l:
                    lst_tag = l[0].split(";")
                lst_tag.append(tag)
                lst_tag = list(set(lst_tag))
                str_tag = ";".join(lst_tag)
                conn.execute('update segments set tag=? where seg_id=?',(str_tag,segid))
            fn = glob(os.path.join(l[1], "*.meta"))
            _update_meta(fn, str_tag)

def get_last(db_file):
    """ Return the last computed segment in a given database.
    
    Return a seg-id

    """
    conn = sqlite3.connect(db_file, check_same_thread=True)
    l = conn.execute('select seg_id,max(ended_on) from tasks').fetchone()
    return l[0]

def _update_meta(fn, str_tag):
        """ Update meta file with current tag value. 
        
        fn: string,  meta file name
        tag: string, tag string
        """
        if fn:
            with closing(file(fn[0], 'r')) as f:
                d = pickle.load(f)
            d['tag'] = str_tag
            with closing(file(fn[0], 'w')) as f:    
                r = pickle.dump(d,f)
def del_tag (db_file, tag):
        """ Delete tag from the database

        """
        print tag
        with conn:
            l = conn.execute('select seg_id, tag, curr_dir from segments where tag == ?',(tag,)).fetchall()
            for s in l:
                lst_tag = s[1].split(";")
                print lst_tag
                lst_tag.remove(tag)
                conn.execute('update segments set tag=? where seg_id=?',(str_tag,s[0]))
                fn = glob(os.path.join(s[2], "*.meta"))
                _update_meta(fn, str_tag)

def _delseg(db_file, lst_seg):
        """ Delete a pipeline instance. 

        Delete all segments and products directories of a pipeline instance. 
        
        """
        ## remove everyboby
        segid = lst_seg[0]
        with conn:
            currdir  = conn.execute('select curr_dir from segments where seg_id = ?',(int(segid),)).fetchone()[0]
            l = conn.execute('select seg_id from segments where curr_dir like ?',(currdir+'%',)) 
            for e in l:
                lst_seg.remove(e[0])

            ## delete from tasks_relations
            l = conn.execute('delete from task_relations where child_id in (select task_id from tasks where seg_id in (select seg_id from segments where segments.curr_dir like ?))',(currdir+'%',))
            ## delete from tasks
            l = conn.execute('delete from tasks where seg_id in (select seg_id from segments where segments.curr_dir like ?)',(currdir+'%',))
            ## delete from segments_relations
            l = conn.execute('delete from segment_relations where child_id in (select seg_id from segments where segments.curr_dir like ?)',(currdir+'%',))
            ## delete from segments
            l = conn.execute('delete from segments where curr_dir like ?',(currdir+'%',))
                
        conn.close()
        try:
            shutil.rmtree(currdir)
        except:
            pass
        if lst_seg:
            _delseg(db_file, lst_seg)

def _deltask(db_file, lst_task):
        """ Delete a tasks instances.

        Delete all products directories of a tasks instance. 
        
        """
        with conn:

            l = conn.execute('delete from task_relations where child_id = ? ',(int(taskid),))
            l = conn.execute('delete from tasks where task_id = ?',(int(taskid),))
        try:
            shutil.rmtree(str_input)
        except:
            pass
        if lst_task:
            _deltask(db_file, lst_task)

def _get_fathers(db_file, segid):
        """ Append upstream pipeline instance paths to the current path.
        
        Return a list which contains all upstream segment instances
        id for a given segment instance. This is used to print a pipeline 
        tree view with all dependencies. 

        """
        lstid = [int(segid)]
        conn  = sqlite3.connect(db_file,check_same_thread=True)
        with conn:
            fids = conn.execute(
                'select father_id from segment_relations where child_id = ?'
                ,(segid,)).fetchall()
        if fids:
            for l in fids:
                lstid += _get_fathers(db_file,l[0])
        return lstid

def _get_children(db_file, segid):
        """ Return the list of pipeline instance paths which depend on seg
        
        Return a list which contains all downstream segment instances
        id for a given segment instance. This is used to delete all dependencies. 

        """
        with conn:
                'select child_id from segment_relations where father_id = ?'
        if fids:
                lstid += _get_children(db_file, l[0])

def _get_children_task(db_file, taskid):
        """ Return the list of tasks which depend on task.
        
        Return a list which contains all downstream task instances
        id for a given task instance. This is used to delete all dependencies. 

        """
        lstid = [int(taskid)]
        with conn:
                'select child_id from task_relations where father_id = ?'
                ,(taskid,)).fetchall()
        if fids:
                lstid += _get_children_task(db_file, l[0])

## 
##
##

import os.path as path
import subprocess
from datetime import datetime
import logging
from utils import str_web
import cPickle as pickle
from pipelet.utils import init_logger, close_logger
from pipelet.repository import get_git_revision
import traceback
import signal
import gc
import inspect
import pipelet

class NullHandler(logging.Handler):
    """ Extension of the logging handler class. 
    """
    def emit(self, record):
        """ Avoid warnings. 
        """
        pass

class EnvironmentBase():
    """ Basic segment's facilities.
    
    An environment contains all segment's facilities provided to the
    user. The base class is almost empty, and should be extended to
    define a completely new environment. The default class can be
    extended to complete the default environment with new utilities.
    """

    def __init__(self, w):
        """ Initialize the base environment with task input. 
        
        w: a worker instance
        """
        self._worker    = w
        self._seg_input = w.task.task_input

    def _get_data_fn(self, x, seg=None):
        """ Complete the filename with the path to the working
        directory.
        
        x: string, filename suffix
        
        Returns
        -------
        string, filename
        """
        if seg is None:
            return path.join(self._worker.pipe.get_data_dir(self._worker.task.seg, prod=(self._worker.task.task_input, self._worker.task.id)), x)
        else:
            lst = [path.join(y, x) for y in self._worker.task.str_parents[seg]]
            if len(lst)==1:
                lst = lst[0]
            return lst

    def _get_namespace(self):
        """ Prepare the namespace for a segment.

        Returns
        ------- 
        Dictionnary, namespace containing all the
        public methods of the class.
        
        """
        glo = {}
        for k in dir(self):
            if k[0] != "_":
                glo[k] = getattr(self, k)
        return glo
    
    def _get_log_file(self):
        """ Return the log file name
        
        """
        return self._get_data_fn(self._worker.task.seg+".log")

    def _hook(self, hook_name, glo):
        """ Execute hook code.
        
        Search for an extra segment code file, and update dictionnary
        with the result of its execution.

        hook_name: string, hook name
        glo: dict, global dictionnary to update. 
        
        """
        code = self._worker.pipe.repository.get_hook_string(self._worker.task.seg, hook_name)
        if code:
            exec(compile(code[0], code[1], "exec"), glo)
        # Other exception occuring in the segment are catched by default
        else:
            print "No hook file named %s for seg %s"%(self._worker.task.seg,hook_name)
        
    def _close(self, glo):
        """ Return segment's output from dictionnary.

        glo: segment execution dictionnary
        """
        return None

class Environment(EnvironmentBase):
    """ Default segment's facilities.

    The default environment contains some basic utilities related to filenames, parameters and subprocesses.
    It can be extended to complete the default environment with new utilities.
    """
    
        """

        ## worker instance
        self._worker = w

        ## segment input
        self.seg_input = w.task.task_input

        ## segment output
        self.seg_output = []

        ## list of temporary files
        self._tmpf   = []

        ## logger instance
        self.logger = init_logger (self._get_data_fn(""), self._get_log_file(), level=[])

        ## list of param to save
        self.lst_par = []
        
        ## list of param to expose
        self.lst_tag = []

        ## dictionnary with external dependencies version or revision number
        self.version = {"pipelet":pipelet.__version__}

    def get_input(self, seg=None):
        """ Return the input coming from segment seg. 
        
        If no segment specified, take the first. This utility replaces
        the seg_input variable which type could vary as described
        below.

        >>> get_input(seg='third')
        the_brave
        """
        inp = None
        if isinstance(self.seg_input, dict):
            if seg is not None:
                if self.seg_input.has_key(seg):
                    inp= self.seg_input[seg]
                else:
                    self.logger.info("No input from parent named %s", seg)
            else:
                if len(self.seg_input.values())>0:
                    inp= self.seg_input.values()[0]
                else:
                    self.logger.info("Empty input list")
        elif isinstance(self.seg_input, list) and  len(self.seg_input)==0:
            self.logger.info("Empty input list")
        else:
            inp= self.seg_input
        return inp

    def set_output(self, lst_o):
        """ Set the segment output as a list. 
        
        If o is not a list, set a list of one element o.
        """ 
        if not isinstance (lst_o, list):
            self.seg_output=[lst_o]
        else:
            self.seg_output=lst_o

    def get_data_fn(self, x, seg=None):
        
        >>> get_data_fn ("test.png")
        ./test/first_test/first_S3CRKVY/data/test.png
        """
        return self._get_data_fn(x, seg)

    def hook(self, hook_name, glo):
        
        """
        self.logger.info ("hooking %s"%hook_name)
        return self._hook(hook_name, glo)
    
    def get_parent(self):
        """ Return list of parent names

        >>> get_parent()
        ['second', 'third'] 
        """
        str_parents = self._worker.task.str_parents
        return str_parents.keys()

    def glob_parent(self, y, segs=None):
        """ Globbing limited to the fatherhood. For unlimited globbing see glob_seg.

        >>> glob_parent ('test*.txt')
        ['./test/first_test/first_S3CRKVY/second_WDSDP3Q/data/7_Robin/test.txt']
        """
        if segs is None:
            segs = str_parents.keys()
        elif isinstance(segs,str):
            segs = [segs]
        res=[]
        if not isinstance (segs, list):
            self.logger.critical ("can't convert string to list")
        for x in segs:
            segx = self._worker.pipe.find_seg(self._worker.task.seg, x)
            if segx is None:
                self.logger.warning("No parent segment matching %s found"%x)
            if not str_parents[segx]:
                res+=glob(path.join(self._worker.pipe.get_data_dir(segx),y))
            else:
                for t in str_parents[segx]:
                    res+=glob(path.join(self._worker.pipe.get_data_dir(segx),t,y))
        return res

    def glob_seg(self, x, y):
        """ Return the list of filename matching y in the working directory of segment x.
        
        Usage of glob_seg should be limited:
        + potentially breaks the dependancy scheme
        + may hurt performances as all task directories of the segment x will be searched

        >>> glob_seg ('second', 'test*.txt')
        ['./test/first_test/first_S3CRKVY/second_WDSDP3Q/data/8_Arthur/test.txt', './test/first_test/first_S3CRKVY/second_WDSDP3Q/data/7_Robin/test.txt']
        """
        
        segx = self._worker.pipe.find_seg(self._worker.task.seg, x)
        if segx is None:
            self.logger.warning("No parent segment matching %s found"%x)
        return glob(path.join(self._worker.pipe.get_data_dir(segx),y))\
            +glob(path.join(self._worker.pipe.get_data_dir(segx),path.join('*/',y)))

    def logged_subprocess(self, args, shell=False, except_on_failure=True, name=None):
        """ Execute a subprocess and log its output.
        
        Create files process_name.log and process_name.err. If shell is set to True, the command is run via a shell. If except_on_failure set to True, raise an exception when command execution return non zero
        
        >>> (o,e)=logged_subprocess (["echo", "$TMP"])
        >>> open(o).readlines()
        ['$BASH']
        """
        proc = args[0]
        if name is None:
            name = path.basename(proc)
        of = self.get_data_fn(name+'.log')
        ef = self.get_data_fn(name+'.err')
        o = file(of,'w')
        e = file(ef,'w')
        e.write('#'+' '.join([str(a) for a in args])+'\n')
        e.flush()
        p=subprocess.Popen(args, stdout=o, stderr=e, shell=shell)
        p.communicate()[0]
        if except_on_failure:
            if p.returncode != 0:
                raise subprocess.CalledProcessError(p.returncode, args[0])
        o.close()
        e.close()
        return (of, ef)

    def get_tmp_fn(self):
        """ Obtain a temporary filename
        
        Note : has to be part of the segment execution environment
        The temporary file is added to the intern list for future removal. 
        
        >>> get_tmp_fn()
        ./test/first_test/third_QA7CX5A/fourth_LGJSKVQ/735200
        """
        tf = path.join(self._worker.work_dir, str(datetime.now().toordinal()))
        self._tmpf.append(tf)
        return tf

    def load_param(self, seg,glo,param_name='*'):
        """ Update the global dictionnary with parameters of a
        segment.
        
        >>> load_param('second', globals(), ["fn"])
        >>> print fn
        test.txt
        """
        segx = self._worker.pipe.find_seg(self._worker.task.seg, seg)
            self.logger.warning("No parent segment matching %s found"%seg)
        if not path.exists(self._worker.pipe.get_param_file(segx)):
            return
        self.load_products(self._worker.pipe.get_param_file(segx), glo, param_name=param_name)

    def save_products(self, filename, glo, param_name='*'):
        """ Use pickle to save a part of a given namespace.
        
        >>> save_products(get_data_fn("test.pkl"), globals(),['i'])
        >>> open(get_data_fn("test.pkl")).readlines()
        ['(dp1'\\n, "S'i'\\n", 'I0\\n', 's.']
        """
        new_dict = {}
        if param_name == '*':
            param_name = glo.keys()
        for k in param_name:
            try:
                new_dict[k] = glo[k]
                self.logger.warning('Fail to save object %s in file %s'%(k,filename))
        f = file(filename,'w')
        pickle.dump(new_dict,f)
        f.close()
        
    def load_products(self, filename, glo, param_name='*'):
        """ Update a namespace by unpickling requested object from the
        file.
        
        >>> load_products(get_data_fn("test.pkl", seg="second"), globals(),['i'])
        >>> print i
        0
        """
        try:
            f = file(filename)
            new_dict  = pickle.load(f)
            f.close()
        except IOError:
            self.logger.warning('No such file: %s'%filename)
        except pickle.UnpicklingError:
            self.logger.warning('Failed to unpickle from file: %s'%filename)
            f.close()
            param_name = new_dict.keys()
            try:
                glo[k] = new_dict[k]
                self.logger.warning( 'Fail to load object %s from file %s'%(k,filename) )

    def clean_tmp(self):
        """ Delete the list of tmp file created by a segment
        """
        for e in self._tmpf:
            subprocess.Popen(['rm', e]).communicate()[0]
        self._tmpf = []

    def _make_tag(self, seg, glo):
        """ Write a tag using parameters of a segment. 
        
        """
        strtag = ''
        if len(self.lst_tag)==0:
            try:
                self.lst_tag = glo['lst_tag']
                self.lst_tag = []
        for param in self.lst_tag:
            if param in glo:
                strtag = strtag + param + '=' + str_web(glo[param])+'    '
            else:
                self.logger.warning( 'parameter '+param+' not in dictionary')
        var_tag = strtag + ' <small>(<b>'+ datetime.today().strftime("%e %m - %R")+'</b>)</small>   '
        return var_tag

    def _save_param(self, glo,param_name='*'):
        """ Save parameters of a segment.
        
        """
            try:
                self.logger.warning('Fail to save object %s '%(k))
        return new_dict
        
    def _save_version(self, glo):
        """ Save version information for all modules
        
        """
        import types
        for k,v in glo.items():
            if isinstance(v, types.ModuleType):
                if hasattr(v, '__version__'):
                    self.version[k] = v.__version__
                elif hasattr(v, '__file__'):
                    v_path = path.dirname(v.__file__)
                    if path.exists(path.join(v_path, '.git')):
                        self.version[k] = get_git_revision (v_path)
        if len(self._worker.pipe.repository._rev)>0:
            self.version[self._worker.task.seg] = self._worker.pipe.repository._rev
        return self.version

    def save_param (self, lst_par):
        """ Set list of parameter to save on disk
        """
        
        if not isinstance(lst_par, list):
            lst_par = [lst_par]
        self.lst_par += lst_par
        
    def expose (self, lst_tag):
        """ Set list of parameter to expose from the web interface. 
        """
        if not isinstance(lst_tag, list):
            lst_tag = [lst_tag]
        self.lst_tag += lst_tag

        """ Close environment. 

        This routine performs post processing like saving tag,
        parameters, etc.
        """
        seg = self._worker.task.seg
        self.clean_tmp()
        param = None
        if len(self.lst_par)==0:
            try: # Save params
                param = self._save_param(glo, param_name=glo['lst_par'])
                self.logger.warning( 'Nothing to save in param file for seg %s' % seg)
            except Exception:
                self.logger.warning( 'Fail to save the param file for seg %s' % seg)
        else:            
            param = self._save_param(glo, param_name=self.lst_par)

        self._worker.task.args    = param ## pickle .args
        self._worker.task.version = self._save_version(glo) ## as comment at the beginning of the code copy
        self._worker.task.tag     = self._make_tag(seg, glo) ## string in database + meta data
        
        if len(self.seg_output) == 0:
            try:
                res = glo["seg_output"]
            except:
                res = []
                self.logger.info("No segment output found, setting seg_output to None")
        else:
            res = self.seg_output
        close_logger (self.logger)
        glo.clear() ## empty segment workspace
        return  res

class SandBoxEnv(Environment):
    def _save_param(self, seg, glo,param_name='*'):
        pass
    
        pass
        
        """
        tf = os.tmpnam()
        return tf

        """
        self.logger = logging.getLogger('sandbox')

## 
##
##

try:
import cherrypy
import os.path
import os
current_dir = os.path.dirname(os.path.abspath(__file__))
import auth
from cherrypy.lib import profiler
from pipelet.treeview import TreeView
from pipelet.pbview import PBView

class PipeIndex():
    """ A pipeline index Web interface.
    
    A pipeline index Web interface prints the lists of available pipelines 
    found in a given path. 
    """
    html_tmp = """
    <html>
    <head>
    <SCRIPT SRC="static/mktree.js" LANGUAGE="JavaScript"></SCRIPT>
    <SCRIPT SRC="static/cook.js" LANGUAGE="JavaScript"></SCRIPT>
    <SCRIPT SRC="static/tag.js" LANGUAGE="JavaScript"></SCRIPT>
    <LINK REL="stylesheet" HREF="static/mktree.css">  
    </head><body>
    """
    
        """ Initialize the Web interface. 
        """
        ## list of pipelines
        import new
        self.pipelist = []
        if self.get_config(cherrypy.config['config_file']):
            for pipe, dbfile in self.get_config(cherrypy.config['config_file']).iteritems():
                view = self.which_view(cherrypy.config['config_file'], pipe)
                setattr(self, pipe, view(os.path.expanduser(dbfile),pipe))
                self.pipelist.append(pipe)

    def which_view(self, config_file, pipe):
        """
        """
        from cherrypy.lib import unrepr
        import ConfigParser
        c = ConfigParser.ConfigParser()
        c.read(config_file)
        if c.has_section('Views'):
            for o in c.options('Views'):
                if o==pipe:
                    return eval(unrepr(c.get('Views', o)))
        return TreeView

    def which_sqlfile(self, path_info):
        """ Return data base filename
        path_info read from configuration file .pipelet. 
        
        """
        try:
            pipe = path_info.split('/')[1]
            if pipe in self.pipelist:
                return getattr(self, pipe).db_file
        except KeyError:
            return None

    @cherrypy.expose
    def index(self):
        """ Pipeline index. 
        
        """
        self.__init__()
        html = self.html_tmp 
        html +='<h1>Pipeline Index </h1>'
        html +='<div class="list" id="pipelist"><h2>Select a pipeline to browse</h2><ul>'
        for k in self.pipelist:
            html += '<li> <a href="%s/"> %s</a></li>'%(k,k)
        html += '</ul></div></body></html>'
        return html

    def get_config(self, config_file):
        """ Get the list of pipelines. 
        """
        if c.has_section('Pipelines'):
            d = {}
            for o in c.options('Pipelines'):
                d[o] = unrepr(c.get('Pipelines', o))
            return d

class MyApp(cherrypy.Application):
    """ Pipelet web interface application. 
    """

    def __init__(self, domain, domaindir, script_name="", config=None, pipeview=TreeView):
        """ Initialize the web application. 
        
        script_name is the URL "mount point" for the application
        config is a dict of {path: pathconf} pairs, where 'pathconf' is itself a dict of {key: value} pairs
        """
        ## string
        self.domain = domain
        ## string
        self.domaindir = domaindir

        ## top most object of the nested set which define the URL space 
        root = PipeIndex()
        if root.self.get_config(cherrypy.config['config_file']):
            for pipe, dbfile in root.self.get_config(cherrypy.config['config_file']).iteritems():
                setattr(root, pipe, pipeview(os.path.expanduser(dbfile)))
        cherrypy.Application.__init__(self, root, script_name, config={'/':{}})

def start(config, fastcgi=False):
    """Start the web server in daemon mode.

    Store the PID in the file specified in config.

    """
    from cherrypy.process.plugins import Daemonizer, PIDFile
    
    cherrypy.config.update(config)
    app = cherrypy.tree.mount(PipeIndex(),"",config)
    if hasattr(app, 'toolboxes'):
        # CherryPy 3.1+
        app.toolboxes['pipeauth'] = auth.pipeauth
    
    if fastcgi:
        from flup.server.fcgi import WSGIServer
        cherrypy.engine.start()
        try:
            WSGIServer(app).run()
        finally:
            # This ensures that any left-over threads are stopped as well.
            cherrypy.engine.stop()
    else:
        d = Daemonizer(cherrypy.engine)
        d.subscribe()
        p = PIDFile(cherrypy.engine, config['global']['server.pidfile'])
        p.subscribe()
    
def stop(config):
    """Stop the web server by sending the TERM signal to the daemon process.

    Look for the PID in the file specified in config.

    """
    import signal
    pidf = open(config['global']['server.pidfile'])
    pid = int(pidf.read())
    pidf.close()
    os.kill(pid, signal.SIGTERM)
    
def update_config(pipename, pipepath, config_file, delete=False, view=None):
    """ Add a new Pipeline entry to the config file.

    If the config file does not exist, create a new one.
    if delete is True remove pipe from list
    """
    import ConfigParser
    c = ConfigParser.ConfigParser()
    c.read(config_file)
    if not c.has_section('Pipelines'):
        c.add_section('Pipelines')
    if not c.has_section('Views'):
        c.add_section('Views')   
    if delete:
        c.remove_option('Pipelines', pipename)
        c.remove_option('Views', pipename)
    elif view is not None:
        if view in ['TreeView', 'PBView']:
            c.set('Views', pipename, '"'+view+'"')
        else:
           print view+ " not available" 
    else:
        if os.path.isfile(pipepath):
            pipepath = os.path.expanduser(pipepath)
            pipepath = os.path.abspath(pipepath)
            c.set('Pipelines', pipename, '"'+pipepath+'"')
        else:
            print pipepath+ " not found"
    f=file(config_file,'w')
    c.write(f)
    f.close()

def main():
    import optparse
    parser = optparse.OptionParser(usage="\nTo start/stop the web server:\n  %prog start | stop\nTo track/untrack a pipeline under the name <name> with the web interface:\n  %prog track | untrack <name> </path/to/sqlfile>\nTo change web interface view type of a tracked pipeline under the name <name>:\n  %prog view <name> <viewname> (in [TreeView, PBView])")
    parser.add_option('-H', '--host',
                      help='serve this address',default='0.0.0.0')
    parser.add_option('-c', '--config-file',
                      help='Read pipelet configuration from this file',default=os.path.expanduser('~/.pipelet/pipelet.conf'))
    parser.add_option('-p', '--port',
                      help='port the server is listenning to', default=8080, type='int')
    parser.add_option('-n', '--no-daemon',
                      help='Run the web server interactively instead of starting a daemon, log errors and access to the screen. Stop with C-c.', default=False, action='store_true')
    parser.add_option('-k', '--krenew',
                      help='In deamon mode ask and renew afs token.', default=False, action='store_true')
    parser.add_option('-l', '--error-file',
                      help='Logging file for errors when run in daemon mode',
                      default=os.path.expanduser('~/.pipelet/pipelet.errors'))
    parser.add_option('-a', '--access-file',
                      help='Logging file for access when run in daemon mode',
                      default=os.path.expanduser('~/.pipelet/pipelet.access'))
    parser.add_option('-A', '--authentication-type',
                      help='What kind of authentication is performed by cherrypy, can be one of ["Digest", "Basic", "ACL", "None"]',
                      default="Digest")
    parser.add_option('-i', '--pid-file',
                      help='Store the pid in daemon mode',
                      default=os.path.expanduser('~/.pipelet/pipelet.pid'))
    parser.add_option('-f', '--fastcgi',
                      help='Run in fastCGI mode', default=False, action='store_true')

    (options, args) = parser.parse_args()

    ## make .pipelet directory
    try:
        os.mkdir (os.path.dirname(options.config_file))
    except:
        pass

    config = {'/static':
                  {'tools.staticdir.on': True,
                   'tools.staticdir.dir': os.path.join(current_dir,'static'),
                   },
              'global':{'server.socket_port':options.port, 'server.socket_host':options.host,
                        'server.thread_pool':10,
                        'log.screen':False, 
                        'log.error_file': options.error_file,
                        'log.access_file':options.access_file,
                        'server.pidfile':options.pid_file,
                        'pipeauth.read_access.auth_type':options.authentication_type,
                        'pipeauth.write_access.auth_type':options.authentication_type,
                        'config_file':options.config_file,              
                        }
              }
    
    if options.no_daemon:
        config['global']['log.screen'] = True
        cherrypy.config.update(config)
        cherrypy.config.update(options.config_file)
        app = cherrypy.tree.mount(PipeIndex(),"",config)
        if hasattr(app, 'toolboxes'):
            # CherryPy 3.1+
            app.toolboxes['pipeauth'] = auth.pipeauth
        cherrypy.engine.block()
        exit(0)
    if options.krenew:
        exit(0)
    if len(args) < 1:
        parser.print_usage()
        exit(-1)
    if args[0] == 'start':
        start(config, fastcgi=options.fastcgi)
    elif args[0] == 'stop':
        stop(config)
    elif args[0] == 'restart':
        restart()
    elif args[0] == 'track':
        if len(args) != 3:
            parser.print_usage()
            exit(-1)
        update_config(args[1], args[2], options.config_file)
    elif args[0] == 'untrack':
        if len(args) < 2:
            exit(-1)
        update_config(args[1], "", options.config_file, delete=True)
    elif args[0] == 'view':
            exit(-1)
        update_config(args[1], "", options.config_file, view=args[2])
    else:
        exit(-1)
    
if __name__ == "__main__":
    main()

## Copyright (C) 2008, 2009, 2010 APC LPNHE CNRS Universite Paris Diderot <lejeune@apc.univ-paris7.fr> <betoule@apc.univ-paris7.fr>
##
##
##

import os

from utils import get_hashkey, crc32
from utils import str_file, flatten
from environment import Environment
from repository import LocalRepository
import re

class PipeException(Exception):
    """ Extension of exception class.
    """
    pass

def _str_join(strs):
    """ Print a node.

    >>> print(_str_join(('sega -> segb','segc','segd')))
    sega -> segb
            segc
            segd
    """
    l = [len(max(a.splitlines(), key=len)) for a in strs]
    if l:
        lmax = max(l)
    else:
        lmax = 0
    s = ""
    for a, l1 in zip(strs, l):
        s += '\n'.join([' ' * (lmax - l1) + i for i in a.splitlines()])
        s += ('\n')
    return s[:-1]

def _str_red(a, b):
    """ Print a sequence.

    >>> print(_str_red('segb->segc', 'sega'))
    sega->segb->segc
    """
    if b:
        return b + '->' + a
    else:
        return a

class Pipeline:
    """ A segment oriented graph.

    A pipeline is a list of code segments with their corresponding
    repository and dependency tree.

    Setting matplotlib to True will turn the matplotlib backend to Agg
    in order to allow the execution in non-interactive environment.
    """

    def __init__(self, seg_list, code_dir='./',
                 prefix='./', sqlfile=None,
                 matplotlib=False, matplotlib_interactive=False,
                 env=Environment, permissive=False):
        """ Initialize a pipeline.

        seg_list : either a list of string (the list of segment for a
        linear pipeline), or a string description of the pipeline
        using a dot-like syntax.
        code_dir : string, the location of the segment source code
        prefix : string, indicates where to save the pipeline products.
        sqlfile: string, sql data base
        matplotlib: boolean, if true, turn the matplotlib backend to Agg
        in order to allow the execution in non-interactive environment.
        matplotlib_interactive: same thing for interactive workers
        env: extension of the EnvironmentBase class.
        permissive: If set no error raise for segment not found.

        >>> p = Pipeline('')
        >>> print p._curr_dirs, p._parents, p._children
        {} {} {}
        >>> p = Pipeline('a;', code_dir=r'/test/', permissive=True)
        >>> print p._curr_dirs, p._parents, p._children # doctest: +ELLIPSIS
        {'a': '.../test/a_AAAAAAA'} {'a': []} {'a': []}
        """
        if isinstance(seg_list, str):
            self.from_dot(seg_list)
        else:
            self._make_tree(seg_list)
        ## string, the location of the segment source code
        self.repository = None
        if code_dir:
            self.repository = LocalRepository(self._seg_list,
                                              code_dir,
                                              permissive=permissive)

        ## string, indicates where to save the pipeline products.
        os.umask(002)
        self._prefix = path.realpath(prefix)
        if not os.path.exists(prefix):
            os.mkdir(prefix)
        logdir = os.path.join(prefix, "log")
        if not os.path.exists(logdir):
            os.mkdir(logdir)

        ## dict, directory corresponding to each segment
        self._curr_dirs = {}
        ## dict, hash key corresponding to each segment
        self._hashes = {}
        self.compute_hash()

        ## string, sql data base
        if sqlfile is None:
            sqlfile = path.join(prefix, ".sqlstatus")
        self.sqlfile = path.expanduser(path.expandvars(sqlfile))

        ## boolean, if true, turn the matplotlib backend to Agg
        self.matplotlib = matplotlib

        self.matplotlib_interactive = matplotlib_interactive

        ## string, pipe name.
        self.name = "pipelet"

        ## Environment Base extension
        self.env  = env

        ## input dictionnary (for orphan segments)
        self._input = {}

        """ Print the segment tree.

        >>> print(Pipeline('first->second->fourth\\nthird->fourth',
        ...                code_dir = '../test/', permissive=True)
        ...                ) #doctest: +NORMALIZE_WHITESPACE
        first -> second -> 
                  third -> fourth -> 
        """
        s = ""
        for k, v in self._children.iteritems():
            if not v:
                s += self.strseg(k)
        return s

    def to_dot(self, filename=None):
        """ Return a string representation of the graph.

        Writen using a subsample of the graphviz dot syntax.

        >>> p = Pipeline("")
        >>> p.from_dot('''a->b->c;
        ... d->c;''')
        >>> print p.to_dot()
        digraph pipelet {
        b -> c;
        d -> c;
        a -> b;
        }
        """
        s = "digraph %s {\n" % self.name
        for k, v in self._parents.iteritems():
            for p in v:
                s += '%s -> %s;\n' % (p, k)
        s += '}'
        if filename:
            with closing(file(filename, 'w')) as f:
                f.write(s)
        return s

    def from_dot(self, s):
        """ Parse a subset of the graphviz dot language.

        >>> p.from_dot('a;')
        >>> print p._parents == {'a':[]}, p._children == {'a':[]}
        True True
        >>> p.from_dot('''a->b;
        ... c->b;
        ... c->d;''')
        >>> print p._children 
        {'a': ['b'], 'c': ['b', 'd'], 'b': [], 'd': []}
        >>> print p._parents
        {'a': [], 'c': [], 'b': ['a', 'c'], 'd': ['c']}
        """
        _edge = re.compile('(\w+)\s*->\s*(\w+)')
        c = 0
        a = _edge.search(s)
        sp = '->'
        if not a:
            _edge = re.compile('(\w+)\s*;')
            a = _edge.search(s)
            sp = ';'
        r = []
        while(a):
            r.append(a.groups())
            c = s.find(sp, a.start())
            a = _edge.search(s, c)
        segs = set(flatten(r))
        self._seg_list = segs
        self._parents  = dict([(seg, []) for seg in segs])
        self._children = dict([(seg, []) for seg in segs])
        for c in r:
            if len(c) > 1:
                self._parents[c[1]].append(c[0])
                self._children[c[0]].append(c[1])

    def strseg(self, seg):
        """ Convert a segment to a string.

        >>> p = Pipeline('a->b', permissive=True)
        >>> print p.strseg('a')
        a -> 
        >>> print p.strseg('b')
        a -> b -> 
        """
        p = []
        for k in self._parents[seg]:
            p.append(self.strseg(k))
        s = _str_join(p)
        s += seg + " -> "
        return s

    def connect(self, seg1, p, seg2):
        """ Connect two pipes.

        First segment becomes parent of second segment.

        seg1: string, segment name
        p: pipeline object
        seg2: string, segment name
        """
        if isinstance(p, Pipeline):
            parents = p._parents
            children = p._children
        else:
            children = dict(zip(p, [[s] for s in p[1:]] + [[]]))
            parents = dict(zip(p, [[]] + [[s] for s in p[:-1]]))
        self._parents.update(parents)
        self._children.update(children)
        self._parents[seg2].append(seg1)
        self._children[seg1].append(seg2)

    def _make_tree(self, seg_list):
        """ Convert a segment list into a segment tree.

        >>> a = Pipeline('')
        >>> a._make_tree(['first','second'])
        >>> print(a._parents)
        {'second': ['first'], 'first': []}
        >>> print(a._children)
        {'second': [], 'first': ['second']}
        """
        self._seg_list = seg_list
        self._children = dict(zip(seg_list,
                                  [[s] for s in seg_list[1:]] + [[]]))
        self._parents = dict(zip(seg_list,
                                 [[]] + [[s] for s in seg_list[:-1]]))

    def get_parents(self, seg, nophantom=False):
        """ Return a list of parent segments.

        >>> a = Pipeline('''first->second->fourth
        ...                 third->fourth''',
        ...              permissive=True)
        >>> print(a.get_parents('fourth'))
        ['second', 'third']
        """
        if not nophantom and not self._parents[seg]:
            l = [seg + "phantom"]
        else:
            l = self._parents[seg]
        return l

    def get_childrens(self, seg):
        """ Return a list of child segments.
        """
        return self._children[seg]

    def push(self, **args):
        """ Add input to orphan segment.
        """
        for k, v in args.iteritems():
            if self._parents[k]:
                raise PipeException("Segment %s not an orphan segment" % k)
            if not isinstance(v, list):
                raise TypeError("Segment input should be a list object")
            self._input[k] = v

    def flatten(self):
        """ Return a generator delivering segments in execution order.

        Perform a topological sort of the dependency graph. For this
        to succeed the graph MUST be acyclic.

        Iterative algorithm from Kahn (1962)
        http://en.wikipedia.org/wiki/Topological_sorting

        >>> T = Pipeline('''first->second->fourth;
        ...                 third->fourth;''', permissive=True)
        >>> for e in T.flatten():
        ...    print e
        third
        first
        second
        fourth
        """
        S = [s for s, p in self._parents.iteritems() if not p]
        # copy the graph
        parents = self._parents.copy()
        for k in parents: parents[k] = list(parents[k])
        children = self._children.copy()
        for k in children: children[k] = list(children[k])
        while S:
            s = S.pop(0)
            yield s
            while children[s]:
                c = children[s].pop(0)
                parents[c].remove(s)
                if not parents[c]:
                    S.append(c)
        for s, p in parents.iteritems():
            if p:
                raise PipeException("Dependency graph has "
                                    "at least one cyle including "
                                    "segment %s" % s)

    def _compute_hash(self, seg):
        """ Compute the hashkey for a given segment.

        The hash function is fed with a cleaned version of the segment
        code and args fatten by the list of subdirectories.

        """
        hash = crc32()
        currdir = self._prefix
        #self.repository.save_all_string(seg)
            h, currdir = self._compute_hash(k)
            hash.update(h)
        s = self.repository.get_all_string(seg)
        h = get_hashkey(s, hash).base32_digest()
        self._hashes[seg] = h
        currdir = path.join(currdir, '%s_%s' % (seg, h))
        self._curr_dirs[seg] = currdir

        return h, currdir

    def compute_hash(self):
        """ Compute the hashkey for all segments of the tree once and for all.
        """
            if not v:
                self._compute_hash(k)

    def get_full_seg_name(self, seg):
        """ Return segment full name (segment name + hashkey).
        """
        return '%s_%s' % (seg, self._hashes[seg])

    def _match(self, seg, re):
        """ Find the first segment which name matches regexp.
        """
        if re.match(seg):
            return seg
            s = self._match(k, re)
            if s is not None:
                return s

    def find_seg(self, seg, expr):
        """ Find segment which name matches regexp.
        """
        if expr in self._parents:
            return expr
        if expr == -1:
            return self._parents[seg][0]
        if isinstance(expr, str):
            _re = re.compile(expr)
            return self._match(seg, _re)

    # Manage directories
    def get_curr_dir(self, seg):
        """ Return the segment directory.

        """
        return self._curr_dirs[seg]

    def get_param_file(self, seg):
        """
        return path.join(self.get_curr_dir(seg), '%s.args' % seg)

    def get_tag_file(self, seg):

        """
        return path.join(self.get_curr_dir(seg), 'stdout')

    def get_data_dir(self, seg, prod=None):
        """ Return the data directory for the segment or a product
        full name.

        """
        if prod is not None:
            task_id = prod[1]
            prod    = str_file(prod[0])
            if prod:
                prod = "%d_%s" % (task_id, prod)
            return '%s/data/%s' % (self._curr_dirs[seg], prod)
        else:
            return '%s/data' % self._curr_dirs[seg]

    def get_log_dir(self):
        """ Return the pipe log directory.
        """
        return path.join(self._prefix, "log")

    def get_log_file(self, seg):
        """ Return the segment log filename.
        """
        return path.join(self.get_curr_dir(seg), '%s.log' % seg)

    def get_meta_file(self, seg, prod=-1):
        """ Return the meta data filename

        This routine is called for segment meta data storage and task
        meta data storage.

        If the first case, meta data are stored in the segment
        curr_dir.

        In the second case, meta data are stored in the task directory
        (prod may be None)
        """
        if (prod == -1):
            dirname = self.get_curr_dir(seg)
        else:
            dirname = self.get_data_dir(seg, prod)
        return path.join(dirname, '%s.meta' % seg)

    import doctest
    doctest.testmod()

## 
##
##

import os
import task
from task import TaskList
from task import Task
import Queue
import tracker
from directive import Multiplex
from pipelet.utils import close_logger, spl
import multiprocessing
import multiprocessing.managers
import re

    """
        """
        pass

logger = logging.getLogger("scheduler")
h = NullHandler()
logger.addHandler(h)

class Scheduler():
    """ Feed tasks to an execution queue. 
    
    A scheduler is made of a execution queue which is feeded with
    tasks. The scheduler coordinates the execution of the pipeline,
    building tasks for each segment and product.
    """

    def __init__(self, pipe):
        """ Initialize a scheduler. 
        
        A Tracker instance is created at this stage.  

        pipe : a Pipeline instance.
        """
        ## pipeline object
        self.pipe = pipe
        ## task queue
        self.task_queue = Queue.Queue(0)
        ## list of tasks
        self.products_list = TaskList()
        ## number of successfull tasks
        self.nb_success = 0
        ## lock on success
        self.success_lock = threading.Lock()
        ## lock on worker
        self.nb_worker_lock = threading.RLock()
        ## number of worker
        self.nb_worker = 0
        ## stoppping event 
        self.stop_event = threading.Event()
        ## stopping occurence value
        self.stopping = False
        ## aborting occurence value
        self.abort = False
        ## tracker object
        self.tracker = None

    def get_pipe(self):
        """ Return the pipe object bound to the scheduler
        """
        logger.info("a worker is trying to recover pipe")
        return self.pipe

    def check_in(self):
        """ Register a worker. 
        
        A worker call this method to check in
        """
        with self.nb_worker_lock:
            if not self.stopping:
                self.nb_worker = self.nb_worker + 1
                res = True
                logger.info('%d workers registered'%self.nb_worker)
            else:
                logger.warning('Reject logging as scheduler is stopping')
                res = False
        return res
    def check_out(self):
        """ Unregister a worker. 
        
        A worker call this method to check out
        """
            self.nb_worker = self.nb_worker - 1
            logger.info('%d workers registered'%self.nb_worker)
            if self.nb_worker == 0:    
                self.stop_event.set()
                self.abort_scheduling(None)

    def put_task(self, t):
        """ Add task to the queue.
        
        There is one task for each input which are the products of the
        parent segment.
        
        """
        t = self.tracker.add_queued(t)
        if not self.abort:
            self.task_queue.put(t)
            
    def get_task(self):
        """ Return the next task of the queue. 
        
        """
        t = self.task_queue.get()
        if t is not None:
            self.tracker.update_status(t,'running')
        return t

    def task_done(self, t):
        """ Tell the tracker when a task has been succesfully
        completed.
        
        The product list is updated with the task outputs.  If the
        completed task is the last of the segment task list, next segment
        is pushed to the queue. 
        
        """
        with self.success_lock:
            self.nb_success = self.nb_success + 1
            if self.nb_success==1:
                self.tracker.update_param(t) ## first successful task registers segment tag
                self.store_meta_seg(t)
        self.tracker.update_status(t,'done') 
        self.products_list.push(t)
        self.task_queue.task_done()

    def store_meta_seg(self, t):
       """ Store meta information for segment. 

       Store meta info at the segment level on disk (to
       rebuild db). 

       """

       ## parents
       parents = self.pipe.get_parents(t.seg, nophantom=True)
       lst_dir = []
       for e in parents:
           lst_dir.append(self.pipe.get_curr_dir(e))

       fn = self.pipe.get_meta_file(t.seg)
       try:
           with closing(file(fn, 'r')) as f:
               d = pickle.load(f)
       except:
           d = {}
       d['parents'] = lst_dir
       try:
           d['param'] = t.tag
       except:
           logger.warning("no tag saved for seg %s"%t.seg)
       with closing(file(fn, 'w')) as f:
           r = pickle.dump(d,f)

       ## other stuff
       try:
           d = t.args
           fn= self.pipe.get_param_file(t.seg)
           if not path.exists(fn):
               with closing(file(fn, 'w')) as f:
                   r = pickle.dump(d,f)
       except:
           logger.warning("no param saved for seg %s"%t.seg)
       
       if t.version:
           d = t.version
           fn= path.join(self.pipe.get_curr_dir(t.seg),path.basename(self.pipe.repository.get_code_string(t.seg)[1]))
           s = "### module version informations: \n"
           for k,v in d.items():
               s += "### '"+str(k)+"' : "+str(v)+'\n'
           with closing(file(fn, 'r+')) as f:
               c = f.read()
               c = s+c
               f.seek(0)
               f.write(c)
    
       else:
           logger.warning("no version info saved for seg %s"%t.seg)

    def push_next_seg(self, seg):
        """ Push the segment task list to the queue. 

        Segment repository is created at this stage, where a copy of
        the code, visu and argument files are saved.
        The parent segment outputs are combined to form the task list. 

        """

        ## some cleaning in the data base first
        self.tracker.clean (seg)
        
        logger.info('Pushing segment %s'%seg)
        os.umask(002) # 022 TODO this is a problem for group shared
                     # pipelines

        ######################################
        # This belongs to a dedicated function
        #######################################
        d = self.pipe.get_curr_dir(seg)
        if not path.exists(d):
            os.mkdir(d)
            r = self.pipe.repository
            f = r.get_code_string(seg)
            if f: 
                dest = os.path.join(d,os.path.basename(f[1]))
                fid = open(dest, "w")
                fid.write(f[0])
                fid.close()
            lst_hook = r.get_hook_list(seg)
            for h in lst_hook:
                f = r.get_hook_string(seg, h)
                if f[0]:
                    dest = os.path.join(d,os.path.basename(f[1]))
                    fid = open(dest, "w")
                    fid.write(f[0])
                    fid.close()
            for dk, dv in r.get_deps_string(seg).items():
                if dk:
                    dest = d+'/'+os.path.basename(dk)
                    fid.write(dv)
        d = self.pipe.get_data_dir(seg)
        ############################################

        parents = self.pipe.get_parents(seg) ## parents segments

        d       = self.tracker.get_done(seg) ## done tasks
        failed  = self.tracker.get_failed(seg) # failed tasks
        
        #reg = re.compile('/(\d*)(_[^/]*)$')
        #dprod   = [reg.sub(r'/0\2', t.str_input) for t in d] ## done products
        dprod   = dict([(spl(t.str_input), t) for t in d])
        #failed_prod = [reg.sub(r'/0\2', t.str_input) for t in failed] # failed products
        failed_prod   = dict([(spl(t.str_input), t) for t in failed])

        dstrp = dict([(str(sorted(d[i].parents)),i) for i in range(len(d))]) ## parents of done
        fstrp = dict([(str(sorted(failed[i].parents)),i) for i in range(len(failed))]) ## parents of done

        logger.debug('Found %d done tasks segment %s'%(len(d),seg))
        logger.debug('Found %d failed tasks segment %s'%(len(failed),seg))
        ## task list to queue
        l = self.products_list.multiplex(seg, parents, self.pipe.repository.get_directive(Multiplex,seg)) 

        # Convention: task with no input and not already done are push once with None as input
        if not l and not d:
            l = [Task(seg)]
            
        # Updating product list for task already done 
        for t in dprod:
            self.products_list.push(dprod[t])

        # Looking for new task to get done 
        logger.info('Found %d tasks in seg %s to get done'%(len(l),seg))
        with self.tracker.conn_lock:
            with self.tracker.conn:
                for t in l: # foreach task of the task list
                    t.str_input = self.pipe.get_data_dir(t.seg, prod=(t.task_input, 0))
                    if (t.str_input in failed_prod): # done but failed
                        continue
                    if (not (t.str_input in dprod)): #not done wrt task_input
                        self.put_task(t)
                    else: # found a task with a matching task_input. Checking whether parents also match
                        if len(t.parents)==0: # orphan task -> product gives the status
                            pass # already done nothing to do
                        else:                 # task with parents
                            try:
                                strp = str(sorted(t.parents))
                                ind  = dstrp[strp] ## find parents set in list of done parents sets
                                tjk  = d[ind]; ## check that parents set exist
                            except KeyError, ValueError: ## parents do not match parents of the done list. This is a new task.
                                self.put_task(t)

        logger.debug("nb_success starts at %d for segment %s"%(self.nb_success,seg))

    def task_failed(self, t):
        """ Tell the tracker when a task has failed.
        
        If the failed task is the last of the segment task list,
        next segment is pushed to the queue.

        """
        self.tracker.update_status(t,'failed')

    def requeue(self, t):
        """ Requeue a running task.

        """
        self.tracker.update_status(t,'queued')

    def abort_scheduling(self, t):
        """ Abort the processing.

        """
            if t is not None:
                logger.warning("Aborting scheduling of task %d..."%t.id)
            else:
                logger.warning("Aborting scheduling ...")
            self.abort = True
                logger.warning("tasks from %s has failed"%t.seg)
                self.task_queue.task_done()
                self.tracker.update_status(t,'failed')
            # empty the task queue
            try:
                logger.info ("Trying to empty task queue")
                while(self.task_queue.get_nowait()):#(block=False)):
                    self.task_queue.task_done()
            except Queue.Empty:
                logger.debug("Empty Queue")
            logger.warning("Scheduler aborted")

    #def run(self):
    #    import cProfile
    #    cProfile.runctx('self._run()', globals(), locals(),'schedprof')

    def run(self):
        """ Start the scheduler. 
        
        Start a tracker and fill the task queue with the first segment task. 
        
        """
        self.stop_event.clear()
        self.tracker = tracker.SqliteTracker(self.pipe)
        self.tracker.start()
        logger.info("Tracker launched")
        for k,v in self.pipe._input.iteritems():
            t = Task(self.pipe.get_parents(k)[0], task_output=v)
            self.products_list.push(t)
            logger.debug("Pushing phantom task %s"%str(t))
        try:
            for s in self.pipe.flatten():
                self.push_next_seg(s)
                self.task_queue.join()
                if self.abort:
                    logger.warning("Interrupting the main loop")
                    break
                if self.nb_success:
                    #self.store_meta_seg(s)
                    self.nb_success = 0
                else:
                    logger.warning("No successfull job in the previous step %s, stopping here."%s)
                    break
        except Exception, e:
            logger.critical(str(e))
            self.abort_scheduling(None)
            raise
        finally:
            # inform the workers that there are no jobs remaining
            logger.info("Asking workers to stop...")
            with self.nb_worker_lock:
                self.stopping = True
                for i in range(self.nb_worker):
                    self.task_queue.put(None)

            # wait for workers to complete
            logger.info("Waiting for workers to complete...")
            if self.nb_worker == 0:
            self.stop_event.wait()
            logger.info("No more workers. Stopping the sqltracker ...")
            self.tracker.stop()
            logger.info("Stopping the scheduler ...")            
            self.tracker.join()
            close_logger (logger)

## 
##
##

try:

class Tracker:
    """ Track the submission of objects through files in the segment directory.
    
    The numerous file access may result in a serious performance
    bottleneck. Nevertheless, it is the only work tracker that may be
    compatible with the ancestral version of pipelet.
    """
    def get_list_status(self, seg, status):
        """ Return the task list for a given segment and status. 
        
        """
        pass
    
    def get_queued(self, seg):
        """ Return the list of queued tasks.
        
        """
        return self.get_list_status(seg,'queued')

    def get_done(self, seg):
        """ Return the list of done tasks.
        
        """
        return self.get_list_status(seg,'done')
    
    def get_failed(self, seg):
        """ Return the list of failed tasks.
        
        """
        return self.get_list_status(seg,'failed')

    def get_running(self, seg):
        """ Return the list of running tasks.
        
        """
        return self.get_list_status(seg,'running')
    
    def add_queued(self, t):
        """ Add a task to the queue. 

        Add a task to the data base, and return it with its new id and status. 
        
        """
        pass
    
    def update_status(self, t, status):
        """ Update the task with new status.

        Update the task status. If not done task_output is set to None.  
        
        """
        pass

    def get_status(self, t):
        """ Return task status. 
    
        """
        pass

class SqliteTracker(Tracker,threading.Thread):
    """ Fast sqlite-based tracker.
    
    Tasks are saved in a sqlite3 data base. 
    Format is: 
    TODO
    """

        """ Initialize the Sqlite tracker. 
        
        Build the different tables of the sqlite db. 
        Register all segments. 
        Initialize the db lock. 
        
        """
        threading.Thread.__init__(self)
        
        ## dictionnary of segment ids
        self.seg_id_cache = {}
        self.seg_name_cache = {}

        ## string, sql filename
        self.sqlfile = path.join(pipe._prefix,'.sqlstatus')
        if self.pipe.sqlfile is not None:
            self.sqlfile = path.expanduser(path.expandvars(self.pipe.sqlfile))
        logger.info("tracking job in %s" % self.sqlfile)

        ## threading lock on connexion 
        self.conn_lock = threading.Lock()  # We use this to avoid database locked exceptions

        ## sql connexion object
        self.conn = sqlite3.connect(self.sqlfile, check_same_thread=False)
        self.conn.text_factory=str
        cur = self.conn.execute('select tbl_name from sqlite_master where tbl_name == "tasks"')
        if not cur.fetchone():
            with self.conn:
                self.conn.execute('create table segments (seg_id INTEGER PRIMARY KEY, seg TEXT NOT NULL, curr_dir TEXT NOT NULL UNIQUE, param TEXT, tag TEXT, comment TEXT)')
                self.conn.execute('create table tasks (task_id INTEGER PRIMARY KEY, seg_id INTEGER NOT NULL, status TEXT NOT NULL, input BLOB NOT NULL, output BLOB, str_input NOT NULL, queued_on TEXT, begun_on TEXT, ended_on TEXT)')
                self.conn.execute('create table task_relations (father_id INTEGER NOT NULL, child_id INTEGER NOT NULL)')
                self.conn.execute('create table segment_relations (father_id INTEGER NOT NULL, child_id INTEGER NOT NULL)')
                # Creating most performance critical indexes
                self.conn.execute('CREATE INDEX task_idx1 on tasks(seg_id, status)') 
                self.conn.execute('CREATE INDEX task_rel_idx1 on task_relations(child_id)')

        with self.conn_lock:
            self.conn.row_factory = sqlite3.Row
            self.segments_registration()

        ## requests queue
        #self.requests = multiprocessing.JoinableQueue(0)
        self.requests = Queue.Queue(0)
        #self.requests = multiprocessing.managers.Queue.Queue(0)

        self.tracking_event = threading.Event()
        self.tracking_delay = 1. # Tracker is awakened every second
        self.max_pending_request = 100 # Tracker is awakened if the number of request in queue exceed 100

    def segments_registration(self):
        """ Register all segments in the db. 
        
        The registration includes connectivity between segments. 
        """
        previous = None
        for s in self.pipe.flatten():
            curr_dir = self.pipe.get_curr_dir(s)
            try:
                docline = self.pipe.repository.get_docline(s)
                docline = ""
            try:
                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
                self.seg_name_cache[seg_id]  = s
                logger.debug("Segment %s instance (%s) already registered in db."%(s, curr_dir))
            except TypeError:
                logger.info("Creating segment %s instance (%s)."%(s, curr_dir))                
                c = self.conn.execute(
                    'insert into segments (seg, curr_dir, comment) values (?, ?, ?)'
                    ,(s, curr_dir, docline))
                seg_id = c.lastrowid
                logger.debug("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))

    def _asynchronous_request(self, sqlrequest, args):
        """ Add an sql request to the queue
        
        We want those requests to be done FIFO.

        """
        self.requests.put((sqlrequest,args))
        if self.requests.qsize() > self.max_pending_request:
            self.tracking_event.set()

        """ Start the tracker. 
        
        Use the shared connection to the sqlite data base, and
        regularly call flush to empty the requests queue, and commit
        the result to the db.
        """
        stopping=False
        n_tries = 0
        while not stopping:
            self.tracking_event.clear()
            self.tracking_event.wait(self.tracking_delay)
            try:
                stopping = self.flush()
            except sqlite3.OperationalError:
                n_tries = n_tries + 1
                logger.error('Catch database is locked exception, retrying: %d...' % n_tries)
                if n_tries > 10:
                    logger.error('The database seems down, abandon tracking')
                    raise
                pass
            else:
                n_tries = 0
        logger.info("Read None, tracker going down")

    def flush(self):
        """ Empty the sql requests queue in one go.
        
        As most of the asynchronous requests are updates, and the
        change are commited to the sqlfile after completion, calling
        flush too often with few requests may seriously hinder the
        performances.
        """ 
        threadconn = self.conn
        with self.conn_lock: # protect the use of the shared sql connection
            with threadconn:
                try:
                    for sqlrequest, args in iter(lambda:self.requests.get(False),(None,None)):
                        threadconn.execute(sqlrequest, args)
                        self.requests.task_done()
                    stopping = True
                except Queue.Empty:
                    stopping = False
        return stopping
        
    def stop(self):
        """ Stop the tracker.
        
        Empty the request queue by adding None. 
        
        """
        logger.info("Stopping the tracker ...")
        self.requests.put((None,None))
        self.tracking_event.set()

    def get_list_status_old(self, seg, status):

        """
        l = []
        s = self.pipe.get_curr_dir(seg)
        logger.debug("Looking for curr_dir=%s, status=%s in database"%(s, status)) 
            ts = self.conn.execute('select * from tasks as t join segments as s on t.seg_id=s.seg_id and s.curr_dir=? and t.status=? ',(s, status))
            for t in ts:
                task_output = pickle.loads(t['output'])
                task_input  = pickle.loads(t['input'])
                tid = t['task_id']
                p = self.conn.execute('select father_id from task_relations where child_id=?', ((tid,)))
                parents = [e[0] for e in p]
                l.append(task.Task(seg,status=status,id=tid, task_input=task_input, task_output=task_output,parents=parents))
        return l

        s = self.seg_id_cache[seg]
        logger.debug("Looking for seg=%d, status=%s in database"%(s, status))
            ts = self.conn.execute('select task_id, str_input, output from tasks where seg_id=? and status=? ',(s, status))
            l = [task.Task(seg, status=status, id=t[0], str_input=t[1], task_output=pickle.loads(t[2]),
                           parents=[e[0] for e in self.conn.execute('select father_id from task_relations where child_id=?', ((t[0],)))])
                 for t in ts]
        return l

    def gettest(self, seg, status):                                                                    
        pl = conn.execute('select t.task_id, t.str_input, t.output, tr.father_id from tasks as t join task_relations as tr on t.seg_id=? and t.status=? and tr.child_id = t.task_id', (s,status)).fetchall()
        index = [e[0] for e in pl]
        parent = [[] for i in range(max(index)+1)]
        for i, s in enumerate(index): parent[s].append(pl[i][3])
        l = [task.Task(seg, status=status, id=pl[i][0], str_input=pl[i][1], task_output=pickle.loads(pl[i][2]),
                       parents=parent[i]) for i in set(index)]
        return l

    def gettest2(self, seg, status):
        pl = self.conn.execute('select t.task_id, t.str_input, t.output, group_concat(tr.father_id) from tasks as t join task_relations as tr on t.seg_id=? and t.status=? and tr.child_id = t.task_id group by t.task_id', (s,status))
        l = [task.Task(seg, status=status, id=t[0], str_input=t[1], task_output=pickle.loads(t[2]),
                       parents=map(int,t[3].split(','))) for t in pl]
        return l

        """ Add a list of task to the queue. 

        """

        t.update_status ('queued')
            
        task_output = pickle.dumps(t.task_output)
        task_input  = pickle.dumps(t.task_input)
        str_input   = self.pipe.get_data_dir(t.seg, prod=(t.task_input, 0))
        for p in self.pipe.get_parents(t.seg):
            t.str_parents[p] = []            
        seg_id = self.seg_id_cache[t.seg]
        c = self.conn.execute('insert into tasks(seg_id, status, input, output, str_input, queued_on) values(?, ?, ?, ?, ?, ?)',
                              (seg_id, t.status, task_input, task_output, str_input, t.queued_on))
        t.id = c.lastrowid
                    #logger.info('for task:\n%s\nparents:\n%s\n'%(str(t),str(t.parents)))

        ### WARNING ! this has been restored due to some error in glob_parent
        ### TO DO : see if it inpacts performance
        str_input   = self.pipe.get_data_dir(t.seg, prod=(t.task_input, t.id))
        self.conn.execute('update tasks set str_input=? where task_id=?',(str_input, t.id))

        self.conn.executemany('insert into task_relations (father_id, child_id) values (?, ?)',((fid, t.id) for fid in t.parents))
        
        for fid in t.parents:
            c = self.conn.execute('select seg_id, str_input from tasks where task_id=?', ((fid,))).fetchone()
            s = self.seg_name_cache[c[0]]
            t.str_parents[s].append(c[1])
        return t
    
        """
            c = self.conn.execute('select status from tasks where task_id=?', t.id)
        return c.fetchone()['status']
    
        Update the task status in data base. If not done task_output is set to None.  
        
        """
        t.update_status (status)
        self._asynchronous_request('update tasks set status=?, str_input=?, output=?, ended_on=?, begun_on=?, queued_on=? where task_id=?',  (status, str_input, task_output, t.ended_on, t.begun_on, t.queued_on,  t.id))
        #self._asynchronous_request('update tasks set status=?, output=?, ended_on=?, begun_on=?, queued_on=? where task_id=?',  (status,  task_output, t.ended_on, t.begun_on, t.queued_on,  t.id))
        
        if status =='done' or status == 'failed':
            t.store_meta(self.pipe.get_meta_file(t.seg, prod=(t.task_input,t.id)))
        return t

    def update_param(self, t):
        """ Update the segment with new tag. 
        
        """
        self._asynchronous_request('update segments set param=? where seg_id =?',
                                   (t.tag, self.seg_id_cache[t.seg]))
        
    def clean(self, seg):
        """ Remove queued and running tasks from data base. 
        
        This is called only at the init of the scheduler.
    
        """
            seg_id = self.seg_id_cache[seg]
            l = self.conn.execute('select task_id, str_input from tasks where (status="running" or status="queued") and seg_id=?',(seg_id,)).fetchall()
            if l:
                self.conn.executemany('delete from tasks where task_id=?',((e[0],) for e in l))
                self.conn.executemany('delete from task_relations where child_id=?',((e[0],) for e in l))
            def logproblem(func, path, exc_info):
                logger.warning('An error occurred when trying to remove the unacomplished task %s'%path)
            for e in l:
                shutil.rmtree(e[1],onerror=logproblem)

    def close(self):
        """ Close database connexion.
        """
        self.conn.close()

# Copyright (C) 2008, 2009, 2010 APC LPNHE CNRS Universite Paris Diderot <lejeune@apc.univ-paris7.fr>  <betoule@apc.univ-paris7.fr>
##
##
##

""" Define various tools used by the pipe engine.
"""

import re
import base64
import zlib
import datetime
try:
import glob
from pipelet.repository import LocalRepository
import os
import sys
try:
    from pipelet.ctools import clean_string
    _sup = re.compile(r"""[\(\)\{\}\[\]"']""")
    _rep = re.compile(r"""\W+""")

    def clean_string(s):
        s = _sup.sub('', s)
        s = _rep.sub('_', s)
        return s

def flatten(seq):
    """
    this generator flattens nested containers such as

    >>> l=( ('John', 'Hunter'), (1,23), [[[[42,(5,23)]]]])
    >>> for i in flatten(l): print i,
    John Hunter 1 23 42 5 23

    By: Composite of Holger Krekel and Luther Blissett
    From: http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/121294
    and Recipe 1.12 in cookbook
    """
    for item in seq:
        if isinstance(item, str): yield item
        else:
            try:
                iter(item)
                for subitem in flatten(item):
                    yield subitem
            except:
                yield item

def spl(s):
    try:
        e = s.rindex('/') + 1
        es = s[e:]
        return s[:e] + '0' + es[es.index('_'):]
    except ValueError:
        return s

def str_file(obj):
    """ Object string representation with special caracters replaced.

    >>> str_file((1,))
    '1_'
    """
    try:
        s = clean_string(str(obj.values()))
    except:
        s = clean_string(str(obj))
    return s

def str_web(a):
    """String representation without the quotes and spaces that
    disturbs the web interface.

    >>> print(str_web(['a', ' b']))
    [a,b]
    """
    s = str(a)
    s = s.replace('"', '')
    s = s.replace("'", '')
    s = s.replace(' ', '')
    return s

def str_date():
    """ Return a string representation of the date time now %y/%m/%d %H:%M:%s

    """
    d = datetime.datetime.now()
    return (d.strftime("%Y-%m-%d %H:%M:%S"))

def is_code_file(s):
    """Check wether the file is a python code file based an the file
    extension.

    """
    return path.splitext(s)[1] in ['.py']

class crc32:
    """ Hash object using the crc32 algorithm.
    """
    def __init__(self, s=""):
        """ Initialize the hash object.

        """
        ## string, checksum given by the crc32 algorithm
        self.checksum = 0
        if s:
            self.checksum = zlib.crc32(s)

    def base32_digest(self):
        """Return a base32 digest of the checksum

        Checksum is computed using the crc32 algorithm.

        >>> print crc32('test').base32_digest()
        3B7X4DA
        """
        return base64.b32encode(
            base64.b16decode(
                "%08X" % (self.checksum & 0xffffffff)
            )).replace('=', '')

    def update(self, s):
        """Update this hash object's state with the provided string.

        """
        self.checksum = zlib.crc32(s, self.checksum)

_rm_comment = re.compile(r'^[ \t]*#.*')
_rm_endline_comments = re.compile(r'#.*')
_rm_blank_lines  = re.compile(r'^\s*$')
_rm_double_space = re.compile(r'(\S+)\s+')
_rm_endline_space = re.compile(r'\s+$')
_rm_useless_space = re.compile(r'\s*([^A-Za-z0-9 \t\n\r\f\v])\s*')

def reduced_code_formatting(s, unsafe=False):
    """ Return a code string with some part of the non syntactic
    information removed.

    if unsafe is True remove some syntactic information like spaces in string literals.

    >>> print reduced_code_formatting('''def pouet():
    ... #lets try it
    ... a,  b = foo ( bar )
    ... ''')
    def pouet():
    a,  b = foo ( bar )
    <BLANKLINE>
    """
    min_code = ""
    for l in s.splitlines():
        if _rm_comment.match(l):
            continue
        if _rm_blank_lines.match(l):
            continue
        l = _rm_endline_comments.sub('', l)
        l = _rm_endline_space.sub(r'', l)
        if unsafe:
            l = _rm_double_space.sub(r'\1 ', l)
            l = _rm_useless_space.sub(r'\1', l)
        min_code += l + '\n'
    return min_code

def get_log_file(pipe, name):
    """ Return log file name

    Current date time is prepend to the log file name.

    """
    strdate = d.strftime("%y-%m-%d_%H-%M")
    return path.join(pipe.get_log_dir(), strdate + "_" + name)

def init_logger(name, filename, level=logging.DEBUG):
    """ Initialize a logger.

    """
    logger = logging.getLogger(name)
    logger.setLevel(logging.DEBUG)
    if (len(logger.handlers) < 2):
        formatter = logging.Formatter("%(asctime)s - %(name)s - "
                                      "%(levelname)s - %(message)s")
            rfhandler = logging.handlers.RotatingFileHandler(
                filename, maxBytes=200000, backupCount=5)
            logger.addHandler(rfhandler)
            rfhandler.setFormatter(formatter)
        if level:
            shandler = logging.StreamHandler(sys.stdout)
            shandler.setLevel(level)
            shandler.setFormatter(formatter)
            logger.addHandler(shandler)
    return logger

def close_logger(logger):
    """ Close a logger

    """
    for h in logger.handlers:
        logger.removeHandler(h)
        h.flush()
        h.close()

def create_pipe(pipename, prefix=[]):
    """ Create an empty pipe environment including main, defaulf
    segment, prefix and log directories.

    This function is called from the script pipeutils.

    """
    import os

    ## make code dir
    code_dir = os.path.join("./" + pipename)
    if not os.path.exists(code_dir):
        os.mkdir(code_dir)

    ## set prefix dir
    if not prefix:
        prefix = code_dir

    if not os.path.exists(os.path.join(code_dir, "main.py")):
        fn = current_dir + "/static/main.py"
        with closing(file(fn, 'r')) as f:
            str = f.read()
        str = str.replace("prefix   = './'", "prefix   = '%s'" % prefix)
        fn = os.path.join(code_dir, "main.py")
            f.write(str)
    if not os.path.exists(os.path.join(code_dir, "default.py")):
        # make a segment file with some doc
        shutil.copy(current_dir + "/static/default.py", code_dir)

    print "\n\n-----------------------------------------------------------"
    print "  Pipeline '%s' has been successfully created  " % pipename
    print "-----------------------------------------------------------\n\n"
    print "1- Change directory to '%s' to set your pipeline scheme\n in the"
    " file named 'main.py'\n" % code_dir
    print "2- Rename and edit the 'default.py' to set the segments content.\n"
    print "3- Run 'python main.py -d' to enter the debugging execution mode."
    print "Type 'python main.py --help' to get the full list of options.\n"
    print "4- Run 'pipeweb track %s %s/.sqlstatus' \n to add the pipe to"
    " the web interface.\n" % (pipename, prefix)
    print "5- Set acl with 'pipeutils -a username -l 2 %s/.sqlstatus'\n"\
        % prefix
    print "6- Launch the web server with 'pipeweb start'"
    print "You should be able to browse the result on the web page"
    " http://localhost:8080\n"
    print "Retype the current command line to get those instructions back."

def get_hashkey(s, starthash):
    """Return a unique key which identify code string
    using cheksum algorithm.

    Comment lines and blanks are not taken into account in code string.

    >>> print(get_hashkey("a = 5", 0).base32_digest())
    ZNBR63A
    """
    if not starthash:
        starthash = crc32()

    starthash.update(reduced_code_formatting(s)) 
    return starthash

def parse_disk(pipedir):
    """ Return list of pipe instance starting from pipedir.
    """
    lstpipe = []
    dircontent = glob.glob(path.join(pipedir, "*_*"))

    ## visit each segment's repository
    for c in dircontent:

        ## this is a node
        if path.isdir(c):
            lstpipe.append(c)
            lstpipe += parse_disk(c)

    return lstpipe

def rebuild_db_from_disk(pipedir, sqlfile=None):
    """ Rebuild db from disk.

    """
    ## set sqlfile
    if sqlfile is None:
        sqlfile = os.path.join(pipedir, ".sqlstatus")

    ## 1---: create all tables
    ## taken from sqlite tracker constructor
    conn = sqlite3.connect(sqlfile)
    cur = conn.execute('select tbl_name from sqlite_master where tbl_name == "tasks"')
    if not cur.fetchone():
        with conn:
            conn.execute('create table segments (seg_id INTEGER PRIMARY KEY, seg TEXT NOT NULL, curr_dir TEXT NOT NULL UNIQUE, param TEXT, tag TEXT, comment TEXT)')
            conn.execute('create table tasks (task_id INTEGER PRIMARY KEY, seg_id INTEGER NOT NULL, status TEXT NOT NULL, input BLOB NOT NULL, output BLOB, str_input NOT NULL, queued_on TEXT, begun_on TEXT, ended_on TEXT)')
            conn.execute('create table task_relations (father_id INTEGER NOT NULL, child_id INTEGER NOT NULL)')
            conn.execute('create table segment_relations (father_id INTEGER NOT NULL, child_id INTEGER NOT NULL)')
        conn.row_factory = sqlite3.Row
        
    ## 2---: segments table from curr_dir 
    lst_dir = parse_disk (pipedir) ## list curr_dir
    seg_id_cache = {}              ## store new seg_id for each curr_dir
    seg_depend_cache = {}          ## store parents for each curr_dir
    for curr_dir in lst_dir:
        curr_dir = path.abspath(curr_dir) 
        #s        = curr_dir.split("_")[-2].split("/")[-1] ## TODO replace with regexp -> done
        #rec      = re.search (r"/([a-zA-Z0-9]+)_([a-zA-Z0-9]_*)([a-zA-Z0-9]+)$", curr_dir)
        gr = re.search (r"/([a-zA-Z0-9]+)(_[a-zA-Z0-9])*_([a-zA-Z0-9]+)$", curr_dir).groups()[:-1] # remove segkey
        s  = gr[0]
        for g in gr[1:]:
            if g:
                s= s+g
        print "Creating segment %s instance (%s)."%(s, curr_dir)
        R        = LocalRepository([s], curr_dir) 
        ## read  curr_dir/seg_s_code.py to get docstring
        try:
            docline = R.get_docline(s)
        except Exception:
            docline = ""
        ## read curr_dir/seg_s.meta to get parents, param, and tag
        fn = path.join(curr_dir, s+".meta") ## 
            meta = pickle.load(f)
            seg_depend_cache[curr_dir] = meta['parents']
            param = meta['param']
            if meta.has_key('tag'):
                tag = meta['tag']
            else:
                tag = ""

        ## insert values in db
        c = conn.execute(
            'insert into segments (seg, curr_dir, comment, param, tag) values (?, ?, ?, ?, ?)'
            ,(s, curr_dir, docline, param, tag))
        ## retrieve the seg_id
        seg_id_cache[curr_dir]  = c.lastrowid
            
    ## 3---: segment_relations table from meta
    ## need to restart the loop in order to use the right seg_id
    for k,v in seg_id_cache.items():
        print "Storing connectivity for segment %s."%k
        for p in seg_depend_cache[k]:
            id = conn.execute('select seg_id from segments where curr_dir = ?',(p,)).fetchone()
            conn.execute('insert into segment_relations (father_id, child_id) values (?, ?)',(id[0], v))
                    
    ## 4---: tasks table from str_input
    task_id_cache = {}     ## store new task_id for each str_input
    task_depend_cache = {} ## store parents for each str_input
        print "Creating tasks of segment %s."%k

        ## Get the list of sub directory (ie str_input)
        lst_task = []
        for root, dir, fn in os.walk(path.join(k, "data")):
            for d in dir:
                lst_task.append(path.abspath(path.join(root, d)))
            break
        if not lst_task:
            lst_task.append(path.abspath(path.join(k, "data"))+"/")

        for t in lst_task:

            print "Creating task %s"%t 

            ## read task propertie from meta file
            fn = glob.glob(path.join(t, "*.meta"))
            if fn:
                with closing (file(fn[0],'r')) as f:
                    meta = pickle.load(f)
            else:
                print "error no meta file found"
            task_depend_cache[t] = meta['parents']

            ## insert values in db
            task_id_cache[t] = meta['id']#t.split("_")[-2].split("/")[-1]

            c = conn.execute('insert into tasks (task_id, seg_id, output, input, begun_on, ended_on, queued_on,status,str_input) values (?, ?, ?, ?, ?, ?, ?, ?, ?)',(task_id_cache[t], v, pickle.dumps(meta["output"]), pickle.dumps(meta["input"]), meta["begun_on"], meta["ended_on"], meta["queued_on"], meta["status"],t))
            
    ## 5---: task_relations from meta + task input 
    ## need to restart the loop in order to use the right task_id
    for k,v in task_id_cache.items():
        print "Storing connectivity for task %s."%k
        for sp in task_depend_cache[k].values():
            for p in sp: 
                id = conn.execute('select task_id,output from tasks where str_input = ?',(p,)).fetchone()
                conn.execute('insert into task_relations (father_id, child_id) values (?, ?)',(id[0], v)) 

    conn.commit()

def make_dict(l):
    """ Convert a list into a dict keeping duplicated entries.
    
    >>> d = make_dict(zip(['a']*10+['b'], range(10)+[1]))
    >>> print d
    {'a': [0, 1, 2, 3, 4, 5, 6, 7, 8, 9], 'b': 1}
    """
    d = {}
    for k, v in l:
        try:
            d[k].append(v)
            d[k] = v
        except AttributeError:
            d[k] = [d[k],v]
    return d

def upgrade_webconfig():
    """ Upgrade pipeweb config with new convention. 
    
    Use .pipelet directory for all config/log files
    """
    try:
        os.system("mv ~/.pipelet ~/pipelet.conf")
        os.system("mkdir ~/.pipelet")
        os.system("mv ~/pipelet.conf ~/.pipelet/")
        os.system("mv ~/pipelet.access ~/.pipelet/")
        os.system("mv ~/pipelet.errors ~/.pipelet/")
        os.system("mv ~/pipelet.pid ~/.pipelet/")
        print "Successfull upgrade !"
    except:
        pass

## 
##
##

try:
import os
from cherrypy.lib.static import serve_file, serve_download
import re
from db_utils import get_lst_tag, get_lst_date, add_tag,del_tag, _delseg, _deltask, _get_children, _get_fathers, _get_children_task, get_lst_seg

class TreeView:
    """ A minimalist pipeline Web interface.
    
    A pipeline Web interface allows to browse all instances of a given pipeline. 
    The informations about the pipeline are retrieve from its data base file. 
    Pipelines are printed through a tree view with links to all segments and products. 
    """
    #p = profiler.Profiler('/home/betoule/pipewebprof/')
    ## boolean, Web object is exposed. 
    exposed = True
    def __init__(self, db_file, name):
        """ Initialize a web instance. 
        
        """
        ## string, pipeline short name
        self.name = name
        ## string, pipeline data base file.
        self.db_file = db_file
        self.js_file = ['../static/mktree.js', '../static/cook.js', '../static/tag.js']

    def get_html_refresh (self, refresh=True):
        """
        """
        html="""
<html>
<head>
"""
        for l in self.js_file:
            html += """
<SCRIPT SRC="%s" LANGUAGE="JavaScript"></SCRIPT>
"""%l
        html += """
<LINK REL="stylesheet" HREF="../static/mktree.css">  
"""
        if refresh:
            html += """
<meta http-equiv="refresh" content="35" > 
"""
        html+= """
</head><body>
"""

    @auth.pipeauth.write_access()
    def addtag (self, segid, tag):
        """ Add new tag to the database
        """
        add_tag(self.db_file, segid, tag)
        raise cherrypy.HTTPRedirect('/'+self.name+'/',303)

    def deltag (self, tag):
        """
        del_tag(self.db_file, tag)
        
        with conn:
                l = conn.execute('select seg_id from segments where seg=?',(seg,)).fetchall()
                for s in l: 
                    lstseg+=s

        html = self.index(highlight=lstseg)
    
    @auth.pipeauth.read_access()
    def filter (self, tag=None, date=None, seg=None,recursive="True"):
        """ Print the pipeline instances matching arguments.
        
        If many parameters are set, the selection
        corresponds to the union.
        Tag are set for leafs only. Upstream parents are 
        added to the list if recursive is set to True. 
        """
        lstseg = []
        recursive = eval(recursive)
        conn = sqlite3.connect(self.db_file,check_same_thread=True)
        with conn:
            if tag is not None:
                l = conn.execute('select seg_id, tag from segments where tag like ?',("%"+tag+"%",)).fetchall()
                for s in l: 
                    lst_tag = s[1].split(";")
                    if tag in lst_tag:
                        if recursive:
                            lstseg+=_get_fathers(self.db_file,s[0])
                        else:
                            lstseg+=s[0]

            if date is not None:
                l = conn.execute('select seg_id from tasks where queued_on = ?',(date,)).fetchall()
                for s in l: 
                    if recursive:
                        lstseg+=_get_fathers(self.db_file,s[0])
                    else:
                        lstseg+=s[0]
                for s in l: 
                    else:
                        lstseg+=s

    def filter_panel (self, l):
        """ Return an html string containing the filter panel
        """
        html = ""
        html += '<div id="left"><fieldset id="filters"><legend><span class="text">Filters</span></legend>'
        html += '<table width="100%"><tr><td><table>'
        
        ## Tag checkbox
        html += '<tr  valign="top" id="tr_tag_id" class="filter"> <td style="width:100px;"><input id="cb_tag" type="checkbox"/><label>Tag</label></td>'

        ## Tag select
        lst_tag = get_lst_tag(self.db_file)
        html += '<td style="width:50px;"><select id="se_tag_id">'
        for t in lst_tag:
            html +='<option value="%s">%s</option>'%(t, t)
        html += '</select></td>'

        ## Tag button
        html += '<td style="width:150px;"><a class="icon delete" href="javascript:del_tag();"><small>Delete</small></a></td>'
        
        ## Date checkbox
        html += '<tr  valign="top" id="tr_date_id" class="filter"> <td style="width:100px;"><input id="cb_date" type="checkbox"/><label>Date</label></td>'

        ## Date select
        lst_date = get_lst_date(self.db_file)
        html += '<td style="width:150px;"><select id="se_date_id">'
        for t in lst_date:
            html +='<option value="%s">%s</option>'%(t[0], t)

       ## rec button
        #html += '<td style="width:100px;"><input id="cb_rec" type="checkbox" checked="checked"/><label>Recursive</label></td>'
        
        str_lst_tag = "\\n"
            str_lst_tag += "- "+t+"\\n"

        ## find log dir
        last_dir = l[0][1]
        log_dir  = False
        while not log_dir and last_dir != '/':
            last_dir = os.path.abspath(os.path.join(last_dir, os.path.pardir))
            log_dir  = os.path.isdir(os.path.join(last_dir, "log"))
        log_dir = os.path.join(last_dir, "log")
        
        ## Buttons
        html += '<tr></tr>'
        html += '</table>'
        html += '</td><td></td></tr></table>'
        html += '<p class="buttons">'
        html += '<a class="icon apply"  href="javascript:filter();"><small>Apply</small></a>'
        html += '<a class="icon clear"  href="javascript:uncheck(\'checkbox\');"><small>Clear</small></a>'
        html += '<a class="icon tag"    href="javascript:edit_tag(\'%s\');"><small>Tag</small></a>'%str_lst_tag
        html += '<a class="icon delete" href="javascript:del_seg();"><small>Delete</small></a>'
        html += '<a class="icon log" href="log?logdir=%s"><small>Browse log</small></a>'%log_dir
        html +='</p></fieldset></div>'

    def thumbnail_panel (self, segid):
        """ Return an html string containing the thumbnail panel
        """
        html = ""
        html += '<fieldset id="thumb"><legend><span class="text">Thumbnails</span></legend>'
        if segid is not None:
            conn = sqlite3.connect(self.db_file,check_same_thread=True)
            with conn:
                seg, currdir = conn.execute(
                    'select seg, curr_dir from segments where seg_id = ?'
                    ,(segid,)).fetchone()
                l = conn.execute('select str_input, task_id from tasks where seg_id=?',(segid,)).fetchall()
            conn.close()
            lstimage = []
            for e in l:
                directory = os.path.relpath(e[0],start=currdir)
                directory = self.check_path (segid, directory)
                lstimage = lstimage+(glob(os.path.join(directory,'*.png')))
            for img in lstimage:
                html += '<a href="download?segid=%s&filepath=%s"><img src="download?segid=%s&filepath=%s" width=100 border=no></a>'%(segid, os.path.abspath(img),segid, os.path.abspath(img))
        html +='</fieldset>'

    def treeview_panel (self, l, tcounts, index, sindex, highlight):
        """ Return an html string containing the tree view
        """
        ## Tree view
        html  = ""
        html += '<br>'
        html +='<div class="list"><ul class="mktree" id="segtree">'
        
        indent = -1
        # select a subset
        if highlight is not None:
            newl = []
            for s in l:
                if s[2] in highlight:
                    newl.append(s)
                else:
                    cherrypy.log.error_log.warning(str(s[2])+str(highlight))
            l = newl
        for s in l:
            #with conn:
            #    e = conn.execute('select status, count(status) from tasks where seg_id=? group by status',(s[2],)).fetchall()
            try:
                e = [tcounts[i] for i in index[s[2]]]
            except IndexError:
                e = []

            ss = s[3]
            ##ss = '%s'%("" if str(s[3])=='None' else s[3]) + '<span class="pipetag">%s</span>'%(" " if str(s[4])=='None' else s[4])
            if ss is None:
                ss = ""
            ss += '<span class="pipetag">%s</span>'%("" if str(s[4])=='None' else s[4])
            ss = '<a class="icon image" href="index?thumbnail=%d"></a>'%(s[2])+ss
            for stat in e:
                ss = '<a href="product?segid=%s&status=%s" class=%s>%d</a>, '%(s[2],stat[1], stat[1], stat[2]) +  ss

            ss += '<INPUT type="checkbox" name="checkbox" id="%d"/>'%(s[2])
            diff = s[1].count('/') - indent
            if diff == 1:
                html += '<ul> <li id=%d ><a href="code?segid=%d">%s</a> : %s\n'%(s[2],s[2],s[0],ss)
            elif diff == 0:
                html += '</li><li id=%d ><a href="code?segid=%d">%s</a> : %s\n'%(s[2],s[2],s[0],ss)
            elif diff > 1:
                html += '<li id=%d ><a href="code?segid=%d">%s</a> : %s\n'%(s[2],s[2],s[0],ss)
            else:
                html += '</li> </ul>\n'*abs(diff) + '<li id=%d> <a href="code?segid=%d">%s</a> : %s'%(s[2],s[2],s[0],ss)
            indent += diff 
        
        html += '</li></ul>'*(indent-l[0][1].count('/')+1)
        if not highlight:
            html += '</div></body></html>'

    def index(self, highlight=None, thumbnail=None):
        """ Pipeline instances tree view

        Print the pipeline instances trough a tree view.
        
        """

        html = self.get_html_refresh ()
        html += '<a class="icon home" href="../%s/"><small>Home</small></a>'%(self.name)
        html += '<h1>Pipelines in %s </h1>'%self.name

        # get all instances
        with conn:
            l = conn.execute('select seg, curr_dir, seg_id, param, tag from segments order by curr_dir').fetchall()
            tcounts = conn.execute('select seg_id, status, count() from tasks group by seg_id, status order by seg_id').fetchall()
        try:
            sindex = [e[0] for e in tcounts]
            index  = [[]   for i in range(max(sindex)+1)]
            for i, s in enumerate(sindex): index[s].append(i)
        except ValueError:
            html += 'No computed pipeline !</div></body></html>'
            return html
        
        ## Filter fieldset
        html += self.filter_panel(l)
        
        ## Thumbnails
        #if thumbnail is not None:
        html += self.thumbnail_panel (thumbnail)

        ## Treeview
        html += self.treeview_panel (l, tcounts, index, sindex, highlight)
        
    def product(self, segid=None, status=None):
        """ Products index.
        
        Print the content of the product directory. status in ('done', 'queued', 'running', 'failed')
        """
        with conn:
            seg, currdir = conn.execute(
                'select seg, curr_dir from segments where seg_id = ?'
                ,(segid,)).fetchone()
            l = conn.execute('select str_input, task_id from tasks where seg_id=? and status=?',(segid, status)).fetchall()
        html = self.get_html_refresh() + '<a class="icon home" href="../%s/"><small>Home</small></a>'%(self.name)
        html += '<h1> Data products for %s tasks in segment %s </h1>' % (status, seg)
        html += '<div class="list"><p>Directory : %s</p>  %d <span class="%s">%s</span> tasks<br><br>  '%( currdir,len(l), status, status)
        html += '<a class="icon delete" href="javascript:del_prod(%d);"><small>Delete</small></a>'%(int(segid))
        html += '<a class="icon apply"  href="javascript:check_all();"><small>Check all</small></a>'
        html += '<a class="icon clear"  href="javascript:uncheck(\'checkbox_p\');"><small>Clear</small></a>'
        html += '<ul>'
        # os.path.relpath is somewhat slow. This is a quick workaround, I hope it's correct.
        #s = ''.join(['<li><a href="pipedir?segid=%s&directory=%s"> %s </a><INPUT type="checkbox" name="checkbox_p" id="%d"></li>'%(segid,os.path.relpath(e[0],start=currdir),e[0],e[1]) for e in l])
        r = os.path.relpath(l[0][0], currdir)
        currdir=l[0][0].replace(r,'')
        s = ''.join(['<li><a href="pipedir?segid=%s&directory=%s"> %s </a><INPUT type="checkbox" name="checkbox_p" id="%d"></li>'%(segid,e[0].replace(currdir,''),e[0],e[1]) for e in l])
        html += s + '</ul></div></body></html>'

    def code(self, segid=None):
        """ Segment's files index. 
        
        Print the content of a segment directory (code files, parameters file, ...)
        
        """
        highlight = _get_fathers(self.db_file,segid)
        with conn:
        
        html = self.index(highlight=highlight) + '<p>%s</p><ul>' % (currdir)
        l = glob(os.path.join(currdir,"*.*"))
        for e in l:
            html += '<li><a class="icon download" href="download?segid=%d&filepath=%s&download=0"></a><a href="download?segid=%d&filepath=%s"> %s </a></li>'%((int(segid),os.path.basename(e),int(segid),)+(os.path.basename(e),)*2)

    def del_lst_seg(self, seglist):

        Delete all segments and products directories of a given pipeline instance.
        The list of pipe id is '-' separated
        """
        cherrypy.log.error_log.warning('called ONCE')
        seglist = seglist.split("-")
        
        ## build list including dependencies
        final_list = []
        for segid in seglist:
            if segid and segid is not None:
                final_list= final_list+_get_children(self.db_file,int(segid))
        final_list = list(set(final_list))
        final_list.sort()

        ## empty the list 
        _delseg(self.db_file, final_list)

    def del_lst_prod(self, segid, taskid):
        """ Delete recursively a list of products. 
Delete products directories of a given task id. The list of task id is '-' separated
        """
        tasklist = taskid.split("-")
        
        for taskid in tasklist:
            if taskid:
               final_list= final_list+_get_children_task(self.db_file,taskid)
                
        _deltask(self.db_file, final_list)

    def pipedir(self, segid=None, directory=None):
        """ Print the content of a directory. 

        directory: string pipeline directory path.
        """
        directory = self.check_path (segid, directory)
        html +=  '<h1> Content of %s </h1> <div class="list"><ul>'%directory
        lstfile = (glob(os.path.join(directory,'*')))
        imglist = []
        datlist = []
        loglist = []
        for filename in lstfile:
            try:
                if (filename.split(".")[1] in ["png", "jpg", "eps", "pdf"]):
                    imglist.append(filename)
                elif (filename.split(".")[1] in ["meta", "log", "err"]):
                    loglist.append(filename)
                else:
                    datlist.append(filename)
                datlist.append(filename)
        kk = directory.split("data")[-1]
        codedir = directory.replace ("data"+kk, "")
        codelist=glob(codedir+"*.*")
        biglist = [sorted(loglist), sorted(datlist), sorted(imglist), sorted(codelist)]
        name = ['LOGS', 'DATA', 'FIGURES', 'CODE']
        for L in biglist:
            if L:
                html += '<span class="text"><small>%s</small></span>'%name[biglist.index(L)]
            for filename in L:
                absPath = os.path.abspath(filename)
                if os.path.islink(absPath):
                    html += '<li><a href="pipedir?segid=%d&directory='%int(segid) + absPath + '">' + os.path.basename(filename)+"("+os.path.realpath(filename)+")" + "</a></li>"
                elif os.path.isdir(absPath):
                    html += '<li><a href="pipedir?segid=%d&directory='%int(segid) + absPath + '">' + os.path.basename(filename) + "</a></li>"
                else:
                    html += '<li><a class="icon download" href="download?segid=%d&filepath=%s&download=0"></a><a href="download?segid=%d&filepath='%(int(segid),  absPath, int(segid)) + absPath + '">' + os.path.basename(filename) + "</a> </li>"
                
        html += """</ul></div></body></html>"""

    ## regular expression which forbids path
    forbidden_path = re.compile('.*\.\..*')

    def log(self, logdir):
        """ Print the content of the log directory. 
        """
        directory = logdir
        html = self.get_html_refresh(refresh=False) + '<a class="icon home" href="../%s/"><small>Home</small></a>'%(self.name)
        html += '<h1> Content of %s </h1> <div class="list"><a class="icon delete" href="delete_log?logdir=%s"><small>Delete logs</small></a><ul>'%(directory,logdir)
        for filename in sorted(glob(os.path.join(directory,'*')), reverse=True):
            absPath = os.path.abspath(filename)
            html += '<li><a href="serve_log?filename='+absPath+ '">' + os.path.basename(filename) + "</a> </li>"

    def serve_log(self, filename):
        """ Print the content of the log file.
        """
        return serve_file(filename, content_type='text/plain', disposition="inline")

    def delete_log(self, logdir):
        """ Delete the content of the log directory.
        """
        for filename in glob(os.path.join(logdir,'*')):
            os.remove(absPath)
        raise cherrypy.HTTPRedirect("log?logdir=%s"%logdir,303)

    def check_path(self, segid, path):
        """Chroot the path to the segid currdir.
        """
        with conn:
        filepath = os.path.realpath(os.path.join(currdir, path))
        if self.forbidden_path.match(os.path.relpath(filepath, start=currdir)) is not None:
            raise cherrypy.HTTPError(403)
        else:
            return filepath
        
    def download(self, segid=None, filepath=None, download=None):
        """ Download a file.
        
        Text files and pickle are printed directly.
        
        """
        filepath = self.check_path(segid, filepath)
        if download is not None:
            return serve_download(filepath)
        else:
            if os.path.splitext(filepath)[1] in ['.log','.txt','.list','.py']:
                return serve_file(filepath, content_type='text/plain', disposition="inline")
            elif os.path.splitext(filepath)[1] in ['.pkl', '.meta', '.args', '.version']:
                return self.serve_pickle(filepath)
            else:
                return serve_file(filepath, disposition="inline")

    @auth.pipeauth.read_access() 
    def serve_pickle (self, filename):
        """ Expose pickle file content
        
        """
        import cPickle as pickle
        f = file(filename)
        d = pickle.load(f)
        if isinstance(d, dict):
            s = ""
            for k,v in d.items():
                s += "'"+str(k)+"' : "+str(v)+'</br><br>'
            s = '<span class="code";><small><br>'+s+'</br></small></span>'
        else:
            s = '<span class="code";><small><br>'+str(d)+'</br></small></span>'
        f.close()
        return s     

""" Simple authentication for pipeweb.
"""

try:

def check_access(auth_type, access_level):
    """ Check whether user has sufficient access rights.

    Depending on the configuration passed through auth_type:
    - grant access without verification
    - check the access level of an otherwise authenticated user
    - perform Basic http authentication and check access rights
    - perform Digest http authentication and check access rights

    auth_type is  either 'None', 'ACL', 'Basic', 'Digest'
    """
    realm = "pipeweb"
    if auth_type == 'None':
        #No autorization check required
        return True

    if not 'authorization' in cherrypy.request.headers:
        return False
    
    ah = cherrypy.lib.httpauth.parseAuthorization(cherrypy.request.headers['authorization'])
    if ah is None:
        raise cherrypy.HTTPError(400, 'Bad Request')

    dic = get_credentials(access_level)
    if len(dic)==0:
    
    if auth_type == 'ACL':
        #Check that otherwise authenticated user has the required
        #access level
        if ah['username'] in dic.keys():
            return True
        else:
            return False    
    password = dic.get(ah["username"], None)
    encrypt = lambda x:x
    if cherrypy.lib.httpauth.checkResponse(ah, password, method=cherrypy.request.method,
                              encrypt=encrypt, realm=realm):
        cherrypy.request.login = ah["username"]
    
    cherrypy.request.login = False

    return False

def get_credentials(access_level=1):
    """ Get users/passwd from pipe data base.
access_level is 1 for read-only or 2 for read-write access. return dictionnary of user and passwd
    """
    db_file = cherrypy.request.app.root.which_sqlfile(cherrypy.request.path_info)
    try:
    except:
        import pdb
        pdb.set_trace()
        cherrypy.log.error_log.warning('unable to open database file %s'%dbfile)
        dic = {}
    else:
        with conn:
            try:
                l = conn.execute('select user, passwd from users where access_level >= ?',(access_level,)).fetchall()
                dic = dict(l)
            except:
                dic = dict([])
    return dic

def ask_auth(auth_type):
    if auth_type == "Basic":
        cherrypy.response.headers['www-authenticate'] = cherrypy.lib.httpauth.basicAuth(realm)
    elif auth_type == "Digest":
        cherrypy.response.headers['www-authenticate'] = cherrypy.lib.httpauth.digestAuth(realm)

def read_access(auth_type="Digest"):
    """ Require read only access on a web page
    """
    if check_access (auth_type, 1):
        return
    ask_auth(auth_type)
    raise cherrypy.HTTPError(401, "You are not authorized to access that resource") 

def write_access(auth_type="Digest"):
    """ Require write accessto a web page
    """
    if check_access (auth_type, 2):
        return

# Making this module a custom cherrypy toolbox
pipeauth = cherrypy._cptools.Toolbox('pipeauth')
pipeauth.read_access = cherrypy.Tool('on_start_resource', read_access)
pipeauth.write_access = cherrypy.Tool('on_start_resource', write_access)

def acl_setup(db_file):
    """ Set user/passwd table in pipe database.
    
    """
    with conn:
        if not conn.execute('select tbl_name from sqlite_master where tbl_name == "users"').fetchone():
            conn.execute('create table users (id INTEGER PRIMARY KEY, user TEXT, passwd TEXT, access_level INTEGER)')

def add_user(db_file, user, passwd, access_level=1):
    """ Add user entry to data base.
    
    """
    acl_setup(db_file)
    with conn:
        conn.execute('insert into users(user, passwd, access_level) values(?,?,?)',(user, passwd, access_level))
    
def del_user(db_file, user):
    """ Delete user entry from data base.
    
    """
    with conn:
        conn.execute('delete from users where user = ?' , (user,))

## 
##
##

import shlex

class Directive:
    """ Parse the segment code to get directives (multiplex, depend,
    ...)
    """
    def parse(self, code):
        """ Parse the code to initialiaze directive names and values.
        
        Directive names and value are class attributs. 

        """
        try:
            l = shlex.split(code)
            raise StopIteration()
        it = iter(l)
        if it.next() != self._name:
        for e in it:
            if self.__dict__.has_key(e):
                try:
                    setattr(self, e, it.next())
                except StopIteration:
                    raise ValueError("The string is not a valid directive\n"+self._usage)
            else:
                try:
                    getattr(self,self._args).append(e)
                except AttributeError:
                    setattr(self, self._args, e)
                    
class Multiplex(Directive):
    """ Multiplex directives are used to control the pipeline
    parallelisation scheme.
    """
    _name  = "#multiplex"
    _usage = "#multiplex method where 'condition' group_by 'value'"
    _args  = "method"

        """ Init the multiplex directive. 
        """

        ## string, multiplex method
        self.method   = "cross_prod"

        ## string, mutliplex condition
        self.where    = ""

        ## string, gathering condition
        self.group_by = ""

class Depend(Directive):
    """ Depend directive are used to include external code to the
    hashkey computation.
    """
    _name  = "#depend"
    _usage = "#depend file1 file2 ..."
    _args  = "deps"
        """ Init the depend directive
        """
        ## list of external code files
        self.deps=[]

## 
##
##

import os
from task import *
from environment import *
import gc
from pipelet.utils import close_logger

    """
        """
        pass

class AbortError(Exception):
    """ Extension of the Exception class. 
    """
    def __init__(self, signal=0):
        """ Initialize an error. 

        """
        ## integer, signal value. 
        self.signal = signal

class Worker(object):
    """ Empty tasks of an execution queue.

    A worker is linked to a scheduler. Its job is to empty the
    scheduler task queue.

    This is the base class for all workers type.
    """
    def __init__(self, scheduler, logger=None):
        """ Initialize a worker. 
        
        """
        ## scheduler object
        self.scheduler = scheduler

        ## pipe object related to the scheduler
        self.pipe = scheduler.get_pipe()
        
        ## task object, current task
        self.task = None

        ## string, current task directory
        self.work_dir = None
        
        ## logger
        self.logger = logger
        if logger is None:
            self.logger = logging.getLogger("worker")
        h = NullHandler()
        self.logger.addHandler(h)

    def matplotlib_hook(self):
        """  Turn the matplotlib backend to Agg.
        """
        if self.pipe.matplotlib:
            self.logger.info('Turning matplotlib backend to Agg')
            import matplotlib
            matplotlib.use('Agg')

    #    cProfile.runctx('self._run()', globals(), locals(),'workprof')

        """ Start the worker. 
        
        The worker executes the tasks given by its scheduler until the
        queue is empty. The task status are updated after completion
        or failure.
        """
        self.logger.info( 'checking in ...')
        if not self.scheduler.check_in():
            self.logger.error( "check_in failed : stopping")
            return
        self.logger.info( 'checked in.')

        self.matplotlib_hook()

        n = 0        
        try:
            for task in iter(self.scheduler.get_task, None):
                self.work_dir = self.pipe.get_curr_dir(task.seg)
                self.logger.info('Processing one task of segment %s'%task.seg)
                task = self.execute_task(task)
                if task.status == "done":
                    self.logger.info('task done')
                    self.scheduler.task_done(task)
                else:
                    self.logger.info('task failed')
                    self.scheduler.task_failed(task)
                n = n+1
            self.task = None
            self.logger.info("%d jobs completed" % n)
        except AbortError, e:
            self.logger.error( "Abort after catching signal %d" % e.signal)
            self.scheduler.requeue(task)
        finally:
            self.terminate()

    def execute_task(self, task):
        """ Execute a task. 
        
        Make task directory and execute segment code string with the
        environment feeded with the task input. Task output is read
        from the environment after execution and updated to the task
        object.

        """
        seg  = task.seg
        prod = task.task_input
        self.make_dir(task)
        glo, env = self.prepare_env(task)
        code = self.pipe.repository.get_code_string(seg)
        self.exec_code(code, glo, env)
        self.task.task_output = env._close(glo)
        return self.task

    def exec_code(self, code, glo, env):
        """ Execute code string with namespace. 

        Task status is updated after execution. 
        Exceptions are read from the environment log file. 
        
        """
        try:
            exec (compile(code[0], code[1], "exec"), glo)
        # We do not want to prevent Abortion
            raise e
            etype, value, tb = traceback.sys.exc_info()
            f = file(env._get_log_file(),"a")
            traceback.print_exception(etype, value, tb, file=f)
            f.close()
            # TODO dump the glo for latter debugging
            self.task.status = "failed"
        else:
            self.task.status = "done"

    def terminate(self):
        """ Disconnect from scheduler. """
        
        self.scheduler.check_out()

    def make_dir(self, task):
        """ Create a directory for a given task. 
        
        """

        # make directories if task dir (otherwise scheduler did it)
        dseg = self.pipe.get_data_dir(seg)
        d    = self.pipe.get_data_dir(seg, prod=(prod, task.id))
        if dseg != d:
            os.umask(002)
            try :
                os.mkdir(d)
            except OSError, e:
                if e.errno == 17: # file exist
                    self.logger.info("A new task is submitted without dedicated directory (%s)"%d) 
                else:
                    raise e
                
    def prepare_env(self, task):
        """ Build the segment global execution environment for a given task. 

        Include parameters and a bunch of utilitary routines
        (see the following members).

        """
        #gc.collect() ## to much time
        self.task = task

        env = self.pipe.env(self)
        glo = env._get_namespace()
        return (glo,env)

class InteractiveWorker(Worker):
    """ Run segments in interactive mode.
    
    Exceptions are not caught to ease debugging. After a debugging
    session you may want to call self.terminate() to properly stop the
    scheduler and update the database.
    """
        if self.pipe.matplotlib_interactive:
            matplotlib.use('Agg')  

        """ Reimplement exec_code from Worker.

        Main difference is that exception occuring in the segment code
        are not caught.
        """
        exec(compile(code[0], code[1], "exec"), glo)
        self.task.status = "done"

        """ Try to stop the processing of the ongoing task after a
        bug.
        """
        self.scheduler.abort_scheduling(self.task)

class ThreadWorker(Worker, threading.Thread):
    """Run segments in a separate thread.

    Jobs are obtained from a QueueScheduler.
    Exceptions are caught and dumped.
    """

    def __init__(self, *args, **keys):
        """ Initialize a thread worker. 
        """
        Worker.__init__(self, *args, **keys)

from multiprocessing.managers import BaseManager
from multiprocessing import Process
class SchedulerManager(BaseManager):
    """ Extension of the BaseManager class. 
    """
    pass
SchedulerManager.register('get_scheduler')

def catch_sigterm(signal, frame):
    raise AbortError(signal)

class ProcessWorker(Worker, Process):
    """ The worker is run as an independant process

    It access the scheduler through the use of managers
    """
    def __init__(self, address=('', 50000), authkey="secret", nice=0, **keys):
        """ Initialize a process worker
        """
        mgr = SchedulerManager(address=address, authkey=authkey)
        mgr.connect()
        Worker.__init__(self, mgr.get_scheduler(), **keys)
        keys['logger'].info( "connected to %s"%str(address))
        Process.__init__(self)
        # Catch all TERM signals
        signal.signal(signal.SIGHUP, catch_sigterm)
        signal.signal(signal.SIGINT, catch_sigterm)
        signal.signal(signal.SIGQUIT, catch_sigterm)
        signal.signal(signal.SIGILL, catch_sigterm)
        signal.signal(signal.SIGABRT, catch_sigterm)
        signal.signal(signal.SIGFPE, catch_sigterm)
        signal.signal(signal.SIGSEGV, catch_sigterm)

        # What to do for SIGPIPE
        
        signal.signal(signal.SIGALRM, catch_sigterm)
        signal.signal(signal.SIGTERM, catch_sigterm)
        
        signal.signal(signal.SIGUSR1, catch_sigterm)
        signal.signal(signal.SIGUSR2, catch_sigterm)

        # GE send XCPU and XFSZ for soft limit exceed
        signal.signal(signal.SIGXCPU, catch_sigterm)
        signal.signal(signal.SIGXFSZ, catch_sigterm)
        self.nice = nice

        os.nice(self.nice)
        Worker.run(self)

class SandBox(Worker):
    """ Provide a mean to come back in the execution state of a segment
    for latter debugging.
    """

    def __init__(self, pipeline):
        self.pipe=pipeline

    def setup(self, seg, global_env):
        """ Setup the working environment for segment seg. 
        """
        self.work_dir = self.pipe.get_curr_dir(seg)
        task = Task(seg)
        global_env.update(glo)

class _FakeWorker(ProcessWorker):
    """ Used for performance tests.
    """
        task.task_output = range(10)
        return task

## 
##
##

""" Utilitary functions to launch Pipes

- launch_interactive : launch a local worker in the interactive session
  (debugger compliant)

- launch_thread : launch a bunch of local workers in separate threads
  (SMP machine compliant)

- launch_process : launch a bunch of local workers in separate
  processes (usefull when the GIL becomes limitant ie most time spend
  in pure python processings)

- launch_ssh : launch a bunch of distant workers through ssh

- launch_pbs : submit a bunch of distant workers to any PBS compliant
  system (keep in mind that the pipe directory must be accessible to
  workers)

- launch_ccali : submit a bunch of distant workers to the BQS system
  at CCIN2P3

- launch_ccali_ge : submit a bunch of distant workers to the Grid Engine system
  at CCIN2P3

- launch_pbspro : submit a bunch of distant workers to the PBSpro
  manager (magique3, ...)

Run as a main, it can be used to launch a worker on a distant machine
"""

import pipelet.scheduler as scheduler
import pipelet.worker as worker
import pipelet.pipeline as pipeline
import logging.handlers
import sys
from pipelet.utils import get_log_file, init_logger
import socket  # to get the hostname

def launch_interactive(pipe, log_level=logging.INFO):
    """ Launch a local worker in the interactive session.

    This is debugger compliant, so that exception in the segment
    execution can be tracked.
    
    log_level values given by logging object: logging.DEBUG=10 logging.WARNING=30,
    logging.CRITICAL=50,etc). Set it to 0 to disable stream logging.

    >>> T = pipeline.Pipeline("", code_dir='./',
    ...                       prefix='./', permissive=True)
    >>> w,t = launch_interactive(T, log_level=logging.ERROR)
    >>> w.run()
    """
    init_logger('scheduler', get_log_file(pipe, 'scheduler'), level=log_level)
    s  = scheduler.Scheduler(pipe)
    wl = init_logger('worker', get_log_file(pipe, 'worker'), level=log_level)
    w  = worker.InteractiveWorker(s, wl)
    import threading
    t = threading.Thread(target=s.run)
    t.start()
    return (w, t)

def launch_thread(pipe, n, log_level=logging.WARNING ):
    """ Launch a bunch of local workers in separate threads.

    This is SMP machine compliant. Exceptions araising in the
    execution of any segment are caught and the corresponding task is
    marked as failed.

    >>> T = pipeline.Pipeline("", permissive=True)
    >>> launch_thread(T, 2)
    """
    s = scheduler.Scheduler(pipe)
    for i in range(n):
        wl = init_logger('worker%d' % i, get_log_file(pipe, 'worker%d' % i),
                         level=log_level)
        w = worker.ThreadWorker(s, wl)
        w.start()
    s.run()

    """ Extension of the BaseManager class.
    """
    pass

def launch_process(pipe, n, address=('', 50000), authkey='secret',
                   log_level=logging.WARNING, nice=0):
    """ Launch a bunch of local workers in separate processes .

    This is usefull (compared to launch_thread) when the GIL becomes
    limitant, which is bound to be the case when most time is spend in
    pure python processings.

    >>> launch_process(T, 2)
    """
    SchedulerManager.register('get_scheduler', callable=lambda: s)

    mgr = SchedulerManager(address=address, authkey=authkey)
    mgr.start()

    processlist = []
        w = worker.ProcessWorker(address=address, authkey=authkey,
                                 logger=wl, nice=nice)
        w.start()
        processlist.append(w)

    sched_proxy = mgr.get_scheduler()
    sched_proxy.run()

    # joining Zombie process
    for w in processlist:
        w.join()

def _scp(file, dest):
    """ Wrapper around the scp command."""
    subprocess.Popen(['scp', file, dest]).communicate()[0]

def launch_ssh(pipe, host_list, address=None, authkey='secret',
               log_level=logging.ERROR):
    """ Launch a bunch of distant workers through ssh.

    This is used mainly for testing purposes. It can be usefull to
    distribute computations on a pool of machine that share an access
    to a NAS.

    >>> launch_ssh(T, ['127.0.0.1'], address=('127.0.0.1',50000))
    """

    for i, h in enumerate(host_list):
        w = subprocess.Popen(
            ['ssh', h,
             "python -m pipelet.launchers -H %s -p %s -s %s"
             " -l %s" % (address[0], address[1], authkey,
                         get_log_file(pipe, 'worker%d' % i))])

    # 'launching the scheduler'

        w.wait()

def launch_pbs(pipe, n, address=None, authkey='secret', job_name="job_", log_level=logging.WARNING, cpu_time="2:00:00", server=False, job_header="""
#/bin/bash
echo $PYTHONPATH
""" ):
    """ Launch a bunch of distant workers through a PBS batch system.
    """
    
    init_logger ('scheduler', get_log_file (pipe, 'scheduler'), level=log_level)
    SchedulerManager.register('get_scheduler', callable=lambda:s)
    if server:
        mgr.start()

        jobfile = get_log_file (pipe, 'job%d'%i)
        errfile = jobfile.replace('job', 'err')
        logfile = jobfile.replace('job', 'log')
        f = file(jobfile, 'w')
        f.write (job_header+"\n")
        f.write ("#PBS -o %s\n"%logfile)
        f.write ("#PBS -e %s\n"%errfile)
        f.write ("#PBS -N %s%d\n"%(job_name,i))
        f.write ("#PBS -l select=1:ncpus=1,walltime=%s\n"%cpu_time)
        f.write ("python -m pipelet.launchers -H %s -p %s -s %s -l %s"%(address[0],address[1],authkey,jobfile.replace('job','worker')))
        f.close()
        subprocess.Popen(['qsub','-o' ,logfile, '-e',  errfile,jobfile]).communicate()[0]

    if server:
        print 'launching the scheduler'
        sched_proxy = mgr.get_scheduler()
        sched_proxy.run()                   

# def launch_pbs(pipe, n, address=None, authkey='secret', job_name="job_", log_level=logging.WARNING, cpu_time="2:00:00", job_header="""
# #/bin/bash
# echo $PYTHONPATH
# """ ):
#     """ Launch a bunch of distant workers through a PBS batch system using mpirun.
    
#     """

#     server_file=job_header+"""
# #PBS -l walltime=%s
# echo "export PIPELETD_HOST=$HOSTNAME" > ~/host_info.sh
# pipeletd -n -l %d -p %d -a $HOSTNAME << 'EOF' &
# %s
# EOF
# sleep 5
# mpirun -np %d python $PIPELETPATH/launchers.py -p %d -s %s -l %s -H $HOSTNAME
# """
#     import cPickle as pickle
#     pipedesc = pickle.dumps(pipe)
#     jobfile = get_log_file (pipe, '.job')
#     errfile = jobfile.replace('job', 'err')
#     logfile = jobfile.replace('job', 'log')

#     with closing(file(jobfile,'w')) as f:
#         f.write(server_file%(n, cpu_time, log_level, address[1], pipedesc, n, address[1], authkey, logfile))
#     subprocess.Popen(['qsub','-o' ,logfile, '-e',  errfile,jobfile]).communicate()[0]

def launch_nersc (pipe, n, ppn=1, port=50000, authkey='secret', job_name="job_", log_level=logging.WARNING, cpu_time="00:30:00", job_header="""
#PBS -q regular
#PBS -l nodes=%d:ppn=%d
"""):
     """ Launch a bunch of distant workers through a PBS batch system using aprun.
     """

#     server_file="""
##PBS -q regular
##PBS -l mppwidth=%d    
##PBS -l walltime=%s
#export CRAY_ROOTFS=DSL
#echo "export PIPELETD_HOST=$HOSTNAME" > ~/host_info.sh
#pipeletd -n -l %d -p %d -a $HOSTNAME << 'EOF' &
#%s
#EOF
#sleep 5
#aprun -np %d python $PIPELETPATH/launchers.py -p %d -s %s -l %s -H $HOST
#"""

     server_file=job_header+"""
#PBS -l walltime=%s
export CRAY_ROOTFS=DSL
echo "export PIPELETD_HOST=$HOSTNAME" > ~/host_info.sh
pipeletd -n -l %d -p %d -a $HOSTNAME << 'EOF' &
%s
EOF
sleep 5
mpirun -np %d python $PIPELETPATH/launchers.py -p %d -s %s -l %s -H $HOSTNAME
"""

     nnode=n/ppn
     if (n%ppn>0):
         nnode=nnode+1

     import cPickle as pickle
     pipedesc = pickle.dumps(pipe)
     jobfile = get_log_file (pipe, '.job')
     errfile = jobfile.replace('job', 'err')
     logfile = jobfile.replace('job', 'log')

     with closing(file(jobfile,'w')) as f:
         f.write(server_file%(nnode, ppn, cpu_time, log_level, port, pipedesc, n, port, authkey, logfile))
     subprocess.Popen(['qsub','-o' ,logfile, '-e',  errfile,jobfile]).communicate()[0]

_job_file = """
#/bin/zsh
export PATH=/afs/in2p3.fr/home/b/betoule/software/photcalib/tools:$PATH
echo $SHELL
python -m pipelet.launchers -H %s -p %s -s %s -l %s
"""

_server_file = """
#/bin/zsh
echo $USER
echo pouet > /sps/snls13/users/betoule/calib5_ref/log/11-06-06_15-13_scheduler_on_ccwl0674
cat /sps/snls13/users/betoule/calib5_ref/log/11-06-06_15-13_scheduler_on_ccwl0674
echo "export PIPELETD_HOST=$HOST\nexport PIPEWEB_HOST=$HOST\nexport PIPELETD_PORT=%d" > ~/host_info.sh
source ~/host_info.sh
pipeletd -n -l %s -p $PIPELETD_PORT -a $PIPELETD_HOST << 'EOF' &
%s
EOF
echo "starting web server at $(date)"
pipeweb start -n &
echo "web server started at $(date)"
sleep 2
echo "launching worker at $(date)"
python -m pipelet.launchers -H $PIPELETD_HOST -p $PIPELETD_PORT -s %s -l %s
echo "worker returned at $(date)
"""

_worker_file = """
#/bin/zsh
while [[ ! -e ~/host_info.sh ]]; do sleep 2; done;
sleep 5
"""

def launch_ccali(pipe, n, address=('127.0.0.1',5000), authkey='secret', job_dir="/sps/snls13/users/betoule/bqs/scripts",
                 job_name="job_", log_dir="/sps/snls13/users/betoule/bqs/logs", 
                 cpu_time=1000, scratch="512MB", mem="512MB",plateform='LINUX',server=False, log_level=logging.INFO,buff=None):
    """submit a bunch of distant workers to the BQS system
    at CCIN2P3.
    """
    if server:
        # Submit a job that will run the scheduler, a temporary web server and a single worker
        if buff is not None:
            b = ['-b', buff]
        else:
            b = []
        name = '%s_sched'%(job_name,)
        pipedesc = pickle.dumps(pipe)
        jobfile = get_log_file (pipe, name+'.job')
        with closing(file(jobfile,'w')) as f:
            f.write(_server_file%(address[1], log_level, pipedesc, authkey, logfile))
        com = ['qsub',
               '-l', 'platform=%s'%plateform,
               '-l', 'T=%d'%cpu_time,
               '-l', 'M=%s'%mem,
               '-l', 'scratch=%s'%scratch,
               '-o', logfile,
               '-e', errfile,
               '-N', name,
               jobfile]
        print " ".join(com)
        subprocess.Popen(com).communicate()[0]

    import re
    l  = subprocess.Popen(['qjob', '-wide','-nh'], stdout=subprocess.PIPE).communicate()[0]
    existing_process = re.findall( '%s([0-9]*)'%job_name,l)
    existing_process = filter(lambda x: x!='', existing_process)
    try:
        starting_num = max([int(p) for p in existing_process])+1
    except:
        starting_num = 0
    for i in range(starting_num,starting_num+n):
        name = '%s%d'%(job_name,i)
        jobfile = get_log_file (pipe, name+'.job%d'%i)
            f.write(_worker_file%(authkey,logfile))
        com=['qsub',
             '-l', 'platform=%s'%plateform,
             '-l', 'T=%d'%cpu_time,
             '-l', 'M=%s'%mem,
             '-l', 'scratch=%s'%scratch,
             '-o', logfile,
             '-e', errfile,
             '-N', name,
             jobfile]

def launch_ccali_ge(pipe, n, address=('127.0.0.1',5000), authkey='secret', job_dir="/sps/snls13/users/betoule/bqs/scripts",
                    job_name="job_", log_dir="/sps/snls13/users/betoule/bqs/logs", 
                    cpu_time=1000, scratch="512MB", mem="512MB",project='snovae',server=False, log_level=logging.INFO,buff=None, ressource=['sps']):
    """
    if server:
        else:
            b = []
               '-P', 'P_%s'%project,
               '-l', 'ct=%d'%cpu_time,
               '-l', 'vmem=%s'%mem,
               '-l', 'fsize=%s'%scratch,
               '-N', name]+reduce(lambda x,y : x+y, [['-l', '%s=1'%res] for res in ressource], [])+[jobfile]

    import re
    l  = subprocess.Popen(['qstat', '-r'], stdout=subprocess.PIPE).communicate()[0]
    try:
    except:
             '-P', 'P_%s'%project,
             '-l', 'ct=%d'%cpu_time,
             '-l', 'vmem=%s'%mem,
             '-l', 'fsize=%s'%scratch,
             '-N', name]+reduce(lambda x,y : x+y, [['-l', '%s=1'%res] for res in ressource], [])+[jobfile]

def launch_ccage_server (pipe, address=('127.0.0.1',5000),  authkey='secret', log_level=logging.WARNING, project="planck", cpu_time='24:00:00', scratch="512M", mem="512M", ressource=['sps']):
    """ Launch a scheduler + web server through a SGE batch system @ccin2p3.                                   
    """
    server_file="""#! /bin/zsh -l 
echo $HOST
echo "export PIPELETD_HOST=$HOST\nexport PIPEWEB_HOST=$HOST\nexport PIPELETD_PORT=%d" > $HOME/host_info.sh
source $HOME/host_info.sh
sleep 5
echo "web server returned at $(date)"
pipeletd -n -l %s -p $PIPELETD_PORT -a $HOST << 'EOF' 
%s
EOF
"""

    port = address[1]
    import cPickle as pickle
    pipedesc = pickle.dumps(pipe)
    name = 'sched'
    jobfile = get_log_file (pipe, name+'.job')
    errfile = jobfile.replace('job', 'err')
    logfile = jobfile.replace('job', 'log')
    
    with closing(file(jobfile,'w')) as f:
        f.write(server_file%(port, log_level, pipedesc))

    com=['qsub', 
         '-P', 'P_%s'%project,
         #'-q', 'demon', 
         #'-l', 'demon=1', 
         '-l', 'sps=1',
         '-l', 'fsize=%s'%scratch,
         '-l', 'vmem=%s'%mem,
         '-l', 'ct=%s'%cpu_time,
         '-o' ,logfile, 
         '-e',  errfile,
         '-N', name]+reduce(lambda x,y : x+y, [['-l', '%s=1'%res] for res in ressource], [])+[jobfile]
    subprocess.Popen(com).communicate()[0]

def launch_ccage_worker (pipe, n, address=('127.0.0.1',5000),  authkey='secret', log_level=logging.WARNING, project="planck", cpu_time='01:00:00', scratch="512M", mem="512M", ressource=['sps'], multicores=False):
    """ Launch n worker using job array
    """        
    worker_file = """#! /bin/zsh -l 
echo $HOST
sleep 1
python -m pipelet.launchers -H $PIPELETD_HOST -p $PIPELETD_PORT -s %s -l %s$SGE_TASK_ID
echo "worker returned at $(date)"
"""
    name = 'worker'
        f.write(worker_file%(authkey,logfile))
    if multicores:
        mc = ['-pe', 'multicores' , '%d'%multicores]
    else:
        mc = []
         '-t', '1-%d'%n,
         '-N', name]+mc+reduce(lambda x,y : x+y, [['-l', '%s=1'%res] for res in ressource], [])+[jobfile]

    parser = optparse.OptionParser()
    parser.add_option('-H', '--host', metavar='hostname',
                      help='hostame or adress of the scheduler server',
                      default=socket.gethostname())
    parser.add_option('-p', '--port', metavar='port',
                      help='port the scheduler is listenning to',
                      default=50000, type='int')
    parser.add_option('-s', '--secret', metavar='key',
                      help='authentication key', default='secret')
    parser.add_option('-l', '--logfile', metavar='logfile',
                      help='worker log filename')
#    parser.add_option('-L', '--log-level', metavar='level',
#                      type=
#                      help='worker log filename')

    # print (options.host, options.port)
    # print options.secret
    name  = options.logfile.split("_")[-1]
    wl    = init_logger(name, options.logfile)
    w = worker.ProcessWorker(address=(options.host, options.port),
                             authkey=options.secret, logger=wl)
    sys.argv = [sys.argv[0]]  # do not propage command line argument to script
    w.run()

##
##
##

from os import path
from directive import Depend

logger = logging.getLogger('scheduler')

def get_git_revision(path):
    """ Return Git revision if any.
    """
    rev = ""
    try:
        P = subprocess.Popen(["git", "rev-parse", "HEAD"],
                             stdout=subprocess.PIPE, cwd=path)
        (rev, jk) = P.communicate()
    except:
        pass
    return rev

class RepositoryError(Exception):
    """ To avoid catching Exception
    """
    def __init__(self, value):
        self.parameter = value

        return repr(self.parameter)

class Repository:
    """ Location of the segment's source code.

    A reposiroty is the location where to find the segment's source
    code files.  It returns the full path of all files that will be
    used to execute the segment (code, hook, ...). By default,
    local repositories are used, but it can also be a version control
    system (git, CVS, ...).
    """

    def __init__(self, lstseg, permissive=False):
        """ Initialize a repository. 

        If permissive is False raise an error when segment script is
        not found. Otherwise default to an empty script
        
        >>> r = Repository([], permissive=True)

        """
        ## dict, code string corresponding to each segment
        self._code   = {}
        ## dict, code string corresponding to each hook
        self._hook   = {}
        ## dict, code string corresponding to each external dependency
        self._deps   = {}
        ## dict, code string
        self._all_string = {}
        self._fill_dict(lstseg)

    @staticmethod
    def _match_fn(fns, seg, hook=None):
        """ Find filename matching pipelet rules for a given segment.

        >>> print Repository._match_fn( ['first.py'], 'first111')
        first.py
        """
        if hook is None:
            for s in range(len(seg)):
                def match(x):
                    return path.basename(x) == seg[0:len(seg) - s] + '.py'
                fn = filter(match, fns)
                if len(fn) > 0:
                    return fn[0]
        else:
                for h in range(len(hook)):
                    def match(x):
                        return path.basename(x) == seg[0:len(seg) - s]\
                            + '_' + hook[0:len(hook) - h] + '.py'
                    fn = filter(match, fns)
                    if len(fn) > 0:
                        return fn[0]
        raise RepositoryError('No source file corresponding to segment '
                              '%s and hook %s' % (seg, hook))

    def _fill_dict(self, lstseg):
        """ Initialize all dictionnaries.
        """
        fns = self.get_fns()
        for s in lstseg:
            ## code
            try:
                fn = self._match_fn(fns, s)
            except RepositoryError, e:
                logger.warning(str(e))
                if not self.permissive:
                    raise e
                else:
                    fn = ""
            if not fn:
                source = ""
                logger.warning("Assuming empty script for segment %s" % s)
            else:
                source = self.get_code_source(fn)
            self._code[s] = (source, fn)  # compile(source, fn, "exec")
            self._all_string[s] = source

            ## hooks
            lsthook = self._get_hook_list(source)
            self._hook[s] = {}
            for h in lsthook:
                try:
                    fn = self._match_fn(fns, s, h)
                    source = self.get_code_source(fn)
                except RepositoryError, e:  # No source file
                    logger.warning(str(e))
                    source = ""
                    fn = ""
                self._hook[s][h] = (source, fn)  # compile(source, fn, "exec")
                self._all_string[s] += source

            ## deps
            depend = self.get_directive(Depend, s)
            self._deps[s] = {}
            for fn in depend.deps:
                source = self.get_file_content(fn)
                self._deps[s][fn] = source

    def get_all_string(self, seg):
        """ Return all code as a string.
        """
        return self._all_string[seg]

    def get_directive(self, Direct, seg):
        """ Initialize a directive object from segment seg.

        >>> r = LocalRepository(['fourth'], "./test/first_test")
        >>> print r.get_directive (Multiplex, 'fourth').__class__
        pipelet.directive.Multiplex
        """
        c = self._all_string[seg]
        d = Direct()
        for l in c.splitlines():
            try:
                if l[0] != "#":
                    continue
                continue
            try:
                d.parse(l)
            except StopIteration:
                pass
        return d

    def _get_hook_list(code):
        """ Return the list of hooks by parsing the code source.
        
        >>> print Repository._get_hook_list ("a=1\nhook('thehook',globals())\n")
        ['thehook']
        """
        hook_list = []
        for l in code.split("hook(")[1:]:
            hook_list.append(l.split(",")[0][1:-1])
        return hook_list

    def get_hook_list(self, seg):
        """ Return the list of hooks for a segment
        """
        return self._hook[seg].keys()

    def get_code_string(self, seg):
        """ Return segment code as a string.

        This is done to prevent file changes after hashkey computation.
        """
        return self._code[seg]

    def get_deps_string(self, seg):
        """ Return external dependency code as a string.

        """
        return self._deps[seg]

    def get_hook_string(self, seg, hook):
        """ Return hook code as a string.

        """
        return self._hook[seg][hook]

    def get_docline(self, seg):
        """ Return the segment synopsis doc line.
        """
        import pydoc
        return pydoc.source_synopsis(self._code[seg].co_filename)

    def get_file_content(filename):
        """ Read content of a given file

        This is the default to read dependencies of a segment.

        >>> print Repository.get_file_content ('./test/first_test/third.py')[0:17]
        adj = 'the_brave'
        """
        fid  = open(path.expandvars(path.expanduser(filename)), "r")
        code = fid.read()
        fid.close()
        return code

    def get_fns(self):
        """ Return filename candidates.
        """
        return []

class LocalRepository(Repository):
    """A local repository is made of a main directory where to find segment's
    code file.
    """

    def __init__(self, lst_seg, src_path, permissive=False):
        """ Initialize a local repository.

        >>> r = LocalRepository(['first','second'], "./test/first_test", permissive=True)
        >>> print len(r._code)
        2
        """

        self.permissive = permissive
        ## string, where to find segment's source code file
        self.src_path = path.expanduser(path.expandvars(src_path))

        self._rev = get_git_revision(self.src_path)
        self._fill_dict(lst_seg)

    def get_code_source(filename):
        """ Read source code from file

        >>> print LocalRepository.get_code_source ('./test/first_test/third.py')[0:17]
        """
        fid  = open(filename, "r")
        # if len(self._rev)>0:
        # code = "### %s: %s\n%s"%(path.basename(filename), self._rev , code)

        >>> print r.get_fns()
        ['./test/first_test/second.py', './test/first_test/fourth.py', './test/first_test/third.py', './test/first_test/main.py', './test/first_test/first.py']
        """
        return glob(path.join(self.src_path, '*.py'))

class GitRepository(LocalRepository):
    """ A Git repository.

    """
        pass

    def get_code_source(self, filename):
        pass

        pass

class CVSRepository(Repository):
    """ A CVS repository.

    """
        pass

        pass

        pass

get_file_content() (pipelet.repository.Repository static method)

get_code_source() (pipelet.repository.LocalRepository static method)