def create_annotation(self, id_scope=None):
        from vistrails.db.domain import IdScope

        if id_scope is None:
            id_scope = IdScope()
        annotation = Annotation(id=id_scope.getNewId(Annotation.vtType), key="akey %s", value="some value %s")
        return annotation

    def create_annotation(self, id_scope=None):
        from vistrails.db.domain import IdScope

        if id_scope is None:
            id_scope = IdScope()
        annotation = ActionAnnotation(
            id=id_scope.getNewId(ActionAnnotation.vtType), key="akey", action_id=1L, value="some value", user="test"
        )
        return annotation

    def create_control_parameter(self, id_scope=None):
        from vistrails.db.domain import IdScope

        if id_scope is None:
            id_scope = IdScope()
        control_parameter = ModuleControlParam(id=id_scope.getNewId(ModuleControlParam.vtType),
                                name='name %s',
                                value='some value %s')
        return control_parameter

    def create_annotation(self, id_scope=None):
        from vistrails.db.domain import IdScope

        if id_scope is None:
            id_scope = IdScope()
        annotation = \
            ActionAnnotation(id=id_scope.getNewId(ActionAnnotation.vtType),
                             key='akey', action_id=1L,
                             value='some value', user='test')
        return annotation

 def create_module(self, id_scope=None):
     from vistrails.core.modules.basic_modules import identifier as basic_pkg
     from vistrails.db.domain import IdScope
     if id_scope is None:
         id_scope = IdScope()
     
     params = [ModuleParam(id=id_scope.getNewId(ModuleParam.vtType),
                               type='Int',
                               val='1')]
     functions = [ModuleFunction(id=id_scope.getNewId(ModuleFunction.vtType),
                                 name='value',
                                 parameters=params)]
     module = Module(id=id_scope.getNewId(Module.vtType),
                     name='Float',
                     package=basic_pkg,
                     functions=functions)
     return module

 def __init__(self, id, vt_version, id_scope=None):
     DBMashuptrail.__init__(self, None, id, version="", vtVersion=vt_version)
     self.db_actions = []
     self.currentVersion = -1
     self.db_annotations = []
     self.db_actionAnnotations = []
     if not id_scope:
         self.id_scope = IdScope(1L)
     else:
         self.id_scope = id_scope

    def create_ops(self, id_scope=IdScope()):
        from vistrails.core.modules.basic_modules import identifier as basic_pkg
        from vistrails.core.vistrail.module import Module
        from vistrails.core.vistrail.module_function import ModuleFunction
        from vistrails.core.vistrail.module_param import ModuleParam
        from vistrails.core.vistrail.annotation import Annotation
        
        if id_scope is None:
            id_scope = IdScope(remap={AddOp.vtType: 'operation',
                                      ChangeOp.vtType: 'operation',
                                      DeleteOp.vtType: 'operation'})

        m = Module(id=id_scope.getNewId(Module.vtType),
                   name='Float',
                   package=basic_pkg)
        add_op = AddOp(id=id_scope.getNewId(AddOp.vtType),
                       what=Module.vtType,
                       objectId=m.id,
                       data=m)
        function = ModuleFunction(id=id_scope.getNewId(ModuleFunction.vtType),
                                  name='value')
        change_op = ChangeOp(id=id_scope.getNewId(ChangeOp.vtType),
                             what=ModuleFunction.vtType,
                             oldObjId=2,
                             newObjId=function.real_id,
                             parentObjId=m.id,
                             parentObjType=Module.vtType,
                             data=function)
        param = ModuleParam(id=id_scope.getNewId(ModuleParam.vtType),
                            type='Float',
                            val='1.0')
        
        delete_op = DeleteOp(id=id_scope.getNewId(DeleteOp.vtType),
                             what=ModuleParam.vtType,
                             objectId=param.real_id,
                             parentObjId=function.real_id,
                             parentObjType=ModuleFunction.vtType)

        annotation = Annotation(id=id_scope.getNewId(Annotation.vtType),
                                key='foo',
                                value='bar')
        add_annotation = AddOp(id=id_scope.getNewId(AddOp.vtType),
                               what=Annotation.vtType,
                               objectId=m.id,
                               data=annotation)
        
        return [add_op, change_op, delete_op, add_annotation]

    def create_action(self, id_scope=None):
        from vistrails.core.modules.basic_modules import identifier as basic_pkg
        from vistrails.core.vistrail.action import Action
        from vistrails.core.vistrail.module import Module
        from vistrails.core.vistrail.module_function import ModuleFunction
        from vistrails.core.vistrail.module_param import ModuleParam
        from vistrails.core.vistrail.operation import AddOp
        from vistrails.db.domain import IdScope
        from datetime import datetime
        
        if id_scope is None:
            id_scope = IdScope()
        param = ModuleParam(id=id_scope.getNewId(ModuleParam.vtType),
                            type='Integer',
                            val='1')
        function = ModuleFunction(id=id_scope.getNewId(ModuleFunction.vtType),
                                  name='value',
                                  parameters=[param])
        m = Module(id=id_scope.getNewId(Module.vtType),
                   name='Float',
                   package=basic_pkg,
                   functions=[function])

        add_op = AddOp(id=id_scope.getNewId('operation'),
                       what='module',
                       objectId=m.id,
                       data=m)
        action = Action(id=id_scope.getNewId(Action.vtType),
                        prevId=0,
                        date=datetime(2007,11,18),
                        operations=[add_op])
        return action

    def createMashupController(self, vt_controller, version, view=DummyView()):
        #print "Manager creating mashup controller ", vt_controller, version
        newvt_controller = MashupsManager.copyVistrailController(vt_controller,
                                                                view)
        mashuptrail = \
         MashupsManager.getMashuptrailforVersionInVistrailController(vt_controller,
                                                                     version)
        if mashuptrail is None:
            (p_mashuptrail, p_version) = \
                     MashupsManager.findClosestParentMashuptrail(vt_controller, 
                                                                 version)
            id_scope = IdScope(1L)
            if p_mashuptrail is not None:
                version_name = vt_controller.get_pipeline_name(p_version)
                (res, mshpv) = MashupsManager.showFoundMashupsDialog(p_mashuptrail, 
                                                            version_name)
                if res in ['Copy', 'Move']:
                    pipeline = newvt_controller.vistrail.getPipeline(version)
                    if res == 'Copy':
                        # we will copy the mashup from the parent trail and 
                        # validate it to the current pipeline before adding
                        # to the current mashup trail
                        mashuptrail = Mashuptrail(self.getNewMashuptrailId(), 
                                                  version, id_scope)
                        p_mashup = p_mashuptrail.getMashup(mshpv)
                        mashup = p_mashup.do_copy()
                        mashup.id_scope = id_scope
                        mashup.version = version
                        mashup.validateForPipeline(pipeline)
                        currVersion = mashuptrail.addVersion(
                                      parent_id=mashuptrail.getLatestVersion(),
                                      mashup=mashup, 
                                      user=vistrails.core.system.current_user(),
                                      date=vistrails.core.system.current_time())
                        mashuptrail.currentVersion = currVersion
                        mashuptrail.updateIdScope()
                        p_tag = p_mashuptrail.getTagForActionId(mshpv)
                        if p_tag == '':
                            tag = "<latest>"
                        tag = "Copy from %s"%p_tag
                        MashupsManager.addMashuptrailtoVistrailController(vt_controller,
                                                                          mashuptrail)    
                        
                    elif res == 'Move':
                        # we will move the parent trail and validate all mashups
                        # for the current pipeline to make sure they will be 
                        # executable for the current version

                        mashuptrail = p_mashuptrail
                        currVersion = mashuptrail.getLatestVersion()
                        mashuptrail.currentVersion = currVersion
                        mashuptrail.validateMashupsForPipeline(version, pipeline)
                        tag = None
                        
                    mashuptrail.vtVersion = version
                    mshpController = MashupController(vt_controller, 
                                                      newvt_controller, 
                                                      version, mashuptrail)
                    mshpController.setCurrentVersion(mashuptrail.currentVersion)
                    # this is to make sure the pipeline displayed in the mashup
                    # view is consistent with the list of aliases in the central
                    # panel
                    mshpController.updatePipelineAliasesFromCurrentMashup()
                    if tag is not None:
                        mshpController.updateCurrentTag(tag)
                    return mshpController
                
            mashuptrail = Mashuptrail(self.getNewMashuptrailId(), version, 
                                      id_scope)
            pipeline = newvt_controller.vistrail.getPipeline(version)
            id = id_scope.getNewId('mashup')
            mashup = Mashup(id=id, name="mashup%s"%id, vtid=vt_controller.locator, 
                        version=version)
            mashup.loadAliasesFromPipeline(pipeline, id_scope)
            currVersion = mashuptrail.addVersion(parent_id=mashuptrail.getLatestVersion(),
                                             mashup=mashup, 
                                             user=vistrails.core.system.current_user(),
                                             date=vistrails.core.system.current_time())
    
            mashuptrail.currentVersion = currVersion
            
            MashupsManager.addMashuptrailtoVistrailController(vt_controller,
                                                              mashuptrail)
            mshpController = MashupController(vt_controller,
                                              newvt_controller, 
                                              version, mashuptrail)
            mshpController.setCurrentVersion(mashuptrail.currentVersion)
            if mshpController.currentVersion == 1L:
                mshpController.updateCurrentTag("ROOT")
        else:
            #print "----> found mashuptrail ", mashuptrail.currentVersion
            mshpController = MashupController(vt_controller, 
                                              newvt_controller, 
                                              version, mashuptrail)
            mshpController.setCurrentVersion(mashuptrail.currentVersion)
            mshpController.updatePipelineAliasesFromCurrentMashup()
        
        return mshpController

    def updateFunctionPort(self):
        """
        Function to be used inside the updateUsptream method of the Map module. It
        updates the module connected to the FunctionPort port, executing it in
        parallel.
        """
        nameInput = self.getInputFromPort('InputPort')
        nameOutput = self.getInputFromPort('OutputPort')
        rawInputList = self.getInputFromPort('InputList')

        # Create inputList to always have iterable elements
        # to simplify code
        if len(nameInput) == 1:
            element_is_iter = False
            inputList = [[element] for element in rawInputList]
        else:
            element_is_iter = True
            inputList = rawInputList

        workflows = []
        module = None
        vtType = None

        # iterating through the connectors
        for connector in self.inputPorts.get('FunctionPort'):
            module = connector.obj

            # pipeline
            original_pipeline = connector.obj.moduleInfo['pipeline']

            # module
            module_id = connector.obj.moduleInfo['moduleId']
            vtType = original_pipeline.modules[module_id].vtType

            # serialize the module for each value in the list
            for i, element in enumerate(inputList):
                if element_is_iter:
                    self.element = element
                else:
                    self.element = element[0]

                # checking type and setting input in the module
                self.typeChecking(connector.obj, nameInput, inputList)
                self.setInputValues(connector.obj, nameInput, element)

                pipeline_db_module = original_pipeline.modules[module_id].do_copy()

                # transforming a subworkflow in a group
                # TODO: should we also transform inner subworkflows?
                if pipeline_db_module.is_abstraction():
                    group = Group(id=pipeline_db_module.id,
                                  cache=pipeline_db_module.cache,
                                  location=pipeline_db_module.location,
                                  functions=pipeline_db_module.functions,
                                  annotations=pipeline_db_module.annotations)

                    source_port_specs = pipeline_db_module.sourcePorts()
                    dest_port_specs = pipeline_db_module.destinationPorts()
                    for source_port_spec in source_port_specs:
                        group.add_port_spec(source_port_spec)
                    for dest_port_spec in dest_port_specs:
                        group.add_port_spec(dest_port_spec)

                    group.pipeline = pipeline_db_module.pipeline
                    pipeline_db_module = group

                # getting highest id between functions to guarantee unique ids
                # TODO: can get current IdScope here?
                if pipeline_db_module.functions:
                    high_id = max(function.db_id
                                  for function in pipeline_db_module.functions)
                else:
                    high_id = 0

                # adding function and parameter to module in pipeline
                # TODO: 'pos' should not be always 0 here
                id_scope = IdScope(beginId=long(high_id+1))
                for elementValue, inputPort in izip(element, nameInput):

                    p_spec = pipeline_db_module.get_port_spec(inputPort, 'input')
                    descrs = p_spec.descriptors()
                    if len(descrs) != 1:
                        raise ModuleError(
                                self,
                                "Tuple input ports are not supported")
                    if not issubclass(descrs[0].module, Constant):
                        raise ModuleError(
                                self,
                                "Module inputs should be Constant types")
                    type = p_spec.sigstring[1:-1]

                    mod_function = ModuleFunction(id=id_scope.getNewId(ModuleFunction.vtType),
                                                  pos=0,
                                                  name=inputPort)
                    mod_param = ModuleParam(id=0L,
                                            pos=0,
                                            type=type,
                                            val=elementValue)

                    mod_function.add_parameter(mod_param)
