Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Parallelize the set command #33

Closed
wants to merge 4 commits into from
Closed

Parallelize the set command #33

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
111421c
Parallelize the set command
dominikl Nov 21, 2019
6f1ca9d
Move batch default value to add_argument call
dominikl Nov 22, 2019
c5cd1e0
Fix flake8
dominikl Nov 22, 2019
602199c
Some improvements and refactoring
dominikl Nov 26, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
170 changes: 102 additions & 68 deletions src/omero_cli_render.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
from builtins import str
from builtins import range
from builtins import object
import concurrent.futures
import sys
import time
import json
Expand Down Expand Up @@ -396,6 +397,11 @@ def _configure(self, parser):
help="Local file or OriginalFile:ID which specifies the "
"rendering settings")

set_cmd.add_argument(
"--batch", type=int, default=1,
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Have you considered a higher default?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could do. Just didn't want to change the default behaviour if the new option is omitted.

help="Batch size to process simultaneously"
)

test.add_argument(
"--force", action="store_true",
help="Force creation of pixel data file in binary "
Expand All @@ -414,7 +420,7 @@ def _lookup(self, gateway, type, oid):
self.ctx.die(110, "No such %s: %s" % (type, oid))
return obj

def render_images(self, gateway, object, batch=100):
def load_images(self, gateway, object, batch=100):
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you explain your thoughts on this rename?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Because the method doesn't render any images, it just loads the images. I found the name render_images totally confusing every time I looked into the omero_cli_render code.

"""
Get the images.

Expand All @@ -429,12 +435,12 @@ def render_images(self, gateway, object, batch=100):

if isinstance(object, list):
for x in object:
for rv in self.render_images(gateway, x, batch):
for rv in self.load_images(gateway, x, batch):
yield rv
elif isinstance(object, Screen):
scr = self._lookup(gateway, "Screen", object.id)
for plate in scr.listChildren():
for rv in self.render_images(gateway, plate._obj, batch):
for rv in self.load_images(gateway, plate._obj, batch):
yield rv
elif isinstance(object, Plate):
plt = self._lookup(gateway, "Plate", object.id)
Expand All @@ -455,7 +461,7 @@ def render_images(self, gateway, object, batch=100):
elif isinstance(object, Project):
prj = self._lookup(gateway, "Project", object.id)
for ds in prj.listChildren():
for rv in self.render_images(gateway, ds._obj, batch):
for rv in self.load_images(gateway, ds._obj, batch):
yield rv

elif isinstance(object, Dataset):
Expand Down Expand Up @@ -485,7 +491,7 @@ def render_images(self, gateway, object, batch=100):
def info(self, args):
""" Implements the 'info' command """
first = True
for img in self.render_images(self.gateway, args.object, batch=1):
for img in self.load_images(self.gateway, args.object, batch=1):
ro = RenderObject(img)
if args.style == 'plain':
self.ctx.out(ro)
Expand All @@ -505,8 +511,8 @@ def info(self, args):
@gateway_required
def copy(self, args):
""" Implements the 'copy' command """
for src_img in self.render_images(self.gateway, args.object, batch=1):
for targets in self.render_images(self.gateway, args.target):
for src_img in self.load_images(self.gateway, args.object, batch=1):
for targets in self.load_images(self.gateway, args.target):
batch = dict()
for target in targets:
if target.id == src_img.id:
Expand Down Expand Up @@ -549,13 +555,9 @@ def _generate_thumbs(self, images):
self.ctx.dbg("Image:%s got thumbnail in %2.2fs" % (
img.id, stop - start))

def _read_default_planes(self, img, data, ignore_errors=False):
def _read_default_planes(self, img, def_z=None, def_t=None, ignore_errors=False):
"""Read and validate the default planes"""

# Read values from dictionary
def_z = data['z'] if 'z' in data else None
def_t = data['t'] if 't' in data else None

# Minimal validation: default planes should be 1-indexed integers
if (def_z is not None) and (def_z < 1 or int(def_z) != def_z):
self.ctx.die(
Expand Down Expand Up @@ -583,23 +585,78 @@ def _read_default_planes(self, img, data, ignore_errors=False):
def_t = None
return (def_z, def_t)

def _set_rend(self, def_z, def_t, img, cindices, greyscale, rangelist,
colorlist, disable, skipthumbs):
"""
Set the rendering settings for one image
:param def_z: The default z plane (int)
:param def_t: The default t plane (int)
:param img: The image (ImageWrapper)
:param cindices: The channel indices (int list)
:param greyscale: If the image is grey scale (boolean)
:param rangelist: The range (min, max) (list of int lists)
:param colorlist: The colors (list of hex strings)
:param disable: Flag if other channels should be disabled (boolean)
:param skipthumbs: Flag if the thumbnail generation should be skipped
(boolean)
:return: The image id (long)
"""
reactivatechannels = []
if not disable:
# Calling set_active_channels will disable channels which
# are not specified, have to keep track of them and
# re-activate them later again
imgchannels = img.getChannels()
for ci, ch in enumerate(imgchannels, 1):
if ci not in cindices and -ci not in cindices \
and ch.isActive():
reactivatechannels.append(ci)

img.set_active_channels(
cindices, windows=rangelist, colors=colorlist)
if greyscale is not None:
if greyscale:
img.setGreyscaleRenderingModel()
else:
img.setColorRenderingModel()

if len(reactivatechannels) > 0:
img.set_active_channels(reactivatechannels)

if def_z:
img.setDefaultZ(def_z - 1)
if def_t:
img.setDefaultT(def_t - 1)

try:
img.saveDefaults()
self.ctx.dbg(
"Updated rendering settings for Image:%s" % img.id)
if not skipthumbs:
img.getThumbnail(size=(96,), direct=False, use_cached_ts=False)
except Exception as e:
self.ctx.err('ERROR: %s' % e)
finally:
img._closeRE()
return img.id

@gateway_required
def set(self, args):
""" Implements the 'set' command """
newchannels = {}
data = pydict_text_io.load(
settings = pydict_text_io.load(
args.channels, session=self.client.getSession())
if 'channels' not in data:
if 'channels' not in settings:
self.ctx.die(104, "ERROR: No channels found in %s" % args.channels)

version = _getversion(data)
version = _getversion(settings)
if version == 0:
self.ctx.die(124, "ERROR: Cannot determine version. Specify"
" version or use either start/end or min/max"
" (not both).")

# Read channel setttings from rendering dictionary
for chindex, chdict in data['channels'].items():
for chindex, chdict in settings['channels'].items():
try:
cindex = int(chindex)
except Exception as e:
Expand All @@ -617,15 +674,19 @@ def set(self, args):
105, "Invalid channel description: %s" % chdict)

try:
greyscale = data['greyscale']
print('greyscale=%s' % data['greyscale'])
greyscale = settings['greyscale']
print('greyscale=%s' % settings['greyscale'])
except KeyError:
greyscale = None

# Read values from dictionary
def_z = settings['z'] if 'z' in settings else None
def_t = settings['t'] if 't' in settings else None

namedict = {}
cindices = []
rangelist = []
colourlist = []
colorlist = []
for (i, c) in newchannels.items():
if c.label:
namedict[i] = c.label
Expand All @@ -634,56 +695,29 @@ def set(self, args):
else:
cindices.append(i)
rangelist.append([c.start, c.end])
colourlist.append(c.color)
colorlist.append(c.color)

iids = []
for img in self.render_images(self.gateway, args.object, batch=1):
iids.append(img.id)

(def_z, def_t) = self._read_default_planes(
img, data, ignore_errors=args.ignore_errors)

reactivatechannels = []
if not args.disable:
# Calling set_active_channels will disable channels which
# are not specified, have to keep track of them and
# re-activate them later again
imgchannels = img.getChannels()
for ci, ch in enumerate(imgchannels, 1):
if ci not in cindices and -ci not in cindices\
and ch.isActive():
reactivatechannels.append(ci)

img.set_active_channels(
cindices, windows=rangelist, colors=colourlist)
if greyscale is not None:
if greyscale:
img.setGreyscaleRenderingModel()
else:
img.setColorRenderingModel()

if len(reactivatechannels) > 0:
img.set_active_channels(reactivatechannels)

if def_z:
img.setDefaultZ(def_z - 1)
if def_t:
img.setDefaultT(def_t - 1)

try:
img.saveDefaults()
self.ctx.dbg(
"Updated rendering settings for Image:%s" % img.id)
if not args.skipthumbs:
self._generate_thumbs([img])
except Exception as e:
self.ctx.err('ERROR: %s' % e)
finally:
img._closeRE()

if not iids:
self.ctx.die(113, "ERROR: No images found for %s %d" %
(args.object.__class__.__name__, args.object.id._val))
n_images = 0
for img_batch in self.load_images(self.gateway, args.object,
batch=args.batch):
n_images += len(img_batch)
with concurrent.futures.ThreadPoolExecutor() as executor:
future_image = {executor.submit(self._set_rend, def_z,
def_t, img, cindices, greyscale,
rangelist, colorlist,
args.disable, args.skipthumbs):
img for img in img_batch}
for future in concurrent.futures.as_completed(future_image):
img = future_image[future]
try:
iid = future.result()
iids.append(iid)
except Exception as exc:
self.ctx.err('ERROR: Image %d: %s' % (img.id, exc))

if len(iids) < n_images:
self.ctx.die(113, "Some errors occurred.")

if namedict:
self._update_channel_names(self.gateway, iids, namedict)
Expand All @@ -695,7 +729,7 @@ def edit(self, args):
def test(self, args):
""" Implements the 'test' command """
self.gateway.SERVICE_OPTS.setOmeroGroup('-1')
for img in self.render_images(self.gateway, args.object, batch=1):
for img in self.load_images(self.gateway, args.object, batch=1):
self.test_per_pixel(
self.client, img.getPrimaryPixels().id, args.force, args.thumb)

Expand Down