diff --git a/go.mod b/go.mod index 7df318e5d2..e9da796e4f 100644 --- a/go.mod +++ b/go.mod @@ -8,6 +8,7 @@ require ( github.com/google/uuid v1.6.0 github.com/gordonklaus/portaudio v0.0.0-20260203164431-765aa7dfa631 github.com/gorilla/websocket v1.5.3 + github.com/makiuchi-d/gozxing v0.1.1 github.com/prometheus/client_golang v1.23.2 github.com/santhosh-tekuri/jsonschema/v6 v6.0.2 github.com/stretchr/testify v1.11.1 @@ -34,6 +35,7 @@ require ( golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842 // indirect golang.org/x/sys v0.35.0 // indirect golang.org/x/text v0.37.0 // indirect + golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect google.golang.org/protobuf v1.36.8 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index 8019933187..a104e4b620 100644 --- a/go.sum +++ b/go.sum @@ -32,6 +32,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= +github.com/makiuchi-d/gozxing v0.1.1 h1:xxqijhoedi+/lZlhINteGbywIrewVdVv2wl9r5O9S1I= +github.com/makiuchi-d/gozxing v0.1.1/go.mod h1:eRIHbOjX7QWxLIDJoQuMLhuXg9LAuw6znsUtRkNw9DU= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= @@ -68,6 +70,8 @@ golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI= golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc= golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38= +golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE= +golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc= google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/internal/hooks/greeting_hook.go b/internal/hooks/greeting_hook.go index 5ec725a757..4e0dd802f3 100644 --- a/internal/hooks/greeting_hook.go +++ b/internal/hooks/greeting_hook.go @@ -32,18 +32,20 @@ const defaultGreetingPrompt = "You are {robot_name}, a friendly robot greeting w "Finish by offering help, for example: \"{help_message}\". " + "Respond with only the greeting text, with no quotes or commentary." -// greetingStartHook handles the start of a greeting conversation by generating a -// greeting message using an LLM and sending it to the TTS provider. func (r *Runner) greetingStartHook(ctx context.Context, cfg, vars map[string]any) error { + return r.announceGenerated(ctx, cfg, vars, defaultGreetingPrompt, "How can I help you today?") +} + +func (r *Runner) announceGenerated(ctx context.Context, cfg, vars map[string]any, defaultPrompt, defaultHelp string) error { provider, err := r.greetingTTSProvider(cfg) if err != nil { - r.log.Error("greeting_start_hook: error", zap.Error(err)) + r.log.Error("greeting hook: error", zap.Error(err)) return err } robotName := formatTemplate(stringVal(cfg, "robot_name"), vars) - helpMessage := "How can I help you today?" + helpMessage := defaultHelp if custom := formatTemplate(stringVal(cfg, "custom_message"), vars); custom != "" { helpMessage = custom } @@ -51,20 +53,20 @@ func (r *Runner) greetingStartHook(ctx context.Context, cfg, vars map[string]any face := providers.NewFacePresenceProvider(providers.FacePresenceConfig{}) snapshot, snapErr := face.FetchSnapshot(ctx) if snapErr != nil { - r.log.Warn("greeting_start_hook: face snapshot failed", zap.Error(snapErr)) + r.log.Warn("greeting hook: face snapshot failed", zap.Error(snapErr)) } memContext := r.recallMemory(ctx, snapshot.ClosestUUID) - if greeting, genErr := r.generateGreeting(ctx, cfg, vars, snapshot, memContext, robotName, helpMessage); genErr != nil { - r.log.Warn("greeting_start_hook: llm generation failed, using static greeting", zap.Error(genErr)) + if greeting, genErr := r.generateGreeting(ctx, cfg, vars, snapshot, memContext, robotName, helpMessage, defaultPrompt); genErr != nil { + r.log.Warn("greeting hook: llm generation failed, using static greeting", zap.Error(genErr)) provider.AddText(staticGreeting(snapshot, snapErr, robotName, helpMessage)) } else { r.log.Info("greeting generated successfully", zap.String("greeting", greeting)) provider.AddText(greeting) } - r.log.Info("greeting start hook executed successfully") + r.log.Info("greeting hook executed successfully") return nil } @@ -92,7 +94,7 @@ func memoryClause(memContext string) string { } // generateGreeting constructs a prompt using the snapshot and other context, calls the LLM to generate a greeting, and returns the greeting text. -func (r *Runner) generateGreeting(ctx context.Context, cfg, vars map[string]any, snapshot providers.PresenceSnapshot, memContext, robotName, helpMessage string) (string, error) { +func (r *Runner) generateGreeting(ctx context.Context, cfg, vars map[string]any, snapshot providers.PresenceSnapshot, memContext, robotName, helpMessage, defaultPrompt string) (string, error) { if robotName == "" { robotName = "a friendly robot" } @@ -106,7 +108,7 @@ func (r *Runner) generateGreeting(ctx context.Context, cfg, vars map[string]any, promptTemplate := stringVal(cfg, "prompt") if strings.TrimSpace(promptTemplate) == "" { - promptTemplate = defaultGreetingPrompt + promptTemplate = defaultPrompt } promptVars := make(map[string]any, len(vars)+6) diff --git a/internal/hooks/luma_hook.go b/internal/hooks/luma_hook.go new file mode 100644 index 0000000000..834c5c05b6 --- /dev/null +++ b/internal/hooks/luma_hook.go @@ -0,0 +1,25 @@ +package hooks + +import ( + "context" +) + +func init() { + RegisterHook("luma_hook", "luma_intro_hook", (*Runner).lumaIntroHook) +} + +const defaultLumaHelp = "Are you registered on Luma?" + +const defaultLumaPrompt = "You are {robot_name}, a friendly robot welcoming a guest to an event. " + + "The current time is {current_time}. " + + "Generate a single warm, natural spoken greeting of one or two short sentences. " + + "Here is what you currently see: {scene}. " + + "Make it feel personal and present by naturally referencing something specific from what you see; never invent anything. " + + "If a specific person is recognized ({closest_name}), greet them by name; otherwise greet generically. " + + "{memory}" + + "Finish by asking the guest whether they are registered on Luma, for example: \"{help_message}\". " + + "Respond with only the greeting text, with no quotes or commentary." + +func (r *Runner) lumaIntroHook(ctx context.Context, cfg, vars map[string]any) error { + return r.announceGenerated(ctx, cfg, vars, defaultLumaPrompt, defaultLumaHelp) +} diff --git a/internal/hooks/luma_hook_test.go b/internal/hooks/luma_hook_test.go new file mode 100644 index 0000000000..a273449033 --- /dev/null +++ b/internal/hooks/luma_hook_test.go @@ -0,0 +1,19 @@ +package hooks + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestLumaHookRegistered(t *testing.T) { + _, ok := lookupHook("luma_hook", "luma_intro_hook") + require.True(t, ok) +} + +func TestLumaPromptMentionsLuma(t *testing.T) { + require.True(t, strings.Contains(defaultLumaPrompt, "{help_message}"), + "prompt must interpolate the help message") + require.Contains(t, strings.ToLower(defaultLumaHelp), "luma") +} diff --git a/internal/providers/luma/checkin.go b/internal/providers/luma/checkin.go new file mode 100644 index 0000000000..be97d8144e --- /dev/null +++ b/internal/providers/luma/checkin.go @@ -0,0 +1,21 @@ +package luma + +import ( + "sync/atomic" + "time" +) + +type CheckIn struct { + Name string + Time time.Time +} + +var lastCheckIn atomic.Pointer[CheckIn] + +func RecordCheckIn(name string, t time.Time) { + lastCheckIn.Store(&CheckIn{Name: name, Time: t}) +} + +func LastCheckIn() *CheckIn { + return lastCheckIn.Load() +} diff --git a/internal/providers/luma/checkin_test.go b/internal/providers/luma/checkin_test.go new file mode 100644 index 0000000000..028e338fe8 --- /dev/null +++ b/internal/providers/luma/checkin_test.go @@ -0,0 +1,31 @@ +package luma + +import ( + "testing" + "time" +) + +func TestRecordAndLastCheckIn(t *testing.T) { + if got := LastCheckIn(); got != nil { + t.Fatalf("expected nil before any publish, got %+v", got) + } + + t1 := time.Unix(1_700_000_000, 0) + RecordCheckIn("Ada", t1) + + got := LastCheckIn() + if got == nil { + t.Fatal("expected a check-in after publish") + } + if got.Name != "Ada" || !got.Time.Equal(t1) { + t.Fatalf("unexpected check-in: %+v", got) + } + + // A later publish replaces the previous value. + t2 := t1.Add(time.Minute) + RecordCheckIn("Grace", t2) + got = LastCheckIn() + if got.Name != "Grace" || !got.Time.Equal(t2) { + t.Fatalf("expected latest check-in to be Grace@t2, got %+v", got) + } +} diff --git a/internal/providers/luma/client.go b/internal/providers/luma/client.go new file mode 100644 index 0000000000..904562aff5 --- /dev/null +++ b/internal/providers/luma/client.go @@ -0,0 +1,234 @@ +package luma + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" + + "github.com/openmind/om1/internal/httpclient" +) + +const ( + DefaultBaseURL = "https://public-api.luma.com" + GetGuestPath = "/v1/event/get-guest" + CheckInURL = "https://api.luma.com/event/admin/update-check-in" +) + +var ( + ErrNotFound = errors.New("luma: guest not found") + ErrUnauthorized = errors.New("luma: unauthorized") +) + +// Guest is a flattened view of the guest record returned by Luma's +// /v1/event/get-guest. Identity fields use the user_-prefixed names from the +// public API schema. Unknown fields are ignored. +type Guest struct { + APIID string `json:"api_id"` + UserAPIID string `json:"user_api_id"` + UserName string `json:"user_name"` + UserFirstName string `json:"user_first_name"` + UserLastName string `json:"user_last_name"` + UserEmail string `json:"user_email"` + EventAPIID string `json:"event_api_id"` + CheckedInAt string `json:"checked_in_at"` + ApprovalStatus string `json:"approval_status"` +} + +// guestEnvelope handles Luma's nested response shape: {"guest": {...}, "event": {...}}. +// Some endpoints return the guest object directly; we try the envelope first. +type guestEnvelope struct { + Guest *Guest `json:"guest"` + Event *struct { + APIID string `json:"api_id"` + } `json:"event"` +} + +type Client struct { + baseURL string + apiKey string + eventAPIID string + sessionKey string + http *http.Client + timeout time.Duration +} + +func NewClient(baseURL, apiKey, eventAPIID string, timeout time.Duration, opts ...func(*Client)) *Client { + if baseURL == "" { + baseURL = DefaultBaseURL + } + if timeout <= 0 { + timeout = 5 * time.Second + } + c := &Client{ + baseURL: strings.TrimRight(baseURL, "/"), + apiKey: apiKey, + eventAPIID: eventAPIID, + http: httpclient.Default(), + timeout: timeout, + } + for _, o := range opts { + o(c) + } + return c +} + +func WithSessionKey(key string) func(*Client) { + return func(c *Client) { c.sessionKey = key } +} + +// EventAPIID returns the configured event id. +func (c *Client) EventAPIID() string { return c.eventAPIID } + +// GetGuest retrieves a guest record by its API ID. +func (c *Client) GetGuest(ctx context.Context, pk string) (*Guest, error) { + ctx, cancel := context.WithTimeout(ctx, c.timeout) + defer cancel() + + q := url.Values{ + "event_id": []string{c.eventAPIID}, + "id": []string{pk}, + } + endpoint := c.baseURL + GetGuestPath + "?" + q.Encode() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) + if err != nil { + return nil, err + } + c.setAuth(req) + + resp, err := c.do(req) + if err != nil { + return nil, err + } + defer func() { _ = resp.Body.Close() }() + + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("luma get-guest read body: %w", err) + } + switch resp.StatusCode { + case http.StatusOK: + case http.StatusNotFound: + return nil, ErrNotFound + case http.StatusUnauthorized, http.StatusForbidden: + return nil, ErrUnauthorized + default: + return nil, fmt.Errorf("luma get-guest %d: %s", resp.StatusCode, string(body)) + } + + var env guestEnvelope + if err := json.Unmarshal(body, &env); err == nil && env.Guest != nil { + g := env.Guest + if g.EventAPIID == "" && env.Event != nil { + g.EventAPIID = env.Event.APIID + } + return g, nil + } + var bare Guest + if err := json.Unmarshal(body, &bare); err != nil { + return nil, fmt.Errorf("luma get-guest decode: %w", err) + } + return &bare, nil +} + +func (c *Client) setAuth(req *http.Request) { + req.Header.Set("x-luma-api-key", c.apiKey) + req.Header.Set("Accept", "application/json") +} + +// do issues req with one retry on 429 after a short backoff. +func (c *Client) do(req *http.Request) (*http.Response, error) { + resp, err := c.http.Do(req) + if err != nil { + return nil, err + } + if resp.StatusCode != http.StatusTooManyRequests { + return resp, nil + } + + _, _ = io.Copy(io.Discard, resp.Body) + _ = resp.Body.Close() + + t := time.NewTimer(500 * time.Millisecond) + defer t.Stop() + select { + case <-req.Context().Done(): + return nil, req.Context().Err() + case <-t.C: + } + + if req.GetBody != nil { + body, berr := req.GetBody() + if berr != nil { + return nil, berr + } + req.Body = body + } + return c.http.Do(req) +} + +func (c *Client) CheckIn(ctx context.Context, guest *Guest) error { + if c.sessionKey == "" { + return fmt.Errorf("luma check-in: no session key configured") + } + if guest == nil || guest.APIID == "" { + return fmt.Errorf("luma check-in: missing guest api_id") + } + + payload := map[string]string{ + "event_api_id": c.eventAPIID, + "rsvp_api_id": guest.APIID, + "check_in_method": "guest-list", + "check_in_status": "checked-in", + "type": "guest", + } + body, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("luma check-in marshal: %w", err) + } + + ctx, cancel := context.WithTimeout(ctx, c.timeout) + defer cancel() + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, CheckInURL, bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("luma check-in request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Cookie", "luma.auth-session-key="+c.sessionKey) + + resp, err := c.do(req) + if err != nil { + return fmt.Errorf("luma check-in: %w", err) + } + defer func() { _ = resp.Body.Close() }() + respBody, _ := io.ReadAll(resp.Body) + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("luma check-in %d: %s", resp.StatusCode, string(respBody)) + } + return nil +} + +// FirstName picks the best available first-name field from a Guest record. +// Falls back to "friend" when nothing usable is set. +func FirstName(g *Guest) string { + if g == nil { + return "friend" + } + if g.UserFirstName != "" { + return g.UserFirstName + } + if g.UserName != "" { + if parts := strings.Fields(g.UserName); len(parts) > 0 { + return parts[0] + } + } + return "friend" +} diff --git a/internal/providers/luma/client_test.go b/internal/providers/luma/client_test.go new file mode 100644 index 0000000000..72983641a0 --- /dev/null +++ b/internal/providers/luma/client_test.go @@ -0,0 +1,223 @@ +package luma + +import ( + "context" + "encoding/json" + "io" + "net/http" + "strings" + "testing" + "time" +) + +// roundTripFunc lets us stub HTTP responses inline. +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) { return f(req) } + +func stubClient(fn roundTripFunc) *http.Client { + return &http.Client{Transport: fn} +} + +func newTestClient(doer *http.Client, opts ...func(*Client)) *Client { + c := NewClient("https://test.luma.com", "test-key", "evt-123", 5*time.Second, opts...) + c.http = doer + return c +} + +func jsonResp(code int, body any) *http.Response { + b, _ := json.Marshal(body) + return &http.Response{ + StatusCode: code, + Body: io.NopCloser(strings.NewReader(string(b))), + Header: http.Header{"Content-Type": []string{"application/json"}}, + } +} + +func TestGetGuest_OK_Envelope(t *testing.T) { + doer := stubClient(func(req *http.Request) (*http.Response, error) { + if !strings.Contains(req.URL.Path, GetGuestPath) { + t.Fatalf("unexpected path: %s", req.URL.Path) + } + if req.Header.Get("x-luma-api-key") != "test-key" { + t.Fatalf("missing api key header") + } + return jsonResp(200, guestEnvelope{ + Guest: &Guest{APIID: "gst-1", UserFirstName: "Alice", UserEmail: "alice@x.com"}, + Event: &struct { + APIID string `json:"api_id"` + }{APIID: "evt-123"}, + }), nil + }) + + c := newTestClient(doer) + g, err := c.GetGuest(context.Background(), "g-abc") + if err != nil { + t.Fatal(err) + } + if g.APIID != "gst-1" { + t.Fatalf("got APIID %q", g.APIID) + } + if g.UserFirstName != "Alice" { + t.Fatalf("got first name %q", g.UserFirstName) + } + if g.EventAPIID != "evt-123" { + t.Fatalf("got event api id %q", g.EventAPIID) + } +} + +func TestGetGuest_OK_Bare(t *testing.T) { + doer := stubClient(func(req *http.Request) (*http.Response, error) { + return jsonResp(200, Guest{APIID: "gst-2", UserName: "Bob Smith"}), nil + }) + + c := newTestClient(doer) + g, err := c.GetGuest(context.Background(), "g-abc") + if err != nil { + t.Fatal(err) + } + if g.APIID != "gst-2" { + t.Fatalf("got APIID %q", g.APIID) + } +} + +func TestGetGuest_NotFound(t *testing.T) { + doer := stubClient(func(req *http.Request) (*http.Response, error) { + return jsonResp(404, map[string]string{"error": "not found"}), nil + }) + + c := newTestClient(doer) + _, err := c.GetGuest(context.Background(), "g-abc") + if err == nil || err.Error() != ErrNotFound.Error() { + t.Fatalf("expected ErrNotFound, got %v", err) + } +} + +func TestGetGuest_Unauthorized(t *testing.T) { + doer := stubClient(func(req *http.Request) (*http.Response, error) { + return jsonResp(401, nil), nil + }) + + c := newTestClient(doer) + _, err := c.GetGuest(context.Background(), "g-abc") + if err == nil || err.Error() != ErrUnauthorized.Error() { + t.Fatalf("expected ErrUnauthorized, got %v", err) + } +} + +func TestGetGuest_429_Retry(t *testing.T) { + calls := 0 + doer := stubClient(func(req *http.Request) (*http.Response, error) { + calls++ + if calls == 1 { + return jsonResp(429, nil), nil + } + return jsonResp(200, guestEnvelope{ + Guest: &Guest{APIID: "gst-retry"}, + }), nil + }) + + c := newTestClient(doer) + g, err := c.GetGuest(context.Background(), "g-abc") + if err != nil { + t.Fatal(err) + } + if g.APIID != "gst-retry" { + t.Fatalf("got APIID %q", g.APIID) + } + if calls != 2 { + t.Fatalf("expected 2 calls, got %d", calls) + } +} + +func TestCheckIn_OK(t *testing.T) { + doer := stubClient(func(req *http.Request) (*http.Response, error) { + if req.URL.String() != CheckInURL { + t.Fatalf("unexpected url: %s", req.URL) + } + if !strings.Contains(req.Header.Get("Cookie"), "luma.auth-session-key=sess-abc") { + t.Fatalf("missing session cookie") + } + body, _ := io.ReadAll(req.Body) + var payload map[string]string + _ = json.Unmarshal(body, &payload) + if payload["rsvp_api_id"] != "gst-1" { + t.Fatalf("wrong rsvp_api_id: %s", payload["rsvp_api_id"]) + } + if payload["check_in_status"] != "checked-in" { + t.Fatalf("wrong status: %s", payload["check_in_status"]) + } + return jsonResp(200, map[string]string{"ok": "true"}), nil + }) + + c := newTestClient(doer, WithSessionKey("sess-abc")) + err := c.CheckIn(context.Background(), &Guest{APIID: "gst-1"}) + if err != nil { + t.Fatal(err) + } +} + +func TestCheckIn_NoSessionKey(t *testing.T) { + c := NewClient("", "key", "evt-1", time.Second) + err := c.CheckIn(context.Background(), &Guest{APIID: "gst-1"}) + if err == nil || !strings.Contains(err.Error(), "no session key") { + t.Fatalf("expected no session key error, got %v", err) + } +} + +func TestCheckIn_NilGuest(t *testing.T) { + c := NewClient("", "key", "evt-1", time.Second, WithSessionKey("s")) + err := c.CheckIn(context.Background(), nil) + if err == nil || !strings.Contains(err.Error(), "missing guest") { + t.Fatalf("expected missing guest error, got %v", err) + } +} + +func TestCheckIn_ServerError(t *testing.T) { + doer := stubClient(func(req *http.Request) (*http.Response, error) { + return jsonResp(500, map[string]string{"error": "internal"}), nil + }) + + c := newTestClient(doer, WithSessionKey("sess")) + err := c.CheckIn(context.Background(), &Guest{APIID: "gst-1"}) + if err == nil || !strings.Contains(err.Error(), "500") { + t.Fatalf("expected 500 error, got %v", err) + } +} + +func TestFirstName(t *testing.T) { + cases := []struct { + name string + guest *Guest + want string + }{ + {"nil guest", nil, "friend"}, + {"first name set", &Guest{UserFirstName: "Alice"}, "Alice"}, + {"only full name", &Guest{UserName: "Bob Smith"}, "Bob"}, + {"empty", &Guest{}, "friend"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := FirstName(tc.guest); got != tc.want { + t.Fatalf("got %q want %q", got, tc.want) + } + }) + } +} + +func TestNewClient_Defaults(t *testing.T) { + c := NewClient("", "key", "evt-1", 0) + if c.baseURL != DefaultBaseURL { + t.Fatalf("got base url %q", c.baseURL) + } + if c.timeout != 5*time.Second { + t.Fatalf("got timeout %v", c.timeout) + } +} + +func TestEventAPIID(t *testing.T) { + c := NewClient("", "key", "evt-42", time.Second) + if c.EventAPIID() != "evt-42" { + t.Fatalf("got %q", c.EventAPIID()) + } +} diff --git a/plugins/backgrounds/backgrounds.go b/plugins/backgrounds/backgrounds.go index d20c9ed521..28300131a6 100644 --- a/plugins/backgrounds/backgrounds.go +++ b/plugins/backgrounds/backgrounds.go @@ -1,6 +1,7 @@ package backgrounds import ( + _ "github.com/openmind/om1/plugins/backgrounds/luma" _ "github.com/openmind/om1/plugins/backgrounds/unitree/go2" _ "github.com/openmind/om1/plugins/backgrounds/vlm" ) diff --git a/plugins/backgrounds/luma/face_size_watch.go b/plugins/backgrounds/luma/face_size_watch.go new file mode 100644 index 0000000000..9e64dd9bb5 --- /dev/null +++ b/plugins/backgrounds/luma/face_size_watch.go @@ -0,0 +1,110 @@ +package luma + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "go.uber.org/zap" + + bg "github.com/openmind/om1/internal/backgrounds" + "github.com/openmind/om1/internal/logger" + "github.com/openmind/om1/internal/providers" + "github.com/openmind/om1/internal/util" +) + +func init() { + bg.Register("FaceSizeWatch", NewFaceSizeWatch) +} + +type faceSizeWatchConfig struct { + BaseURL string `json:"face_http_base_url"` + RecentSec float64 `json:"face_recent_sec"` + PollSec float64 `json:"face_poll_interval_sec"` + MinFaceArea float64 `json:"min_face_area"` +} + +type FaceSizeWatch struct { + log *zap.Logger + provider *providers.FacePresenceProvider + period time.Duration + minArea float64 +} + +func NewFaceSizeWatch(configMap map[string]any) (bg.Background, error) { + var cfg faceSizeWatchConfig + if b, err := json.Marshal(configMap); err == nil { + _ = json.Unmarshal(b, &cfg) + } + if cfg.BaseURL == "" { + cfg.BaseURL = "http://127.0.0.1:6793" + } + if cfg.RecentSec <= 0 { + cfg.RecentSec = 1.0 + } + if cfg.PollSec <= 0 { + cfg.PollSec = 0.5 + } + if cfg.MinFaceArea <= 0 { + cfg.MinFaceArea = 3000 + } + + log := logger.Get().Named("FaceSizeWatch") + + provider := providers.NewFacePresenceProvider(providers.FacePresenceConfig{ + BaseURL: cfg.BaseURL, + RecentSec: cfg.RecentSec, + Timeout: 2 * time.Second, + }) + + log.Info("initialized", + zap.String("base_url", cfg.BaseURL), + zap.Float64("min_face_area", cfg.MinFaceArea), + zap.Float64("poll_sec", cfg.PollSec), + ) + + return &FaceSizeWatch{ + log: log, + provider: provider, + period: time.Duration(cfg.PollSec * float64(time.Second)), + minArea: cfg.MinFaceArea, + }, nil +} + +func (f *FaceSizeWatch) Run(ctx context.Context) { + snap, err := f.provider.FetchSnapshot(ctx) + if err != nil { + if ctx.Err() == nil { + f.log.Warn("failed to fetch snapshot", zap.Error(err)) + } + util.Sleep(ctx, f.period) + return + } + + var largestArea int + var largestTrackID int + for _, face := range snap.Faces { + if face.Area > largestArea { + largestArea = face.Area + largestTrackID = face.TrackID + } + } + if float64(largestArea) >= f.minArea { + f.log.Info("face close enough, triggering transition", + zap.Int("area", largestArea), + zap.Int("track_id", largestTrackID), + ) + providers.IO().AddInput("PrimaryGuestTrackID", + fmt.Sprintf("%d", largestTrackID), + time.Now(), + ) + providers.ModeContext().Publish(map[string]any{"face_close_enough": true}) + } + + util.Sleep(ctx, f.period) +} + +func (f *FaceSizeWatch) Stop() { + f.log.Info("stopping") +} diff --git a/plugins/backgrounds/luma/face_size_watch_test.go b/plugins/backgrounds/luma/face_size_watch_test.go new file mode 100644 index 0000000000..a6f5ff5961 --- /dev/null +++ b/plugins/backgrounds/luma/face_size_watch_test.go @@ -0,0 +1,37 @@ +package luma + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestNewFaceSizeWatch_Defaults(t *testing.T) { + bg, err := NewFaceSizeWatch(map[string]any{}) + require.NoError(t, err) + require.NotNil(t, bg) + + fsw := bg.(*FaceSizeWatch) + require.Equal(t, float64(3000), fsw.minArea) + require.NotNil(t, fsw.provider) + require.NotNil(t, fsw.log) +} + +func TestNewFaceSizeWatch_CustomConfig(t *testing.T) { + bg, err := NewFaceSizeWatch(map[string]any{ + "face_http_base_url": "http://localhost:9999", + "face_recent_sec": 2.0, + "face_poll_interval_sec": 1.0, + "min_face_area": 5000.0, + }) + require.NoError(t, err) + + fsw := bg.(*FaceSizeWatch) + require.Equal(t, float64(5000), fsw.minArea) +} + +func TestFaceSizeWatch_Stop(t *testing.T) { + bg, err := NewFaceSizeWatch(map[string]any{}) + require.NoError(t, err) + bg.(*FaceSizeWatch).Stop() +} diff --git a/plugins/backgrounds/luma/guest_lingering.go b/plugins/backgrounds/luma/guest_lingering.go new file mode 100644 index 0000000000..2acacfcff4 --- /dev/null +++ b/plugins/backgrounds/luma/guest_lingering.go @@ -0,0 +1,154 @@ +package luma + +import ( + "context" + "encoding/json" + "strconv" + "time" + + "go.uber.org/zap" + + bg "github.com/openmind/om1/internal/backgrounds" + "github.com/openmind/om1/internal/logger" + "github.com/openmind/om1/internal/providers" + "github.com/openmind/om1/internal/providers/luma" + "github.com/openmind/om1/internal/providers/tts" + "github.com/openmind/om1/internal/util" +) + +func init() { + bg.Register("GuestLingering", NewGuestLingering) +} + +type guestLingeringConfig struct { + FaceBaseURL string `json:"face_http_base_url"` + FaceRecentSec float64 `json:"face_recent_sec"` + MinFaceArea float64 `json:"min_face_area"` + PollSec float64 `json:"poll_interval_sec"` + GracePeriodSec float64 `json:"grace_period_sec"` +} + +type GuestLingering struct { + log *zap.Logger + face *providers.FacePresenceProvider + period time.Duration + gracePeriod time.Duration + minArea float64 + lastHandled time.Time +} + +func NewGuestLingering(configMap map[string]any) (bg.Background, error) { + var cfg guestLingeringConfig + if b, err := json.Marshal(configMap); err == nil { + _ = json.Unmarshal(b, &cfg) + } + if cfg.FaceBaseURL == "" { + cfg.FaceBaseURL = "http://127.0.0.1:6793" + } + if cfg.FaceRecentSec <= 0 { + cfg.FaceRecentSec = 1.0 + } + if cfg.MinFaceArea <= 0 { + cfg.MinFaceArea = 3000 + } + if cfg.PollSec <= 0 { + cfg.PollSec = 1.0 + } + if cfg.GracePeriodSec <= 0 { + cfg.GracePeriodSec = 5.0 + } + + log := logger.Get().Named("GuestLingering") + + face := providers.NewFacePresenceProvider(providers.FacePresenceConfig{ + BaseURL: cfg.FaceBaseURL, + RecentSec: cfg.FaceRecentSec, + Timeout: 2 * time.Second, + }) + + providers.IO().AddInput("CheckinStatus", "", time.Now()) + + log.Info("initialized", + zap.Float64("min_face_area", cfg.MinFaceArea), + zap.Float64("grace_period_sec", cfg.GracePeriodSec), + ) + + return &GuestLingering{ + log: log, + face: face, + period: time.Duration(cfg.PollSec * float64(time.Second)), + gracePeriod: time.Duration(cfg.GracePeriodSec * float64(time.Second)), + minArea: cfg.MinFaceArea, + lastHandled: time.Now(), + }, nil +} + +func (g *GuestLingering) Run(ctx context.Context) { + checkin := luma.LastCheckIn() + if checkin == nil || !checkin.Time.After(g.lastHandled) { + util.Sleep(ctx, g.period) + return + } + + if time.Since(checkin.Time) < g.gracePeriod { + util.Sleep(ctx, g.period) + return + } + + // Read the track_id of the primary guest stored by FaceSizeWatch. + var primaryTrackID int + if in := providers.IO().GetInput("PrimaryGuestTrackID"); in != nil && in.Input != "" { + primaryTrackID, _ = strconv.Atoi(in.Input) + } + + snap, err := g.face.FetchSnapshot(ctx) + if err != nil { + if ctx.Err() == nil { + g.log.Warn("failed to fetch face snapshot", zap.Error(err)) + } + util.Sleep(ctx, g.period) + return + } + + // Check if the primary guest's face is still present (by track_id or any large face as fallback). + var guestPresent bool + for _, face := range snap.Faces { + if float64(face.Area) < g.minArea { + continue + } + if primaryTrackID > 0 && face.TrackID == primaryTrackID { + guestPresent = true + break + } + if primaryTrackID == 0 { + guestPresent = true + break + } + } + + if guestPresent { + // Only emit a fresh nudge when TTS is idle, so nudges never pile up in the + // queue while the robot is still speaking or has speech queued. + if tts.Busy() { + util.Sleep(ctx, g.period) + return + } + g.log.Info("guest lingering after check-in", + zap.String("name", checkin.Name), + zap.Int("primary_track_id", primaryTrackID), + ) + providers.IO().AddInput("CheckinStatus", + "checkin_status: guest_lingering name="+checkin.Name, + time.Now(), + ) + } else { + providers.IO().AddInput("CheckinStatus", "", time.Now()) + g.lastHandled = checkin.Time + } + + util.Sleep(ctx, g.period) +} + +func (g *GuestLingering) Stop() { + g.log.Info("stopping") +} diff --git a/plugins/backgrounds/luma/luma_checkin.go b/plugins/backgrounds/luma/luma_checkin.go new file mode 100644 index 0000000000..95f2dc9b31 --- /dev/null +++ b/plugins/backgrounds/luma/luma_checkin.go @@ -0,0 +1,143 @@ +package luma + +import ( + "context" + "encoding/json" + "strconv" + "time" + + "go.uber.org/zap" + + bg "github.com/openmind/om1/internal/backgrounds" + "github.com/openmind/om1/internal/logger" + "github.com/openmind/om1/internal/providers" + "github.com/openmind/om1/internal/providers/luma" + "github.com/openmind/om1/internal/util" +) + +func init() { + bg.Register("LumaCheckin", LumaCheckin) +} + +type checkinCompleteConfig struct { + FaceBaseURL string `json:"face_http_base_url"` + FaceRecentSec float64 `json:"face_recent_sec"` + FaceMinArea float64 `json:"min_face_area"` + PollSec float64 `json:"poll_interval_sec"` + GracePeriodSec float64 `json:"grace_period_sec"` +} + +type CheckinComplete struct { + log *zap.Logger + face *providers.FacePresenceProvider + period time.Duration + gracePeriod time.Duration + minArea float64 + lastHandled time.Time // timestamp of the scan we've already acted on +} + +func LumaCheckin(configMap map[string]any) (bg.Background, error) { + var cfg checkinCompleteConfig + if b, err := json.Marshal(configMap); err == nil { + _ = json.Unmarshal(b, &cfg) + } + if cfg.FaceBaseURL == "" { + cfg.FaceBaseURL = "http://127.0.0.1:6793" + } + if cfg.FaceRecentSec <= 0 { + cfg.FaceRecentSec = 1.0 + } + if cfg.FaceMinArea <= 0 { + cfg.FaceMinArea = 3000 + } + if cfg.PollSec <= 0 { + cfg.PollSec = 1.0 + } + if cfg.GracePeriodSec <= 0 { + cfg.GracePeriodSec = 5.0 + } + + log := logger.Get().Named("CheckinComplete") + + face := providers.NewFacePresenceProvider(providers.FacePresenceConfig{ + BaseURL: cfg.FaceBaseURL, + RecentSec: cfg.FaceRecentSec, + Timeout: 2 * time.Second, + }) + + log.Info("initialized", + zap.Float64("grace_period_sec", cfg.GracePeriodSec), + ) + + return &CheckinComplete{ + log: log, + face: face, + period: time.Duration(cfg.PollSec * float64(time.Second)), + gracePeriod: time.Duration(cfg.GracePeriodSec * float64(time.Second)), + minArea: cfg.FaceMinArea, + lastHandled: time.Now(), + }, nil +} + +func (c *CheckinComplete) Run(ctx context.Context) { + checkin := luma.LastCheckIn() + if checkin == nil || !checkin.Time.After(c.lastHandled) { + util.Sleep(ctx, c.period) + return + } + + if time.Since(checkin.Time) < c.gracePeriod { + util.Sleep(ctx, c.period) + return + } + + var primaryTrackID int + if in := providers.IO().GetInput("PrimaryGuestTrackID"); in != nil && in.Input != "" { + primaryTrackID, _ = strconv.Atoi(in.Input) + } + + snap, err := c.face.FetchSnapshot(ctx) + if err != nil { + if ctx.Err() == nil { + c.log.Warn("failed to fetch face snapshot", zap.Error(err)) + } + util.Sleep(ctx, c.period) + return + } + + c.log.Debug("departure check", + zap.Int("primary_track_id", primaryTrackID), + zap.Int("num_faces", len(snap.Faces)), + zap.String("checkin_name", checkin.Name), + ) + + // Check if primary guest is still present by track_id or fallback to any large face. + for _, face := range snap.Faces { + if float64(face.Area) < c.minArea { + continue + } + if primaryTrackID > 0 && face.TrackID == primaryTrackID { + util.Sleep(ctx, c.period) + return + } + if primaryTrackID == 0 { + util.Sleep(ctx, c.period) + return + } + } + + c.log.Info("guest departed after successful check-in, triggering transition", + zap.String("name", checkin.Name), + zap.Int("primary_track_id", primaryTrackID), + ) + // Clear any lingering signal so no further nudge ticks fire while we transition. + providers.IO().AddInput("CheckinStatus", "", time.Now()) + providers.ModeContext().Publish(map[string]any{"checkin_complete": true}) + c.lastHandled = checkin.Time + + util.Sleep(ctx, c.period) +} + +func (c *CheckinComplete) Stop() { + c.log.Info("stopping") +} diff --git a/plugins/backgrounds/luma/luma_checkin_test.go b/plugins/backgrounds/luma/luma_checkin_test.go new file mode 100644 index 0000000000..628a06abb7 --- /dev/null +++ b/plugins/backgrounds/luma/luma_checkin_test.go @@ -0,0 +1,38 @@ +package luma + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestLumaCheckin_Defaults(t *testing.T) { + bg, err := LumaCheckin(map[string]any{}) + require.NoError(t, err) + require.NotNil(t, bg) + + cc := bg.(*CheckinComplete) + require.Equal(t, float64(3000), cc.minArea) + require.True(t, cc.lastHandled.IsZero()) + require.NotNil(t, cc.face) + require.NotNil(t, cc.log) +} + +func TestLumaCheckin_CustomConfig(t *testing.T) { + bg, err := LumaCheckin(map[string]any{ + "face_http_base_url": "http://localhost:9999", + "min_face_area": 8000.0, + "poll_interval_sec": 2.0, + "grace_period_sec": 10.0, + }) + require.NoError(t, err) + + cc := bg.(*CheckinComplete) + require.Equal(t, float64(8000), cc.minArea) +} + +func TestCheckinComplete_Stop(t *testing.T) { + bg, err := LumaCheckin(map[string]any{}) + require.NoError(t, err) + bg.(*CheckinComplete).Stop() +} diff --git a/plugins/inputs/checkin_status.go b/plugins/inputs/checkin_status.go new file mode 100644 index 0000000000..022376e9da --- /dev/null +++ b/plugins/inputs/checkin_status.go @@ -0,0 +1,73 @@ +package inputs + +import ( + "context" + "time" + + "go.uber.org/zap" + + "github.com/openmind/om1/internal/inputs" + "github.com/openmind/om1/internal/logger" + "github.com/openmind/om1/internal/providers" +) + +func init() { + inputs.Register("CheckinStatus", NewCheckinStatus) +} + +type CheckinStatusSensor struct { + log *zap.Logger +} + +func NewCheckinStatus(_ map[string]any) (inputs.Sensor, error) { + log := logger.Get().Named("CheckinStatus") + log.Info("initializing") + return &CheckinStatusSensor{log: log}, nil +} + +func (s *CheckinStatusSensor) Listen(ctx context.Context) (<-chan any, error) { + out := make(chan any) + go func() { + defer close(out) + ticker := time.NewTicker(2 * time.Second) + defer ticker.Stop() + for { + select { + case <-ticker.C: + in := providers.IO().GetInput("CheckinStatus") + if in != nil && in.Input != "" { + select { + case out <- in.Input: + default: + } + } + case <-ctx.Done(): + return + } + } + }() + return out, nil +} + +func (s *CheckinStatusSensor) Poll(_ context.Context) (any, error) { + return nil, nil +} + +func (s *CheckinStatusSensor) RawToText(_ context.Context, raw any) (*inputs.Message, error) { + msg, ok := raw.(string) + if !ok || msg == "" { + return nil, nil + } + return &inputs.Message{Message: msg}, nil +} + +func (s *CheckinStatusSensor) FormattedLatestBuffer() string { + in := providers.IO().GetInput("CheckinStatus") + if in == nil || in.Input == "" { + return "" + } + return "\n" + in.Input + "\n" +} + +func (s *CheckinStatusSensor) TriggersTick() bool { return true } +func (s *CheckinStatusSensor) Stop() {} diff --git a/plugins/inputs/inputs.go b/plugins/inputs/inputs.go index b612cbf00b..990308b657 100644 --- a/plugins/inputs/inputs.go +++ b/plugins/inputs/inputs.go @@ -2,6 +2,7 @@ package inputs import ( _ "github.com/openmind/om1/plugins/inputs/asr" + _ "github.com/openmind/om1/plugins/inputs/luma_checkin" _ "github.com/openmind/om1/plugins/inputs/unitree/go2" _ "github.com/openmind/om1/plugins/inputs/vlm" ) diff --git a/plugins/inputs/luma_checkin/debounce.go b/plugins/inputs/luma_checkin/debounce.go new file mode 100644 index 0000000000..6b0c3c2a57 --- /dev/null +++ b/plugins/inputs/luma_checkin/debounce.go @@ -0,0 +1,47 @@ +package luma_checkin + +import ( + "sync" + "time" +) + +// debouncer drops repeated `pk` values seen within a sliding time window. +type debouncer struct { + mu sync.Mutex + window time.Duration + seen map[string]time.Time + now func() time.Time +} + +func newDebouncer(window time.Duration) *debouncer { + return &debouncer{ + window: window, + seen: make(map[string]time.Time), + now: time.Now, + } +} + +// TryRecord returns true if pk was not seen within the configured window. +// Recording also prunes entries older than 10x the window to bound memory. +func (d *debouncer) TryRecord(pk string) bool { + d.mu.Lock() + defer d.mu.Unlock() + + now := d.now() + d.pruneLocked(now) + + if last, ok := d.seen[pk]; ok && now.Sub(last) < d.window { + return false + } + d.seen[pk] = now + return true +} + +func (d *debouncer) pruneLocked(now time.Time) { + cutoff := now.Add(-10 * d.window) + for k, t := range d.seen { + if t.Before(cutoff) { + delete(d.seen, k) + } + } +} diff --git a/plugins/inputs/luma_checkin/debounce_test.go b/plugins/inputs/luma_checkin/debounce_test.go new file mode 100644 index 0000000000..17ebcb4460 --- /dev/null +++ b/plugins/inputs/luma_checkin/debounce_test.go @@ -0,0 +1,62 @@ +package luma_checkin + +import ( + "testing" + "time" +) + +func TestDebouncerAcceptsThenRejects(t *testing.T) { + d := newDebouncer(30 * time.Second) + now := time.Unix(1_700_000_000, 0) + d.now = func() time.Time { return now } + + if !d.TryRecord("g-1") { + t.Fatalf("first record should be accepted") + } + if d.TryRecord("g-1") { + t.Fatalf("immediate re-record should be rejected") + } + + now = now.Add(29 * time.Second) + if d.TryRecord("g-1") { + t.Fatalf("re-record within window should be rejected") + } + + now = now.Add(2 * time.Second) + if !d.TryRecord("g-1") { + t.Fatalf("re-record after window should be accepted") + } +} + +func TestDebouncerDistinctKeysIndependent(t *testing.T) { + d := newDebouncer(30 * time.Second) + if !d.TryRecord("g-1") { + t.Fatalf("g-1 first should be accepted") + } + if !d.TryRecord("g-2") { + t.Fatalf("g-2 first should be accepted") + } +} + +func TestDebouncerPrunesOldEntries(t *testing.T) { + d := newDebouncer(1 * time.Second) + now := time.Unix(1_700_000_000, 0) + d.now = func() time.Time { return now } + + has := func(key string) bool { _, ok := d.seen[key]; return ok } + + d.TryRecord("g-old") + if !has("g-old") { + t.Fatalf("expected g-old to be recorded") + } + + now = now.Add(15 * time.Second) + d.TryRecord("g-fresh") + + if has("g-old") { + t.Fatalf("expected g-old to be pruned after 15x window") + } + if !has("g-fresh") { + t.Fatalf("expected g-fresh to be present") + } +} diff --git a/plugins/inputs/luma_checkin/decode.go b/plugins/inputs/luma_checkin/decode.go new file mode 100644 index 0000000000..4614fd9160 --- /dev/null +++ b/plugins/inputs/luma_checkin/decode.go @@ -0,0 +1,34 @@ +package luma_checkin + +import ( + "bytes" + "errors" + "image" + _ "image/jpeg" + + "github.com/makiuchi-d/gozxing" + "github.com/makiuchi-d/gozxing/qrcode" +) + +// errQRNotFound signals that no QR code was found in the frame. Callers should +// treat this as the common case and skip silently. +var errQRNotFound = errors.New("luma_checkin: no qr code in frame") + +// decodeQR decodes a single QR code from a JPEG-encoded frame. It returns the +// raw text payload of the code, or errQRNotFound if no code is present. +func decodeQR(jpegBytes []byte) (string, error) { + img, _, err := image.Decode(bytes.NewReader(jpegBytes)) + if err != nil { + return "", err + } + bmp, err := gozxing.NewBinaryBitmapFromImage(img) + if err != nil { + return "", err + } + reader := qrcode.NewQRCodeReader() + result, err := reader.Decode(bmp, nil) + if err != nil { + return "", errQRNotFound + } + return result.GetText(), nil +} diff --git a/plugins/inputs/luma_checkin/decode_test.go b/plugins/inputs/luma_checkin/decode_test.go new file mode 100644 index 0000000000..cb9ccfb2fe --- /dev/null +++ b/plugins/inputs/luma_checkin/decode_test.go @@ -0,0 +1,71 @@ +package luma_checkin + +import ( + "bytes" + "errors" + "image" + "image/color" + "image/jpeg" + "testing" + + "github.com/makiuchi-d/gozxing" + "github.com/makiuchi-d/gozxing/qrcode" +) + +func TestDecodeQRRoundTrip(t *testing.T) { + const payload = "https://luma.com/check-in/evt-test?pk=g-test-123" + + jpegBytes := encodeQRAsJPEG(t, payload, 360) + + got, err := decodeQR(jpegBytes) + if err != nil { + t.Fatalf("DecodeQR: %v", err) + } + if got != payload { + t.Errorf("payload: got %q want %q", got, payload) + } +} + +func TestDecodeQRReturnsNotFoundOnBlankFrame(t *testing.T) { + jpegBytes := encodeBlankJPEG(t, 100, 100) + + _, err := decodeQR(jpegBytes) + if !errors.Is(err, errQRNotFound) { + t.Fatalf("expected ErrQRNotFound, got %v", err) + } +} + +func encodeQRAsJPEG(t *testing.T, payload string, size int) []byte { + t.Helper() + writer := qrcode.NewQRCodeWriter() + bm, err := writer.Encode(payload, gozxing.BarcodeFormat_QR_CODE, size, size, nil) + if err != nil { + t.Fatalf("encode QR: %v", err) + } + var buf bytes.Buffer + if err := jpeg.Encode(&buf, bm, &jpeg.Options{Quality: 90}); err != nil { + t.Fatalf("jpeg encode: %v", err) + } + return buf.Bytes() +} + +func encodeBlankJPEG(t *testing.T, w, h int) []byte { + t.Helper() + img := newBlankImage(w, h) + var buf bytes.Buffer + if err := jpeg.Encode(&buf, img, &jpeg.Options{Quality: 90}); err != nil { + t.Fatalf("jpeg encode: %v", err) + } + return buf.Bytes() +} + +func newBlankImage(w, h int) image.Image { + img := image.NewGray(image.Rect(0, 0, w, h)) + white := color.Gray{Y: 255} + for y := 0; y < h; y++ { + for x := 0; x < w; x++ { + img.SetGray(x, y, white) + } + } + return img +} diff --git a/plugins/inputs/luma_checkin/luma_checkin.go b/plugins/inputs/luma_checkin/luma_checkin.go new file mode 100644 index 0000000000..3f42a30b6d --- /dev/null +++ b/plugins/inputs/luma_checkin/luma_checkin.go @@ -0,0 +1,374 @@ +package luma_checkin + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sync" + "time" + + "go.uber.org/zap" + + "github.com/openmind/om1/internal/inputs" + "github.com/openmind/om1/internal/logger" + "github.com/openmind/om1/internal/providers" + "github.com/openmind/om1/internal/providers/luma" + video "github.com/openmind/om1/internal/providers/vlm" +) + +const ( + scannerName = "LumaCheckin" + scannerRTSPName = "LumaCheckinRTSP" + scannerDescriptor = "Luma Check-In" + scannerMaxMessages = 8 + scanChannelBuffer = 4 + + defaultLumaTimeout = 4 * time.Second +) + +func init() { + inputs.Register(scannerName, NewLumaCheckin) + inputs.Register(scannerRTSPName, NewLumaCheckinRTSP) +} + +type LumaConfig struct { + APIKey string `json:"api_key"` + BaseURL string `json:"base_url"` + EventAPIID string `json:"event_api_id"` + RequestTimeoutSeconds float64 `json:"request_timeout_seconds"` + SessionKey string `json:"session_key"` +} + +// Config holds the JSON configuration for the LumaCheckin input plugin. +type Config struct { + CameraIndex int `json:"camera_index"` + RTSPURL string `json:"rtsp_url"` + CaptureFPS int `json:"capture_fps"` + DecodeFPS int `json:"decode_fps"` + Width int `json:"resolution_width"` + Height int `json:"resolution_height"` + JPEGQuality int `json:"jpeg_quality"` + DedupeWindowSeconds float64 `json:"dedupe_window_seconds"` + Luma *LumaConfig `json:"luma"` +} + +type frameSource interface { + Start(ctx context.Context) <-chan video.Frame + Stop() +} + +type guestLookup interface { + GetGuest(ctx context.Context, pk string) (*luma.Guest, error) + CheckIn(ctx context.Context, guest *luma.Guest) error +} + +type sensor struct { + name string + cfg Config + log *zap.Logger + source frameSource + debouncer *debouncer + + luma guestLookup + lumaTimeout time.Duration + expectedEventID string + + mu sync.Mutex + messages []inputs.Message + stopped bool + cancel context.CancelFunc +} + +// NewLumaCheckin constructs a camera-backed Luma check-in sensor. +func NewLumaCheckin(configMap map[string]any) (inputs.Sensor, error) { + cfg := parseConfig(configMap) + + log := logger.Get().Named(scannerName) + log.Info("initializing", + zap.Int("camera_index", cfg.CameraIndex), + zap.String("rtsp_url", cfg.RTSPURL), + zap.Int("capture_fps", cfg.CaptureFPS), + zap.Int("decode_fps", cfg.DecodeFPS), + zap.Int("width", cfg.Width), + zap.Int("height", cfg.Height), + zap.Float64("dedupe_window_seconds", cfg.DedupeWindowSeconds), + ) + + source := video.NewVideoStream(video.VideoStreamConfig{ + DeviceIndex: cfg.CameraIndex, + FPS: cfg.CaptureFPS, + Width: cfg.Width, + Height: cfg.Height, + JPEGQuality: cfg.JPEGQuality, + }) + return newSensor(cfg, log, source), nil +} + +// NewLumaCheckinRTSP constructs an RTSP-backed Luma check-in sensor. +func NewLumaCheckinRTSP(configMap map[string]any) (inputs.Sensor, error) { + cfg := parseConfig(configMap) + if cfg.RTSPURL == "" { + cfg.RTSPURL = "rtsp://localhost:8554/top_camera_raw" + } + + log := logger.Get().Named(scannerRTSPName) + log.Info("initializing", + zap.Int("camera_index", cfg.CameraIndex), + zap.String("rtsp_url", cfg.RTSPURL), + zap.Int("capture_fps", cfg.CaptureFPS), + zap.Int("decode_fps", cfg.DecodeFPS), + zap.Int("width", cfg.Width), + zap.Int("height", cfg.Height), + zap.Float64("dedupe_window_seconds", cfg.DedupeWindowSeconds), + ) + + source := video.NewVideoRTSPStream(video.VideoRTSPStreamConfig{ + RTSPURL: cfg.RTSPURL, + FPS: cfg.CaptureFPS, + Width: cfg.Width, + Height: cfg.Height, + JPEGQuality: cfg.JPEGQuality, + }) + return newSensor(cfg, log, source), nil +} + +func newSensor(cfg Config, log *zap.Logger, source frameSource) *sensor { + window := time.Duration(cfg.DedupeWindowSeconds * float64(time.Second)) + s := &sensor{ + name: log.Name(), + cfg: cfg, + log: log, + source: source, + debouncer: newDebouncer(window), + } + + if cfg.Luma != nil && cfg.Luma.APIKey != "" && cfg.Luma.EventAPIID != "" { + timeout := time.Duration(cfg.Luma.RequestTimeoutSeconds * float64(time.Second)) + if timeout <= 0 { + timeout = defaultLumaTimeout + } + var opts []func(*luma.Client) + if cfg.Luma.SessionKey != "" { + opts = append(opts, luma.WithSessionKey(cfg.Luma.SessionKey)) + } + s.luma = luma.NewClient(cfg.Luma.BaseURL, cfg.Luma.APIKey, cfg.Luma.EventAPIID, timeout, opts...) + s.lumaTimeout = timeout + s.expectedEventID = cfg.Luma.EventAPIID + log.Info("luma lookup enabled", + zap.String("event_api_id", cfg.Luma.EventAPIID), + zap.Duration("timeout", timeout), + ) + } + + return s +} + +func parseConfig(configMap map[string]any) Config { + var cfg Config + if b, err := json.Marshal(configMap); err == nil { + _ = json.Unmarshal(b, &cfg) + } + if cfg.CaptureFPS <= 0 { + cfg.CaptureFPS = 15 + } + if cfg.DecodeFPS <= 0 { + cfg.DecodeFPS = 5 + } + if cfg.DecodeFPS > cfg.CaptureFPS { + cfg.DecodeFPS = cfg.CaptureFPS + } + if cfg.Width <= 0 { + cfg.Width = 640 + } + if cfg.Height <= 0 { + cfg.Height = 480 + } + if cfg.JPEGQuality <= 0 { + cfg.JPEGQuality = 60 + } + if cfg.DedupeWindowSeconds <= 0 { + cfg.DedupeWindowSeconds = 30 + } + return cfg +} + +func (s *sensor) Listen(ctx context.Context) (<-chan any, error) { + ctx, cancel := context.WithCancel(ctx) + s.mu.Lock() + s.cancel = cancel + s.mu.Unlock() + + frames := s.source.Start(ctx) + out := make(chan any, scanChannelBuffer) + + stride := s.cfg.CaptureFPS / s.cfg.DecodeFPS + if stride < 1 { + stride = 1 + } + + go func() { + defer close(out) + defer s.Stop() + + var counter uint64 + for { + select { + case <-ctx.Done(): + return + case frame, ok := <-frames: + if !ok { + return + } + counter++ + if counter%uint64(stride) != 0 { + continue + } + + text, err := decodeQR(frame.JPEG) + if err != nil { + if !errors.Is(err, errQRNotFound) { + s.log.Debug("decode error", zap.Error(err)) + } + continue + } + eventID, pk, valid := parseLumaCheckinURL(text) + if !valid { + s.log.Debug("ignoring non-luma qr", zap.String("text", truncate(text, 80))) + continue + } + if !s.debouncer.TryRecord(pk) { + s.log.Debug("debounced", zap.String("pk", pk)) + continue + } + + msg := s.formatScanMessage(ctx, pk, eventID) + s.log.Info("emitted scan", zap.String("pk", pk), zap.String("event", eventID)) + select { + case out <- msg: + default: + s.log.Warn("scan channel full, dropping", zap.String("pk", pk)) + } + } + } + }() + + return out, nil +} + +func (s *sensor) Poll(context.Context) (any, error) { return nil, nil } + +// RawToText converts a raw scan event into a timestamped Message and appends it +// to the bounded in-memory history. +func (s *sensor) RawToText(_ context.Context, raw any) (*inputs.Message, error) { + text, ok := raw.(string) + if !ok || text == "" { + return nil, nil + } + msg := inputs.NewMessage(text) + + s.mu.Lock() + s.messages = append(s.messages, *msg) + if len(s.messages) > scannerMaxMessages { + s.messages = s.messages[len(s.messages)-scannerMaxMessages:] + } + s.mu.Unlock() + + return msg, nil +} + +// FormattedLatestBuffer returns the newest scan formatted for the LLM prompt +// and clears the history. Returns "" when empty. +func (s *sensor) FormattedLatestBuffer() string { + s.mu.Lock() + defer s.mu.Unlock() + + if len(s.messages) == 0 { + return "" + } + + latest := s.messages[len(s.messages)-1] + result := fmt.Sprintf("\n%s: '%s'\n", scannerDescriptor, latest.Message) + + ts := time.Unix(0, int64(latest.Timestamp*1e9)) + providers.IO().AddInput(s.name, latest.Message, ts) + s.messages = nil + + return result +} + +// TriggersTick opts the scanner into waking the cortex loop on every fresh scan. +func (s *sensor) TriggersTick() bool { return true } + +func (s *sensor) Stop() { + s.mu.Lock() + if s.stopped { + s.mu.Unlock() + return + } + s.stopped = true + cancel := s.cancel + s.mu.Unlock() + + if cancel != nil { + cancel() + } + s.source.Stop() + s.log.Info("stopping sensor") +} + +func truncate(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n] + "..." +} + +// formatScanMessage performs the Luma guest lookup and check-in. +func (s *sensor) formatScanMessage(ctx context.Context, pk, eventID string) string { + if s.luma == nil { + return fmt.Sprintf("qr_scan: pk=%s event=%s", pk, eventID) + } + + if s.expectedEventID != "" && eventID != "" && eventID != s.expectedEventID { + s.log.Info("scan event mismatch", + zap.String("expected", s.expectedEventID), + zap.String("got", eventID), + ) + return fmt.Sprintf("qr_scan_failed: pk=%s reason=event_mismatch", pk) + } + + lookupCtx, cancel := context.WithTimeout(ctx, s.lumaTimeout) + defer cancel() + + guest, err := s.luma.GetGuest(lookupCtx, pk) + if err != nil { + switch { + case errors.Is(err, luma.ErrNotFound): + s.log.Info("luma lookup: not found", zap.String("pk", pk)) + return fmt.Sprintf("qr_scan_failed: pk=%s reason=guest_not_registered", pk) + case errors.Is(err, luma.ErrUnauthorized): + s.log.Error("luma lookup: unauthorized") + return fmt.Sprintf("qr_scan_failed: pk=%s reason=luma_auth", pk) + default: + s.log.Warn("luma lookup failed", zap.String("pk", pk), zap.Error(err)) + return fmt.Sprintf("qr_scan_failed: pk=%s reason=lookup_error", pk) + } + } + if guest == nil { + return fmt.Sprintf("qr_scan_failed: pk=%s reason=empty_response", pk) + } + + name := luma.FirstName(guest) + + if err := s.luma.CheckIn(ctx, guest); err != nil { + s.log.Warn("luma check-in failed", zap.String("pk", pk), zap.Error(err)) + } else { + s.log.Info("luma check-in ok", zap.String("pk", pk), zap.String("name", name)) + } + + luma.RecordCheckIn(name, time.Now()) + + s.log.Info("luma lookup ok", zap.String("pk", pk), zap.String("name", name)) + return fmt.Sprintf("qr_scan: name=%s", name) +} diff --git a/plugins/inputs/luma_checkin/url_parse.go b/plugins/inputs/luma_checkin/url_parse.go new file mode 100644 index 0000000000..70ef08f012 --- /dev/null +++ b/plugins/inputs/luma_checkin/url_parse.go @@ -0,0 +1,30 @@ +package luma_checkin + +import ( + "net/url" + "strings" +) + +// parseLumaCheckinURL extracts the event ID and guest/ticket key from a Luma check-in URL. +func parseLumaCheckinURL(s string) (eventID, pk string, ok bool) { + u, err := url.Parse(strings.TrimSpace(s)) + if err != nil { + return "", "", false + } + if u.Scheme != "http" && u.Scheme != "https" { + return "", "", false + } + host := strings.ToLower(u.Hostname()) + if host != "luma.com" && host != "www.luma.com" && host != "lu.ma" { + return "", "", false + } + parts := strings.Split(strings.Trim(u.Path, "/"), "/") + if len(parts) != 2 || parts[0] != "check-in" || parts[1] == "" { + return "", "", false + } + pk = u.Query().Get("pk") + if pk == "" { + return "", "", false + } + return parts[1], pk, true +} diff --git a/plugins/inputs/luma_checkin/url_parse_test.go b/plugins/inputs/luma_checkin/url_parse_test.go new file mode 100644 index 0000000000..cd930a6282 --- /dev/null +++ b/plugins/inputs/luma_checkin/url_parse_test.go @@ -0,0 +1,116 @@ +package luma_checkin + +import ( + "testing" +) + +func TestParseLumaCheckinURL(t *testing.T) { + cases := []struct { + name string + input string + wantEvent string + wantPK string + wantOK bool + }{ + { + name: "happy path luma.com", + input: "https://luma.com/check-in/evt-abc?pk=g-12345", + wantEvent: "evt-abc", + wantPK: "g-12345", + wantOK: true, + }, + { + name: "lu.ma short host", + input: "https://lu.ma/check-in/evt-abc?pk=tk_xyz", + wantEvent: "evt-abc", + wantPK: "tk_xyz", + wantOK: true, + }, + { + name: "www.luma.com", + input: "https://www.luma.com/check-in/evt-abc?pk=g-12345", + wantEvent: "evt-abc", + wantPK: "g-12345", + wantOK: true, + }, + { + name: "trailing slash on path", + input: "https://luma.com/check-in/evt-abc/?pk=g-12345", + wantEvent: "evt-abc", + wantPK: "g-12345", + wantOK: true, + }, + { + name: "http scheme accepted", + input: "http://luma.com/check-in/evt-abc?pk=g-12345", + wantEvent: "evt-abc", + wantPK: "g-12345", + wantOK: true, + }, + { + name: "percent-encoded pk", + input: "https://luma.com/check-in/evt-abc?pk=g%2D12345", + wantEvent: "evt-abc", + wantPK: "g-12345", + wantOK: true, + }, + { + name: "missing pk", + input: "https://luma.com/check-in/evt-abc", + wantOK: false, + }, + { + name: "empty pk", + input: "https://luma.com/check-in/evt-abc?pk=", + wantOK: false, + }, + { + name: "wrong host", + input: "https://example.com/check-in/evt-abc?pk=g-12345", + wantOK: false, + }, + { + name: "wrong path", + input: "https://luma.com/event/evt-abc?pk=g-12345", + wantOK: false, + }, + { + name: "extra path segment", + input: "https://luma.com/check-in/evt-abc/extra?pk=g-12345", + wantOK: false, + }, + { + name: "missing event id", + input: "https://luma.com/check-in/?pk=g-12345", + wantOK: false, + }, + { + name: "non-url garbage", + input: "hello world", + wantOK: false, + }, + { + name: "ftp scheme rejected", + input: "ftp://luma.com/check-in/evt-abc?pk=g-12345", + wantOK: false, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + gotEvent, gotPK, gotOK := parseLumaCheckinURL(tc.input) + if gotOK != tc.wantOK { + t.Fatalf("ok mismatch: got %v want %v", gotOK, tc.wantOK) + } + if !tc.wantOK { + return + } + if gotEvent != tc.wantEvent { + t.Errorf("event: got %q want %q", gotEvent, tc.wantEvent) + } + if gotPK != tc.wantPK { + t.Errorf("pk: got %q want %q", gotPK, tc.wantPK) + } + }) + } +}