How to use the cellprofiler.pipeline function in CellProfiler

def make_workspace(pixel_data, mask=None):
    image = cellprofiler.image.Image(pixel_data, mask)
    image_set_list = cellprofiler.image.ImageSetList()

    image_set = image_set_list.get_image_set(0)
    image_set.add(IMAGE_NAME, image)

    module = cellprofiler.modules.identifydeadworms.IdentifyDeadWorms()
    module.set_module_num(1)
    module.image_name.value = IMAGE_NAME
    module.object_name.value = OBJECTS_NAME

    pipeline = cellprofiler.pipeline.Pipeline()

    def callback(caller, event):
        assert not isinstance(event, cellprofiler.pipeline.LoadExceptionEvent)
        assert not isinstance(event, cellprofiler.pipeline.RunExceptionEvent)

    pipeline.add_listener(callback)
    pipeline.add_module(module)

    workspace = cellprofiler.workspace.Workspace(
        pipeline,
        module,
        image_set,
        cellprofiler.object.ObjectSet(),
        cellprofiler.measurement.Measurements(),
        image_set_list,
    )

def check(module, url, dd, keys=None, xml=None):
    """Check that running the metadata module on a url generates the expected dictionary"""
    pipeline = cellprofiler.pipeline.Pipeline()
    imgs = cellprofiler.modules.images.Images()
    imgs.filter_choice.value = cellprofiler.modules.images.FILTER_CHOICE_NONE
    imgs.module_num = 1
    pipeline.add_module(imgs)
    module.set_module_num(2)
    pipeline.add_module(module)
    pipeline.add_urls([url])
    m = cellprofiler.measurement.Measurements()
    workspace = cellprofiler.workspace.Workspace(pipeline, module, None, None, m, None)
    file_list = workspace.file_list
    file_list.add_files_to_filelist([url])
    if xml is not None:
        file_list.add_metadata(url, xml)
    ipds = pipeline.get_image_plane_details(workspace)
    assert len(ipds) == len(dd)
    for d, ipd in zip(dd, ipds):

def make_workspace(labels, mode, distance=0, neighbors_labels=None):
    """Make a workspace for testing MeasureObjectNeighbors"""
    module = cellprofiler.modules.measureobjectneighbors.MeasureObjectNeighbors()
    module.set_module_num(1)
    module.object_name.value = OBJECTS_NAME
    module.distance_method.value = mode
    module.distance.value = distance
    pipeline = cellprofiler.pipeline.Pipeline()
    pipeline.add_module(module)
    object_set = cellprofiler.object.ObjectSet()
    image_set_list = cellprofiler.image.ImageSetList()
    image_set = image_set_list.get_image_set(0)
    measurements = cellprofiler.measurement.Measurements()
    measurements.group_index = 1
    measurements.group_number = 1
    workspace = cellprofiler.workspace.Workspace(
        pipeline, module, image_set, object_set, measurements, image_set_list
    )
    objects = cellprofiler.object.Objects()
    objects.segmented = labels
    object_set.add_objects(objects, OBJECTS_NAME)
    if neighbors_labels is None:
        module.neighbors_name.value = OBJECTS_NAME
    else:

