Example Programs

pyzm v2 quick-start

#!/usr/bin/env python3
"""pyzm v2 quick-start examples.

Each section is self-contained.  Comment/uncomment what you need.

See also:
    Full documentation: https://pyzmv2.readthedocs.io/en/latest/
    Quick-start guide:  https://pyzmv2.readthedocs.io/en/latest/guide/quickstart.html
    Detection options:  https://pyzmv2.readthedocs.io/en/latest/guide/detection.html
    Remote server:      https://pyzmv2.readthedocs.io/en/latest/guide/serve.html
"""

import pyzm
print(f"pyzm {pyzm.__version__}")


# ============================================================================
# 1. ZM API CLIENT (no ML needed)
# ============================================================================

from pyzm import ZMClient

zm = ZMClient(
    api_url="https://demo.zoneminder.com/zm/api",
    user="zmuser",
    password="zmpass",
    # verify_ssl=False,  # for self-signed certs
)

print(f"ZM {zm.zm_version}, API {zm.api_version}")

# -- Monitors --
for m in zm.monitors():
    print(f"  Monitor {m.id}: {m.name} ({m.function}) {m.width}x{m.height}")

# -- Events (last hour) --
events = zm.events(since="1 hour ago", limit=5)
for ev in events:
    print(f"  Event {ev.id}: {ev.cause} ({ev.length:.1f}s, {ev.alarm_frames} alarm frames)")

# -- Single event --
if events:
    ev = zm.event(events[0].id)
    print(f"  Event detail: {ev.name} notes={ev.notes!r}")

# -- Zones for a monitor --
if zm.monitors():
    m = zm.monitors()[0]
    zones = m.get_zones()
    for z in zones:
        print(f"  Zone: {z.name} ({len(z.points)} points)")


# ============================================================================
# 2. ML DETECTION (no ZM needed)
# ============================================================================

from pyzm import Detector

# Quick start -- model names are resolved against base_path on disk
detector = Detector(models=["yolo11s"])

# Or with explicit config:
#
# from pyzm import DetectorConfig, ModelConfig
# from pyzm.models.config import ModelFramework, ModelType, Processor
#
# detector = Detector(config=DetectorConfig(
#     models=[ModelConfig(
#         type=ModelType.OBJECT,
#         framework=ModelFramework.OPENCV,
#         processor=Processor.GPU,
#         weights="/path/to/yolo11s.onnx",
#         labels="/path/to/coco.names",
#         min_confidence=0.5,
#     )],
# ))

# Detect on a local image
result = detector.detect("/tmp/image.jpg")

if result.matched:
    print(f"Detections: {result.summary}")
    # e.g. "person:97% car:85%"

    for det in result.detections:
        print(f"  {det.label}: {det.confidence:.0%} at ({det.bbox.x1},{det.bbox.y1})-({det.bbox.x2},{det.bbox.y2})")

    # Draw boxes and save
    annotated = result.annotate()
    import cv2
    cv2.imwrite("/tmp/detected.jpg", annotated)
else:
    print("No detections")


# ============================================================================
# 3. ZM + ML TOGETHER (detect on a ZM event)
# ============================================================================

from pyzm import ZMClient, Detector, StreamConfig

zm = ZMClient(api_url="https://zm.example.com/zm/api", user="admin", password="secret")
detector = Detector(models=["yolo11s"])

event_id = 12345
m = zm.monitor(1)
zones = m.get_zones()

result = detector.detect_event(
    zm,
    event_id,
    zones=zones,
    stream_config=StreamConfig(
        frame_set=["snapshot", "alarm", "1"],
        resize=800,
    ),
)

if result.matched:
    print(result.summary)
    result.annotate()  # draw boxes on the image

    # Update ZM event notes
    ev = zm.event(event_id)
    ev.update_notes(result.summary)


# ============================================================================
# 4. AUDIO DETECTION (BirdNET)
# ============================================================================
# Requires: /opt/zoneminder/venv/bin/pip install birdnet-analyzer

from pyzm import Detector

