Python实时图像分类问题与神经网络
我正在尝试使用caffe和python来进行实时图像分类。 我使用OpenCV从一个进程中的网络摄像头进行流式处理，并在一个单独的进程中使用caffe对从网络摄像头拉出的帧进行图像分类。 然后，我将分类结果返回给主线程，以标注摄像头流。
问题是，即使我有一个NVIDIA GPU，并且正在GPU上执行caffe预测，主线程也会被忽略。 通常不做任何预测，我的摄像头流以30 fps运行; 然而，根据预测，我的摄像头流最多只能达到15 fps。
我已经验证了caffe在执行预测时的确在使用GPU，并且我的GPU或GPU内存不是最大的。 我还证实，在编程期间，我的CPU内核没有任何问题。 我想知道如果我做错了什么，或者如果没有办法让这两个过程真正分开。 任何建议表示赞赏。 这里是我的代码供参考
class Consumer(multiprocessing.Process):
def __init__(self, task_queue, result_queue):
multiprocessing.Process.__init__(self)
self.task_queue = task_queue
self.result_queue = result_queue
#other initialization stuff
def run(self):
caffe.set_mode_gpu()
caffe.set_device(0)
#Load caffe net -- code omitted
while True:
image = self.task_queue.get()
#crop image -- code omitted
text = net.predict(image)
self.result_queue.put(text)
return
import cv2
import caffe
import multiprocessing
import Queue
tasks = multiprocessing.Queue()
results = multiprocessing.Queue()
consumer = Consumer(tasks,results)
consumer.start()
#Creating window and starting video capturer from camera
cv2.namedWindow("preview")
vc = cv2.VideoCapture(0)
#Try to get the first frame
if vc.isOpened():
rval, frame = vc.read()
else:
rval = False
frame_copy[:] = frame
task_empty = True
while rval:
if task_empty:
tasks.put(frame_copy)
task_empty = False
if not results.empty():
text = results.get()
#Add text to frame
cv2.putText(frame,text)
task_empty = True
#Showing the frame with all the applied modifications
cv2.imshow("preview", frame)
#Getting next frame from camera
rval, frame = vc.read()
frame_copy[:] = frame
#Getting keyboard input
key = cv2.waitKey(1)
#exit on ESC
if key == 27:
break
我非常肯定这是咖啡预测会放慢一切，因为当我评论预测并在过程之间来回传递虚拟文本时，我会再次获得30 fps。
class Consumer(multiprocessing.Process):
def __init__(self, task_queue, result_queue):
multiprocessing.Process.__init__(self)
self.task_queue = task_queue
self.result_queue = result_queue
#other initialization stuff
def run(self):
caffe.set_mode_gpu()
caffe.set_device(0)
#Load caffe net -- code omitted
while True:
image = self.task_queue.get()
#crop image -- code omitted
#text = net.predict(image)
text = "dummy text"
self.result_queue.put(text)
return
import cv2
import caffe
import multiprocessing
import Queue
tasks = multiprocessing.Queue()
results = multiprocessing.Queue()
consumer = Consumer(tasks,results)
consumer.start()
#Creating window and starting video capturer from camera
cv2.namedWindow("preview")
vc = cv2.VideoCapture(0)
#Try to get the first frame
if vc.isOpened():
rval, frame = vc.read()
else:
rval = False
frame_copy[:] = frame
task_empty = True
while rval:
if task_empty:
tasks.put(frame_copy)
task_empty = False
if not results.empty():
text = results.get()
#Add text to frame
cv2.putText(frame,text)
task_empty = True
#Showing the frame with all the applied modifications
cv2.imshow("preview", frame)
#Getting next frame from camera
rval, frame = vc.read()
frame_copy[:] = frame
#Getting keyboard input
key = cv2.waitKey(1)
#exit on ESC
if key == 27:
break
一些解释和一些反思：
我使用Intel Core i5-6300HQ @2.3GHz CPU， 8 GB RAM和NVIDIA GeForce GTX 960M gpu（2GB内存）在笔记本电脑上运行我的代码，结果如下：
无论我是否使用caffe运行代码（通过注释掉或不是net_output = this->net_->Forward(net_input)和一些必要的东西在void Consumer::entry() ），我总能获得30 fps左右主线程。
在具有Intel Core i5-4440 cpu， 8 GB RAM ， NVIDIA GeForce GT 630 gpu（1GB内存）的PC上也获得了类似的结果。
我在同一台笔记本电脑上运行了@ user3543300的代码，结果是：
无论咖啡是否在运行（在GPU上），我也可以达到30帧/秒左右。
根据@ user3543300的反馈，使用上述代码的2个版本，@ user3543300只能运行caffe（在Nvidia GeForce 940MX GPU and Intel® Core™ i7-6500U CPU @ 2.50GHz × 4笔记本电脑）。 当caffe作为独立程序运行时，网络摄像头的帧速率也会有所下降。
所以我仍然认为这个问题很可能存在于硬件I / O限制，如DMA带宽（这个关于DMA的线程可能暗示）或RAM带宽。 希望@ user3543300可以检查这个或找出我没有意识到的真正问题。
如果问题确实是我上面想到的，那么明智的想法就是减少CNN网络引入的内存I / O开销。 事实上，为了解决硬件资源有限的嵌入式系统中的类似问题，对这个问题进行了一些研究，例如Qautization结构稀疏深度神经网络，SqueezeNet，深度压缩。 所以希望通过应用这样的技巧，它也有助于提高摄像头在帧率上的速度。
原始答案：
试试这个c ++解决方案。 它使用线程来处理任务中的I / O开销，我使用bvlc_alexnet.caffemodel和deploy.prototxt进行了测试，以进行图像分类，并且在caffe运行时（在GPU上）没有看到明显的主线程（网络摄像头流） ）：
#include <stdio.h>
#include <iostream>
#include <string>
#include <boost/thread.hpp>
#include <boost/shared_ptr.hpp>
#include "caffe/caffe.hpp"
#include "caffe/util/blocking_queue.hpp"
#include "caffe/data_transformer.hpp"
#include "opencv2/opencv.hpp"
using namespace cv;
//Queue pair for sharing image/results between webcam and caffe threads
template<typename T>
class QueuePair {
public:
explicit QueuePair(int size);
~QueuePair();
caffe::BlockingQueue<T*> free_;
caffe::BlockingQueue<T*> full_;
DISABLE_COPY_AND_ASSIGN(QueuePair);
};
template<typename T>
QueuePair<T>::QueuePair(int size) {
// Initialize the free queue
for (int i = 0; i < size; ++i) {
free_.push(new T);
}
}
template<typename T>
QueuePair<T>::~QueuePair(){
T *data;
while (free_.try_pop(&data)){
delete data;
}
while (full_.try_pop(&data)){
delete data;
}
}
template class QueuePair<Mat>;
template class QueuePair<std::string>;
//Do image classification(caffe predict) using a subthread
class Consumer{
public:
Consumer(boost::shared_ptr<QueuePair<Mat>> task
, boost::shared_ptr<QueuePair<std::string>> result);
~Consumer();
void Run();
void Stop();
void entry(boost::shared_ptr<QueuePair<Mat>> task
, boost::shared_ptr<QueuePair<std::string>> result);
private:
bool must_stop();
boost::shared_ptr<QueuePair<Mat> > task_q_;
boost::shared_ptr<QueuePair<std::string> > result_q_;
//caffe::Blob<float> *net_input_blob_;
boost::shared_ptr<caffe::DataTransformer<float> > data_transformer_;
boost::shared_ptr<caffe::Net<float> > net_;
std::vector<std::string> synset_words_;
boost::shared_ptr<boost::thread> thread_;
};
Consumer::Consumer(boost::shared_ptr<QueuePair<Mat>> task
, boost::shared_ptr<QueuePair<std::string>> result) :
task_q_(task), result_q_(result), thread_(){
//for data preprocess
caffe::TransformationParameter trans_para;
//set mean
trans_para.set_mean_file("/path/to/imagenet_mean.binaryproto");
//set crop size, here is cropping 227x227 from 256x256
trans_para.set_crop_size(227);
//instantiate a DataTransformer using trans_para for image preprocess
data_transformer_.reset(new caffe::DataTransformer<float>(trans_para
, caffe::TEST));
//initialize a caffe net
net_.reset(new caffe::Net<float>(std::string("/path/to/deploy.prototxt")
, caffe::TEST));
//net parameter
net_->CopyTrainedLayersFrom(std::string("/path/to/bvlc_alexnet.caffemodel"));
std::fstream synset_word("path/to/caffe/data/ilsvrc12/synset_words.txt");
std::string line;
if (!synset_word.good()){
std::cerr << "synset words open failed!" << std::endl;
}
while (std::getline(synset_word, line)){
synset_words_.push_back(line.substr(line.find_first_of(' '), line.length()));
}
//a container for net input, holds data converted from cv::Mat
//net_input_blob_ = new caffe::Blob<float>(1, 3, 227, 227);
}
Consumer::~Consumer(){
Stop();
//delete net_input_blob_;
}
void Consumer::entry(boost::shared_ptr<QueuePair<Mat>> task
, boost::shared_ptr<QueuePair<std::string>> result){
caffe::Caffe::set_mode(caffe::Caffe::GPU);
caffe::Caffe::SetDevice(0);
cv::Mat *frame;
cv::Mat resized_image(256, 256, CV_8UC3);
cv::Size re_size(resized_image.cols, resized_image.rows);
//for caffe input and output
const std::vector<caffe::Blob<float> *> net_input = this->net_->input_blobs();
std::vector<caffe::Blob<float> *> net_output;
//net_input.push_back(net_input_blob_);
std::string *res;
int pre_num = 1;
while (!must_stop()){
std::stringstream result_strm;
frame = task->full_.pop();
cv::resize(*frame, resized_image, re_size, 0, 0, CV_INTER_LINEAR);
this->data_transformer_->Transform(resized_image, *net_input[0]);
net_output = this->net_->Forward();
task->free_.push(frame);
res = result->free_.pop();
//Process results here
for (int i = 0; i < pre_num; ++i){
result_strm << synset_words_[net_output[0]->cpu_data()[i]] << " "
<< net_output[0]->cpu_data()[i + pre_num] << "n";
}
*res = result_strm.str();
result->full_.push(res);
}
}
void Consumer::Run(){
if (!thread_){
try{
thread_.reset(new boost::thread(&Consumer::entry, this, task_q_, result_q_));
}
catch (std::exception& e) {
std::cerr << "Thread exception: " << e.what() << std::endl;
}
}
else
std::cout << "Consumer thread may have been running!" << std::endl;
};
void Consumer::Stop(){
if (thread_ && thread_->joinable()){
thread_->interrupt();
try {
thread_->join();
}
catch (boost::thread_interrupted&) {
}
catch (std::exception& e) {
std::cerr << "Thread exception: " << e.what() << std::endl;
}
}
}
bool Consumer::must_stop(){
return thread_ && thread_->interruption_requested();
}
int main(void)
{
int max_queue_size = 1000;
boost::shared_ptr<QueuePair<Mat>> tasks(new QueuePair<Mat>(max_queue_size));
boost::shared_ptr<QueuePair<std::string>> results(new QueuePair<std::string>(max_queue_size));
char str[100], info_str[100] = " results: ";
VideoCapture vc(0);
if (!vc.isOpened())
return -1;
Consumer consumer(tasks, results);
consumer.Run();
Mat frame, *frame_copy;
namedWindow("preview");
double t, fps;
while (true){
t = (double)getTickCount();
vc.read(frame);
if (waitKey(1) >= 0){
consuer.Stop();
break;
}
if (tasks->free_.try_peek(&frame_copy)){
frame_copy = tasks->free_.pop();
*frame_copy = frame.clone();
tasks->full_.push(frame_copy);
}
std::string *res;
std::string frame_info("");
if (results->full_.try_peek(&res)){
res = results->full_.pop();
frame_info = frame_info + info_str;
frame_info = frame_info + *res;
results->free_.push(res);
}
t = ((double)getTickCount() - t) / getTickFrequency();
fps = 1.0 / t;
sprintf(str, " fps: %.2f", fps);
frame_info = frame_info + str;
putText(frame, frame_info, Point(5, 20)
, FONT_HERSHEY_SIMPLEX, 0.5, Scalar(0, 255, 0));
imshow("preview", frame);
}
}
在src / caffe / util / blocking_queue.cpp中，稍作修改并重建caffe：
...//Other stuff
template class BlockingQueue<Batch<float>*>;
template class BlockingQueue<Batch<double>*>;
template class BlockingQueue<Datum*>;
template class BlockingQueue<shared_ptr<DataReader::QueuePair> >;
template class BlockingQueue<P2PSync<float>*>;
template class BlockingQueue<P2PSync<double>*>;
//add these 2 lines below
template class BlockingQueue<cv::Mat*>;
template class BlockingQueue<std::string*>;
看起来caffe的python包装器阻止了Global Interpreter Lock（GIL）。 因此调用任何caffe python命令都会阻止所有python线程。
解决方法（风险自负）将禁用GIL以实现特定的caffe功能。 例如，如果您希望能够无锁地forward运行，您可以编辑$CAFFE_ROOT/python/caffe/_caffe.cpp 。 添加此功能：
void Net_Forward(Net<Dtype>& net, int start, int end) {
Py_BEGIN_ALLOW_THREADS; // <-- disable GIL
net.ForwardFromTo(start, end);
Py_END_ALLOW_THREADS; // <-- restore GIL
}
并用以下代替.def("_forward", &Net<Dtype>::ForwardFromTo)
.def("_forward", &Net_Forward)
改变之后不要忘记make pycaffe 。
看到这个更多细节。
有人认为你的代码可能会发生，也就是说，它在第一次调用时可以在gpu模式下工作，并且在后面的调用中，它会计算cpu模式下的分类，因为它是默认模式。 在较早版本的caffe中，设置一次gpu模式就足够了，现在需要更新版本，每次都需要设置模式。 您可以尝试以下更改：
def run(self):
#Load caffe net -- code omitted
while True:
caffe.set_mode_gpu()
caffe.set_device(0)
image = self.task_queue.get()
#crop image -- code omitted
text = net.predict(image)
self.result_queue.put(text)
return
当消费者线程正在运行时，请查看GPU时序。 您可以使用以下命令为nvidia：
nvidia-smi
以上命令将显示运行时的GPU利用率。
如果它不能解决另一个解决方案，则在一个线程下创建opencv帧提取代码。 由于它与I / O和设备访问相关，您可能会从GUI线程/主线程在单独的线程上运行它。 该线程将推送队列中的帧，并且当前消费者线程将预测。 在这种情况下，请仔细处理关键块的队列。
链接地址: http://www.djcxy.com/p/55207.html上一篇: Python real time image classification problems with Neural Networks