ot_agent_initial

korobka.opentrons.agents.ot_agent_initial

Agent designated to run the scanning procedure on the Opentrons OT-2.

Has functions for both the preparation of the reaction mix and the move of the sampler attached to the gantry.

Attributes

logger = getLogger(__name__) module-attribute

This code runs inside the OT Python environment and executes the physical operations based on the system commands in a form of transmitted files:

batch_output.json for the reaction mix preparation pos.txt for sampler moves

Classes

Opentrons(protocol)

TODO: add docstring.

TODO: add docstring.

Source code in korobka/opentrons/agents/ot_agent_initial.py

def __init__(self, protocol: protocol_api.ProtocolContext) -> None:
    """TODO: add docstring."""
    self.protocol = protocol
    # TODO: type hint for these attributes
    self.core_plates = {}  # Plates where substances are mixed
    self.support_plates = []  # Tipracks
    self.max_volumes = {}
    self.stock_amounts = defaultdict(list)  # Stock well information
    self.core_amounts = defaultdict(list)  # Core well information

    with open("error.txt", "w") as error_file:
        error_file.write("None")

    # Set gantry speeds
    for ax in ["X", "Y", "Z"]:
        self.protocol.max_speeds[ax] = 350

    # Load core plates (returns plates + fills core_amounts)
    self.load_assigned_plates("core_assigned.json", is_stock=False)

    # Load stock plates (ONLY fills stock_amounts, no plate objects)
    self.load_assigned_plates("stock_assigned.json", is_stock=True)

    self.scan_pipette = protocol.load_instrument("p300_multi_gen2", mount="right")
    self.make_pipette = protocol.load_instrument("p300_single_gen2", mount="left", tip_racks=[*self.support_plates])
    self.make_pipette.flow_rate.aspirate = 200
    self.make_pipette.flow_rate.dispense = 200
    self.make_pipette.swelled = False

Functions

batch_maker()

TODO: add docstring summary line.

For the opentron we actually want to transfer all the substances at once. This saves pipettes and operational time most importantly. We first pick uni_solv and add it to all the catalyst powders. Then we take sub_0 and add it to the vial needed. Then we add sub_1 ... Once all subs are mixed with catalyst in uni_solv, we pick alc_0 and add it to everywhere needed. Guess what happens next with alc_1 and so one!

Source code in korobka/opentrons/agents/ot_agent_initial.py

def batch_maker(
    self,
) -> tuple[int, dict[str, list[dict]], dict[str, float], list[str]]:
    """TODO: add docstring summary line.

    For the opentron we actually want to transfer all the substances at once.
    This saves pipettes and operational time most importantly.
    We first pick uni_solv and add it to all the catalyst powders. Then we take
    sub_0 and add it to the vial needed. Then we add sub_1 ... Once all subs are
    mixed with catalyst in uni_solv, we pick alc_0 and add it to everywhere needed.
    Guess what happens next with alc_1 and so one!
    """
    # TODO refactor this function to improve readability and maintainability
    # too many nested loops and conditions
    with open("batch_output.json") as batch_file:
        # generated by hitro_tools.prepare
        tasks = json.load(batch_file)
        batch = defaultdict(list)
        wells = {}

        for plate, value in tasks.items():
            for val in value:
                total_amt = 0
                for name, v in val[1].items():
                    if name == "uni_solv":
                        # uni_solv always goes to the first
                        # plate in a pair and always have first priority
                        batch[name].append({val[0]: [v, [0, 0]]})
                        total_amt += v
                    else:
                        for alias, p in val[2].items():
                            if alias == name:
                                # this tells how much should go
                                # to which well of which plate in a pair
                                # with which priority
                                batch[name].append({val[0]: [v, p]})
                                if p[0] == 0:
                                    total_amt += v
                # this calculates how much volume we have in the first plate
                # from where we transfer to the second one
                wells[val[0]] = round(total_amt, 2)

    def sort_substances(key: str) -> tuple[int, int, str]:
        """This function enforces the ordering.

        first uni_solv added, then subs, then alcs
        """
        if key == "uni_solv":
            return (0, key)
        elif key.startswith("sub_"):
            return (1, int(key.split("_")[1]), key)
        elif key.startswith("alc_"):
            return (2, int(key.split("_")[1]), key)
        #  TODO: add final else clause

    order = sorted(batch.keys(), key=sort_substances)

    """Resulting batch dict is like { 'sub_0': [ {'A1':[1.54, [1,0]]}, {'C1':[2.28, [0,0]]} ],
                                   'alc_0': [ {'D1':[3.21, [1,1]]}, {'C1':[1.23,[0,1]]} ] }
    """

    # TODO: fix typing
    return int(plate), batch, wells, order