#.........这里部分代码省略.........

class Mashuptrail(DBMashuptrail):
    """ MashupTrail is a class that stores versions of Mashups.
    For now it keeps a linear history."""
    def __init__(self, id, vt_version, id_scope=None):
        DBMashuptrail.__init__(self, None, id, version="", vtVersion=vt_version)
        self.db_actions = []
        self.currentVersion = -1
        self.db_annotations = []
        self.db_actionAnnotations = []
        if not id_scope:
            self.id_scope = IdScope(1L)
        else:
            self.id_scope = id_scope
        
    id = DBMashuptrail.db_name
    vtVersion = DBMashuptrail.db_vtVersion
    actions = DBMashuptrail.db_actions
    annotations = DBMashuptrail.db_annotations
    actionAnnotations = DBMashuptrail.db_actionAnnotations
    
    def _get_actionMap(self):
        return self.db_actions_id_index
    actionMap = property(_get_actionMap)
    
    @staticmethod
    def convert(_mtrail):
        _mtrail.__class__ = Mashuptrail

        for action in _mtrail.actions:
            Action.convert(action)

        for annotation in _mtrail.annotations:
            Annotation.convert(annotation)

        for aannotation in _mtrail.actionAnnotations:
            ActionAnnotation.convert(aannotation)
        _mtrail.id_scope = IdScope(1L)
        _mtrail.updateIdScope()
            
    def addVersion(self, parent_id, mashup, user, date):
        id = self.getLatestVersion() + 1
        mashup.id_scope = self.id_scope
        mashup.id = id
        mashup.version = self.vtVersion
        action = Action(id=id, prevId=parent_id, mashup=mashup,
                        user=user, date=date)
        self.db_add_action(action)
        
        return action.id
    
    def __copy__(self):
        return Mashuptrail.do_copy(self)
    
    def do_copy(self, new_ids=False, id_scope=None, id_remap=None):
        """do_copy() -> Mashuptrail 
        returns a clone of itself"""
        cp = DBMashuptrail.do_copy(self, new_ids, id_scope, id_remap)
        Mashuptrail.convert(cp)        
        cp.currentVersion = self.currentVersion        
        return cp
    
    def getLatestVersion(self):
        if not self.actions:
            return 0
        max_ver = max(a.id for a in self.actions)
        return max_ver

    def getMashup(self, version):
        if version in self.actionMap.keys():
            return self.actionMap[version].mashup
        else:
            return None
        
    def validateMashupsForPipeline(self, version, pipeline):
        """validateMashupsForPipeline(version:long, pipeline:Pipeline)->None
        This will make sure that the aliases present in all mashups are 
        consistent with the current pipeline. 
        
        """
        for action in self.actions:
            action.mashup.id_scope = self.id_scope
            action.mashup.validateForPipeline(pipeline)
            action.mashup.version = version
    
    ####################################################################
    ## Tag manipulation
    ##    
    def hasTagWithName(self, name):
        for a in self.actionAnnotations:
            if a.key == "__tag__":
                if a.value == name:
                    return True
        return False
        
    def hasTagForActionId(self, action_id):
        for a in self.actionAnnotations:
            if a.key == "__tag__" and a.action_id == action_id:
                return True
        return False
    