# Audio-only config -- only audio models in the sequence
audio_opts = {
    "general": {"model_sequence": "audio"},
    "audio": {
        "general": {"pattern": ".*", "same_model_sequence_strategy": "first"},
        "sequence": [
            {
                "name": "BirdNET",
                "enabled": "yes",
                "audio_framework": "birdnet",
                "birdnet_min_conf": 0.5,
                "birdnet_lat": -1,       # -1 = no location filtering
                "birdnet_lon": -1,
                "birdnet_sensitivity": 1.0,
                "birdnet_overlap": 0.0,
            },
        ],
    },
}

detector = Detector.from_dict(audio_opts)

# detect_audio() works on any format ffmpeg can read (WAV, MP3, MP4, etc.)
result = detector.detect_audio("/tmp/recording.wav")

if result.matched:
    print(f"Birds: {result.summary}")
    for det in result.detections:
        print(f"  {det.label}: {det.confidence:.0%}")
else:
    print("No bird species detected")


# ============================================================================
# 5. LOAD FROM YAML CONFIG (ml_sequence dict)
# ============================================================================

# If you already have an ml_sequence dict from your YAML config:
ml_sequence = {
    "general": {
        "model_sequence": "object,face",
    },
    "object": {
        "general": {"pattern": "(person|car|dog)", "same_model_sequence_strategy": "first"},
        "sequence": [
            {
                "object_framework": "coral_edgetpu",
                "object_weights": "/path/to/model.tflite",
                "object_labels": "/path/to/labels.txt",
                "object_min_confidence": 0.3,
            },
            {
                "object_framework": "opencv",
                "object_weights": "/path/to/yolo11s.onnx",
                "object_labels": "/path/to/coco.names",
                "object_processor": "gpu",
                "object_min_confidence": 0.5,
            },
        ],
    },
    "face": {
        "general": {"same_model_sequence_strategy": "first"},
        "sequence": [
            {
                "face_detection_framework": "dlib",
                "known_images_path": "/var/lib/zmeventnotification/known_faces",
                "face_model": "cnn",
                "face_recog_dist_threshold": 0.6,
            },
        ],
    },
}

detector = Detector.from_dict(ml_sequence)
result = detector.detect("/tmp/image.jpg")


# ============================================================================
# 6. LOGGING
# ============================================================================

# Standalone (ML only, no ZoneMinder):
import logging
logging.basicConfig(level=logging.DEBUG)
# All pyzm internals log to the "pyzm" logger -- this is all you need.

# With ZoneMinder (reads zm.conf + DB Config table automatically):
from pyzm.log import setup_zm_logging
adapter = setup_zm_logging(name="myapp", override={"dump_console": True})
adapter.Info("Hello from pyzm")
adapter.Debug(3, "Detail")

Detecting a ZM event stream

#!/usr/bin/env python3
"""pyzm v2 -- detect objects in a ZoneMinder event stream.

Usage:
    python stream.py <event_id> [<monitor_id>]

See also:
    StreamConfig options, frame strategies, multi-model pipelines:
        https://pyzmv2.readthedocs.io/en/latest/guide/detection.html
    Quick-start guide:
        https://pyzmv2.readthedocs.io/en/latest/guide/quickstart.html
"""

import sys

import yaml

from pyzm import Detector, ZMClient, StreamConfig

if len(sys.argv) < 2:
    eid = input("Enter event ID to analyze: ")
    mid = input("Enter monitor ID (for zones): ")
else:
    eid = sys.argv[1]
    mid = sys.argv[2] if len(sys.argv) > 2 else input("Enter monitor ID: ")

# Read connection details from secrets
with open("/etc/zm/secrets.yml") as f:
    conf = yaml.safe_load(f) or {}
secrets = conf.get("secrets", {})

zm = ZMClient(
    api_url=secrets.get("ZM_API_PORTAL"),
    portal_url=secrets.get("ZM_PORTAL"),
    user=secrets.get("ZM_USER"),
    password=secrets.get("ZM_PASSWORD"),
)

stream_cfg = StreamConfig(
    frame_set=["snapshot", "alarm"],
    resize=800,
)

# Get zones for the monitor
m = zm.monitor(int(mid)) if mid else None
zones = m.get_zones() if m else None