batch_runner()

This function takes the task produced in batch_maker and executes it.

Source code in korobka/opentrons/agents/ot_agent_initial.py

def batch_runner(self) -> None:
    """This function takes the task produced in batch_maker and executes it."""
    with open("status.txt", "w") as status_file:
        status_file.write("operating")

    with open("where.txt", "w") as lag_file:
        lag_file.write(" ")

    plate, batch, wells, order = self.batch_maker()
    ts = defaultdict(list)
    for priority in [0, 1]:
        self.protocol.comment(f"Running {priority} priority cycle")
        for smth in order:
            self.protocol.comment(smth)
            # it can be that algo didn't chose some sub this time
            tip_flag = False
            for task in batch[smth]:
                for well, spec in task.items():
                    if spec[1][1] != priority:
                        continue
                    else:
                        if not tip_flag:
                            self.make_pipette.pick_up_tip()
                            tip_flag = True
                        self.protocol.comment(well)
                        amt = spec[0]
                        if spec[1][0] == 0:
                            plate_used = plate
                        else:
                            plate_used = plate + 1
                        self.transfer_execution([smth], [f"core_{plate_used}", well], amt)
                        ts[well].append({smth: datetime.now().strftime("%H:%M:%S")})
            if tip_flag:
                self.make_pipette.drop_tip()

    # now we have all prepared in both plates
    # so we mix them together

    self.protocol.comment("Running cross-mixing")
    for well, amount in wells.items():
        self.make_pipette.pick_up_tip()
        # TODO: avoid hardcoding numbers
        self.transfer_execution([f"core_{plate}", well], [f"core_{plate + 1}", well], amount - 200)
        ts[well].append({"rmx": datetime.now().strftime("%H:%M:%S")})
        self.make_pipette.drop_tip()

    with open("timings.json", "w") as file:
        json.dump(ts, file, indent=4)

    with open("status.txt", "w") as file:
        file.write("success")

    os.remove("batch_output.json")

lift_sampler(which_plate, wash_plate)

TODO: add docstring summary line.

For sampler we do not use complex functions that incorporate lifting up before going anywhere else, so we need to explicitly order it.

Source code in korobka/opentrons/agents/ot_agent_initial.py

def lift_sampler(self, which_plate, wash_plate) -> None:
    """TODO: add docstring summary line.

    For sampler we do not use complex functions that incorporate lifting up
    before going anywhere else, so we need to explicitly order it.
    """
    # this file stores the last location sampler was at
    with open("where.txt") as file:
        where_am_I = file.read()
        if where_am_I == " ":
            pass
        elif where_am_I != "wash":
            self.scan_pipette.move_to(self.core_plates[f"core_{int(which_plate) + 1}"][where_am_I].top(100))
        else:
            self.scan_pipette.move_to(wash_plate.top(100))

liquid_handling()

TODO: add docstring summary line.

Source code in korobka/opentrons/agents/ot_agent_initial.py

