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 project data folder

execution_dir() is for scratch that a run throws away. When you want output to stay with the project and reach the cloud, write it under the project data folder instead. This is where run output that outlives the run belongs: captured images, saved API responses, and mirrored run logs land here when you ask for them.

project_data_dir

project_data_dir(subdir: str | None = None, create: bool = False) -> Path | None

Returns a Path inside your project at <project_root>/data/, or a subfolder of it when you pass subdir. Pass create=True to create the folder and its parents. Returns None when no project is bound, in which case skip the write rather than falling back somewhere else.

Unlike execution_dir(), which lives outside the project tree and is discarded, files here are part of the project. Key your entries by the run so repeated or concurrent runs do not clobber each other:

d = project_data_dir(f"api/{ExecutionInfoContext.get().execution_id}", create=True)
if d:
    (d / "reading.json").write_text(json.dumps(data))

subdir is sanitized one path segment at a time, so .. and absolute paths cannot escape the data/ folder. Every character outside letters, digits, ., _ and - becomes an underscore, so project_data_dir("my capture") gives you data/my_capture. The point is that an awkward name still saves rather than being dropped, not that it survives unchanged.

How the folder syncs

The data/ folder is pushed to the cloud automatically when a run you launched in the web app reaches a terminal state, whether it completed, failed, or you stopped it. There is no separate command to push it up. A few things to know:

• It syncs to the cloud, not to your disk. The push updates the cloud copy of the project, not the files on your computer. To pull a run's generated data down locally, run zeon sync (or zeon pull) after the run finishes.
• Kept per run. The built-in writers key their output by the run's execution id, so data/captures/<run>/, data/logs/<run>/, and data/api/<run>/ keep each run's output apart.
• It accumulates. A push preserves data already in the cloud from earlier runs, so a new run does not erase the previous run's output.
• Large files are skipped. A single file over 16 MB is left out of the push with a warning, rather than failing the whole push.
• data/runs/ is reserved. That subfolder is local-only runtime state and never syncs. Use the paths above for output you want to keep.

Three built-in functions can write here when you opt in with save_to_project=True: capture_image saves the image into data/captures/<run>/, print_log(runlog=True) mirrors the run-log line into data/logs/<run>/, and api_request writes the response into data/api/<run>/.

Two places run output can go

Those same three functions also take an export flag, and it is a different destination, not another name for the same one. Pick whichever fits, or both:

export is the newer of the two. save_to_project still works everywhere and is the only option in the cloud, so nothing you have already written needs changing.

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 is flagged done for this run, and False otherwise. An unknown well or operation reads as not done, so guarded work will run.

"Flagged done" is slightly broader than "already performed". An mm or water operation whose planned volume is zero is flagged done when the ledger is built, because there is nothing to dispense. Steps always start out not done.

• 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", save_to_project=False, export=False, 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.
• save_to_project: set True (with runlog=True) to also mirror the run-log entry into the project data folder under data/logs/<run>/, so it syncs with the project. Default False. A console-only call (no runlog) writes nothing regardless.
• export: set True to mark this run for data export. Default False.
• sep, end: joining and trailing behaviour, as with print.

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

Console output

A print_log call without runlog=True goes to the terminal, and on a skill run it also appears live in the Skills Editor's Console Logs tab. You do not have to opt in: writing print_log("...") the way you already would is enough to watch a skill think while it runs, and the output stays readable after the run ends.

Two limits are worth knowing:

• Skill runs only. A workflow run behaves exactly as before, so a plain print_log inside a workflow does not reach the run log. Use runlog=True for anything a scientist needs to see during a workflow.
• The first 2000 lines. A skill printing inside a tight loop would otherwise flood the log. After 2000 console lines one entry says the output was truncated, and the rest is dropped. The run itself is unaffected.

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