#.........这里部分代码省略.........

def create_opm(workflow, version, log, reg):
    id_scope = IdScope()
    processes = []
    # conn_artifacts = {}
    artifacts = []
    dependencies = []
    accounts = []
    depth_accounts = {}
    file_artifacts = {}
    db_artifacts = {}

    def do_create_process(workflow, item_exec, account, module_processes):
        process = create_process(item_exec, account, id_scope)
        print 'adding process', process.db_id,
        if hasattr(item_exec, 'db_module_name'):
            print item_exec.db_module_name
        elif hasattr(item_exec, 'db_group_nane'):
            print item_exec.db_module_name
        processes.append(process)
        module = workflow.db_modules_id_index[item_exec.db_module_id]
        module_processes[module.db_id] = (module, process)

    def get_package(reg, pkg_identifier, pkg_version=''):
        if not pkg_version:
            # spin and get current package
            for pkg in reg.db_packages:
                if pkg.db_identifier == pkg_identifier:
                    break
                pkg = None
        else:
            pkg = reg.db_packages_identifier_index[(pkg_identifier,
                                                    pkg_version)]
        return pkg

    def process_exec(item_exec, workflow, account, upstream_lookup,
                     downstream_lookup, depth, conn_artifacts=None,
                     function_artifacts=None, module_processes=None,
                     in_upstream_artifacts={}, in_downstream_artifacts={},
                     add_extras=False):

        print 'in_upstream:', [(n, x.db_id) 
                               for n, x_list in in_upstream_artifacts.iteritems() for x in x_list]
        print 'in_downstream:', [(n, x.db_id)  
                                 for n, x_list in in_downstream_artifacts.iteritems() for x in x_list]
        # FIXME merge conn_artifacts and function_artifacts
        # problem is that a conn_artifact is OUTPUT while function_artifact
        # is INPUT
        if conn_artifacts is None:
            conn_artifacts = {}
        if function_artifacts is None:
            function_artifacts = {}
        if module_processes is None:
            module_processes = {}
