From f3568ce25d2da09d747d564975bca24b9ef1ffd2 Mon Sep 17 00:00:00 2001 From: danielbotros Date: Wed, 24 Jun 2026 13:00:18 -0400 Subject: [PATCH 1/4] sensor-monitor: fire arbitrary DoCommands on trigger and resolve MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add per-rule on_trigger / on_resolve action lists, each naming a resource and a DoCommand payload. On the triggering edge the monitor fires every on_trigger action; on the clearing edge it fires every on_resolve action — best-effort, edge-only (not repeated on cooldown). Action resources are declared as required dependencies in Validate and resolved at construction by short name across any API (every resource implements DoCommand), so an action can target a board, sensor, generic service, etc. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 12 +- resources/sensormonitor/sensormonitor.go | 161 ++++++++++++++++-- resources/sensormonitor/sensormonitor_test.go | 159 +++++++++++++++++ 3 files changed, 316 insertions(+), 16 deletions(-) diff --git a/README.md b/README.md index bf2bdb4..b916e81 100644 --- a/README.md +++ b/README.md @@ -79,6 +79,8 @@ When a rule fires, the monitor calls `DoCommand` on the notifier with the payloa Notifications are **edge-triggered**: a message is sent when a rule transitions from not-triggered to triggered. While the rule stays triggered no further message is sent, unless `cooldown_seconds` is set, in which case the message repeats at most once per cooldown window. When the reading clears the threshold and crosses it again, a new notification is sent. +**Actions.** Besides notifying, a rule can fire arbitrary `DoCommand`s on other resources when it triggers and when it resolves, via the `on_trigger` and `on_resolve` lists. Each action names a `resource` (any component or service on the machine, by name) and a `command` (the `DoCommand` payload). Actions fire on the edges only — once when the rule transitions to triggered, once when it clears — not on `cooldown_seconds` repeats. Each named resource is automatically added as a dependency. Actions are best-effort: a failure is logged and never affects monitoring. + ### Configuration ```json @@ -92,7 +94,9 @@ Notifications are **edge-triggered**: a message is sent when a rule transitions "key": "", "operator": "", "threshold": , - "message": "" + "message": "", + "on_trigger": [{ "resource": "", "command": { } }], + "on_resolve": [{ "resource": "", "command": { } }] } ] } @@ -114,6 +118,8 @@ Each entry in `rules`: | `operator` | string | **Yes** | Comparison applied between the reading and `threshold`. One of `>`, `>=`, `<`, `<=`, `==`, `!=` (aliases: `gt`, `gte`, `lt`, `lte`, `eq`, `ne`). | | `threshold` | number | **Yes** | The value the reading is compared against. | | `message` | string | No | Notification template. Supports placeholders `{key}`, `{value}`, `{threshold}`, `{operator}`. If omitted, a message like `temperature is 95 (> 90)` is used. | +| `on_trigger` | array | No | DoCommands to fire when the rule transitions to triggered. Each is `{ "resource": , "command": }`. | +| `on_resolve` | array | No | DoCommands to fire when the rule clears (reading returns to the non-triggered side of the threshold). Same shape as `on_trigger`. | ### Example Configuration @@ -128,7 +134,9 @@ Each entry in `rules`: "key": "temperature", "operator": ">", "threshold": 90, - "message": "High temperature alert: {key} is {value} (limit {threshold})" + "message": "High temperature alert: {key} is {value} (limit {threshold})", + "on_trigger": [{ "resource": "cooling-fan", "command": { "set": { "power": 1 } } }], + "on_resolve": [{ "resource": "cooling-fan", "command": { "set": { "power": 0 } } }] }, { "key": "humidity", diff --git a/resources/sensormonitor/sensormonitor.go b/resources/sensormonitor/sensormonitor.go index e91f601..a8f7170 100644 --- a/resources/sensormonitor/sensormonitor.go +++ b/resources/sensormonitor/sensormonitor.go @@ -1,6 +1,8 @@ // Package sensormonitor implements the viam:sensor-bundle:sensor-monitor model: a -// sensor that watches another sensor's readings against numeric trigger rules and -// sends a notification via a generic service's DoCommand when a threshold is crossed. +// sensor that watches another sensor's readings against numeric trigger rules, +// sends a notification via a generic service's DoCommand when a threshold is +// crossed, and can fire arbitrary DoCommands on other resources when a rule +// triggers or resolves. package sensormonitor import ( @@ -34,6 +36,16 @@ func init() { // defaultPollInterval is used when poll_interval_seconds is not set. const defaultPollInterval = 10 * time.Second +// Action is a DoCommand to fire on a resource when a rule changes state. The +// call is always DoCommand, so the resource may be any type (component or +// service). +type Action struct { + // Resource is the name of the resource to call. + Resource string `json:"resource"` + // Command is the DoCommand payload sent to the resource. + Command map[string]interface{} `json:"command"` +} + // Rule describes a single numeric trigger on one reading key. type Rule struct { // Key is the reading key to watch, e.g. "temperature". @@ -46,6 +58,11 @@ type Rule struct { // Message is an optional notification template. It supports the placeholders // {key}, {value}, {threshold} and {operator}. If empty, a message is generated. Message string `json:"message,omitempty"` + // OnTrigger lists DoCommands to fire when the rule transitions to triggered. + OnTrigger []Action `json:"on_trigger,omitempty"` + // OnResolve lists DoCommands to fire when the rule clears (its reading returns + // to the non-triggered side of the threshold). + OnResolve []Action `json:"on_resolve,omitempty"` } // Config is the configuration for the sensor-monitor model. @@ -76,6 +93,18 @@ func (cfg *Config) Validate(path string) ([]string, []string, error) { if len(cfg.Rules) == 0 { return nil, nil, fmt.Errorf("%s: at least one rule is required", path) } + + // Required deps: the watched sensor, the notifier, and every resource named by + // an action. Collected in a deterministic order with duplicates removed. + deps := []string{cfg.Sensor, cfg.Notifier} + seen := map[string]bool{cfg.Sensor: true, cfg.Notifier: true} + addDep := func(name string) { + if !seen[name] { + seen[name] = true + deps = append(deps, name) + } + } + for i, r := range cfg.Rules { if r.Key == "" { return nil, nil, fmt.Errorf("%s: rules[%d] missing required field 'key'", path, i) @@ -83,8 +112,31 @@ func (cfg *Config) Validate(path string) ([]string, []string, error) { if _, err := parseOperator(r.Operator); err != nil { return nil, nil, fmt.Errorf("%s: rules[%d] %w", path, i, err) } + for j, a := range r.OnTrigger { + if err := validateAction(a); err != nil { + return nil, nil, fmt.Errorf("%s: rules[%d].on_trigger[%d] %w", path, i, j, err) + } + addDep(a.Resource) + } + for j, a := range r.OnResolve { + if err := validateAction(a); err != nil { + return nil, nil, fmt.Errorf("%s: rules[%d].on_resolve[%d] %w", path, i, j, err) + } + addDep(a.Resource) + } } - return []string{cfg.Sensor, cfg.Notifier}, nil, nil + return deps, nil, nil +} + +// validateAction checks a single action's required fields. +func validateAction(a Action) error { + if a.Resource == "" { + return fmt.Errorf("missing required field 'resource'") + } + if a.Command == nil { + return fmt.Errorf("missing required field 'command'") + } + return nil } // ruleState tracks the runtime state of a single rule across polls. @@ -103,6 +155,10 @@ type sensorMonitor struct { sensorDep sensor.Sensor notifierDep generic.Service + // actionResources holds every resource named by a rule action, resolved once + // at construction and keyed by name. Any resource type works — actions only + // ever call DoCommand. + actionResources map[string]resource.Resource pollInterval time.Duration cooldown time.Duration @@ -150,6 +206,33 @@ func newMonitor(deps resource.Dependencies, name resource.Name, conf *Config, lo return nil, fmt.Errorf("failed to get notifier dependency %q: %w", conf.Notifier, err) } + // Resolve every resource named by an action up front (any API), so firing is a + // simple map lookup + DoCommand. + actionResources := map[string]resource.Resource{} + resolveAction := func(a Action) error { + if _, ok := actionResources[a.Resource]; ok { + return nil + } + res, err := lookupResource(deps, a.Resource) + if err != nil { + return fmt.Errorf("action resource %q: %w", a.Resource, err) + } + actionResources[a.Resource] = res + return nil + } + for _, r := range conf.Rules { + for _, a := range r.OnTrigger { + if err := resolveAction(a); err != nil { + return nil, err + } + } + for _, a := range r.OnResolve { + if err := resolveAction(a); err != nil { + return nil, err + } + } + } + pollInterval := defaultPollInterval if conf.PollIntervalSec > 0 { pollInterval = time.Duration(conf.PollIntervalSec * float64(time.Second)) @@ -159,20 +242,41 @@ func newMonitor(deps resource.Dependencies, name resource.Name, conf *Config, lo cancelCtx, cancelFunc := context.WithCancel(context.Background()) return &sensorMonitor{ - Named: name.AsNamed(), - logger: logger, - cfg: conf, - sensorDep: sensorDep, - notifierDep: notifierDep, - pollInterval: pollInterval, - cooldown: cooldown, - cancelCtx: cancelCtx, - cancelFunc: cancelFunc, - lastReadings: map[string]interface{}{}, - ruleStates: make([]ruleState, len(conf.Rules)), + Named: name.AsNamed(), + logger: logger, + cfg: conf, + sensorDep: sensorDep, + notifierDep: notifierDep, + actionResources: actionResources, + pollInterval: pollInterval, + cooldown: cooldown, + cancelCtx: cancelCtx, + cancelFunc: cancelFunc, + lastReadings: map[string]interface{}{}, + ruleStates: make([]ruleState, len(conf.Rules)), }, nil } +// lookupResource finds a dependency by its short name regardless of API, so an +// action can target any resource type. Errors if no dependency — or more than +// one — matches the name. +func lookupResource(deps resource.Dependencies, shortName string) (resource.Resource, error) { + var found resource.Resource + for n, r := range deps { + if n.Name != shortName { + continue + } + if found != nil { + return nil, fmt.Errorf("matches multiple dependencies") + } + found = r + } + if found == nil { + return nil, fmt.Errorf("not found in dependencies") + } + return found, nil +} + // run is the background polling loop. It exits when the resource is closed. func (m *sensorMonitor) run() { defer m.wg.Done() @@ -225,6 +329,8 @@ func (m *sensorMonitor) poll(ctx context.Context) { fired := cmp(value, rule.Threshold) notify := false + fireTrigger := false + fireResolve := false m.mu.Lock() st := &m.ruleStates[i] st.lastValue = value @@ -233,10 +339,13 @@ func (m *sensorMonitor) poll(ctx context.Context) { st.triggered = true st.lastNotified = now notify = true + fireTrigger = true case fired && st.triggered && m.cooldown > 0 && now.Sub(st.lastNotified) >= m.cooldown: st.lastNotified = now notify = true case !fired: + // Edge from triggered to cleared fires the resolve actions once. + fireResolve = st.triggered st.triggered = false } m.mu.Unlock() @@ -244,6 +353,30 @@ func (m *sensorMonitor) poll(ctx context.Context) { if notify { m.sendNotification(ctx, rule, value) } + if fireTrigger { + m.runActions(ctx, rule.OnTrigger) + } + if fireResolve { + m.runActions(ctx, rule.OnResolve) + } + } +} + +// runActions fires each action's DoCommand on its resolved resource. Best-effort: +// a failure is logged and never affects monitoring. +func (m *sensorMonitor) runActions(ctx context.Context, actions []Action) { + for _, a := range actions { + res, ok := m.actionResources[a.Resource] + if !ok { + // Every action resource is resolved at construction, so this is unexpected. + m.logger.Warnf("action resource %q not resolved; skipping", a.Resource) + continue + } + if _, err := res.DoCommand(ctx, a.Command); err != nil { + m.logger.Warnf("action DoCommand on %q failed: %v", a.Resource, err) + continue + } + m.logger.Infof("action fired on %q", a.Resource) } } diff --git a/resources/sensormonitor/sensormonitor_test.go b/resources/sensormonitor/sensormonitor_test.go index d4d279f..cb77ad8 100644 --- a/resources/sensormonitor/sensormonitor_test.go +++ b/resources/sensormonitor/sensormonitor_test.go @@ -83,14 +83,53 @@ func (f *fakeNotifier) texts() []string { return out } +// fakeTarget is a generic resource standing in for an action target. It records +// the DoCommands it receives so tests can assert what the monitor fired. +type fakeTarget struct { + resource.Named + resource.AlwaysRebuild + + mu sync.Mutex + received []map[string]interface{} +} + +func newFakeTarget(name string) *fakeTarget { + return &fakeTarget{Named: generic.Named(name).AsNamed()} +} + +func (f *fakeTarget) DoCommand(ctx context.Context, cmd map[string]interface{}) (map[string]interface{}, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.received = append(f.received, cmd) + return map[string]interface{}{}, nil +} + +func (f *fakeTarget) Close(context.Context) error { return nil } + +func (f *fakeTarget) commands() []map[string]interface{} { + f.mu.Lock() + defer f.mu.Unlock() + return append([]map[string]interface{}{}, f.received...) +} + // newTestMonitor builds a monitor wired to the given fakes WITHOUT starting the // background polling loop, so tests drive poll() deterministically. func newTestMonitor(t *testing.T, cfg *Config, src *fakeSensor, notifier *fakeNotifier) *sensorMonitor { + t.Helper() + return newTestMonitorWithTargets(t, cfg, src, notifier, nil) +} + +// newTestMonitorWithTargets is newTestMonitor plus a set of action-target +// resources, registered as generic dependencies under their names. +func newTestMonitorWithTargets(t *testing.T, cfg *Config, src *fakeSensor, notifier *fakeNotifier, targets map[string]*fakeTarget) *sensorMonitor { t.Helper() deps := resource.Dependencies{ sensor.Named(cfg.Sensor): src, generic.Named(cfg.Notifier): notifier, } + for name, tgt := range targets { + deps[generic.Named(name)] = tgt + } name := resource.NewName(sensor.API, "monitor") m, err := newMonitor(deps, name, cfg, logging.NewTestLogger(t)) if err != nil { @@ -137,6 +176,28 @@ func TestValidate(t *testing.T) { cfg: Config{Sensor: "s", Notifier: "n", Rules: []Rule{{Key: "t", Operator: "=>", Threshold: 1}}}, wantErr: true, }, + { + name: "with actions", + cfg: Config{Sensor: "s", Notifier: "n", Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1, + OnTrigger: []Action{{Resource: "r1", Command: map[string]interface{}{"x": 1}}}, + OnResolve: []Action{{Resource: "r2", Command: map[string]interface{}{"x": 0}}}, + }}}, + reqDeps: []string{"s", "n", "r1", "r2"}, + }, + { + name: "action missing resource", + cfg: Config{Sensor: "s", Notifier: "n", Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1, + OnTrigger: []Action{{Command: map[string]interface{}{"x": 1}}}, + }}}, + wantErr: true, + }, + { + name: "action missing command", + cfg: Config{Sensor: "s", Notifier: "n", Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1, + OnResolve: []Action{{Resource: "r1"}}, + }}}, + wantErr: true, + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -324,3 +385,101 @@ func TestDoCommandCheckForcesPoll(t *testing.T) { t.Fatal("expected error for unknown command") } } + +func TestActionsFireOnTriggerAndResolve(t *testing.T) { + ctx := context.Background() + src := newFakeSensor("src") + notifier := newFakeNotifier("notify") + relay := newFakeTarget("relay") + m := newTestMonitorWithTargets(t, &Config{ + Sensor: "src", + Notifier: "notify", + Rules: []Rule{{Key: "usage", Operator: ">=", Threshold: 15, + OnTrigger: []Action{{Resource: "relay", Command: map[string]interface{}{"set": true}}}, + OnResolve: []Action{{Resource: "relay", Command: map[string]interface{}{"set": false}}}, + }}, + }, src, notifier, map[string]*fakeTarget{"relay": relay}) + + // Below threshold: nothing fired. + src.set(map[string]interface{}{"usage": 5.0}) + m.poll(ctx) + if got := len(relay.commands()); got != 0 { + t.Fatalf("expected no actions below threshold, got %d", got) + } + + // Cross above: the on_trigger action fires once. + src.set(map[string]interface{}{"usage": 20.0}) + m.poll(ctx) + cmds := relay.commands() + if len(cmds) != 1 || cmds[0]["set"] != true { + t.Fatalf("expected one on_trigger command {set:true}, got %v", cmds) + } + + // Still triggered: on_trigger does not refire. + m.poll(ctx) + if got := len(relay.commands()); got != 1 { + t.Fatalf("expected no refire while triggered, got %d", got) + } + + // Reading returns below threshold: the on_resolve action fires once. + src.set(map[string]interface{}{"usage": 0.0}) + m.poll(ctx) + cmds = relay.commands() + if len(cmds) != 2 || cmds[1]["set"] != false { + t.Fatalf("expected on_resolve command {set:false}, got %v", cmds) + } + + // Stays resolved: on_resolve does not refire. + m.poll(ctx) + if got := len(relay.commands()); got != 2 { + t.Fatalf("expected no refire while resolved, got %d", got) + } +} + +func TestActionTargetsAnyResourceByName(t *testing.T) { + ctx := context.Background() + src := newFakeSensor("src") + notifier := newFakeNotifier("notify") + // Two distinct targets, fired on different edges. + a := newFakeTarget("alpha") + b := newFakeTarget("beta") + m := newTestMonitorWithTargets(t, &Config{ + Sensor: "src", + Notifier: "notify", + Rules: []Rule{{Key: "usage", Operator: ">=", Threshold: 15, + OnTrigger: []Action{{Resource: "alpha", Command: map[string]interface{}{"go": 1}}}, + OnResolve: []Action{{Resource: "beta", Command: map[string]interface{}{"go": 2}}}, + }}, + }, src, notifier, map[string]*fakeTarget{"alpha": a, "beta": b}) + + src.set(map[string]interface{}{"usage": 20.0}) + m.poll(ctx) + src.set(map[string]interface{}{"usage": 0.0}) + m.poll(ctx) + + if cmds := a.commands(); len(cmds) != 1 || cmds[0]["go"] != 1 { + t.Fatalf("alpha (on_trigger) not fired correctly: %v", cmds) + } + if cmds := b.commands(); len(cmds) != 1 || cmds[0]["go"] != 2 { + t.Fatalf("beta (on_resolve) not fired correctly: %v", cmds) + } +} + +func TestActionResourceMissingFromDeps(t *testing.T) { + src := newFakeSensor("src") + notifier := newFakeNotifier("notify") + deps := resource.Dependencies{ + sensor.Named("src"): src, + generic.Named("notify"): notifier, + } + cfg := &Config{ + Sensor: "src", + Notifier: "notify", + Rules: []Rule{{Key: "usage", Operator: ">=", Threshold: 15, + OnTrigger: []Action{{Resource: "ghost", Command: map[string]interface{}{"x": 1}}}, + }}, + } + if _, err := newMonitor(deps, resource.NewName(sensor.API, "monitor"), cfg, logging.NewTestLogger(t)); err == nil { + t.Fatal("expected construction to fail when an action resource is not a dependency") + } +} From 4bf78ac54de8217afe29fc018b16e777ca9ed04d Mon Sep 17 00:00:00 2001 From: danielbotros Date: Wed, 24 Jun 2026 13:21:33 -0400 Subject: [PATCH 2/4] sensor-monitor: trim action doc comments Co-Authored-By: Claude Opus 4.8 (1M context) --- resources/sensormonitor/sensormonitor.go | 16 +++++----------- 1 file changed, 5 insertions(+), 11 deletions(-) diff --git a/resources/sensormonitor/sensormonitor.go b/resources/sensormonitor/sensormonitor.go index a8f7170..a9ee893 100644 --- a/resources/sensormonitor/sensormonitor.go +++ b/resources/sensormonitor/sensormonitor.go @@ -36,14 +36,10 @@ func init() { // defaultPollInterval is used when poll_interval_seconds is not set. const defaultPollInterval = 10 * time.Second -// Action is a DoCommand to fire on a resource when a rule changes state. The -// call is always DoCommand, so the resource may be any type (component or -// service). +// Action is a DoCommand to fire on a resource when a rule changes state. type Action struct { - // Resource is the name of the resource to call. - Resource string `json:"resource"` - // Command is the DoCommand payload sent to the resource. - Command map[string]interface{} `json:"command"` + Resource string `json:"resource"` + Command map[string]interface{} `json:"command"` } // Rule describes a single numeric trigger on one reading key. @@ -60,8 +56,7 @@ type Rule struct { Message string `json:"message,omitempty"` // OnTrigger lists DoCommands to fire when the rule transitions to triggered. OnTrigger []Action `json:"on_trigger,omitempty"` - // OnResolve lists DoCommands to fire when the rule clears (its reading returns - // to the non-triggered side of the threshold). + // OnResolve lists DoCommands to fire when the rule clears OnResolve []Action `json:"on_resolve,omitempty"` } @@ -156,8 +151,7 @@ type sensorMonitor struct { sensorDep sensor.Sensor notifierDep generic.Service // actionResources holds every resource named by a rule action, resolved once - // at construction and keyed by name. Any resource type works — actions only - // ever call DoCommand. + // at construction and keyed by name. actionResources map[string]resource.Resource pollInterval time.Duration From 598fe9c3a60f2d3899c752e6da5413a1dae7968b Mon Sep 17 00:00:00 2001 From: danielbotros Date: Wed, 24 Jun 2026 16:11:08 -0400 Subject: [PATCH 3/4] sensor-monitor: actions-only model with capture/reference data passing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Remove the dedicated notifier (notifier field, message templating, sendNotification): notifying is now just an on_trigger action targeting a notification service. Each rule's on_trigger/on_resolve actions fire DoCommands on any resource, on the trigger edge (and cooldown repeats) and the resolve edge respectively. Action commands support ${...} references resolved before sending — against rule context (${value}, ${key}, ${threshold}, ${operator}) and against responses captured from earlier actions via "capture". This lets a message sent on trigger be reacted to on resolve by capturing the send response and referencing ${msg.ts}/${msg.channel}. An unresolved reference skips that action. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 62 +-- resources/sensormonitor/sensormonitor.go | 264 ++++++++---- resources/sensormonitor/sensormonitor_test.go | 406 +++++++++--------- 3 files changed, 427 insertions(+), 305 deletions(-) diff --git a/README.md b/README.md index b916e81..41a6e78 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,7 @@ The `viam:sensor-bundle` module provides these models: 1. **`viam:sensor-bundle:stateful-sensor`** - A sensor component that holds a value you set via `DoCommand`, serves it through the Readings API, and persists it to a file on disk so it survives restarts. -2. **`viam:sensor-bundle:sensor-monitor`** - A sensor component that watches another sensor's readings against numeric trigger rules and sends a notification via a generic service's `DoCommand` when a threshold is crossed. +2. **`viam:sensor-bundle:sensor-monitor`** - A sensor component that watches another sensor's readings against numeric trigger rules and fires `DoCommand` actions on other resources when a rule triggers or resolves. --- @@ -69,24 +69,26 @@ Returns: **API:** `rdk:component:sensor` -Monitors the readings of another sensor and fires notifications when a numeric reading crosses a configured threshold. It takes two dependencies — the sensor to watch and a generic service to notify — and evaluates a set of numeric trigger rules on every poll. +Monitors the readings of another sensor and, when a numeric reading crosses a configured threshold, fires `DoCommand` **actions** on other resources. It evaluates a set of numeric trigger rules on every poll; the only dependency is the sensor to watch (plus whatever resources the actions target). -When a rule fires, the monitor calls `DoCommand` on the notifier with the payload: +**Actions.** Each rule has an `on_trigger` and an `on_resolve` list. Every action names a `resource` (any component or service on the machine, by name) and a `command` (the `DoCommand` payload sent to it). Notifying is just an action — point one at a notification service with `{ "command": "send", "text": "..." }`. -```json -{"command": "send", "text": ""} -``` +- `on_trigger` fires when the rule transitions from not-triggered to triggered, and again at most once per `cooldown_seconds` while it stays triggered (with the default `cooldown_seconds: 0`, that's edge-only). +- `on_resolve` fires once when the reading clears the threshold. +- Actions are best-effort: a failure is logged and never affects monitoring. Each named resource is automatically added as a dependency. -Notifications are **edge-triggered**: a message is sent when a rule transitions from not-triggered to triggered. While the rule stays triggered no further message is sent, unless `cooldown_seconds` is set, in which case the message repeats at most once per cooldown window. When the reading clears the threshold and crosses it again, a new notification is sent. +**References and captures.** String values in a `command` may embed `${...}` references, resolved just before the command is sent: -**Actions.** Besides notifying, a rule can fire arbitrary `DoCommand`s on other resources when it triggers and when it resolves, via the `on_trigger` and `on_resolve` lists. Each action names a `resource` (any component or service on the machine, by name) and a `command` (the `DoCommand` payload). Actions fire on the edges only — once when the rule transitions to triggered, once when it clears — not on `cooldown_seconds` repeats. Each named resource is automatically added as a dependency. Actions are best-effort: a failure is logged and never affects monitoring. +- Rule/reading context: `${value}` (the current reading), `${key}`, `${threshold}`, `${operator}`. +- Captured responses: set `"capture": ""` on an action to store its `DoCommand` response, then reference its fields from later actions (including `on_resolve`) as `${.}`. + +A value that is exactly one reference keeps the referenced value's type; a reference embedded in a larger string is substituted as text. If a reference can't be resolved (e.g. nothing was captured), that action is skipped. This is how a message sent on trigger gets reacted to on resolve — capture the send response, then reference its `ts`/`channel` in the resolve action. A capture is overwritten on each cooldown re-fire (so resolve references the most recent response) and cleared once the rule resolves. ### Configuration ```json { "sensor": "", - "notifier": "", "poll_interval_seconds": , "cooldown_seconds": , "rules": [ @@ -94,8 +96,7 @@ Notifications are **edge-triggered**: a message is sent when a rule transitions "key": "", "operator": "", "threshold": , - "message": "", - "on_trigger": [{ "resource": "", "command": { } }], + "on_trigger": [{ "resource": "", "command": { }, "capture": "" }], "on_resolve": [{ "resource": "", "command": { } }] } ] @@ -105,10 +106,9 @@ Notifications are **edge-triggered**: a message is sent when a rule transitions | Name | Type | Required | Description | | ----------------------- | ------ | -------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------- | | `sensor` | string | **Yes** | Name of the sensor dependency whose readings are monitored. | -| `notifier` | string | **Yes** | Name of the generic service dependency that receives notification `DoCommand`s. | | `rules` | array | **Yes** | One or more numeric trigger rules (see below). At least one rule is required. | | `poll_interval_seconds` | number | No | How often the sensor is polled, in seconds. Defaults to `10`. | -| `cooldown_seconds` | number | No | Minimum time between repeat notifications while a rule stays triggered. `0` (default) means no repeat until the rule clears and fires again. | +| `cooldown_seconds` | number | No | Minimum time between repeat `on_trigger` firings while a rule stays triggered. `0` (default) means fire only on the edge. | Each entry in `rules`: @@ -117,31 +117,43 @@ Each entry in `rules`: | `key` | string | **Yes** | The reading key to watch, e.g. `"temperature"`. Non-numeric or missing values are skipped. | | `operator` | string | **Yes** | Comparison applied between the reading and `threshold`. One of `>`, `>=`, `<`, `<=`, `==`, `!=` (aliases: `gt`, `gte`, `lt`, `lte`, `eq`, `ne`). | | `threshold` | number | **Yes** | The value the reading is compared against. | -| `message` | string | No | Notification template. Supports placeholders `{key}`, `{value}`, `{threshold}`, `{operator}`. If omitted, a message like `temperature is 95 (> 90)` is used. | -| `on_trigger` | array | No | DoCommands to fire when the rule transitions to triggered. Each is `{ "resource": , "command": }`. | -| `on_resolve` | array | No | DoCommands to fire when the rule clears (reading returns to the non-triggered side of the threshold). Same shape as `on_trigger`. | +| `on_trigger` | array | No | Actions to fire when the rule triggers (and on each cooldown window). Each is `{ "resource": , "command": , "capture": }`. | +| `on_resolve` | array | No | Actions to fire when the rule clears. Same shape as `on_trigger`. | ### Example Configuration +A low-bean alert that posts to Slack on trigger and adds a ✅ reaction to that same message on resolve, plus a fan toggled directly: + ```json { - "sensor": "outdoor-temp", - "notifier": "slack-notifier", + "sensor": "usage-sensor", "poll_interval_seconds": 30, "cooldown_seconds": 3600, "rules": [ + { + "key": "regular_grinds", + "operator": ">=", + "threshold": 20, + "on_trigger": [ + { + "resource": "slack-notifier", + "command": { "command": "send", "text": "☕ Regular beans are low (${value}), please refill." }, + "capture": "msg" + } + ], + "on_resolve": [ + { + "resource": "slack-notifier", + "command": { "command": "react", "name": "white_check_mark", "ts": "${msg.ts}", "channel": "${msg.channel}" } + } + ] + }, { "key": "temperature", "operator": ">", "threshold": 90, - "message": "High temperature alert: {key} is {value} (limit {threshold})", "on_trigger": [{ "resource": "cooling-fan", "command": { "set": { "power": 1 } } }], "on_resolve": [{ "resource": "cooling-fan", "command": { "set": { "power": 0 } } }] - }, - { - "key": "humidity", - "operator": "<", - "threshold": 20 } ] } @@ -162,7 +174,7 @@ Returns the most recent readings from the monitored sensor, plus a `_trigge ### DoCommand -**`check`** - Force an immediate poll of the sensor (instead of waiting for the next interval), evaluating all rules and sending any notifications. +**`check`** - Force an immediate poll of the sensor (instead of waiting for the next interval), evaluating all rules and firing any actions. ```json {"check": true} diff --git a/resources/sensormonitor/sensormonitor.go b/resources/sensormonitor/sensormonitor.go index a9ee893..c5971c6 100644 --- a/resources/sensormonitor/sensormonitor.go +++ b/resources/sensormonitor/sensormonitor.go @@ -1,14 +1,15 @@ // Package sensormonitor implements the viam:sensor-bundle:sensor-monitor model: a -// sensor that watches another sensor's readings against numeric trigger rules, -// sends a notification via a generic service's DoCommand when a threshold is -// crossed, and can fire arbitrary DoCommands on other resources when a rule -// triggers or resolves. +// sensor that watches another sensor's readings against numeric trigger rules and +// fires configurable DoCommand actions on other resources when a rule triggers or +// resolves. An action can capture its response, and later actions can reference +// it — so, for example, a message sent on trigger can be reacted to on resolve. package sensormonitor import ( "context" "encoding/json" "fmt" + "regexp" "strconv" "strings" "sync" @@ -17,12 +18,11 @@ import ( sensor "go.viam.com/rdk/components/sensor" "go.viam.com/rdk/logging" "go.viam.com/rdk/resource" - generic "go.viam.com/rdk/services/generic" ) // Model is the sensor-monitor model triplet. It watches the readings of a sensor -// and fires notifications through a generic service when a numeric reading crosses -// a configured threshold. +// and fires DoCommand actions on other resources when a numeric reading crosses a +// configured threshold. var Model = resource.NewModel("viam", "sensor-bundle", "sensor-monitor") func init() { @@ -38,8 +38,14 @@ const defaultPollInterval = 10 * time.Second // Action is a DoCommand to fire on a resource when a rule changes state. type Action struct { - Resource string `json:"resource"` - Command map[string]interface{} `json:"command"` + // Resource is the name of the resource to call (any component or service). + Resource string `json:"resource"` + // Command is the DoCommand payload. String values may contain ${...} + // references that are resolved before the command is sent — see resolveValue. + Command map[string]interface{} `json:"command"` + // Capture, when set, stores this action's response under this name so later + // actions (including on_resolve) can reference its fields as ${name.field}. + Capture string `json:"capture,omitempty"` } // Rule describes a single numeric trigger on one reading key. @@ -51,12 +57,11 @@ type Rule struct { Operator string `json:"operator"` // Threshold is the value the reading is compared against. Threshold float64 `json:"threshold"` - // Message is an optional notification template. It supports the placeholders - // {key}, {value}, {threshold} and {operator}. If empty, a message is generated. - Message string `json:"message,omitempty"` - // OnTrigger lists DoCommands to fire when the rule transitions to triggered. + // OnTrigger lists actions to fire when the rule transitions to triggered, and + // again on each cooldown window while it stays triggered. OnTrigger []Action `json:"on_trigger,omitempty"` - // OnResolve lists DoCommands to fire when the rule clears + // OnResolve lists actions to fire when the rule clears (its reading returns to + // the non-triggered side of the threshold). OnResolve []Action `json:"on_resolve,omitempty"` } @@ -64,35 +69,27 @@ type Rule struct { type Config struct { // Sensor is the name of the sensor dependency whose readings are monitored. Sensor string `json:"sensor"` - // Notifier is the name of the generic service dependency that receives - // notification DoCommands of the form {"command": "send", "text": }. - Notifier string `json:"notifier"` // Rules is the set of numeric trigger rules. At least one is required. Rules []Rule `json:"rules"` // PollIntervalSec is how often the sensor is polled, in seconds. Defaults to 10. PollIntervalSec float64 `json:"poll_interval_seconds,omitempty"` - // CooldownSec is the minimum time between repeat notifications while a rule - // stays triggered, in seconds. 0 (default) means do not re-notify until the - // rule clears and fires again. + // CooldownSec is the minimum time between repeat on_trigger firings while a + // rule stays triggered, in seconds. 0 (default) means fire only on the edge. CooldownSec float64 `json:"cooldown_seconds,omitempty"` } -// Validate checks the config and returns the required dependency names. +// Validate checks the config and returns the required dependency names: the +// watched sensor plus every resource named by an action. func (cfg *Config) Validate(path string) ([]string, []string, error) { if cfg.Sensor == "" { return nil, nil, fmt.Errorf("%s: missing required field 'sensor'", path) } - if cfg.Notifier == "" { - return nil, nil, fmt.Errorf("%s: missing required field 'notifier'", path) - } if len(cfg.Rules) == 0 { return nil, nil, fmt.Errorf("%s: at least one rule is required", path) } - // Required deps: the watched sensor, the notifier, and every resource named by - // an action. Collected in a deterministic order with duplicates removed. - deps := []string{cfg.Sensor, cfg.Notifier} - seen := map[string]bool{cfg.Sensor: true, cfg.Notifier: true} + deps := []string{cfg.Sensor} + seen := map[string]bool{cfg.Sensor: true} addDep := func(name string) { if !seen[name] { seen[name] = true @@ -136,9 +133,13 @@ func validateAction(a Action) error { // ruleState tracks the runtime state of a single rule across polls. type ruleState struct { - triggered bool - lastNotified time.Time - lastValue float64 + triggered bool + lastFired time.Time + lastValue float64 + // vars holds the responses of actions that set "capture", keyed by capture + // name, so on_resolve actions can reference values produced on trigger. Reset + // when the rule resolves. + vars map[string]interface{} } type sensorMonitor struct { @@ -148,10 +149,10 @@ type sensorMonitor struct { logger logging.Logger cfg *Config - sensorDep sensor.Sensor - notifierDep generic.Service + sensorDep sensor.Sensor // actionResources holds every resource named by a rule action, resolved once - // at construction and keyed by name. + // at construction and keyed by name. Any resource type works — actions only + // ever call DoCommand. actionResources map[string]resource.Resource pollInterval time.Duration @@ -195,10 +196,6 @@ func newMonitor(deps resource.Dependencies, name resource.Name, conf *Config, lo if err != nil { return nil, fmt.Errorf("failed to get sensor dependency %q: %w", conf.Sensor, err) } - notifierDep, err := generic.FromProvider(deps, conf.Notifier) - if err != nil { - return nil, fmt.Errorf("failed to get notifier dependency %q: %w", conf.Notifier, err) - } // Resolve every resource named by an action up front (any API), so firing is a // simple map lookup + DoCommand. @@ -240,7 +237,6 @@ func newMonitor(deps resource.Dependencies, name resource.Name, conf *Config, lo logger: logger, cfg: conf, sensorDep: sensorDep, - notifierDep: notifierDep, actionResources: actionResources, pollInterval: pollInterval, cooldown: cooldown, @@ -291,7 +287,8 @@ func (m *sensorMonitor) run() { } } -// poll reads the sensor once, evaluates every rule, and sends notifications. +// poll reads the sensor once, evaluates every rule, and fires the rule's actions +// on the trigger and resolve edges (and on cooldown repeats while triggered). func (m *sensorMonitor) poll(ctx context.Context) { readings, err := m.sensorDep.Readings(ctx, nil) if err != nil { @@ -322,7 +319,6 @@ func (m *sensorMonitor) poll(ctx context.Context) { cmp, _ := parseOperator(rule.Operator) fired := cmp(value, rule.Threshold) - notify := false fireTrigger := false fireResolve := false m.mu.Lock() @@ -331,34 +327,56 @@ func (m *sensorMonitor) poll(ctx context.Context) { switch { case fired && !st.triggered: st.triggered = true - st.lastNotified = now - notify = true + st.lastFired = now + fireTrigger = true + case fired && st.triggered && m.cooldown > 0 && now.Sub(st.lastFired) >= m.cooldown: + st.lastFired = now fireTrigger = true - case fired && st.triggered && m.cooldown > 0 && now.Sub(st.lastNotified) >= m.cooldown: - st.lastNotified = now - notify = true case !fired: - // Edge from triggered to cleared fires the resolve actions once. fireResolve = st.triggered st.triggered = false } m.mu.Unlock() - if notify { - m.sendNotification(ctx, rule, value) - } if fireTrigger { - m.runActions(ctx, rule.OnTrigger) + m.runActions(ctx, i, rule, rule.OnTrigger, value) } if fireResolve { - m.runActions(ctx, rule.OnResolve) + m.runActions(ctx, i, rule, rule.OnResolve, value) + // The episode is over; drop captured vars so the next one starts fresh. + m.mu.Lock() + m.ruleStates[i].vars = nil + m.mu.Unlock() } } } -// runActions fires each action's DoCommand on its resolved resource. Best-effort: -// a failure is logged and never affects monitoring. -func (m *sensorMonitor) runActions(ctx context.Context, actions []Action) { +// runActions fires each action's DoCommand on its resolved resource. Before each +// call it resolves ${...} references in the command against the rule/reading +// context and any captured responses; after a successful call it stores the +// response under the action's capture name (if set) for later actions to +// reference. Best-effort: an unresolved reference, missing resource, or DoCommand +// error logs a warning and skips that action without affecting monitoring. +func (m *sensorMonitor) runActions(ctx context.Context, ruleIdx int, rule Rule, actions []Action, value float64) { + if len(actions) == 0 { + return + } + + // Substitution context: the rule/reading values plus any previously captured + // responses for this rule. + subCtx := map[string]interface{}{ + "value": value, + "key": rule.Key, + "threshold": rule.Threshold, + "operator": rule.Operator, + } + m.mu.RLock() + for k, v := range m.ruleStates[ruleIdx].vars { + subCtx[k] = v + } + m.mu.RUnlock() + + captured := map[string]interface{}{} for _, a := range actions { res, ok := m.actionResources[a.Resource] if !ok { @@ -366,26 +384,36 @@ func (m *sensorMonitor) runActions(ctx context.Context, actions []Action) { m.logger.Warnf("action resource %q not resolved; skipping", a.Resource) continue } - if _, err := res.DoCommand(ctx, a.Command); err != nil { + resolved, err := resolveValue(a.Command, subCtx) + if err != nil { + m.logger.Warnf("skipping action on %q: %v", a.Resource, err) + continue + } + cmd, _ := resolved.(map[string]interface{}) + resp, err := res.DoCommand(ctx, cmd) + if err != nil { m.logger.Warnf("action DoCommand on %q failed: %v", a.Resource, err) continue } + if a.Capture != "" { + // Available to later actions in this batch and (after the loop) persisted + // for on_resolve. + subCtx[a.Capture] = resp + captured[a.Capture] = resp + } m.logger.Infof("action fired on %q", a.Resource) } -} -// sendNotification renders the rule's message and calls DoCommand on the notifier. -func (m *sensorMonitor) sendNotification(ctx context.Context, rule Rule, value float64) { - msg := renderMessage(rule, value) - cmd := map[string]interface{}{ - "command": "send", - "text": msg, - } - if _, err := m.notifierDep.DoCommand(ctx, cmd); err != nil { - m.logger.Errorf("failed to notify via %q: %v", m.cfg.Notifier, err) - return + if len(captured) > 0 { + m.mu.Lock() + if m.ruleStates[ruleIdx].vars == nil { + m.ruleStates[ruleIdx].vars = map[string]interface{}{} + } + for k, v := range captured { + m.ruleStates[ruleIdx].vars[k] = v + } + m.mu.Unlock() } - m.logger.Infof("notification sent: %s", msg) } // Readings returns the most recent sensor readings plus, per rule, a @@ -463,26 +491,88 @@ func toFloat64(v interface{}) (float64, bool) { } } -// renderMessage builds the notification text for a fired rule. -func renderMessage(rule Rule, value float64) string { - if rule.Message == "" { - return fmt.Sprintf("%s is %s (%s %s)", - rule.Key, - formatFloat(value), - rule.Operator, - formatFloat(rule.Threshold), - ) +// refExact matches a string that is exactly one reference, e.g. "${msg.ts}". +// refAny matches every reference embedded anywhere in a string. +var ( + refExact = regexp.MustCompile(`^\$\{([^}]+)\}$`) + refAny = regexp.MustCompile(`\$\{([^}]+)\}`) +) + +// resolveValue walks a command value and substitutes ${name} / ${name.path} +// references against ctx. A value that is exactly one reference keeps the +// referenced value's type (so a captured map or number passes through intact); a +// reference embedded in a larger string is substituted textually. An unresolved +// reference returns an error so the caller can skip the action. +func resolveValue(v interface{}, ctx map[string]interface{}) (interface{}, error) { + switch x := v.(type) { + case string: + if mm := refExact.FindStringSubmatch(x); mm != nil { + val, ok := lookupPath(ctx, mm[1]) + if !ok { + return nil, fmt.Errorf("unresolved reference ${%s}", mm[1]) + } + return val, nil + } + var refErr error + out := refAny.ReplaceAllStringFunc(x, func(match string) string { + path := refAny.FindStringSubmatch(match)[1] + val, ok := lookupPath(ctx, path) + if !ok { + refErr = fmt.Errorf("unresolved reference ${%s}", path) + return match + } + return stringify(val) + }) + if refErr != nil { + return nil, refErr + } + return out, nil + case map[string]interface{}: + out := make(map[string]interface{}, len(x)) + for k, val := range x { + rv, err := resolveValue(val, ctx) + if err != nil { + return nil, err + } + out[k] = rv + } + return out, nil + case []interface{}: + out := make([]interface{}, len(x)) + for i, e := range x { + rv, err := resolveValue(e, ctx) + if err != nil { + return nil, err + } + out[i] = rv + } + return out, nil + default: + return v, nil } - r := strings.NewReplacer( - "{key}", rule.Key, - "{value}", formatFloat(value), - "{threshold}", formatFloat(rule.Threshold), - "{operator}", rule.Operator, - ) - return r.Replace(rule.Message) } -// formatFloat renders a float without a trailing ".0" for whole numbers. -func formatFloat(f float64) string { - return strconv.FormatFloat(f, 'f', -1, 64) +// lookupPath walks a dot-separated path into a context map. +func lookupPath(ctx map[string]interface{}, path string) (interface{}, bool) { + var cur interface{} = ctx + for _, part := range strings.Split(path, ".") { + m, ok := cur.(map[string]interface{}) + if !ok { + return nil, false + } + cur, ok = m[part] + if !ok { + return nil, false + } + } + return cur, true +} + +// stringify renders a referenced value for embedding in a larger string, +// trimming trailing zeros from floats. +func stringify(v interface{}) string { + if f, ok := v.(float64); ok { + return strconv.FormatFloat(f, 'f', -1, 64) + } + return fmt.Sprint(v) } diff --git a/resources/sensormonitor/sensormonitor_test.go b/resources/sensormonitor/sensormonitor_test.go index cb77ad8..ff6b36b 100644 --- a/resources/sensormonitor/sensormonitor_test.go +++ b/resources/sensormonitor/sensormonitor_test.go @@ -4,6 +4,7 @@ import ( "context" "sync" "testing" + "time" sensor "go.viam.com/rdk/components/sensor" "go.viam.com/rdk/logging" @@ -45,52 +46,16 @@ func (f *fakeSensor) Readings(ctx context.Context, extra map[string]interface{}) func (f *fakeSensor) Close(context.Context) error { return nil } -// fakeNotifier is a generic.Service that records the DoCommands it receives. -type fakeNotifier struct { - resource.Named - resource.AlwaysRebuild - - mu sync.Mutex - received []map[string]interface{} -} - -func newFakeNotifier(name string) *fakeNotifier { - return &fakeNotifier{Named: generic.Named(name).AsNamed()} -} - -func (f *fakeNotifier) DoCommand(ctx context.Context, cmd map[string]interface{}) (map[string]interface{}, error) { - f.mu.Lock() - defer f.mu.Unlock() - f.received = append(f.received, cmd) - return map[string]interface{}{}, nil -} - -func (f *fakeNotifier) Close(context.Context) error { return nil } - -// texts returns the notification message strings received so far. -func (f *fakeNotifier) texts() []string { - f.mu.Lock() - defer f.mu.Unlock() - out := make([]string, 0, len(f.received)) - for _, cmd := range f.received { - if cmd["command"] != "send" { - continue - } - if text, ok := cmd["text"].(string); ok { - out = append(out, text) - } - } - return out -} - // fakeTarget is a generic resource standing in for an action target. It records -// the DoCommands it receives so tests can assert what the monitor fired. +// the DoCommands it receives and returns a configurable response, so tests can +// assert what the monitor fired and exercise the capture/reference mechanism. type fakeTarget struct { resource.Named resource.AlwaysRebuild mu sync.Mutex received []map[string]interface{} + resp map[string]interface{} } func newFakeTarget(name string) *fakeTarget { @@ -101,6 +66,9 @@ func (f *fakeTarget) DoCommand(ctx context.Context, cmd map[string]interface{}) f.mu.Lock() defer f.mu.Unlock() f.received = append(f.received, cmd) + if f.resp != nil { + return f.resp, nil + } return map[string]interface{}{}, nil } @@ -112,20 +80,13 @@ func (f *fakeTarget) commands() []map[string]interface{} { return append([]map[string]interface{}{}, f.received...) } -// newTestMonitor builds a monitor wired to the given fakes WITHOUT starting the -// background polling loop, so tests drive poll() deterministically. -func newTestMonitor(t *testing.T, cfg *Config, src *fakeSensor, notifier *fakeNotifier) *sensorMonitor { - t.Helper() - return newTestMonitorWithTargets(t, cfg, src, notifier, nil) -} - -// newTestMonitorWithTargets is newTestMonitor plus a set of action-target -// resources, registered as generic dependencies under their names. -func newTestMonitorWithTargets(t *testing.T, cfg *Config, src *fakeSensor, notifier *fakeNotifier, targets map[string]*fakeTarget) *sensorMonitor { +// newTestMonitor builds a monitor wired to the given sensor and action targets +// WITHOUT starting the background polling loop, so tests drive poll() +// deterministically. +func newTestMonitor(t *testing.T, cfg *Config, src *fakeSensor, targets map[string]*fakeTarget) *sensorMonitor { t.Helper() deps := resource.Dependencies{ - sensor.Named(cfg.Sensor): src, - generic.Named(cfg.Notifier): notifier, + sensor.Named(cfg.Sensor): src, } for name, tgt := range targets { deps[generic.Named(name)] = tgt @@ -148,52 +109,47 @@ func TestValidate(t *testing.T) { }{ { name: "valid", - cfg: Config{Sensor: "s", Notifier: "n", Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1}}}, - reqDeps: []string{"s", "n"}, + cfg: Config{Sensor: "s", Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1}}}, + reqDeps: []string{"s"}, }, { name: "missing sensor", - cfg: Config{Notifier: "n", Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1}}}, - wantErr: true, - }, - { - name: "missing notifier", - cfg: Config{Sensor: "s", Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1}}}, + cfg: Config{Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1}}}, wantErr: true, }, { name: "no rules", - cfg: Config{Sensor: "s", Notifier: "n"}, + cfg: Config{Sensor: "s"}, wantErr: true, }, { name: "rule missing key", - cfg: Config{Sensor: "s", Notifier: "n", Rules: []Rule{{Operator: ">", Threshold: 1}}}, + cfg: Config{Sensor: "s", Rules: []Rule{{Operator: ">", Threshold: 1}}}, wantErr: true, }, { name: "rule bad operator", - cfg: Config{Sensor: "s", Notifier: "n", Rules: []Rule{{Key: "t", Operator: "=>", Threshold: 1}}}, + cfg: Config{Sensor: "s", Rules: []Rule{{Key: "t", Operator: "=>", Threshold: 1}}}, wantErr: true, }, { name: "with actions", - cfg: Config{Sensor: "s", Notifier: "n", Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1, + cfg: Config{Sensor: "s", Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1, OnTrigger: []Action{{Resource: "r1", Command: map[string]interface{}{"x": 1}}}, OnResolve: []Action{{Resource: "r2", Command: map[string]interface{}{"x": 0}}}, }}}, - reqDeps: []string{"s", "n", "r1", "r2"}, + reqDeps: []string{"s", "r1", "r2"}, }, { name: "action missing resource", - cfg: Config{Sensor: "s", Notifier: "n", Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1, + cfg: Config{Sensor: "s", Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1, OnTrigger: []Action{{Command: map[string]interface{}{"x": 1}}}, }}}, wantErr: true, }, { name: "action missing command", - cfg: Config{Sensor: "s", Notifier: "n", Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1, + cfg: Config{Sensor: "s", Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1, OnResolve: []Action{{Resource: "r1"}}, }}}, wantErr: true, @@ -283,122 +239,79 @@ func TestToFloat64(t *testing.T) { } } -func TestRenderMessage(t *testing.T) { - value := 95.0 - def := renderMessage(Rule{Key: "temperature", Operator: ">", Threshold: 90}, value) - if def != "temperature is 95 (> 90)" { - t.Fatalf("default message = %q", def) - } - custom := renderMessage(Rule{Key: "temperature", Operator: ">", Threshold: 90, Message: "ALERT {key}={value} over {threshold}"}, value) - if custom != "ALERT temperature=95 over 90" { - t.Fatalf("custom message = %q", custom) - } -} - -func TestEdgeTriggeredNotifies(t *testing.T) { - ctx := context.Background() - src := newFakeSensor("src") - src.set(map[string]interface{}{"temperature": 50.0}) - notifier := newFakeNotifier("notify") - - m := newTestMonitor(t, &Config{ - Sensor: "src", - Notifier: "notify", - Rules: []Rule{{Key: "temperature", Operator: ">", Threshold: 90}}, - }, src, notifier) - - // Below threshold: no notification. - src.set(map[string]interface{}{"temperature": 50.0}) - m.poll(ctx) - if got := len(notifier.texts()); got != 0 { - t.Fatalf("expected 0 notifications below threshold, got %d", got) +func TestResolveValue(t *testing.T) { + ctx := map[string]interface{}{ + "value": 95.0, + "threshold": 90.0, + "key": "temperature", + "msg": map[string]interface{}{"ts": "100.1", "channel": "C1"}, } - // Cross above threshold: exactly one notification. - src.set(map[string]interface{}{"temperature": 95.0}) - m.poll(ctx) - if got := notifier.texts(); len(got) != 1 { - t.Fatalf("expected 1 notification on edge, got %d: %v", len(got), got) - } - - // Still above threshold, cooldown 0: no repeat. - m.poll(ctx) - m.poll(ctx) - if got := len(notifier.texts()); got != 1 { - t.Fatalf("expected no repeat while triggered, got %d", got) - } - - // Clear then re-fire: a second notification. - src.set(map[string]interface{}{"temperature": 50.0}) - m.poll(ctx) - src.set(map[string]interface{}{"temperature": 99.0}) - m.poll(ctx) - if got := len(notifier.texts()); got != 2 { - t.Fatalf("expected 2 notifications after re-fire, got %d", got) - } -} - -func TestReadingsExposesTriggerState(t *testing.T) { - ctx := context.Background() - src := newFakeSensor("src") - notifier := newFakeNotifier("notify") - m := newTestMonitor(t, &Config{ - Sensor: "src", - Notifier: "notify", - Rules: []Rule{{Key: "humidity", Operator: "<", Threshold: 30}}, - }, src, notifier) - - src.set(map[string]interface{}{"humidity": 20.0}) - m.poll(ctx) - - got, err := m.Readings(ctx, nil) - if err != nil { - t.Fatalf("Readings: %v", err) - } - if got["humidity"] != 20.0 { - t.Fatalf("expected humidity reading passed through, got %v", got["humidity"]) - } - if triggered, ok := got["humidity_triggered"].(bool); !ok || !triggered { - t.Fatalf("expected humidity_triggered=true, got %v", got["humidity_triggered"]) - } -} + t.Run("exact reference keeps type", func(t *testing.T) { + got, err := resolveValue("${value}", ctx) + if err != nil { + t.Fatal(err) + } + if f, ok := got.(float64); !ok || f != 95.0 { + t.Fatalf("expected float64 95, got %T %v", got, got) + } + }) -func TestDoCommandCheckForcesPoll(t *testing.T) { - ctx := context.Background() - src := newFakeSensor("src") - src.set(map[string]interface{}{"temperature": 95.0}) - notifier := newFakeNotifier("notify") - m := newTestMonitor(t, &Config{ - Sensor: "src", - Notifier: "notify", - Rules: []Rule{{Key: "temperature", Operator: ">", Threshold: 90}}, - }, src, notifier) + t.Run("exact reference into capture", func(t *testing.T) { + got, err := resolveValue("${msg.ts}", ctx) + if err != nil { + t.Fatal(err) + } + if got != "100.1" { + t.Fatalf("expected \"100.1\", got %v", got) + } + }) - if _, err := m.DoCommand(ctx, map[string]interface{}{"check": true}); err != nil { - t.Fatalf("DoCommand check: %v", err) - } - if got := len(notifier.texts()); got == 0 { - t.Fatal("expected a notification after forced check") - } + t.Run("embedded references are stringified", func(t *testing.T) { + got, err := resolveValue("temp ${value} over ${threshold}", ctx) + if err != nil { + t.Fatal(err) + } + if got != "temp 95 over 90" { + t.Fatalf("got %q", got) + } + }) + + t.Run("recurses into maps and slices", func(t *testing.T) { + got, err := resolveValue(map[string]interface{}{ + "a": "${value}", + "b": []interface{}{"${key}"}, + }, ctx) + if err != nil { + t.Fatal(err) + } + m := got.(map[string]interface{}) + if m["a"] != 95.0 { + t.Fatalf("nested map not resolved: %v", m["a"]) + } + if m["b"].([]interface{})[0] != "temperature" { + t.Fatalf("nested slice not resolved: %v", m["b"]) + } + }) - if _, err := m.DoCommand(ctx, map[string]interface{}{"bogus": 1}); err == nil { - t.Fatal("expected error for unknown command") - } + t.Run("unresolved reference errors", func(t *testing.T) { + if _, err := resolveValue("${nope.field}", ctx); err == nil { + t.Fatal("expected error for unresolved reference") + } + }) } func TestActionsFireOnTriggerAndResolve(t *testing.T) { ctx := context.Background() src := newFakeSensor("src") - notifier := newFakeNotifier("notify") relay := newFakeTarget("relay") - m := newTestMonitorWithTargets(t, &Config{ - Sensor: "src", - Notifier: "notify", + m := newTestMonitor(t, &Config{ + Sensor: "src", Rules: []Rule{{Key: "usage", Operator: ">=", Threshold: 15, OnTrigger: []Action{{Resource: "relay", Command: map[string]interface{}{"set": true}}}, OnResolve: []Action{{Resource: "relay", Command: map[string]interface{}{"set": false}}}, }}, - }, src, notifier, map[string]*fakeTarget{"relay": relay}) + }, src, map[string]*fakeTarget{"relay": relay}) // Below threshold: nothing fired. src.set(map[string]interface{}{"usage": 5.0}) @@ -407,7 +320,7 @@ func TestActionsFireOnTriggerAndResolve(t *testing.T) { t.Fatalf("expected no actions below threshold, got %d", got) } - // Cross above: the on_trigger action fires once. + // Cross above: on_trigger fires once. src.set(map[string]interface{}{"usage": 20.0}) m.poll(ctx) cmds := relay.commands() @@ -415,13 +328,13 @@ func TestActionsFireOnTriggerAndResolve(t *testing.T) { t.Fatalf("expected one on_trigger command {set:true}, got %v", cmds) } - // Still triggered: on_trigger does not refire. + // Still triggered, cooldown 0: no refire. m.poll(ctx) if got := len(relay.commands()); got != 1 { t.Fatalf("expected no refire while triggered, got %d", got) } - // Reading returns below threshold: the on_resolve action fires once. + // Reading returns below threshold: on_resolve fires once. src.set(map[string]interface{}{"usage": 0.0}) m.poll(ctx) cmds = relay.commands() @@ -429,52 +342,112 @@ func TestActionsFireOnTriggerAndResolve(t *testing.T) { t.Fatalf("expected on_resolve command {set:false}, got %v", cmds) } - // Stays resolved: on_resolve does not refire. + // Stays resolved: no refire. m.poll(ctx) if got := len(relay.commands()); got != 2 { t.Fatalf("expected no refire while resolved, got %d", got) } } -func TestActionTargetsAnyResourceByName(t *testing.T) { +func TestOnTriggerRepeatsOnCooldown(t *testing.T) { ctx := context.Background() src := newFakeSensor("src") - notifier := newFakeNotifier("notify") - // Two distinct targets, fired on different edges. - a := newFakeTarget("alpha") - b := newFakeTarget("beta") - m := newTestMonitorWithTargets(t, &Config{ - Sensor: "src", - Notifier: "notify", + tgt := newFakeTarget("tgt") + m := newTestMonitor(t, &Config{ + Sensor: "src", + CooldownSec: 60, // long; we backdate lastFired to force a repeat Rules: []Rule{{Key: "usage", Operator: ">=", Threshold: 15, - OnTrigger: []Action{{Resource: "alpha", Command: map[string]interface{}{"go": 1}}}, - OnResolve: []Action{{Resource: "beta", Command: map[string]interface{}{"go": 2}}}, + OnTrigger: []Action{{Resource: "tgt", Command: map[string]interface{}{"ping": 1}}}, }}, - }, src, notifier, map[string]*fakeTarget{"alpha": a, "beta": b}) + }, src, map[string]*fakeTarget{"tgt": tgt}) + + src.set(map[string]interface{}{"usage": 20.0}) + m.poll(ctx) // edge fire + if got := len(tgt.commands()); got != 1 { + t.Fatalf("expected 1 fire on edge, got %d", got) + } + + // Backdate so the cooldown has elapsed; next poll re-fires. + m.ruleStates[0].lastFired = m.ruleStates[0].lastFired.Add(-time.Hour) + m.poll(ctx) + if got := len(tgt.commands()); got != 2 { + t.Fatalf("expected a cooldown repeat, got %d", got) + } +} + +func TestCaptureCarriesAcrossTriggerAndResolve(t *testing.T) { + ctx := context.Background() + src := newFakeSensor("src") + // The notifier returns a message identity on send; the monitor must carry it + // to the react command on resolve. + notifier := newFakeTarget("notifier") + notifier.resp = map[string]interface{}{"ok": true, "ts": "100.1", "channel": "C1"} + m := newTestMonitor(t, &Config{ + Sensor: "src", + Rules: []Rule{{Key: "usage", Operator: ">=", Threshold: 15, + OnTrigger: []Action{{ + Resource: "notifier", + Command: map[string]interface{}{"command": "send", "text": "low (${value})"}, + Capture: "msg", + }}, + OnResolve: []Action{{ + Resource: "notifier", + Command: map[string]interface{}{ + "command": "react", "name": "white_check_mark", + "ts": "${msg.ts}", "channel": "${msg.channel}", + }, + }}, + }}, + }, src, map[string]*fakeTarget{"notifier": notifier}) src.set(map[string]interface{}{"usage": 20.0}) m.poll(ctx) src.set(map[string]interface{}{"usage": 0.0}) m.poll(ctx) - if cmds := a.commands(); len(cmds) != 1 || cmds[0]["go"] != 1 { - t.Fatalf("alpha (on_trigger) not fired correctly: %v", cmds) + cmds := notifier.commands() + if len(cmds) != 2 { + t.Fatalf("expected send then react, got %v", cmds) + } + if cmds[0]["command"] != "send" || cmds[0]["text"] != "low (20)" { + t.Fatalf("send command not as expected: %v", cmds[0]) } - if cmds := b.commands(); len(cmds) != 1 || cmds[0]["go"] != 2 { - t.Fatalf("beta (on_resolve) not fired correctly: %v", cmds) + react := cmds[1] + if react["command"] != "react" || react["name"] != "white_check_mark" || + react["ts"] != "100.1" || react["channel"] != "C1" { + t.Fatalf("react command did not carry captured message: %v", react) } } -func TestActionResourceMissingFromDeps(t *testing.T) { +func TestResolveActionSkippedWhenCaptureMissing(t *testing.T) { + ctx := context.Background() src := newFakeSensor("src") - notifier := newFakeNotifier("notify") - deps := resource.Dependencies{ - sensor.Named("src"): src, - generic.Named("notify"): notifier, + notifier := newFakeTarget("notifier") + // on_trigger does NOT capture, so the on_resolve reference can't resolve. + m := newTestMonitor(t, &Config{ + Sensor: "src", + Rules: []Rule{{Key: "usage", Operator: ">=", Threshold: 15, + OnTrigger: []Action{{Resource: "notifier", Command: map[string]interface{}{"command": "send"}}}, + OnResolve: []Action{{Resource: "notifier", Command: map[string]interface{}{"command": "react", "ts": "${msg.ts}"}}}, + }}, + }, src, map[string]*fakeTarget{"notifier": notifier}) + + src.set(map[string]interface{}{"usage": 20.0}) + m.poll(ctx) + src.set(map[string]interface{}{"usage": 0.0}) + m.poll(ctx) + + cmds := notifier.commands() + if len(cmds) != 1 || cmds[0]["command"] != "send" { + t.Fatalf("expected only the send (react skipped, unresolved ref), got %v", cmds) } +} + +func TestActionResourceMissingFromDeps(t *testing.T) { + src := newFakeSensor("src") + deps := resource.Dependencies{sensor.Named("src"): src} cfg := &Config{ - Sensor: "src", - Notifier: "notify", + Sensor: "src", Rules: []Rule{{Key: "usage", Operator: ">=", Threshold: 15, OnTrigger: []Action{{Resource: "ghost", Command: map[string]interface{}{"x": 1}}}, }}, @@ -483,3 +456,50 @@ func TestActionResourceMissingFromDeps(t *testing.T) { t.Fatal("expected construction to fail when an action resource is not a dependency") } } + +func TestReadingsExposesTriggerState(t *testing.T) { + ctx := context.Background() + src := newFakeSensor("src") + m := newTestMonitor(t, &Config{ + Sensor: "src", + Rules: []Rule{{Key: "humidity", Operator: "<", Threshold: 30}}, + }, src, nil) + + src.set(map[string]interface{}{"humidity": 20.0}) + m.poll(ctx) + + got, err := m.Readings(ctx, nil) + if err != nil { + t.Fatalf("Readings: %v", err) + } + if got["humidity"] != 20.0 { + t.Fatalf("expected humidity reading passed through, got %v", got["humidity"]) + } + if triggered, ok := got["humidity_triggered"].(bool); !ok || !triggered { + t.Fatalf("expected humidity_triggered=true, got %v", got["humidity_triggered"]) + } +} + +func TestDoCommandCheckForcesPoll(t *testing.T) { + ctx := context.Background() + src := newFakeSensor("src") + src.set(map[string]interface{}{"temperature": 95.0}) + tgt := newFakeTarget("tgt") + m := newTestMonitor(t, &Config{ + Sensor: "src", + Rules: []Rule{{Key: "temperature", Operator: ">", Threshold: 90, + OnTrigger: []Action{{Resource: "tgt", Command: map[string]interface{}{"go": 1}}}, + }}, + }, src, map[string]*fakeTarget{"tgt": tgt}) + + if _, err := m.DoCommand(ctx, map[string]interface{}{"check": true}); err != nil { + t.Fatalf("DoCommand check: %v", err) + } + if got := len(tgt.commands()); got == 0 { + t.Fatal("expected an action to fire after forced check") + } + + if _, err := m.DoCommand(ctx, map[string]interface{}{"bogus": 1}); err == nil { + t.Fatal("expected error for unknown command") + } +} From 601fbd9809c9e191a013854fdab61f9101b8dd91 Mon Sep 17 00:00:00 2001 From: danielbotros Date: Wed, 24 Jun 2026 17:22:06 -0400 Subject: [PATCH 4/4] sensor-monitor: trim doc comments Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 4 ++-- resources/sensormonitor/sensormonitor.go | 22 ++++++++-------------- 2 files changed, 10 insertions(+), 16 deletions(-) diff --git a/README.md b/README.md index 41a6e78..90c3719 100644 --- a/README.md +++ b/README.md @@ -71,7 +71,7 @@ Returns: Monitors the readings of another sensor and, when a numeric reading crosses a configured threshold, fires `DoCommand` **actions** on other resources. It evaluates a set of numeric trigger rules on every poll; the only dependency is the sensor to watch (plus whatever resources the actions target). -**Actions.** Each rule has an `on_trigger` and an `on_resolve` list. Every action names a `resource` (any component or service on the machine, by name) and a `command` (the `DoCommand` payload sent to it). Notifying is just an action — point one at a notification service with `{ "command": "send", "text": "..." }`. +**Actions.** Each rule has an `on_trigger` and an `on_resolve` list. Every action names a `resource` (any component or service on the machine, by name) and a `command` (the `DoCommand` payload sent to it). - `on_trigger` fires when the rule transitions from not-triggered to triggered, and again at most once per `cooldown_seconds` while it stays triggered (with the default `cooldown_seconds: 0`, that's edge-only). - `on_resolve` fires once when the reading clears the threshold. @@ -82,7 +82,7 @@ Monitors the readings of another sensor and, when a numeric reading crosses a co - Rule/reading context: `${value}` (the current reading), `${key}`, `${threshold}`, `${operator}`. - Captured responses: set `"capture": ""` on an action to store its `DoCommand` response, then reference its fields from later actions (including `on_resolve`) as `${.}`. -A value that is exactly one reference keeps the referenced value's type; a reference embedded in a larger string is substituted as text. If a reference can't be resolved (e.g. nothing was captured), that action is skipped. This is how a message sent on trigger gets reacted to on resolve — capture the send response, then reference its `ts`/`channel` in the resolve action. A capture is overwritten on each cooldown re-fire (so resolve references the most recent response) and cleared once the rule resolves. +A value that is exactly one reference keeps the referenced value's type; a reference embedded in a larger string is substituted as text. If a reference can't be resolved (e.g. nothing was captured), that action is skipped. A capture is overwritten on each cooldown re-fire (so resolve references the most recent response) and cleared once the rule resolves. ### Configuration diff --git a/resources/sensormonitor/sensormonitor.go b/resources/sensormonitor/sensormonitor.go index c5971c6..fe49f31 100644 --- a/resources/sensormonitor/sensormonitor.go +++ b/resources/sensormonitor/sensormonitor.go @@ -1,8 +1,7 @@ // Package sensormonitor implements the viam:sensor-bundle:sensor-monitor model: a // sensor that watches another sensor's readings against numeric trigger rules and // fires configurable DoCommand actions on other resources when a rule triggers or -// resolves. An action can capture its response, and later actions can reference -// it — so, for example, a message sent on trigger can be reacted to on resolve. +// resolves. package sensormonitor import ( @@ -38,13 +37,13 @@ const defaultPollInterval = 10 * time.Second // Action is a DoCommand to fire on a resource when a rule changes state. type Action struct { - // Resource is the name of the resource to call (any component or service). + // Resource is the name of the resource to call. Resource string `json:"resource"` // Command is the DoCommand payload. String values may contain ${...} // references that are resolved before the command is sent — see resolveValue. Command map[string]interface{} `json:"command"` // Capture, when set, stores this action's response under this name so later - // actions (including on_resolve) can reference its fields as ${name.field}. + // actions can reference its fields as ${name.field}. Capture string `json:"capture,omitempty"` } @@ -78,8 +77,7 @@ type Config struct { CooldownSec float64 `json:"cooldown_seconds,omitempty"` } -// Validate checks the config and returns the required dependency names: the -// watched sensor plus every resource named by an action. +// Validate checks the config and returns the required dependency names. func (cfg *Config) Validate(path string) ([]string, []string, error) { if cfg.Sensor == "" { return nil, nil, fmt.Errorf("%s: missing required field 'sensor'", path) @@ -137,8 +135,7 @@ type ruleState struct { lastFired time.Time lastValue float64 // vars holds the responses of actions that set "capture", keyed by capture - // name, so on_resolve actions can reference values produced on trigger. Reset - // when the rule resolves. + // name, so later actions can reference values produced. Reset when the rule resolves. vars map[string]interface{} } @@ -151,8 +148,7 @@ type sensorMonitor struct { sensorDep sensor.Sensor // actionResources holds every resource named by a rule action, resolved once - // at construction and keyed by name. Any resource type works — actions only - // ever call DoCommand. + // at construction and keyed by name. actionResources map[string]resource.Resource pollInterval time.Duration @@ -197,8 +193,7 @@ func newMonitor(deps resource.Dependencies, name resource.Name, conf *Config, lo return nil, fmt.Errorf("failed to get sensor dependency %q: %w", conf.Sensor, err) } - // Resolve every resource named by an action up front (any API), so firing is a - // simple map lookup + DoCommand. + // Resolve every resource named by an action up front so firing is a simple map lookup + DoCommand. actionResources := map[string]resource.Resource{} resolveAction := func(a Action) error { if _, ok := actionResources[a.Resource]; ok { @@ -287,8 +282,7 @@ func (m *sensorMonitor) run() { } } -// poll reads the sensor once, evaluates every rule, and fires the rule's actions -// on the trigger and resolve edges (and on cooldown repeats while triggered). +// poll reads the sensor once, evaluates every rule, and fires the rule's actions. func (m *sensorMonitor) poll(ctx context.Context) { readings, err := m.sensorDep.Readings(ctx, nil) if err != nil {