Code: Select all
from ultralytics import YOLO
import cv2
import requests
import time
import json
import os
from datetime import datetime, timedelta
# ZoneMinder API details
ZM_URL = "http://172.16.32.139/zm/"
ZM_MONITOR_ID = 1 # Monitor ID
YOLO_MODEL_PATH = "yolo11n.pt" # Path to YOLO model
model = YOLO(YOLO_MODEL_PATH)
def get_last_event_id():
"""Get the last event ID by checking all events."""
events_url = f"{ZM_URL}api/events.json"
try:
response = requests.get(events_url)
if response.status_code != 200:
print(f"Error: Status code {response.status_code}")
return None
events_data = response.json()
if "events" not in events_data:
print("Invalid JSON structure. 'events' key not found.")
return None
events = events_data["events"]
if not events:
print("No events found.")
return None
try:
max_id_event = max(events, key=lambda x: x["Event"]["Id"])
last_event_id = max_id_event["Event"]["Id"]
print(f"Last event ID: {last_event_id}")
return last_event_id
except KeyError as e:
print(f"Invalid event structure. Key {e} not found.")
return None
except requests.exceptions.RequestException as e:
print(f"Request error: {e}")
return None
def create_zm_event():
"""Create a new event in ZoneMinder and return its ID."""
try:
start_time = datetime.now()
end_time = start_time + timedelta(seconds=3)
start_time_str = start_time.strftime("%Y-%m-%d %H:%M:%S")
end_time_str = end_time.strftime("%Y-%m-%d %H:%M:%S")
event_url = f"{ZM_URL}api/events.json"
event_data = {
"MonitorId": ZM_MONITOR_ID,
"Cause": "Human detected",
"StateId": 1,
"StartDateTime": start_time_str,
"EndDateTime": end_time_str,
"Frames": 3,
"Width": 640,
"Height": 480
}
response = requests.post(event_url, data=event_data)
print(f"Status Code: {response.status_code}")
print(f"Response Text: {response.text}")
if response.status_code in (200, 201):
try:
response_json = response.json()
if "event" in response_json and "Id" in response_json["event"]:
event_id = response_json["event"]["Id"]
print(f"Event created! ID: {event_id}")
return event_id
else:
print("Invalid JSON structure. Missing 'event' or 'Id'.")
return None
except json.JSONDecodeError:
print("Invalid JSON response.")
return None
else:
print(f"Error: API returned status code {response.status_code}")
return None
except requests.exceptions.RequestException as e:
print(f"Request failed: {e}")
return None
def trigger_zm_event(event_id, frame):
"""Register the event in ZoneMinder."""
event_folder_path = f"/var/cache/zoneminder/events/{ZM_MONITOR_ID}/{time.strftime('%Y-%m-%d')}/{event_id}"
os.makedirs(event_folder_path, exist_ok=True)
create_zm_event()
filename1 = f"{event_folder_path}/00001-capture.jpg"
filename2 = f"{event_folder_path}/00002-capture.jpg"
filename3 = f"{event_folder_path}/00003-capture.jpg"
filename4 = f"{event_folder_path}/snapshot.jpg"
filename5 = f"{event_folder_path}/alarm.jpg"
cv2.imwrite(filename1, frame)
cv2.imwrite(filename2, frame)
cv2.imwrite(filename3, frame)
cv2.imwrite(filename4, frame)
cv2.imwrite(filename5, frame)
print(f"Frame saved to {filename1}")
def detect_human(frame):
"""Detect humans in the frame using YOLOv8."""
results = model(frame)
for result in results:
for det in result.boxes:
if model.names[int(det.cls)] == "person": # Check for human class
return True
return False
def main():
cap = cv2.VideoCapture(f"{ZM_URL}cgi-bin/nph-zms?mode=jpeg&monitor={ZM_MONITOR_ID}&scale=100&maxfps=10")
last_event_id = 93
while True:
ret, frame = cap.read()
if not ret:
print("Failed to capture frame from ZoneMinder stream.")
break
if detect_human(frame):
last_event_id += 1
print("Human detected! Registering event in ZoneMinder...")
trigger_zm_event(last_event_id, frame)
time.sleep(10) # Cooldown to avoid multiple events
cap.release()
if __name__ == "__main__":
main()