Do the workflow training
494352
def train(self, **kwargs):
        
        log.info("Starting to train...")
        if not self.setup_run:
            self.setup()
        self.trained_tasks = []
        for task in self.tasks:
            data = self.reformatted_input[task.data_format]['data']
            target = self.reformatted_input[task.data_format]['target']
            if data is None:
                raise Exception("Data cannot be none.  Check the config file to make sure the right input is being read.")
            kwargs['data']=data
            kwargs['target']=target
            trained_task = self.execute_train_task_with_dependencies(task, **kwargs)
            self.trained_tasks.append(trained_task)
            #If the trained task alters the data in any way, pass it down the chain to the next task
            if hasattr(trained_task, 'data'):
                self.reformatted_input[task.data_format]['data'] = trained_task.data
        log.info("Finished training.")
965860
def run(self):
        """"""
        # Try to create the directory
        if not os.path.exists(self.output):
            try:
                os.mkdir(self.output)
            except:
                print 'failed to create output directory %s' % self.output

        # Be sure it is a directory
        if not os.path.isdir(self.output):
            print 'invalid output directory %s' % self.output
            sys.exit(1)

        # Create the CSV handlers
        visitors = [
            _CompaniesCSV(self.output),
            _ActivitiesCSV(self.output),
            _ActivitiesSeenCSV(self.output),
            _QSACSV(self.output),
        ]

        # Run by each company populating the CSV files
        for path in glob.glob(os.path.join(self.input, '*.json')):
            with open(path, 'r') as f:
                try:
                    data = json.load(f, encoding='utf-8')
                except ValueError:
                    continue

                for visitor in visitors:
                    visitor.visit(data)
765442
void performPostCreationTasks()
    {
        StatusAction statusAction = new StatusAction();
        try
        {
            ResponseEvents events = CoherentManagerConnection.sendEventGeneratingAction(statusAction, 1000);
            for (ResponseEvent event : events.getEvents())
            {
                if (event instanceof StatusEvent)
                {
                    // do nothing. Creating the events will register the
                    // channels, which is after all what we are trying to do.
                }
            }

        }
        catch (IllegalArgumentException | IllegalStateException | IOException | TimeoutException e)
        {
            logger.error(e, e);
        }

    }
129350
def _execute_workflow(cls, process_args):
        
        workflow, input_args, log_path = process_args

        if log_path:
            logger = logging.getLogger()
            logger.setLevel(logging.DEBUG)
            handler = cls._get_log_handler(log_path)
            logger.addHandler(handler)

        stats = {'start_time': dt.datetime.now()}
        try:
            _ = workflow.execute(input_args, monitor=True)
        except BaseException:
            stats['error'] = traceback.format_exc()
        stats['end_time'] = dt.datetime.now()

        if log_path:
            handler.close()
            logger.removeHandler(handler)

        return stats
