# Run your own scripts

# 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.

```python
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](https://phloem.ai/docs/inputs/push/) 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:

```python
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:

```python
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](https://phloem.ai/docs/inputs/push/).

## 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 |

> **Tip:** 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.