Source code for

Utility functions and classes
import itertools
import logging
import operator
import typing as t
from functools import wraps
from pathlib import Path
from time import perf_counter
from typing import Callable

import cv2

[docs]def repr_obj(obj, indent=0): """ Helper function to display info about OMERO objects. Not all objects will have a "name" or owner field. """ string = """%s%s:%s Name:"%s" (owner=%s)""" % ( " " * indent, obj.OMERO_CLASS, obj.getId(), obj.getName(), obj.getAnnotation(), ) return string
[docs]def imread(path): return cv2.imread(str(path), -1)
[docs]class Cache: """ Fixed-length mapping to use as a cache. Deletes items in FIFO manner when maximum allowed length is reached. """
[docs] def __init__(self, max_len=5000, load_fn: Callable = imread): """ :param max_len: Maximum number of items in the cache. :param load_fn: The function used to load new items if they are not available in the Cache """ self._dict = dict() self._queue = [] self.load_fn = load_fn self.max_len = max_len
def __getitem__(self, item): if item not in self._dict: self.load_item(item) return self._dict[item] def load_item(self, item): self._dict[item] = self.load_fn(item) # Clean up the queue self._queue.append(item) if len(self._queue) > self.max_len: del self._dict[self._queue.pop(0)] def clear(self): self._dict.clear() self._queue.clear()
[docs]def accumulate(list_: list) -> t.Generator: """Accumulate list based on the first value""" list_ = sorted(list_) it = itertools.groupby(list_, operator.itemgetter(0)) for key, sub_iter in it: yield key, [x[1] for x in sub_iter]
[docs]def get_store_path(save_dir, store, name): """Create a path to a position-specific store. This combines the name and the store's base name into a file path within save_dir. For example. >>> get_store_path('data', 'baby_seg.h5', 'pos001') Path(data/pos001baby_seg.h5') :param save_dir: The root directory in which to save the file, absolute path. :param store: The base name of the store :param name: The name of the position :return: Path(save_dir) / name+store """ store = Path(save_dir) / store store = store.with_name(name + return store
[docs]def parametrized(dec): def layer(*args, **kwargs): def repl(f): return dec(f, *args, **kwargs) return repl return layer
[docs]@parametrized def timed(f, name=None): @wraps(f) def decorated(*args, **kwargs): t = perf_counter() res = f(*args, **kwargs) to_print = name or f.__name__ logging.debug(f"Timing:{to_print}:{perf_counter() - t}s") return res return decorated