694204
def sos_run(workflow=None,
            targets=None,
            shared=None,
            args=None,
            source=None,
            **kwargs):
    ''''''
    if '__std_out__' in env.sos_dict and '__std_err__' in env.sos_dict:
        raise RuntimeError(
            'Executing nested workflow (action sos_run) in tasks is not supported.'
        )

    if isinstance(workflow, str):
        workflows = [workflow]
    elif isinstance(workflow, Sequence):
        workflows = list(workflow)
    elif workflow is not None:
        raise ValueError(
            'workflow has to be None, a workflow name, or a list of workflow names'
        )

    if source is None:
        script = SoS_Script(env.sos_dict['__step_context__'].content,
                            env.sos_dict['__step_context__'].filename)
        wfs = [script.workflow(wf, use_default=not targets) for wf in workflows]
    else:
        # reading workflow from another file
        script = SoS_Script(filename=source)
        wfs = [script.workflow(wf, use_default=not targets) for wf in workflows]
    # if wf contains the current step or one of the previous one, this constitute
    # recusive nested workflow and should not be allowed
    all_parameters = set()
    for wf in wfs:
        all_parameters |= set(wf.parameters())
        if env.sos_dict['step_name'] in [
                f'{x.name}_{x.index}' for x in wf.sections
        ]:
            raise RuntimeError(
                f'Nested workflow {workflow} contains the current step {env.sos_dict["step_name"]}'
            )

    # args can be specified both as a dictionary or keyword arguments
    if args is None:
        args = kwargs
    else:
        args.update(kwargs)

    for key in args.keys():
        if key not in all_parameters and key not in SOS_ACTION_OPTIONS:
            raise ValueError(
                f'No parameter {key} is defined for workflow {workflow}')

    if shared is None:
        shared = []
    elif isinstance(shared, str):
        shared = [shared]

    # for nested workflow, _input would becomes the input of workflow.
    env.sos_dict.set('__step_output__',
                     copy.deepcopy(env.sos_dict.get('_input', None)))
    shared.append('__step_output__')
    try:
        my_name = env.sos_dict['step_name']
        args_output = ', '.join(f'{x}={short_repr(y)}' for x, y in args.items()
                                if not x.startswith('__'))
        if 'ACTION' in env.config['SOS_DEBUG'] or 'ALL' in env.config['SOS_DEBUG']:
            env.log_to_file(
                'ACTION',
                'Executing workflow ``{}`` with input ``{}`` and {}'.format(
                    workflow,
                    short_repr(env.sos_dict.get('_input', None), True),
                    'no args' if not args_output else args_output))

        if not hasattr(env, '__socket__'):
            raise RuntimeError(
                'sos_run function cannot be executed in scratch cell.')
        # tell the master process to receive a workflow
        # really send the workflow
        shared = {
            x: (env.sos_dict[x] if x in env.sos_dict else None) for x in shared
        }

        wf_ids = [str(uuid.uuid4()) for wf in wfs]

        blocking = not env.sos_dict.get('__concurrent_subworkflow__', False)
        env.__socket__.send_pyobj([
            'workflow', wf_ids, wfs, targets, args, shared, env.config, blocking
        ])

        if not blocking:
            return {'pending_workflows': wf_ids}
        res = {}
        for wf in wfs:
            wf_res = env.__socket__.recv_pyobj()
            res.update(wf_res)
            if wf_res is None:
                sys.exit(0)
            elif isinstance(wf_res, Exception):
                raise wf_res
            else:
                env.sos_dict.quick_update(wf_res['shared'])
        return res
    finally:
        # restore step_name in case the subworkflow re-defines it
        env.sos_dict.set('step_name', my_name)
