recognizer_opencv3.py

from pi_video_stream_thread import PiVideoStreamThread
import numpy as np
import cv2
import os
import time
import datetime
from threading import Thread
import subprocess
import signal
import math


class FaceRecognizer:
	
	def __init__(self):
		
		self.folders = 0
		
		self.treshold = 3700
		self.network_stream = None
		
		self.timestamp = None
		self.ts = None
		self.lastsaved = datetime.datetime.now()
		self.framescount = 0
		self.dropbox_mail_path = '/tmp/dropbox_mail/'
		self.opencv_stream_path = '/tmp/opencv_stream/'
		
		self.database_path = '/home/pi/Video_Doorphone_V1/database/'
		self.database_negatives_path = '/home/pi/Video_Doorphone_V1/caltech_vertical_off_012/'
		self.images = []
		self.labels = []
		self.names = {}
		
		# Choose algorith
		self.model = cv2.face.createEigenFaceRecognizer()
#		self.model = cv2.face.createFisherFaceRecognizer()
#		self.model = cv2.face.createLBPHFaceRecognizer()
		
		self.face_cascade = cv2.CascadeClassifier('./haarcascade_frontalface_default.xml')
		self.eyes_cascade = cv2.CascadeClassifier('./haarcascade_eye.xml')
		self.glasses_cascade = cv2.CascadeClassifier('./haarcascade_eye_tree_eyeglasses.xml')
		self.thread_stream = PiVideoStreamThread()
		self.image = None
		self.face_detected = False
		self.prediction = None
	
		self.monitor_doorbell = None
		
  # load database
	def load_database(self):		
		
		print 'Ladowanie obrazow...'
		id = 0
		
		# POSITIVE DATABASE
		for dirname, dirnames, filenames in os.walk(self.database_path):
			self.folders += len(dirnames)
			for subdirname in dirnames:
				self.names[id] = subdirname
				subject_path = os.path.join(dirname, subdirname)
				for filename in os.listdir(subject_path):
					abs_path = "%s/%s" % (subject_path, filename)
					img = cv2.imread(abs_path, 0)
					self.images.append(img)
					self.labels.append(int(id))
					cv2.imshow("Ladowanie obrazow...", img)
					cv2.waitKey(50)
				id += 1
		
		# NEGATIVE DATABASE
	#	for dirname, dirnames, filenames in os.walk(self.database_negatives_path):
	#		for subdirname in dirnames:
	#			self.names[id] = subdirname
	#			subject_path = os.path.join(dirname, subdirname)
	#			for filename in os.listdir(subject_path):
	#				abs_path = "%s/%s" % (subject_path, filename)
	#				img = cv2.imread(abs_path, 0)
	#				self.images.append(img)
	#				self.labels.append(int(id))
	#				cv2.imshow("Ladowanie obrazow...", img)
	#				cv2.waitKey(50)
	#			id += 1
		
		print 'Foldery (osoby) w positive database = ', self.folders
		
		cv2.destroyAllWindows()
	
	# Train
	def train_recognizer(self):
		
		print 'Uczenie modelu...'
		self.model.train(np.asarray(self.images), np.asarray(self.labels))
		print 'Trening zakonczony pomyslnie!'
		
	# Test 
	def test_recognizer(self):
		
		self.load_database()
		self.train_recognizer()
		
		print 'Inicjalizacja kamery...'
		self.thread_stream.start()
		time.sleep(2)
	
		print 'Przetwarzanie obrazu w toku...\n'
		start = time.time()
	
		while True:
			self.timestamp = datetime.datetime.now()
			self.ts = self.timestamp.strftime('%a %d-%b-%Y %H:%M:%S')
			
			self.image = self.thread_stream.read()
			gray_orig_size = cv2.cvtColor(self.image, cv2.COLOR_BGR2GRAY)
			ratio = 2
			gray = cv2.resize(gray_orig_size, (gray_orig_size.shape[1] / ratio, gray_orig_size.shape[0] / ratio), interpolation = cv2.INTER_AREA)
			#equalizegray = cv2.equalizeHist(gray)
		
			faces = self.face_cascade.detectMultiScale(gray, scaleFactor = 1.2, minNeighbors = 5, minSize = (30,30))
		
			for (x,y,w,h) in faces:
				x_ = x * ratio
				y_ = y * ratio
				x_w_ = (x + w) * ratio
				y_h_ = (y + h) * ratio
				
				self.face_detected = True
		
				roi_color = self.image[y_:y_h_, x_:x_w_]
				roi_gray = gray_orig_size[y_:y_h_, x_:x_w_]
		
				horizontal_offset = int(0.2 * w * ratio)
				vertical_offset = int(0.15 * h * ratio)
				extracted_face = gray_orig_size[y_ + vertical_offset: y_h_ - vertical_offset, x_ + horizontal_offset: x_w_ - horizontal_offset]
				extracted_face = cv2.resize(extracted_face, (92, 112), interpolation = cv2.INTER_AREA)
				extracted_face = cv2.equalizeHist(extracted_face)
			
				cv2.imshow('Wykryta twarz', extracted_face)
				
				collector = cv2.face.StandardCollector_create()
				self.model.predict_collect(extracted_face, collector)
				list_of_conf_id_tuples = collector.getResults()
				print '\n' + str(list_of_conf_id_tuples) + '\n'
				print '\n' + str(list_of_conf_id_tuples[0]) + '\n'
				print '\n' + str(list_of_conf_id_tuples[0][0]) + '\n'
				print '\n' + str(list_of_conf_id_tuples[0][1]) + '\n'
				print '\n' + str(list_of_conf_id_tuples[4]) + '\n'
				
				self.prediction = self.model.predict(extracted_face)
		
				if self.prediction[0] < self.folders:
					if self.prediction[1] < self.treshold:
						known_face = cv2.rectangle(self.image, (x_ + horizontal_offset, y_ + vertical_offset), (x_w_ - horizontal_offset, y_h_ - vertical_offset), (0, 255, 0), 2)
						cv2.putText(self.image, '%s - %.0f' % (self.names[self.prediction[0]], self.prediction[1]), (x_, y_ - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2)
					else:
						unknown_face = cv2.rectangle(self.image, (x_ + horizontal_offset, y_ + vertical_offset), (x_w_ - horizontal_offset, y_h_ - vertical_offset), (0, 0, 255), 2)
						cv2.putText(self.image, 'Brak danych - %s - %.0f' % (self.names[self.prediction[0]], self.prediction[1]), (x_, y_ - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 2)
				else:
					unknown_face = cv2.rectangle(self.image, (x_ + horizontal_offset, y_ + vertical_offset), (x_w_ - horizontal_offset, y_h_ - vertical_offset), (0, 0, 255), 2)
					cv2.putText(self.image, 'Brak danych - %s - %.0f' % (self.names[self.prediction[0]], self.prediction[1]), (x_, y_ - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 2)
			
			cv2.putText(self.image, self.ts, (10, self.image.shape[0] - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 255), 2)
			cv2.imshow('Rozpoznawanie twarzy', self.image)
		
			end = time.time()
			print 'FPS: ', int(round(1 / (end - start)))
			start = time.time()
		
			if cv2.waitKey(1) & 0xFF == ord("q"):
				break

		print 'Zamykanie programu...'
		cv2.destroyAllWindows()
		self.thread_stream.stop()
	
	
def main():
	app = FaceRecognizer()
	app.test_recognizer()
	
if __name__ == '__main__':
	main()