From b461ec8ded3e08db0dea37804b77c6043ca188a6 Mon Sep 17 00:00:00 2001 From: Sebastian Norling <1932208+Bazze@users.noreply.github.com> Date: Thu, 21 May 2026 14:30:50 +0200 Subject: [PATCH] feat(event/stream): decode NotificationMessage into normalized Event MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the Decode entry point that converts the ONVIF wire form into the package's typed Event. The agent (and any other consumer) no longer has to walk NotificationMessage.Message.Message.Data.SimpleItem chains and hand-special-case per-vendor data item names. Decoding rules -------------- * Topic -> Kind via the verified Classify table. * PropertyOperation parses Initialized/Changed/Deleted; absent or unrecognised -> PropertyUnknown (the attribute is optional per WS-Notification). * UtcTime parses RFC3339Nano first, RFC3339 second, normalised to UTC. Absent or unparseable -> DeviceTime is zero. Camera clocks drift; the type doc already steers callers to prefer Timestamp. * State extraction scans Data items in order for the first boolean-like value (true/false/1/0/active/inactive, case-insensitive). This handles every vendor data item in the verified table — IsMotion, State, IsTamper, LogicalState, active, Motion, triggered, SoundDetection, TamperingDetection — without a per-kind switch. * Edge-triggered topics (LineDetector/Crossed with only ObjectId) yield StateUnknown, matching the topic-rule doc note. * Source and Data are full ONVIF SimpleItem name->value maps so callers retain multi-item info (AXIS AOA active+classType+confidence, digital I/O InputToken+LogicalState, analytics VideoSourceConfigurationToken+ Rule). Empty notifications yield nil maps, matching the Event zero-value contract from types_test.go. * Topic, Source and Data are always populated even when Kind is KindUnknown, so consumers can log/route unclassified events. Tests cover the AXIS motion happy path, the inactive case, the Hanwha numeric-string variant, the Avigilon 'active' literal, multi-item AOA decode, the LineDetector edge-trigger semantic, unknown-topic wire preservation, every PropertyOperation literal, RFC3339 with sub-second and timezone offsets, and case-insensitive State extraction. --- event/stream/decode.go | 104 +++++++++++++++ event/stream/decode_test.go | 249 ++++++++++++++++++++++++++++++++++++ 2 files changed, 353 insertions(+) create mode 100644 event/stream/decode.go create mode 100644 event/stream/decode_test.go diff --git a/event/stream/decode.go b/event/stream/decode.go new file mode 100644 index 0000000..1d96941 --- /dev/null +++ b/event/stream/decode.go @@ -0,0 +1,104 @@ +package stream + +import ( + "strings" + "time" + + "github.com/kerberos-io/onvif/event" +) + +// Decode converts a single ONVIF NotificationMessage into the package's +// normalized Event representation. +// +// deviceID is supplied by the caller because the message itself does not +// identify the originating camera. observedAt is recorded verbatim as +// Event.Timestamp; the camera-reported wsnt:UtcTime attribute (when +// present and parseable) populates Event.DeviceTime. +// +// When the Topic does not match any classifier rule the returned Event +// has Kind == KindUnknown but Source, Data and Topic are still populated +// so consumers can fall back to inspecting the wire form. +func Decode(msg event.NotificationMessage, deviceID string, observedAt time.Time) Event { + topic := string(msg.Topic.TopicKinds) + desc := msg.Message.Message + return Event{ + Kind: Classify(topic), + State: extractState(desc.Data.SimpleItem), + Operation: parsePropertyOperation(string(desc.PropertyOperation)), + DeviceID: deviceID, + Source: simpleItemsToMap(desc.Source.SimpleItem), + Data: simpleItemsToMap(desc.Data.SimpleItem), + Topic: topic, + Timestamp: observedAt, + DeviceTime: parseDeviceTime(string(desc.UtcTime)), + } +} + +// simpleItemsToMap collapses ONVIF SimpleItem lists to a Name->Value map. +// Returns nil for an empty list so empty notifications do not allocate +// and match the Event zero-value contract. +func simpleItemsToMap(items []event.SimpleItem) map[string]string { + if len(items) == 0 { + return nil + } + m := make(map[string]string, len(items)) + for _, it := range items { + m[string(it.Name)] = string(it.Value) + } + return m +} + +// extractState scans Data items for a boolean-like value and returns the +// first one as a State. Returns StateUnknown when no item parses — this +// is the correct outcome for edge-triggered topics such as +// LineDetector/Crossed whose Data carries only an ObjectId. +// +// Iteration order over the original []SimpleItem is preserved so the +// behaviour stays deterministic per notification. (Map iteration is not +// involved; simpleItemsToMap is a separate path.) +func extractState(items []event.SimpleItem) State { + for _, it := range items { + switch strings.ToLower(strings.TrimSpace(string(it.Value))) { + case "true", "1", "active": + return StateActive + case "false", "0", "inactive": + return StateInactive + } + } + return StateUnknown +} + +// parsePropertyOperation parses the wsnt:PropertyOperation attribute. +// The attribute is optional per WS-Notification; an empty or unrecognised +// value yields PropertyUnknown. +func parsePropertyOperation(s string) PropertyOperation { + switch s { + case "Initialized": + return PropertyInitialized + case "Changed": + return PropertyChanged + case "Deleted": + return PropertyDeleted + default: + return PropertyUnknown + } +} + +// parseDeviceTime parses the wsnt:UtcTime attribute, returning the zero +// time when the attribute is absent or unparseable. The result is +// normalised to UTC so equality comparisons across timezones work. +// +// xsd:dateTime in ONVIF messages is RFC 3339 in practice; we try +// time.RFC3339Nano first (covers sub-second precision) and fall back to +// time.RFC3339 for cameras that drop the fractional part. +func parseDeviceTime(s string) time.Time { + if s == "" { + return time.Time{} + } + for _, layout := range []string{time.RFC3339Nano, time.RFC3339} { + if t, err := time.Parse(layout, s); err == nil { + return t.UTC() + } + } + return time.Time{} +} diff --git a/event/stream/decode_test.go b/event/stream/decode_test.go new file mode 100644 index 0000000..edce4c8 --- /dev/null +++ b/event/stream/decode_test.go @@ -0,0 +1,249 @@ +package stream + +import ( + "testing" + "time" + + "github.com/kerberos-io/onvif/event" + "github.com/kerberos-io/onvif/xsd" + "github.com/stretchr/testify/assert" +) + +// msg builds a NotificationMessage from the topic and a (PropertyOperation, +// UtcTime, source items, data items) tuple so tests stay short and intent +// is visible at the call site. +func msg(topic, propOp, utcTime string, source, data map[string]string) event.NotificationMessage { + toItems := func(m map[string]string) []event.SimpleItem { + if len(m) == 0 { + return nil + } + items := make([]event.SimpleItem, 0, len(m)) + for k, v := range m { + items = append(items, event.SimpleItem{ + Name: xsd.AnyType(k), + Value: xsd.AnyType(v), + }) + } + return items + } + return event.NotificationMessage{ + Topic: event.Topic{TopicKinds: xsd.String(topic)}, + Message: event.MessageBody{ + Message: event.MessageDescription{ + PropertyOperation: xsd.AnyType(propOp), + UtcTime: xsd.AnyType(utcTime), + Source: event.Source{SimpleItem: toItems(source)}, + Data: event.Data{SimpleItem: toItems(data)}, + }, + }, + } +} + +func TestDecode_MotionActive(t *testing.T) { + observedAt := time.Date(2026, 5, 21, 10, 30, 1, 0, time.UTC) + in := msg( + "tns1:RuleEngine/CellMotionDetector/Motion", + "Changed", + "2026-05-21T10:30:00Z", + map[string]string{ + "VideoSourceConfigurationToken": "VideoSourceConfigToken0", + "Rule": "MyMotionRule", + }, + map[string]string{"IsMotion": "true"}, + ) + + ev := Decode(in, "axis-cam-01", observedAt) + + assert.Equal(t, KindMotion, ev.Kind) + assert.Equal(t, StateActive, ev.State) + assert.Equal(t, PropertyChanged, ev.Operation) + assert.Equal(t, "axis-cam-01", ev.DeviceID) + assert.Equal(t, "VideoSourceConfigToken0", ev.Source["VideoSourceConfigurationToken"]) + assert.Equal(t, "MyMotionRule", ev.Source["Rule"]) + assert.Equal(t, "true", ev.Data["IsMotion"]) + assert.Equal(t, "tns1:RuleEngine/CellMotionDetector/Motion", ev.Topic) + assert.True(t, ev.Timestamp.Equal(observedAt)) + assert.Equal(t, time.Date(2026, 5, 21, 10, 30, 0, 0, time.UTC), ev.DeviceTime) +} + +func TestDecode_MotionInactive(t *testing.T) { + in := msg( + "tns1:VideoSource/MotionAlarm", + "Changed", + "", + nil, + map[string]string{"State": "false"}, + ) + ev := Decode(in, "dev", time.Now()) + assert.Equal(t, KindMotion, ev.Kind) + assert.Equal(t, StateInactive, ev.State) +} + +func TestDecode_HanwhaNumericMotionValue(t *testing.T) { + // Hanwha emits xsd:string values "0"/"1" instead of xsd:boolean. + in := msg( + "tns1:VideoAnalytics/tnssamsung:MotionDetection", + "Changed", + "", + nil, + map[string]string{"Motion": "1"}, + ) + ev := Decode(in, "dev", time.Now()) + assert.Equal(t, KindMotion, ev.Kind) + assert.Equal(t, StateActive, ev.State) +} + +func TestDecode_AvigilonActiveLiteral(t *testing.T) { + // Avigilon and a handful of older firmwares emit "active"/"inactive" + // as the Data value rather than a boolean. + in := msg( + "tns1:Device/tns1:Trigger/tns1:Relay", + "Changed", + "", + map[string]string{"RelayToken": "Relay-1"}, + map[string]string{"LogicalState": "active"}, + ) + ev := Decode(in, "dev", time.Now()) + assert.Equal(t, KindDigitalOutput, ev.Kind) + assert.Equal(t, StateActive, ev.State) + assert.Equal(t, "Relay-1", ev.Source["RelayToken"]) +} + +func TestDecode_AxisObjectAnalyticsMultiItem(t *testing.T) { + // AOA emits active + classType + confidence in the same Data list. + // The decoder must preserve every item; State picks the first + // boolean-like value, which is 'active'. + in := msg( + "tnsaxis:CameraApplicationPlatform/ObjectAnalytics/Device1Scenario1", + "Changed", + "", + map[string]string{"Source": "device1Scene1"}, + map[string]string{ + "active": "1", + "classType": "Human", + "confidence": "92", + }, + ) + ev := Decode(in, "dev", time.Now()) + assert.Equal(t, KindObjectDetected, ev.Kind) + assert.Equal(t, StateActive, ev.State) + assert.Equal(t, "Human", ev.Data["classType"]) + assert.Equal(t, "92", ev.Data["confidence"]) + assert.Equal(t, "1", ev.Data["active"]) +} + +func TestDecode_LineDetectorCrossedHasNoState(t *testing.T) { + // Edge-triggered topic — Data carries ObjectId, not a boolean. State + // must remain Unknown so consumers do not misread it as level-Active. + in := msg( + "tns1:RuleEngine/LineDetector/Crossed", + "Changed", + "", + map[string]string{"VideoSourceConfigurationToken": "vsct0", "Rule": "LineRule"}, + map[string]string{"ObjectId": "42"}, + ) + ev := Decode(in, "dev", time.Now()) + assert.Equal(t, KindObjectDetected, ev.Kind) + assert.Equal(t, StateUnknown, ev.State) + assert.Equal(t, "42", ev.Data["ObjectId"]) +} + +func TestDecode_UnknownTopicStillPreservesWireData(t *testing.T) { + // Kind unknown does not mean discard: consumers may want to log or + // route on the raw topic when classification misses. + in := msg( + "tns1:UserAlarm/IVA", + "", + "", + nil, + map[string]string{"Custom": "true"}, + ) + ev := Decode(in, "dev", time.Now()) + assert.Equal(t, KindUnknown, ev.Kind) + assert.Equal(t, "tns1:UserAlarm/IVA", ev.Topic) + assert.Equal(t, "true", ev.Data["Custom"]) +} + +func TestDecode_PropertyOperationVariants(t *testing.T) { + tests := []struct { + name string + in string + want PropertyOperation + }{ + {"initialized", "Initialized", PropertyInitialized}, + {"changed", "Changed", PropertyChanged}, + {"deleted", "Deleted", PropertyDeleted}, + {"absent", "", PropertyUnknown}, + {"unrecognised", "Bogus", PropertyUnknown}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + in := msg("tns1:VideoSource/MotionAlarm", tc.in, "", nil, nil) + ev := Decode(in, "dev", time.Now()) + assert.Equal(t, tc.want, ev.Operation) + }) + } +} + +func TestDecode_DeviceTimeParsing(t *testing.T) { + tests := []struct { + name string + in string + want time.Time + }{ + {"rfc3339_utc", "2026-05-21T10:30:00Z", time.Date(2026, 5, 21, 10, 30, 0, 0, time.UTC)}, + {"rfc3339_with_offset", "2026-05-21T12:30:00+02:00", time.Date(2026, 5, 21, 10, 30, 0, 0, time.UTC)}, + {"rfc3339_subsecond", "2026-05-21T10:30:00.500Z", time.Date(2026, 5, 21, 10, 30, 0, 500_000_000, time.UTC)}, + {"absent", "", time.Time{}}, + {"unparseable", "not-a-date", time.Time{}}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + in := msg("tns1:VideoSource/MotionAlarm", "Changed", tc.in, nil, nil) + ev := Decode(in, "dev", time.Now()) + if tc.want.IsZero() { + assert.True(t, ev.DeviceTime.IsZero(), "DeviceTime=%v", ev.DeviceTime) + } else { + assert.True(t, ev.DeviceTime.Equal(tc.want), "got=%v want=%v", ev.DeviceTime, tc.want) + } + }) + } +} + +func TestDecode_EmptySourceAndDataYieldNilMaps(t *testing.T) { + // Matches the zero-value contract in types_test.go: callers can + // safely len() and index into Source/Data without nil-checking, but + // we do not allocate an empty map for empty notifications. + in := msg("tns1:VideoSource/MotionAlarm", "Changed", "", nil, nil) + ev := Decode(in, "dev", time.Now()) + assert.Nil(t, ev.Source) + assert.Nil(t, ev.Data) +} + +func TestDecode_StateValueIsCaseInsensitive(t *testing.T) { + tests := []struct { + name string + value string + want State + }{ + {"true_lower", "true", StateActive}, + {"true_upper", "TRUE", StateActive}, + {"true_mixed", "True", StateActive}, + {"false_lower", "false", StateInactive}, + {"false_mixed", "False", StateInactive}, + {"active_mixed", "Active", StateActive}, + {"inactive_mixed", "Inactive", StateInactive}, + {"one", "1", StateActive}, + {"zero", "0", StateInactive}, + {"empty", "", StateUnknown}, + {"nonsense", "maybe", StateUnknown}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + in := msg("tns1:VideoSource/MotionAlarm", "Changed", "", + nil, map[string]string{"State": tc.value}) + ev := Decode(in, "dev", time.Now()) + assert.Equal(t, tc.want, ev.State) + }) + } +}