Run your own scripts
Keep the test scripts you already have. A recipe step runs your script with
exec. The script publishes what it measured as channels, and the recipe judges
them. Instrument SDKs, vendor tools and SSH sessions all work as they are.
recipe("unit_accept", setup="load_test", unit="serial")
ch = channels()
reading = ch["/dut/reading"]
level = value("reading_at_rest")
require(level.below(0.5))
step("read", exec("python3 read.py --source running", timeout="60s"),
measure(level, reading))
plan(steps=["read"])
What exec does
| Command | written in full: "python3 read.py --source running" |
| Where it runs | in the script folder this device allows in Settings. A command that leaves that folder is refused when you compile |
| Interpreter | python3 from the device's environment. Point it at the environment that has your SDKs |
| Shell | none. The command is split into words: no pipes, no $(...), no ;. If you need a pipeline, put it in a script |
timeout= | optional. The device has its own default |
| Output | stdout, stderr and the exit code are saved on the run record |
The exit code decides nothing. A script can exit 0 after failing to do its job. The verdict comes from measurements taken afterwards, from an independent read of the unit.
Your script gets the run's details as environment variables:
| Variable | Holds |
|---|---|
PHLOEM_RUN_ID | the run's id. File your own log under it |
PHLOEM_UUT | the unit under test (the serial scanned at start) |
PHLOEM_FIELD_<KEY> | every run field the run started with: PHLOEM_FIELD_ITEM, PHLOEM_FIELD_BATCH, … An empty field is absent |
Read fields from the environment, not the command line. exec splits at
spaces, so "python3 cfg.py --batch " + field("batch")
breaks on a batch called Batch 7, while os.environ["PHLOEM_FIELD_BATCH"]
always holds the whole value.
Publishing readings from the script
Your script publishes channels like any input, by POSTing readings to the device's push endpoint. See Send your own data for the address, body and responses.
Set up once:
| A push input | on Inputs, add Push endpoint, codec batch, with a Channel paths prefix your script publishes under (/dut/) |
| Cadence | episodic for a script that publishes once per step. The default, continuous, treats silence as a fault, which suits a sensor stream. Left continuous, the quiet between steps makes the measurement unavailable |
Declare the same cadence in the setup that reads it:
ch.read("/dut/output_v", unit="V", cadence="episodic")
A setup arms only after it has seen every channel it reads. Your script's
channel has no data before the first step. Without cadence="episodic", Start
run stays blocked on "waiting for first data" until someone enters the value
by hand.
Declared episodic, the setup arms without it and never calls it stale. The
recipe judges it with measure() in the step that produced it. An episodic
channel can't carry checks= or an interlock, since nothing watches it between
steps. Both declarations must agree: a setup that calls a stream episodic is
refused at deploy.
A minimal reader:
import json, subprocess, time, urllib.request
def read_unit():
out = subprocess.run(["my-tool", "status"], capture_output=True, text=True).stdout
try:
return float(out.split("reading=", 1)[1].split()[0])
except (IndexError, ValueError):
return None # could not read it — publish nothing
def publish(readings):
t_us = time.time_ns() // 1000
body = [{"path": p, "samples": [{"t_us": t_us, "value": {"f64": v}}]}
for p, v in readings.items() if v is not None]
if not body:
return
req = urllib.request.Request(PUSH_URL, data=json.dumps(body).encode(),
headers={"Content-Type": "application/json"})
urllib.request.urlopen(req).read()
publish({"/dut/reading": read_unit()})
PUSH_URL is the endpoint on Send your own data.
See the shape first
Before you write a recipe for a script, run the script once and read what it
prints (its describe verb, or a --dry-run read). Probing scripts, on
Settings → Document scripts, lets the assistant do this for you. It runs a
file in the script folder once, never as a test run. While probing it can read
anywhere, write only a scratch folder, reach no network, and start only the
script or python. It is off by default; a refusal names the switch.
Rules that keep a verdict honest
| Rule | Why |
|---|---|
| If you couldn't read it, publish nothing | not a zero, not -1. The run then records unavailable, which is true. A zero is a measurement: it can pass a bad unit or fail a good one |
| Tag every value | {"f64": 0.13}, {"i64": 3}, {"bool": true}. A bare number is refused, with the reason |
| Change in one step, measure in another | step("configure", exec(...)) then step("verify", exec(read), measure(...)). A script that grades its own change reports what it believes it did |
| Name channels the script publishes | a bound on a path nothing sends compiles, then reports unavailable on every run |
Run the read script three times with a different --source: as
found, after configuring, after a power cycle. One parser then gives every
before/after comparison.
Updated