def liquid_handling(self) -> None:
    """TODO: add docstring summary line."""
    with open("status.txt", "w") as status_file:
        status_file.write("operating")

    try:
        with open("action.txt") as file:
            lines = [line.strip() for line in file.readlines() if line.strip()]

        if len(lines) != 3:
            raise ValueError(f"Invalid format in {'action.txt'}. Expected 3 lines, got {len(lines)}.")

        source = lines[0].split()  # Split into parts (either ["alc_1"] or ["Sylvan", "A1"])
        receiver = lines[1].split()  # Always ["Plate", "Well"]
        amount = float(lines[2])  # Parse amount

        if len(receiver) != 2:
            raise ValueError(f"Invalid receiver format in 'action.txt': {lines[1]}")

        os.remove("action.txt")  # Remove the file immediately after reading

        self.make_pipette.pick_up_tip()
        self.transfer_execution(source, receiver, amount)
        self.make_pipette.drop_tip()

        with open("state.json", "w") as file:
            json.dump(self.core_amounts, file, indent=4)

        with open("status.txt", "w") as status_file:
            status_file.write("completed")

    except Exception as e:
        with open("error.txt", "w") as f:
            f.write(str(e))

liquid_transfer(to, fr, amt)

This function universally receives from where to where move what.

Assumes that the tip is on and everything is tip-top (hah).

Source code in korobka/opentrons/agents/ot_agent_initial.py

