سیستم‌های تشخیص و ردیابی عابرین پیاده در دوربین های نظارتی می‌توانند در قالب نرم‌افزارهای مختلفی ظاهر شوند. برای مثال:
در این مقاله، قصد دارم توضیح دهم که چطور با استفاده از کتابخانه PyTorch و چارچوب flask یک لایه تعاملی را در سیستم ردیابی برخط عابرین پیاده در دوربین‌های نظارتی به‌کار گرفته‌ام.
در این مقاله، ابتدا به‌صورت خلاصه الگوریتم deep sort را معرفی می‌کنم و پس از آن عمدتاً به نکات طراحی وب‌سرور خواهم پرداخت. کدهای مربوط به این پروژه را می‌توانید در این لینک ملاحظه فرمایید.
براساس کتاب «Practical Computer Vision»، روش‌های کلاسیک ردیابی چند شیء به دو بخش تقسیم می‌شوند:
شناسایی: ابتدا تمامی اشیاء موردنظر شناسایی می‌شوند.
مطابقت: سپس اشیاء مشابه با توجه به فریم پیشین، با یک‌دیگر تطابق داده می‌شوند. پس از آن این فریم‌های تطبیق‌یافته دنبال می‌شوند تا ردپای یک شیء به دست آید.
اما این روش در الگوریتم deep sort به سه مرحله تقسیم می‌شود:
برای آشنایی بیشتر با الگوریتم deep sort، پیشنهاد می‌کنم صفحات ۱۲۶ تا ۱۲۹ از کتاب «Practical Computer Vision» و این وبلاگ را مطالعه نمایید.
مدتی بود که در وبلاگ‌های مختلف به دنبال کدهای متن‌باز برنامه‌های شناسایی عابرین پیاده در ویدیوهای پخش زنده بودم و متأسفانه نتوانستم منبعی پیدا کنم که هم هوش مصنوعی و هم بخش‌های مربوط به سمت سرور را شامل شود. بنابراین، تصمیم گرفتم کدهای مربوطه را بنویسم به این امید که سایرین بتوانند با کمک این پروژه ساده، پروژه‌های پیشرفته‌تری تعریف کنند.
پروتکل پخش زنده ((RTSP یک پروتکل برای کنترل شبکه در سیستم‌های ارتباطی است که به منظور کنترل پخش ویدیویی از سرورهای رسانه‌ای طراحی شده است. من در این نوشتار آموزشی فرض را بر این گذاشته‌ام که به لینک مربوط به پخش زنده دوربین‌هایی که قصد دارید این سرویس را روی آن‌ها آزمایش کنید، دسترسی دارید. درغیر این صورت، می‌توانید از ویدیوهای وب‌سایت earthcam.com استفاده کنید.
من در حوزه طراحی الگو و مهندسی سخت‌افزار تجربه کمی دارم، به همین دلیل در تلاش اول به این ساختار رسیدم:
اما فوراً متوجه ضعف‌های این رویکرد شدم:
به این ترتیب، من متوجه شدم که اغلب مشکلات به دلیل همبستگی شدید فرمان‌های هوش مصنوعی با ماژول‌های لایه تعاملی است و تصمیم گرفتم که یک ماژول غیر همزمانبه نام رِدیس (Redis) را به مدل اضافه کنم تا از شدت این همبستگی بکاهد.
به لطف مقاله آقای آدریان رزبروک به نام «Deep learning in production with Keras, Redis, Flask, and Apache»، متوجه اهمیت استفاده از ماژول ردیس برای ذخیره‌سازی فریم‌ها در وب‌سرور شدم و به ساختار زیر رسیدم:
هر یک دایره‌های سیاه‌رنگ در تصویر بالا نشان‌دهنده یکی از مراحل زیر است:
من در این پروژه از کتابخانه‌های زیر استفاده کرده‌ام:
Python 3.7
Opencv-python
Sklearn
Torch > 0.4
Torchvision >=0.1
Pillow
Easydict
Redis
Dotenv
Flask
من این پروژه را روی سیستم عامل Ubuntu 16.04 و GPU مدل Nvidia GeForce GTX 2080 اجرا کردم.
برای عملیات ذخیره‌سازی نیز ماژول Redis را نصب کردم و آن را روی پورت ۶۳۷۹ اجرا کردم. آموزش کامل نصب Redis را می‌توانید در این لینک مشاهده نمایید.
در مرحله اول تنها سعی کردم فریم‌های ویدیو را ضبط کرده و در ردیس ذخیره کنم. اشیای cv2.VideoCapture در کتابخانه OpenCV به‌طور پیش‌فرض می‌توانند به داده‌هایی که دوربین تولید می‌کند، رسیدگی کنند. اما براساس این نوشتار، ظاهراً احتمال از دست رفتن بسته‌های داده‌ای در این ماژول وجود دارد. بنابراین، من از نخ‌ها برای غلبه بر این مشکل استفاده کردم. در کادر زیر می‌توانید کد اولیه را ملاحظه فرمایید:
import cv2
from concurrent.futures import ThreadPoolExecutor
from redis import Redis
redis_cache = Redis('۱۲۷.۰.۰.۱')
class RealTimeTracking(object):
def __init__(self, src):
self.vdo = cv2.VideoCapture(self.args.input)
self.status, self.frame = None, None
self.output_frame = None
self.thread = ThreadPoolExecutor(max_workers=۱)
self.thread.submit(self.update)
def update(self):
while True:
if self.vdo.isOpened():
(self.status, self.frame) = self.vdo.read()
def run(self):
print('streaming started ...')
while True:
try:
redis_cache.set('frame', frame_to_bytes)
except AttributeError:
pass
خواندن فریم‌های RTSP با استفاده از نخ‌ها برای حل مسئله از دست رفتن بسته‌های داده‌ای
کد سرویس شناسایی عابرین پیاده در ویدیوهای ساده را می‌توانید در این لینک مشاهده کنید. با ترکیب این کد و کدی که در مرحله اول نوشتم، به کد زیر میرسیم:
import warnings
from os import getenv
import sys
from os.path import dirname, abspath
sys.path.append(dirname(dirname(abspath(__file__))))
import torch
from deep_sort import build_tracker
from detector import build_detector
import cv2
from utils.draw import compute_color_for_labels
from concurrent.futures import ThreadPoolExecutor
from redis import Redis
redis_cache = Redis('۱۲۷.۰.۰.۱')
class RealTimeTracking(object):
"""
This class is built to get frame from rtsp link and continuously
assign each frame to an attribute namely as frame in order to
compensate the network packet loss. then we use flask to give it
as service to client.
Args:
args: parse_args inputs
cfg: deepsort dict and yolo-model cfg from server_cfg file
"""
def __init__(self, cfg, args):
# Create a VideoCapture object
self.cfg = cfg
self.args = args
use_cuda = self.args.use_cuda and torch.cuda.is_available()
if not use_cuda:
warnings.warn(UserWarning("Running in cpu mode!"))
self.detector = build_detector(cfg, use_cuda=use_cuda)
self.deepsort = build_tracker(cfg, use_cuda=use_cuda)
self.class_names = self.detector.class_names
self.vdo = cv2.VideoCapture(self.args.input)
self.status, self.frame = None, None
self.im_width = int(self.vdo.get(cv2.CAP_PROP_FRAME_WIDTH))
self.im_height = int(self.vdo.get(cv2.CAP_PROP_FRAME_HEIGHT))
self.output_frame = None
self.thread = ThreadPoolExecutor(max_workers=۱)
self.thread.submit(self.update)
def update(self):
"""
Repeatedly reading frames from camera using threads
"""
while True:
if self.vdo.isOpened():
(self.status, self.frame) = self.vdo.read()
def run(self):
"""
Until the in_progress flag is not set as off, keep reading and processing
pedestrian detection.
"""
print('streaming started ...')
while getenv('in_progress') != 'off':
try:
frame = self.frame.copy()
self.detection(frame=frame)
frame_to_bytes = cv2.imencode('.jpg', frame)[۱].tobytes()
redis_cache.set('frame', frame_to_bytes)
except AttributeError:
pass
print('streaming stopped ...')
def detection(self, frame):
"""
Detection, tracking and drawing the bboxes on pedestrians
"""
im = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
# do detection
bbox_xywh, cls_conf, cls_ids = self.detector(im)
if bbox_xywh is not None:
# select person class
mask = cls_ids == ۰
bbox_xywh = bbox_xywh[mask]
bbox_xywh[:, ۳:] *= ۱.۲ # bbox dilation just in case bbox too small
cls_conf = cls_conf[mask]
# do tracking
outputs = self.deepsort.update(bbox_xywh, cls_conf, im)
# draw boxes for visualization
if len(outputs) > ۰:
self.draw_boxes(img=frame, output=outputs)
@staticmethod
def draw_boxes(img, output, offset=(۰, ۰)):
# Draws bboxes on the detcted pedestrians
for i, box in enumerate(output):
x1, y1, x2, y2, identity = [int(ii) for ii in box]
x1 += offset[۰]
x2 += offset[۰]
y1 += offset[۱]
y2 += offset[۱]
# box text and bar
color = compute_color_for_labels(identity)
label = '{}{:d}'.format("", identity)
t_size = cv2.getTextSize(label, cv2.FONT_HERSHEY_PLAIN, ۲, ۲)[۰]
cv2.rectangle(img, (x1, y1), (x2, y2), color, ۳)
cv2.rectangle(img, (x1, y1), (x1 + t_size[۰] + ۳, y1 + t_size[۱] + ۴), color, -۱)
cv2.putText(img, label, (x1, y1 + t_size[۱] + ۴), cv2.FONT_HERSHEY_PLAIN, ۲, [۲۵۵, ۲۵۵, ۲۵۵], ۲)
return img
کدهای تشخیص عابرین پیاده
در ادامه به تشریح برخی از تغییرات مهم می‌پردازیم:
اساساً در وب‌سرور برای اجرای این برنامه به ۲ ابزار احتیاج داریم. یک ابزار برای بازیابی فریم‌های ذخیره‌شده در Redis و تبدیل آن‌ها به لینک HTTP. و یک ابزار دیگر که به ما امکان فعال‌سازی و متوقف کردن سرویس هوش مصنوعی را بدهد. به‌منظور تهیه ابزار اول یک قالب HTML به نام index.html ساختم و سپس آن را در مسیر templates قرار دادم. با استفاده از این قالب می‌توان تصاویری که توسط وب‌سرور تهیه شده‌اند را به کاربران نمایش داد.
<html lang="en">
<head>
<title>Camera #۱ </title>
</head>
<body>
<h1>Floor ۲ </h1>
<img src="{{ url_for('video_feed') }}">
</body>
</html>
سمت کاربر
و برای پیکربندی پارامترهای deep sort و مدل تشخیص عابر پیاده YOLO، فایل کدهای server_cfg.py را نیز به کدهای پیشین اضافه کردم:
""" Configuring deep learning models parameters in dictionary variables.
"""
import sys
from os.path import dirname, abspath, isfile
sys.path.append(dirname(dirname(abspath(__file__))))
from dotenv import load_dotenv
from utils.asserts import assert_in_env
from os import getenv
from os.path import join
load_dotenv('.env')
# Configure deep sort info
deep_sort_info = dict(REID_CKPT=join(getenv('project_root'), getenv('reid_ckpt')),
MAX_DIST=۰.۲,
MIN_CONFIDENCE=.۳,
NMS_MAX_OVERLAP=۰.۵,
MAX_IOU_DISTANCE=۰.۷,
N_INIT=۳,
MAX_AGE=۷۰,
NN_BUDGET=۱۰۰)
deep_sort_dict = {'DEEPSORT': deep_sort_info}
# Configure yolov3 info
yolov3_info = dict(CFG=join(getenv('project_root'), getenv('yolov3_cfg')),
WEIGHT=join(getenv('project_root'), getenv('yolov3_weight')),
CLASS_NAMES=join(getenv('project_root'), getenv('yolov3_class_names')),
SCORE_THRESH=۰.۵,
NMS_THRESH=۰.۴
)
yolov3_dict = {'YOLOV3': yolov3_info}
# Configure yolov3-tiny info
yolov3_tiny_info = dict(CFG=join(getenv('project_root'), getenv('yolov3_tiny_cfg')),
WEIGHT=join(getenv('project_root'), getenv('yolov3_tiny_weight')),
CLASS_NAMES=join(getenv('project_root'), getenv('yolov3_class_names')),
SCORE_THRESH=۰.۵,
NMS_THRESH=۰.۴
)
yolov3_tiny_dict = {'YOLOV3': yolov3_tiny_info}
check_list = ['project_root', 'reid_ckpt', 'yolov3_class_names', 'model_type', 'yolov3_cfg', 'yolov3_weight',
'yolov3_tiny_cfg', 'yolov3_tiny_weight', 'yolov3_class_names']
if assert_in_env(check_list):
assert isfile(deep_sort_info['REID_CKPT'])
if getenv('model_type') == 'yolov3':
assert isfile(yolov3_info['WEIGHT'])
assert isfile(yolov3_info['CFG'])
assert isfile(yolov3_info['CLASS_NAMES'])
model = yolov3_dict.copy()
elif getenv('model_type') == 'yolov3_tiny':
assert isfile(yolov3_tiny_info['WEIGHT'])
assert isfile(yolov3_tiny_info['CFG'])
assert isfile(yolov3_tiny_info['CLASS_NAMES'])
model = yolov3_tiny_dict.copy()
else:
raise ValueError("Value '{}' for model_type is not valid".format(getenv('model_type')))
پیکربندی مدل یادگیری عمیق
من در تمامی پروژه‌هایم یک فایل .env ایجاد می‌کنم و اطلاعات مربوط به فایل‌های ایستا از جمله وزن‌های مدل، فایل‌های پیکربندی و دیتاست‌ها را در آن قرار می‌دهم.
و درنهایت کد وب‌سرور به این شکل خواهد بود:
This code handles the pedestrian detection service on specified camera.
Also provides stream images for clients.
"""
from os.path import join
from os import getenv, environ
from dotenv import load_dotenv
import argparse
from threading import Thread
from redis import Redis
from flask import Response, Flask, jsonify, request, abort
from rtsp_threaded_tracker import RealTimeTracking
from server_cfg import model, deep_sort_dict
from config.config import DevelopmentConfig
from utils.parser import get_config
redis_cache = Redis('۱۲۷.۰.۰.۱')
app = Flask(__name__)
environ['in_progress'] = 'off'
def parse_args():
"""
Parses the arguments
Returns:
argparse Namespace
"""
assert 'project_root' in environ.keys()
project_root = getenv('project_root')
parser = argparse.ArgumentParser()
parser.add_argument("--input",
type=str,
default=getenv('camera_stream'))
parser.add_argument("--model",
type=str,
default=join(getenv('model_type')))
parser.add_argument("--cpu",
dest="use_cuda",
action="store_false", default=True)
args = parser.parse_args()
return args
def gen():
"""
Returns: video frames from redis cache
"""
while True:
frame = redis_cache.get('frame')
if frame is not None:
yield b'--frame\r\n'b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n'
def pedestrian_tracking(cfg, args):
"""
starts the pedestrian detection on rtsp link
Args:
cfg:
args:
Returns:
"""
tracker = RealTimeTracking(cfg, args)
tracker.run()
def trigger_process(cfg, args):
"""
triggers pedestrian_tracking process on rtsp link using a thread
Args:
cfg:
args:
Returns:
"""
try:
t = Thread(target=pedestrian_tracking, args=(cfg, args))
t.start()
return jsonify({"message": "Pedestrian detection started successfully"})
except Exception:
return jsonify({'message': "Unexpected exception occured in process"})
@app.errorhandler(۴۰۰)
def bad_argument(error):
return jsonify({'message': error.description['message']})
# Routes
@app.route('/stream', methods=['GET'])
def stream():
"""
Provides video frames on http link
Returns:
"""
return Response(gen(),
mimetype='multipart/x-mixed-replace; boundary=frame')
@app.route("/run", methods=['GET'])
def process_manager():
"""
request parameters:
run (bool): 1 -> start the pedestrian tracking
0 -> stop it
camera_stream: str -> rtsp link to security camera
:return:
"""
# data = request.args
data = request.args
status = data['run']
status = int(status) if status.isnumeric() else abort(۴۰۰, {'message': f"bad argument for run {data['run']}"})
if status == ۱:
# if pedestrian tracking is not running, start it off!
try:
if environ.get('in_progress', 'off') == 'off':
global cfg, args
vdo = data.get('camera_stream')
if vdo is not None:
args.input = int(vdo)
environ['in_progress'] = 'on'
return trigger_process(cfg, args)
elif environ.get('in_progress') == 'on':
# if pedestrian tracking is running, don't start another one (we are short of gpu resources)
return jsonify({"message": " Pedestrian detection is already in progress."})
except Exception:
environ['in_progress'] = 'off'
return abort(۵۰۳)
elif status == ۰:
if environ.get('in_progress', 'off') == 'off':
return jsonify({"message": "pedestrian detection is already terminated!"})
else:
environ['in_progress'] = 'off'
return jsonify({"message": "Pedestrian detection terminated!"})
if __name__ == '__main__':
load_dotenv()
app.config.from_object(DevelopmentConfig)
# BackProcess Initialization
args = parse_args()
cfg = get_config()
cfg.merge_from_dict(model)
cfg.merge_from_dict(deep_sort_dict)
# Start the flask app
app.run()
برنامه وب‌سرور که کنترل سرویس مبتنی بر هوش مصنوعی را برعهده دارد
حال بیایید این کد را با جزییات بررسی کنیم:
در آخر باید چند مجوز و سند نیز به این تابع اضافه شود که آن‌ها را در Github قرار خواهم داد.
در این مقاله سرویس شناسایی عابرین پیاده را به روشی ساده روی دوربین‌های نظارتی اجرا کردیم و آموختیم که نحوه طراحی معماری در تعمیم‌پذیریاین برنامه تا چه حد تأثیرگذار است. در این پروژه یک ابزار ذخیره‌سازی به نام ردیس به ما کمک کرد تا بتوانیم با جداسازی فرآیند شناسایی عابرین از بخش وب‌سرور، ساختار بهتری طراحی کنیم. همچنین، یک API ساده تهیه کردیم که با استفاده از آن توانستیم فرآیند شناسایی عابرین و ورودی‌های آن را کنترل کنیم.
در آخر باید از آقای آدریان رزبروک بابت پست‌ها عالی وبلاگ‌شان (به ویژه این پست) که مرا در مسیر اجرای این پروژه راهنمایی کرد، تشکر کنم. همچنین باید از آقای ژیانگ پی نیز بابت در دسترس قرار دادن رویکرد خود در زمینه شناسایی عابرین به صورت متن‌باز، تشکر کنم.
منبع: hooshio.com
