reality_exploration/backup/video_demo05Nov
2020-11-11 18:03:00 +01:00

327 lines
11 KiB
Text

from __future__ import division
import time
from flirpy.camera.boson import Boson
# import torch
# import torch.nn as nn
# from torch.autograd import Variable
import numpy as np
import cv2
from matplotlib import pyplot as plt
from util import *
from darknet import Darknet
from preprocess import prep_image, inp_to_image, letterbox_image
import pickle as pkl
import argparse
from scipy.signal import savgol_filter
from pythonosc import udp_client
from pythonosc import osc_server
from pythonosc import dispatcher
import threading
# import sounddevice as sd
import soundfile as sf
from playsound import playsound
client = udp_client.SimpleUDPClient('192.168.21.200', 10000 )
# client1 = udp_client.SimpleUDPClient('192.168.21.200', 30000)
client1 = udp_client.SimpleUDPClient('127.0.0.1', 15000)
client_is_person_present = udp_client.SimpleUDPClient('192.168.21.200', 20000)
# data, fs = sf.read('bell_short.wav', dtype='float32')
# sd.play(data.T, fs, device=2)
print("""
Choose Object detection:
Nose = 'n'
Face = 'f'
""")
detection_model = input()
if detection_model == 'f':
weight_file = "data/yolov3-wider_face_16000.weights"
cfg_file = "cfg/yolov3-face.cfg"
print("Activating FACE Detection")
else:
weight_file = "data/yolov3_training_final.weights"
cfg_file = "cfg/yolov3_pysource_testing.cfg"
print("Activating NOSE Detection")
def get_test_input(input_dim, CUDA):
img = cv2.imread("data/dog-cycle-car.png")
img = cv2.resize(img, (input_dim, input_dim))
img_ = img[:,:,::-1].transpose((2,0,1))
img_ = img_[np.newaxis,:,:,:]/255.0
img_ = torch.from_numpy(img_).float()
img_ = Variable(img_)
if CUDA:
img_ = img_.cuda()
return img_
def prep_image(img, inp_dim):
"""
Prepare image for inputting to the neural network.Returns a Variable
"""
orig_im = img
dim = orig_im.shape[1], orig_im.shape[0]
img = (letterbox_image(orig_im, (inp_dim, inp_dim)))
img_ = img[:,:,::-1].transpose((2,0,1)).copy()
img_ = torch.from_numpy(img_).float().div(255.0).unsqueeze(0)
return img_, orig_im, dim
prev_vals = []
inhale_val = 0
exhale_val = 0
is_val_up = False
is_exaling = False
Is_person_present = 0
def write(x, img, abs_image):
global prev_vals, inhale_val, exhale_val, Is_person_present
c1 = tuple(x[1:3].int())
c2 = tuple(x[3:5].int())
cls = int(x[-1])
label = "person"#"{0}".format(classes[cls])
color = (0, 0, 255)#random.choice(colors)
if (label=='person'):
if c1[0]>10 and c1[0]<600:
client_is_person_present.send_message("/python/person", 1)
else:
client1.send_message("/sound", (0,0,0))
if detection_model == 'f':
cv2.rectangle(img, c1, c2, (0,0,0), 2)
c1 = tuple([c1[0]+20, c1[1]+25])
c2 = tuple([c2[0]-15, c2[1]-15])
cropped_abs = abs_image[c1[1]:c2[1], c1[0]:c2[0]]
cropped_abs = np.ndarray.flatten(cropped_abs)
number_of_min_pixl = int(len(cropped_abs) * 0.1)
idx = np.argpartition(cropped_abs, number_of_min_pixl)
cv2.rectangle(img, c1, c2, color, 2)
else:
cropped_abs = abs_image[c1[1]:c2[1], c1[0]:c2[0]]
cropped_abs = np.ndarray.flatten(cropped_abs)
number_of_min_pixl = int(len(cropped_abs)*0.2)
idx = np.argpartition(cropped_abs, number_of_min_pixl)
cv2.rectangle(img, c1, c2,color, 2)
if cropped_abs.any():
# obs = np.min(cropped_abs)
obs = np.mean(cropped_abs[idx[:number_of_min_pixl]])
if len(prev_vals) >= 100:
prev_vals = prev_vals[1:]
prev_vals.append(obs)
sav_filtered = savgol_filter(prev_vals, 11, 3)
prev_vals_mean = np.ones(100) * np.mean(sav_filtered[40:])#((np.max(sav_filtered[40:]) - np.min(sav_filtered[40:]))/2 + np.min(sav_filtered[40:]))#
# print(prev_vals_mean)
if sav_filtered[-1] > prev_vals_mean[-1]:
exhale_val = abs(sav_filtered[-1] - prev_vals_mean[-1])
if exhale_val>0.1:
inhale_val = 0
# print('Exhaling')
exhale_val = exhale_val/(160)
client.send_message("/chan", (inhale_val, exhale_val))
# if is_exaling == True:
# sd.play(data, fs, device=1)
# print(exhale_val)
# ////////////// Sound
exhale_val = exhale_val + 0.20
if c1[0] > 10 and c1[0] < 600:
client1.send_message("/sound", (1, inhale_val, exhale_val))
else:
client1.send_message("/sound", (0,inhale_val, exhale_val))
# //////////////
else:
inhale_val = abs(prev_vals_mean[-1] - sav_filtered[-1])
if inhale_val>0.1:
exhale_val = 0
# print('Inhaling')
# ////////////Sound
inhale_val = inhale_val/(160)
if c1[0] > 10 and c1[0] < 600:
client1.send_message("/sound", (1, inhale_val, exhale_val))
else:
client1.send_message("/sound", (0, inhale_val, exhale_val))
# //////////////
client.send_message("/chan", (inhale_val, exhale_val))
plt.subplot(211)
plt.plot(sav_filtered)
plt.plot(prev_vals_mean)
plt.subplot(212)
plt.bar(['Inhale', 'exhale'], [inhale_val, exhale_val])
plt.ylim([0,1])
plt.draw()
plt.pause(0.0011)
plt.clf()
else:
prev_vals.append(obs)
t_size = cv2.getTextSize(label, cv2.FONT_HERSHEY_PLAIN, 1 , 1)[0]
c2 = c1[0] + t_size[0] + 3, c1[1] + t_size[1] + 4
return img
def arg_parse():
parser = argparse.ArgumentParser(description='YOLO v3 Video Detection Module')
parser.add_argument("--video", dest = 'video', help =
"Video to run detection upon",
default = "video.avi", type = str)
parser.add_argument("--dataset", dest = "dataset", help = "Dataset on which the network has been trained", default = "pascal")
parser.add_argument("--confidence", dest = "confidence", help = "Object Confidence to filter predictions", default = 0.5)
parser.add_argument("--nms_thresh", dest = "nms_thresh", help = "NMS Threshhold", default = 0.4)
# parser.add_argument("--cfg", dest = 'cfgfile', help =
# "Config file",
# default = "cfg/yolov3_pysource_testing.cfg", type = str)#cfg/yolov3.cfg yolov3_pysource_testing.cfg
# parser.add_argument("--weights", dest = 'weightsfile', help =
# "weightsfile",
# default = "data/yolov3_training_final_17_10_2020.weights", type = str)#yolov3.weights yolov3-wider_face_16000 yolov3_training_final_17_10_2020.weights
parser.add_argument("--reso", dest = 'reso', help =
"Input resolution of the network. Increase to increase accuracy. Decrease to increase speed",
default = "416", type = str)
return parser.parse_args()
val_reset = False
def osc_command_receiver(unused_addr):
global val_reset
val_reset = True
print('Reset')
if __name__ == '__main__':
args = arg_parse()
confidence = float(args.confidence)
nms_thesh = float(args.nms_thresh)
start = 0
CUDA = torch.cuda.is_available()
num_classes = 1
CUDA = torch.cuda.is_available()
bbox_attrs = 5 + num_classes
# print("Loading Face Landmark detector.....")
print("Loading network.....")
model = Darknet(cfg_file)#args.cfgfile)
model.load_weights(weight_file)#args.weightsfile)
print("Network successfully loaded")
print(CUDA)
model.net_info["height"] = args.reso
inp_dim = int(model.net_info["height"])
assert inp_dim % 32 == 0
assert inp_dim > 32
if CUDA:
model.cuda()
print("x")
model(get_test_input(inp_dim, CUDA), CUDA)
model.eval()
videofile = args.video
# cap = cv2.VideoCapture("vids/breathing_cap.mov")#'http://192.168.21.200:81/')
# cap = cv2.VideoCapture(0)
camera = Boson()
# assert cap.isOpened(), 'Cannot capture source'
frames = 0
start = time.time()
dispatcher = dispatcher.Dispatcher()
dispatcher.map("/python/reset", osc_command_receiver)
server = osc_server.ThreadingOSCUDPServer(('192.168.21.87', 12000), dispatcher)
thread = threading.Thread(target=server.serve_forever)
thread.start()
while True:
image = camera.grab()
img = (image - np.min(image)) / (np.max(image) - np.min(image)) * 255
img = img.astype(np.uintc)
frame = np.uint8(img)
frame= cv2.cvtColor(frame, cv2.COLOR_GRAY2BGR)
if len(image):
img, orig_im, dim = prep_image(frame, inp_dim)
im_dim = torch.FloatTensor(dim).repeat(1,2)
if CUDA:
im_dim = im_dim.cuda()
img = img.cuda()
with torch.no_grad():
output = model(Variable(img), CUDA)
output = write_results(output, confidence, num_classes, nms = True, nms_conf = nms_thesh)
if type(output) == int:
cv2.imshow("frame", orig_im)
key = cv2.waitKey(1)
if key & 0xFF == ord('q'):
break
continue
im_dim = im_dim.repeat(output.size(0), 1)
scaling_factor = torch.min(inp_dim/im_dim,1)[0].view(-1,1)
output[:,[1,3]] -= (inp_dim - scaling_factor*im_dim[:,0].view(-1,1))/2
output[:,[2,4]] -= (inp_dim - scaling_factor*im_dim[:,1].view(-1,1))/2
output[:,1:5] /= scaling_factor
for i in range(output.shape[0]):
output[i, [1,3]] = torch.clamp(output[i, [1,3]], 0.0, im_dim[i,0])
output[i, [2,4]] = torch.clamp(output[i, [2,4]], 0.0, im_dim[i,1])
classes = load_classes('data/coco.names')
colors = pkl.load(open("data/pallete", "rb"))
list(map(lambda x: write(x, orig_im,image), output))
cv2.imshow("frame", orig_im)
key = cv2.waitKey(1)
if key & 0xFF == ord('r') or val_reset == True:
print("______Resetting Breathing Values_______")
prev_vals.clear()
val_reset = False
plt.draw()
plt.pause(0.0011)
plt.clf()
if key & 0xFF == ord('q'):
print("Closing Down the program")
camera.close()
plt.close()
cv2.destroyAllWindows()
break
frames += 1
if frames % 100 == 0:
print("FPS of the video is {:5.2f}".format( frames / (time.time() - start)))
else:
break