def test_load_v1():
    with open("./tests/resources/modules/unmixcolors/v1.pipeline", "r") as fd:
        data = fd.read()

    pipeline = cellprofiler.pipeline.Pipeline()

    def callback(caller, event):
        assert not isinstance(event, cellprofiler.pipeline.LoadExceptionEvent)

    pipeline.add_listener(callback)
    pipeline.load(six.moves.StringIO(data))
    assert len(pipeline.modules()) == 1
    module = pipeline.modules()[0]
    assert isinstance(module, cellprofiler.modules.unmixcolors.UnmixColors)
    assert module.input_image_name == "Color"
    assert module.stain_count.value == 13
    assert module.outputs[0].image_name == "Hematoxylin"
    assert module.outputs[-1].image_name == "RedWine"
    for i, stain in enumerate(
        (
            cellprofiler.modules.unmixcolors.CHOICE_HEMATOXYLIN,

object_set = cellprofiler.object.ObjectSet()
    image = cellprofiler.image.Image(pixel_data, dimensions=dimensions)
    if not mask is None:
        image.mask = mask
    image_set.add(IMAGES_NAME, image)
    if not objects is None:
        o = cellprofiler.object.Objects()
        o.segmented = objects
        object_set.add_objects(o, OBJECTS_NAME)
    module = cellprofiler.modules.measureimagequality.MeasureImageQuality()
    module.images_choice.value = cellprofiler.modules.measureimagequality.O_SELECT
    module.image_groups[0].include_image_scalings.value = False
    module.image_groups[0].image_names.value = IMAGES_NAME
    module.image_groups[0].use_all_threshold_methods.value = False
    module.set_module_num(1)
    pipeline = cellprofiler.pipeline.Pipeline()
    pipeline.add_module(module)
    workspace = cellprofiler.workspace.Workspace(
        pipeline,
        module,
        image_set,
        object_set,
        cellprofiler.measurement.Measurements(),
        image_set_list,
    )
    return workspace

importPackage(Packages.org.cellprofiler.imageset);
            importPackage(Packages.org.cellprofiler.imageset.filter);
            var imageFile=new ImageFile(new java.net.URI(url));
            var imageFileDetails = new ImageFileDetails(imageFile);
            var imageSeries=new ImageSeries(imageFile, series);
            var imageSeriesDetails = new ImageSeriesDetails(imageSeries, imageFileDetails);
            var imagePlane=new ImagePlane(imageSeries, index, channel);
            var ipd = new ImagePlaneDetails(imagePlane, imageSeriesDetails);
            for (var entry in Iterator(metadata.entrySet())) {
                ipd.put(entry.getKey(), entry.getValue());
            }
            ipd;
            """,
        dict(url=url, metadata=jmetadata, series=series, index=index, channel=channel),
    )
    return cellprofiler.pipeline.ImagePlaneDetails(jipd)

def callback(caller, event):
            self.assertFalse(
                isinstance(event, cellprofiler.pipeline.LoadExceptionEvent)
            )

def make_workspace(mdict, controls_measurement, dose_measurements=[]):
    """Make a workspace and module for running CalculateStatistics

    mdict - a two-level dictionary that mimics the measurements structure
            for instance:
            mdict = { cpmeas.Image: { "M1": [ 1,2,3] }}
            for the measurement M1 with values for 3 image sets
    controls_measurement - the name of the controls measurement
    """
    module = cellprofiler.modules.calculatestatistics.CalculateStatistics()
    module.set_module_num(1)
    module.grouping_values.value = controls_measurement

    pipeline = cellprofiler.pipeline.Pipeline()
    pipeline.add_module(module)

    m = cellprofiler.measurement.Measurements()
    nimages = None
    for object_name in list(mdict.keys()):
        odict = mdict[object_name]
        for feature in list(odict.keys()):
            m.add_all_measurements(object_name, feature, odict[feature])
            if nimages is None:
                nimages = len(odict[feature])
            else:
                assert nimages == len(odict[feature])
            if (
                object_name == cellprofiler.measurement.IMAGE
                and feature in dose_measurements
            ):

def enter_batch_mode(self, workspace):
        """Restore the image set list from its setting as we go into batch mode"""
        pipeline = workspace.pipeline
        assert isinstance(pipeline, cpp.Pipeline)
        assert not self.distributed_mode, "Distributed mode no longer supported"
        default_output_directory = self.custom_output_directory.value
        default_image_directory = self.default_image_directory.value
        if os.path.isdir(default_output_directory):
            cpprefs.set_default_output_directory(default_output_directory)
        else:
            logger.info(
                'Batch file default output directory, "%s", does not exist'
                % default_output_directory
            )
        if os.path.isdir(default_image_directory):
            cpprefs.set_default_image_directory(default_image_directory)
        else:
            logger.info(
                'Batch file default input directory "%s", does not exist'
                % default_image_directory

'email':lookup('email', 'user@broadinstitute.org'),
         'queue':lookup('queue', 'broad'),
         'project':lookup('project','imaging'),
         'write_data':lookup('write_data','no'),
         'batch_size':lookup('batch_size','10'),
         'memory_limit':lookup('memory_limit','2000'),
         'timeout':lookup('timeout','30'),
         'revision':lookup('revision','8009'),
         'url':myself
         }

batch_file = os.path.join(keys['data_dir'], F_BATCH_DATA)
grouping_keys = None
error_message = None
if os.path.exists(batch_file):
    pipeline = cpp.Pipeline()
    print "<span style="visibility:hidden">"
    try:
        had_problem = [False]
        def error_callback(event, caller):
            if (isinstance(event, cpp.LoadExceptionEvent) or
                isinstance(event, cpp.RunExceptionEvent)):
                sys.stderr.write("Handling exception: %s\n"%str(event))
                sys.stderr.write(traceback.format_exc())
                had_problem = [True]
        pipeline.add_listener(error_callback)
        pipeline.load(batch_file)
        if had_problem[0]:
            raise RuntimeError("Failed to load batch file")
        image_set_list = pipeline.prepare_run(None)
        if had_problem[0]:
            raise RuntimeError("Failed to prepare batch file")</span>