import re
from abc import abstractmethod
from collections.abc import Sequence
from random import choice
import math
from typing import Type, Optional, TypeVar, List, Any, Dict, Tuple, Callable, Union
import pandas as pd
import requests
from airium import Airium
from tqdm import tqdm
from tqdm.asyncio import tqdm as aio_tqdm
import asyncio
from LabGuruAPI._base import LabGuruItem, LGI, LGInt, LGFloat, LGStr, LGBool, UNITS, SESSION, Session, HasWells, \
LGList, LGDict
import LabGuruAPI._collections as collections
from LabGuruAPI._datasets import Dataset
from LabGuruAPI._search_api import make_lg_searchable
S = TypeVar('S', bound='Storage')
[docs]
class Storage(LabGuruItem):
_api_name = 'storages'
class_name = 'System::Storage::Storage'
class_display_name = 'Storages'
_attribute_dict = {
'storage' : '_storage_dict',
'name_with_hierarchy': 'full_location',
'barcode': '_barcode'
}
_storage_dict = LGDict()
_storage: "Storage" = None
_barcode = LGStr()
full_location = LGStr(lg_name='name_with_hierarchy')
"""A string providing the full breadcrumb trail for the item's location"""
[docs]
def to_dict(self, **kwargs) -> Dict[str, Any]:
to_dict = super().to_dict(**kwargs)
if self._storage_dict:
to_dict['storage_id'] = self._storage_dict.get('id')
del to_dict['storage']
del to_dict['name_with_hierarchy']
return to_dict
@property
def storage(self):
"""The parent `Storage` object"""
if self._storage is None and 'id' in self._storage_dict:
self._storage = SESSION.get_object(Storage, self._storage_dict['id'], proxy=False)
return self._storage
@storage.setter
def storage(self, value: "Storage"):
self._storage_dict = {'name': value.name, 'url': value.api_url, 'title': value.name, 'id': value.id}
self._storage = value
[docs]
def get_boxes(self) -> List["Box"]:
"""Return a list of boxes contained in the current storage element"""
r_json = SESSION.get(f"/storages/{self.id}/storage_data", what='boxes')
return [SESSION.get_object(Box, r['id']) for r in r_json]
[docs]
class StockLocationError(ValueError):
"""
Represents a custom exception for stock location errors.
This exception is intended to handle errors specific to stock location
processing within an inventory or logistics system. It extends
the ValueError to provide more context-specific error handling.
"""
pass
[docs]
class LocationFullError(StockLocationError):
"""
Exception raised when attempting to exceed location capacity in stock management.
This exception is intended to signal that a specific stock location has reached
or exceeded its full capacity, preventing further additions of items to that
location. It is a subclass of `StockLocationError` and can be used to distinguish
capacity issues from other stock-related errors.
"""
pass
[docs]
class Stock(LabGuruItem):
_api_name = 'stocks'
_attribute_dict = {
'storage' : '_storage_dict',
'stockable' : 'stockable',
'marked_as_output' : 'marked_as_output',
'location_in_box' : 'location_in_box',
'barcode' : 'barcode',
'concentration' : '_concentration',
'concentration_unit_id': '_concentration_unit_id',
'container_type' : 'container_type',
'Buffer' : 'buffer',
'Quant Method' : 'quant_method',
'Reaction Role' : 'reaction_role'
}
xlsx_collection = 'Stock'
stockable = collections.Parent(collections.Collections)
"""The collection item held by the stock"""
marked_as_output = LGBool(False)
"""If true, the stock was generated by a LabGuru experiment"""
location_in_box = LGInt()
"""The LG position value for the well/position of the stock in the plate/box"""
barcode = LGStr('')
"""The stock's barcode. Searchable."""
container_type = LGStr('Tube')
"""A physical description of the container. Searchable."""
_concentration = LGFloat(lg_name='concentration_to_display')
_concentration_unit_id = LGInt(15, lg_name='units')
buffer = LGStr(lg_name='custom1')
"""The buffer that the sample is stored in. Searchable."""
quant_method = LGStr(lg_name='custom2')
"""The method by which the concentration was obtained. Searchable."""
reaction_role = LGStr(lg_name='custom3')
"""The role that the stockable item played in a reaction. Searchable."""
def __init__(self, name: str = '', storage: S = None, stockable: LGI = None):
super().__init__()
self.name = name
self._storage_dict = {}
self._storage: Optional[S] = None
if storage:
self.storage = storage
if stockable:
self.stockable = stockable
try:
self.name = stockable.name
except AttributeError:
self.name = stockable
self.storage_location = {}
self._ngul = None
self._molarity = None
[docs]
@classmethod
def from_barcode(cls, barcode) -> Optional["Stock"]:
"""Queries the LG API for a stock by its barcode"""
bc_stocks = SESSION.search_api(Stock, 'barcode', 'eq', str(barcode), 1)
try:
return bc_stocks[0]
except IndexError:
return None
[docs]
@classmethod
async def aio_from_barcode(cls, barcode) -> Optional["Stock"]:
"""
Asynchronously fetches a `Stock` object from the database matching the provided barcode.
Args:
barcode: The stock's barcode to search for in the database.
Returns:
Optional[Stock]: The `Stock` object matching the provided barcode if found,
or `None` if no matching stock exists.
"""
bc_stocks = await SESSION.aio_search_api(Stock, 'barcode', 'eq', str(barcode), 1)
try:
return bc_stocks[0]
except IndexError:
return None
@property
def storage(self) -> S:
if not self._storage and self._storage_dict:
s_type = Plate if 'box' in self._storage_dict['url'] else Storage
self._storage = SESSION.get_object(s_type, name=self._storage_dict['name'])
return self._storage
@storage.setter
def storage(self, value: S):
self._storage = value
self._storage_dict = {'name' : value.name, 'url': value.other_properties.get('url', ''),
'title': value.name, 'id': value.id}
storage: Storage = make_lg_searchable(storage, 'storage_location')
"""The box/plate where the `Stock` is located. Searchable."""
[docs]
def storage_box(self) -> Optional["Box"]:
"""Old method. No longer needed. :meta private:"""
if isinstance(self.storage, (Box, type(None))):
return self.storage
else:
return None
[docs]
def get_plate_position(self) -> Optional[int]:
"""Old method. No longer needed. :meta private:"""
if isinstance(self.storage, GridlessPlate) and self.description:
return self.storage.well_name_to_position(self.description)
elif isinstance(self.storage, Plate) and self.location_in_box:
return self.location_in_box
return None
[docs]
def get_plate_well(self, short=True) -> Optional[str]:
"""Old method. No longer needed. :meta private:"""
dlc = getattr(self.storage, 'description_line_count', 0)
if dlc == 0:
return None
dlc = getattr(self.storage, 'description_line_count', 0)
if dlc == 3 and bool(self.description):
return self.description
elif dlc == 2 and self.location_in_box:
return self.storage.position_to_well_name(self.location_in_box, short=short)
return None
[docs]
def get_plate_tecan_position(self) -> Optional[int]:
"""Old method. No longer needed. :meta private:"""
plate_position = self.get_plate_position()
if plate_position:
return self.storage.position_to_tecan_well(plate_position)
return None
[docs]
def update_api(self: "Stock", session: Session = SESSION, **kwargs) -> "Stock":
try_count = 1
body = {
'token': session.token,
'item' : {
'name' : self.name,
'stockable_id' : int(self.stockable.id),
'stockable_type' : self.stockable.class_name,
'description' : self.description,
'marked_as_output' : self.marked_as_output,
'barcode' : self.barcode,
'concentration' : self.concentration if self.concentration == self.concentration else None,
# NaN check
'concentration_unit_id': self.concentration_unit_id,
'Buffer' : self.buffer or None,
'Quant Method' : self.quant_method or None,
'Reaction Role' : self.reaction_role or None,
'container_type' : self.container_type
}
}
if isinstance(self.storage, Box):
body['item']['location_in_box'] = self.location_in_box
# body['item'].update(self.other_properties)
if self.storage:
body['item'].update({
'storage_id' : int(self.storage.id),
'storage_type': self.storage.class_name
})
else:
try:
del body['item']['storage_id']
del body['item']['storage_type']
except KeyError:
pass
while try_count <= 3:
if self.id:
r = requests.put(self.item_api_url(self.id), json=body)
else:
r = requests.post(self.get_api_url(), json=body)
if not r.ok:
if r.status_code == 422 and b'not located correctly' in r.content:
if try_count >= 3:
raise StockLocationError(r.content)
if try_count >= 3:
raise IOError(f'{r.status_code}\n{r.content}')
try_count += 1
else:
break
data = r.json()
new_item = self.parse_api_data(data, session)
if new_item.marked_as_output != self.marked_as_output:
if self.marked_as_output:
new_item = new_item.mark_as_output()
else:
new_item = new_item.unmark_as_output()
self.bulk_property_update(**new_item.to_dict(True))
return self
[docs]
def mark_as_consumed(self, session: "Session" = SESSION) -> "Stock":
"""
Marks the current stock item as consumed in the system using the provided session
for authentication and returns the updated stock item.
Args:
session (Session): The session object containing authentication details,
including the token needed to authenticate the API request.
Returns:
Stock: The updated stock instance after marking it as consumed.
"""
body = {
'token': session.token,
'item' : {
'id' : self.id,
'exp_id': None
}
}
r = requests.post(f'https://my.labguru.com/api/v1/stocks/{self.id}/mark_as_consumed.json', json=body)
data = r.json()
new_item = self.parse_api_data(data, session)
self.__dict__.update(new_item.__dict__)
return self
[docs]
async def aio_mark_as_consumed(self, session: "Session" = SESSION) -> "Stock":
"""
Marks the current stock item as consumed asynchronously via an API call and updates
the current object's attributes with the data received from the server.
Args:
session: Represents the session object that holds authentication and connection
details. This is used to interact with the API and must be provided.
Returns:
Stock: The updated Stock instance with attributes modified as per the server's response.
"""
body = {
'token': session.token,
'item' : {
'id' : self.id,
'exp_id': None
}
}
data = await session.aio_post(f'stocks/{self.id}/mark_as_consumed.json', **body)
new_item = self.parse_api_data(data, session)
self.__dict__.update(new_item.__dict__)
return self
[docs]
async def aio_mark_as_output(self, session: "Session" = SESSION) -> "Stock":
"""
Marks the current stock as output using the provided session.
Args:
session (Session): The session object used to authenticate the API request.
It should contain a valid token.
Returns:
Stock: The updated stock instance after it has been marked as output.
"""
body = {
'token': session.token,
'item' : {
'id' : self.id,
'exp_id': None
}
}
data = await session.aio_post(f'stocks/{self.id}/mark_as_output.json', **body)
new_item = self.parse_api_data(data, session)
self.__dict__.update(new_item.__dict__)
return self
[docs]
def mark_as_output(self) -> "Stock":
"""
Marks the stock instance as output.
Returns:
Stock: The updated stock instance after it has been marked as output.
"""
return asyncio.run(self.aio_mark_as_output())
[docs]
async def aio_unmark_as_output(self, session: "Session" = SESSION) -> "Stock":
"""
Asynchronously unmarks a stock item as output in the inventory system.
Args:
session (Session, optional): The session object containing the authentication
token and configuration needed to make API requests. Defaults to SESSION.
Returns:
Stock: The updated instance of the Stock class with the modified attributes
as per the API response data.
"""
body = {
'token': session.token,
'item' : {
'id' : self.id,
'exp_id': None
}
}
data = await session.aio_post(f'stocks/{self.id}/unmark_output.json', **body)
new_item = self.parse_api_data(data, session)
self.__dict__.update(new_item.__dict__)
return self
[docs]
def unmark_as_output(self) -> "Stock":
"""
Marks the instance as no longer being an output stock object in the system,
modifying its metadata accordingly.
Returns:
Stock: Returns an updated instance of the Stock object.
"""
return asyncio.run(self.aio_unmark_as_output())
def __repr__(self):
return f'<{self.stockable.class_display_name} Stock {self.id or ""}: {self.name}>'
@property
def concentration(self) -> float:
"""The magnitude of the stock's concentration"""
return self._concentration
@concentration.setter
def concentration(self, value: float):
self._concentration = value
self._ngul = None
self._molarity = None
@property
def concentration_unit_id(self) -> int:
"""The id of the stock's concentration unit"""
return self._concentration_unit_id
@concentration_unit_id.setter
def concentration_unit_id(self, value: int):
self._concentration_unit_id = value
self._ngul = None
self._molarity = None
@property
def concentration_unit_name(self) -> str:
"""The name of the stock's concentration unit"""
return UNITS.ids_to_units[self._concentration_unit_id]
@concentration_unit_name.setter
def concentration_unit_name(self, value: str):
self.concentration_unit_id = UNITS.names_to_ids[value]
@property
def molarity(self) -> Optional[float]:
"""
The molarity of the stock.
If the stock's concentration is recorded in a variant of molarity (nM, μM, etc.)
the value will be converted to standard mol/L. If the stock's concentration is recorded in ng/μL and it is a
:py:class:`~LabGuruAPI._collections.Weighted` object, it will also be converted to mol/L.
"""
if self._molarity is None:
if self.concentration and self.concentration_unit_id:
UNITS._get_units()
if UNITS.ids_to_units[self.concentration_unit_id][-1:] == 'M':
self._molarity = UNITS.convert(self.concentration, self.concentration_unit_id, UNITS.molar)
elif isinstance(self.stockable, collections.Weighted) and self.ng_ul:
# mol/L = ng/μL * g/ng (-9) * mol/g * μL/L (6)
self._molarity = (10.0 ** -3) * self._ngul / self.stockable.mol_weight()
return self._molarity
@molarity.setter
def molarity(self, value: float):
self.concentration = value
self.concentration_unit_id = UNITS.molar
self._molarity = value
@property
def ng_ul(self) -> Optional[float]:
"""
The stock's concentration in ng/μL.
If the stock's concentration is recorded in a variant of molarity (nM, μM, etc.) and it is a
:py:class:`~LabGuruAPI._collections.Weighted` object, it will be converted to ng/μL.
"""
if self._ngul is None:
if self.concentration and self.concentration_unit_id:
if self.concentration_unit_id == UNITS.ng_ul:
self._ngul = self.concentration
elif isinstance(self.stockable, collections.Weighted) and self.molarity:
# ng/μL = mol/L * g/mol * L/μL (-6) * ng/g (9)
return self.molarity * self.stockable.mol_weight() * 1_000
return self._ngul
@ng_ul.setter
def ng_ul(self, value: float):
self.concentration = value
self.concentration_unit_id = UNITS.ng_ul
self._ngul = value
@property
def micromolarity(self) -> Optional[float]:
"""
The micromolarity of the stock.
If the stock's concentration is recorded in a variant of molarity (nM, μM, etc.)
the value will be converted to micromolar. If the stock's concentration is recorded in ng/μL and it is a
:py:class:`~LabGuruAPI._collections.Weighted` object, it will also be converted.
"""
return UNITS.convert(self.molarity, UNITS.molar, UNITS.micromolar) if self.molarity else None
@micromolarity.setter
def micromolarity(self, value: float):
self.concentration = value
self.concentration_unit_id = UNITS.micromolar
self._molarity = value * 1e-6
@property
def nanomolarity(self) -> Optional[float]:
"""
The nanomolarity of the stock.
If the stock's concentration is recorded in a variant of molarity (nM, μM, etc.)
the value will be converted to nanomolar. If the stock's concentration is recorded in ng/μL and it is a
:py:class:`~LabGuruAPI._collections.Weighted` object, it will also be converted.
"""
return UNITS.convert(self.molarity, UNITS.molar, UNITS.nanomolar) if self.molarity else None
@nanomolarity.setter
def nanomolarity(self, value: float):
self.concentration = value
self.concentration_unit_id = UNITS.nanomolar
self._molarity = value * 1e-9
[docs]
def ul_needed(self, ratio: float = 1.0, final_nmolar: float = 2.0, total_ul: float = 10.0) -> Optional[float]:
"""
Calculates the μL of stock needed to achieve a given molarity at the final volume. It can be scaled by a given
ratio.
Args:
ratio: Scalar for the parts ratio needed in the reaction. Default = 1.0
final_nmolar: Target nanomolarity for a "standard" part in the reaction. Default = 2.0
total_ul: Target volume for the reaction in microliters. Default = 10.0
Returns:
The volume of stock to be used in microliters.
"""
if self.molarity is None:
return None
# final_nM * uL_total = part_nmolarity * uL_needed
# Volume: Round((20/[Parts]![nmolar]*1000)/2.5)*2.5 (for nL)
return (final_nmolar * total_ul) / self.nanomolarity * ratio
[docs]
def nl_needed(self, ratio: float = 1.0, final_nmolar: float = 2.0, total_ul: float = 10.0,
step: float = 2.5) -> Optional[float]:
"""
Calculates the nL of stock needed to achieve a given molarity at the final volume. It can be scaled by a given
ratio.
Args:
ratio: Scalar for the parts ratio needed in the reaction. Default = 1.0
final_nmolar: Target nanomolarity for a "standard" part in the reaction. Default = 2.0
total_ul: Target volume for the reaction in nanoliters. Default = 10.0
step: Step size for nL calculation if the dispenses are quantized (for the Echo or Mantis). Default = 2.5
Returns:
The volume of stock to be used in nanoliters.
"""
if self.molarity is None:
return None
exact_nL = self.ul_needed(ratio, final_nmolar, total_ul) * 1000
echo_nl = math.ceil(exact_nL / step) * step
return echo_nl
[docs]
def ul_for_ng(self, ng_needed: float) -> Optional[float]:
"""
The microliters needed to dispense a particular weight of sample.
Args:
ng_needed: The nanograms of sample needed for the reaction.
Returns:
The volume of sample in microliters
"""
if self.molarity is None:
return None
return float(ng_needed) / self.ng_ul
[docs]
def nl_for_ng(self, ng_needed: float, step: float = 2.5) -> Optional[float]:
"""
The nanoliters needed to dispense a particular weight of sample.
Args:
ng_needed: The nanograms of sample needed for the reaction.
step: Step size for nL calculation if the dispenses are quantized (for the Echo or Mantis). Default = 2.5
Returns:
The volume of sample in nanoliters
"""
if self.molarity is None:
return None
exact_nL = self.ul_for_ng(ng_needed) * 1000
return math.ceil(exact_nL / step) * step
@property
def concentration_string(self):
"""A human-readable string describing the concentration of the stock."""
if self.concentration:
return f"{self.concentration:0.02f} {UNITS.ids_to_units[self.concentration_unit_id]}"
return ''
@concentration_string.setter
def concentration_string(self, value: str):
if ' ' not in value:
for i, c in enumerate(value):
if c.isalpha():
value = value[:i] + ' ' + value[i:]
break
conc_str, unit_name = value.split(' ', 1)
self.concentration = float(conc_str)
self.concentration_unit_id = UNITS[unit_name]
@property
def is_rxn_product(self):
"""True if the stock represents a reaction product"""
return self.reaction_role == 'Product'
@is_rxn_product.setter
def is_rxn_product(self, value: bool):
if value:
self.reaction_role = 'Product'
elif self.is_rxn_product:
self.reaction_role = None
@property
def is_rxn_side_product(self):
"""True if the stock represents a reaction side product"""
return self.reaction_role == 'Side Product'
@is_rxn_side_product.setter
def is_rxn_side_product(self, value: bool):
if value:
self.reaction_role = 'Side Product'
elif self.is_rxn_product:
self.reaction_role = None
@property
def is_rxn_reactant(self):
"""True if the stock represents a reactant"""
return self.reaction_role == 'Reactant'
@is_rxn_reactant.setter
def is_rxn_reactant(self, value: bool):
if value:
self.reaction_role = 'Reactant'
elif self.is_rxn_reactant:
self.reaction_role = None
@property
def is_rxn_catalyst(self):
"""True if the stock represents a reaction catalyst"""
return self.reaction_role == 'Catalyst'
@is_rxn_catalyst.setter
def is_rxn_catalyst(self, value: bool):
if value:
self.reaction_role = 'Catalyst'
elif self.is_rxn_catalyst:
self.reaction_role = None
@property
def is_rxn_bystander(self):
"""True if the stock represents a reaction bystander (like a buffer)"""
return self.reaction_role == 'Bystander'
@is_rxn_bystander.setter
def is_rxn_bystander(self, value: bool):
if value:
self.reaction_role = 'Bystander'
elif self.is_rxn_bystander:
self.reaction_role = None
[docs]
class StockOptions(Sequence):
def __init__(self, stocks: List[Stock]):
self.stocks = stocks
def __getitem__(self, i: int) -> Optional[Stock]:
if self.stocks:
return self.stocks[i]
else:
return None
def __len__(self) -> int:
return len(self.stocks)
def random(self) -> Stock:
return choice(self.stocks)
def first(self) -> Stock:
return self[0]
[docs]
class LGStockList(LGList[List["Stock"]]):
base_type = Stock
def __init__(self) -> None:
super().__init__()
self.base_type = Stock
def __get__(self, instance, owner) -> List["Stock"]:
return super().__get__(instance, owner)
@classmethod
def new_value_function(cls, value: Any) -> "Stock":
if isinstance(value, dict) and 'stock' in value:
value = value['stock']
return super().new_value_function(value)
def __set__(self, instance, value):
if isinstance(value, list):
id_list = []
for item in value:
if isinstance(item, dict) and 'stock' in item:
id_list.append(item['stock']['id'])
setattr(instance, '_stock_ids', id_list)
return super().__set__(instance, value)
[docs]
class Box(HasWells, Storage):
_api_name = 'boxes'
class_name = 'System::Storage::Box'
class_display_name = 'Boxes'
xlsx_collection = 'Box'
_attribute_dict = {'rows': 'rows', 'cols': 'cols', 'shared': 'shared', 'stocks': 'stocks'}
shared = LGBool(True)
stocks = LGStockList()
def __init__(self, rows=8, cols=12):
super().__init__(rows=rows, cols=cols)
self._stock_ids = []
@classmethod
def _parse_api_data(cls: Type[LGI], json_data: Dict[str, Any], session: "Session" = SESSION,
include_custom=False) -> LGI:
return super().parse_api_data(json_data, session, include_custom)
[docs]
@classmethod
def from_api(cls: Type["Box"], session: "Session" = SESSION, item_id: int = None, name: str = None,
uuid: str = None,
api_url: str = None, auto_name: str = None, include_custom=False) -> Optional[LGI]:
if item_id:
results = SESSION.search_api(cls, 'id', 'eq', str(item_id))
if results:
return results[0]
return super().from_api(session, item_id, name, uuid, api_url, auto_name, include_custom)
[docs]
def to_dict(self, **kwargs) -> Dict[str, Any]:
self_dict = super().to_dict(**kwargs)
del self_dict['stocks']
return self_dict
[docs]
def stocks_from_position(self, position: int, well_sample_filter: Callable[[Stock], bool] = None,
**kwargs) -> List[Stock]:
well_sample_filter = well_sample_filter or (lambda x: True)
for s in self.stocks:
if s.location_in_box == position:
return [s] if well_sample_filter(s) else []
return []
[docs]
def copy_stock_to_position(self, stock: "Stock", position: int) -> Stock:
current_stock = self.stocks_from_position(position)
if current_stock:
new_stock = current_stock[0]
new_stock.stockable = stock.stockable
new_stock.name = stock.name
else:
new_stock = Stock(name=stock.name, storage=self, stockable=stock.stockable)
stock_props = stock.to_dict(False)
del stock_props['id']
new_stock.bulk_property_update(**stock_props)
new_stock.location_in_box = position
new_stock = SESSION.update(new_stock) if current_stock else SESSION.add(new_stock)
self.stocks.append(new_stock)
return new_stock
[docs]
def update_stock_at_position(self, stock: "Stock", position: int) -> Stock:
return self.copy_stock_to_position(stock, position)
[docs]
def add_sample_to_position(self, sample: LGI, position: int) -> "Stock":
"""
Adds a collection item to a position. A `Stock` will be generated automatically
:param sample: the LabGuruItem to add
:param position: the stock's position in the box
:return: the new Stock
"""
return self.copy_stock_to_position(Stock(stockable=sample), position)
[docs]
def add_sample_to_well(self, sample: LGI, well_name: str) -> "Stock":
"""
Adds a collection item to a well. A `Stock` will be generated automatically
:param sample: the LabGuruItem to add
:param well_name: the stock's position in the box
:return: the new Stock
"""
return self.add_sample_to_position(sample, self.well_name_to_position(well_name))
[docs]
def add_sample_to_tecan_position(self, sample: LGI, tecan_position: int) -> "Stock":
"""
Adds a collection item to a tecan position. A `Stock` will be generated automatically
:param sample: the LabGuruItem to add
:param tecan_position: the stock's position in the box
:return: the new Stock
"""
return self.add_sample_to_position(sample, self.tecan_well_to_position(tecan_position))
[docs]
def resovle_stocks(self, verbose=False):
"""Please use `aio_resolve_stocks` instead. It's much faster."""
if verbose:
iter_fxn = tqdm(self.stocks, desc=f"Resolving {self.name}")
else:
iter_fxn = iter(self.stocks)
for s in iter_fxn:
try:
_ = s.stockable.sequence
except AttributeError:
_ = s.stockable.id
finally:
del _
[docs]
def aio_resolve_stocks(self, verbose=True):
"""
An upfront query of all the stocks and collection items in a box. This greatly speeds up working with them later
:param verbose: if true, a progress bar will be shown for stock and item resolutions. Default: true
"""
if verbose:
gather_fxn = aio_tqdm.gather
else:
async def gather_fxn(*coros):
return await asyncio.gather(*coros)
print_fxn = aio_tqdm.write if verbose else str
future_objects = [SESSION.aio_get_object(Stock, item_id=i) for i in self._stock_ids]
print_fxn(f'Resolving Stocks in {repr(self)}')
results = asyncio.run(gather_fxn(*future_objects))
print_fxn('Resolving Stock Contents')
unique_keys = {getattr(s.stockable, '_proxy_target') for s in results if hasattr(s.stockable, '_proxy_target')}
future_objects = [SESSION.aio_get_object_from_cache_key(k) for k in unique_keys]
results2 = asyncio.run(gather_fxn(*future_objects))
return results
[docs]
class GridlessBox(Box):
[docs]
def stocks_from_well(self, well_name: str, well_sample_filter: Callable[[Stock], bool] = None,
**kwargs) -> List[Stock]:
well_name = self._force_short_name(well_name)
well_sample_filter = well_sample_filter or (lambda x: True)
return [s for s in self.stocks if s.description == well_name and well_sample_filter(s)]
[docs]
def stocks_from_position(self, position: int, well_sample_filter: Callable[[Stock], bool] = None,
**kwargs) -> List[Stock]:
return self.stocks_from_well(self.position_to_well_name(position, True), well_sample_filter, **kwargs)
[docs]
def update_stock_at_well(self, stock: Stock, well_name: str) -> Stock:
well_name = self._force_short_name(well_name)
current_stocks = [s for s in self.stocks_from_well(well_name) if s.stockable.id == stock.stockable.id]
if current_stocks:
new_stock = current_stocks[0]
else:
new_stock = Stock(name=stock.name, storage=self, stockable=stock.stockable)
new_stock.description = well_name
new_stock.concentration = stock.concentration
new_stock.concentration_unit_id = stock.concentration_unit_id
new_stock = SESSION.update(new_stock) if current_stocks else SESSION.add(new_stock)
self.stocks.append(new_stock)
return new_stock
[docs]
def update_stock_at_position(self, stock: Stock, position: int) -> Stock:
return self.update_stock_at_well(stock, self.position_to_well_name(position, True))
[docs]
def copy_stock_to_well(self, stock: Stock, well_name: str) -> Stock:
return self.update_stock_at_well(stock, well_name)
[docs]
def copy_stock_to_position(self, stock: Stock, position: int) -> Stock:
return self.copy_stock_to_well(stock, self.position_to_well_name(position, True))
def _get_dimensions_from_description(self) -> Tuple[int, int]:
m = re.search(r"(\d+) rows x (\d+) columns", self.description)
if m:
return int(m.group(1)), int(m.group(2))
else:
return 0, 0
def _add_dimensions_to_description(self):
r, c = self._get_dimensions_from_description()
if r == 0:
self.description = f'{self.description}\n{self.rows:d} rows x {self.cols:d} columns'
[docs]
@classmethod
def parse_api_data(cls: Type[LGI], json_data: Dict[str, Any], session: "Session" = SESSION,
include_custom=False) -> LGI:
new_item: GridlessBox = super()._parse_api_data(json_data, session, include_custom)
new_item.rows, new_item.cols = new_item._get_dimensions_from_description()
return new_item
[docs]
def to_dict(self, **kwargs) -> Dict[str, Any]:
self._add_dimensions_to_description()
to_dict = super().to_dict(**kwargs)
to_dict['rows'] = to_dict['cols'] = 0
return to_dict
[docs]
class PlateDescriptorMixin:
"""
Mixin class for plate descriptors to define attributes and methods used for
manipulating plate-related properties.
This class provides functionality for handling plate identification,
description parsing and updating, as well as extracting attributes like
barcode, label, and plate type. It includes an abstract method `get_type`
that must be implemented by subclasses to specify the plate type. It also
offers mechanisms for working with descriptions in a formatted manner,
allowing detailed parsing of labels and plate types from delimited strings.
Attributes:
description_delimiter (str): The delimiter used to split and join the
plate description.
description_line_count (int): The expected number of lines after
splitting the description.
name (LGStr): The name or barcode of the plate.
description (LGStr): The detailed description of the plate.
"""
description_delimiter = ' | '
description_line_count = 2
name: LGStr
description: LGStr
def __new__(cls, *args, **kwargs):
cls.name = cls.name if hasattr(cls, 'name') else LGStr('')
cls.description = cls.description if hasattr(cls, 'description') else LGStr('')
return super().__new__(cls)
def __init__(self) -> None:
super().__init__()
@property
def barcode(self) -> str:
return self.name
@barcode.setter
def barcode(self, value: str):
self.name = value
barcode: str = make_lg_searchable(barcode, 'name')
"""The plate's barcode. Searchable."""
@abstractmethod
def get_type(self) -> Type["Plate"]:
pass
def _split_description(self) -> List[str]:
if self.description_delimiter not in self.description:
self.description = self.description.replace(' / ', self.description_delimiter)
if "<p>" in self.description:
self.description = self.description.replace("</p><p>", self.description_delimiter) \
.replace("<p>", "").replace("</p>", "")
splits = self.description.split(self.description_delimiter)
if len(splits) < self.description_line_count:
self.description += self.description_delimiter
return self._split_description()
return splits
@property
def label(self) -> str:
"""The human-readable label on the plate"""
return self._split_description()[0]
@label.setter
def label(self, value: str):
desc_vals = self._split_description()
desc_vals[0] = value
self.description = self.description_delimiter.join(desc_vals)
@property
def plate_type(self) -> str:
"""The consumable name of the plate"""
return self._split_description()[1]
@plate_type.setter
def plate_type(self, value: str):
if isinstance(value, LabGuruItem):
value = value.name
desc_vals = self._split_description()
desc_vals[1] = value
self.description = self.description_delimiter.join(desc_vals)
[docs]
def make_real(self) -> "Plate":
"""Only implemented for TheoreticalPlate. Returns itself otherwise."""
assert isinstance(self, Plate)
return self
[docs]
class Plate(Box, PlateDescriptorMixin):
def __init__(self, rows=8, cols=12):
super().__init__(rows, cols)
self._dataset_item: Optional[Dataset] = None
[docs]
@classmethod
def parse_api_data(cls: Type[LGI], json_data: Dict[str, Any], session: "Session" = SESSION,
include_custom=False) -> LGI:
if json_data['rows'] == 0:
return GridlessPlate.parse_api_data(json_data, session, include_custom)
else:
return cls._parse_api_data(json_data, session, include_custom)
[docs]
def get_type(self) -> Type["Plate"]:
"""
Determines and returns the type of the current instance.
This method provides the ability to access the type of the instance it is
called on, which can be used for type comparisons or introspection.
Returns:
Type[Plate]: The specific type of the current class instance.
"""
return type(self)
def _get_description(self) -> str:
return self.description
def _set_description(self, value: str):
self.description = value
[docs]
@classmethod
def from_barcode(cls: Type[LGI], barcode: str) -> LGI:
"""Query the LG api for a plate by its barcode"""
return cls.from_name(barcode)
[docs]
@classmethod
async def async_from_barcode(cls: Type[LGI], barcode: str) -> LGI:
"""
Creates an instance of the class asynchronously using the provided barcode.
"""
return await cls.async_from_name(barcode)
def __repr__(self):
return f'<Storage Plate {self.id}: {self.name}>'
[docs]
def update_linked_elements(self):
"""Updates the HTML of all linked experiment elements"""
from LabGuruAPI import Element
data = self.to_html()
for e in self.get_linked_items(Element):
e.update_data(data)
@property
def dataframe(self) -> pd.DataFrame:
"""A dataframe containing plate samples and reaction roles"""
if self._dataset_item is None:
self._dataset_item = Dataset.make_new(name=self.name)
df = self._dataset_item.get_data()
if df is None:
df = pd.DataFrame()
return df
@classmethod
def _parse_api_data(cls: Type[LGI], json_data: Dict[str, Any], session: "Session" = SESSION,
include_custom=False) -> LGI:
new_dataplate = super().parse_api_data(json_data, session, include_custom)
for c_dataset in new_dataplate.get_linked_items(Dataset):
if c_dataset.name == new_dataplate.name:
new_dataplate._dataset_item = c_dataset
break
else:
new_dataplate._dataset_item = Dataset.make_new(name=new_dataplate.name,
description=f"Samples found in {new_dataplate.barcode}")
return new_dataplate
[docs]
def update_api(self: "Plate", session: Session, retries=0) -> LGI:
new_item = super().update_api(session, retries)
if self._dataset_item is not None:
if self._dataset_item.id:
new_item._dataset_item = SESSION.update(self._dataset_item)
else:
new_item._dataset_item = SESSION.add(self._dataset_item)
SESSION.link_objects(new_item, new_item._dataset_item)
return new_item
def _append_to_dataset(self, row_dict: dict, merge_on: List[str]):
df = self.dataframe
df = pd.concat([df, pd.DataFrame({k: [v] for k, v in row_dict.items()})])
self._dataset_item.set_data(df, merge_on)
[docs]
def add_reactant_to_well(self, item: Union[Stock, LGI], well_name: str, **stock_properties) -> Stock:
"""
Adds a reactant to a specified well and updates the dataset with the relevant information. This method is designed to
manage reactants within a biochemical or chemical workflow by either using an existing `Stock` object or creating
a new one based on input parameters. Once the reactant is added to the specified well, it updates the dataset with
the necessary details including the well name, associated stock attributes, and properties for workflow tracking.
Args:
item (Union[Stock, LGI]): The input item to be added. This can be an existing `Stock` object or another
stockable entity of type `LGI` from which a new Stock will be created.
well_name (str): The name of the well where the reactant is to be added.
**stock_properties: Optional keyword arguments used to create a new `Stock` object if `item` is not already of
type `Stock`. These properties include additional attributes related to the stock itself.
Returns:
Stock: The `Stock` object associated with the added reactant, either newly created or the existing one.
"""
if isinstance(item, Stock):
stock = item
else:
stock = Stock.make_new(stockable=item, **stock_properties)
vector_dict = {
'Well' : well_name,
stock.stockable.xlsx_collection: stock.stockable.name,
'Box' : self.barcode,
'Role' : stock.reaction_role or 'Reactant'
}
if stock.concentration_string:
vector_dict['Concentration'] = stock.concentration_string
if stock.buffer:
vector_dict['Buffer'] = stock.buffer
if stock.quant_method:
vector_dict['Quant Method'] = stock.quant_method
if stock.id:
vector_dict['Stock ID'] = stock.id
self._append_to_dataset(vector_dict, ['Well', 'Plate', stock.stockable.xlsx_collection, 'Role'])
return stock
[docs]
def add_reactant_to_position(self, item: Union[Stock, LGI], position: int, **stock_properties) -> Stock:
"""
Adds a reactant to a specified position in a well plate. This method serves as a convenience
function that maps a given position to a corresponding well name before adding the reactant
to the identified well. It supports positional mapping and additional stock properties to
customize the reactant being added to the well.
Args:
item: A reactant object, either of type Stock or LGI, that needs to be added to the specified
position in the well plate.
position: An integer specifying the target position on the well plate where the reactant will
be added.
**stock_properties: Arbitrary keyword arguments representing additional properties or
specifications for the stock to be added. These properties can be used to customize the
behavior or attributes of the stock.
Returns:
Stock: The Stock object that was added to the specified well, after the position has been
mapped and the stock properties have been applied.
"""
return self.add_reactant_to_well(item, self.position_to_well_name(position), **stock_properties)
[docs]
def add_reactant_to_tecan_position(self, item: Union[Stock, LGI], position: int, **stock_properties) -> Stock:
"""
Adds a reactant to a TECAN position with optional stock properties.
Args:
item (Stock | LGI): The reactant to be added. Can be an instance of either
a Stock or LGI class.
position (int): The TECAN position where the reactant is to be added.
**stock_properties: Additional properties to define or modify the stock.
Returns:
Stock: The created or updated Stock object after adding the reactant.
"""
return self.add_reactant_to_well(item, self.tecan_well_to_name(position), **stock_properties)
[docs]
def add_product_to_well(self, item: Union[Stock, LGI], well_name: str, **stock_properties) -> Stock:
"""
Adds a product to a specified well in the system. This method ensures that the
item is processed as a product before adding it to the corresponding well. If
the provided item is already an instance of Stock, its reaction role is updated
accordingly. Otherwise, the method creates a new Stock object from the provided
item and properties, ensuring that it is configured as a product. Once this
preparation is complete, the product is copied into the specified well.
Args:
item (Union[Stock, LGI]): The item to add as a product to the well. If the
item is an instance of Stock, it will be updated as a product. If it is
not, a new Stock object is created from it.
well_name (str): The name of the well to which the product should be added.
**stock_properties: Additional properties required for creating a new Stock
object if the provided item is not of type Stock.
Returns:
Stock: The Stock object that has been added to the specified well after being
processed or created.
"""
if isinstance(item, Stock):
item.reaction_role = 'Product'
item.id = None
else:
stock_properties['reaction_role'] = 'Product'
item = Stock.make_new(stockable=item, name=item.name, **stock_properties)
stock = self.copy_stock_to_well(item, well_name)
self.add_reactant_to_well(stock, well_name)
return stock
[docs]
def add_product_to_position(self, item: Union[Stock, LGI], position: int, **stock_properties) -> Stock:
"""
Adds a product to a specific position, assigning it additional stock properties if provided.
This method places a product, specified as either a `Stock` or `LGI` object, into a designated
position in a storage system. The item's properties are updated if additional stock attributes
are supplied.
Args:
item: The product to be added to the position, either of type `Stock` or `LGI`.
position: The numeric position where the product should be placed.
**stock_properties: Optional keyword arguments representing additional attributes or
properties to assign to the product.
Returns:
Stock: The added product represented as a `Stock` object equipped with updated properties.
"""
return self.add_product_to_well(item, self.position_to_well_name(position), **stock_properties)
[docs]
def add_product_to_tecan_position(self, item: Union[Stock, LGI], position: int, **stock_properties) -> Stock:
"""
Adds a product to the specified Tecan position on the platform.
Args:
item (Union[Stock, LGI]): The product to add, which must be of type `Stock`
or `LGI`.
position (int): The position on the Tecan platform where the product will be
added. This is identified by an integer corresponding to a well.
**stock_properties: Additional properties for the stock item, such as
metadata or parameters, which are passed as keyword arguments.
Returns:
Stock: The `Stock` instance that has been added to the Tecan position.
"""
return self.add_product_to_well(item, self.tecan_well_to_name(position), **stock_properties)
[docs]
def add_side_product_to_well(self, item: Union[Stock, LGI], well_name: str, **stock_properties) -> Stock:
"""
Adds a side product to the specified well.
Args:
item: A `Stock` object or an `LGI` object that represents the substance to be added
as a side product.
well_name: The name of the well where the side product will be added.
**stock_properties: Additional attributes for creating a new `Stock` object when
the input `item` is not an instance of `Stock`.
Returns:
Stock: The `Stock` object added to the well.
"""
if isinstance(item, Stock):
item.reaction_role = 'Side Product'
item.id = None
else:
stock_properties['reaction_role'] = 'Side Product'
item = Stock.make_new(stockable=item, name=item.name, **stock_properties)
return self.add_reactant_to_well(item, well_name)
[docs]
def add_side_product_to_position(self, item: Union[Stock, LGI], position: int, **stock_properties) -> Stock:
"""
Adds a side product to a specified position on the well plate. This method facilitates adding a
product by associating it with a specific position using its attributes and additional stock
properties provided.
Args:
item (Union[Stock, LGI]): The product to be added, which can either be of type Stock or
LGI depending on the implementation.
position (int): The index on the well plate where the product should be placed. Position
should correspond to specific plate coordinates.
**stock_properties: Arbitrary set of additional properties related to the stock, which are
passed as keyword arguments.
Returns:
Stock: The new stock instance reflecting the added product along with its updated attributes.
"""
return self.add_product_to_well(item, self.position_to_well_name(position), **stock_properties)
[docs]
def add_side_product_to_tecan_position(self, item: Union[Stock, LGI], position: int, **stock_properties) -> Stock:
"""
Adds a side product to a specified Tecan position in the well.
This method places a stock or LGI item into a specific position of a
Tecan well. Additional stock-related properties may also be specified
via keyword arguments to further define the item placement.
Args:
item: The item to be added to the Tecan position. Can be an instance
of Stock or LGI.
position: The integer position index within the Tecan well where the
item will be added.
**stock_properties: Arbitrary keyword arguments specifying additional
properties of the stock item during placement.
Returns:
Stock: The resultant stock item added to the specified position in
the Tecan well.
"""
return self.add_product_to_well(item, self.tecan_well_to_name(position), **stock_properties)
[docs]
def add_catalyst_to_well(self, item: Union[Stock, LGI], well_name: str, **stock_properties) -> Stock:
"""
Adds a catalyst to a specified well.
If the given item is an instance of the `Stock` class, it directly assigns its reaction role
to 'Catalyst' and resets its ID. Otherwise, it creates a new `Stock` object by specifying
the reaction role as 'Catalyst' along with the additional properties provided. The resulting
`Stock` object is then added to the specified well using the `add_reactant_to_well` method.
Args:
item (Union[Stock, LGI]): The item to be added as a catalyst. If it's an instance
of `Stock`, its reaction role is updated. If it's an `LGI`, a new `Stock`
instance is created using the given item and stock properties.
well_name (str): The name of the well to which the catalyst will be added.
**stock_properties: Additional properties for the `Stock` object, applicable
only when the given item is not already a `Stock`.
Returns:
Stock: The `Stock` object representing the catalyst that was added to the specified well.
"""
if isinstance(item, Stock):
item.reaction_role = 'Catalyst'
item.id = None
else:
stock_properties['reaction_role'] = 'Catalyst'
item = Stock.make_new(stockable=item, name=item.name, **stock_properties)
return self.add_reactant_to_well(item, well_name)
[docs]
def add_catalyst_to_position(self, item: Union[Stock, LGI], position: int, **stock_properties) -> Stock:
"""
Adds a catalyst item to a specified position on a well plate. The item can be of a specific stock type or a custom LGI type
and additional stock properties may be passed for flexibility in processing.
Args:
item (Union[Stock, LGI]): The catalyst item to be added, which could be a Stock or LGI object.
position (int): The numeric position on the well plate where the item is to be added, typically corresponding to
the plate's index or coordinates.
**stock_properties: Optional keyword arguments that define additional properties of the stock, which can include
details such as concentration, volume, or other chemical properties.
Returns:
Stock: Returns the stock object that has been added to the well plate.
"""
return self.add_product_to_well(item, self.position_to_well_name(position), **stock_properties)
[docs]
def add_catalyst_to_tecan_position(self, item: Union[Stock, LGI], position: int, **stock_properties) -> Stock:
"""
Adds a catalyst to a specified Tecan position.
This function adds a given catalyst `item` to a Tecan position identified by
`position`, applying any additional `stock_properties` provided.
Args:
item: The catalyst stock or LGI item to be added to the Tecan position.
position: The integer identifier for the specific Tecan well position.
**stock_properties: Additional details or configurations for the catalyst
being added.
Returns:
Stock: The resulting Stock object after the catalyst has been added.
"""
return self.add_product_to_well(item, self.tecan_well_to_name(position), **stock_properties)
[docs]
def add_bystander_to_well(self, item: Union[Stock, LGI], well_name: str, **stock_properties) -> Stock:
"""
Adds a bystander to a specified well and updates its properties.
If the provided item is of type `Stock`, its `reaction_role` is set to 'Bystander'
and its `id` is cleared. Otherwise, a new `Stock` is created with the given
properties and assigned a `reaction_role` of 'Bystander'. The item is then added
as a reactant to the specified well.
Args:
item: The item to be added as a bystander to the well. It can be either an
instance of `Stock` or `LGI`.
well_name: The name of the well where the bystander should be added.
**stock_properties: Additional properties for creating a new `Stock`
instance for non-`Stock` items.
Returns:
Stock: The updated bystander `Stock` object added to the well.
"""
if isinstance(item, Stock):
item.reaction_role = 'Bystander'
item.id = None
else:
stock_properties['reaction_role'] = 'Bystander'
item = Stock.make_new(stockable=item, name=item.name, **stock_properties)
return self.add_reactant_to_well(item, well_name)
[docs]
def add_bystander_to_position(self, item: Union[Stock, LGI], position: int, **stock_properties) -> Stock:
"""
Adds a bystander to a specified position in the stock.
This method is used to add an item to a specific position in the well plate
by converting the position to the corresponding well name. It accepts a
product item and its properties and stores them accordingly. The method
ensures integration with the existing well-based system by using a helper
method for adding the product to the corresponding well.
Args:
item (Union[Stock, LGI]): The product to add, which can be of type Stock
or LGI, representing different product types.
position (int): The position in the well plate where the product needs to
be added. The position is typically an integer index.
**stock_properties: Arbitrary keyword arguments containing additional
properties or configurations for the stock or product.
Returns:
Stock: The added stock after successfully placing it into the specified
position in the well.
"""
return self.add_product_to_well(item, self.position_to_well_name(position), **stock_properties)
[docs]
def add_bystander_to_tecan_position(self, item: Union[Stock, LGI], position: int, **stock_properties) -> Stock:
"""
Adds a bystander product, either a Stock or LGI object, to a specific Tecan well position. This function
associates the given `item` with a designated Tecan well position, defined by its numerical `position`,
and allows passing additional stock-related properties if required.
Args:
item: A Stock or LGI object representing the product to be added to the specified Tecan position.
position: An integer representing the Tecan well numeric ID that the `item` will be linked to.
**stock_properties: Arbitrary set of key-value properties related to the stock for customization or
configuration purposes.
Returns:
Stock: The Stock object after being added to the specified Tecan well position.
"""
return self.add_product_to_well(item, self.tecan_well_to_name(position), **stock_properties)
def _dataset_vector_to_stock(self, vector_data: Dict[str, Any]) -> Stock:
if 'Stock ID' in vector_data and vector_data['Stock ID'] is not None and pd.notna(vector_data['Stock ID']):
return SESSION.get_object(Stock, vector_data['Stock ID'])
relevant_data = {k: v for k, v in vector_data.items() if k not in ['Plate', 'Stock ID', 'id']}
stock_dict = {}
stock_dict['location_in_box'] = self.well_name_to_position(relevant_data.pop('Well'))
stock_dict['marked_as_output'] = relevant_data.pop('Role') == 'Product'
stock_dict['concentration_string'] = relevant_data.pop('Concentration', None)
stock_dict['buffer'] = relevant_data.pop('Buffer', None)
stock_dict['quant_method'] = relevant_data.pop('Quant Method', None)
for k, v in relevant_data.items():
if v and (k in collections.COLLECTIONS_BY_NAME):
stock_dict['stockable'] = SESSION.get_object(collections.COLLECTIONS_BY_NAME[k], name=v)
stock_dict['name'] = v
break
else:
raise ValueError(f'Could not find a sample in the dataset vector: {repr(vector_data)}')
return Stock.make_new(**stock_dict)
def _get_component_from_well(self, well_name: str, component_type: str) -> List[Stock]:
try:
reactant_rows = self.dataframe.groupby(['Well', 'Role']).get_group((well_name, component_type))
return [self._dataset_vector_to_stock(v) for v in reactant_rows.to_dict('records')]
except KeyError:
return []
[docs]
def get_reactants_from_well(self, well_name) -> List[Stock]:
"""
Get the reactants from a specified well.
Args:
well_name: The name of the well from which reactant components are to be
retrieved.
Returns:
List[Stock]: A list of Stock objects representing the reactants in the
specified well.
"""
return self._get_component_from_well(well_name, 'Reactant')
[docs]
def get_products_from_well(self, well_name: str) -> List[Stock]:
"""
Gets the list of products associated with a specified well.
This method retrieves all the products linked with the given well name by
querying relevant data. If no products are found, it fetches the stocks
associated with the well instead.
Args:
well_name: The name of the well from which products or stocks should be
retrieved.
Returns:
List[Stock]: A list containing product objects obtained from the well.
If no products are found, it returns stocks associated with the well.
"""
products = self._get_component_from_well(well_name, 'Product')
if products:
return products
else:
return self.stocks_from_well(well_name)
[docs]
def get_side_products_from_well(self, well_name: str) -> List[Stock]:
"""
Gets the side products from a specific well.
Args:
well_name: The name of the well from which the side products are to be retrieved.
Returns:
List[Stock]: A list of Stock objects representing the side products available in the
specified well.
"""
return self._get_component_from_well(well_name, 'Side Product')
[docs]
def get_catalysts_from_well(self, well_name: str) -> List[Stock]:
"""
Fetches all catalysts from a specified well.
Args:
well_name: The name of the well from which to retrieve catalyst
components.
Returns:
List[Stock]: A list of catalyst components retrieved from the
specified well.
"""
return self._get_component_from_well(well_name, 'Catalyst')
[docs]
def get_bystanders_from_well(self, well_name: str) -> List[Stock]:
"""
Retrieves bystander stocks from the specified well.
Args:
well_name: The name of the well containing the bystander stocks.
Returns:
List[Stock]: A list of Stock objects representing the bystanders
found in the specified well.
"""
return self._get_component_from_well(well_name, 'Bystander')
[docs]
def get_reactants_from_position(self, position) -> List[Stock]:
"""
Retrieves the reactants from a given position.
Args:
position: The position to be converted to a well name.
Returns:
List[Stock]: A list of Stock objects representing the reactants.
"""
well_name = self.position_to_well_name(position)
return self.get_reactants_from_well(well_name)
[docs]
def get_reactants_from_tecan_position(self, tecan_position) -> List[Stock]:
"""
Retrieves the reactants from a given Tecan position.
Args:
tecan_position: The Tecan position to be converted to a well name.
Returns:
List[Stock]: A list of Stock objects representing the reactants.
"""
well_name = self.tecan_well_to_name(tecan_position)
return self.get_reactants_from_well(well_name)
[docs]
def get_products_from_position(self, position) -> List[Stock]:
"""
Retrieves the products from a given position.
Args:
position: The position to be converted to a well name.
Returns:
List[Stock]: A list of Stock objects representing the products.
"""
well_name = self.position_to_well_name(position)
return self.get_products_from_well(well_name)
[docs]
def get_products_from_tecan_position(self, tecan_position) -> List[Stock]:
"""
Retrieves the products from a given Tecan position.
Args:
tecan_position: The Tecan position to be converted to a well name.
Returns:
List[Stock]: A list of Stock objects representing the products.
"""
well_name = self.position_to_well_name(tecan_position)
return self.get_products_from_well(well_name)
[docs]
def get_side_products_from_position(self, position) -> List[Stock]:
"""
Retrieves the side products from a given position.
Args:
position: The position to be converted to a well name.
Returns:
List[Stock]: A list of Stock objects representing the side products.
"""
well_name = self.position_to_well_name(position)
return self.get_side_products_from_well(well_name)
[docs]
def get_side_products_from_tecan_position(self, tecan_position) -> List[Stock]:
"""
Retrieves the side products from a given Tecan position.
Args:
tecan_position: The Tecan position to be converted to a well name.
Returns:
List[Stock]: A list of Stock objects representing the side products.
"""
well_name = self.tecan_well_to_name(tecan_position)
return self.get_side_products_from_well(well_name)
[docs]
def get_catalysts_from_position(self, position) -> List[Stock]:
"""
Retrieves the catalysts from a given position.
Args:
position: The position to be converted to a well name.
Returns:
List[Stock]: A list of Stock objects representing the catalysts.
"""
well_name = self.position_to_well_name(position)
return self.get_catalysts_from_well(well_name)
[docs]
def get_catalysts_from_tecan_position(self, tecan_position) -> List[Stock]:
"""
Retrieves the catalysts from a given Tecan position.
Args:
tecan_position: The Tecan position to be converted to a well name.
Returns:
List[Stock]: A list of Stock objects representing the catalysts.
"""
well_name = self.tecan_well_to_name(tecan_position)
return self.get_catalysts_from_well(well_name)
[docs]
def get_bystanders_from_position(self, position) -> List[Stock]:
"""
Retrieves the bystanders from a given position.
Args:
position: The position to be converted to a well name.
Returns:
List[Stock]: A list of Stock objects representing the bystanders.
"""
well_name = self.position_to_well_name(position)
return self.get_bystanders_from_well(well_name)
[docs]
def get_bystanders_from_tecan_position(self, tecan_position) -> List[Stock]:
"""
Retrieves the bystanders from a given Tecan position.
Args:
tecan_position: The Tecan position to be converted to a well name.
Returns:
List[Stock]: A list of Stock objects representing the bystanders.
"""
well_name = self.tecan_well_to_name(tecan_position)
return self.get_bystanders_from_well(well_name)
[docs]
def to_html(self):
"""Renders a platemap as an HTML table"""
# tqdm.write(f'Resolving stocks in Plate {self.barcode}')
self.aio_resolve_stocks(True)
tqdm.write(f'Generating HTML for Plate {self.barcode}')
a = Airium()
with a.div():
link = self.other_properties.get('url', None)
if link:
link = f'https://my.labguru.com{link}'
with a.h3(_t=self.barcode):
a.a(href=link, target='_blank', style='text-decoration:none') \
.i(klass="fa fa-link", aria_hidden='true')
else:
a.h3(_t=self.barcode)
a.p(_t=self.description)
with a.table():
with a.thead(style='vertical-align:top;white-space:nowrap;text-align:center').tr():
for c in range(self.cols + 1):
a.th(_t=c or '', style='text-align:center')
with a.tbody(style='vertical-align:top;white-space:nowrap;text-align:center'):
for r in 'ABCDEFGHIJKLMNOPQRSTUVWXYZ'[:self.rows]:
with a.tr():
a.th(_t=r, style='text-align:center;vertical-align:middle')
for c in range(self.cols):
with a.td():
cur_well = f'{r}{c + 1:d}'
cur_stock: Stock
for cur_stock in self.get_reactants_from_well(cur_well) + \
self.get_products_from_well(cur_well):
text = cur_stock.name
if cur_stock.concentration:
text += f'</br>({cur_stock.concentration:0.2f} ' \
f'{UNITS.ids_to_units[cur_stock.concentration_unit_id]})'
link = cur_stock.stockable.web_url
if link:
# link = f'https://my.labguru.com{link}'
a.p(style='border:#ddd;border-style:solid none') \
.a(href=link, target='_blank', _t=text, style='font-size:10px',
title=cur_stock.stockable.description)
else:
a.p(_t=text, style='font-size:10px;border:#ddd;border-style:solid none')
return str(a)
[docs]
class GridlessPlate(GridlessBox, Plate):
description_line_count = 3
def _get_dimensions_from_description(self) -> Tuple[int, int]:
m = re.search(r"(\d+) rows x (\d+) columns", self._split_description()[2])
if m:
return int(m.group(1)), int(m.group(2))
else:
return 0, 0
def _add_dimensions_to_description(self):
desc_vals = self._split_description()
desc_vals[2] = f'{self.rows:d} rows x {self.cols:d} columns'
self.description = self.description_delimiter.join(desc_vals)
def __repr__(self):
return f'<Reaction Plate {self.id}: {self.name}>'
[docs]
@classmethod
def make_new(cls: Type[LGI], overwrite=False, **properties) -> LGI:
# if 'barcode' in properties:
# properties['name'] = properties['barcode']
# del properties['barcode']
new_item: GridlessPlate = super().make_new(overwrite=overwrite, **properties)
new_item.barcode = new_item.name
new_item._add_dimensions_to_description()
return new_item
def convert_to_reactionplate(plate: Plate) -> GridlessPlate:
# add wells to stock descriptions
stocks_to_update = []
for cur_well in plate.iter_well_names(short=True):
cur_stock: Stock
for cur_stock in plate.stocks_from_well(cur_well):
cur_stock.description = cur_well
cur_stock.location_in_box = None
cur_stock.container_type = 'Other'
stocks_to_update.append(cur_stock)
# convert box to gridless
plate.description = GridlessPlate.description_delimiter.join([plate.label or 'No Label',
plate.plate_type or 'Unknown Type',
f'{plate.rows:d} rows x {plate.cols:d} columns'])
plate.rows = 0
plate.cols = 0
new_plate = SESSION.update(plate)
list(map(SESSION.update, stocks_to_update))
return new_plate
[docs]
class TheoreticalPlate(HasWells, PlateDescriptorMixin):
[docs]
def __init__(self, barcode='', label='', plate_type='', rows=8, cols=12):
"""A plate that can be manipulated without waiting for API calls. When done, update all plate stocks in
parallel using the ``make_real()`` method."""
super().__init__(rows, cols)
self.barcode = barcode
self.label = label
self.plate_type = plate_type
self.stocks: Dict[str, List[Stock]] = {w: [] for w in self.iter_well_names(short=True)}
[docs]
def stocks_from_position(self, position: int, **kwargs) -> List[Stock]:
return self.stocks[self.position_to_well_name(position, short=True)]
[docs]
def copy_stock_to_position(self, stock: Stock, position: int) -> Stock:
well_name = self.position_to_well_name(position, short=True)
for s in self.stocks_from_position(position):
if s.stockable == stock.stockable:
self.stocks[well_name].remove(s)
break
self.stocks[well_name].append(stock)
return stock
[docs]
def update_stock_at_position(self, stock: Stock, position: int) -> Stock:
return self.copy_stock_to_position(stock, position)
def get_type(self) -> Type[Plate]:
if any(len(stock_list) > 1 for stock_list in self.stocks.values()):
return GridlessPlate
return Plate
[docs]
def make_real(self) -> Plate:
"""Converts a ``TheoreticalPlate`` into a ``Plate`` by adding the stocks to LG in parallel."""
plate_type = self.get_type()
tqdm.write(f'\nGetting/Making {self.barcode}')
real_plate = SESSION.get_object(plate_type, name=self.barcode)
if not real_plate:
real_plate = plate_type.make_new(barcode=self.barcode, label=self.label, plate_type=self.plate_type,
rows=self.rows, cols=self.cols)
real_plate = SESSION.add(real_plate)
async def add_to_well(stock: Stock, well_name: str):
await asyncio.sleep(0.1)
return await asyncio.get_event_loop().run_in_executor(None, real_plate.copy_stock_to_well, stock, well_name)
tqdm.write(f'\nAdding stocks to {self.barcode}')
futures = []
last_s = Stock()
last_well = ''
for cur_well, stock_list in self.stocks.items():
for s in stock_list:
futures.append(add_to_well(s, cur_well))
last_s, last_well = s, cur_well
asyncio.run(aio_tqdm.gather(*futures))
if last_well:
real_plate.copy_stock_to_well(last_s, last_well)
return real_plate
if __name__ == '__main__':
dplate = SESSION.get_object(Plate, 2357)
print(repr(dplate))
print(dplate.dataframe.head())