93 lines
3.5 KiB
Python
93 lines
3.5 KiB
Python
import cv2
|
|
import time
|
|
import os
|
|
import requests
|
|
import json
|
|
from retinaface import RetinaFace
|
|
|
|
treashold = float(os.getenv('FACE_TRASHOLD', '0.9'))
|
|
sever = os.getenv('API_ROOT', 'localhost:8000')
|
|
root_api_url = 'http://{}/local'.format(sever)
|
|
base_path = '/app/data/'
|
|
|
|
print("API_ROOT: " + str(root_api_url), flush=True)
|
|
print("FACE_TRASHOLD: " + str(treashold), flush=True)
|
|
|
|
|
|
try:
|
|
while True:
|
|
print("LOOP Started !!!", flush=True)
|
|
|
|
r = requests.get('{}/anonymize'.format(root_api_url))
|
|
# print(r.text, flush=True)
|
|
response = r.json()
|
|
|
|
for detection in response:
|
|
print(str(detection['id']) + " Anonymization Started", flush=True)
|
|
|
|
img_path = "{}{}".format(base_path, detection['source_image_path'])
|
|
#print(img_path, flush=True)
|
|
#Skip Non existing files
|
|
if (not os.path.exists(img_path)):
|
|
continue
|
|
|
|
img = cv2.imread(img_path)
|
|
faces = RetinaFace.detect_faces(img, treashold)
|
|
|
|
#Count faces is null dont anonimize file
|
|
print(str(len(faces)) + " Faces Found" , flush=True)
|
|
if (len(faces) <= 0):
|
|
data = {
|
|
"id": detection['id'],
|
|
"source_image_path": img_path.replace(base_path,""),
|
|
"face_found": 0,
|
|
}
|
|
response = requests.post('{}/anonymize'.format(root_api_url), json=data)
|
|
print(str(detection['id']) + " Anonymized 0 faces" , flush=True)
|
|
continue
|
|
|
|
i = 0
|
|
for (key) in faces:
|
|
confidence = faces[key]["score"]
|
|
print(str(confidence) + " Faces Confidence" , flush=True)
|
|
x1, y1 = min(faces[key]["facial_area"][0], faces[key]["facial_area"][2]) - 2, min(faces[key]["facial_area"][1], faces[key]["facial_area"][3]) - 2
|
|
x2, y2 = max(faces[key]["facial_area"][0], faces[key]["facial_area"][2]) + 2, max(faces[key]["facial_area"][1], faces[key]["facial_area"][3]) + 2
|
|
x1, x2 = max(x1, 0), min(x2, img.shape[1])
|
|
y1, y2 = max(y1, 0), min(y2, img.shape[0])
|
|
face_roi = img[y1:y2, x1:x2]
|
|
blurred_face = cv2.GaussianBlur(face_roi, (99, 99), 50)
|
|
img[y1:y2, x1:x2] = blurred_face
|
|
print("Faces Anonymized" , flush=True)
|
|
|
|
i = i +1
|
|
|
|
#cv2.imshow("test_img", img)
|
|
#cv2.waitKey(0)
|
|
extension = img_path[img_path.index(".")+1:]
|
|
|
|
new_img_path = img_path
|
|
if (not img_path.endswith(('_anonym.{}'.format(extension)))):
|
|
new_img_path = img_path.replace((".{}".format(extension)), ('_anonym.{}'.format(extension)))
|
|
else:
|
|
print("Path already anonymized !!!")
|
|
|
|
cv2.imwrite(new_img_path, img)
|
|
|
|
data = {
|
|
"id": detection['id'],
|
|
"source_image_path": new_img_path.replace(base_path,""),
|
|
"face_found": len(faces),
|
|
}
|
|
response = requests.post('{}/anonymize'.format(root_api_url), json=data)
|
|
print(str(detection['id']) + " Anonymized" , flush=True)
|
|
time.sleep(10)
|
|
|
|
print("sleep", flush=True)
|
|
time.sleep(900)
|
|
|
|
except Exception as e:
|
|
print("An exception occurred", flush=True)
|
|
print(str(e), flush=True)
|
|
|
|
data = {"trace": str(e)}
|
|
response = requests.post('{}/log'.format(root_api_url), json=data) |