# -*- coding: utf-8 -*-
#
# Sample-related models.
#
# ------------------------------------------------
# imports
# -------
from enum import Enum
import json
import os
import warnings
import esp
from ..utils import normalize_resource_link_type, convert_date_value
try:
from urllib.parse import quote_plus
except ImportError:
from urllib import quote_plus
import dateparser
from gems import cached, composite
import six
from .. import base
from .. import utils
from .__base__ import BaseModel, LinkedModel, create_and_return_uuid, export_fixed_id, cached_uuid
from .__base__ import raw, compile_report, espfile_or_value, export_variable_definitions, push_variable_definitions
def _export_resource_link_val(r_val_value):
resource_link_value = json.loads(r_val_value)
link_type = resource_link_value["linkType"]
link_type = normalize_resource_link_type(link_type)
if not hasattr(esp.models, link_type):
# link_type can be a pre-supplied/Out-of-box Entity Class.
# Customers and implementations may create other entity classes like L7Product
link_type = "Sample"
# reference resource by fixed_id and name
uuid = resource_link_value.pop("uuid")
obj = getattr(esp.models, link_type)(uuid)
if obj.fixed_id:
resource_link_value["fixed_id"] = obj.fixed_id
resource_link_value["name"] = obj.name
return json.dumps(resource_link_value, default=str)
def _convert_date_value(value, resource_value):
if not value:
return value
if isinstance(value, str):
if "{{" in value:
return value
ret = dateparser.parse(value)
# dateparser can return null if it can't parse value...
if not ret:
return value
else:
ret = value
# strip time value if it isn't conveying information.
if hasattr(ret, "hour") and ret.hour == 0 and ret.minute == 0 and hasattr(ret, "date"):
ret = ret.date()
ret = ret.isoformat()
if "T" in ret and not ret.endswith("Z"):
ret += "Z"
base.logger.info("Converting value `{}` to `{}`".format(value, ret))
return ret
# models
# ------
class EntityType(BaseModel):
"""
Object for interacting with EntityTypes 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.
.. note:: Previous versions of ESP used the word ``SampleType`` instead
of ``EntityType``. The python client now uses ``EntityType`` but
``SampleType``, though deprecated, is still supported for backwards
comatibility with older code.
Configuration:
Simple entity type:
.. code-block:: yaml
name: Library
desc: Entity type for Library
tags: [library, demo]
Entity type with auto-naming sequence:
.. code-block:: yaml
name: Library
desc: Entity type for Library
tags: [library, demo]
sequences:
- LIBRARY SEQUENCE
Entity type with variables and sequence:
.. code-block:: yaml
name: Library
desc: Entity type for Library
tags: [library, demo]
sequences:
- LIBRARY SEQUENCE
variables:
- Entity Type:
rule: string
value: Illumina Library
- Numeric Value:
rule: numeric
value: 0
Create Entity type and new entities:
.. code-block:: yaml
name: Library
variables:
- Entity Type:
rule: string
value: Illumina Library
- Numeric Value:
rule: numeric
value: 0
create:
- Library 1
- name: Library 2
desc: My special library.
variables:
Entity Type: Non-Illumina Library
Numeric Value: 2
Configuration Notes:
* Variables specified for entities can take the same format as variables
defined for protocols within ESP.
* Entity creation can be nested in EntityType configuration using the
``create`` parameter.
* For any new EntityType object, corresponding information
MUST be included in the ``lab7.conf`` configuration file. Here
are lines in that file that's relevant to the examples above:
.. code-block:: yaml
lims:
auto_sample_id_format: "ESP{sample_number:06}"
sample_id_sequences:
- name: "ESP SEQUENCE"
format: ESP{sample_number:06}
- name: "LIBRARY SEQUENCE"
format: LIB{sample_number:03}
sequence: library_seq
sequences:
lab7_sample_auto_id_seq: 1
library_seq: 1
Examples:
.. code-block:: python
>>> from esp.models import EntityType
>>> et = EntityType('Library')
>>> et.name, et.created_at
('Demo Consumable', '2019-06-21T16:04:01.199076Z')
>>> # show relationships
>>> et.entities
[<Entity(name=LIB0001)>, <Entity(name=LIB0002)>, <Entity(name=LIB0003)>]
>>> et.variables
[{'name': 'Value 1', 'value': 1, ...}, {'name': 'Value 2', 'value': 2, ...}]
Args:
ident (str): Name or uuid for object.
"""
__api__ = "sample_types"
__api_cls__ = "SampleType"
__version_api__ = "sample_type_definitions"
__type_aliases__ = ["SampleType"]
__push_format__ = {
"meta": lambda x: {
"lab7_id_sequence": list(x.sequences),
},
"resource_vars": lambda x: push_variable_definitions(raw(x.variables)),
}
__mutable__ = BaseModel.__mutable__ + ["resource_vars", "icon_svg", "view_template", "fixed_id"]
__exportable__ = [
"name",
"tags",
"desc",
"sequences",
"variables",
"class",
"icon_svg",
"view_template",
"fixed_id",
]
def _export_class(self):
if self.workflowable_class.name == "Sample":
return ""
return self.workflowable_class.name
def _export_variables(self):
return export_variable_definitions(self.variables)
__export_format__ = {
"sequences": lambda x: raw(x.sequences),
"variables": _export_variables,
"class": _export_class,
"fixed_id": export_fixed_id,
}
@classmethod
def parse_import(cls, config, overwrite=False, allow_snapshot_uuid_remap=False):
"""
Create new object 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.
"""
meta = {} if "meta" not in config else config["meta"]
if "sequences" in config:
meta["lab7_id_sequence"] = config["sequences"]
else:
meta["lab7_id_sequence"] = ["ESP SEQUENCE"]
config["meta"] = meta
config["resource_vars"] = push_variable_definitions(config.pop("variables", []))
compiled = config.pop("compiled", False)
if "view_template" in config:
config["view_template"] = compile_report(config["view_template"], compiled)
if "icon_svg" in config and config["icon_svg"] is not None:
config["icon_svg"] = espfile_or_value(config["icon_svg"], True)
if "workflowable" in config:
config.setdefault("class", config.pop("workflowable"))
if "class" in config:
config["workflowable_resource_class"] = create_and_return_uuid(
WorkflowableClass, config.pop("class"), overwrite=overwrite
)
cls._create = config.pop("create", [])
return config
@classmethod
def parse_local_import(cls, data):
if "icon_svg" in data:
raise ValueError(
"Custom icons (`icon_svg`) in EntityType definitions " "are not currently supported for local imports."
)
class_ = None
if "class" in data:
class_ = data.pop("class")
data = cls.parse_import(data)
if class_:
data["workflowable_resource_class"] = cached_uuid(WorkflowableClass, class_)
data["cls"] = cls.__api_cls__
return data
@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
st = cls.from_data(data)
for idx, conf in enumerate(create):
if isinstance(conf, six.string_types):
create[idx] = {"name": conf}
create[idx]["sample_type"] = st
if len(create):
Sample.create(create, overwrite=False)
return st
def drop(self, deep=False):
"""
Issue DELETE request to remove object in ESP database.
"""
if deep:
for obj in self.samples:
obj.drop()
# don't drop default generic sample type via client
if self.name.lower() != "Generic sample".lower() and self.name.lower() != "MES Product".lower():
super(SampleType, self).drop()
return
@cached
def sequences(self):
"""
Return sequence ID used within ESP. This id is used throughout the system
in a variety of different ways, so being able to query it is useful.
"""
return list(self.data["meta"].lab7_id_sequence)
@cached
def variables(self):
"""
Proxy for weird API schema requiring strange 'meta' field.
"""
return self.resource_vars
@cached
def workflowable_class(self):
"""
Proxy for accessing SampleType WorkflowableClass
"""
return WorkflowableClass(self.data.workflowable_resource_class)
@cached
def entities(self):
"""
Return Sammple object associated with sample type.
"""
result = base.SESSION.get("/api/samples?sample_type={}".format(quote_plus(self.name)))
return [Entity.from_data(s) for s in result.json()]
@property
def samples(self):
return self.entities
def _data_by_uuid(self):
# Work around fact that sample -> sample_type association is based
# on the /definition/ uuid, NOT the head uuid. In general, hard to
# know if we're getting head or definition. Try head first.
data = super(SampleType, self)._data_by_uuid()
if not data:
# have to use query here b/c the permission setup on
# get_sample_type_definition tries to resolve a
# SampleType instead of a SampleTypeDefinition.
res = base.SESSION.get("/api/sample_type_definitions?uuids={}".format(json.dumps([self.ident]))).json()
if len(res) == 0:
return {}
return res[0]
else:
return data
SampleType = EntityType
class DuplicateWorkflowableTypeNames(UserWarning):
def __init__(self, typeName, resolvedTypes, selectedType):
resolvedTypesDisplay = [{"uuid": x["uuid"], "cls": x["cls"]} for x in resolvedTypes]
selectedTypeDisplay = {"uuid": selectedType["uuid"], "cls": selectedType["cls"]}
super().__init__(
"WorkflowableTypeName `{}` is shared by more than one type:\n\t* {}\nUsing:\n\t* {}".format(
typeName, "\n\t* ".join(str(x) for x in resolvedTypesDisplay), selectedTypeDisplay
)
)
class CoreWorkflowableClass(Enum):
"""
Enum for dealing with core types the backend considers "Workflowable".
The value determines the relative priority of the classes for situations where
a type name resolves to more than one class. ESP 3.0.0 prevents some
duplications but not others because we have to walk existing implementations forward
where some type names may be deliberately duplicated since ESP < 3.0.0 didn't support
Containers and Items as workflowable types.
Note:
This enum differs from the "WorkflowableClass" in that these three classes are back-end
objects. All user-defined workflowable classes are actually "SampleType" on the backend.
"""
SampleType = 1
ContainerType = 2
ItemType = 3
class WorkflowableType(EntityType):
"""
Generalized EntityType, including ContainerType and ItemType.
This class is normally not used directly by end-users, but is used
within the client for resolving references to SampleType, ContainerType,
and ItemType that were formerly limited to SampleType prior to 3.0.0.
Since: 3.0.2.
"""
def _data_by_name(self):
# Use names b/c sample_types?name=X is a wildcard search for %X%.
query = "/api/sample_types?names={}&clses={}".format(
json.dumps([self.ident]), json.dumps([x.name for x in CoreWorkflowableClass])
)
res = base.SESSION.get(query).json()
if len(res) == 0:
return {}
# Can't make an entity type the same name as containertype and itemtype.
# But it _is_ possible, currently, to make a containertype or itemtype with the
# same name as an entity type.
if len(res) == 1:
return res[0]
# preference order for 3.0.2: SampleType, ItemType, ContainerType.
# Prefer this order because SampleType is the most common; ItemType is allowed for
# sample protocols, but containertype is not.
# Or else we can raise an exception... should be a rare case. However, raising ->
# You can't import, which means you can't correct; this way, you can fix it after.
# But _do_ warn...
res = sorted(res, key=lambda x: CoreWorkflowableClass[x["cls"]].value)
warnings.warn(DuplicateWorkflowableTypeNames(self.ident, res, res[0]))
return res[0]
[docs]class WorkflowableClass(BaseModel):
"""
Object for interacting with WorkflowableClass 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:
Simple workflowable type:
.. code-block:: yaml
name: Patient
desc: A Human Patient
tags: [patient, demo]
plural_name: Patients
view_template: <h1>My View Template</h1>
list_view_template: <h1>My View Template</h1>
detail_panel_template: <h1>My View Template</h1>
icon_svg: <svg xmlns=“http://www.w3.org/2000/svg” width=“24" height=“24” viewBox=“0 0 24 24"><g fill=“none” fill-rule=“evenodd”><path d=“M0 0h24v24H0z”/><path fill=“#550185" fill-rule=“nonzero” d=“M6 6v14h12V6h-3v1H9V6H6zm4-1a2 2 0 1 1 4 0h5v16H5V5h5zm1 5h5v1h-5v-1zm0 3h5v1h-5v-1zm-3-3h2v1H8v-1zm0 3h2v1H8v-1zm0 3h2v1H8v-1zm3 0h5v1h-5v-1z”/>
Examples:
.. code-block:: python
>>> from esp.models import WorkflowableClass
>>> wf = WorkflowableClass('Patient')
>>> wf.name, wf.created_at
('Library', '2019-06-21T16:04:01.199076Z')
Args:
ident (str): Name or uuid for object.
"""
__api__ = "workflowable_resource_classes"
__api_cls__ = "WorkflowableResourceClass"
__mutable__ = BaseModel.__mutable__ + [
"plural_name",
"view_template",
"list_view_template",
"detail_panel_template",
"icon_svg",
]
__exportable__ = [
"name",
"tags",
"desc",
"plural_name",
"view_template",
"list_view_template",
"detail_panel_template",
"icon_svg",
]
__type_aliases__ = ["EntityClass"]
[docs] @classmethod
def parse_import(cls, config, overwrite=False, allow_snapshot_uuid_remap=False):
"""
Create new object 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.
"""
# parse view_template contents
compiled = config.pop("compiled", False)
if config.get("view_template") is not None:
config["view_template"] = compile_report(config["view_template"], compiled)
# parse list_view_template contents
if config.get("list_view_template") is not None:
config["list_view_template"] = compile_report(config["list_view_template"], compiled)
if config.get("detail_panel_template") is not None:
config["detail_panel_template"] = compile_report(config["detail_panel_template"], compiled)
# parse icon
if "icon_svg" in config:
config["icon_svg"] = espfile_or_value(config["icon_svg"], True)
return config
@classmethod
def parse_import_hub(cls, config):
# Pop off 'icon_svg' until we can find a reasonable way to handle it
# that's not ESP File
icon_svg = None
if "icon_svg" in config:
icon_svg = config.pop("icon_svg")
config = cls.parse_import(config)
config["icon_svg"] = icon_svg
config.pop("_format")
return config
[docs] def drop(self, *args, **kwargs):
if self.name.lower() in ["sample", "container", "item", "mesproduct"]:
return
return super(WorkflowableClass, self).drop(*args, **kwargs)
EntityClass = WorkflowableClass
class Entity(LinkedModel):
"""
Object for interacting with Entities 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.
.. note:: Previous versions of ESP used the word ``Sample`` instead
of ``Entity``. The python client currently uses the old
``Sample`` nomenclature for ``Entity`` objects in the system,
which will be deprecated in favor of ``Entity`` in future versions.
Configuration:
Simple entity:
.. code-block:: yaml
name: LIB0001
Entity of specific type with note and tags:
.. code-block:: yaml
name: LIB0001
desc: Library sample
tags: [library, special-sample-1]
type: Library
Entity with variable defaults:
.. code-block:: yaml
name: LIB0001
type: Library
variables:
Entity Type: Illumina Library
Numeric Value: 10
Entity batch:
.. code-block:: yaml
count: 10
type: Illumina Library
Entity batch with variable defaults:
.. code-block:: yaml
count: 10
type: Illumina Library
variables:
Entity Type: Illumina Library
Numeric Value: 10
Configuration Notes:
* Either single entities or batches of entities can be created with
the Entity model ``create`` method.
* Default entity values can be set for entities using the
``variables`` parameter.
Examples:
.. code-block:: python
>>> from esp.models import Entity
>>> library = Entity('Library 1')
>>> library.name, library.created_at
('Demo Consumable', '2019-06-21T16:04:01.199076Z')
>>> # show relationships
>>> library.projects
[<Project(name=My Project)>]
>>> library.experiments
[<Experiment(name=My Experiment 1)>, <Experiment(name=My Experiment 2)>]
>>> library.containers
[<Container(name=Freezer 1)>, <Container(name=Freezer 3)>]
>>> library.parents
[<Entity(name=Parent 1)>]
>>> library.children
[<Entity(name=Child 1)>, <Entity(name=Child 2)>]
>>> # data accessors
>>> library.variables
{'Entity Type': 'Illumina Library', 'Numeric Value': 2}
>>> library.locations
{
'Freezer 1': 'A01',
'Freezer 3': 'B02'
}
>>> library.metadata
[{'My Workflow': {'My Protocol': {'Column 1': 'Value 1', 'Column 2': 'Value 2'}}]
>>> # parent-child relationships
>>> other = Entity('Other Entity')
>>> other.add_children(library)
>>> other.add_children([library, Entity('Other Entity')])
>>> other.add_parents(library)
>>> other.add_parents([library, Entity('Other Entity')])
Args:
ident (str): Name or uuid for object.
"""
__api__ = "samples"
__api_cls__ = "Sample"
__mutable__ = LinkedModel.__mutable__ + [
"sample_type_uuid",
"resource_vals",
]
__create_params__ = ["sample_type"]
__push_format__ = {"resource_vals": lambda x: x._push_variables(x)}
__exportable__ = BaseModel.__base_exportable__ + ["barcode", "type", "variables"]
__export_format__ = {"type": lambda x: x.sample_type.name, "variables": lambda x: x._export_variables(x)}
__pristine__ = ["variables"]
_converters = {
"date": _convert_date_value,
"resource_link": utils.format_resourcelink_value,
"attachment": utils.format_attachment_value,
}
__type_aliases__ = ["Sample"]
@classmethod
def _export_variables(cls, obj):
result = {}
for r_val in obj.resource_vals:
r_val_value = r_val["value"]
r_val_type = r_val["var_type"]
r_val_name = r_val["name"]
exporter = "_export_{}_val".format(r_val_type)
if r_val_value is not None and exporter in globals():
result[r_val_name] = globals()[exporter](r_val_value)
else:
result[r_val_name] = r_val_value
return result
@classmethod
def _push_variables(cls, obj):
vals = raw(obj.resource_vals)
rvars = {x["name"]: x for x in obj.sample_type.resource_vars}
# TODO: Consolidate/centralize value -> string value conversions.
if isinstance(vals, list):
keys = [x["name"] for x in vals]
elif isinstance(vals, dict):
# ensure we send in the correct order.
keys = [x["name"] for x in obj.sample_type.resource_vars]
else:
raise ValueError("Unhandled resource_vals format: `{}`. Expected list or dict.".format(vals))
ret = []
for k in keys:
value = obj.variables[k]
var = rvars.get(k, {"name": k, "var_type": "string"})
if var["var_type"] in cls._converters:
value = cls._converters[var["var_type"]](value, var)
ret.append({"name": k, "value": value})
obj.data.pop("resource_vals")
return ret
@classmethod
def _get_sample_type(cls, config):
alternate_type_keys = ["entity_type", "sample_type"]
tkey = "type"
for type_key in alternate_type_keys:
if type_key in config:
tkey = type_key
break
if isinstance(config.get(tkey), BaseModel):
return config.get(tkey)
else:
return SampleType(config.pop(tkey, "Generic sample"))
@classmethod
def parse_bulk_import(cls, config, overwrite=False):
"""
Create new object in ESP database using config file or other data.
This method should be overwritten by subclasses of LinkedModel for
model-specific logic.
Examples:
.. code-block:: python
>>> # multiple entities with specified entity type.
>>> objs = Sample.create(
>>> count=3,
>>> type='Illumina Sample'
>>> )
>>> print([o.uuid for o in objs])
['8552afc3-882b-41e3-b0cf-1a4cf3f28d33', '96e63b83-1ddc-4e5b-884c-13fcedf5322d']
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.
"""
stype = cls._get_sample_type(config)
names = config.pop("names", [])
count = config.pop("count", len(names))
descs = config.pop("descs", [])
tags = config.pop("tags", [])
idseq = config.pop("lab7_id_sequence", stype.sequences[0])
variables = config.get("variables", {})
if count == 0:
raise AssertionError("`count` or `names` must be provided and > 0!")
if tags and len(tags) != len(names):
raise ValueError("tags must be the same length as names!")
# create stubs for samples
samples = [{} for x in range(count)]
# set sample type for all samples
for idx, sample in enumerate(samples):
sample["sample_type_uuid"] = stype.def_uuid
sample["lab7_id_sequence"] = idseq
sample["resource_vals"] = []
if names:
sample["name"] = names[idx]
if descs:
sample["desc"] = descs[idx]
for rv in stype.variables:
key = rv["name"]
v = rv["default_val"]
if isinstance(variables, (list, tuple)):
v = variables[idx].get(key, v)
elif isinstance(variables.get(key), (list, tuple)):
v = variables[key][idx]
elif key in variables:
v = variables[key]
sample["resource_vals"].append({"name": key, "value": v})
if tags:
sample["tags"] = tags[idx]
return {
"autogen_names": not bool(names),
"lab7_id_sequence": idseq,
"samples": samples,
}
@classmethod
def samples_from_lims_group(cls, config):
group_name = config["from"]
group_type = config.get("type", "experiment")
type_norm = str(group_type).lower()
protocol_name = config.get("protocol_name")
if type_norm == "experiment":
from .project import Experiment as Group
elif type_norm == "worksheet":
from .project import SampleSheet as Group
else:
raise ValueError('"type" must be one of experiment or worksheet when using the `from` form of samples')
group = Group(group_name)
if not group.exists():
raise ValueError("No such {}: {}".format(group_type, group_name))
if protocol_name is None:
return group.protocols[-1].samples
protocol = [x for x in group.protocols if x.name == protocol_name]
if not protocol:
raise ValueError("No such protocol `{}` in {} `{}`".format(protocol_name, group_type, group_name))
return protocol[0].samples
@classmethod
def parse_import(cls, config, overwrite=False, allow_snapshot_uuid_remap=False):
"""
Create new object in ESP database using config file or other data.
This method should be overwritten by subclasses of LinkedModel for
model-specific logic.
Examples:
>>> # create single sample using default sample type
>>> obj = Sample.create('ESP000001')
>>> print(obj.uuid)
8552afc3-882b-41e3-b0cf-1a4cf3f28d33
>>>
>>> # multiple samples with specific names
>>> objs = Sample.create(['ESP0000001', 'ESP0000002'])
>>> print([o.uuid for o in objs])
['8552afc3-882b-41e3-b0cf-1a4cf3f28d33', '96e63b83-1ddc-4e5b-884c-13fcedf5322d']
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.
"""
# bulk import
if "count" in config or "names" in config:
return cls.parse_bulk_import(config, overwrite=overwrite)
if "from" in config:
return cls.samples_from_lims_group(config)
# reformat string input
if isinstance(config, str):
config = {"name": config}
# insert sample with specific name
stype = cls._get_sample_type(config)
config["sample_type"] = stype.name
config["sample_type_uuid"] = stype.def_uuid
variables = config.pop("variables", {})
config["resource_vals"] = []
for rv in stype.variables:
rv_name = rv.get("name")
rv_fixed_id = rv.get("fixed_id")
rv_uuid = rv.get("uuid")
default_val = rv["default_val"]
# try to get value by fixed id first
# else try to get value by name
v = variables.get(rv_fixed_id, variables.get(rv_name, default_val))
if rv["var_type"] in cls._converters and v is not None:
v = cls._converters[rv["var_type"]](v, rv)
config["resource_vals"].append({"uuid": rv_uuid, "name": rv_name, "value": v})
config["meta"] = config.pop("data", {})
cls._children = config.pop("children", [])
return config
@classmethod
def parse_local_import(cls, data):
from .__base__ import CACHE
stype_uuid = cached_uuid(SampleType, data["type"])
stype_mock = SampleType.from_data(CACHE[stype_uuid])
data["type"] = stype_mock
data = cls.parse_import(data)
data["cls"] = cls.__api_cls__
return data
@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:
"""
# bulk response
if "samples" in data:
return [cls.from_data(x) for x in data["samples"]]
# single response
children = cls._children
del cls._children
sample = cls.from_data(data)
if len(children):
sample.add_children(Sample.create(children, overwrite=False))
return sample
@classmethod
def ingest(cls, filename, pipeline=None, container=None, slot=None, tags=list()):
"""
Do sample ingest via sample registry.
"""
import pandas
from .analysis import Pipeline, Analysis
# resolve inputs
if isinstance(filename, (list, tuple)) and isinstance(filename[0], dict):
filename = pandas.DataFrame(filename)
if isinstance(filename, pandas.DataFrame):
from io import StringIO
df = filename
filename = "ingest.csv"
fi = StringIO()
df.to_csv(fi, sep=",", index=False)
fh = fi.getvalue().encode("utf-8")
fi.close()
elif isinstance(filename, six.string_types):
with open(filename, "rb") as fi:
fh = fi.read()
else:
raise AssertionError("Error: no rule for ingesting object of type {}".format(str(type(filename))))
# configure pipeline for ingest
if pipeline is None:
pipeline = Pipeline.create(
name="Generic Ingest",
desc="Pipeline for performing generic data ingests.",
tasks=[
dict(
name="Generic Ingest",
desc="Task for performing generic data ingests.",
cmd="{{param('apps', 'client')}} ingest --tags='{{tags}}' --container-uuid='{{container_uuid}}' --container-slot='{{container_slot}}' generic {{infile}}",
)
],
overwrite=False,
)
if not isinstance(pipeline, Pipeline):
pipeline = Pipeline(pipeline)
# configure container and slots for ingest
if container is not None:
from .container import Container
container = Container(container)
if not container.exists():
raise AssertionError("Error: specified container {} does not exist!".format(container.ident))
if slot is None:
slot = container.slots[0]
container = container.uuid
# construct and send payload
payload = {
"format": "text",
"has_header": True,
"pipeline_uuid": pipeline.uuid,
"container_uuid": container,
"container_slot": slot,
"sample_tags": tags,
}
files = {
"metadata": (
None,
json.dumps(payload),
),
"data": (
os.path.basename(filename),
fh,
),
}
try:
ctype = base.SESSION.session.headers.pop("Content-Type", None)
res = base.SESSION.post("/api/samples", files=files)
finally:
if ctype is not None:
base.SESSION.session.headers["Content-Type"] = ctype
# wait for ingest pipeline to finish
data = res.json()
if "PipelineInstances" not in data:
raise OSError(
"Pipeline failed to execute, either because of " "insufficient privileges or some other reason."
)
pool = Analysis(data["PipelineInstances"][0])
pool.join()
# TODO: search for sample by tag and return those created via ingest
return pool
@cached.tag("refresh")
def variables(self):
"""
Proxy for weird API schema requiring strange 'meta' field.
"""
# many sample queries return Sample-compatible data structures
# except that the resource_vals have already been coalesced into
# a dict.
rvals = self.resource_vals
if isinstance(rvals, dict) or (isinstance(rvals, composite) and rvals.meta_type == "dict"):
return dict(self.resource_vals)
return {x["name"]: x["value"] for x in self.resource_vals}
@cached.tag("refresh")
def variables_by_id(self):
"""
# returning a dictionary with key fixed id. If fixed id has no value, it defaults to name as key
# returns None if key not found
from esp.models import Sample
sample = Sample('ESP000009')
value = sample.variables_by_id['fixed_id_here']
"""
from esp.utils import FixedIdOrName
return FixedIdOrName(self.variables, self.entity_type.variables)
@cached
def entity_type(self):
"""
Return associated SampleType object.
"""
# Note: sample_type_uuid for an entity is _always_ a definition UUID.
# Using .from_definition drops # of backend hits from 2 (or more) to 1.
# return EntityType.from_definition(self.data.sample_type_uuid)
return EntityType.from_definition(self.data.sample_type_uuid)
@property
def sample_type(self):
return self.entity_type
@cached
def projects(self):
"""
Return projects associated with sample.
"""
res = {}
for exp in self.experiments:
if exp.project.name not in res:
res[exp.project.name] = exp.project
return list(res.values())
@cached
def experiments(self):
"""
Return Experiment objects associated with sample.
"""
from .project import Experiment
deps = self.dependencies["dependents"]
ret = []
for cat in deps:
if cat[0]["cls"] == Experiment.__api_cls__:
for dep in cat:
if not isinstance(dep, six.string_types):
ret.append(Experiment(dep["uuid"]))
return ret
@cached
def experiment_chains(self):
"""
Return all ExperimentChain objects associated with sample.
"""
from .project import ExperimentChain
chains = []
for exp in self.experiments:
res = base.SESSION.get("/api/workflow_chain_instances", params=dict(workflow_instance=exp.uuid)).json()
for chain in res:
chain = ExperimentChain(chain["uuid"], data=chain)
if chain not in chains:
chains.append(chain)
return chains
@cached
def containers(self):
"""
Return container objects associated with sample.
.. note:: This could be sped up with a backend accessor, but
this implementation will work until there is a need
for speeding it up.
"""
from .container import Container
res = []
for container in Container.all():
for slot in container.items:
items = slot.item_uuid
if not isinstance(raw(items), list):
items = [items] if items is not None else []
if self.uuid in items:
res.append(container)
break
return res
@cached
def locations(self):
"""
Return locations dictionary showing containers and slots
associated with sample.
"""
res = {c.name: [] for c in self.containers}
for container in self.containers:
for slot in container.items:
items = slot.item_uuid
if not isinstance(raw(items), list):
items = [items] if items is not None else []
if self.uuid in items:
if slot.label not in res[container.name]:
res[container.name].append(slot.label)
return res
@cached
def metadata(self):
"""
Return worksheet metadata for each sample.
"""
res = base.SESSION.get("/api/samples/{}/data".format(self.uuid))
data = res.json()
return data.get("results", {})
def metadata_df(self):
"""
Return worksheet metadata for the sample as a pandas data frame.
"""
metadata = self.metadata
rows = []
for experiment in metadata:
for protocol in metadata[experiment]:
for column in metadata[experiment][protocol]:
rows.append(
{
"Experiment": experiment,
"Protocol": protocol,
"Column": column,
"Value": metadata[experiment][protocol][column],
}
)
import pandas as pd
return pd.DataFrame(rows, columns=["Experiment", "Protocol", "Column", "Value"])
Sample = Entity