Skip to main content

ViewtronServer

Overview​

ViewtronServer is an HTTP server that listens for webhook events from Viewtron IP cameras and NVRs. When a camera detects something — a license plate, face, intrusion, line crossing, etc. — it sends an HTTP POST with XML data to your server. ViewtronServer parses the XML and calls your callback with a ViewtronEvent object.

The server handles all Viewtron camera connection requirements automatically: HTTP/1.1 persistent connections, keepalive responses, XML success replies, and multi-threaded request handling.

from viewtron import ViewtronServer

def handle_event(event, client_ip):
    print(f"[{event.category}] {event.get_alarm_description()} from {client_ip}")

server = ViewtronServer(port=5050, on_event=handle_event)
server.serve_forever()

Constructor​

ViewtronServer(port=5050, on_event=None, on_connect=None, on_raw=None)
ParameterTypeDefaultDescription
portint5050TCP port to listen on
on_eventcallableNoneCalled for each parsed event: on_event(event, client_ip)
on_connectcallableNoneCalled when a camera first connects: on_connect(client_ip)
on_rawcallableNoneCalled with raw XML before parsing: on_raw(xml_text, client_ip)

Methods​

serve_forever()​

Start the server and block until interrupted (Ctrl+C) or shutdown() is called. Prints the server's LAN IP and port on startup.

server = ViewtronServer(port=5050, on_event=handle_event)
server.serve_forever()
# Output:
# Viewtron Event Server
# Listening on http://192.168.1.100:5050
# Ready for camera events...

shutdown()​

Stop the server from another thread. Call this when you need to stop the server programmatically rather than with Ctrl+C.

import threading

server = ViewtronServer(port=5050, on_event=handle_event)
thread = threading.Thread(target=server.serve_forever)
thread.start()

# Later...
server.shutdown()

Callbacks​

on_event(event, client_ip)​

The primary callback. Called once for each parsed event.

  • event — a ViewtronEvent object (LPR, FaceDetection, IntrusionDetection, Traject, etc.)
  • client_ip — str, the IP address of the camera or NVR that sent the event

Called for all detection events including:

  • LPR (license plate recognition)
  • Face detection
  • Intrusion / line crossing / region entry / region exit
  • Loitering, illegal parking
  • Target counting (by line or area)
  • Video metadata
  • Traject (smart tracking data)

Not called for:

  • Keepalive heartbeats (cameras send these every ~30 seconds)
  • Alarm status on/off messages

Note on traject events: Traject data is high-volume — cameras send multiple events per second for each tracked target. If you don't need tracking data, filter it out early:

def handle_event(event, client_ip):
    if event.category == "traject":
        return  # Skip high-volume tracking data

    print(f"[{event.category}] {event.get_alarm_description()} from {client_ip}")

    if event.category == "lpr":
        print(f"  Plate: {event.get_plate_number()}")
        print(f"  Group: {event.get_plate_group()}")

on_connect(client_ip)​

Optional. Called the first time a camera sends a keepalive (empty-body POST) from a new IP address. Useful for logging which cameras are connected.

Each camera IP is tracked — the callback fires once per IP, not on every keepalive.

def handle_connect(client_ip):
    print(f"Camera connected: {client_ip}")

server = ViewtronServer(
    port=5050,
    on_event=handle_event,
    on_connect=handle_connect
)

on_raw(xml_text, client_ip)​

Optional. Called with the raw XML string before parsing. Useful for debugging or logging raw camera data.

  • xml_text — str, the raw XML body from the HTTP POST
  • client_ip — str, the IP address of the camera or NVR

Note: on_raw is not called for traject data. Traject posts are high-volume and would flood raw logging. Traject events are still delivered to on_event as parsed Traject objects.

def handle_raw(xml_text, client_ip):
    with open(f"raw_{client_ip}.xml", "a") as f:
        f.write(xml_text + "\n---\n")

server = ViewtronServer(
    port=5050,
    on_event=handle_event,
    on_raw=handle_raw
)

Threading Model​

ViewtronServer uses Python's ThreadingMixIn with HTTPServer — each incoming HTTP request is handled in a separate thread. All threads are daemon threads, so they won't prevent your program from exiting.

Your callbacks (on_event, on_connect, on_raw) run in request-handling threads, not the main thread. If your callbacks modify shared state, use proper synchronization:

import threading
from collections import defaultdict

lock = threading.Lock()
plate_counts = defaultdict(int)

def handle_event(event, client_ip):
    if event.category == "lpr":
        plate = event.get_plate_number()
        with lock:
            plate_counts[plate] += 1

Camera Connection Requirements​

ViewtronServer handles these automatically, but they're useful to understand when troubleshooting:

  • HTTP/1.1 persistent connections — cameras use Connection: keep-alive and maintain a single long-lived TCP connection. The server sets protocol_version = "HTTP/1.1" to support this.
  • XML success response — every POST receives a \<status>success\</status> XML reply. This tells the camera the connection is alive.
  • Keepalive heartbeats — cameras send an empty POST every ~30 seconds to confirm the connection. The server handles these silently.
  • Reboot after config changes — after changing HTTP Post settings on a camera or NVR, reboot the device for changes to take effect. The camera caches connection settings and won't reconnect properly without a reboot.

For instructions on configuring cameras to send events, see the HTTP Post Setup guide.

Common Patterns​

Running in a Background Thread​

Run the server without blocking your main program:

import threading
from viewtron import ViewtronServer

def handle_event(event, client_ip):
    print(f"[{event.category}] {event.get_alarm_description()}")

server = ViewtronServer(port=5050, on_event=handle_event)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()

# Your main program continues here...

Routing by Event Category​

Dispatch events to separate handler functions based on their category:

def handle_lpr(event, client_ip):
    plate = event.get_plate_number()
    group = event.get_plate_group()
    print(f"Plate: {plate} ({group}) from {client_ip}")

def handle_intrusion(event, client_ip):
    print(f"Intrusion: {event.get_alarm_type()} from {client_ip}")

def handle_face(event, client_ip):
    print(f"Face detected from {client_ip}")

def handle_event(event, client_ip):
    handlers = {
        "lpr": handle_lpr,
        "intrusion": handle_intrusion,
        "face": handle_face,
    }
    handler = handlers.get(event.category)
    if handler:
        handler(event, client_ip)

Saving Images to Disk​

Most detection events include snapshot images. Save them from within your callback:

import os

def handle_event(event, client_ip):
    if event.category == "lpr":
        plate = event.get_plate_number()

        source = event.get_source_image_bytes()
        if source:
            os.makedirs("captures", exist_ok=True)
            with open(f"captures/{plate}_full.jpg", "wb") as f:
                f.write(source)

        target = event.get_target_image_bytes()
        if target:
            with open(f"captures/{plate}_crop.jpg", "wb") as f:
                f.write(target)