#         while item_exec.vtType == DBLoopExec.vtType:
#             item_exec = item_exec.db_item_execs[0]
        (module, process) = module_processes[item_exec.db_module_id]

        def process_connection(conn):
            source = conn.db_ports_type_index['source']
            source_t = (source.db_moduleId, source.db_name)
            in_cache = False
            print '!!! processing', source_t
            if source_t in conn_artifacts:
                artifact = conn_artifacts[source_t]
                in_cache = True
            else:
                # key off source module and port name
                # get descriptor from registry and then port_spec
                # store port_spec as artifact

                if source.db_moduleId < 0:
                    dest = conn.db_ports_type_index['destination']
                    module = source.db_module
                else:
                    module = workflow.db_modules_id_index[source.db_moduleId]
                print module.db_name, module.db_id

                pkg = get_package(reg, module.db_package, module.db_version)

                if not module.db_namespace:
                    module_namespace = ''
                else:
                    module_namespace = module.db_namespace
                module_desc = \
                    pkg.db_module_descriptors_name_index[(module.db_name,
                                                          module_namespace,
                                                          '')]
                # FIXME make work for module port_specs, too
                # for example, a PythonSource with a given port in 
                # module.db_portSpecs
                port_spec = None
                spec_t = (source.db_name, 'output')
                if spec_t in module.db_portSpecs_name_index:
                    port_spec = module.db_portSpecs_name_index[spec_t]
                while port_spec is None and \
                        module_desc.db_id != reg.db_root_descriptor_id:
                    if spec_t in module_desc.db_portSpecs_name_index:
                        port_spec = module_desc.db_portSpecs_name_index[spec_t]
                    base_id = module_desc.db_base_descriptor_id

