Example Program(s)¶
Simple event extraction example¶
import pyzm
import pyzm.api as zmapi
import getpass
import traceback
import pyzm.ZMMemory as zmmemory
import pyzm.helpers.utils as utils
import pyzm.helpers.globals as g
print ('Using pyzm version: {}'.format(pyzm.__version__))
g.logger.set_level(2)
conf = utils.read_config('/etc/zm/secrets.ini')
api_options = {
'apiurl': utils.get(key='ZM_API_PORTAL', section='secrets', conf=conf),
'portalurl':utils.get(key='ZM_PORTAL', section='secrets', conf=conf),
'user': utils.get(key='ZM_USER', section='secrets', conf=conf),
#'disable_ssl_cert_check': True
}
zmapi = zmapi.ZMApi(options=api_options)
event_filter = {
'from': '9 am',
'to': '7 pm',
'object_only':False,
'min_alarmed_frames': 0,
'max_events':5,
}
cam_events = zmapi.events(event_filter)
print ('I got {} events'.format(len(cam_events.list())))
for e in cam_events.list():
print ('Event:{} Cause:{} Notes:{}'.format(e.name(), e.cause(), e.notes()))
#cam_events.list()[0].download_image()
Various examples¶
import pyzm
import pyzm.api as zmapi
import getpass
import traceback
import pyzm.ZMMemory as zmmemory
import time
import pyzm.helpers.globals as g
use_zmlog = True
use_zmes = True
has_zmes = False
has_zmlog = False
print ('Using pyzm version: {}'.format(pyzm.__version__))
if use_zmlog:
try:
import pyzm.ZMLog as zmlog #only if you want to log to ZM
has_zmlog = True
except ImportError as e:
print ('Could not import ZMLog, function will be disabled:'+str(e))
zmlog = None
if use_zmes:
try:
from pyzm.ZMEventNotification import ZMEventNotification as ZMES
has_zmes = True
except ImportError as e:
print ('Could not import ZMEventNotification, function will be disabled:'+str(e))
def on_es_message(msg):
print ('======> APP GOT MESSAGE FROM ES: {}'.format(msg))
def on_es_error(err):
print ('======> APP GOT ERROR FROM ES: {}'.format(err))
# ----------------- MAIN -------------------------
# Assuming you want to log to ZM
# You can override default ZM Log settings
# programatically
zm_log_override = {
'log_level_syslog' : 3,
'log_level_db': -5,
'log_debug': 1,
#'log_level_file': -5,
'log_debug_target': None
}
if has_zmlog:
zmlog.init(name='apitest',override=zm_log_override)
print ("Log inited")
i = input ('Try machine learning tests? [y/N]').lower()
if i == 'y':
import cv2
fname = input ('Enter full path to image file:')
if fname:
print ('Reading {}'.format(fname))
img = cv2.imread(fname)
else:
print ('Reading /tmp/image.jpg')
img = cv2.imread('/tmp/image.jpg')
i = input ('Try TPU tests? [y/N]').lower()
if i == 'y':
options = {
'object_weights':'/var/lib/zmeventnotification/models/coral_edgetpu/ssd_mobilenet_v2_coco_quant_postprocess_edgetpu.tflite',
'object_labels': '/var/lib/zmeventnotification/models/coral_edgetpu/coco_indexed.names',
'object_min_confidence': 0.3
}
import pyzm.ml.coral_edgetpu as tpu
m = tpu.Tpu(options=options)
b,l,c = m.detect(img)
print (b,l,c)
i = input ('Try OpenCV tests? [y/N]').lower()
if i == 'y':
options = {
'object_weights':'/var/lib/zmeventnotification/models/yolov4/yolov4.weights',
'object_labels': '/var/lib/zmeventnotification/models/yolov4/coco.names',
'object_config': '/var/lib/zmeventnotification/models/yolov4/yolov4.cfg',
'object_processor': 'cpu',
'object_min_confidence': 0.3
}
import pyzm.ml.yolo as yolo
m = yolo.Yolo(options=options)
b,l,c = m.detect(img)
print (b,l,c)
i = input ('Try Face recognition tests? [y/N]').lower()
if i == 'y':
options = {
'known_images_path': '/var/lib/zmeventnotification/known_faces',
'face_recog_dist_threshold':0.6,
'unknown_face_name':'klingon',
'save_unknown_faces':'no',
'save_unknown_faces_leeway_pixels':100,
'unknown_images_path':'/var/lib/zmeventnotification/unknown_faces',
'face_detection_framework': 'dlib',
'face_recognition_framework': 'dlib',
}
import pyzm.ml.face as face
m = face.Face(options=options)
b,l,c = m.detect(img)
print (b,l,c)
cam_name='DemoVirtualCam1'
api_options = {
'apiurl': 'https://demo.zoneminder.com/zm/api',
'portalurl': 'https://demo.zoneminder.com/zm',
'user': 'zmuser',
'password': 'zmpass',
#'disable_ssl_cert_check': True
}
print ('Running examples on {}'.format(api_options[
'apiurl'
]))
i = input ('Try monitor shared memory tests? [y/N]').lower()
if i == 'y':
mid = int(input ('Enter monitor ID:'))
while True:
k = 'y'
try:
m = zmmemory.ZMMemory(mid=mid)
break
except Exception as e:
print ('Error initing: {}'.format(e))
k = input ('try again, or \'q\' to quit...')
if k == 'q': break
while True and k != 'q':
if m.is_valid():
print (m.get())
else:
print ('Memory not valid')
try:
m.reload()
except Exception as e:
print ('Error reloading: {}'.format(e))
k = input ('Try to read again [\'q\' to quit this test]')
if has_zmes:
i = input ('Test the Event Server? [y/N]').lower()
if i=='y':
ES_URL=None
ES_USER=None
ES_PASSWORD=None
ALLOW_UNTRUSTED=True
if not ES_URL: ES_URL = input ('Enter ES URL (example wss://foo:9000):')
if not ES_USER: ES_USER = input ('Enter ES user (example admin):')
if not ES_PASSWORD: ES_PASSWORD = getpass.getpass('Enter ES password:')
es = ZMES({
'url':ES_URL,
'password': ES_PASSWORD,
'user': ES_USER,
'allow_untrusted': ALLOW_UNTRUSTED,
'on_es_message': on_es_message,
'on_es_error': on_es_error
})
# send a legit command
print ("Sending a valid login")
es.send({"event":"control","data":{"type":"filter","monlist":"1,2,5,6,7,8,9,10", "intlist":"0,0,0,0,0,0,0,0"}})
print ("Sleeping for 3 seconds...")
time.sleep(3)
# send a bad command
print ("Sending an invalid command")
es.send ('Hi From ES')
input ('press a key to proceed with the rest...')
# lets init the API
try:
zmapi = zmapi.ZMApi(options=api_options)
except Exception as e:
print ('Error: {}'.format(str(e)))
print(traceback.format_exc())
exit(1)
# Various getter tests
print ("--------| Getting Monitors |-----------")
ms = zmapi.monitors()
for m in ms.list():
print ('Name:{} Enabled:{} Type:{} Dims:{}'.format(m.name(), m.enabled(), m.type(), m.dimensions()))
print (m.status())
print ("--------| Getting Events |-----------")
print ("Getting events across all monitors")
event_filter = {
'from': '24 hours ago', # this will use localtimezone, use 'tz' for other timezones
'object_only':False,
'min_alarmed_frames': 1,
'max_events':5,
}
es = zmapi.events(event_filter)
print ('I got {} events'.format(len(es.list())))
input ("Now revoke the access token in ZM and I'll try the same api again. Press ENTER when ready....(only applies to ZM 1.33+)")
es = zmapi.events(event_filter)
print ('repeat API - I got {} events'.format(len(es.list())))
input ('press ENTER to continue')
print ('Getting events for {} with filter: {}'.format(cam_name, event_filter))
cam_events = ms.find(name=cam_name).events(options=event_filter)
for e in cam_events.list():
print ('Event:{} Cause:{} Notes:{}'.format(e.name(), e.cause(), e.notes()))
print ('Now trying to download an image from the first event')
print(cam_events.list())
if cam_events.list():
e = cam_events.list()[0]
print (e.name())
e.download_image(dir='/tmp')
e.download_video(dir='/tmp')
else:
print ('No events found')
print ('Getting event summaries')
m = ms.find(name=cam_name)
# These will use server timezone
print ('Monitor {} has {} events {}'.format(m.name(), m.eventcount(options={'from':'1 hour ago','tz': zmapi.tz()}), '1 hour ago'))
print ('Monitor {} has {} events {}'.format(m.name(), m.eventcount(options={'from':'1 day ago','tz': zmapi.tz()}), '1 day ago'))
print ("--------| Getting ZM States |-----------")
states = zmapi.states()
for state in states.list():
print ('State:{}[{}], active={}, details={}'.format(state.name(), state.id(), state.active(), state.definition()))
i = input ('Test Monitor State Change? [y/N]').lower()
if i=='y':
print ("--------| Setting Monitors |-----------")
m = ms.find(name=cam_name)
try:
old_function = m.function()
input ('Going to change state of {}[{}] to Monitor from {}'.format(m.name(),m.id(), old_function))
print (m.set_parameter(options={'function':'Monitor'}))
input ('Switching back to {}'.format(old_function))
print (m.set_parameter(options={'function':old_function}))
except Exception as e:
print ('Error: {}'.format(str(e)))
print ("--------| Setting Alarms |-----------")
try:
input ('Arming {}, press enter'.format(m.name()))
print (m.arm())
input ('Disarming {}, press enter'.format(m.name()))
print (m.disarm())
except Exception as e:
print ('Error: {}'.format(str(e)))
i = input ('Test ZM State Changes? [y/N]').lower()
if i=='y':
print ("--------| Setting States |-----------")
try:
input ('Stopping ZM press enter')
print (zmapi.stop())
input ('Starting ZM press enter')
print (zmapi.start())
for idx,state in enumerate(states.list()):
print ('{}:{}'.format(idx,state.name()))
i=int(input('enter state number to switch to:'))
name = states.list()[i].name()
print ('Changing state to: {}'.format(name))
print (zmapi.set_state(state=name))
except Exception as e:
print ('Error: {}'.format(str(e)))
print ("--------| Configs Test |-----------")
try:
conf = zmapi.configs()
print (conf.find(name='ZM_AUTH_HASH_LOGINS'))
except Exception as e:
print ('Error: {}'.format(str(e)))
zmlog.close()
Detecting a video stream¶
from pyzm import __version__ as pyzmversion
import pyzm.api as zmapi
import getpass
import traceback
import pyzm.ZMMemory as zmmemory
import time
#import pyzm.ml.object as ObjectDetect
from pyzm.ml.detect_sequence import DetectSequence
import pyzm.helpers.utils as utils
import sys
import pyzm.helpers.globals as g
import pyzm.ZMLog as log
print ('Using pyzm version: {}'.format(pyzmversion))
#log.init(name='stream', override={'dump_console': True})
g.logger.set_level(5)
#time.sleep(1000)
mid = None
if len(sys.argv) == 1:
eid = input ('Enter event ID to analyze:')
mid = input ('Enter MID to use:')
else:
eid = sys.argv[1]
if len(sys.argv) == 2:
print ('Event to analyze:{}'.format(eid))
mid = input ('Enter MID to use:')
else:
mid = sys.argv[2]
'''
api_options = {
'apiurl': 'https://demo.zoneminder.com/zm/api',
'portalurl': 'https://demo.zoneminder.com/zm',
'user': 'zmuser',
'password': 'zmpass',
#'disable_ssl_cert_check': True
}
'''
conf = utils.read_config('/etc/zm/secrets.ini')
api_options = {
'apiurl': utils.get(key='ZM_API_PORTAL', section='secrets', conf=conf),
'portalurl':utils.get(key='ZM_PORTAL', section='secrets', conf=conf),
'user': utils.get(key='ZM_USER', section='secrets', conf=conf),
'password': utils.get(key='ZM_PASSWORD', section='secrets', conf=conf),
# 'basic_auth_user': 'bob',
# 'basic_auth_password': 'bobs password'
#'disable_ssl_cert_check': True
}
zmapi = zmapi.ZMApi(options=api_options)
ml_options = {
'general': {
'model_sequence': 'object,face,alpr',
'disable_locks': 'no'
},
'object': {
'general':{
'pattern':'.*',
'same_model_sequence_strategy': 'most' # also 'most', 'most_unique's
},
'sequence': [{
#First run on TPU
'name': 'TPU for object detection', # descriptor (optional)
'enabled': 'no', # skips TPU. Easy way to keep configs but not enable
'object_weights':'/var/lib/zmeventnotification/models/coral_edgetpu/ssd_mobilenet_v2_coco_quant_postprocess_edgetpu.tflite',
'object_labels': '/var/lib/zmeventnotification/models/coral_edgetpu/coco_indexed.names',
'object_min_confidence': 0.3,
'object_framework':'coral_edgetpu'
},
{
# YoloV4 on GPU if TPU fails (because sequence strategy is 'first')
'name': 'GPU Yolov4 for object detection', # descriptor (optional)
'enabled': 'no', # skips. Easy way to keep configs but not enable
'object_config':'/var/lib/zmeventnotification/models/yolov4/yolov4.cfg',
'object_weights':'/var/lib/zmeventnotification/models/yolov4/yolov4.weights',
'object_labels': '/var/lib/zmeventnotification/models/yolov4/coco.names',
'object_min_confidence': 0.3,
'object_framework':'opencv',
'object_processor': 'gpu',
#'car_past_det_max_diff_area': '10%',
#'match_past_detections': 'yes',
#'car_max_detection_size': '13000',
#'truck_max_detection_size': '13000',
'image_path': '/var/lib/zmeventnotification/images',
#'model_width': 512,
#'model_height': 512
}]
},
'face': {
'general':{
'pattern': '.*',
'same_model_sequence_strategy': 'union'
},
'sequence': [{
'name': 'DLIB face recognition',
'enabled': 'yes',
'face_detection_framework': 'dlib',
'known_images_path': '/var/lib/zmeventnotification/known_faces',
'face_model': 'cnn',
'face_train_model': 'cnn',
'face_recog_dist_threshold': 0.6,
'face_num_jitters': 1,
'face_upsample_times':1,
'max_size': 800
},
{
'name': 'TPU face detection',
'enabled': 'yes',
'face_detection_framework': 'tpu',
'face_weights':'/var/lib/zmeventnotification/models/coral_edgetpu/ssd_mobilenet_v2_face_quant_postprocess_edgetpu.tflite',
'face_min_confidence': 0.3,
}]
},
'alpr': {
'general':{
'same_model_sequence_strategy': 'first',
'pre_existing_labels':['car', 'motorbike', 'bus', 'truck', 'boat'],
},
'sequence': [{
'alpr_api_type': 'cloud',
'alpr_service': 'plate_recognizer',
'alpr_key': utils.get(key='PLATEREC_ALPR_KEY', section='secrets', conf=conf),
'platrec_stats': 'no',
'platerec_min_dscore': 0.1,
'platerec_min_score': 0.2,
}]
}
} # ml_options
stream_options = {
#'frame_skip':2,
#'start_frame': 21,
#'max_frames':20,
'strategy': 'most_models',
#'strategy': 'first',
'api': zmapi,
'download': False,
'frame_set': 'snapshot,alarm',
'resize': 800,
'save_frames': False,
'save_analyzed_frames': False,
'save_frames_dir': '/tmp',
'contig_frames_before_error': 5,
'max_attempts': 3,
'sleep_between_attempts': 4,
'disable_ssl_cert_check': True,
'mid':mid
}
#input ('Enter...')
m = DetectSequence(options=ml_options)
#m = ObjectDetect.Object(options=ml_options)
matched_data,all_data = m.detect_stream(stream=eid, options=stream_options)
print('ALL FRAMES: {}\n\n'.format(all_data))
print ('SELECTED FRAME: {}, SIZE: {} LABELS: {} BOXES:{} CONFIDENCES:{}'.format(matched_data['frame_id'],matched_data['image_dimensions'],matched_data['labels'],matched_data['boxes'],matched_data['confidences']))