def liquid_transfer(self, to, fr, amt: float) -> None:
    """This function universally receives from where to where move what.

    Assumes that the tip is on and everything is tip-top (hah).
    """
    # TODO: type hint for `to` and `fr`
    pipette_max_vol = 280
    amts = [pipette_max_vol for i in range(int(amt // pipette_max_vol))]
    res = amt % pipette_max_vol
    if res > 0:
        amts.append(res)
    # we have sliced the vol to discrete additions by max vols
    # e.g. 2.73ml -> [0.95, 0.95, 0.83]
    for a in amts:
        self.make_pipette.aspirate(a, fr)
        self.make_pipette.air_gap(15)
        self.make_pipette.dispense(location=to.top(z=1))
        self.make_pipette.blow_out(location=to.top(z=1))

load_assigned_plates(filename, is_stock)

Load assigned plates from JSON, ensuring missing well values are filled.

Resulting dict is like { 'sub_0': [{'position':object of opentron plates, 'amount': 5000}], 'sub_1': [{'position':object of opentron plates, 'amount': 4999}] }

Source code in korobka/opentrons/agents/ot_agent_initial.py

def load_assigned_plates(self, filename: str, is_stock: bool) -> None:
    """Load assigned plates from JSON, ensuring missing well values are filled.

    Resulting dict is like { 'sub_0': [{'position':object of opentron plates, 'amount': 5000}],
                             'sub_1': [{'position':object of opentron plates, 'amount': 4999}] }
    """
    # TODO: refactor this method to simplify the logic
    with open(filename) as assigned_file:
        assigned_data = json.load(assigned_file)

    for plate_name, plate_info in assigned_data.items():
        if plate_name.startswith("tiprack_"):
            plate = self.protocol.load_labware(plate_info["type"], location=plate_info["place"])
            plate.set_offset(x=0.7, y=1.4, z=0.2)
            self.support_plates.append(plate)
            continue

        # Load labware definition
        with open(plate_info["type"]) as labware_file:
            labware_def = json.load(labware_file)

        plate = self.protocol.load_labware_from_definition(labware_def=labware_def, location=plate_info["place"])

        if not is_stock:
            plate.set_offset(x=-1.0, y=3.5, z=0.0)

        # Get max volume for core plates
        if not is_stock and "max_volume" in plate_info:
            self.max_volumes[plate_name] = plate_info["max_volume"]

        # Ensure all wells have substance and amount values
        well_defaults = {well: {"substance": {"initial": None}, "amount": 0} for well in labware_def["wells"]}
        if plate_info.get("content"):
            for well, well_data in plate_info["content"].items():
                well_defaults[well]["amount"] = well_data["amount"]
                well_defaults[well]["substance"] = {"initial": well_data["substance"]}

        # Store well data
        if is_stock:
            for well, data in well_defaults.items():
                self.stock_amounts[data["substance"]["initial"]].append(
                    {"position": plate[well], "amount": data["amount"]}
                )
        else:
            self.core_amounts[plate_name] = well_defaults  # Stores plate data
            self.core_plates[plate_name] = plate  # Stores plate objects

monitor_signal()

This is listening function that makes our agent work all the time.

It responds to small discrete commands upon receiving respective file.

Source code in korobka/opentrons/agents/ot_agent_initial.py

def monitor_signal(self) -> None:
    """This is listening function that makes our agent work all the time.

    It responds to small discrete commands upon receiving respective file.
    """
    while True:
        if os.path.exists("pos.txt"):
            self.scan_move()

        elif os.path.exists("batch_output.json"):
            self.batch_runner()

        elif os.path.exists("action.txt"):
            self.liquid_handling()

        time.sleep(1)

receiver_validation(receiver, amt)

Ensures the receiver well does not exceed max volume after transfer.

Source code in korobka/opentrons/agents/ot_agent_initial.py

def receiver_validation(self, receiver: list[str], amt: float) -> None:
    """Ensures the receiver well does not exceed max volume after transfer."""
    plate_name, well = receiver
    max_vol = self.max_volumes.get(plate_name, float("inf"))  # Defaults to unlimited if not defined

    current_amount = self.core_amounts[plate_name][well]["amount"]
    if (current_amount + amt) > max_vol:
        raise RuntimeError(
            f"Overflow risk: {plate_name} {well} has {current_amount}μL, adding {amt}μL exceeds max {max_vol}μL."
        )

scan_move()

TODO: add docstring summary line.

This function opens the signal file from system and then moves the scan pipette to the specified location (e.g. (core_0, A1) or wash). Once all the scans are done system sends "enough" and this whole protocol is shut down.

Source code in korobka/opentrons/agents/ot_agent_initial.py

def scan_move(self) -> None:
    """TODO: add docstring summary line.

    This function opens the signal file from system and then moves the scan pipette to the
    specified location (e.g. (core_0, A1) or wash). Once all the scans are done system
    sends "enough" and this whole protocol is shut down.
    """
    self.stock_validation("wash_solv", 2800.0)
    wash_plate = self.stock_amounts["wash_solv"][0]["position"]

    with open("pos.txt") as file:
        lines = file.readlines()
        which_plate = lines[0].strip()
        position = lines[1].strip()

    os.remove("pos.txt")

    self.lift_sampler(which_plate, wash_plate)

    if position != "lift":
        # system first sends "ok now go to that spot to pull" and after a while triggers pump
        if position not in ["enough", "wash"]:
            self.scan_pipette.move_to(self.core_plates[which_plate][position].top(35))
            time.sleep(3)
            # record this position for future uplifting
            with open("where.txt", "w") as file:
                file.write(position)

        # once system reports successful scan it sends "now go to washing spot"
        elif position == "wash":
            self.scan_pipette.move_to(wash_plate.top(40))
            time.sleep(3)
            with open("where.txt", "w") as file:
                file.write(position)
            self.stock_amounts["wash_solv"][0]["amount"] -= 2800

        else:
            with open("where.txt", "w") as file:
                file.write(" ")
            self.protocol.home()
            exit()

source_validation(source, amt)

Validates that the source core well has enough volume.

Source code in korobka/opentrons/agents/ot_agent_initial.py

def source_validation(self, source: list[str], amt: float) -> None:
    """Validates that the source core well has enough volume."""
    plate_name, well = source
    try:
        well_data = self.core_amounts[plate_name][well]
    except KeyError:
        raise RuntimeError(f"Core well {plate_name} {well} not found.")

    if well_data["amount"] < amt:
        raise RuntimeError(
            f"Insufficient volume in {plate_name} {well}. Available: {well_data['amount']}, required: {amt}."
        )

stock_validation(what, amt)

This function checks that if stocks are addressed we have a sufficient amount of whatever is needed.

Source code in korobka/opentrons/agents/ot_agent_initial.py

def stock_validation(self, what: str, amt: float) -> None:
    """This function checks that if stocks are addressed we have a sufficient amount of whatever is needed."""
    min_vol = 200
    approved = False

    while not approved:
        try:
            substance = self.stock_amounts[what][0]
        except:  # NOQA  # TODO: specify the exception type
            logger.error("Stock amounts: %s", self.stock_amounts)
            raise RuntimeError(f"Substance {what} not found in deck.")

        if amt > (substance["amount"] - min_vol):
            logger.warning(
                f"Amount of {what} needed is greater than the amount in the well.\n"
                f"Well {substance['position']} is now out of scope. \n"
                f"Trying again by moving to the next well containing {what}."
            )
            # Change the well plate to move from
            self.stock_amounts[what].pop(0)
            if len(self.stock_amounts[what]) == 0:
                self.protocol.home()
                raise RuntimeError(f"No more {what} left on the deck.")
        else:
            approved = True

swell_tip(with_what)

Better wetting I guess so no drip and better precision.

Sacred knowledge inherited from an elder, more developed civilization.

Source code in korobka/opentrons/agents/ot_agent_initial.py

def swell_tip(self, with_what: str) -> None:
    """Better wetting I guess so no drip and better precision.

    Sacred knowledge inherited from an elder, more developed civilization.
    """
    spot = self.stock_amounts[with_what][0]["position"]
    self.make_pipette.aspirate(100, spot)
    time.sleep(10)
    self.make_pipette.move_to(spot.top())
    self.make_pipette.dispense(100, location=spot)
    self.make_pipette.swelled = with_what

transfer_execution(source, receiver, amt)

Handles liquid transfers between wells.

• From stock → Core well: Validates stock, updates core_amounts.
• From core well → Another core well: Validates both source & receiver, updates core_amounts.

Source code in korobka/opentrons/agents/ot_agent_initial.py

def transfer_execution(self, source: list[str], receiver: list[str], amt: float) -> None:
    """Handles liquid transfers between wells.

    - From **stock** → Core well: Validates stock, updates `core_amounts`.
    - From **core well** → Another core well: Validates both source & receiver, updates `core_amounts`.
    """
    # Receiver is always in a core plate
    rmx = self.core_plates[receiver[0]][receiver[1]]

    if len(source) == 1:
        # Stock transfer case
        what = source[0]
        self.stock_validation(what, amt)
        substance = self.stock_amounts[what][0]

        if self.make_pipette.swelled != what:
            self.swell_tip(with_what=what)

        self.receiver_validation(receiver, amt)  # Prevent overflow
        self.liquid_transfer(rmx, substance["position"], amt)

        # Update stock volume
        substance["amount"] -= amt

        # Update receiver well in core_amounts
        receiver_well = self.core_amounts[receiver[0]][receiver[1]]
        receiver_well["amount"] += amt

        # Track history with timestamp
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
        receiver_well["substance"][timestamp] = (
            what,
            substance["position"].well_name,
            amt,
        )

    elif len(source) == 2:
        # Core-to-core transfer case
        self.source_validation(source, amt)
        self.receiver_validation(receiver, amt)

        src = self.core_plates[source[0]][source[1]]
        self.liquid_transfer(rmx, src, amt)

        # Update source and receiver well volumes
        source_well = self.core_amounts[source[0]][source[1]]
        receiver_well = self.core_amounts[receiver[0]][receiver[1]]

        source_well["amount"] -= amt
        receiver_well["amount"] += amt

        # Track history with timestamp
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
        receiver_well["substance"][timestamp] = (
            *source,
            amt,
        )  # Store as (plate, well)

    else:
        raise RuntimeError(
            "Wrong transfer parameters. \n"
            "Either use a masked stock name (e.g. source=['alc_1']) \n"
            "or core coordinates (e.g. source=['core_0', 'A1'])."
        )

Functions

run(protocol)

TODO: add docstring summary line.

Source code in korobka/opentrons/agents/ot_agent_initial.py

def run(protocol: protocol_api.ProtocolContext) -> None:
    """TODO: add docstring summary line."""
    try:
        ot = Opentrons(protocol=protocol)
        ot.monitor_signal()
    except Exception as ex:
        with open("error.txt", "w") as f:
            f.write(str(ex))
        raise