#.........这里部分代码省略.........

    def execute(self, *args, **kwargs):
        """Execute the pipeline.

        Positional arguments are either input values (created from
        ``module == value``, where `module` is a Module from the pipeline and
        `value` is some value or Function instance) for the pipeline's
        InputPorts, or Module instances (to select sink modules).

        Keyword arguments are also used to set InputPort by looking up inputs
        by name.

        Example::

           input_bound = pipeline.get_input('higher_bound')
           input_url = pipeline.get_input('url')
           sinkmodule = pipeline.get_module(32)
           pipeline.execute(sinkmodule,
                            input_bound == vt.Function(Integer, 10),
                            input_url == 'http://www.vistrails.org/',
                            resolution=15)  # kwarg: only one equal sign
        """
        sinks = set()
        inputs = {}

        reg = get_module_registry()
        InputPort_desc = reg.get_descriptor_by_name(
                get_vistrails_basic_pkg_id(),
                'InputPort')

        # Read args
        for arg in args:
            if isinstance(arg, ModuleValuePair):
                if arg.module.id in inputs:
                    raise ValueError(
                            "Multiple values set for InputPort %r" %
                            get_inputoutput_name(arg.module))
                if not reg.is_descriptor_subclass(arg.module.module_descriptor,
                                                  InputPort_desc):
                    raise ValueError("Module %d is not an InputPort" %
                                     arg.module.id)
                inputs[arg.module.id] = arg.value
            elif isinstance(arg, Module):
                sinks.add(arg.module_id)

        # Read kwargs
        for key, value in kwargs.iteritems():
            key = self.get_input(key)  # Might raise KeyError
            if key.module_id in inputs:
                raise ValueError("Multiple values set for InputPort %r" %
                                 get_inputoutput_name(key.module))
            inputs[key.module_id] = value

        reason = "API pipeline execution"
        sinks = sinks or None

        # Use controller only if no inputs were passed in
        if (not inputs and self.vistrail is not None and
                self.vistrail.current_version == self.version):
            controller = self.vistrail.controller
            results, changed = controller.execute_workflow_list([[
                    controller.locator,  # locator
                    self.version,  # version
                    self.pipeline,  # pipeline
                    DummyView(),  # view
                    None,  # custom_aliases
                    None,  # custom_params
                    reason,  # reason
                    sinks,  # sinks
                    None,  # extra_info
                    ]])
            result, = results
        else:
            pipeline = self.pipeline
            if inputs:
                id_scope = IdScope(1)
                pipeline = pipeline.do_copy(False, id_scope)

                # A hach to get ids from id_scope that we know won't collide:
                # make them negative
                id_scope.getNewId = lambda t, g=id_scope.getNewId: -g(t)

                create_module = \
                        VistrailController.create_module_from_descriptor_static
                create_function = VistrailController.create_function_static
                create_connection = VistrailController.create_connection_static
                # Fills in the ExternalPipe ports
                for module_id, values in inputs.iteritems():
                    module = pipeline.modules[module_id]
                    if not isinstance(values, (list, tuple)):
                        values = [values]

                    # Guess the type of the InputPort
                    _, sigstrings, _, _, _ = get_port_spec_info(pipeline, module)
                    sigstrings = parse_port_spec_string(sigstrings)

                    # Convert whatever we got to a list of strings, for the
                    # pipeline
                    values = [reg.convert_port_val(val, sigstring, None)
                              for val, sigstring in izip(values, sigstrings)]

#.........这里部分代码省略.........