Ticket #2893: wells44.py

import omero
from omero_sys_ParametersI import ParametersI

import time
import sys
import itertools
import logging

LOGFORMAT =  """%(asctime)s %(levelname)-5.5s [%(name)40s] (%(threadName)-10s) %(message)s"""
logging.basicConfig(level=10, format=LOGFORMAT)

class Test(object):

    def __init__(self, user, password, host="localhost"):
        self.client = omero.client(host)
        self.session = self.client.createSession(user, password)
        self.update = self.session.getUpdateService()
        self.query = self.session.getQueryService()

    def create_screen(self):
        s = omero.model.ScreenI()
        s.name = omero.rtypes.rstring("test-screen")
        s = self.update.saveAndReturnObject(s)
        return s.id

    def create_acquisition(self, plate_id):
        ac = omero.model.PlateAcquisitionI()
        ac.startTime = omero.rtypes.rtime(time.time() * 1e3)
        ac.endTime = omero.rtypes.rtime(time.time() * 1e3)
        ac.plate = omero.model.PlateI(plate_id, False)
        ac = self.update.saveAndReturnObject(ac)
        return ac.id

    def create_plate(self, rows=1, cols=1):
        p = omero.model.PlateI()
        p.name = omero.rtypes.rstring("test-plate")
        for r in range(rows):
            for c in range(cols):
                w = omero.model.WellI()
                w.plate = p
                w.column = omero.rtypes.rint(c)
                w.row = omero.rtypes.rint(r)
                p.addWell(w)

        p = self.update.saveAndReturnObject(p)
        return p.id

    def get_well(self, plate_id, row, col):
        query = '''
        select w from Well w
        left outer join fetch w.wellSamples
        where w.row = :row and
        w.column = :col and
        w.plate.id = :pid'''
        params = ParametersI()
        params.add('pid', plate_id)
        params.add('row', omero.rtypes.rint(row))
        params.add('col', omero.rtypes.rint(col))
        return self.query.findByQuery(query, params)

    def create_image(self):
        i = omero.model.ImageI()
        i.name = omero.rtypes.rstring("test-image")
        i.acquisitionDate = omero.rtypes.rtime(time.time()*1e3)
        i = self.update.saveAndReturnObject(i)
        return i.id

    def create_well_sample(self, plate_id, acquisition_id, row, col):
        w = self.get_well(plate_id, row, col)
        image_id = self.create_image()

        ws = omero.model.WellSampleI()
        ws.well = omero.model.WellI(w.id, False)
        ws.image = omero.model.ImageI(image_id, False)
        ws.plateAcquisition = omero.model.PlateAcquisitionI(acquisition_id, False)
        w.addWellSample(ws)
        w = self.update.saveAndReturnObject(w)

        ws = w.getWellSample(w.sizeOfWellSamples()-1)
        ws = omero.model.WellSampleI(ws.id.val, False)

        return ws.id


def main(rows, cols, samples):
    logging.info("login")
    test = Test(*sys.argv[1:])
    logging.info("create screen")
    screen_id = test.create_screen()
    logging.info("create plate")
    plate_id = test.create_plate(rows, cols)
    logging.info("create acquisition")
    acquisition_id = test.create_acquisition(plate_id)
    for r, c, s in itertools.product(range(rows),
                                     range(cols),
                                     range(samples)):
        logging.info("creating sample-%i for well %i,%i" % (s, r, c))
        test.create_well_sample(plate_id, acquisition_id, r, c)


if __name__ == "__main__":
    # create 1x1 plate with 2 samples
    import threading
    import random

    stop=[False]
    class T(threading.Thread):
        def run(self):
            while not stop[0]:
                cols=random.choice([2,3,4])
                rows=random.choice([2,3,4])
                samples=random.choice([2,4,8])
                main(2,2,8)


threads = [T() for x in range(8)]
for t in threads: t.start()

try:
    sys.stdin.read()
except KeyboardInterrupt:
    pass
stop[0] = True

for t in threads: t.join()