mirror of
https://github.com/kerberos-io/onvif.git
synced 2026-08-23 15:08:33 +00:00
feat(event/stream): decode NotificationMessage into normalized Event
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.
This commit is contained in:
Sebastian Norling
104
event/stream/decode.go
Normal file
104
event/stream/decode.go
Normal file
@@ -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{}
|
||||
}
|
||||
249
event/stream/decode_test.go
Normal file
249
event/stream/decode_test.go
Normal file
@@ -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)
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user