682089
def train(target, dataset, cluster_spec, ctx):
  """"""
  # Number of workers and parameter servers are infered from the workers and ps
  # hosts string.
  num_workers = len(cluster_spec.as_dict()['worker'])
  num_parameter_servers = len(cluster_spec.as_dict()['ps'])
  # If no value is given, num_replicas_to_aggregate defaults to be the number of
  # workers.
  if FLAGS.num_replicas_to_aggregate == -1:
    num_replicas_to_aggregate = num_workers
  else:
    num_replicas_to_aggregate = FLAGS.num_replicas_to_aggregate

  # Both should be greater than 0 in a distributed training.
  assert num_workers > 0 and num_parameter_servers > 0, (' num_workers and '
                                                         'num_parameter_servers'
                                                         ' must be > 0.')

  # Choose worker 0 as the chief. Note that any worker could be the chief
  # but there should be only one chief.
  is_chief = (FLAGS.task_id == 0)

  # Ops are assigned to worker by default.
  with tf.device('/job:worker/task:%d' % FLAGS.task_id):
    # Variables and its related init/assign ops are assigned to ps.
    with slim.scopes.arg_scope(
          [slim.variables.variable, slim.variables.global_step],
          device=slim.variables.VariableDeviceChooser(num_parameter_servers)):
      # Create a variable to count the number of train() calls. This equals the
      # number of updates applied to the variables.
      global_step = slim.variables.global_step()

      # Calculate the learning rate schedule.
      num_batches_per_epoch = (dataset.num_examples_per_epoch() /
                               FLAGS.batch_size)
      # Decay steps need to be divided by the number of replicas to aggregate.
      decay_steps = int(num_batches_per_epoch * FLAGS.num_epochs_per_decay /
                        num_replicas_to_aggregate)

      # Decay the learning rate exponentially based on the number of steps.
      lr = tf.train.exponential_decay(FLAGS.initial_learning_rate,
                                      global_step,
                                      decay_steps,
                                      FLAGS.learning_rate_decay_factor,
                                      staircase=True)
      # Add a summary to track the learning rate.
      tf.summary.scalar('learning_rate', lr)

      # Create an optimizer that performs gradient descent.
      opt = tf.train.RMSPropOptimizer(lr,
                                      RMSPROP_DECAY,
                                      momentum=RMSPROP_MOMENTUM,
                                      epsilon=RMSPROP_EPSILON)

      if FLAGS.input_mode == 'spark':
        def feed_dict(feed_batch):
          # extract TFRecords, since feed_batch is [(TFRecord, None)]
          tfrecords = []
          for elem in feed_batch:
            tfrecords.append(str(elem[0]))
          return tfrecords

        batch = tf.placeholder(tf.string, [FLAGS.batch_size / FLAGS.num_preprocess_threads])

        # The following is adapted from image_processing.py to remove Readers/QueueRunners.
        # Note: this removes the RandomShuffledQueue, so the incoming data is not shuffled.
        # Presumably, this could be done on the Spark side or done in additional TF code.
        examples = tf.unstack(batch)
        images, labels = [], []
        for example_serialized in examples:
          for thread_id in range(FLAGS.num_preprocess_threads):
            # Parse a serialized Example proto to extract the image and metadata.
            image_buffer, label_index, bbox, _ = image_processing.parse_example_proto(example_serialized)
            image = image_processing.image_preprocessing(image_buffer, bbox, train, thread_id)
            images.append(image)
            labels.append(label_index)
        height = FLAGS.image_size
        width = FLAGS.image_size
        depth = 3
        images = tf.cast(images, tf.float32)
        images = tf.reshape(images, shape=[FLAGS.batch_size, height, width, depth])
        tf.summary.image('images', images)
        labels = tf.reshape(labels, [FLAGS.batch_size])
      else:
        images, labels = image_processing.distorted_inputs(
            dataset,
            batch_size=FLAGS.batch_size,
            num_preprocess_threads=FLAGS.num_preprocess_threads)

      # Number of classes in the Dataset label set plus 1.
      # Label 0 is reserved for an (unused) background class.
      num_classes = dataset.num_classes() + 1
      logits = inception.inference(images, num_classes, for_training=True)
      # Add classification loss.
      inception.loss(logits, labels)

      # Gather all of the losses including regularization losses.
      losses = tf.get_collection(slim.losses.LOSSES_COLLECTION)
      losses += tf.get_collection(tf.GraphKeys.REGULARIZATION_LOSSES)

      total_loss = tf.add_n(losses, name='total_loss')

      if is_chief:
        # Compute the moving average of all individual losses and the
        # total loss.
        loss_averages = tf.train.ExponentialMovingAverage(0.9, name='avg')
        loss_averages_op = loss_averages.apply(losses + [total_loss])

        # Attach a scalar summmary to all individual losses and the total loss;
        # do the same for the averaged version of the losses.
        for l in losses + [total_loss]:
          loss_name = l.op.name
          # Name each loss as '(raw)' and name the moving average version of the
          # loss as the original loss name.
          tf.summary.scalar(loss_name + ' (raw)', l)
          tf.summary.scalar(loss_name, loss_averages.average(l))

        # Add dependency to compute loss_averages.
        with tf.control_dependencies([loss_averages_op]):
          total_loss = tf.identity(total_loss)

      # Track the moving averages of all trainable variables.
      # Note that we maintain a 'double-average' of the BatchNormalization
      # global statistics.
      # This is not needed when the number of replicas are small but important
      # for synchronous distributed training with tens of workers/replicas.
      exp_moving_averager = tf.train.ExponentialMovingAverage(
          inception.MOVING_AVERAGE_DECAY, global_step)

      variables_to_average = (
          tf.trainable_variables() + tf.moving_average_variables())

      # Add histograms for model variables.
      for var in variables_to_average:
        tf.summary.histogram(var.op.name, var)

      # Create synchronous replica optimizer.
      opt = tf.train.SyncReplicasOptimizer(
          opt,
          replicas_to_aggregate=num_replicas_to_aggregate,
          total_num_replicas=num_workers,
          variable_averages=exp_moving_averager,
          variables_to_average=variables_to_average)

      batchnorm_updates = tf.get_collection(slim.ops.UPDATE_OPS_COLLECTION)
      assert batchnorm_updates, 'Batchnorm updates are missing'
      batchnorm_updates_op = tf.group(*batchnorm_updates)
      # Add dependency to compute batchnorm_updates.
      with tf.control_dependencies([batchnorm_updates_op]):
        total_loss = tf.identity(total_loss)

      # Compute gradients with respect to the loss.
      grads = opt.compute_gradients(total_loss)

      # Add histograms for gradients.
      for grad, var in grads:
        if grad is not None:
          tf.summary.histogram(var.op.name + '/gradients', grad)

      apply_gradients_op = opt.apply_gradients(grads, global_step=global_step)

      with tf.control_dependencies([apply_gradients_op]):
        train_op = tf.identity(total_loss, name='train_op')

      # Get chief queue_runners, init_tokens and clean_up_op, which is used to
      # synchronize replicas.
      # More details can be found in sync_replicas_optimizer.
      chief_queue_runners = [opt.get_chief_queue_runner()]
      init_tokens_op = opt.get_init_tokens_op()

      # Create a saver.
      saver = tf.train.Saver()

      # Build the summary operation based on the TF collection of Summaries.
      summary_op = tf.summary.merge_all()

      # Build an initialization operation to run below.
      init_op = tf.global_variables_initializer()

      # We run the summaries in the same thread as the training operations by
      # passing in None for summary_op to avoid a summary_thread being started.
      # Running summaries and training operations in parallel could run out of
      # GPU memory.
      summary_writer = tf.summary.FileWriter("tensorboard_%d" % ctx.worker_num, graph=tf.get_default_graph())
      sv = tf.train.Supervisor(is_chief=is_chief,
                               logdir=FLAGS.train_dir,
                               init_op=init_op,
                               summary_op=None,
                               global_step=global_step,
                               summary_writer=summary_writer,
                               saver=saver,
                               save_model_secs=FLAGS.save_interval_secs)

      tf.logging.info('%s Supervisor' % datetime.now())

      sess_config = tf.ConfigProto(
          allow_soft_placement=True,
          log_device_placement=FLAGS.log_device_placement)

      # Get a session.
      sess = sv.prepare_or_wait_for_session(target, config=sess_config)

      # Start the queue runners.
      queue_runners = tf.get_collection(tf.GraphKeys.QUEUE_RUNNERS)
      sv.start_queue_runners(sess, queue_runners)
      tf.logging.info('Started %d queues for processing input data.',
                      len(queue_runners))

      if is_chief:
        sv.start_queue_runners(sess, chief_queue_runners)
        sess.run(init_tokens_op)

      # Train, checking for Nans. Concurrently run the summary operation at a
      # specified interval. Note that the summary_op and train_op never run
      # simultaneously in order to prevent running out of GPU memory.
      next_summary_time = time.time() + FLAGS.save_summaries_secs
      tf_feed = TFNode.DataFeed(ctx.mgr)
      while not sv.should_stop():
        try:
          start_time = time.time()
          if FLAGS.input_mode == 'spark':
            tmp = feed_dict(tf_feed.next_batch(FLAGS.batch_size / FLAGS.num_preprocess_threads))
            feed = {batch: tmp}
            loss_value, step = sess.run([train_op, global_step], feed_dict=feed)
          else:
            loss_value, step = sess.run([train_op, global_step])
          assert not np.isnan(loss_value), 'Model diverged with loss = NaN'
          if step > FLAGS.max_steps:
            break
          duration = time.time() - start_time

          if step % 30 == 0:
            examples_per_sec = FLAGS.batch_size / float(duration)
            format_str = ('Worker %d: %s: step %d, loss = %.2f'
                          '(%.1f examples/sec; %.3f  sec/batch)')
            tf.logging.info(format_str %
                            (FLAGS.task_id, datetime.now(), step, loss_value,
                             examples_per_sec, duration))

          # Determine if the summary_op should be run on the chief worker.
          if FLAGS.input_mode == 'tf' and is_chief and next_summary_time < time.time():
            tf.logging.info('Running Summary operation on the chief.')
            summary_str = sess.run(summary_op)
            sv.summary_computed(sess, summary_str)
            tf.logging.info('Finished running Summary operation.')

            # Determine the next time for running the summary.
            next_summary_time += FLAGS.save_summaries_secs
        except:
          if is_chief:
            tf.logging.info('About to execute sync_clean_up_op!')
          raise

      # Stop the TFNode data feed
      if FLAGS.input_mode == 'spark':
        tf_feed.terminate()

      # Stop the supervisor.  This also waits for service threads to finish.
      sv.stop()

      # Save after the training ends.
      if is_chief:
        saver.save(sess,
                   os.path.join(FLAGS.train_dir, 'model.ckpt'),
                   global_step=global_step)
