# -*- coding: utf-8 -*-
#
# Container-related models.
#
# ------------------------------------------------
# imports
# -------
import functools
import itertools
import json
import six
from gems import cached, composite
import esp
from .__base__ import BaseModel
from .__base__ import (
yaml_str,
yaml_str_list,
raw,
export_variable_definitions,
export_fixed_id,
push_variable_definitions,
)
from .project import Experiment
# models
# ------
[docs]class ContainerType(BaseModel):
"""
Object for interacting with container types from the ESP database.
See the `Usage <./usage.html>`_ and `Examples <./examples.html>`_ pages
of the documentation for more context and comprehensive examples of
how to create and use this type of objects.
Configuration:
Single-element container:
.. code-block:: yaml
name: 96-Well Plate
desc: 96-Well Plate for sample aliquots
tags: [plate, lab]
label_format: '%s%s'
slot_capacity: element
axes:
- Rows: [A, B, C, D, E, F, G, H]
- Cols: [01, 02, 03, 04, 05, 06, 07, 08, 09, 10, 11, 12]
contains:
- Sample
Multi-element container:
.. code-block:: yaml
name: Freezer
desc: Freezer for samples and racks
tags: [freezer, lab]
label_format: 'Shelf %s'
slot_capacity: list
axes:
- Shelf: [1, 2, 3, 4, 5]
contains:
- Sample
- Container
Type with default containers:
.. code-block:: yaml
name: Freezer
desc: Freezer for samples and racks
tags: [freezer, lab]
label_format: 'Shelf %s'
slot_capacity: list
axes:
- Shelf: [1, 2, 3, 4, 5]
contains:
- Sample
- Container
create:
- Freezer 1
- Freezer 2
Nested config with containers and samples to fill:
.. code-block:: yaml
name: Freezer
desc: Freezer for samples and racks
tags: [freezer, lab]
label_format: 'Shelf %s'
slot_capacity: list
axes:
- Shelf: [1, 2, 3, 4, 5]
contains:
- Sample
- Container
create:
- Freezer 1
- name: Freezer 2
barcode: 1234
fill:
Shelf 1: [ESP0001, ESP0002]
Examples:
.. code-block:: python
>>> from esp.models import ContainerType
>>> ct = ContainerType('Freezer')
>>> ct.name, ct.created_at
('Freezer', '2019-06-21T16:04:01.199076Z')
>>> ct.containers
[<Container(name=Freezer 1)>, <Container(name=Freezer 2)>]
Arguments:
ident (str): Name or uuid for object.
"""
__api__ = "container_types"
__api_cls__ = "ContainerType"
__version_api__ = "container_type_definitions"
__defaults__ = {
"renderer": "defaultcontainer.js",
"contains": ["Sample", "Container"],
"slot_capacity": "list",
}
__mutable__ = BaseModel.__mutable__ + [
"contains",
"group_meta",
"resource_vars",
"fixed_id",
"contained_type_params",
]
def _push_meta(self):
# make sure we invert the axes back into meta.
# all other properties are already properly pushed.
for i, axis in enumerate(self.axes):
if i >= len(self.meta.axis_names):
self.meta.axis_names.append(axis.name)
self.meta.dims.append(axis.size)
self.meta.axis_labels.append(axis.labels)
else:
self.meta.axis_names[i] = axis.name
self.meta.dims[i] = axis.size
self.meta.axis_labels[i] = axis.labels
if isinstance(self.meta, composite):
return self.meta.json()
return self.meta
__push_format__ = {
"meta": _push_meta,
"resource_vars": lambda x: push_variable_definitions(raw(x.variables)),
}
__exportable__ = BaseModel.__base_exportable__ + [
"contains",
"slot_capacity",
"label_format",
"renderer",
"axes",
"variables",
"fixed_id",
"contained_type_params",
]
def _export_meta(self):
meta_copy = dict(self.meta.json())
for k in [
"axis_labels",
"axis_names",
"contains_cls",
"dims",
"format",
"label_map",
"labels",
"location_holds",
]:
meta_copy.pop(k, None)
return meta_copy
def _export_variables(self):
return export_variable_definitions(self.variables)
def _export_contained_type_params(self):
result = []
for contained_type in self.contained_type_params:
workflowable_resource_type = contained_type.workflowable_resource_type
name = workflowable_resource_type.name
max_contained = contained_type.max_contained
min_contained = contained_type.min_contained
transfer_item_qty = contained_type.transfer_item_qty
sample_type = esp.models.SampleType(workflowable_resource_type.uuid)
fixed_id = sample_type.fixed_id
result.append(
{
"workflowable_resource_type": {"name": name, "fixed_id": fixed_id, "type": sample_type.cls},
"max_contained": max_contained,
"min_contained": min_contained,
"transfer_item_qty": transfer_item_qty,
}
)
return result
__export_format__ = {
"axes": lambda x: [{yaml_str(axis.name): yaml_str_list(axis.labels)} for axis in x.axes],
"meta": _export_meta,
"variables": _export_variables,
"fixed_id": export_fixed_id,
"contained_type_params": _export_contained_type_params,
}
@staticmethod
def format_contained_type_params_for_import(config):
contained_type_params = config.get("contained_type_params", [])
for contained_type in contained_type_params:
workflowable_resource_type = contained_type.get("workflowable_resource_type")
if workflowable_resource_type is None:
raise ValueError(
"`workflowable_resource_type` property is required for {}".format(
json.dumps(contained_type, default=str)
)
)
name_or_fixed_id = workflowable_resource_type.pop("fixed_id", workflowable_resource_type.get("name"))
if name_or_fixed_id is None:
raise ValueError(
"`name` or `fixed_id` property is required for {}".format(json.dumps(contained_type, default=str))
)
esp_model_type = workflowable_resource_type.get("type")
if esp_model_type is None:
raise ValueError("`type` property is required for workflowable_resource_type")
# normalize model type in case of spaces (ex. Container Type -> ContainerType)
esp_model_type = esp_model_type.replace(" ", "")
sample_type = getattr(esp.models, esp_model_type)(name_or_fixed_id)
workflowable_resource_type["uuid"] = sample_type.uuid
[docs] @classmethod
def parse_import(cls, config, overwrite=False, allow_snapshot_uuid_remap=False):
"""
Create new ContainerType in ESP database using config file or other data.
Args:
config (str, dict, list): Config file or information to use in
creating new object.
overwrite (bool): Whether or not to delete current entry in
the ESP database.
"""
if "axes" not in config:
raise ValueError("axes property must be specified for ContainerType objects!")
# meta
axes = config.pop("axes")
meta = config.get("meta", {})
meta["axis_labels"] = [list(x.values())[0] for x in axes]
meta["dims"] = [len(x) for x in meta["axis_labels"]]
meta["axis_names"] = [list(x.keys())[0] for x in axes]
meta["format"] = config.pop("label_format", "%s" * len(axes))
cls._verify_label_format(meta["format"], len(axes))
meta["location_holds"] = config.pop("slot_capacity")
cls._verify_slot_capacity(meta["location_holds"])
config["meta"] = meta
# resource_vars
variables = config.pop("variables", config.pop("resource_vars", []))
if variables:
config["resource_vars"] = push_variable_definitions(variables)
cls.format_contained_type_params_for_import(config)
# group meta
config["group_meta"] = {"customrenderer": "#include " + config.pop("renderer")}
cls._create = config.get("create", [])
return config
[docs] @classmethod
def parse_response(cls, data, overwrite=False, prompt=False, allow_snapshot_uuid_remap=False):
"""
Method for parsing request and returning
:param allow_snapshot_uuid_remap:
"""
create = cls._create
del cls._create
ct = cls.from_data(data)
for idx, conf in enumerate(create):
if isinstance(conf, six.string_types):
create[idx] = {"name": conf}
create[idx]["container_type"] = ct
if len(create):
Container.create(create, overwrite=False)
return ct
@staticmethod
def _verify_slot_capacity(value):
if value not in ["list", "element"]:
raise ValueError("slot_capacity must be one of `list` or `element`. " "Received: {}".format(value))
@staticmethod
def _verify_label_format(label_format, naxes):
values = tuple([str(x) for x in range(naxes)])
try:
label_format % values
except TypeError:
raise ValueError(
(
"Label format `{}` is incompatible with number of axes ({}). "
"Hint: the label format should have one %s for each axis dimension."
).format(label_format, naxes)
)
[docs] def drop(self, deep=False):
"""
Overwrite drop to delete all corresponding containers as well.
"""
if self.exists():
if deep:
for ct in self.containers:
ct.drop()
return super(ContainerType, self).drop()
@cached
def containers(self):
"""
Property for accessing all related containers.
"""
ret = []
for ct in Container.all():
if ct.container_type.uuid == self.uuid:
ret.append(ct)
return ret
@property
def renderer(self):
"""
The name of the container rendering script.
"""
return self.group_meta.customrenderer.replace("#include ", "")
@renderer.setter
def renderer(self, value):
"""
Set the name of the container rendering script.
value should just the the name of the script.
"""
self.group_meta.customrenderer = "#include " + value
@property
def dimensions(self):
"""
Return the container dimensions.
"""
return self.meta.dims
@dimensions.setter
def dimensions(self, value):
"""
Set the container dimensions.
"""
self.meta.dims = value
@cached
def axes(self):
"""
Return the axes structure List[Dict[str,object]]. Each list entry
is a dict of information about a single axis, including keys:
* name (str): the axis name
* size (int): the axis size
* labels (List[str]): the axis labels.
"""
ret = []
for name, dim, labels in zip(self.meta.axis_names, self.meta.dims, self.meta.axis_labels):
ret.append({"name": name, "size": dim, "labels": labels})
return composite(ret)
[docs] def slot_label(self, *idxdims, **kwdims):
"""
Build individual label components.
Examples:
>>> ct = ContainerType.create({'name': '96-well Plate', axes=[
... {'Row': [x for x in 'ABCDEFGH']},
... {'Col': ['{:02}'.format(x) for x in range(1, 12)]}],
... 'slot_capacity': 'element'})
>>> ct.slot_label(1, 1)
'A01'
>>> ct.slot_label(Row=1, Col=1)
'A01'
Args:
*idxdims (list): List of 1-based dimension indices in the same
order as axes or dimensions
**kwdims (dict): Dict of axis-name: axis-index
"""
components = []
if idxdims:
if kwdims:
raise ValueError("Dimensions must be specified exclusively by " "either index or dimension name")
if len(idxdims) != len(self.axes):
raise ValueError(
"Expected {} index{}, got {}".format(
len(self.axes), "" if len(self.axes) == 1 else "es", len(idxdims)
)
)
for axis, idx in zip(self.axes, idxdims):
if idx < 1 or idx > axis.size:
raise ValueError("Dimension for `{}` must between 1 and {}".format(axis.name, axis.size))
components.append(axis.labels[idx - 1])
elif kwdims:
for axis in self.axes:
if axis.name not in kwdims:
raise ValueError("Need to specify `{}`".format(axis.name))
idx = kwdims[axis.name]
if idx < 1 or idx > axis.size:
raise ValueError("Dimension for `{}` must be between 1 and {}".format(axis.name, axis.size))
components.append(axis.labels[idx - 1])
else:
raise ValueError("Must specify dimensions!")
return self.label_format % tuple(components)
@property
def label_format(self):
return self.meta.format
@label_format.setter
def label_format(self, value):
self.meta.format = value
@property
def slot_capacity(self):
return self.meta.location_holds
@slot_capacity.setter
def slot_capacity(self, value):
self._verify_slot_capacity(value)
self.meta.location_holds = value
@cached
def total_capacity(self):
"""
Return the total capacity of the container.
If the slot capacity is "list", returns "Inf", otherwise, returns
the product of the length of all dimensions.
"""
if self.slot_capacity == "list":
# Use numpy's inf?
return "Inf"
return functools.reduce(lambda a, b: a * b, self.dimensions)
def _slot_for_position_1d(self, position, by):
# Note: in this function, position is 0-based!
return [self.axes[0][position]]
def _slot_for_position_2d(self, position, by):
# Note: in this function, position is 0-based!
# for now, just 2d.
slot = [None] * len(self.dimensions)
# assume 2d for now. Generalization would allow "by" to be a list of
# indices.
notby = int(not by)
slot[by] = position // self.dimensions[notby]
slot[notby] = position % self.dimensions[notby]
return slot
[docs] def slot_for_position(self, position, by, recycle=False, label=True):
"""
Convert 1d index to 1d or 2d coordinates.
Note that x may be > slot_count, so this can be used when laying out a list
of samples into a list of locations.
Args:
position (int): the 1-based 1d position (e.g.: for a 96 well plate,
position is 1-96)
by (0|1): The fastest rotating axis for a 2d container. Not used
for 1d containers, where 0 is the first axis and 1 is the second.
recycle (bool): If true, position may be > total capacity of
the container type under the assumption that the positions will
be "recycled" across multiple containers of the same type.
label (bool): If true, the position label is returned. Otherwise,
the position coordinates (0-based) are returned as a list-like
object.
Examples:
>>> ct = ContainerType.create({'name': '96-well Plate', axes=[
... {'Row': [x for x in 'ABCDEFGH']},
... {'Col': ['{:02}'.format(x) for x in range(1, 12)]}],
... 'slot_capacity': 'element'})
>>> ct.slot_for_position(25, 0)
"C01"
>>> ct.slot_for_position(25, 1)
"A04"
>>> ct.slot_for_position(25, 0, label=False)
[3, 1]
>>> ct.slot_for_position(25, 1, label=False)
[1, 4]
>>> ct.slot_for_position(100, 0)
ValueError: 100 does not fit into 96 slots
>>> ct.slot_for_position(100, 0, recycle=True)
"A04"
"""
# TODO: Refactor location_support, this code, and various invokables
# into a cohesive set of code for supporting auto-layout either in the
# client or on the server.
if self.total_capacity == "Inf":
raise ValueError("No auto ordering with infinite-capacity slots.")
# convert to 0-based for internal sanity.
position -= 1
if position >= self.total_capacity:
if not recycle:
raise ValueError("{} does not fit into {} slots".format(position + 1, self.total_capacity))
position -= self.total_capacity * (position // self.total_capacity)
if len(self.dimensions) == 1:
ret = self._slot_for_position_1d(position, by)
elif len(self.dimensions) == 2:
ret = self._slot_for_position_2d(position, by)
else:
raise ValueError("Only support slot_for_position in 1 or 2 dimensions at this time.")
# back to 1-based indexing.
ret = [x + 1 for x in ret]
if label:
# note: slot_for_position returns 0-based indexes, but slot_label
# expects 1-based indexes.
return self.slot_label(*ret)
return ret
# TODO: This should probably be present on the base class, or a mixin.
# but sticking with copy/paste for the moment.
@classmethod
def _push_variables(cls, data):
"""
Format variables data structure for parsing import and
pushing updates. This is particularly useful for updating complex
variables like containers or other inventory items.
"""
from .container import ContainerType
res = []
for var in data:
# location
if var.get("var_type") == "location":
vmeta = var.setdefault("meta", {})
if "container" not in var and "container_type" not in vmeta:
raise ValueError("Location variables must define container")
if "container" in var:
vmeta["container_type"] = ContainerType(var.pop("container")).uuid
if "fields" in var:
vmeta["location_fields"] = var.pop("fields")
# digital approval
if var.get("var_type") == "approval":
var.setdefault("meta", {})
if "workgroup" in var:
var["meta"]["workgroup"] = var.pop("workgroup")
if "workgroup" in var["meta"]:
from esp.models import Workgroup
wg = Workgroup(var["meta"]["workgroup"])
var["meta"]["workgroup"] = wg.uuid
# barcode
if var.get("var_type") == "barcode":
var.setdefault("meta", {})
if "barcode_type" in var:
var["meta"]["barcode_type"] = var.pop("barcode_type")
# itemqtyadj
if var.get("var_type") == "itemqtyadj" or var.get("var_type") == "itemadj":
var.setdefault("meta", {})
if "item_type" in var:
var["meta"]["item_type"] = var.pop("item_type")
# onchange
if "onchange" in var:
var.setdefault("meta", {})["onchange"] = var.pop("onchange")
res.append(var)
return res
@property
def variables(self):
return self.resource_vars
@variables.setter
def variables(self, variables):
self.resource_vars = variables
[docs]class Container(BaseModel):
"""
Object for interacting with containers from the ESP database.
See the `Usage <./usage.html>`_ and `Examples <./examples.html>`_ pages
of the documentation for more context and comprehensive examples of
how to create and use this type of objects.
Configuration:
Create container:
.. code-block:: yaml
name: Freezer 1
desc: Freezer in lab A
tags: [freezer, lab]
type: Freezer
barcode: 12345
Container with samples to fill:
.. code-block:: yaml
name: Freezer 1
desc: Freezer in lab A
tags: [freezer, lab]
type: Freezer
barcode: 12345
fill:
Shelf 1:
- ESP0001
- name: ESP0002
desc: special new sample
tags: [one, two]
Configuration Notes:
* Upon creation, samples can be used to fill container slots
via the ``fill`` parameter.
* Default container values can be set for samples using the
``variables`` parameter.
Examples:
.. code-block:: python
>>> from esp.models import Container
>>> ct = Container('Freezer 1')
>>> ct.name, ct.created_at
('Freezer 1', '2019-06-21T16:04:01.199076Z')
>>> # show relationships
>>> ct.slots
['Shelf 1', 'Shelf 2', 'Shelf 3', 'Shelf 4', 'Shelf 5']
>>> ct.samples
[<Sample(name=ESP0001)>, <Sample(name=ESP0002)>]
>>> ct.slot_samples
{
'Shelf 1': [<Sample(name=ESP0001)>, <Sample(name=ESP0002)>],
'Shelf 2': [],
'Shelf 3': [],
...
}
>>> # put items in specific slots
>>> ct.put_items_in_slots(
>>> items=[Sample('ESP0001'), Sample('ESP0002')],
>>> slots=['Shelf 1', 'Shelf 2']
>>> )
Arguments:
ident (str): Name or uuid for object.
"""
__api__ = "containers"
__api_cls__ = "Container"
__mutable__ = BaseModel.__mutable__ + [
"items",
"barcode",
"resource_vals",
]
__exportable__ = BaseModel.__base_exportable__ + ["type", "barcode", "variables", "fill", "status"]
def _export_fill(self):
from .sample import Sample
from .inventory import Item
ret = {}
if len(self.container_type.contains) == 1:
ret = {slot: [x.name for x in self.slot_objects[slot]] for slot in self.slot_objects}
else:
for slot in self.slot_objects:
ret[slot] = []
for obj in self.slot_objects[slot]:
if isinstance(obj, Sample):
ret[slot].append({"name": obj.name, "sample_type": obj.sample_type.name})
elif isinstance(obj, Item):
ret[slot].append({"name": obj.name, "item_type": obj.itemtype.name})
else: # assuming its a container
ret[slot].append({"name": obj.name, "container_type": obj.container_type.name})
return ret
__export_format__ = {
"type": lambda x: x.container_type.name,
"variables": lambda x: {val["name"]: val["value"] for val in x.resource_vals},
"fill": _export_fill,
}
__push_format__ = {
"resource_vals": lambda x: [{"name": k["name"], "value": x.variables[k["name"]]} for k in x.resource_vals]
}
[docs] @classmethod
def parse_import(cls, config, overwrite=False, allow_snapshot_uuid_remap=False):
"""
Create new Container in ESP database using config file or other data.
Args:
config (str, dict, list): Config file or information to use in
creating new object.
overwrite (bool): Whether or not to delete current entry in
the ESP database.
"""
if "type" in config:
config["container_type"] = config.pop("type")
if "container_type" not in config:
raise ValueError("Container creation requires container_type key.")
if isinstance(config["container_type"], (dict, list)):
ctype = ContainerType.create(config["container_type"], overwrite=overwrite)
elif isinstance(config["container_type"], six.string_types):
ctype = ContainerType(config["container_type"])
elif isinstance(config["container_type"], ContainerType):
ctype = config["container_type"]
else:
raise AssertionError("Specified data for container_type not supported!")
if not ctype.exists():
raise ValueError("Specified ContainerType {} does " "not exist!".format(config["container_type"]))
config["container_type"] = ctype.def_uuid
# generate barcode (if specified)
if "barcode" in config:
config["barcode"] = str(config["barcode"])
if "${HEX}" in config["barcode"]:
import hashlib
ident = ctype.name + ":" + config["name"]
config["barcode"] = str(hashlib.md5(ident.encode()).hexdigest())[:7]
variables = config.pop("variables", config.pop("resource_vals", {}))
config["resource_vals"] = []
for rv in ctype.variables:
key = rv["name"]
v = variables.get(key, rv["default_val"])
config["resource_vals"].append({"name": key, "value": v})
cls._fill = config.pop("fill", None)
return config
@classmethod
def _guess_fill_types(cls, ct, items):
if len(ct.container_type.contains) == 1:
if ct.container_type.contains[0] == "Sample":
return items, [], []
elif ct.container_type.contains[0] == "Item":
return [], [], items
else:
return [], items, []
containers = []
samples = []
inventory_items = []
for x in items:
if isinstance(x, dict) and "container_type" in x:
containers.append(x)
elif isinstance(x, dict) and "item_type" in x:
inventory_items.append(x)
else:
samples.append(x)
return samples, containers, inventory_items
[docs] @classmethod
def parse_response(cls, data, overwrite=False, prompt=False, allow_snapshot_uuid_remap=False):
"""
Parse response and put items in slots (if specified).
:param allow_snapshot_uuid_remap:
"""
from esp.models import Sample
from esp.models import Item
fill = cls._fill
del cls._fill
ct = cls.from_data(data)
if fill is not None:
objects, slots = [], []
for key, items in fill.items():
slots.extend([key] * len(items))
samples, containers, inventory_items = cls._guess_fill_types(ct, items)
if samples:
samples = Sample.create(samples, overwrite=False)
if not isinstance(samples, (list, tuple)):
samples = [samples]
samples = {x.name: x for x in samples}
if inventory_items:
inventory_items = Item.create(inventory_items, overwrite=False)
if not isinstance(inventory_items, (list, tuple)):
inventory_items = [inventory_items]
inventory_items = {x.name: x for x in inventory_items}
if containers:
containers = cls.create(containers, overwrite=False)
if not isinstance(containers, (list, tuple)):
containers = [containers]
containers = {x.name: x for x in containers}
# reconstruct the original list as hydrated objects.
objs = []
for item in items:
if not isinstance(item, str):
item = item["name"]
if item in samples:
objs.append(samples[item])
elif item in inventory_items:
objs.append(inventory_items[item])
else:
objs.append(containers[item])
objects.extend(objs)
ct.put_items_in_slots(objects, slots)
return ct
[docs] def drop(self, deep=False):
"""
Remove all items from container and drop.
"""
for idx, item in enumerate(self.items):
self.items[idx]["item_uuid"] = []
try:
self.push()
except AssertionError:
# backend can get into strange stage with resources
# pass for now and update backend to more gracefully
# handle this later
pass
return super(Container, self).drop()
@cached
def container_type(self):
"""
Return container type definition associated with item.
.. note:: Using container type definition here is unintuitive
and backend storage of data should be resolved.
"""
return ContainerType.from_definition(self.data.container_type)
@cached
def slots(self):
"""
Return array of container slots.
"""
return [slot.label for slot in self.items]
@cached
def samples(self):
"""
Return array of samples associated with container.
"""
from .sample import Sample
res = []
for key in self.slot_objects:
res.extend([x for x in self.slot_objects[key] if isinstance(x, Sample)])
return res
@cached
def slot_objects(self):
from .sample import Sample
from .container import Container
from .inventory import Item
uuids = []
# use a 2-pass scan to avoid N API hits.
for slot in self.items:
if slot["item_uuid"] is None:
continue
if self.container_type.slot_capacity == "element":
uuids.append(slot["item_uuid"])
else:
uuids.extend(slot["item_uuid"])
objects = []
# empty uuids can trigger a search of the entire samples/container registry.
if len(uuids) > 0:
if "Sample" in self.container_type.contains:
objects.extend(Sample.search(uuids=uuids))
if len(objects) != len(uuids) and "Container" in self.container_type.contains:
objects.extend(Container.search(uuids=uuids))
if len(objects) != len(uuids) and "Item" in self.container_type.contains:
objects.extend(Item.search(uuids=uuids))
uuid_to_obj = {x.uuid: x for x in objects}
def vet_uuid(uuid):
if uuid not in uuid_to_obj:
raise ValueError(
"Unable to find object for uuid `{}` in container `{}` ({})".format(uuid, self.name, self.uuid)
)
return uuid_to_obj[uuid]
ret = {k: [] for k in self.slots}
for slot in self.items:
uuid = slot["item_uuid"]
if uuid is None:
continue
if self.container_type.slot_capacity == "element":
ret[slot["label"]].append(vet_uuid(uuid))
else:
ret[slot["label"]].extend([vet_uuid(x) for x in uuid])
return ret
@cached
def slot_samples(self):
"""
Return array of container slots, filtered to only samples.
"""
from .sample import Sample
return {k: [x for x in self.slot_objects[k] if isinstance(x, Sample)] for k in self.slot_objects}
[docs] def put_items_in_slots(self, items, slots, overwrite=False, worksheet=None, fields=None, save=True):
"""
Puts an item into the specified slot.
Args:
item (Sample|Container|List[Sample|Container]) - Instance of object
that may be placed in a container slot, or a list of instances.
slots (str|List[str]): Name(s) of slot(s) to put item in.
overwrite (bool) - If the slots hold single elements and the slot
is already occupied, overwrite=True will kick that item out and
put this item in; overwrite=False will raise a ValueError.
If the slot is empty or allows multiple items, the value of
overwrite is ignored - new items are always appended to the
slot.
worksheet (esp.models.SampleSheet) - Instance of SampleSheet.
fields (dict[str,str]) - Dictionary of slot-specific fields to set
for the sample. Note that fields can only be set if worksheet
is specified.
save (bool) - If True (the default), the new item set is
immediately saved to the backend. Note that callers are still
responsible for pushing resource vals to the backend even if
worksheet is specified.
Return:
None|Dict[UUID, List[Dict]] - If worksheet is None, None is
returned. Otherwise, a Dict is returned mapping UUIDs to
location structures. The location structures can be directly
converted to string via json.dumps to be used as resource val
values.
Items can be placed into a container inside or outside of a worksheet.
Within the context of a worksheet, the backend currently stores the
container information in both the lims val and the container, so
synchronizing across both is required. Hence, if you're putting an item
into a slot, and that information will be recorded in a worksheet, pass
the worksheet object to this method and an appropriate lims val will
be created.
"""
def normalize(input_):
if isinstance(input_, (tuple, list)):
return input_
return [input_]
def prepare(input_, maxlen):
if len(input_) == maxlen:
return input_
return itertools.cycle(input_)
slots = normalize(slots)
items = normalize(items)
fields = normalize(fields)
maxlen = max([len(slots), len(items), len(fields)])
if (
(len(slots) != maxlen and len(slots) != 1)
or (len(items) != maxlen and len(slots) != 1)
or (len(fields) != maxlen and len(fields) != 1)
):
raise ValueError(
"Must provide either a single item, a single slot, or "
"equal-length lists of both. Found len(slots): "
"{}; len(items): {}; len(fields): {}".format(len(slots), len(items), len(fields))
)
slots = prepare(slots, maxlen)
items = prepare(items, maxlen)
fields = prepare(fields, maxlen)
ret = {} if worksheet else None
for item, slotname, fielddict in zip(items, slots, fields):
slot = self._slot_for_label(slotname)
tries = [item.__class__.__name__] + item.__type_aliases__
if not any(x in self.container_type.contains for x in tries):
raise ValueError("Items of type `{}` may not be put in `{}`".format(item.__class__.__name__, self.name))
if self.container_type.slot_capacity == "element":
self._update_element_slot(item, slot, worksheet, overwrite)
else:
self._update_list_slot(item, slot, worksheet)
if worksheet is None:
continue
rval = self._build_resource_val(item, slot, fielddict, worksheet)
if item.uuid in ret:
ret[item.uuid]["locations"].extend(rval["locations"])
else:
ret[item.uuid] = rval
if save:
self.push()
if worksheet:
ret = {x: ret[x].to_value() for x in ret}
return ret
def _update_element_slot(self, item, slot, worksheet, overwrite):
if slot["item_uuid"] is None or overwrite:
slot["item_uuid"] = item.uuid
if worksheet:
experiment = worksheet.experiment_for_sample(item)
slot["wi_uuid"] = experiment.uuid
# not none and not overwrite _and_ it's a different object than
# what's already in there.
elif slot["item_uuid"] != item.uuid:
raise ValueError(
(
"Container slot `{}` is occupied by `{}` and overwrite is " "False while putting `{}` into the slot"
).format(slot["label"], slot["item_uuid"], item.uuid)
)
def _update_list_slot(self, item, slot, worksheet):
if slot["item_uuid"] is None:
slot["item_uuid"] = []
slot["item_uuid"].append(item.uuid)
if worksheet:
experiment = worksheet.experiment_for_sample(item)
if "wi_uuid" in slot and slot["wi_uuid"] != experiment.uuid:
raise ValueError(
"When adding multiple items to the same slot at the same "
"time, all values must derive from the same experiment. "
"This is a limitation of the backend API."
)
slot["wi_uuid"] = experiment.uuid
def _build_resource_val(self, item, slot, fields, worksheet):
# build the LIMS val and also update meta.
# note that if we're dealing with a worksheet, we get to assume
# the item is a sample.
value = ContainerValue("[]")
value.experiment_uuid = worksheet.experiment_for_sample(item).uuid
value.container_uuid = self.uuid
value.container_name = self.name
value.locations.append({"label": slot["label"], "fields": {} if fields is None else fields})
# TODO: The backend really needs to be cleaned up/de-duplicated
# here b/c we're storing information in both the resource_val AND the
# container metadata.
if fields:
# as far as I can tell from watching the DB, the fields meta
# value is last-set-wins. Fields in a worksheet are read from
# the resource_val.
if "fields" not in self.meta:
self.meta["fields"] = {}
return value
def _slot_for_label(self, label):
"""Given a label, return the 1-d slot index for that label.
This is because the Container storage format is, stupidly, a 1
dimensional list rather than a dict."""
# STUPID IMPLMENTATION! FIXME LATER!
for slot in self.items:
if slot.label == label:
return slot
raise ValueError(
"`{}` is not a vald slot in container `{}`, type `{}`".format(label, self.name, self.container_type.name)
)
@cached
def variables(self):
return {x["name"]: x["value"] for x in self.resource_vals}
class ContainerValue(object):
"""
Client-side representation of a container resource value.
Args:
value (str): The container slot resource val string.
"""
def __init__(self, value):
if value:
try:
value = json.loads(value)
except Exception as e:
raise ValueError("Unable to json-parse `{}`!".format(value))
# no support for multi-container values yet.
if not isinstance(value, list) or len(value) > 1:
raise ValueError("Expected value to be a list of length <= 1! ({})".format(value))
if value:
self.container_uuid = value[0]["container"]
if "name" in value[0]:
self._container_name = value[0]["name"]
else:
self._container_name = None
self._container = None
self._experiment = None
if "workflow_instance" in value[0]:
self.experiment_uuid = value[0]["workflow_instance"]
else:
self.experiment_uuid = None
self.locations = []
if "locations" in value[0]:
for location in value[0]["locations"]:
self.locations.append(composite(location))
else:
self.container_uuid = None
self._container = None
self.experiment_uuid = None
self._experiment = None
self.locations = []
self._container_name = None
@property
def container_name(self):
if self._container_name is not None:
return self._container_name
elif self._container is not None:
return self._container.name
else:
return "Unresolved"
@container_name.setter
def container_name(self, name):
self._container_name = name
@property
def container(self):
if self.container_uuid is None:
return None
if self._container is None or self._container.uuid != self.container_uuid:
self._container = Container(self.container_uuid)
return self._container
@container.setter
def container(self, container):
self._container = container
self.container_uuid = None if container is None else container.uuid
@property
def experiment(self):
if self.experiment_uuid is None:
return None
if self._experiment is None or self._experiment.uuid != self.experiment_uuid:
self._experiment = Experiment(self.experiment_uuid)
return self._experiment
@experiment.setter
def experiment(self, experiment):
self._experiment = experiment
self.experiment_uuid = None if experiment is None else experiment.uuid
def occupy_slot(self, slot, fields=None):
val = composite({})
val.label = slot
if fields:
val.fields = fields
self.locations.append(val)
def to_value(self):
if self.locations:
if not self.experiment_uuid or not self.locations:
raise AssertionError("locations, container, and experiment must all be set, or " "all be None/Falsey.")
if not self.locations:
return json.dumps([])
struct = {"container": self.container_uuid, "workflow_instance": self.experiment_uuid, "locations": []}
if self._container is not None:
# for ESP 2.0, the client-side renderer likes having name in
# the value, and putting it there saves a client-side fetch
# for each cell in the table.
struct["name"] = self._container.name
elif self._container_name is not None:
struct["name"] = self._container_name
for location in self.locations:
struct["locations"].append(dict(location))
return json.dumps([struct])