Source code for transcriptic.jupyter.instruction

import warnings

import pandas as pd

[docs]class Instruction(object): """ An Instruction object contains information related to the current instruction such as the start, completed time as well as warps associated with the instruction. Note that Instruction objects are usually created as part of a run and not created explicitly. Additionally, if diagnostic information is available, one can click on the `Show Diagnostics Data` button to view relevant diagnostic information. Example Usage: .. code-block:: python myRun = Run('r12345') myRun.instructions # Access instruction object myRun.Instructions[1] myRun.Instructions[1].warps Attributes ---------- id : str Instruction id name: str Instruction name warps : DataFrame DataFrame of warps in the instruction started_at : str Time where instruction begun completed_at : str Time where instruction ended device_id: str Id of device which instruction was executed on attributes: dict Master attributes dictionary connection: transcriptic.config.Connection Transcriptic Connection object associated with this specific object """ def __init__(self, attributes, connection=None): """ Parameters ---------- attributes : dict Instruction attributes connection: Optional[transcriptic.config.Connection] Connection context. The default context object will be used unless explicitly provided """ self.connection = connection self.attributes = attributes = attributes["id"] = attributes["operation"]["op"] self.started_at = attributes["started_at"] self.completed_at = attributes["completed_at"] self.generated_containers = attributes["generated_containers"] if len(attributes["warps"]) > 0: device_id_set = set( [warp["device_id"] for warp in self.attributes["warps"]] ) self.device_id = device_id_set.pop() if len(device_id_set) > 1: warnings.warn( "There is more than one device involved in this instruction. Please" " contact Transcriptic for assistance." ) else: self.device_id = None self._warps = pd.DataFrame() self._warp_events = pd.DataFrame() @property def warps(self): if self._warps.empty: warp_list = self.attributes["warps"] if len(warp_list) != 0: self._warps = pd.DataFrame(x["command"] for x in warp_list) self._warps.columns = [x.title() for x in self._warps.columns.tolist()] # Rearrange columns to start with `Name` if "Name" in self._warps.columns: col_names = ["Name"] + [ col for col in self._warps.columns if col != "Name" ] self._warps = self._warps[col_names] self._warps.insert(1, "WarpId", [x["id"] for x in warp_list]) self._warps.insert( 2, "Completed", [x["reported_completed_at"] for x in warp_list] ) self._warps.insert( 3, "Started", [x["reported_started_at"] for x in warp_list] ) else: warnings.warn( "There are no warps associated with this instruction. Please " "contact Transcriptic for assistance." ) return self._warps @property def warp_events(self): """ Warp events include discrete monitoring events such as liquid sensing events for a particular instruction. """ # Note: We may consider adding special classes for specific warp # events, with more specific annotations/fields. if self._warp_events.empty: self._warp_events = self.monitoring(data_type="events") return self._warp_events
[docs] def monitoring(self, data_type="pressure", grouping=None): """ View monitoring data of a given instruction Parameters ---------- data_type: Optional[str] Monitoring data type, defaults to 'pressure' grouping: Optional[str] Determines whether the values will be grouped, defaults to None. E.g. "5:ms" Returns ------- DataFrame Returns a pandas dataframe of the monitoring data if present. Returns an empty dataframe if no data can be found due to errors. """ response = self.connection.monitoring_data(, data_type=data_type, grouping=grouping ) # Handle errors by returning empty dataframe if "error" in response: warnings.warn(response["error"]) return pd.DataFrame() res = pd.DataFrame(response["results"]) # re-order so that "name" column is always leading if "name" in res.columns: rearr_cols = ["name"] + res.columns[res.columns != "name"].tolist() return res[rearr_cols] return res
def _repr_html_(self): return """<iframe src="%s" frameborder="0" allowtransparency="true" \ style="width:450px" seamless></iframe>""" % self.connection.get_route( "view_instruction", run_id=self.attributes["run_id"], project_id=self.attributes["project_id"],, )