851334
def add_jobflow_steps(self, jobflow_id, steps):
        
        if type(steps) != types.ListType:
            steps = [steps]
        params = {}
        params['JobFlowId'] = jobflow_id

        # Step args
        step_args = [self._build_step_args(step) for step in steps]
        params.update(self._build_step_list(step_args))

        return self.get_object(
            'AddJobFlowSteps', params, RunJobFlowResponse, verb='POST')
53973
def retrain(self):
        """"""
        folder = TrainData.from_folder(self.args.folder)
        train_data, test_data = folder.load(True, not self.args.no_validation)

        train_data = TrainData.merge(train_data, self.sampled_data)
        test_data = TrainData.merge(test_data, self.test)
        train_inputs, train_outputs = train_data
        print()
        try:
            self.listener.runner.model.fit(
                train_inputs, train_outputs, self.args.batch_size, self.epoch + self.args.epochs,
                validation_data=test_data, callbacks=self.callbacks, initial_epoch=self.epoch
            )
        finally:
            self.listener.runner.model.save(self.args.model)
1359981
def build(args, parser):
    

    if args is None:
        raise Exception("arguments not provided")

    try:
        json_spec = _parse_executable_spec(args.src_dir, "dxworkflow.json", parser)
        workflow_id = _build_or_update_workflow(json_spec, args)
        _print_output(workflow_id, args)
    except WorkflowBuilderException as e:
        print("Error: %s" % (e.args,), file=sys.stderr)
        sys.exit(3)
