Clean up events

axis/interfaces/mqtt.py

@@ -2,8 +2,6 @@
 
 from typing import Any
 
-import orjson
-
 from ..models.api_discovery import ApiId
 from ..models.mqtt import (
     API_VERSION,
@@ -25,28 +23,6 @@
 DEFAULT_TOPICS = ["//."]
 
 
-def mqtt_json_to_event(msg: bytes | str) -> dict[str, Any]:
-    """Convert JSON message from MQTT to event format."""
-    message = orjson.loads(msg)
-    topic = message["topic"].replace("onvif", "tns1").replace("axis", "tnsaxis")
-
-    source = source_idx = ""
-    if message["message"]["source"]:
-        source, source_idx = next(iter(message["message"]["source"].items()))
-
-    data_type = data_value = ""
-    if message["message"]["data"]:
-        data_type, data_value = next(iter(message["message"]["data"].items()))
-
-    return {
-        "topic": topic,
-        "source": source,
-        "source_idx": source_idx,
-        "type": data_type,
-        "value": data_value,
-    }
-
-
 class MqttClientHandler(ApiHandler[Any]):
     """MQTT Client for Axis devices."""
 

axis/models/event.py

@@ -10,18 +10,6 @@
 LOGGER = logging.getLogger(__name__)
 
 
-class EventGroup(enum.StrEnum):
-    """Logical grouping of events."""
-
-    INPUT = "input"
-    LIGHT = "light"
-    MOTION = "motion"
-    OUTPUT = "output"
-    PTZ = "ptz"
-    SOUND = "sound"
-    NONE = "none"
-
-
 class EventOperation(enum.StrEnum):
     """Possible operations of an event."""
 
@@ -41,22 +29,22 @@ def _missing_(cls, value: object) -> "EventOperation":
 class EventTopic(enum.StrEnum):
     """Supported event topics."""
 
-    DAY_NIGHT_VISION = "tns1:VideoSource/tnsaxis:DayNightVision"
-    FENCE_GUARD = "tnsaxis:CameraApplicationPlatform/FenceGuard"
-    LIGHT_STATUS = "tns1:Device/tnsaxis:Light/Status"
-    LOITERING_GUARD = "tnsaxis:CameraApplicationPlatform/LoiteringGuard"
-    MOTION_DETECTION = "tns1:VideoAnalytics/tnsaxis:MotionDetection"
-    MOTION_DETECTION_3 = "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1"
-    MOTION_DETECTION_4 = "tnsaxis:CameraApplicationPlatform/VMD"
-    MOTION_GUARD = "tnsaxis:CameraApplicationPlatform/MotionGuard"
-    OBJECT_ANALYTICS = "tnsaxis:CameraApplicationPlatform/ObjectAnalytics"
-    PIR = "tns1:Device/tnsaxis:Sensor/PIR"
-    PORT_INPUT = "tns1:Device/tnsaxis:IO/Port"
-    PORT_SUPERVISED_INPUT = "tns1:Device/tnsaxis:IO/SupervisedPort"
-    PTZ_IS_MOVING = "tns1:PTZController/tnsaxis:Move"
-    PTZ_ON_PRESET = "tns1:PTZController/tnsaxis:PTZPresets"
-    RELAY = "tns1:Device/Trigger/Relay"
-    SOUND_TRIGGER_LEVEL = "tns1:AudioSource/tnsaxis:TriggerLevel"
+    DAY_NIGHT_VISION = "onvif:VideoSource/axis:DayNightVision"
+    FENCE_GUARD = "axis:CameraApplicationPlatform/FenceGuard"
+    LIGHT_STATUS = "onvif:Device/axis:Light/Status"
+    LOITERING_GUARD = "axis:CameraApplicationPlatform/LoiteringGuard"
+    MOTION_DETECTION = "onvif:VideoAnalytics/axis:MotionDetection"
+    MOTION_DETECTION_3 = "onvif:RuleEngine/axis:VMD3/vmd3_video_1"
+    MOTION_DETECTION_4 = "axis:CameraApplicationPlatform/VMD"
+    MOTION_GUARD = "axis:CameraApplicationPlatform/MotionGuard"
+    OBJECT_ANALYTICS = "axis:CameraApplicationPlatform/ObjectAnalytics"
+    PIR = "onvif:Device/axis:Sensor/PIR"
+    PORT_INPUT = "onvif:Device/axis:IO/Port"
+    PORT_SUPERVISED_INPUT = "onvif:Device/axis:IO/SupervisedPort"
+    PTZ_IS_MOVING = "onvif:PTZController/axis:Move"
+    PTZ_ON_PRESET = "onvif:PTZController/axis:PTZPresets"
+    RELAY = "onvif:Device/Trigger/Relay"
+    SOUND_TRIGGER_LEVEL = "onvif:AudioSource/axis:TriggerLevel"
     UNKNOWN = "unknown"
 
     @classmethod
@@ -67,25 +55,6 @@ def _missing_(cls, value: object) -> "EventTopic":
         return EventTopic.UNKNOWN
 
 
-TOPIC_TO_GROUP = {
-    EventTopic.DAY_NIGHT_VISION: EventGroup.LIGHT,
-    EventTopic.FENCE_GUARD: EventGroup.MOTION,
-    EventTopic.LIGHT_STATUS: EventGroup.LIGHT,
-    EventTopic.LOITERING_GUARD: EventGroup.MOTION,
-    EventTopic.MOTION_DETECTION: EventGroup.MOTION,
-    EventTopic.MOTION_DETECTION_3: EventGroup.MOTION,
-    EventTopic.MOTION_DETECTION_4: EventGroup.MOTION,
-    EventTopic.MOTION_GUARD: EventGroup.MOTION,
-    EventTopic.OBJECT_ANALYTICS: EventGroup.MOTION,
-    EventTopic.PIR: EventGroup.MOTION,
-    EventTopic.PORT_INPUT: EventGroup.INPUT,
-    EventTopic.PORT_SUPERVISED_INPUT: EventGroup.INPUT,
-    EventTopic.PTZ_IS_MOVING: EventGroup.PTZ,
-    EventTopic.PTZ_ON_PRESET: EventGroup.PTZ,
-    EventTopic.RELAY: EventGroup.OUTPUT,
-    EventTopic.SOUND_TRIGGER_LEVEL: EventGroup.SOUND,
-}
-
 TOPIC_TO_STATE = {
     EventTopic.LIGHT_STATUS: "ON",
     EventTopic.RELAY: "active",
@@ -102,8 +71,8 @@ def _missing_(cls, value: object) -> "EventTopic":
 NOTIFICATION_MESSAGE = ("MetadataStream", "Event", "NotificationMessage")
 MESSAGE = (*NOTIFICATION_MESSAGE, "Message", "Message")
 TOPIC = (*NOTIFICATION_MESSAGE, "Topic", "#text")
-TIMESTAMP = (*MESSAGE, "@UtcTime")
-OPERATION = (*MESSAGE, "@PropertyOperation")
+TIMESTAMP = (*MESSAGE, "UtcTime")
+OPERATION = (*MESSAGE, "PropertyOperation")
 SOURCE = (*MESSAGE, "Source")
 DATA = (*MESSAGE, "Data")
 
@@ -128,23 +97,21 @@ def extract_name_value(
     item = data.get("SimpleItem", {})
     if isinstance(item, list):
         item = item[0]
-    return item.get("@Name", ""), item.get("@Value", "")
-    # return item.get("Name", ""), item.get("Value", "")
+    return item.get("Name", ""), item.get("Value", "")
 
 
 @dataclass
 class Event:
     """Event data from Axis device."""
 
-    data: dict[str, Any]
-    group: EventGroup
     id: str
     is_tripped: bool
     operation: EventOperation
     source: str
     state: str
     topic: str
     topic_base: EventTopic
+    data: dict[str, Any]
 
     @classmethod
     def decode(cls, data: bytes | dict[str, Any]) -> Self:
@@ -162,41 +129,44 @@ def _decode_from_dict(cls, data: dict[str, Any]) -> Self:
         source_idx = data.get(EVENT_SOURCE_IDX, "")
         value = data.get(EVENT_VALUE, "")
 
-        if (topic_base := EventTopic(topic)) == EventTopic.UNKNOWN:
+        if (topic_base := EventTopic(topic)) is EventTopic.UNKNOWN:
             _topic_base, _, _source_idx = topic.rpartition("/")
             topic_base = EventTopic(_topic_base)
             if source_idx == "":
                 source_idx = _source_idx
 
         if source_idx == "-1":
-            source_idx = "ANY" if source != "port" else ""
+            source_idx = "ANY"
 
         return cls(
-            data=data,
-            group=TOPIC_TO_GROUP.get(topic_base, EventGroup.NONE),
             id=source_idx,
             is_tripped=value == TOPIC_TO_STATE.get(topic_base, "1"),
             operation=operation,
             source=source,
             state=value,
             topic=topic,
             topic_base=topic_base,
+            data=data,
         )
 
     @classmethod
     def _decode_from_bytes(cls, data: bytes) -> Self:
         """Parse metadata xml."""
         raw = xmltodict.parse(
             data,
-            # attr_prefix="",
+            attr_prefix="",
             process_namespaces=True,
             namespaces=XML_NAMESPACES,
         )
 
         if raw.get("MetadataStream") is None:
             return cls._decode_from_dict({})
 
-        topic = traverse(raw, TOPIC)
+        topic = (
+            str(traverse(raw, TOPIC))
+            .replace("tns1", "onvif")
+            .replace("tnsaxis", "axis")
+        )
         # timestamp = traverse(raw, TIMESTAMP)
         operation = traverse(raw, OPERATION)
 

axis/models/mqtt.py

@@ -4,16 +4,44 @@
 
 from dataclasses import dataclass
 import enum
-from typing import Literal, NotRequired, Self
+from typing import Any, Literal, NotRequired, Self
 
 import orjson
 from typing_extensions import TypedDict
 
 from .api import CONTEXT, ApiRequest, ApiResponse
+from .event import (
+    EVENT_SOURCE,
+    EVENT_SOURCE_IDX,
+    EVENT_TOPIC,
+    EVENT_TYPE,
+    EVENT_VALUE,
+)
 
 API_VERSION = "1.0"
 
 
+def mqtt_json_to_event(msg: bytes | str) -> dict[str, Any]:
+    """Convert JSON message from MQTT to event format."""
+    message = orjson.loads(msg)
+
+    source = source_idx = ""
+    if message["message"]["source"]:
+        source, source_idx = next(iter(message["message"]["source"].items()))
+
+    data_type = data_value = ""
+    if message["message"]["data"]:
+        data_type, data_value = next(iter(message["message"]["data"].items()))
+
+    return {
+        EVENT_TOPIC: message["topic"],
+        EVENT_SOURCE: source,
+        EVENT_SOURCE_IDX: source_idx,
+        EVENT_TYPE: data_type,
+        EVENT_VALUE: data_value,
+    }
+
+
 class ErrorDataT(TypedDict):
     """Error data in response."""