diff --git a/.gitignore b/.gitignore index 87d03c1..0bb4e46 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,4 @@ greengrass-build/ *.avro /data/ /out/ +/logs/ diff --git a/docs/README.md b/docs/README.md index 1610c36..73ccaf7 100644 --- a/docs/README.md +++ b/docs/README.md @@ -10,7 +10,8 @@ wherever you deploy it — as a Greengrass v2 component, a standalone process, o | Doc | Start here when you want to… | |-----|------------------------------| | **[Tutorial](tutorial.md)** | learn by doing — bring the processor up against a local broker and watch it downsample and archive telemetry, end to end | -| **[How-to guides](how-to-guides.md)** | accomplish a specific task — filter, downsample, window-aggregate, archive to Parquet, forward alarms northbound, deploy | +| **[How-to guides](how-to-guides.md)** | accomplish a specific task — filter, downsample, window-aggregate, handle array signals, script, archive to Parquet, forward alarms northbound, deploy | +| **[Scripting](scripting.md)** | write Rhai `filter`/`script` logic — the scope (incl. runtime context), return semantics, a language primer, array handling, and a worked cookbook | | **[Reference](reference/)** | look up an exact option, topic, payload, or column type | | **[Explanation](explanation.md)** | understand how it works and why — the route/worker model, the processing-and-timing pipeline, targets and the file sink | @@ -21,6 +22,8 @@ wherever you deploy it — as a Greengrass v2 component, a standalone process, o - **"What message do I subscribe to / publish, and on which topic?"** → [Reference — Messaging Interface](reference/messaging-interface.md). - **"How are values stored in the Parquet / AVRO files?"** → [Reference — Data Types](reference/data-types.md). - **"How do I downsample a fast signal or aggregate per signal?"** → [How-to guides](how-to-guides.md). +- **"How do I write a script, and what can it see/return?"** → [Scripting](scripting.md). +- **"My signal's value is an array — how do I handle it?"** → [Handle array-valued signals](how-to-guides.md#handle-array-valued-signals). - **"How do I control file rotation — by size, time, or count?"** → [How-to guides](how-to-guides.md). - **"Why is my data too fast / slow / laggy, and how does windowing work?"** → [Explanation — the processing-and-timing pipeline](explanation.md). - **"How does one route share a topic with another?"** → [Explanation — one route, one worker](explanation.md). diff --git a/docs/explanation.md b/docs/explanation.md index dd0be90..34931d5 100644 --- a/docs/explanation.md +++ b/docs/explanation.md @@ -80,90 +80,31 @@ caught at startup; runtime errors) drops the message rather than crashing the ro The built-in stages cover the common shapes — quality gating, value thresholds, downsampling, windowed reduction, whitelisting. **Scripting is the escape hatch** for logic they don't express: -a derived field, a conditional drop, a reshape into a bespoke body, a multi-condition predicate. -[Rhai](https://rhai.rs) is a small, sandboxed expression language embedded in the processor; you -reach for it only when a built-in won't do, because the built-ins are faster and need no parsing. +a derived engineering unit, a conditional drop, a reshape of a bespoke payload, a predicate that +spans several samples or inspects an array. [Rhai](https://rhai.rs) is a small, sandboxed, +Rust-native language embedded in the processor; a script is compiled **once at startup** and then run +per message, with no I/O and a bounded op budget, so it can shape data but can't reach outside the +pipeline. -Scripting appears in **two places**, both backed by the same engine and the same message view: +Scripting appears in **two roles**, both backed by the same engine and the same scope: -- a **`filter` `script`** — a predicate that evaluates to a **boolean**; `true` keeps the message. +- a **`filter` `script`** — a predicate that evaluates to a **boolean**; `true` keeps the message + (it fails *closed* — an error or non-boolean drops). - a **`script` stage** — a transform that evaluates to the **new body** (a map), or to `()` to **drop** the message. -### Where the source lives — inline or a file - -A script's source is given **inline** (`{"script": ""}`) or read from an external **file** -(`{"script": {"file": "rules/derive.rhai"}}`). A relative file path resolves against -`component.global.defaults.scriptsDir` (default: the working directory); an absolute path is used -as-is. Both forms are **compiled once at startup** — a missing file or a syntax error fails the -component immediately, before any telemetry flows, rather than at the first message. Inline is for a -one-liner; once a script grows past a line or two, move it to a file (it version-controls cleanly, -needs no JSON string-escaping, and ships as a deployment artifact — see [the how-to](how-to-guides.md#use-an-external-script-file)). - -### The scope a script sees - -Before each evaluation the stage builds a fresh **scope** from the current message and binds these -variables (the same set for a `filter` predicate and a `script` transform): - -| Binding | Type | What it is | -|---|---|---| -| `topic` | string | the source topic the message arrived on | -| `body` | map | the full message body (`body.signal`, `body.samples`, `body.device`, …) | -| `tags` | map | the message-**envelope** `tags` metadata (`tags.site`, `tags.thing`, …) — **not** the signal | -| `samples` | array | `body.samples` (or `[]` if absent) — each element a map with `value`, `quality`, … | -| `value` | any | a convenience: the **first** sample's `value` (number, string, or bool) | -| `quality` | string | a convenience: the **first** sample's `quality` (`""` if absent) | - -`body` and `tags` are the two roots. `value`/`quality`/`samples` are conveniences derived from the -southbound shape — on a non-southbound payload `samples` is `[]` and `value`/`quality` are -empty/`""`, but `body` always holds whatever JSON arrived, so a script is **payload-agnostic**: read -your own paths off `body`. - -### What a script returns - -The return value is the whole contract — there is no other output channel: - -- **`filter` `script`** → a **boolean**. `true` keeps the message, `false` drops it. A non-boolean - result or a runtime error is treated as `false` (drop) and logged at WARN — a filter fails *closed*. -- **`script` stage** → the **new body**. A map (`#{ … }`) replaces `body`; the envelope (`header`, - `tags`) is preserved. **`()`** (Rhai unit) **drops** the message. A result that can't convert to - JSON, or a runtime error, also drops the message (logged at WARN). - -So a transform either reshapes (`return a map`) or drops (`return ()`); a predicate either keeps -(`true`) or drops (`false`). Nothing partially mutates the message in place. - -### Scripts are stateless (one message in, one decision out) - -Each evaluation gets a **fresh scope** and sees **only the current message** — there is no variable -that persists from one message to the next, no counter, no rolling window inside a script. This is -deliberate: it keeps a script a pure function (easy to reason about, impossible to leak memory -across messages, safe to share one engine across every route). **Cross-message state lives in the -built-in stages**, which the worker owns and drives on a timer: use `sample` for rate state and -`aggregate` for windowed counters/min/max/avg. (A future revision may add an opt-in per-key script -state object; it is intentionally not in this version.) - -### Worked examples - -```jsonc -// filter: keep only when every sample is GOOD and under 100 -{ "filter": { "script": "samples.all(|s| s.quality == \"GOOD\" && s.value < 100.0)" } } - -// transform: rescale the first value and stamp the source topic, keeping the signal identity -{ "script": "#{ \"signal\": body.signal, \"scaled\": value * 0.1, \"src\": topic }" } - -// conditional drop: keep the body as-is below the limit, drop it above -{ "script": "if value > 1000.0 { () } else { body }" } - -// payload-agnostic: a non-southbound body, read your own paths and fold in envelope metadata -{ "script": "#{ \"site\": tags.site, \"tempF\": body.temperature * 1.8 + 32.0 }" } - -// from a file (multi-line logic): rules/derive.rhai resolved under scriptsDir -{ "script": { "file": "rules/derive.rhai" } } -``` - -`#{ … }` is a Rhai object-map literal; `||` is a closure; arrays have `.all` / `.any` / `.map`; `()` -is unit (drop). JSON numbers arrive as integers or floats and `quality` as a string. Keep a script -short — it runs on **every** message on the route's hot path, under the shared 1,000,000-op budget. +A script sees the **message view** (`topic`, `body`, `tags`, `samples`, and the first-sample +conveniences `value`/`quality`) plus the **runtime context** (`thingName`, `componentName`, +`componentFullName`, `routeId`, `recvMs`) so a generic, reusable script can branch on which +component/route/thing it runs in. Scripts are **stateless** — each evaluation sees only the current +message; cross-message state belongs in `sample`/`aggregate`. Array-valued fields arrive as Rhai +arrays, so a script can `for`/`map`/`filter`/`reduce` over them like any collection. + +> **Scripting has its own guide.** The dedicated **[Scripting page](scripting.md)** is the full +> treatment: every scope binding, return and error semantics, a Rhai language primer (functions, +> loops, ranges, `switch`, array methods), array handling, and a **cookbook of worked examples** +> (derived units, array mean/peak/RMS, rate-of-change, reusable identity-stamping, payload +> normalization, status mapping) — each explained goal → how → why, and each backed by a test. ## The two flows: from adapter to cloud diff --git a/docs/how-to-guides.md b/docs/how-to-guides.md index df20b8b..54599c7 100644 --- a/docs/how-to-guides.md +++ b/docs/how-to-guides.md @@ -78,6 +78,38 @@ A `filter` stage takes exactly one form, checked in this order: --- + +## Handle array-valued signals + +**Goal:** process a signal whose sample `value` is an **array** (an OPC UA array node, a batched +register read, a vector) — the OPC UA adapter and the wire format both carry these. + +Array values are first-class across the stages; pick the tool for the job: + +- **Aggregate across the elements.** The `aggregate` stage folds an array value **element-wise**, so + the numeric reducers span every element of every sample in the window — no special syntax: + + ```jsonc + { "aggregate": { "window": "10s", "by": "body.signal.id", "fn": ["avg", "min", "max", "count"] } } + // value [10,20,30] then [40,50] in a window → avg 30, min 10, max 50, count 5 + ``` + +- **Filter on the elements.** A trailing `[]` flattens the array so an **any-element** predicate works: + + ```jsonc + { "filter": { "field": "body.samples[].value[]", "op": "gt", "value": 100 } } // keep if any element > 100 + ``` + +- **Compute over the array in a script.** An array `value` arrives as a Rhai array — `map`/`filter`/ + `reduce`/`for` over it to emit mean, peak, RMS, counts, etc. See the + [Scripting cookbook](scripting.md#2-array-node-mean-peak-and-rms). + +- **Archive the array.** The file sink's default `rows` projection stores an array as JSON in the + `valueString` column; to spread it into one row per element, declare a + [`rows` projection](reference/data-types.md#rows-user-projection) with `explode`. + +--- + ## Reshape a message **Goal:** keep a whitelist of fields and/or stamp in literals. @@ -107,12 +139,14 @@ A `filter` stage takes exactly one form, checked in this order: - A **`filter` `script`** returns a boolean — `true` keeps the message. - A **`script`** stage returns the **new body** map, or `()` to **drop** the message. -- Scope exposed to both: `topic` (string), `body` and `tags` (maps — `tags` is envelope metadata, not - the signal), `samples` (array), and the convenience bindings `value` / `quality` (the first - sample's). An eval error or a non-JSON result drops the message (logged at WARN). - -For the full scripting model — bindings, return values, statelessness, more examples — see -[Scripting with Rhai](explanation.md#scripting-with-rhai). +- Scope exposed to both: the message view — `topic`, `body` and `tags` (maps; `tags` is envelope + metadata, not the signal), `samples`, and the conveniences `value` / `quality` (the first sample's) — + **plus the runtime context** `thingName` / `componentName` / `componentFullName` / `routeId` / + `recvMs`. An eval error or a non-JSON result drops the message (logged at WARN). + +For the full scripting model — every scope binding (incl. the runtime context `thingName` / +`componentName` / `routeId` / `recvMs`), return/error semantics, a Rhai language primer, array +handling, and a cookbook of worked examples — see the dedicated **[Scripting guide](scripting.md)**. --- diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index cf37a00..655540f 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -140,10 +140,11 @@ source is given **inline** or from an **external file**: Both forms are **compiled once at startup** (a bad path or a compile error fails fast, before any message flows); the shared engine is bounded (1,000,000 ops/eval) so a runaway script cannot wedge a -worker. For the full scripting model — scope, state, return values, examples — see the -[Scripting explanation](../explanation.md#scripting-with-rhai). +worker. For the full scripting model — scope, state, return values, the Rhai language, and a cookbook +of worked examples — see the dedicated **[Scripting guide](../scripting.md)**. -**Rhai scope** (available to both `filter` `script` and the `script` stage): +**Rhai scope** (available to both `filter` `script` and the `script` stage) — +the per-message **message view** plus the constant **runtime context**: | Binding | Type | Value | |---------|------|-------| @@ -151,8 +152,13 @@ worker. For the full scripting model — scope, state, return values, examples | `body` | map | the message body | | `tags` | map | the envelope `tags{}` (message metadata — *not* the signal) | | `samples` | array | `body.samples` (or `[]`) | -| `value` | any | the first sample's `value` | +| `value` | any | the first sample's `value` (scalar **or array**) | | `quality` | string | the first sample's `quality` | +| `thingName` | string | the IoT Thing name (`{ThingName}`) | +| `componentName` | string | the short component name (`{ComponentName}`) | +| `componentFullName` | string | the fully-qualified component name (`{ComponentFullName}`) | +| `routeId` | string | the id of the route running the script | +| `recvMs` | integer | this message's broker receive time (Unix ms) | > **Key paths** are dotted paths over the message: roots `body.` (the default when no known root diff --git a/docs/reference/data-types.md b/docs/reference/data-types.md index 77f4646..e9b3e5d 100644 --- a/docs/reference/data-types.md +++ b/docs/reference/data-types.md @@ -127,6 +127,10 @@ A sample `value` is narrowed to one typed column: > may lose precision for `|value| > 2^53`. An unsigned value above `2^63` is cast into the signed > int64 column. +> **Array values** land in `valueString` as compact JSON (`valueType = "string"`) under the default +> projection. To spread an array into one row per element, or to fold it in `aggregate`, filter, or a +> script instead, see [Handle array-valued signals](../how-to-guides.md#handle-array-valued-signals). + ## Quality & timestamps - **`quality`** is the normalized, protocol-independent verdict — `GOOD` \| `BAD` \| `UNCERTAIN` diff --git a/docs/scripting.md b/docs/scripting.md new file mode 100644 index 0000000..8b3a380 --- /dev/null +++ b/docs/scripting.md @@ -0,0 +1,358 @@ +# Scripting with Rhai + +The built-in stages (`filter`, `sample`, `aggregate`, `project`) cover the shapes most routes need, and +they are the right tool when they fit: they compile to fixed forms and cost nothing per message beyond +the work they do. But telemetry logic is open-ended — a derived engineering unit, a reading you only +care about when it moves, a vendor payload that doesn't match the southbound shape, a decision that +depends on the *relationship* between several samples. **Scripting is the escape hatch for logic the +built-ins don't express.** + +The processor embeds [Rhai](https://rhai.rs) — a small, sandboxed, Rust-native scripting language. A +script is compiled **once at startup** and then run per message on the route's hot path. It has no I/O, +no access to the filesystem or network, and runs under a bounded operation budget, so a script can shape +data but cannot reach outside the pipeline or wedge a worker. + +This page is the complete guide: the two places scripts run, everything a script can see, what it must +return, the language features you'll actually use, how array-typed values behave, and a cookbook of +worked examples — each one explained (goal → how → why), and each one backed by a test in the +processor's own suite so the syntax is known-good. + +> **New to a specific task?** For *where* scripts live (inline vs. an external `.rhai` file) and how to +> ship them to a device or pod, see [Use an external script file](how-to-guides.md#use-an-external-script-file) +> and [Ship script files with a deployment](how-to-guides.md#ship-script-files-with-a-deployment). For +> the config field reference, see [`script`](reference/configuration.md#script-stage). + +## The two places a script runs + +Scripting appears in two roles. Both use the same engine and the same [scope](#the-scope-what-a-script-sees); +they differ only in what they return. + +- A **`filter` `script`** is a **predicate**: it evaluates to a boolean, and the message is kept when the + result is `true`. Use it when the built-in `field`/`op`/`value` or `quality` filters can't express the + condition — e.g. a test that spans several samples or inspects an array. + + ```jsonc + { "filter": { "script": "value.filter(|x| x > 50).len() >= 2" } } + ``` + +- A **`script` stage** is a **transform**: it evaluates to the **new message body** (a map), or to `()` + (Rhai's unit value) to **drop** the message. Use it to derive fields, reshape a payload, or conditionally + discard. + + ```jsonc + { "script": "#{ \"tempF\": value * 1.8 + 32.0 }" } + ``` + +Inline source is fine for a one-liner. Anything longer belongs in a `.rhai` file referenced as +`{"script": {"file": "rules/derive.rhai"}}`, which version-controls cleanly, needs no JSON string-escaping, +and is compiled and validated at startup — a missing file or a syntax error stops the component +immediately rather than failing silently at the first message. + +## The scope: what a script sees + +Before each evaluation the stage builds a fresh **scope** and binds two groups of variables: the +**message view** (the data in front of you) and the **runtime context** (constant facts about where the +script is running). Everything is a plain value — read them, combine them, return something new. + +### The message view + +| Binding | Type | What it is | +|---|---|---| +| `topic` | string | the source topic the message arrived on | +| `body` | map | the full message body — `body.signal`, `body.samples`, `body.device`, or *any* JSON your payload carries | +| `tags` | map | the message-**envelope** metadata (`tags.site`, `tags.thing`, …) — identity, **not** the signal | +| `samples` | array | `body.samples` (or `[]` when absent); each element is a map with `value`, `quality`, `sourceTs`, … | +| `value` | any | convenience: the **first** sample's `value` (a number, string, bool, **or array**) | +| `quality` | string | convenience: the **first** sample's `quality` (`""` when absent) | + +`body` and `tags` are the two roots; `value`/`quality`/`samples` are conveniences derived from the +southbound shape. On a payload that *isn't* southbound-shaped, `samples` is `[]` and `value`/`quality` +are empty — but `body` always holds whatever arrived, so **a script is payload-agnostic**: read your own +paths off `body`. + +### The runtime context + +These are **constant for the life of a route** — the processor's own identity and which route is +running. They mirror the config template variables (`{ThingName}` etc.), and they exist so a **single +generic script can be reused across components** and still stamp or branch on where it runs. + +| Binding | Type | What it is | +|---|---|---| +| `thingName` | string | the IoT Thing name (`{ThingName}`) | +| `componentName` | string | the short component name (`{ComponentName}`, the segment after the last `.`) | +| `componentFullName` | string | the fully-qualified component name (`{ComponentFullName}`) | +| `routeId` | string | the id of the route this script belongs to | +| `recvMs` | integer | the broker **receive time** of this message, in Unix milliseconds | + +`recvMs` is per-message (it changes each call); the four identity bindings are the same for every message +on the route. + +## What a script returns + +The return value is the entire contract — there is no other output channel, and a script never mutates +the message in place. + +- A **`filter` `script`** returns a **boolean**. `true` keeps the message, `false` drops it. A + non-boolean result, or a runtime error, is treated as `false` (drop) and logged at WARN — **a filter + fails closed**, so a broken predicate discards rather than leaks. +- A **`script` stage** returns the **new body**. A map (`#{ … }`) replaces `body`; the envelope + (`header`, `tags`) is preserved. Returning **`()`** drops the message. A result that can't convert to + JSON, or a runtime error, also drops the message (logged at WARN). + +## Scripts are stateless + +Each evaluation gets a **fresh scope** and sees **only the current message** — there is no variable that +survives from one message to the next, no counter, no rolling buffer inside a script. This is deliberate: +a script stays a pure function of its input, which makes it easy to reason about, impossible to leak +memory across messages, and safe to share one engine across every route. + +**Cross-message state lives in the built-in stages**, which the route worker owns and drives on a timer: +use `sample` for per-key rate limiting and `aggregate` for windowed counters, min/max/avg. A common, +powerful pattern is to let a script *shape* each message and an `aggregate` stage *accumulate* across +them — see the cookbook's [rate-of-change](#3-rate-of-change-across-consecutive-samples) example, which a +downstream `aggregate` can then reduce. + +## A Rhai primer for the processor + +You don't need to learn all of Rhai. This is the subset that matters here; the [Rhai +book](https://rhai.rs/book/) has the full language. + +**Values.** Integers (`42`), floats (`3.14`), strings (`"ok"`), booleans, arrays (`[1, 2, 3]`), object +maps (`#{ "a": 1 }`), and `()` (unit — "nothing", used to drop a message). JSON from the message maps +directly: a JSON object → a Rhai map, a JSON array → a Rhai array, a JSON number → an integer or float. + +**Maps and arrays.** Build a body with a map literal `#{ "k": v, … }`; build a list with `[a, b, …]`. +Read fields with `.` (`body.signal.id`) and index with `[]` (`samples[0].value`). + +**Math.** The usual `+ - * / %`. Integers and floats mix (an integer is promoted), so `value * 1.8` +works whether `value` is `20` or `20.0`. Float methods like `.sqrt()`, `.round()`, `.abs()` are +available. + +**Control flow.** `if cond { … } else { … }`; `switch x { "A" => 1, _ => 0 }` for value dispatch; +`for x in xs { … }` and `for i in 0..n { … }` (ranges are exclusive of the end); `while cond { … }`. Use +`return v;` to exit a function or the script early. + +**Functions.** Define reusable helpers with `fn name(args) { … }` — the last expression is the return +value. Functions keep a script readable when the logic is more than a line. + +**Array methods.** `len()`, `is_empty()`, `push(x)`, and the higher-order trio you'll reach for +constantly: `map(|x| …)` (transform each element), `filter(|x| …)` (keep matching elements), and +`reduce(|acc, x| …, seed)` (fold to a single value). Also `all(|x| …)`, `some(|x| …)`, `contains(x)`, +`index_of(x)`. + +**The operation budget.** Every evaluation runs under a cap of **1,000,000 operations**, so a runaway +loop can't hang a worker — but it also means a script that loops over a very large array on every message +has a real cost. Keep scripts tight; push heavy accumulation into `aggregate`. + +## Working with array-typed values + +A sample's `value` is not always a scalar. An OPC UA array node, a batched register read, or a vector +signal produces `value` as a **JSON array**, and the wire format carries it faithfully. In a script that +array arrives as an ordinary **Rhai array**, so all the array machinery above applies — iterate it, +`map`/`filter`/`reduce` it, index it. The [cookbook](#2-array-node-mean-peak-and-rms) shows mean, peak, +and RMS over an array value. + +Array values are first-class in the built-ins too: the **`aggregate` stage folds an array value +element-wise** (every element counts and feeds `avg`/`min`/`max`/`sum`), and a **filter/key path can +flatten an array** with a trailing `[]` (`body.samples[].value[]`). For archiving, the file sink's +default projection stores an array as a JSON string in `valueString`, and a +[declared `rows` projection](reference/data-types.md#rows-user-projection) can `explode` an array into one +row per element. So an array signal flows through filtering, scripting, aggregation, and archival without +being dropped or flattened to an opaque blob. + +## Cookbook + +Real patterns, each with the goal, the script, how it works, and why you'd write it this way. Every +example here is exercised by a test in `src/proc/script.rs`, so the syntax is verified. + +### 1. Derive an engineering unit, dropping empty reads + +**Goal:** convert a raw Celsius reading to Fahrenheit, but drop a message that carries no sample rather +than emitting a bogus value. + +```rhai +fn to_fahrenheit(c) { c * 1.8 + 32.0 } + +if samples.is_empty() { return (); } // no reading → drop the message + +#{ "signal": body.signal, "tempF": to_fahrenheit(value) } +``` + +**How it works.** A helper `fn` names the conversion so the intent is obvious and the formula lives in +one place. The guard runs first: if there are no samples, `return ()` drops the message before anything +tries to use the (absent) `value`. Otherwise the script returns a new body that keeps the source +`signal` identity and adds the derived `tempF`. + +**Why.** Deriving units at the edge means the cloud stores query-ready values, not raw counts. The guard +matters because `value` is `()` when `samples` is empty — computing on it would produce a garbage row; +dropping is the honest outcome. + +### 2. Array node: mean, peak, and RMS + +**Goal:** an OPC UA array node delivers `value` as `[10.0, 20.0, 30.0]`; emit summary statistics for it. + +```rhai +fn mean(xs) { + if xs.is_empty() { return 0.0; } + let s = 0.0; + for x in xs { s += x; } + s / xs.len() +} + +let readings = value; // the array value +let peak = readings[0]; +for x in readings { if x > peak { peak = x; } } + +#{ "mean": mean(readings), "peak": peak, "n": readings.len() } +``` + +RMS (root-mean-square) is the same shape with a `.sqrt()`: + +```rhai +let sumsq = 0.0; +for x in value { sumsq += x * x; } +#{ "rms": (sumsq / value.len()).sqrt() } +``` + +**How it works.** Because `value` is an array, `for x in value` iterates its elements, `readings[0]` +indexes it, and `readings.len()` counts them — exactly as you'd expect for a list. `mean` is factored +into a helper; `peak` is a running maximum; RMS accumulates squares then takes the root with the float +`.sqrt()` method. + +**Why.** Shipping the raw array to the cloud pushes the reduction downstream and multiplies storage. Computing +mean/peak/RMS at the edge turns a burst of numbers into the few figures an operator actually watches. + +### 3. Rate of change across consecutive samples + +**Goal:** a batched message carries several samples; emit the deltas between consecutive readings. + +```rhai +let deltas = []; +for i in 1..samples.len() { + deltas.push(samples[i].value - samples[i - 1].value); +} +#{ "deltas": deltas } +``` + +**How it works.** The range `1..samples.len()` walks the sample indices from the second to the last; +each step subtracts the previous sample's value from the current one and appends the difference. The +built-in filters can't do this — they test one value at a time and have no notion of "the previous +sample". + +**Why.** Rate of change often matters more than the absolute value — a temperature *climbing* fast is the +alarm, not the temperature itself. Emitting deltas lets a downstream `aggregate` (max delta per window) +or `filter` catch it. + +### 4. Keep only when an array crosses a threshold enough times + +**Goal:** a `filter` that keeps a message only when **at least two** elements of an array value exceed +50. + +```rhai +value.filter(|x| x > 50).len() >= 2 +``` + +**How it works.** `value.filter(|x| x > 50)` produces the sub-array of over-threshold elements; `.len()` +counts them; `>= 2` is the boolean the filter needs. One expression, no loop. + +**Why.** A single spike might be noise; two or more elements over the line is a signal. Expressing +"how many crossed" is exactly what the scalar built-in filters can't do. + +### 5. A generic, reusable script that stamps identity + +**Goal:** one script, deployed to many components, that tags each message with where it came from and +which route handled it — without hard-coding those values. + +```rhai +#{ + "signal": body.signal, + "value": value, + "thing": thingName, + "component": componentName, + "route": routeId, + "ingestedMs": recvMs +} +``` + +**How it works.** `thingName`, `componentName`, `routeId`, and `recvMs` come from the [runtime +context](#the-runtime-context), not the message — so the *same* script text produces +`"thing": "edge-42"` on one device and `"thing": "edge-77"` on another. `recvMs` records when the +processor received the message. + +**Why.** Reusable scripts are how you avoid a bespoke script per component. Because identity is injected +rather than written into the script, one file in a ConfigMap or artifact bundle can serve a whole fleet. + +### 6. Normalize a non-southbound vendor payload + +**Goal:** an upstream publisher sends `{"dev": "pump-7", "metric": "vibration", "raw": 325}`; reshape it +into the southbound `signal` shape so the built-in stages and the file sink's default projection work. + +```rhai +#{ + "signal": #{ "id": body.dev, "name": body.metric }, + "samples": [ #{ "value": body.raw * 0.1, "quality": "GOOD" } ] +} +``` + +**How it works.** The script reads the vendor fields straight off `body` and constructs a proper +`SouthboundSignalUpdate`-shaped body: a `signal` map with `id`/`name`, and a one-element `samples` array +carrying the scaled value. Nested map literals build the structure inline. + +**Why.** This is the adapter-of-last-resort: rather than teach every downstream stage a new payload +shape, one `script` stage at the front of the route normalizes into the shape everything already +understands. It's the payload-agnostic model in action. + +### 7. Map a status string to a code + +**Goal:** translate a vendor status string into a compact numeric code. + +```rhai +let code = switch body.status { + "RUNNING" => 1, + "IDLE" => 0, + "FAULT" => -1, + _ => 99, +}; +#{ "statusCode": code } +``` + +**How it works.** `switch` dispatches on the value of `body.status`; each arm maps a known string to its +code, and `_` catches anything unexpected (here, `99`). The result is bound to `code` and returned in the +new body. + +**Why.** Numeric codes are cheaper to store and easier to threshold/alarm on than free-text status +strings, and the `_` arm makes "unknown" explicit instead of silently dropping. + +### 8. Sum an array with `reduce` + +**Goal:** total an array value in one expression instead of a loop. + +```rhai +#{ "total": value.reduce(|a, v| a + v, 0.0) } +``` + +**How it works.** `reduce` folds the array: it starts the accumulator `a` at the seed `0.0`, then calls +`|a, v| a + v` for each element, threading the running total through. The final accumulator is the sum. + +**Why.** For a simple fold, `reduce` is clearer than a `for` loop with an external accumulator, and it +composes — swap the closure for `if v > a { v } else { a }` and you have `max` instead. + +## Limits and gotchas + +- **Stateless.** No memory across messages; use `sample`/`aggregate` for cross-message state. +- **Fail-closed / drop-on-error.** A filter that errors drops the message; a transform that errors or + returns non-JSON drops it (both logged at WARN). Scripts never crash the route. +- **The 1,000,000-op budget** bounds each evaluation — deep loops over large arrays on every message add + up; push heavy work into `aggregate`. +- **No I/O.** Scripts can't read files, call the network, or see other messages — by design. +- **Integer vs. float.** JSON `20` arrives as an integer and `20.0` as a float; mixed arithmetic promotes + to float, but if you need a float result from integer inputs, multiply by a float (`x * 1.0`). +- **Compiled once at startup.** Editing a `.rhai` file needs a restart (a new deployment / pod rollout), + not a live reload. + +## See also + +- [Use an external script file](how-to-guides.md#use-an-external-script-file) — inline vs. file, `scriptsDir`. +- [Ship script files with a deployment](how-to-guides.md#ship-script-files-with-a-deployment) — Greengrass artifacts / k8s ConfigMap. +- [`script` configuration reference](reference/configuration.md#script-stage) — the config fields. +- [Explanation — the pipeline](explanation.md) — where scripting sits among the stages. diff --git a/src/app.rs b/src/app.rs index 5258026..31b36ce 100644 --- a/src/app.rs +++ b/src/app.rs @@ -33,14 +33,20 @@ struct RouteWire { tx: mpsc::Sender, } -/// Per-route-invariant build context: the shared Rhai engine, the script-file loader, and the -/// cross-route defaults. Bundled so `build_route` takes the route plus one context, not a long -/// argument list. +/// Per-route-invariant build context: the shared Rhai engine, the script-file loader, the +/// cross-route defaults, and the component identity injected into scripts. Bundled so `build_route` +/// takes the route plus one context, not a long argument list. struct RouteBuildCtx<'a> { engine: &'a Arc, loader: &'a crate::proc::script::ScriptLoader, default_key: &'a str, default_target: Option<&'a str>, + /// `{ThingName}` — raw (not topic-sanitized), injected into scripts as `thingName`. + thing_name: &'a str, + /// `{ComponentName}` — the short name (segment after the last `.`), injected as `componentName`. + component_name: &'a str, + /// `{ComponentFullName}` — the fully-qualified name, injected as `componentFullName`. + component_full_name: &'a str, } /// The running processor: its subscriptions, channel senders, and worker tasks. @@ -76,11 +82,20 @@ impl ProcessorApp { .map(|d| resolve(&config, d)) .unwrap_or_else(|| ".".to_string()); let loader = crate::proc::script::ScriptLoader::new(scripts_dir); + // Component identity for the script runtime context. Read the raw values (the template + // resolver would sanitize them for topic/path safety, which we don't want in a script). + let component_full_name = config.component_name.clone(); + let component_name = + component_full_name.rsplit('.').next().unwrap_or(&component_full_name).to_string(); + let thing_name = config.thing_name.clone(); let ctx = RouteBuildCtx { engine: &engine, loader: &loader, default_key: &default_key, default_target: defaults.target.as_deref(), + thing_name: &thing_name, + component_name: &component_name, + component_full_name: &component_full_name, }; let mut app = Self { @@ -163,7 +178,14 @@ impl ProcessorApp { _ => None, }; - let pipeline = Pipeline::build(&route.pipeline, &route_key, ctx.engine, ctx.loader)?; + let script_ctx = Arc::new(crate::proc::script::ScriptContext { + thing_name: ctx.thing_name.to_string(), + component_name: ctx.component_name.to_string(), + component_full_name: ctx.component_full_name.to_string(), + route_id: route.id.clone(), + }); + let pipeline = + Pipeline::build(&route.pipeline, &route_key, ctx.engine, ctx.loader, &script_ctx)?; let dispatcher = Dispatcher::new( self.messaging.clone(), target, diff --git a/src/proc/aggregate.rs b/src/proc/aggregate.rs index f66b0c5..acea84d 100644 --- a/src/proc/aggregate.rs +++ b/src/proc/aggregate.rs @@ -51,7 +51,17 @@ impl Acc { } } + /// Fold one value into the accumulator. An **array** value is folded element-wise (each element + /// counts and contributes to the numeric reducers), so an array-typed signal — an OPC UA array + /// node, or a `value` path that resolves to an array — aggregates across its elements instead of + /// being dropped as non-numeric. Nested arrays recurse; an empty array folds nothing. fn fold_value(&mut self, v: &Value) { + if let Some(arr) = v.as_array() { + for el in arr { + self.fold_value(el); + } + return; + } self.count += 1; if self.first.is_none() { self.first = Some(v.clone()); @@ -273,6 +283,38 @@ mod tests { assert_eq!(out[0].msg.body["agg"]["count"], json!(1)); } + #[test] + fn aggregate_folds_array_values_elementwise() { + // An array-typed signal (e.g. an OPC UA array node): each sample's value is an array. The + // reducers fold across ALL elements of ALL samples in the window (a time window here, so the + // close is time-driven and independent of how many elements each message carries). + let spec = AggregateSpec { + window: "1s".into(), + by: None, + fns: vec!["avg".into(), "min".into(), "max".into(), "sum".into(), "count".into()], + value: None, + }; + let mut s = AggregateStage::build(&spec, "body.signal.id").unwrap(); + let mk = |arr: serde_json::Value, recv: u64| { + let m = MessageBuilder::new("SouthboundSignalUpdate", "1.0") + .payload(json!({ "signal": { "id": "a" }, "samples": [ { "value": arr, "quality": "GOOD" } ] })) + .build(); + ProcMsg { topic: "t".into(), msg: m, recv_ms: recv } + }; + // Window [0,1000): two array-valued messages. + assert!(s.process(mk(json!([1.0, 2.0, 3.0]), 100)).is_empty()); + assert!(s.process(mk(json!([4.0, 5.0]), 200)).is_empty()); + let out = s.on_tick(1000); + assert_eq!(out.len(), 1); + let agg = &out[0].msg.body["agg"]; + // elements folded across both messages: 1,2,3,4,5 → avg 3, min 1, max 5, sum 15, count 5 + assert_eq!(agg["avg"], json!(3.0)); + assert_eq!(agg["min"], json!(1.0)); + assert_eq!(agg["max"], json!(5.0)); + assert_eq!(agg["sum"], json!(15.0)); + assert_eq!(agg["count"], json!(5)); + } + #[test] fn aggregate_custom_value_path_non_southbound() { // A non-SouthboundSignalUpdate payload: aggregate `body.temp`, keyed by `body.id`. diff --git a/src/proc/filter.rs b/src/proc/filter.rs index b307ee8..4baa659 100644 --- a/src/proc/filter.rs +++ b/src/proc/filter.rs @@ -13,7 +13,7 @@ use smallvec::smallvec; use crate::config::FilterSpec; use crate::json_path::resolve_values; -use crate::proc::script::{RhaiEval, ScriptLoader}; +use crate::proc::script::{RhaiEval, ScriptContext, ScriptLoader}; use crate::proc::{Out, ProcMsg, Processor}; enum Op { @@ -45,9 +45,10 @@ impl FilterStage { spec: &FilterSpec, engine: &Arc, loader: &ScriptLoader, + ctx: &Arc, ) -> anyhow::Result { let pred = if let Some(src) = &spec.script { - Predicate::Rhai(RhaiEval::compile(engine, &loader.load(src)?)?) + Predicate::Rhai(RhaiEval::compile(engine, &loader.load(src)?, ctx)?) } else if let Some(q) = &spec.quality { Predicate::QualityAll(q.clone()) } else if let Some(field) = &spec.field { @@ -157,6 +158,10 @@ mod tests { ProcMsg { topic: "southbound/x".into(), msg: m, recv_ms: now_ms() } } + fn ctx() -> Arc { + Arc::new(ScriptContext::default()) + } + fn engine() -> Arc { Arc::new(Engine::new()) } @@ -164,7 +169,7 @@ mod tests { #[test] fn quality_all_keeps_only_all_good() { let spec = FilterSpec { quality: Some("GOOD".into()), ..Default::default() }; - let mut s = FilterStage::build(&spec, &engine(), &ScriptLoader::default()).unwrap(); + let mut s = FilterStage::build(&spec, &engine(), &ScriptLoader::default(), &ctx()).unwrap(); let good = msg(json!([{ "value": 1, "quality": "GOOD" }, { "value": 2, "quality": "GOOD" }])); let mixed = msg(json!([{ "value": 1, "quality": "GOOD" }, { "value": 2, "quality": "BAD" }])); assert_eq!(s.process(good).len(), 1); @@ -179,7 +184,7 @@ mod tests { value: Some(json!(50)), ..Default::default() }; - let mut s = FilterStage::build(&spec, &engine(), &ScriptLoader::default()).unwrap(); + let mut s = FilterStage::build(&spec, &engine(), &ScriptLoader::default(), &ctx()).unwrap(); assert_eq!(s.process(msg(json!([{ "value": 99 }]))).len(), 1); assert_eq!(s.process(msg(json!([{ "value": 10 }]))).len(), 0); } @@ -190,7 +195,7 @@ mod tests { script: Some(ScriptSource::Inline("samples.all(|s| s.quality == \"GOOD\")".into())), ..Default::default() }; - let mut s = FilterStage::build(&spec, &engine(), &ScriptLoader::default()).unwrap(); + let mut s = FilterStage::build(&spec, &engine(), &ScriptLoader::default(), &ctx()).unwrap(); assert_eq!(s.process(msg(json!([{ "value": 1, "quality": "GOOD" }]))).len(), 1); assert_eq!(s.process(msg(json!([{ "value": 1, "quality": "BAD" }]))).len(), 0); } @@ -205,6 +210,7 @@ mod tests { }, &engine(), &ScriptLoader::default(), + &ctx(), ) .unwrap() } @@ -227,7 +233,7 @@ mod tests { #[test] fn build_and_op_parse_errors() { // No predicate form configured. - assert!(FilterStage::build(&FilterSpec::default(), &engine(), &ScriptLoader::default()).is_err()); + assert!(FilterStage::build(&FilterSpec::default(), &engine(), &ScriptLoader::default(), &ctx()).is_err()); // Unknown op. assert!(Op::parse("bogus").is_err()); // Symbolic aliases parse. diff --git a/src/proc/mod.rs b/src/proc/mod.rs index 171d6f3..a2a2f03 100644 --- a/src/proc/mod.rs +++ b/src/proc/mod.rs @@ -51,19 +51,21 @@ pub struct Pipeline { impl Pipeline { /// Build the pipeline from config. `route_key` is the default aggregation/sample key; `engine` - /// is the shared Rhai engine for `filter`/`script` stages. + /// is the shared Rhai engine and `ctx` the per-route runtime context (identity + route id) + /// bound into every `filter`/`script` evaluation. pub fn build( stages: &[StageConfig], route_key: &str, engine: &Arc, loader: &script::ScriptLoader, + ctx: &Arc, ) -> anyhow::Result { let mut built: Vec> = Vec::with_capacity(stages.len()); let mut tick_hints: Vec = Vec::new(); for sc in stages { let stage: Box = match sc { StageConfig::Filter(spec) => { - Box::new(filter::FilterStage::build(spec, engine, loader)?) + Box::new(filter::FilterStage::build(spec, engine, loader, ctx)?) } StageConfig::Sample(spec) => Box::new(sample::SampleStage::build(spec, route_key)?), StageConfig::Aggregate(spec) => { @@ -74,7 +76,7 @@ impl Pipeline { } StageConfig::Project(spec) => Box::new(project::ProjectStage::build(spec)), StageConfig::Script(src) => { - Box::new(script::ScriptStage::build(src, engine, loader)?) + Box::new(script::ScriptStage::build(src, engine, loader, ctx)?) } }; built.push(stage); @@ -145,7 +147,7 @@ mod tests { value: None, }), ]; - let mut p = Pipeline::build(&stages, "body.signal.id", &engine(), &script::ScriptLoader::default()).unwrap(); + let mut p = Pipeline::build(&stages, "body.signal.id", &engine(), &script::ScriptLoader::default(), &Arc::new(script::ScriptContext::default())).unwrap(); assert_eq!(p.min_tick_ms(), None, "a count window needs no flush timer"); // BAD is filtered out; two GOODs fill the count=2 window and emit on the second. @@ -168,7 +170,7 @@ mod tests { value: None, }), ]; - let mut p = Pipeline::build(&stages, "body.signal.id", &engine(), &script::ScriptLoader::default()).unwrap(); + let mut p = Pipeline::build(&stages, "body.signal.id", &engine(), &script::ScriptLoader::default(), &Arc::new(script::ScriptContext::default())).unwrap(); assert_eq!(p.min_tick_ms(), Some(1000)); p.run(one("a", 1.0, "GOOD", 100), None); let out = p.run(SmallVec::new(), Some(2000)); diff --git a/src/proc/script.rs b/src/proc/script.rs index bbaa931..74fecbf 100644 --- a/src/proc/script.rs +++ b/src/proc/script.rs @@ -4,9 +4,12 @@ //! message). The same [`RhaiEval`] backs the Rhai `filter` option (evaluating to a boolean). The //! engine is shared across all routes; each stage compiles its source to an `AST` once at build. //! -//! Scope exposed to a script: `topic` (string), `body` / `tags` (maps), `samples` (array), and the -//! convenience bindings `value` / `quality` (the first sample's). A `filter` script returns a -//! boolean; a `script` stage returns the new body map (or `()` to drop). +//! Scope exposed to a script: the message view (`topic`, `body` / `tags` maps, `samples` array, and +//! the convenience bindings `value` / `quality` — the first sample's), plus the **runtime context** +//! (`thingName`, `componentName`, `componentFullName`, `routeId`, `recvMs`) so a generic script can +//! branch on which component/route/thing it runs in. A `filter` script returns a boolean; a `script` +//! stage returns the new body map (or `()` to drop). Array-valued fields arrive as Rhai arrays, so a +//! script can `for`/`map`/`filter`/`reduce` over them like any other collection. use std::path::{Path, PathBuf}; use std::sync::Arc; @@ -19,6 +22,27 @@ use smallvec::smallvec; use crate::config::ScriptSource; use crate::proc::{Out, ProcMsg, Processor}; +/// Per-route runtime context injected into every `filter`/`script` evaluation as **constant** +/// bindings, alongside the per-message view. It carries the component identity and the route id so a +/// single reusable script can behave differently per component/route (e.g. stamp `thingName`, or gate +/// on `routeId`) without hard-coding those values in the script text. +/// +/// The identity mirrors the config template variables: `thing_name` = `{ThingName}`, `component_name` +/// = `{ComponentName}` (the short name — the segment after the last `.`), `component_full_name` = +/// `{ComponentFullName}`. Values are the **raw** identity (not topic-sanitized like the template +/// resolver's output). Cheap to clone; the app builds one `Arc` per route. +#[derive(Debug, Default, Clone)] +pub struct ScriptContext { + /// IoT Thing name — exposed to scripts as `thingName`. + pub thing_name: String, + /// Short component name (after the last `.`) — exposed as `componentName`. + pub component_name: String, + /// Fully-qualified component name — exposed as `componentFullName`. + pub component_full_name: String, + /// The owning route's id — exposed as `routeId`. + pub route_id: String, +} + /// Resolves [`ScriptSource`]s to Rhai source text. `File` paths resolve against `base` (the /// `global.defaults.scriptsDir`) when relative, or are used as-is when absolute. pub struct ScriptLoader { @@ -50,24 +74,35 @@ impl Default for ScriptLoader { } } -/// A compiled Rhai program plus the shared engine. +/// A compiled Rhai program plus the shared engine and the per-route runtime context. pub struct RhaiEval { engine: Arc, ast: AST, + ctx: Arc, } impl RhaiEval { - /// Compile `src` against the shared engine. - pub fn compile(engine: &Arc, src: &str) -> anyhow::Result { + /// Compile `src` against the shared engine, binding the runtime `ctx` into every evaluation. + pub fn compile( + engine: &Arc, + src: &str, + ctx: &Arc, + ) -> anyhow::Result { let ast = engine .compile(src) .map_err(|e| anyhow::anyhow!("rhai compile error in `{src}`: {e}"))?; - Ok(Self { engine: engine.clone(), ast }) + Ok(Self { engine: engine.clone(), ast, ctx: ctx.clone() }) } fn scope_for(&self, m: &ProcMsg) -> Scope<'static> { let mut scope = Scope::new(); scope.push("topic", m.topic.clone()); + // Runtime context — constant per route, so a generic/reused script can branch on identity. + scope.push("thingName", self.ctx.thing_name.clone()); + scope.push("componentName", self.ctx.component_name.clone()); + scope.push("componentFullName", self.ctx.component_full_name.clone()); + scope.push("routeId", self.ctx.route_id.clone()); + scope.push("recvMs", m.recv_ms as i64); scope.push_dynamic("body", to_dyn(&m.msg.body)); if let Ok(tags) = serde_json::to_value(&m.msg.tags) { scope.push_dynamic("tags", to_dyn(&tags)); @@ -130,9 +165,10 @@ impl ScriptStage { src: &ScriptSource, engine: &Arc, loader: &ScriptLoader, + ctx: &Arc, ) -> anyhow::Result { let text = loader.load(src)?; - Ok(Self { eval: RhaiEval::compile(engine, &text)? }) + Ok(Self { eval: RhaiEval::compile(engine, &text, ctx)? }) } } @@ -164,12 +200,17 @@ mod tests { Arc::new(Engine::new()) } + fn ctx() -> Arc { + Arc::new(ScriptContext::default()) + } + #[test] fn script_transforms_body_using_value_binding() { let mut s = ScriptStage::build( &ScriptSource::Inline(r#"#{ "doubled": value * 2 }"#.into()), &engine(), &ScriptLoader::default(), + &ctx(), ) .unwrap(); let out = s.process(pm(json!({ "samples": [{ "value": 21, "quality": "GOOD" }] }))); @@ -183,6 +224,7 @@ mod tests { &ScriptSource::Inline(r#"#{ "thing": tags.thing, "q": quality }"#.into()), &engine(), &ScriptLoader::default(), + &ctx(), ) .unwrap(); let out = s.process(pm(json!({ "samples": [{ "value": 1, "quality": "GOOD" }] }))); @@ -196,6 +238,7 @@ mod tests { &ScriptSource::Inline("()".into()), &engine(), &ScriptLoader::default(), + &ctx(), ) .unwrap(); assert_eq!(s.process(pm(json!({ "samples": [] }))).len(), 0); @@ -207,6 +250,7 @@ mod tests { &ScriptSource::Inline("this is not valid rhai @@".into()), &engine(), &ScriptLoader::default(), + &ctx(), ) .is_err()); } @@ -221,14 +265,193 @@ mod tests { let loader = ScriptLoader::new(&dir); let src = ScriptSource::File { file: "double.rhai".into() }; - let mut s = ScriptStage::build(&src, &engine(), &loader).unwrap(); + let mut s = ScriptStage::build(&src, &engine(), &loader, &ctx()).unwrap(); let out = s.process(pm(json!({ "samples": [{ "value": 21, "quality": "GOOD" }] }))); assert_eq!(out[0].msg.body["doubled"], json!(42)); // A missing file is a build error (surfaced at startup, not silently ignored). let missing = ScriptSource::File { file: "nope.rhai".into() }; - assert!(ScriptStage::build(&missing, &engine(), &loader).is_err()); + assert!(ScriptStage::build(&missing, &engine(), &loader, &ctx()).is_err()); std::fs::remove_dir_all(&dir).ok(); } + + // A `ScriptContext` with real identity, for the runtime-context test. + fn ctx_with(thing: &str, comp: &str, full: &str, route: &str) -> Arc { + Arc::new(ScriptContext { + thing_name: thing.into(), + component_name: comp.into(), + component_full_name: full.into(), + route_id: route.into(), + }) + } + + #[test] + fn script_sees_runtime_context() { + // The runtime identity is available so one generic script can stamp/branch on where it runs. + let src = ScriptSource::Inline( + r#"#{ "t": thingName, "c": componentName, "cf": componentFullName, "r": routeId, "gotTs": recvMs > 0 }"# + .into(), + ); + let ctx = ctx_with("edge-42", "TelemetryProcessor", "com.acme.TelemetryProcessor", "archive"); + let mut s = ScriptStage::build(&src, &engine(), &ScriptLoader::default(), &ctx).unwrap(); + let out = s.process(pm(json!({ "samples": [] }))); + let b = &out[0].msg.body; + assert_eq!(b["t"], json!("edge-42")); + assert_eq!(b["c"], json!("TelemetryProcessor")); + assert_eq!(b["cf"], json!("com.acme.TelemetryProcessor")); + assert_eq!(b["r"], json!("archive")); + assert_eq!(b["gotTs"], json!(true)); + } + + #[test] + fn script_processes_array_value_with_fn_and_loop() { + // An array-typed sample value arrives as a Rhai array; a user fn + `for` loop reduce it. + // Goal: emit the mean and peak of an OPC UA array node's readings. + let src = ScriptSource::Inline( + r#" + fn mean(xs) { + if xs.is_empty() { return 0.0; } + let s = 0.0; + for x in xs { s += x; } + s / xs.len() + } + let readings = value; // the first sample's value — an array here + let peak = readings[0]; + for x in readings { if x > peak { peak = x; } } + #{ "mean": mean(readings), "peak": peak, "n": readings.len() } + "# + .into(), + ); + let mut s = ScriptStage::build(&src, &engine(), &ScriptLoader::default(), &ctx()).unwrap(); + let out = s.process(pm(json!({ "samples": [{ "value": [10.0, 20.0, 30.0], "quality": "GOOD" }] }))); + let b = &out[0].msg.body; + assert_eq!(b["mean"], json!(20.0)); + assert_eq!(b["peak"], json!(30.0)); + assert_eq!(b["n"], json!(3)); + } + + #[test] + fn script_filters_array_with_map_filter() { + // A filter script over an array value: keep only when ≥2 elements exceed a threshold. + // Demonstrates array `.filter` + `.len` in a boolean predicate. + let src = ScriptSource::Inline( + r#"value.filter(|x| x > 50).len() >= 2"#.into(), + ); + let e = RhaiEval::compile(&engine(), &loader_load(&src), &ctx()).unwrap(); + let keep = pm(json!({ "samples": [{ "value": [10, 60, 70, 20], "quality": "GOOD" }] })); + let drop = pm(json!({ "samples": [{ "value": [10, 60, 20, 30], "quality": "GOOD" }] })); + assert!(e.eval_bool(&keep)); + assert!(!e.eval_bool(&drop)); + } + + #[test] + fn script_maps_status_with_switch() { + // Map a vendor status string to a numeric code with a `switch` expression. + let src = ScriptSource::Inline( + r#" + let code = switch body.status { + "RUNNING" => 1, + "IDLE" => 0, + "FAULT" => -1, + _ => 99, + }; + #{ "statusCode": code } + "# + .into(), + ); + let mut s = ScriptStage::build(&src, &engine(), &ScriptLoader::default(), &ctx()).unwrap(); + let out = s.process(pm(json!({ "status": "FAULT", "samples": [] }))); + assert_eq!(out[0].msg.body["statusCode"], json!(-1)); + } + + #[test] + fn script_reduces_array_with_reduce() { + // Sum an array value with `reduce` (seed 0.0). + let src = + ScriptSource::Inline(r#"#{ "total": value.reduce(|a, v| a + v, 0.0) }"#.into()); + let mut s = ScriptStage::build(&src, &engine(), &ScriptLoader::default(), &ctx()).unwrap(); + let out = s.process(pm(json!({ "samples": [{ "value": [1.5, 2.5, 4.0], "quality": "GOOD" }] }))); + assert_eq!(out[0].msg.body["total"], json!(8.0)); + } + + #[test] + fn script_computes_deltas_over_samples() { + // Rate-of-change: the delta between each pair of consecutive samples, via a range loop. + let src = ScriptSource::Inline( + r#" + let deltas = []; + for i in 1..samples.len() { + deltas.push(samples[i].value - samples[i - 1].value); + } + #{ "deltas": deltas } + "# + .into(), + ); + let mut s = ScriptStage::build(&src, &engine(), &ScriptLoader::default(), &ctx()).unwrap(); + let out = s.process(pm(json!({ + "samples": [ { "value": 10.0 }, { "value": 13.0 }, { "value": 12.0 } ] + }))); + assert_eq!(out[0].msg.body["deltas"], json!([3.0, -1.0])); + } + + #[test] + fn script_normalizes_non_southbound_payload() { + // Reshape a vendor body into the southbound signal shape so downstream stages/sinks work. + let src = ScriptSource::Inline( + r#" + #{ + "signal": #{ "id": body.dev, "name": body.metric }, + "samples": [ #{ "value": body.raw * 0.1, "quality": "GOOD" } ] + } + "# + .into(), + ); + let mut s = ScriptStage::build(&src, &engine(), &ScriptLoader::default(), &ctx()).unwrap(); + let out = s.process(pm(json!({ "dev": "pump-7", "metric": "vibration", "raw": 325 }))); + let b = &out[0].msg.body; + assert_eq!(b["signal"]["id"], json!("pump-7")); + assert_eq!(b["signal"]["name"], json!("vibration")); + assert_eq!(b["samples"][0]["value"], json!(32.5)); + } + + #[test] + fn script_derives_unit_with_helper_and_guard() { + // A helper fn for the conversion + an early guard that drops a reading-less message. + let src = ScriptSource::Inline( + r#" + fn to_fahrenheit(c) { c * 1.8 + 32.0 } + if samples.is_empty() { return (); } + #{ "signal": body.signal, "tempF": to_fahrenheit(value) } + "# + .into(), + ); + let mut s = ScriptStage::build(&src, &engine(), &ScriptLoader::default(), &ctx()).unwrap(); + let out = s.process(pm(json!({ "signal": { "id": "t" }, "samples": [{ "value": 20.0 }] }))); + assert_eq!(out[0].msg.body["tempF"], json!(68.0)); + // No sample → the guard drops the message. + assert_eq!(s.process(pm(json!({ "signal": { "id": "t" }, "samples": [] }))).len(), 0); + } + + #[test] + fn script_computes_rms_with_sqrt() { + // Root-mean-square across an array value — uses the float `.sqrt()` from Rhai's math package. + let src = ScriptSource::Inline( + r#" + let sumsq = 0.0; + for x in value { sumsq += x * x; } + #{ "rms": (sumsq / value.len()).sqrt() } + "# + .into(), + ); + let mut s = ScriptStage::build(&src, &engine(), &ScriptLoader::default(), &ctx()).unwrap(); + let out = s.process(pm(json!({ "samples": [{ "value": [3.0, 4.0], "quality": "GOOD" }] }))); + let rms = out[0].msg.body["rms"].as_f64().unwrap(); + assert!((rms - 3.535_533).abs() < 1e-4, "rms was {rms}"); + } + + // Resolve an inline ScriptSource to text for a direct RhaiEval compile in tests. + fn loader_load(src: &ScriptSource) -> String { + ScriptLoader::default().load(src).unwrap() + } }