ax-devil-mqtt 0.2.2

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

ax-devil-mqtt

Python package for working with Axis devices MQTT functionality.

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
🔌 MQTT Connection	Connect to MQTT brokers and Axis devices	RawMQTTManager	ax-devil-mqtt device monitor
📊 Analytics Streaming	Stream analytics data from Axis devices via MQTT	AnalyticsManager	ax-devil-mqtt device monitor
💾 Data Recording	Record any 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_device_api import DeviceConfig

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

def message_callback(message):
    print(message)

# 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

def message_callback(message):
    print(message)

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

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

CLI Usage Examples

🔍 (Optional): Use ax-devil-device-api CLI to find available analytics streams.

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

📊 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.