How to use the luigi.build function in luigi

def _test_mode(self, mode, threshold=.5):
        from cluster_tools.thresholded_components import ThresholdedComponentsWorkflow
        task = ThresholdedComponentsWorkflow(tmp_folder=self.tmp_folder,
                                             config_dir=self.config_folder,
                                             target=self.target, max_jobs=self.max_jobs,
                                             input_path=self.input_path,
                                             input_key=self.input_key,
                                             output_path=self.output_path,
                                             output_key=self.output_key,
                                             assignment_key=self.assignment_key,
                                             threshold=threshold, threshold_mode=mode)
        ret = luigi.build([task], local_scheduler=True)
        self.assertTrue(ret)
        self._check_result(mode, threshold=threshold)

i = luigi.Parameter()

    def run(self):  # Use work instead of run to DEBUG
        logger.info('Running test job...')
        with open(self.output().path, 'w') as f:
            f.write('This is a test job.')
            f.close()

    def output(self):
        return luigi.LocalTarget(path=os.path.join(os.getcwd(), 'testjob_' + str(self.i) + '.txt'))


if __name__ == '__main__':
    tasks = [MyJobTask(i=str(i), select=i+1) for i in range(3)]
    luigi.build(tasks, local_scheduler=True, workers=3)

input_path=self.input_path,
                                     input_key=self.input_key,
                                     output_path=self.output_path,
                                     output_key=self.output_key,
                                     threshold=.5,
                                     dependency=DummyTask())
        offset_path = './tmp/offsets.json'
        with z5py.File(self.input_path) as f:
            shape = f[self.input_key].shape
        task = MergeOffsetsLocal(tmp_folder=self.tmp_folder,
                                 config_dir=self.config_folder,
                                 max_jobs=8,
                                 shape=shape,
                                 save_path=offset_path,
                                 dependency=task1)
        ret = luigi.build([task], local_scheduler=True)
        self.assertTrue(ret)
        self.assertTrue(os.path.exists(offset_path))

        # checks
        # load offsets from file
        with open(offset_path) as f:
            offsets_dict = json.load(f)
            offsets = offsets_dict['offsets']
            max_offset = int(offsets_dict['n_labels']) - 1

        # load output segmentation
        with z5py.File(self.output_path) as f:
            seg = f[self.output_key][:]

        blocking = nt.blocking([0, 0, 0], list(shape), self.block_shape)
        for block_id in range(blocking.numberOfBlocks):

def test_start_handler(self):
        saved_tasks = []

        @EmptyTask.event_handler(Event.START)
        def save_task(task):
            print("Saving task...")
            saved_tasks.append(task)

        t = EmptyTask(True)
        build([t], local_scheduler=True)
        self.assertEqual(saved_tasks, [t])

def test_region_features(self):
        from cluster_tools.features import RegionFeaturesWorkflow
        ret = luigi.build([RegionFeaturesWorkflow(input_path=self.input_path,
                                                  input_key=self.input_key,
                                                  labels_path=self.input_path,
                                                  labels_key=self.seg_key,
                                                  output_path=self.output_path,
                                                  output_key=self.output_key,
                                                  config_dir=self.config_folder,
                                                  tmp_folder=self.tmp_folder,
                                                  target=self.target,
                                                  max_jobs=self.max_jobs)],
                          local_scheduler=True)
        self.assertTrue(ret)
        feature_names = self.check_subresults()
        self.check_result(feature_names)

def region_feats(path, ws_key, pmap_key, out_path,
                 tmp_folder, prefix, target, n_jobs):

    feat_key = f'region_features/{prefix}'
    task = RegionFeaturesWorkflow
    t = task(max_jobs=n_jobs, target=target,
             tmp_folder=tmp_folder, config_dir=os.path.join(tmp_folder, 'configs'),
             input_path=path, input_key=pmap_key,
             labels_path=path, labels_key=ws_key,
             output_path=out_path, output_key=feat_key,
             prefix=prefix)
    ret = luigi.build([t], local_scheduler=True)
    assert ret

luigi.interface.setup_interface_logging.has_run = True  # turn off Luigi's default logging setup
    log_cfg.get_logger('luigi-interface', 20)  # just calling log_cfg.get_logger registers the luigi-interface

    dataset.resolve_pipeline_and_toolset()
    dataset.start()
    final_stage = dataset.pipeline.build_pipeline(dataset)

    luigi_params = {
        'tasks': [final_stage],
        'local_scheduler': cfg.query('luigi', 'local_scheduler'),
        'workers': cfg.query('luigi', 'max_parallel_jobs', ret_default=4)
    }
    if luigi_params['local_scheduler'] is not True:
        luigi_params['scheduler_url'] = cfg['luigi']['scheduler_url']

    success = luigi.build(**luigi_params)

    # if any exception occurred during the pipeline raise them here again
    dataset.raise_exceptions()

    return 0 if success is True else 9

config = SkeletonWorkflow.get_config()
    global_config = config['global']
    shebang = '#! /g/kreshuk/pape/Work/software/conda/miniconda3/envs/cluster-new/bin/python'
    global_config.update({'shebang': shebang})
    with open(os.path.join(config_dir, 'global.config'), 'w') as f:
        json.dump(global_config, f)

    resolution = [40, 4, 4]
    size_threshold = 2500

    task = SkeletonWorkflow(tmp_folder=tmp_folder, config_dir=config_dir,
                            max_jobs=max_jobs, target=target,
                            input_path=path, input_key=input_key,
                            output_path=path, output_key=output_key,
                            resolution=resolution, size_threshold=size_threshold)
    success = luigi.build([task], local_scheduler=True)
    assert success

def pipeline(dataset):
    analysis_driver.pipeline._dataset = dataset
    final_stage = Stage4()
    luigi.build(
        [final_stage],
        local_scheduler=True
    )
    return final_stage.exit_status

def main(nbar_path, outpath, nnodes=1, nodenum=1):
    nbar_files = sorted([pjoin(nbar_path, f) for f in os.listdir(nbar_path) if
                        '_NBAR_' in f])
    nbar_files = [basename(f) for f in scatter(l1t_files, nnodes, nodenum)]
    ncpus = int(os.getenv('PBS_NCPUS', '1'))
    tasks = []
    for nbar_file in nbar_files:
        nbar_dataset_path = pjoin(nbar_path, nbar_file)
        fc_dataset_path = os.path.join(outpath, fc_name_from_nbar(nbar_file))
        tasks.append(FractionalCoverTask(nbar_dataset_path,
                                         fc_dataset_path))

    luigi.build(tasks, local_scheduler=True, workers=16)