Skill state and logging

This page covers the plumbing a skill uses around its real work: sharing values with other skills in the same run, writing files a run produces, keeping a resume ledger for liquid handling, logging, and checking whether you are in simulation. Import them with from execution.execution_functions import *.

Passing values between skills

A node parameter does not read the previous node's output. If one skill computes a value and a later skill in the same run needs it, put it in the shared store.

set_skill_variable, get_skill_variable, clear_skill_variable

set_skill_variable(key: str, value) -> None
get_skill_variable(key: str, default=None)
clear_skill_variable(key: str) -> None

set_skill_variable stores a value under a string key. The value can be any Python type. get_skill_variable reads it back, returning default (which is None when you do not pass one) if the key was never set. clear_skill_variable removes one key, and does nothing if the key is not present.

# In an early skill:
set_skill_variable("selected_well", "A1")

# In a later skill in the same run:
well = get_skill_variable("selected_well", "A1")

Run artifacts

execution_dir

execution_dir(create: bool = False) -> Path | None

Returns the directory for this run's artifacts: the folder where files a run produces belong (parsed maps, generated JSON, your own outputs). Pass create=True to create the directory and its parents if they do not exist yet.

Returns a Path, or None when there is no current run. It returns None rather than falling back to somewhere inside your project.

d = execution_dir(create=True)
if d:
    (d / "result.json").write_text(json.dumps(data))

The liquid-handling resume ledger

These three functions let a liquid-handling run that fails partway be resumed without repeating transfers it already completed. Before each transfer you ask whether it is already done; after a successful transfer you record it. On a resumed run the recorded transfers are skipped.

The ledger is scoped to one run and stored among that run's artifacts, so it follows the same lifetime as execution_dir().

record_transfer

record_transfer(
    rxn_well: str,
    kind: str,
    *,
    index: int | None = None,
    part: str | None = None,
    vol: float | None = None,
) -> None

Marks one completed transfer in the ledger and saves it. Call it right after a successful dispense.

• rxn_well: the destination well identifier, for example "A1".
• kind: which planned operation completed. Use "mm", "water", or "step".
• index (keyword-only): for kind="step", which step in the well's plan completed. Ignored for "mm" and "water".
• part (keyword-only): the part label for a step. If omitted for a step, it is filled from the plan.
• vol (keyword-only): the dispensed volume. The units are whatever your plan and dispense functions use; the source does not state a unit, and the value is stored as given. If omitted, it is filled from the planned volume for that operation.

Returns None. This call is best-effort: it never raises back to you, because the physical transfer has already happened by the time you record it. If recording fails it is logged and swallowed.

is_transfer_done

is_transfer_done(rxn_well: str, kind: str, index: int | None = None) -> bool

Returns True if the named operation was already recorded complete for this run, and False otherwise. An unknown well or operation reads as not done, so guarded work will run.

• rxn_well: the destination well identifier.
• kind: "mm", "water", or "step".
• index: for kind="step", which step to check. An out-of-range or missing index reads as not done.

if not is_transfer_done("A1", "mm"):
    # ... command the dispense (this moves the arm) ...
    record_transfer("A1", "mm", vol=5.0)

load_liquid_state

load_liquid_state() -> dict | None

Returns the whole ledger for this run as a dict, or None when there is no run folder or no plan to seed the ledger from. On the first call in a run it builds the ledger from the run's plan; on a resumed run it returns the existing ledger unchanged, so completion flags are preserved.

The returned dict includes an execution_id, a created_at and updated_at timestamp, and a wells map keyed by well identifier. Each well entry carries its mm and water operations (each with a planned volume and a done flag), a list of steps (each with anchor, part, vol, and done), and a contents map accumulating dispensed volume by part or operation. You rarely read this directly. is_transfer_done and record_transfer are the intended interface. Reach for load_liquid_state when you need to inspect the full picture.

liquid_state_path

liquid_state_path() -> Path | None

Returns the Path to this run's ledger file, or None when there is no run. This is a location helper. The ledger contents come from load_liquid_state.

Logging

print_log

print_log(*args, runlog=False, runlog_type="transfer", sep=" ", end="\n")

Logs a message to both the run's execution log and the terminal, formatting its arguments the way print does. Use it instead of print inside a skill.

• *args: the values to log, joined by sep.
• runlog: when True, the message is also recorded in the scientist-facing run log for this run and pushed to the frontend. Default False.
• runlog_type: the entry type used when runlog=True, for example "transfer" or "event". Default "transfer". Ignored when runlog is False.
• sep, end: joining and trailing behaviour, as with print.

print_log("Computed offset:", dx, dy)
print_log("Dispensed into A1", runlog=True)

Simulation guard

is_sim_mode

is_sim_mode() -> bool

Returns True when the run is in simulation and False on real hardware. Use it to guard code that has no simulated counterpart, such as reading a physical sensor, so simulation can substitute a stand-in value.

if is_sim_mode():
    reading = 0.0  # stand-in value in simulation
else:
    reading = read_physical_sensor()

Related