Methods

Note: Examples in this guide assume you have already configured LicensePlateTrackerConfig with model specifications. See Configuration Guide for complete setup details.

Methods Overview

Method	Purpose	Use Case
start_tracking_pipeline()	Real-time monitoring with automated alerting	Traffic monitoring, parking enforcement, security surveillance
predict_batch()	Stream processing with programmatic access	Custom analytics, logging, integration with other systems
find_plates_in_file()	Analyze local video files	Post-incident review, batch processing of footage
find_plates_in_clip()	Analyze cloud storage clips	Review alert clips, batch processing from S3

---

start_tracking_pipeline()

Run continuous real-time video monitoring with automated alerting, clip recording, and live streaming.

When to use: This is the primary method for production deployments where you need a fully automated system that runs continuously, handles alerts, saves video clips, and sends notifications. While you can attach a custom sink to access results programmatically, the main value of this method is the automated handling—use predict_batch() if programmatic result processing is your primary goal.

Signature

start_tracking_pipeline(
    frame_iterator: Optional[Iterable] = None,
    sink: Optional[SinkGizmo] = None
) -> Tuple[Composition, Watchdog]

Parameters

• frame_iterator (Optional) - Custom frame source; if None, uses config.video_source

• sink (Optional) - Custom output sink for results

Returns

• Composition - Pipeline composition object (call .wait() to run). See Composition Documentation for details

• Watchdog - Monitoring object for pipeline health. See Watchdog Documentation for details

Example: Traffic Monitoring

Monitor RTSP camera, save clips, and send notifications:

import degirum_vehicle
import degirum_tools

# Configure tracking with alerts and clip recording
config = degirum_vehicle.LicensePlateTrackerConfig(
    # ... model specs (see configuration guide) ...
    video_source="rtsp://192.168.1.100/stream",
    alert_mode=degirum_vehicle.AlertMode.ON_ALL,
    clip_duration=150,
    clip_storage_config=degirum_tools.ObjectStorageConfig(
        endpoint="./traffic_clips",
        bucket="plates"
    ),
    notification_config="mailto://[email protected]",
    notification_message="🚗 License plate detected: ${text} at ${time}",
    live_stream_mode="WEB",
    live_stream_rtsp_url="gate_camera",
)

tracker = degirum_vehicle.LicensePlateTracker(config)
composition, watchdog = tracker.start_tracking_pipeline()
composition.wait()  # Run until Ctrl+C

How the Pipeline Works

1. Frame Acquisition - Read frame from video source

2. License Plate Detection - Detect license plates with bounding boxes

3. Object Tracking - Assign persistent track IDs to plates

4. Plate Filtering - Apply quality filters (zone filtering, size thresholds)

5. Plate Extraction - Crop license plate regions

6. OCR Extraction - Extract text from license plates

7. Bayesian Text Aggregation - Fuse OCR results across frames for improved accuracy

8. Alert Evaluation - Check credence count and alert mode

9. Clip Recording - Save video clips when alerts trigger

10. Notification - Send alerts via configured channels

11. Live Streaming - Output annotated video to display/RTSP

---

predict_batch()

Process video streams and return inference results for each frame with programmatic access.

When to use: Use this when you need fine-grained control over the processing pipeline. Unlike start_tracking_pipeline(), this method gives you access to results for every frame, allowing you to implement custom logic, logging, integration with other systems, or build your own alerting mechanisms.

Signature

predict_batch(stream: Iterable) -> Iterator[InferenceResults]

Parameters

• stream - Iterator yielding video frames as numpy arrays

Returns

• Iterator of InferenceResults objects. InferenceResults objects support standard PySDK methods like image_overlay(), results, etc. See InferenceResults documentation

Each result contains:

• results - List of detection dictionaries with tracking data (access via result.results[i].get("track_id"))

• license_plates property - List of LPRResult objects. See LPRResult Reference

Note: Tracking data (track_id, frame_id) is in the results list, not as properties of LPRResult. To correlate, use the same index for both lists.

Example: Custom Logging

Process video and log all license plate detections:

import cv2
import degirum_vehicle

def frame_generator(video_path):
    cap = cv2.VideoCapture(video_path)
    while True:
        ret, frame = cap.read()
        if not ret:
            break
        yield frame
    cap.release()

# Create tracker
config = degirum_vehicle.LicensePlateTrackerConfig.from_hardware(
    "TFLITE/CPU",
    inference_host_address="@cloud"
)
tracker = degirum_vehicle.LicensePlateTracker(config)

