ax-devil-mqtt 0.3.0

Python package for setting up and retrieving data from Axis devices using MQTT

ax-devil-mqtt

Python package for retrieving analytics data from Axis devices over MQTT.

See also: ax-devil-device-api for device API integration.

---

📋 Contents

• Feature Overview
• Quick Start
• Usage Examples
• Disclaimer
• License

---

🔍 Feature Overview

Feature	Description	Python API	CLI Tool
🔌 Device Setup	Configure Axis devices for analytics MQTT publishing	RawMQTTManager	ax-devil-mqtt device monitor
📊 Analytics Streaming	Stream analytics data from Axis devices with automated setup	AnalyticsManager	ax-devil-mqtt device monitor
💾 Data Recording	Record analytics MQTT data for later replay and analysis	manager.start(recording_file)	ax-devil-mqtt device monitor --record
⏯️ Replay	Replay recorded MQTT data for testing and development	ReplayManager	ax-devil-mqtt replay

---

🚀 Quick Start

Installation

pip install ax-devil-mqtt

Environment Variables

For an easier experience, you can set the following environment variables:

export AX_DEVIL_TARGET_ADDR=<device-ip>
export AX_DEVIL_TARGET_USER=<username>
export AX_DEVIL_TARGET_PASS=<password>
export AX_DEVIL_USAGE_CLI="safe" # Set to "unsafe" to skip SSL certificate verification for CLI calls

---

💻 Usage Examples

Python API Usage

🔌 MQTT Connection and Analytics Streaming

import time
from ax_devil_mqtt import AnalyticsManager
from ax_devil_mqtt.core.types import Message
from ax_devil_device_api import DeviceConfig

# Configure device
device_config = DeviceConfig.http(
    host="192.168.1.200",
    username="root",
    password="pass"
)

def message_callback(message: Message):
    print(f"Topic: {message.topic}")
    print(f"Payload: {message.payload}")
    print(f"Timestamp: {message.timestamp}")

# Create analytics manager
manager = AnalyticsManager(
    broker_host="192.168.1.100",
    broker_port=1883,
    device_config=device_config,
    analytics_data_source_key="com.axis.analytics_scene_description.v0.beta#1",
    message_callback=message_callback
)

manager.start()
# or manager.start(recording_file="recordings/some_file_name.jsonl")
time.sleep(10)
manager.stop()

⏯️ Replay

import time
from ax_devil_mqtt import ReplayManager
from ax_devil_mqtt.core.types import Message, ReplayStats

def message_callback(message: Message):
    print(f"Topic: {message.topic}")
    print(f"Payload: {message.payload}")
    print(f"Timestamp: {message.timestamp}")

def on_replay_complete(stats: ReplayStats):
    print(f"Replay completed!")
    print(f"  Total messages: {stats.message_count}")
    print(f"  Average drift: {stats.avg_drift:.2f}ms")
    print(f"  Max drift: {stats.max_drift:.2f}ms")

# Create a replay manager
manager = ReplayManager(
    recording_file="recordings/device_recording.jsonl",
    message_callback=message_callback,
    on_replay_complete=on_replay_complete
)

# Start the manager
manager.start()
time.sleep(10)
manager.stop()

CLI Usage Examples

🔍 Discover Available Analytics Streams

Using ax-devil-device-api:

ax-devil-device-api-analytics-mqtt sources

Or discover and list with ax-devil-mqtt:

ax-devil-mqtt device list-sources --device-ip <device-ip> --username <username> --password <password>

📊 Streaming Analytics Data Source

ax-devil-mqtt device monitor \
    --device-ip <device-ip> \
    --username <username> \
    --password <password> \
    --broker <broker-ip> \
    --port 1883 \
    --stream "com.axis.analytics_scene_description.v0.beta#1" \
    --duration 3600

💾 Recording MQTT Data

ax-devil-mqtt device monitor \
    --device-ip <device-ip> \
    --username <username> \
    --password <password> \
    --broker <broker-ip> \
    --port 1883 \
    --stream "com.axis.analytics_scene_description.v0.beta#1" \
    --record \
    --duration 3600

⏯️ Replaying Recorded Data

ax-devil-mqtt replay recordings/device_recording.jsonl

Example Scripts

Analytics Monitor Example

python src/ax_devil_mqtt/examples/analytics_monitor.py --host <broker-ip>

Replay Example

python src/ax_devil_mqtt/examples/replay.py recordings/device_recording.jsonl

Note: For more examples, check the examples directory in the source code.

---

⚠️ Disclaimer

This project is an independent, community-driven implementation and is not affiliated with or endorsed by Axis Communications AB. For official APIs and development resources, please refer to Axis Developer Community.

📄 License

MIT License - See LICENSE file for details.