# Run detection
detector = Detector(models=["yolo11s"])
result = detector.detect_event(zm, int(eid), zones=zones, stream_config=stream_cfg)

if result.matched:
    print(f"FRAME: {result.frame_id}")
    print(f"SUMMARY: {result.summary}")
    for det in result.detections:
        print(f"  {det.label}: {det.confidence:.0%}")
else:
    print("No detections")

Detecting a local image

#!/usr/bin/env python3
"""pyzm v2 -- detect objects in a local image file.

Usage:
    python image.py <image_path>

See also:
    Multi-model configs, zones, from_dict():
        https://pyzmv2.readthedocs.io/en/latest/guide/detection.html
    Quick-start guide:
        https://pyzmv2.readthedocs.io/en/latest/guide/quickstart.html
"""

import sys

from pyzm import Detector

if len(sys.argv) < 2:
    image_path = input("Enter filename to analyze: ")
else:
    image_path = sys.argv[1]

# Model names are resolved against base_path on disk
detector = Detector(models=["yolo11s"])
result = detector.detect(image_path)

if result.matched:
    print(f"SUMMARY: {result.summary}")
    for det in result.detections:
        print(f"  {det.label}: {det.confidence:.0%} at ({det.bbox.x1},{det.bbox.y1})-({det.bbox.x2},{det.bbox.y2})")
else:
    print("No detections")

Audio detection (BirdNET)

#!/usr/bin/env python3
"""pyzm v2 -- detect bird species in an audio file using BirdNET.

Prerequisites:
    /opt/zoneminder/venv/bin/pip install birdnet-analyzer

Usage:
    python audio.py <audio_file>
    python audio.py /path/to/recording.wav
    python audio.py /path/to/event.mp4

Any format ffmpeg can read (WAV, MP3, MP4, etc.) is supported.

See also:
    Detection options, multi-model pipelines:
        https://pyzmv2.readthedocs.io/en/latest/guide/detection.html
"""

import sys

from pyzm import Detector

if len(sys.argv) < 2:
    audio_path = input("Enter audio file path: ")
else:
    audio_path = sys.argv[1]

# BirdNET via the ml_sequence dict format (same as objectconfig.yml)
ml_options = {
    "general": {
        "model_sequence": "audio",
    },
    "audio": {
        "general": {
            "pattern": ".*",
            "same_model_sequence_strategy": "first",
        },
        "sequence": [
            {
                "name": "BirdNET",
                "enabled": "yes",
                "audio_framework": "birdnet",
                "birdnet_min_conf": 0.5,
                # Set lat/lon for seasonal species filtering (or -1 to disable)
                "birdnet_lat": -1,
                "birdnet_lon": -1,
                "birdnet_sensitivity": 1.0,
                "birdnet_overlap": 0.0,
            },
        ],
    },
}

detector = Detector.from_dict(ml_options)

# detect_audio() runs BirdNET on the audio file directly
result = detector.detect_audio(audio_path)

if result.matched:
    print(f"Species detected: {result.summary}")
    for det in result.detections:
        print(f"  {det.label}: {det.confidence:.0%}")
else:
    print("No bird species detected")

Remote detection via pyzm.serve

#!/usr/bin/env python3
"""pyzm v2 -- detect objects via a remote pyzm.serve server.

Two modes are supported:

  URL mode (default):
    Client sends ZM frame URLs to the server, which fetches
    images directly from ZoneMinder.  Useful when the GPU box
    has direct network access to ZM.

  Image mode:
    Client JPEG-encodes the image and uploads it to the server.
    Use when the server can't reach ZM directly.

Prerequisites:
    Start the server on the GPU box (no auth):
        python -m pyzm.serve --models yolo11s --port 5000

    With auth:
        python -m pyzm.serve --models yolo11s --port 5000 --auth --auth-password secret

    For YAML config, see examples/objectconfig.yml or the serve guide.

Usage:
    python remote.py <image_path> [gateway_url]

See also:
    Server setup, authentication, YAML config, API reference:
        https://pyzmv2.readthedocs.io/en/latest/guide/serve.html
    Quick-start guide:
        https://pyzmv2.readthedocs.io/en/latest/guide/quickstart.html
"""

