Processes
Laboratory workflows are expressed as processes and use the PythonLab package to express the steps. A process defines the labware that is used and what the interactions should be. pythonLab processes are denoted in a python like syntax, but they are not directly executed by a python interpreter. They are rather parsed into a workflow graph, which can be used by a Scheduler to calculate an optimal schedule (=order of execution). This order of execution might be different from the initial notation. An Orchestrator executes then the schedule and supervises the device communication, e.g. to SiLA servers/devices.
Basic process
If you are planning to define multiple processes within the same lab it is advised to start with a basic process that subsequent processes can inherit from. This is the basis that other processes can build on.
from abc import ABC
from pythonlab.resources.services.moving import MoverServiceResource
from pythonlab.resources.services.human import HumanServiceResource
from pythonlab.resources.services.hello_world import GreeterServiceResource
from pythonlab.resources.services.labware_storage import LabwareStorageResource
from pythonlab.resource import LabwareResource
# TODO: add whatever resources you need
from pythonlab.process import PLProcess
class BasicProcess(PLProcess, ABC):
def __init__(
self, process_name: str, num_plates: int = 0, priority=7
): # 0 has highest priority
self.num_mw_plates = num_plates
self.name = process_name
super().__init__(priority=priority)
def create_resources(self):
# the device names should match the ones in the platform_config
self.hotel1 = LabwareStorageResource(proc=self, name="Hotel1")
self.hotel2 = LabwareStorageResource(proc=self, name="Hotel2")
self.hotel3 = LabwareStorageResource(proc=self, name="Hotel3")
self.robot_arm = MoverServiceResource(proc=self, name="GenericArm")
self.human = HumanServiceResource(proc=self, name="Human")
self.greeter = GreeterServiceResource(proc=self, name="Greeter")
# TODO: add your resources here with names matching the platform_config
# the continers are automatically named/enumerated. You can change the naming without causing problems
self.containers = [
LabwareResource(
proc=self, name=f"{self.name}_cont_{cont}", lidded=True, filled=False
)
for cont in range(self.num_mw_plates)
]
def process(self):
raise NotImplementedError
In this case, BasicProcess defines a number of resources that are present in this setup. There are 3 hotels for
storage, a robot arm to move plates, a human being and a "greeter" entity. It also initialized an arbitrary amount of
plates. The names in your BasicProcess need to match with the names defined in your platform_config.
Simple one-step process
The process GreeterTest defines a simple process that only includes one step.
The process function implements the actual steps that should be executed. In this case, it is only one call to the
Greeter SiLA server
"""
Duplicate this file and add/modify the missing parts to create new processes
"""
from lab_adaption.processes.basic_process import BasicProcess
class GreeterTest(BasicProcess):
def __init__(self):
super().__init__(num_plates=1, process_name="GreeterTest")
def init_service_resources(self):
# setting start position of containers
super().init_service_resources()
for i, cont in enumerate(self.containers):
cont.set_start_position(self.hotel1, i)
def process(self):
self.greeter.wave(self.containers[0])
A more elaborate process
InterestingExample is a more elaborate process with multiple interactions.
"""
Duplicate this file and add/modify the missing parts to create new processes
"""
from lab_adaption.processes.basic_process import BasicProcess
class InterestingExample(BasicProcess):
def __init__(self):
super().__init__(priority=3, num_plates=3, process_name="InterestingExample")
def init_service_resources(self):
# setting start position of containers
super().init_service_resources()
for i, cont in enumerate(self.containers):
cont.set_start_position(self.hotel1, i)
def judge_answer(self, answer) -> bool:
# extract the number from the response
number = answer.Response.value
# return whether the number is even
return number % 2 == 0
def process(self):
# loop through all containers
for cont in self.containers:
# move all containers to hotel2 and read their barcodes
self.robot_arm.move(cont, self.hotel2, read_barcode=True)
# move all containers to the human for inspection (it can hold up to two)
self.robot_arm.move(cont, self.human)
# have the human assign a number to each container
answer = self.human.request_number(
cont, message=f"assign number to {cont.name}!"
)
# do some computation on the result
judgement = self.judge_answer(answer)
# depending on the result, put the labware in hotel1 or hotel 3
if judgement:
self.robot_arm.move(cont, self.hotel3)
else:
self.robot_arm.move(cont, self.hotel1)
One by one, it reads the barcodes of the containers and holds it up for a human being to inspect. Based on their answer, the container is moved either to hotel 1 or 3.
Inspecting processes
Pythonlab converts process descriptions to a graph-based workflow. Quickstart shows how you can access the UI to inspect the workflow graph.
More Examples
Below are some more condensed examples on how to describe workflows in pythonLab
Simple linear workflow
protocol_path = "protocols/evacuation_speroids.lhc"
self.robot_arm.move(cont, self.dispenser)
self.dispenser.run_protocol(labware=cont, protocol=protocol_path)
This process is parsed into the following workflow graph:
Implicit movements for reagents and multi labware steps
bravo_positions = [4, 6, 7, 8]
for i in range(len(growth_plates)):
self.robot_arm.move(cont, target_loc=self.pipetter, lidded=False, position=bravo_positions[i])
self.pipetter.executeProtocol(growth_plates[0], protocol=sup_rem_protocol, duration=200,
reagents=growth_plates[1:],
self.pipetter.executeProtocol(growth_plates, protocol=lysis_protocol, duration=240,
reagents=[self.lysis_buffer],
reagent_pos=[3])
for cont in growth_plates:
self.robot_arm.move(cont, self.incubator2, lidded=True)
The reagent plates in reagents=[...] are moved to and from the liquid handler implicitly:
Simple for-loop
for plate in self.target_plates:
self.robot_arm.move(plate, self.echo, role="destination", read_barcode=True, lidded=False)
self.echo.execute_transfer_protocol(self.source_plate, plate, protocol)
self.robot_arm.move(plate, self.hotel2, lidded=True)
This process is parsed into the following workflow graph:
Single conditional
def is_acceptable(answer) -> bool:
return answer.Response.value >= 5
cont = self.containers[0]
self.robot_arm.move(cont, self.human)
answer = self.human.request_number(cont, message="assign quality score")
acceptable = self.is_acceptable(answer)
if acceptable:
self.robot_arm.move(cont, self.incubator1)
else:
self.robot_arm.move(cont, self.hotel1)
This process is parsed into the following workflow graph:
Nested for-loop and timing comstraints
meas_time = 65
meas_points = [0, 5, 15, 30]
for cont in self.containers:
self.robot_arm.move(cont, self.reader)
self.reader.single_read(cont, method="protocol", label=f"read_{cont.name}_{0}")
for i in range(1, len(meas_points)):
self.robot_arm.move(cont, self.incubator1)
self.incubator1.incubate(cont, duration=10, temperature=295, shaking_frequency=400)
self.robot_arm.move(cont, self.reader)
wait = 60*(meas_points[i] - meas_points[i-1]) - meas_time
self.reader.single_read(cont, method="protocol", label=f"read_{cont.name}_{i}",
relations=[("min_wait", f"read_{cont.name}_{i-1}", [wait]),
("max_wait", f"read_{cont.name}_{i-1}", [wait+20])
]
)
self.robot_arm.move(cont, self.hotel)
This process is parsed into the following workflow graph:
Nested for-loop with conditional and break
# cultured plates in incubator 37°, 200rpm, 45-90 min until aver_OD is >= 0.4.
for cont in cultured_plates:
for j in range(3):
self.robot_arm.move(cont, target_loc=self.reader2, lidded=False)
od = self.reader2.single_read(cont, method=od_600)
if j < max_cult_intervals - 1:
aver_od = self.compute_average(od)
if aver_od > 0.4:
break
else:
# incubate for another interval
self.robot_arm.move(cont, target_loc=self.incubator2, lidded=True)
self.incubator2.incubate(cont, duration=cult_time_interval, temperature=cult_temp,
shaking_frequency=cult_shaking_freq)
self.robot_arm.move(cont, target_loc=self.pipetter, lidded=False, position=3)
This process is parsed into the following workflow graph:
