feat(gateway): automation rule engine and reminder scheduler
- engine: 5-field cron parser (*/n, ranges, lists) firing at most once per minute; device-state triggers with edge detection; manual trigger via POST /automation/rules/:id/trigger; actions run through the AI pipeline so rules can call any tool (IoT control, search, ...) - reminders: due-scan scheduler delivering via WebSocket broadcast, optional LLM-polished reminder text with plain-title fallback; Reminder gains Done flag with ListDue/MarkDone - tests: cron matching and validation, once-per-minute dedup, device edge triggering, manual trigger, reminder delivery
This commit is contained in:
1 parent
9f30cdd086
commit
d80710d85a
7 files changed
+679
-2
No files matched your search
@@ -0,0 +1,178 @@
|
||||
package engine
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/plugin"
|
||||
)
|
||||
|
||||
// Rule is an automation rule. Trigger types: "time" (cron), "event" (device state).
|
||||
// The action is a natural-language instruction executed through the AI pipeline,
|
||||
// so it can call any tool (IoT control, search, etc.).
|
||||
type Rule struct {
|
||||
ID string
|
||||
Name string
|
||||
Trigger string // "time" | "event" | "webhook"
|
||||
Condition map[string]string // time: {"cron": "..."}; event: {"device_id","field","value"}
|
||||
Action string
|
||||
Enabled bool
|
||||
}
|
||||
|
||||
// RuleSource provides the current automation rules.
|
||||
type RuleSource interface {
|
||||
List() []Rule
|
||||
}
|
||||
|
||||
// AIFunc executes an action prompt through the AI pipeline (fire and forget).
|
||||
type AIFunc func(ctx context.Context, sessionID, userID, message string)
|
||||
|
||||
// DeviceQuerier polls IoT device states for event triggers.
|
||||
type DeviceQuerier interface {
|
||||
ListDevices(ctx context.Context) ([]plugin.Device, error)
|
||||
}
|
||||
|
||||
// Engine evaluates automation rules on a ticker.
|
||||
type Engine struct {
|
||||
src RuleSource
|
||||
ai AIFunc
|
||||
devices DeviceQuerier
|
||||
interval time.Duration
|
||||
|
||||
mu sync.Mutex
|
||||
lastCronFire map[string]string // ruleID → fired minute key
|
||||
deviceMatches map[string]bool // ruleID → condition held last poll
|
||||
stop chan struct{}
|
||||
stopOnce sync.Once
|
||||
}
|
||||
|
||||
// New creates an automation engine.
|
||||
func New(src RuleSource, ai AIFunc, devices DeviceQuerier, interval time.Duration) *Engine {
|
||||
if interval <= 0 {
|
||||
interval = 30 * time.Second
|
||||
}
|
||||
return &Engine{
|
||||
src: src,
|
||||
ai: ai,
|
||||
devices: devices,
|
||||
interval: interval,
|
||||
lastCronFire: map[string]string{},
|
||||
deviceMatches: map[string]bool{},
|
||||
stop: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
// Start begins the evaluation loop.
|
||||
func (e *Engine) Start(ctx context.Context) {
|
||||
go func() {
|
||||
ticker := time.NewTicker(e.interval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-e.stop:
|
||||
return
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case t := <-ticker.C:
|
||||
e.check(ctx, t)
|
||||
}
|
||||
}
|
||||
}()
|
||||
logger.Infof("[automation] engine started (interval=%v)", e.interval)
|
||||
}
|
||||
|
||||
// Stop shuts down the engine.
|
||||
func (e *Engine) Stop() { e.stopOnce.Do(func() { close(e.stop) }) }
|
||||
|
||||
// Trigger fires a rule immediately (manual / webhook).
|
||||
func (e *Engine) Trigger(ctx context.Context, id string) error {
|
||||
for _, r := range e.src.List() {
|
||||
if r.ID == id {
|
||||
e.fire(ctx, r, "manual")
|
||||
return nil
|
||||
}
|
||||
}
|
||||
return fmt.Errorf("rule %s not found", id)
|
||||
}
|
||||
|
||||
// check evaluates all rules once.
|
||||
func (e *Engine) check(ctx context.Context, now time.Time) {
|
||||
for _, r := range e.src.List() {
|
||||
if !r.Enabled {
|
||||
continue
|
||||
}
|
||||
switch r.Trigger {
|
||||
case "time":
|
||||
e.checkCron(ctx, r, now)
|
||||
case "event":
|
||||
e.checkDevice(ctx, r)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (e *Engine) checkCron(ctx context.Context, r Rule, now time.Time) {
|
||||
expr := r.Condition["cron"]
|
||||
if expr == "" {
|
||||
return
|
||||
}
|
||||
sched, err := ParseCron(expr)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if !sched.Match(now) {
|
||||
return
|
||||
}
|
||||
// Fire at most once per minute per rule
|
||||
minuteKey := now.Format("2006-01-02 15:04")
|
||||
e.mu.Lock()
|
||||
if e.lastCronFire[r.ID] == minuteKey {
|
||||
e.mu.Unlock()
|
||||
return
|
||||
}
|
||||
e.lastCronFire[r.ID] = minuteKey
|
||||
e.mu.Unlock()
|
||||
e.fire(ctx, r, "cron")
|
||||
}
|
||||
|
||||
func (e *Engine) checkDevice(ctx context.Context, r Rule) {
|
||||
if e.devices == nil {
|
||||
return
|
||||
}
|
||||
devID, field, want := r.Condition["device_id"], r.Condition["field"], r.Condition["value"]
|
||||
if devID == "" || field == "" {
|
||||
return
|
||||
}
|
||||
|
||||
devices, err := e.devices.ListDevices(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
held := false
|
||||
for _, d := range devices {
|
||||
if d.ID != devID {
|
||||
continue
|
||||
}
|
||||
if v, ok := d.State[field]; ok && fmt.Sprint(v) == want {
|
||||
held = true
|
||||
}
|
||||
}
|
||||
|
||||
// Edge trigger: only fire when the condition starts holding
|
||||
e.mu.Lock()
|
||||
was := e.deviceMatches[r.ID]
|
||||
e.deviceMatches[r.ID] = held
|
||||
e.mu.Unlock()
|
||||
if held && !was {
|
||||
e.fire(ctx, r, "device_state")
|
||||
}
|
||||
}
|
||||
|
||||
func (e *Engine) fire(ctx context.Context, r Rule, reason string) {
|
||||
logger.Infof("[automation] rule %q fired (%s): %s", r.Name, reason, r.Action)
|
||||
if e.ai != nil && r.Action != "" {
|
||||
go e.ai(ctx, "automation-"+r.ID, "automation", r.Action)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user