import json
from pathlib import Path
from typing import Any, Dict, Type, Optional, List, Sequence, Union, Tuple, Iterable
import requests
from tqdm import tqdm
from LabGuruAPI import Stock, LGI, Plate, TheoreticalPlate
from LabGuruAPI._base import LabGuruItem, SESSION, LGStr, LGList, LGBool, LGInt, LGJSONStr, Attachment
from LabGuruAPI._collections import Collections, Plasmid, Strain
from Robofiles import CSVRobofile
[docs]
class Element(LabGuruItem):
_api_name = 'elements'
class_name = 'Element'
class_display_name = 'Experiments'
_attribute_dict = {
'element_type': 'element_type',
'data' : 'data',
'position' : 'position'
}
element_type = LGStr('')
"""equipment/sample/text/steps/plate/excel"""
data = LGJSONStr('')
"""The element's data. Will vary depending on the type"""
position = LGInt()
"""The element's position within a section"""
def __repr__(self):
return f'<Element {self.id}: {self.element_type.capitalize()}>'
def update_data(self, new_data_str: str) -> "Element":
r = SESSION.put(Element.item_api_url(self.id, True), data=new_data_str)
self.data = new_data_str
return SESSION.refresh(self)
@property
def json_data(self):
return json.loads(self.data)
@json_data.setter
def json_data(self, value):
self.data = json.dumps(value)
def duplicate(self):
new_ele = Element()
for py_att in list(self._attribute_dict.values()) + ['name', 'description', 'links']:
setattr(new_ele, py_att, getattr(self, py_att))
new_ele.other_properties.update(self.other_properties)
return new_ele
[docs]
class LGElementList(LGList[List[Element]]):
base_type = Element
def __get__(self, instance, owner) -> List[Element]:
return super().__get__(instance, owner)
class Sample(object):
def __init__(self, collection_item: Collections, container: Union[Element, "Section"],
stocks: List[Stock] = None):
self.item = collection_item
self.container = container
self._id: Optional[int] = None
self.stocks = stocks or []
def add_to_api(self, existing_data: Dict[str, Any] = None) -> Dict[str, Any]:
# Get an ID
if not self._id:
req_data = {
'token': SESSION.token,
'item' : {
"item_id" : int(self.item.id),
"item_type" : self.item.class_name,
"name" : self.item.name,
"container_type": self.container.class_name,
"container_id" : self.container.id
}
}
r = requests.post('https://my.labguru.com/api/v1/samples', json=req_data)
j = r.json()
self._id = j['id']
out_json = existing_data or {"headers": {}, "samples": []}
out_json_samples = []
for s in self.stocks:
r = requests.get('https://my.labguru.com/api/v1/sample_stocks/add_stock_by_barcode.json',
params=dict(token=SESSION.token, barcode=int(s.id), element_id=int(self.container.id)))
if 'already added' in str(r.content):
continue
out_json['headers'].update(r.json()['headers'])
out_json_samples = r.json()['samples']
out_json['samples'].extend(out_json_samples)
return out_json
@property
def id(self):
if self._id:
return self._id
else:
self.add_to_api()
return self._id
@id.setter
def id(self, value: int):
self._id = value
@staticmethod
def make_key(item: Union["Sample", Collections]) -> Tuple[int, str]:
if isinstance(item, Sample):
return int(item.item.id), item.item.class_name
elif isinstance(item, Collections):
return int(item.id), item.class_name
else:
raise TypeError(f'Cannot generate Sample Key for type {type(item).__name__}')
def stocks_data(self) -> List[Dict[str, Any]]:
return [s.to_dict() for s in self.stocks]
[docs]
class Section(LabGuruItem):
_api_name = 'sections'
class_name = 'ExperimentProcedure'
class_display_name = 'Sections'
_attribute_dict = {
'elements' : 'elements',
'section_type': 'section_type',
'collapsed' : 'collapsed',
'container_id' : 'container_id',
'container_type': 'container_type',
'data': 'data'
}
elements = LGElementList()
"""A list of elements contained in the section"""
section_type = LGStr()
collapsed = LGBool(False)
"""True if the section should be collapsed"""
container_id = LGInt()
"""ID of the container experiment/protocol"""
container_type = LGStr()
"""the type of the container - Projects::Experiment, Knowledgebase::Protocol, Knowledgebase::Report"""
data = LGJSONStr()
_samples: Dict[Tuple[int, str], Sample] = {}
_samples_element: Element = None
[docs]
@classmethod
def parse_api_data(cls: Type["Section"], json_data: Dict[str, Any], session: "Session" = SESSION,
include_custom=False) -> "Section":
out_item: Section = super().parse_api_data(json_data, session, include_custom)
for cur_element in out_item.elements:
if cur_element.element_type == 'samples':
out_item._samples_element = cur_element
break
return out_item
def duplicate(self):
new_sect = Section()
for py_att in (list(self._attribute_dict.values()) + ['name', 'description', 'links']):
if py_att == 'elements':
new_sect.elements = [e.duplicate() for e in self.elements]
else:
setattr(new_sect, py_att, getattr(self, py_att))
new_sect.other_properties.update(self.other_properties)
return new_sect
def _add_element(self, item_data: Optional[str], element_type: str, update_id: int = None,
element_wrap=False, **kwargs) -> Element:
data = {
'token': SESSION.token,
'item' : {
'container_id': int(self.id), 'container_type': self.class_name,
'element_type': element_type, 'data': item_data
}
}
data['item'].update(kwargs)
if update_id:
data.update(data['item'])
del data['item']
data['id'] = int(update_id)
if element_wrap:
r0 = requests.get(Element.item_api_url(update_id), params={'token': SESSION.token})
base_data = r0.json()
base_data['data'] = item_data
data = {'token': SESSION.token, 'element': base_data}
r = requests.put(Element.item_api_url(update_id, True), json=data)
else:
r = requests.post(Element.get_api_url(), json=data)
j = r.json()
new_element = Element.parse_api_data(j)
if not update_id:
self.elements.append(new_element)
return new_element
def update_element_data(self, e: Element) -> Element:
return self._add_element(e.data, e.element_type, e.id)
[docs]
def add_text_element(self, html: str, name=None) -> Element:
"""
Adds a "text" element to the end of the current section.
:param html: The HTML that will be displayed by the element
:param name: Optional value. Will only appear in a link.
:return: the new element
"""
return self._add_element(html, 'text', name=name)
[docs]
def add_steps_element(self, html: str) -> Element:
"""Add a "steps" element"""
return self._add_element(html, 'steps')
def _get_sample(self, item: Union[Sample, Collections], element: Element = None) -> Sample:
samp_key = Sample.make_key(item)
try:
return self._samples[samp_key]
except KeyError:
new_sample = Sample(item, element or self)
_ = new_sample.id
self._samples[Sample.make_key(new_sample)] = new_sample
return new_sample
[docs]
def add_plate_element(self, plate: Plate, add_stocks=True) -> Element:
"""
Add a plate element to a section. Actual LG plates are awful to work with, so this actually creates a "text"
element and adds the platemap via an HTML table.
:param plate: the Plate object that will be displayed as a map
:param add_stocks: if True, add the plate's stocks as samples to the section
:return: the text element
"""
# check for samples element
samp_id = self._samples_element.id if self._samples_element else ''
# add text element with plate html
link_name = f'Experiment {int(self.container_id):04d} - {self.name}'
plate_ele = self.add_text_element(plate.to_html(), name=link_name)
# link text element to the plate
SESSION.link_objects(plate, plate_ele)
# add stocks
if add_stocks:
self.add_stocks(plate.stocks)
if samp_id:
element_ids = [e.id for e in self.elements if e.id != samp_id]
sorted_ids = sorted(element_ids) + [samp_id]
r = requests.get('https://my.labguru.com/elements/sort', params=dict(token=SESSION.token, list=sorted_ids))
return plate_ele
[docs]
def add_samples_element(self, items: List[Union[Sample, Collections]]) -> Element:
"""Adds a "samples" element. Can be initialized with a list of samples to add."""
if not self._samples_element:
self._samples_element = self._add_element(None, 'samples')
for i in items:
self._get_sample(i, self._samples_element)
self._samples_element = SESSION.get_object(Element, item_id=self._samples_element.id)
return self._samples_element
[docs]
def add_stock(self, item: Stock):
"""Adds a `Stock` to the section's sample element. Will create an element if it does not exist."""
samp_ele = self._samples_element or self.add_samples_element([])
existing_sample = self._get_sample(item.stockable, samp_ele)
if item in existing_sample.stocks:
return
all_stocks = existing_sample.stocks + [item]
existing_sample.stocks = [item]
new_samp_ele_data = existing_sample.add_to_api(samp_ele.json_data)
samp_ele.json_data = new_samp_ele_data
existing_sample.stocks = all_stocks
[docs]
def add_stocks(self, items: Iterable[Stock]):
"""Adds a list of `Stock` items to the section's sample element. Will create an element if it does not exist."""
tqdm.write(f'Adding {len(items):d} stocks')
for s in tqdm(items):
self.add_stock(s)
[docs]
def add_attachments(self, *paths: Union[Path, str, CSVRobofile], att_id: int = None) -> Element:
"""
Adds one or many files to an attachment element in the section.
:param paths: Paths to files to upload (Path or str). Can also add CSVRobofile objects, which will render as a
csv
:param att_id: The ID of an existing attachment element. One will be created if this is None.
:return: the attachment element
"""
if not att_id:
att_ele = self._add_element('', 'attachments')
att_id = att_ele.id
att_id = int(att_id)
for p in paths:
if isinstance(p, CSVRobofile):
att = Attachment()
att_path = att.make_file(p.filename)
with att_path.open('w') as of:
p.format(outfile=of)
p = att_path
p = Path(p)
form_data = {
"item[attachable_type]": self.container_type,
"item[attachable_id]": int(self.container_id),
"item[attach_to_uuid]": int(self.container_id),
"item[section_id]": int(self.id),
"item[element_id]": att_id
}
r = requests.post(f'https://my.labguru.com/api/v1/attachments.json?token={SESSION.token}',
files={"item[attachment]": p.open('rb')},
data=form_data)
if r.ok:
r_data = r.json()
return SESSION.get_object(Element, item_id=att_id)
[docs]
class LGSectionList(LGList[List[Section]]):
base_type = Section
def __get__(self, instance, owner) -> List[Section]:
return super().__get__(instance, owner)
@classmethod
def new_value_function(cls, value: Any) -> Section:
if isinstance(value, dict) and 'experiment_procedure' in value:
value = value['experiment_procedure']
return super().new_value_function(value)
[docs]
class Protocol(LabGuruItem):
_api_name = 'protocols'
_attribute_dict = {
'experiment_procedures': 'sections'
}
sections = LGSectionList()
"""A list of sections contained in the protocol"""
[docs]
class Experiment(LabGuruItem):
_api_name = 'experiments'
class_name = 'Projects::Experiment'
class_display_name = 'Experiments'
_attribute_dict = {
'title' : 'name',
'experiment_procedures': 'sections'
}
sections = LGSectionList()
"""A list of Sections contained in the experiment"""
[docs]
@classmethod
def new_from_protocol(cls, title: str, protocol_id: int,
project_id: int = 18, milestone_id: int = 38) -> "Experiment":
"""
Creates a new LabGuru experiment from the specified protocol and appends the experiment ID to the beginning
of the experiment's title.
Args:
title: A string representing the title of the experiment.
protocol_id: An integer representing the ID of the protocol associated with the experiment.
project_id: An optional integer representing the ID of the project the experiment belongs to.
Default is 18 for the "Foundry" project.
milestone_id: An optional integer representing the ID of the milestone the experiment belongs to.
Default is 38 for the "Requests" folder.
Returns:
An instance of the Experiment class.
Raises:
None
Example:
experiment = Experiment.new_from_protocol(title="Experiment 1", protocol_id=1234)
"""
result_json = SESSION.post(cls._api_name, item=dict(title=title, project_id=project_id,
milestone_id=milestone_id, protocol_id=protocol_id))
result_item = cls.parse_api_data(result_json)
result_item.name = f"{int(result_item.id):04d} - {result_item.name}"
return result_item.lg_sync()
@property
def base_id(self) -> int:
"""ID of this experiment's base experiment (converts the first 4 characters of the expt name into an int)"""
try:
return int(self.name[:4])
except:
return int(self.id)
[docs]
def add_section(self, name: str, position: int = 0) -> Section:
"""
Adds a new section to the experiment.
:param name: The title of the new section
:param position: Where the new section should be located on the page. -1 will put it at the end, but above "Results"
:return: the newly created section
"""
data = {
'token': SESSION.token,
'item' : {
'name' : name,
'container_id' : self.id,
'container_type': self.class_name
}
}
r = requests.post(Section.get_api_url(), json=data)
new_section = Section.parse_api_data(r.json())
new_self = SESSION.get_object(Experiment, item_id=self.id)
self.sections = new_self.sections
return new_section
def old_main(): # trying to add samples. Epic fail.
p = SESSION.get_object(Protocol, item_id=65)
template_section = p.sections[0]
text_element = template_section.elements[0]
filled_text = text_element.format_data(source_plate_list="0645-EKO-0001", dest_plate="0645-PCR-0001",
num_cycles=30, annealing_temp=63, extension_time="1:00")
print(filled_text)
expt = SESSION.get_object(Experiment, item_id=645)
new_section = expt.add_section(f'New Wave Samples', -1)
new_section.add_text_element('SAMPLES!!!')
new_section.add_samples_element([SESSION.get_object(Strain, item_id=i) for i in [7881, 7882, 7883]])
print(expt.name)
if __name__ == '__main__':
expt = SESSION.get_object(Experiment, item_id=665)
p = SESSION.get_object(Plate, item_id=1394)
sect = expt.add_section('Linked Plate?')
sect.add_plate_element(p)
tqdm.write(sect.data)