155421
def _initialize_workflow(self):
        

        try:

            self._prof.prof('initializing workflow', uid=self._uid)

            for p in self._workflow:
                p._assign_uid(self._sid)

            self._prof.prof('workflow initialized', uid=self._uid)

        except Exception, ex:
            self._logger.exception(
                'Fatal error while initializing workflow: %s' % ex)
            raise
1364958
def write_workflow(self, request, opts, cwd, wftype='cwl'):
        """"""

        workflow_url = request.get("workflow_url")

        # link the cwl and json into the cwd
        if workflow_url.startswith('file://'):
            os.link(workflow_url[7:], os.path.join(cwd, "wes_workflow." + wftype))
            workflow_url = os.path.join(cwd, "wes_workflow." + wftype)
        os.link(self.input_json, os.path.join(cwd, "wes_input.json"))
        self.input_json = os.path.join(cwd, "wes_input.json")

        extra_options = self.sort_toil_options(opts.getoptlist("extra"))
        if wftype == 'cwl':
            command_args = ['toil-cwl-runner'] + extra_options + [workflow_url, self.input_json]
        elif wftype == 'wdl':
            command_args = ['toil-wdl-runner'] + extra_options + [workflow_url, self.input_json]
        elif wftype == 'py':
            command_args = ['python'] + extra_options + [workflow_url]
        else:
            raise RuntimeError('workflow_type is not "cwl", "wdl", or "py": ' + str(wftype))

        return command_args
1349151
def upgrade():
    """"""
    op.create_table(
        'workflows_workflow',
        sa.Column(
            'uuid',
            UUIDType,
            primary_key=True,
            nullable=False,
            default=uuid.uuid4()
        ),
        sa.Column(
            'name',
            sa.String(255),
            default='Default workflow',
            nullable=False
        ),
        sa.Column(
            'created',
            sa.DateTime,
            default=datetime.now,
            nullable=False
        ),
        sa.Column(
            'modified',
            sa.DateTime,
            default=datetime.now,
            onupdate=datetime.now,
            nullable=False
        ),
        sa.Column('id_user', sa.Integer, default=0, nullable=False),
        sa.Column(
            'extra_data',
            JSONType().with_variant(
                postgresql.JSON(none_as_null=True),
                'postgresql',
            ),
            default=lambda: dict(),
            nullable=False
        ),
        sa.Column(
            'status',
            ChoiceType(WorkflowStatus, impl=sa.Integer()),
            default=WorkflowStatus.NEW,
            nullable=False
        )
    )

    op.create_table(
        'workflows_object',
        sa.Column('id', sa.Integer, primary_key=True),
        sa.Column(
            'data',
            JSONType().with_variant(
                postgresql.JSON(none_as_null=True),
                'postgresql',
            ),
            default=lambda: dict(),
            nullable=False
        ),
        sa.Column(
            'extra_data',
            JSONType().with_variant(
                postgresql.JSON(none_as_null=True),
                'postgresql',
            ),
            default=lambda: dict(),
            nullable=False
        ),
        sa.Column(
            'id_workflow',
            UUIDType,
            sa.ForeignKey('workflows_workflow.uuid', ondelete='CASCADE'),
            nullable=True,
            index=True
        ),
        sa.Column(
            'status',
            ChoiceType(ObjectStatus, impl=sa.Integer()),
            default=ObjectStatus.INITIAL,
            nullable=False,
            index=True
        ),
        sa.Column(
            'id_parent',
            sa.Integer,
            sa.ForeignKey('workflows_object.id', ondelete='CASCADE'),
            default=None,
            index=True
        ),
        sa.Column('id_user', sa.Integer, default=0, nullable=False),
        sa.Column(
            'created',
            sa.DateTime,
            default=datetime.now,
            nullable=False
        ),
        sa.Column(
            'modified',
            sa.DateTime,
            default=datetime.now,
            onupdate=datetime.now,
            nullable=False
        ),
        sa.Column(
            'data_type',
            sa.String(150),
            default='',
            nullable=True,
            index=True
        ),
        sa.Column('id_user', sa.Integer, default=0, nullable=False),
        sa.Column(
            'callback_pos',
            JSONType().with_variant(
                postgresql.JSON(none_as_null=True),
                'postgresql',
            ),
            default=lambda: list(),
            nullable=True
        )
    )