import sys

from pyzm import Detector

gateway = "http://localhost:5000"

if len(sys.argv) < 2:
    image_path = input("Enter filename to analyze: ")
else:
    image_path = sys.argv[1]

if len(sys.argv) >= 3:
    gateway = sys.argv[2]

# URL mode is the default -- detect_event() sends frame URLs and the
# server fetches them directly from ZoneMinder.
# Use gateway_mode="image" if the server can't reach ZM directly.
detector = Detector(models=["yolo11s"], gateway=gateway)
# detector = Detector(models=["yolo11s"], gateway=gateway, gateway_mode="image")

# With authentication:
# detector = Detector(
#     models=["yolo11s"],
#     gateway=gateway,
#     gateway_username="admin",
#     gateway_password="secret",
# )

# detect() always uploads the image (single-image mode)
result = detector.detect(image_path)

print(f"SUMMARY: {result.summary}")
for det in result.detections:
    print(f"  {det.label}: {det.confidence:.0%}")

# detect_event() uses URL mode by default -- sends frame URLs,
# server fetches them directly from ZM:
#
# from pyzm import ZMClient, StreamConfig
#
# zm = ZMClient(api_url="https://zm.example.com/zm/api",
#               user="admin", password="secret")
#
# result = detector.detect_event(
#     zm, event_id=12345,
#     stream_config=StreamConfig(frame_set=["snapshot", "alarm"]),
# )
# print(result.summary)

Testing a fine-tuned model

#!/usr/bin/env python3
"""Quick-test a trained pyzm model on an image with bounding-box output.

Usage:
    python test_model.py <image> <onnx_weights> [--labels classes.txt] [--confidence 0.3] [--out output.jpg]

Example after `pyzm train`:
    python test_model.py photo.jpg ~/.pyzm/training/my_project/best.onnx
"""

import argparse
import sys
from pathlib import Path

import cv2

from pyzm import Detector
from pyzm.models.config import DetectorConfig, ModelConfig, ModelFramework, Processor


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Test a trained model on an image and save annotated output."
    )
    parser.add_argument("image", help="Path to the input image")
    parser.add_argument("weights", help="Path to the ONNX weights file")
    parser.add_argument(
        "--labels", default=None,
        help="Path to labels/classes text file (optional — extracted from ONNX metadata if omitted)",
    )
    parser.add_argument(
        "--confidence", type=float, default=0.3,
        help="Minimum confidence threshold (default: 0.3)",
    )
    parser.add_argument(
        "--out", default=None,
        help="Output image path (default: <image>_detections.jpg)",
    )
    args = parser.parse_args()

    image_path = Path(args.image)
    if not image_path.exists():
        sys.exit(f"Image not found: {image_path}")

    weights_path = Path(args.weights)
    if not weights_path.exists():
        sys.exit(f"Weights not found: {weights_path}")

    if args.labels and not Path(args.labels).exists():
        sys.exit(f"Labels file not found: {args.labels}")

    out_path = Path(args.out) if args.out else image_path.with_name(
        f"{image_path.stem}_detections{image_path.suffix}"
    )

    # Build detector with the trained model
    detector = Detector(config=DetectorConfig(models=[
        ModelConfig(
            name=weights_path.stem,
            framework=ModelFramework.OPENCV,
            processor=Processor.CPU,
            weights=str(weights_path),
            labels=args.labels,
            min_confidence=args.confidence,
        ),
    ]))

    result = detector.detect(str(image_path))

    if not result.matched:
        print("No detections found.")
        return

    print(f"Detections: {result.summary}")
    for det in result.detections:
        print(f"  {det.label}: {det.confidence:.0%}  "
              f"[{det.bbox.x1},{det.bbox.y1} -> {det.bbox.x2},{det.bbox.y2}]")

    annotated = result.annotate()
    cv2.imwrite(str(out_path), annotated)
    print(f"\nSaved annotated image to: {out_path}")


if __name__ == "__main__":
    main()