Custom Code

The PythonMethod action lets you run custom Python code when labware reaches a device.

Basic Usage

from orca.sdk.actions import PythonMethod
from orca.sdk.workflow import MethodTemplate

async def my_custom_logic(resource, inputs, outputs, options):
    # Your code here
    pass

method = MethodTemplate(
    name="custom_method",
    actions=[
        PythonMethod(
            resource=device,
            method=my_custom_logic,
            inputs=[plate],
            outputs=[plate]
        )
    ]
)

Function Signature

Your function must be async and accept exactly these four parameters:

from typing import Any, Dict, List
from orca.sdk.devices import Device
from orca.sdk.labware import LabwareInstance

async def my_method(
    resource: Device,
    inputs: List[LabwareInstance],
    outputs: List[LabwareInstance],
    options: Dict[str, Any] | None = None
) -> None:
    ...

Parameters

Parameter	Type	Description
resource	Device	The device where labware is located
inputs	List[LabwareInstance]	Labware instances arriving at the device
outputs	List[LabwareInstance]	Labware instances leaving the device
options	Dict[str, Any] | None	Custom options passed to the action

Accessing Labware

Inputs and outputs are runtime instances, not templates:

async def process_plate(resource, inputs, outputs, options):
    for labware in inputs:
        print(f"Processing: {labware.name}")
        print(f"Type: {labware.labware_type}")
        print(f"ID: {labware.id}")

Accessing PyLabRobot Objects

For plates, you can access the underlying PyLabRobot Plate object:

async def dispense_samples(resource, inputs, outputs, options):
    plate_instance = inputs[0]
    plr_plate = plate_instance.PLR()  # Returns pylabrobot.resources.Plate

    # Now use PLR plate methods
    well_a1 = plr_plate.get_well("A1")

Passing Options

Pass custom data to your function via the options parameter:

async def transfer_volume(resource, inputs, outputs, options):
    volume = options.get("volume", 100)
    source_wells = options.get("source_wells", "A1:H12")
    print(f"Transferring {volume}µL from {source_wells}")

method = MethodTemplate(
    name="transfer",
    actions=[
        PythonMethod(
            resource=liquid_handler,
            method=transfer_volume,
            inputs=[source_plate, dest_plate],
            outputs=[source_plate, dest_plate],
            options={"volume": 50, "source_wells": "A1:A12"}
        )
    ]
)

Working with the Resource

The resource parameter gives you access to the device:

async def custom_operation(resource, inputs, outputs, options):
    print(f"Running on device: {resource.name}")

    # Check device state
    if resource.is_initialized:
        # Access device driver for low-level operations
        driver = resource.driver
        await driver.some_method()

Async Operations

Since the function is async, you can use await:

import asyncio

async def timed_operation(resource, inputs, outputs, options):
    duration = options.get("duration", 5)

    print("Starting operation...")
    await asyncio.sleep(duration)
    print("Operation complete")

Error Handling

Exceptions in your function will propagate and halt the workflow:

async def validated_operation(resource, inputs, outputs, options):
    if not inputs:
        raise ValueError("No input labware provided")

    plate = inputs[0]
    if plate.labware_type != "96_well":
        raise TypeError(f"Expected 96-well plate, got {plate.labware_type}")

    # Continue with operation...

Complete Example

import asyncio
import logging
from typing import Any, Dict, List

from orca.devices.devices import LiquidHandler
from orca.sdk.labware import PlateTemplate, LabwareInstance
from orca.sdk.workflow import MethodTemplate, ThreadTemplate, WorkflowTemplate
from orca.sdk.system import SdkToSystemBuilder, WorkflowExecutor, ResourceRegistry, SystemMap
from orca.sdk.events import EventBus
from orca.sdk.actions import PythonMethod
from orca.sdk.devices import Device
from orca.resource_models.transporter import Transporter
from cheshire_drivers import SimLiquidHandlerDriver, SimTransporterDriver
from pylabrobot.resources.thermo_fisher.plates import Thermo_Nunc_96_well_plate_1300uL_Rb

logger = logging.getLogger("orca")

# Define the custom method
async def dispense_samples(
    resource: Device,
    inputs: List[LabwareInstance],
    outputs: List[LabwareInstance],
    options: Dict[str, Any] | None = None
):
    """Custom dispense logic."""
    logger.info(f"Running dispense on {resource.name}")
    logger.info(f"Processing {len(inputs)} input plate(s)")

    volume = options.get("volume", 100) if options else 100
    logger.info(f"Dispensing {volume}µL per well")

    # Access PLR plate for well information
    for plate_instance in inputs:
        plr_plate = plate_instance.PLR()
        logger.info(f"Plate {plate_instance.name} has {plr_plate.num_items} wells")

    # Simulate operation time
    await asyncio.sleep(1)
    logger.info("Dispense complete")

# Setup
plate = PlateTemplate("sample_plate", Thermo_Nunc_96_well_plate_1300uL_Rb, None)
liquid_handler = LiquidHandler("liquid_handler", SimLiquidHandlerDriver("lh"))
transporter = Transporter("arm", SimTransporterDriver("arm"), "teachpoints.json")

resources = ResourceRegistry()
resources.add_resources([liquid_handler, transporter])

system_map = SystemMap(resources)
system_map.assign_resources({"liquid_handler": liquid_handler})

# Create method with PythonMethod action
dispense_method = MethodTemplate(
    name="dispense",
    actions=[
        PythonMethod(
            resource=liquid_handler,
            method=dispense_samples,
            inputs=[plate],
            outputs=[plate],
            options={"volume": 50}
        )
    ]
)

# Thread and workflow
thread = ThreadTemplate(
    plate,
    system_map.get_location("input_stack"),
    system_map.get_location("output_stack"),
    [dispense_method]
)

workflow = WorkflowTemplate("custom_workflow")
workflow.add_thread(thread, is_start=True)

# Build and run
builder = SdkToSystemBuilder(
    "Custom Code Example", "Demo of PythonMethod",
    [plate], resources, system_map,
    [dispense_method], [workflow], EventBus()
)
system = builder.get_system()

async def run():
    executor = WorkflowExecutor(workflow, system)
    await executor.start(sim=True)

asyncio.run(run())

Use Cases

Custom liquid handling logic:

async def cherry_pick(resource, inputs, outputs, options):
    source_plate = inputs[0]
    dest_plate = inputs[1]
    well_map = options.get("well_map", {})  # {"A1": "B3", "C5": "D7"}
    # Implement cherrypicking logic...

Integration with external services:

async def update_lims(resource, inputs, outputs, options):
    plate = inputs[0]
    # Call LIMS API to update plate status
    await lims_client.update_plate_status(plate.id, "processed")

Conditional processing:

async def conditional_process(resource, inputs, outputs, options):
    plate = inputs[0]
    sample_count = options.get("sample_count", 96)

    if sample_count < 48:
        # Use different protocol for partial plates
        await resource.driver.run_partial_plate_protocol()
    else:
        await resource.driver.run_full_plate_protocol()

Next Steps

• Actions - All action types
• Methods - Grouping actions into methods
• PyLabRobot - PLR integration for labware access