# Process video and access results programmatically
for result in tracker.predict_batch(frame_generator("parking_lot.mp4")):
    for i, plate in enumerate(result.license_plates):
        track_id = result.results[i].get("track_id")
        print(f"Track {track_id}: {plate.plate_number} (OCR: {plate.ocr_score:.2f})")

---

find_plates_in_file()

Analyze local video files to find and track all license plates, optionally creating annotated output.

When to use: This is the go-to method for offline analysis of recorded video. Use it for post-incident review, batch processing local video files, or analyzing archived footage. The method automatically handles tracking, applies Bayesian text aggregation for improved OCR accuracy, and produces annotated videos for easy review.

Signature

find_plates_in_file(
    file_path: str,
    save_annotated: bool = True,
    output_video_path: Optional[str] = None
) -> Dict[int, PlateOCRResult]

Parameters

• file_path (str) - Path to input video file

• save_annotated (bool) - Whether to save annotated video (default: True)

• output_video_path (Optional[str]) - Path for annotated output video; if None, no video is saved

Returns

• Dict[int, PlateOCRResult] - Dictionary mapping track IDs to plate OCR data:

  • Each PlateOCRResult object contains the aggregated plate number and OCR score

Example: Analyze Video File

import degirum_vehicle

# Create tracker
config = degirum_vehicle.LicensePlateTrackerConfig.from_hardware(
    "TFLITE/CPU",
    inference_host_address="@cloud"
)
tracker = degirum_vehicle.LicensePlateTracker(config)

# Analyze video file
plates = tracker.find_plates_in_file(
    "parking_lot.mp4",
    save_annotated=True,
    output_video_path="parking_lot_annotated.mp4"
)

# Access detected plates by track ID
for track_id, plate in plates.items():
    print(f"Track {track_id}: {plate.plate_number} (OCR: {plate.ocr_score:.2f})")

---

find_plates_in_clip()

Analyze video clips from object storage (S3 or local), similar to find_plates_in_file() but for cloud/storage-based clips.

When to use: Use this to review video clips saved by start_tracking_pipeline() or stored in S3/object storage. It's particularly useful for reviewing alert clips generated by the automated pipeline, allowing you to verify detections and analyze incidents from cloud-stored footage.

Signature

find_plates_in_clip(
    clip_object_name: str,
    save_annotated: bool = True
) -> Dict[int, PlateOCRResult]

Parameters

• clip_object_name (str) - Name of video clip in object storage

• save_annotated (bool) - Whether to save annotated video back to storage (default: True)

Returns

• Dict[int, PlateOCRResult] - Dictionary mapping track IDs to plate OCR data (same as find_plates_in_file())

Requirements

• clip_storage_config must be configured in LicensePlateTrackerConfig

Example: Review Alert Clips

Process license plate clips from cloud storage:

import degirum_vehicle

# Assumes tracker is configured with clip_storage_config

# List all clips
clip_manager = degirum_vehicle.LPRClipManager(tracker.config.clip_storage_config)
clips = clip_manager.list_clips()

print(f"Found {len(clips)} clips to review")

# Analyze each clip
for clip_name in clips.keys():
    print(f"\nAnalyzing: {clip_name}")
    
    plates = tracker.find_plates_in_clip(
        clip_object_name=clip_name,
        save_annotated=True
    )
    
    print(f"  Found {len(plates)} unique vehicles")
    
    for track_id, plate in plates.items():
        print(f"    Track {track_id}: {plate.plate_number} (OCR: {plate.ocr_score:.2f})")

Note: Annotated videos are saved with _annotated suffix in the same storage location.

Complete Example

import degirum_vehicle

# Create configuration
config = degirum_vehicle.LicensePlateTrackerConfig()
config.video_source = "examples/assets/Highway_720p.mp4"
config.live_stream_mode = "LOCAL"
config.save_video_path = "output_annotated.mp4"
config.show_ocr_score = True
config.ocr_score_threshold = 0.6

# Create tracker
tracker = degirum_vehicle.LicensePlateTracker(config)

# Start pipeline
composition, watchdog = tracker.start_tracking_pipeline()

# Wait for completion
try:
    composition.wait()
except KeyboardInterrupt:
    composition.stop()
    
print(f"Saved annotated video to: {config.save_video_path}")