364876
def run():
    """"""
    
    # Grab server args
    args = client_helper.grab_server_args()

    # Start up workbench connection
    workbench = zerorpc.Client(timeout=300, heartbeat=60)
    workbench.connect('tcp://'+args['server']+':'+args['port'])

    # Upload the file into workbench
    my_file = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                           '../data/pe/bad/033d91aae8ad29ed9fbb858179271232')
    with open(my_file,'rb') as f:

        # Throw file into workbench
        filename = os.path.basename(my_file)
        raw_bytes = f.read()
        md5 = workbench.store_sample(raw_bytes, filename, 'exe')
        results = workbench.work_request('view', md5)
        print 'Filename: %s' % filename
        pprint.pprint(results)
427974
def training_loop(self):
    """"""
    if not self.restarting:
      self._write_counters(self._local_step_at_start, self._global_step)

    tf.logging.info(
        "Training %s up to %d, %d to go", self.model_mode,
        self.target_local_step, self.steps_to_go
    )

    yield

    self._write_counters(self.target_local_step, -1)
672725
function WorkflowProcessStepsController($scope, $state, wfmService, $timeout, $stateParams) {
  var self = this;
  var workorderId = $stateParams.workorderId;

  function updateWorkflowState(workorder) {
    //If the workflow is complete, then we can switch to the summary view.
    if (wfmService.isCompleted(workorder)) {
      return $state.go('app.workflowProcess.complete', {
        workorderId: workorder.id
      });
    }

    if (!wfmService.isOnStep(workorder)) {
      return $state.go('app.workflowProcess.begin', {
        workorderId: workorder.id
      });
    }

    $timeout(function() {
      self.workorder = workorder;
      self.workflow = workorder.workflow;
      self.result = workorder.results;
      self.stepIndex = wfmService.getCurrentStepIdx(workorder);
      self.stepCurrent = wfmService.getCurrentStep(workorder);
    });
  }

  self.back = function() {
    wfmService.previousStep(workorderId)
      .then(updateWorkflowState)
      .catch(console.error);
  };

  self.triggerCompleteStep = function(submission) {
    wfmService.completeStep(workorderId, submission)
      .then(updateWorkflowState)
      .catch(console.error);
  };

  self.triggerBackStep = function() {
    self.back();
  };

  wfmService.readWorkOrder(workorderId).then(updateWorkflowState);
}
435837
def start_workflow(flow)
      @http.post(
        instance_worker_url,
        ATTR_PROG => PROG,
        ATTR_CANONICAL => canonical_name(flow),
        ATTR_NAME => class_name(flow),
        ATTR_DATA => @serializer.encode(@properties.from(flow)),
        ATTR_ID => parse_custom_id_from(flow)
      )
    end
525795
def add_workflow(self, workflow, commit=False):
        
        if workflow.workflow_id in self.workflows:
            raise KeyError("Workflow with id {} already exists".format(workflow.workflow_id))

        self.workflows[workflow.workflow_id] = workflow
        logging.info("Added workflow {} to workflow manager".format(workflow.workflow_id))

        # Optionally also save the workflow to database
        if commit:
            self.commit_workflow(workflow.workflow_id)
        else:
            self.uncommitted_workflows.add(workflow.workflow_id)