An OMERO.script that imports an image via the CLI

test_import_cli.py

import sys
import subprocess
from path import path
import omero.scripts as scripts
from omero.gateway import BlitzGateway
import omero.util.script_utils as script_utils
import omero

from omero.rtypes import rint, rlong, rstring, robject, unwrap


def import_image(conn, scriptParams):

    dataset = conn.getObject("Dataset", scriptParams["IDs"][0])

    # Create test image to import
    image = list(dataset.listChildren())[0]
    pil_image = image.renderImage(0, 0)
    image_name = "test.png"
    pil_image.save(image_name)

    # import via cli
    path_to_bin_omero = "/Users/wmoore/Desktop/OMERO/openmicroscopy/dist/bin"
    client = conn.c
    # server = client.getProperty("omero.host")
    server = "localhost"
    port = client.getProperty("omero.port")
    key = client.getSessionId()

    args = [sys.executable]
    args.append(path(path_to_bin_omero) / "omero")
    args.extend(["-s", server, "-k", key, "-p", port, "import"])
    # Import into current Dataset
    args.extend(["-d", str(dataset.id)])
    args.append(image_name)

    popen = subprocess.Popen(args,
                             stdout=subprocess.PIPE,
                             stderr=subprocess.PIPE)
    out, err = popen.communicate()
    print 'out', out
    print 'err', err
    rc = popen.wait()
    if rc != 0:
        raise Exception("import failed: [%r] %s\n%s" % (args, rc, err))
    image_ids = []
    for x in out.split("\n"):
        if "Image:" in x:
            image_ids.append(long(x.replace('Image:', '')))
    return image_ids


def runAsScript():
    """
    The main entry point of the script, as called by the client via the
    scripting service, passing the required parameters.
    """

    dataTypes = [rstring('Dataset')]

    client = scripts.client(
        'Import_CLI_Test.py',
        """Testing import via CLI from script""",

        scripts.String(
            "Data_Type", optional=False, grouping="1",
            description="Choose source of images (only Dataset supported)",
            values=dataTypes, default="Dataset"),

        scripts.List(
            "IDs", optional=False, grouping="2",
            description="Dataset ID").ofType(rlong(0)),
    )

    try:
        scriptParams = client.getInputs(unwrap=True)
        print scriptParams

        # wrap client to use the Blitz Gateway
        conn = BlitzGateway(client_obj=client)

        # convert Dataset(s) to Plate. Returns new plate if created
        image_ids = import_image(conn, scriptParams)
        if len(image_ids) > 0:
            message = "Imported %s images" % len(image_ids)
            # return link to first image
            newObj = omero.model.ImageI(image_ids[0], False)
            client.setOutput("New_Object", robject(newObj))
        else:
            message = "No images imported"

        client.setOutput("Message", rstring(message))
            
    finally:
        client.closeSession()

if __name__ == "__main__":
    runAsScript()