mirror of
https://github.com/EyeTrackVR/EyeTrackVR.git
synced 2025-11-04 14:39:42 +08:00
Delete RANSAC3d directory
This commit is contained in:
parent
12d60826ff
commit
eed5288b5e
@ -1,569 +0,0 @@
|
||||
from threading import Thread
|
||||
import cv2
|
||||
import numpy as np
|
||||
from pye3dcustom.detector_3d import CameraModel, Detector3D, DetectorMode
|
||||
from pythonosc import udp_client
|
||||
import time
|
||||
import os
|
||||
from scipy import ndimage
|
||||
import pyttsx3
|
||||
import PIL
|
||||
engine = pyttsx3.init()
|
||||
|
||||
##################
|
||||
|
||||
## User Configuration variables
|
||||
## (Things prohurtz may tell you to change)
|
||||
|
||||
# Change to a larger number (i.e. 60) to increase search area
|
||||
focal_length = 30
|
||||
# Arrived at this value by experimentation, may need tweaking depending on avatar eye model.
|
||||
eye_position_scalar = 3000
|
||||
|
||||
##################
|
||||
|
||||
## OSC Configuration
|
||||
|
||||
OSCip="127.0.0.1"
|
||||
OSCport=9000 #VR Chat OSC port
|
||||
client = udp_client.SimpleUDPClient(OSCip, OSCport)
|
||||
|
||||
##################
|
||||
|
||||
def vc():
|
||||
vc.height = 1
|
||||
vc.width = 1
|
||||
vc.roicheck = 1
|
||||
|
||||
vc.xoff = 1
|
||||
vc.yoff = 1
|
||||
vc.eyeoffset = 300
|
||||
vc.eyeoffx = 1
|
||||
vc.setoff = 0
|
||||
vc.x = 1
|
||||
vc.y = 1
|
||||
vc.w = 1
|
||||
vc.h = 1
|
||||
|
||||
vc.xmax = 69420
|
||||
vc.xmin = -69420
|
||||
vc.ymax = 69420
|
||||
vc.ymin = -69420
|
||||
|
||||
vc()
|
||||
|
||||
|
||||
|
||||
|
||||
try:
|
||||
with open("camaddr.cfg", 'r') as camr:
|
||||
lines = camr.readlines()
|
||||
camaddrport = lines[0].strip()
|
||||
camr.close()
|
||||
|
||||
|
||||
except:
|
||||
camaddr = input('Camera Address or Port:> ')
|
||||
ipdetect = ".."
|
||||
|
||||
if ipdetect in camaddr:
|
||||
camaddrport = f"http://{camaddr}:81/stream"
|
||||
else:
|
||||
camaddrport = camaddr
|
||||
|
||||
with open('camaddr.cfg', 'w+') as cam:
|
||||
cam.write(str(camaddr))
|
||||
|
||||
try:
|
||||
with open("settings.cfg", 'r') as camr:
|
||||
lines = camr.readlines()
|
||||
thrsh = int(lines[0].strip())
|
||||
rv = int(lines[1].strip())
|
||||
|
||||
except:
|
||||
print('[ERROR] No Threshold or Rotation Set, Run GUI.')
|
||||
thrsh = input('Threshold:> ')
|
||||
rv = 0
|
||||
with open('settings.cfg', 'w+') as rf:
|
||||
rf.write(str(thrsh))
|
||||
rf.write('\n')
|
||||
rf.write(str(0))
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
try:
|
||||
with open("roi.cfg", 'r+') as roicnfg:
|
||||
lines = roicnfg.readlines()
|
||||
vc.x = float(lines[0].strip())
|
||||
vc.y = float(lines[1].strip())
|
||||
vc.w = float(lines[2].strip())
|
||||
vc.h = float(lines[3].strip())
|
||||
roicnfg.close()
|
||||
except:
|
||||
print('[ERROR] No ROI Set.')
|
||||
cap = cv2.VideoCapture(camaddrport)
|
||||
ret, img = cap.read()
|
||||
img = ndimage.rotate(img, int(rv), reshape=True)
|
||||
roibb = cv2.selectROI("image", img, fromCenter=False, showCrosshair=True)
|
||||
cv2.destroyAllWindows()
|
||||
print('X', roibb[0])
|
||||
print('Y', roibb[1])
|
||||
print('Width', roibb[2])
|
||||
print('Height', roibb[3])
|
||||
|
||||
with open('roi.cfg', 'w+') as rf:
|
||||
rf.write(str(roibb[0]))
|
||||
rf.write('\n')
|
||||
rf.write(str(roibb[1]))
|
||||
rf.write('\n')
|
||||
rf.write(str(roibb[2]))
|
||||
rf.write('\n')
|
||||
rf.write(str(roibb[3]))
|
||||
with open("roi.cfg", 'r+') as roicnfg:
|
||||
lines = roicnfg.readlines()
|
||||
vc.x = float(lines[0].strip())
|
||||
vc.y = float(lines[1].strip())
|
||||
vc.w = float(lines[2].strip())
|
||||
vc.h = float(lines[3].strip())
|
||||
roicnfg.close()
|
||||
|
||||
try:
|
||||
cam = cv2.VideoCapture(camaddrport)
|
||||
print('[INFO] Press ESC when rotation is set.')
|
||||
while True:
|
||||
#time.sleep(0.1)############################################################## < REMOVE ####################################################
|
||||
ret_val, img = cam.read()
|
||||
with open("settings.cfg", 'r') as camr:
|
||||
lines = camr.readlines()
|
||||
thrsh = int(lines[0].strip())
|
||||
rv = int(lines[1].strip())
|
||||
img = img[int(vc.y): int(vc.y+vc.h), int(vc.x): int(float(vc.x+vc.w))]
|
||||
rows, cols, _ = img.shape
|
||||
img_center = (cols / 2, rows / 2)
|
||||
M = cv2.getRotationMatrix2D(img_center, rv, 1)
|
||||
img = cv2.warpAffine(img, M, (cols, rows),
|
||||
borderMode=cv2.BORDER_CONSTANT,
|
||||
borderValue=(255,255,255))
|
||||
#img = ndimage.rotate(img, int(rv), reshape=True)
|
||||
cv2.imshow('cam', img)
|
||||
if cv2.waitKey(1) == 27:
|
||||
break # esc to quit
|
||||
cv2.destroyAllWindows()
|
||||
except:
|
||||
print('[INFO] Rotation sucsessfully set')
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
try:
|
||||
with open("center_offset.cfg") as offr:
|
||||
lines = offr.readlines()
|
||||
vc.xoff = float(lines[0].strip())
|
||||
vc.yoff = float(lines[1].strip())
|
||||
|
||||
offr.close()
|
||||
except:
|
||||
print('[WARN] No eye offset has been detected.')
|
||||
engine.say("No eye offset has been detected please move your eye around and wait for audio prompt.")
|
||||
|
||||
engine.runAndWait()
|
||||
vc.eyeoffset = 300
|
||||
vc.setoff = 1
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
def fit_rotated_ellipse_ransac(
|
||||
data, iter=80, sample_num=10, offset=80 # 80.0, 10, 80
|
||||
): # before changing these values, please read up on the ransac algorithm
|
||||
# However if you want to change any value just know that higher iterations will make processing frames slower
|
||||
count_max = 0
|
||||
effective_sample = None
|
||||
|
||||
for i in range(iter):
|
||||
sample = np.random.choice(len(data), sample_num, replace=False)
|
||||
|
||||
xs = data[sample][:, 0].reshape(-1, 1)
|
||||
ys = data[sample][:, 1].reshape(-1, 1)
|
||||
|
||||
J = np.mat(
|
||||
np.hstack((xs * ys, ys**2, xs, ys, np.ones_like(xs, dtype=np.float)))
|
||||
)
|
||||
Y = np.mat(-1 * xs**2)
|
||||
P = (J.T * J).I * J.T * Y
|
||||
|
||||
# fitter a*x**2 + b*x*y + c*y**2 + d*x + e*y + f = 0
|
||||
a = 1.0
|
||||
b = P[0, 0]
|
||||
c = P[1, 0]
|
||||
d = P[2, 0]
|
||||
e = P[3, 0]
|
||||
f = P[4, 0]
|
||||
ellipse_model = (
|
||||
lambda x, y: a * x**2 + b * x * y + c * y**2 + d * x + e * y + f
|
||||
)
|
||||
|
||||
# thresh
|
||||
ran_sample = np.array(
|
||||
[[x, y] for (x, y) in data if np.abs(ellipse_model(x, y)) < offset]
|
||||
)
|
||||
|
||||
if len(ran_sample) > count_max:
|
||||
count_max = len(ran_sample)
|
||||
effective_sample = ran_sample
|
||||
|
||||
return fit_rotated_ellipse(effective_sample)
|
||||
|
||||
|
||||
def fit_rotated_ellipse(data):
|
||||
|
||||
xs = data[:, 0].reshape(-1, 1)
|
||||
ys = data[:, 1].reshape(-1, 1)
|
||||
|
||||
J = np.mat(np.hstack((xs * ys, ys**2, xs, ys, np.ones_like(xs, dtype=np.float))))
|
||||
Y = np.mat(-1 * xs**2)
|
||||
P = (J.T * J).I * J.T * Y
|
||||
|
||||
a = 1.0
|
||||
b = P[0, 0]
|
||||
c = P[1, 0]
|
||||
d = P[2, 0]
|
||||
e = P[3, 0]
|
||||
f = P[4, 0]
|
||||
theta = 0.5 * np.arctan(b / (a - c))
|
||||
|
||||
cx = (2 * c * d - b * e) / (b**2 - 4 * a * c)
|
||||
cy = (2 * a * e - b * d) / (b**2 - 4 * a * c)
|
||||
|
||||
cu = a * cx**2 + b * cx * cy + c * cy**2 - f
|
||||
w = np.sqrt(
|
||||
cu
|
||||
/ (
|
||||
a * np.cos(theta) ** 2
|
||||
+ b * np.cos(theta) * np.sin(theta)
|
||||
+ c * np.sin(theta) ** 2
|
||||
)
|
||||
)
|
||||
h = np.sqrt(
|
||||
cu
|
||||
/ (
|
||||
a * np.sin(theta) ** 2
|
||||
- b * np.cos(theta) * np.sin(theta)
|
||||
+ c * np.cos(theta) ** 2
|
||||
)
|
||||
)
|
||||
|
||||
ellipse_model = lambda x, y: a * x**2 + b * x * y + c * y**2 + d * x + e * y + f
|
||||
|
||||
error_sum = np.sum([ellipse_model(x, y) for x, y in data])
|
||||
|
||||
return (cx, cy, w, h, theta)
|
||||
|
||||
|
||||
|
||||
|
||||
cap = cv2.VideoCapture(camaddrport) # change this to the video you want to test
|
||||
result_2d = {}
|
||||
result_2d_final = {}
|
||||
|
||||
ret, img = cap.read()
|
||||
frame_number = cap.get(cv2.CAP_PROP_POS_FRAMES)
|
||||
fps = cap.get(cv2.CAP_PROP_FPS)
|
||||
width = cap.get(cv2.CAP_PROP_FRAME_WIDTH)
|
||||
height = cap.get(cv2.CAP_PROP_FRAME_HEIGHT)
|
||||
|
||||
#print(cv2.selectROI("image", img, fromCenter=False, showCrosshair=True))
|
||||
vc.width = vc.w
|
||||
vc.height = vc.h
|
||||
|
||||
camera = CameraModel(focal_length=focal_length, resolution=[vc.width,vc.height])
|
||||
|
||||
detector_3d = Detector3D(camera=camera, long_term_mode=DetectorMode.blocking)
|
||||
|
||||
if cap.isOpened() == False:
|
||||
print("Error opening video stream or file")
|
||||
while cap.isOpened():
|
||||
|
||||
|
||||
|
||||
try:
|
||||
rvo = rv
|
||||
with open("settings.cfg", 'r') as camr:
|
||||
lines = camr.readlines()
|
||||
thrsh = int(lines[0].strip())
|
||||
rv = int(lines[1].strip())
|
||||
if rv != rvo:
|
||||
print("[WARN] Rotation Detected. Center Calibration will start in a few seconds.")
|
||||
vc.eyeoffset = 300
|
||||
except:
|
||||
print('[WARN] Config Over-read Detected.')
|
||||
#try:
|
||||
try:
|
||||
ret, img = cap.read()
|
||||
img = img[int(vc.y): int(vc.y+vc.h), int(vc.x): int(float(vc.x+vc.w))]
|
||||
|
||||
|
||||
except:
|
||||
img = imgo[int(vc.y): int(vc.y+vc.h), int(vc.x): int(float(vc.x+vc.w))]
|
||||
print('[SEVERE WARN] Frame Issue Detected.')
|
||||
#print(cv2.selectROI("image", img, fromCenter=False, showCrosshair=True))
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
frame_number = cap.get(cv2.CAP_PROP_POS_FRAMES)
|
||||
fps = cap.get(cv2.CAP_PROP_FPS)
|
||||
if ret == True:
|
||||
rows, cols, _ = img.shape
|
||||
img_center = (cols / 2, rows / 2)
|
||||
M = cv2.getRotationMatrix2D(img_center, rv, 1)
|
||||
img = cv2.warpAffine(img, M, (cols, rows),
|
||||
borderMode=cv2.BORDER_CONSTANT,
|
||||
borderValue=(255,255,255))
|
||||
newImage2 = img.copy()
|
||||
kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (3, 3))
|
||||
image_gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
|
||||
ret, backupthresh = cv2.threshold(
|
||||
image_gray, (int(thrsh) + 5) , 255, cv2.THRESH_BINARY
|
||||
)
|
||||
ret, thresh = cv2.threshold(
|
||||
image_gray, int(thrsh), 255, cv2.THRESH_BINARY
|
||||
) # this will need to be adjusted everytime hardwere is changed (brightness of IR, Camera postion, etc)
|
||||
opening = cv2.morphologyEx(thresh, cv2.MORPH_OPEN, kernel)
|
||||
closing = cv2.morphologyEx(opening, cv2.MORPH_CLOSE, kernel)
|
||||
image = 255 - closing
|
||||
contours, hierarchy = cv2.findContours(
|
||||
image, cv2.RETR_TREE, cv2.CHAIN_APPROX_NONE
|
||||
)
|
||||
hull = []
|
||||
for i in range(len(contours)):
|
||||
hull.append(cv2.convexHull(contours[i], False))
|
||||
try:
|
||||
cv2.drawContours(img, contours, -1, (255, 0, 0), 1)
|
||||
cnt = sorted(hull, key=cv2.contourArea)
|
||||
maxcnt = cnt[-1]
|
||||
ellipse = cv2.fitEllipse(maxcnt)
|
||||
cx, cy, w, h, theta = fit_rotated_ellipse_ransac(maxcnt.reshape(-1, 2))
|
||||
#get axis and angle of ellipse pupil labs 2d
|
||||
result_2d["center"] = (cx, cy)
|
||||
result_2d["axes"] = (w, h)
|
||||
result_2d["angle"] = theta * 180.0 / np.pi
|
||||
result_2d_final["ellipse"] = result_2d
|
||||
result_2d_final["diameter"] = w
|
||||
result_2d_final["location"] = (cx, cy)
|
||||
result_2d_final["confidence"] = 0.99
|
||||
result_2d_final["timestamp"] = frame_number / fps
|
||||
result_3d = detector_3d.update_and_detect(result_2d_final, image_gray)
|
||||
ellipse_3d = result_3d["ellipse"]
|
||||
|
||||
# draw pupil
|
||||
cv2.ellipse(
|
||||
image_gray,
|
||||
tuple(int(v) for v in ellipse_3d["center"]),
|
||||
tuple(int(v) for v in ellipse_3d["axes"]),
|
||||
ellipse_3d["angle"],
|
||||
0,
|
||||
360, # start/end angle for drawing
|
||||
(0, 255, 0), # color (BGR): red
|
||||
)
|
||||
projected_sphere = result_3d["projected_sphere"]
|
||||
|
||||
# draw eyeball
|
||||
#image_gray1 = img
|
||||
cv2.ellipse(
|
||||
image_gray,
|
||||
tuple(int(v) for v in projected_sphere["center"]),
|
||||
tuple(int(v) for v in projected_sphere["axes"]),
|
||||
projected_sphere["angle"],
|
||||
0,
|
||||
360, # start/end angle for drawing
|
||||
(0, 255, 0), # color (BGR): red
|
||||
)
|
||||
|
||||
# draw line from center of eyeball to center of pupil
|
||||
cv2.line(
|
||||
image_gray,
|
||||
tuple(int(v) for v in projected_sphere["center"]),
|
||||
tuple(int(v) for v in ellipse_3d["center"]),
|
||||
(0, 255, 0), # color (BGR): red
|
||||
)
|
||||
|
||||
exm = ellipse_3d["center"][0]
|
||||
eym = ellipse_3d["center"][1]
|
||||
|
||||
xrl = (cx - projected_sphere["center"][0]) / projected_sphere["axes"][0]
|
||||
|
||||
eyey = (cy - projected_sphere["center"][1]) / projected_sphere["axes"][1]
|
||||
|
||||
if vc.eyeoffset == 1 and vc.setoff == 1:
|
||||
engine.say("Eye offset not found look straight forward")
|
||||
engine.runAndWait()
|
||||
time.sleep(3)
|
||||
|
||||
|
||||
|
||||
if vc.eyeoffset == 0:
|
||||
vc.eyeoffset = vc.eyeoffset - 1
|
||||
vc.xoff = ellipse_3d["center"][0]
|
||||
vc.yoff = ellipse_3d["center"][1]
|
||||
|
||||
|
||||
with open('center_offset.cfg', 'w+') as rf:
|
||||
rf.write(str(vc.xoff))
|
||||
rf.write('\n')
|
||||
rf.write(str(vc.yoff))
|
||||
engine.say("Eye offset has been set.")
|
||||
engine.runAndWait()
|
||||
|
||||
else:
|
||||
if vc.eyeoffset > 0:
|
||||
|
||||
|
||||
if exm > vc.xmax:
|
||||
vc.xmax = exm
|
||||
if exm < vc.xmin:
|
||||
vc.xmin = exm
|
||||
|
||||
if eym > vc.ymax:
|
||||
vc.ymax = eym
|
||||
if eym < vc.xmin:
|
||||
vc.ymin = eym
|
||||
|
||||
|
||||
vc.eyeoffset = vc.eyeoffset - 1
|
||||
|
||||
|
||||
|
||||
|
||||
xl = float(((cx - vc.xoff) * eye_position_scalar) / (vc.xmax - vc.xoff))
|
||||
|
||||
xr = float(((cx - vc.xoff) * eye_position_scalar) / (vc.xmin - vc.xoff))
|
||||
|
||||
|
||||
yu = float(((cy - vc.yoff) * eye_position_scalar) / (vc.ymin - vc.yoff))
|
||||
|
||||
yd = float(((cy - vc.yoff) * eye_position_scalar) / (vc.ymax - vc.yoff))
|
||||
|
||||
|
||||
|
||||
if xr > 0:
|
||||
if xr > 1:
|
||||
xr = 1.0
|
||||
client.send_message("/avatar/parameters/RightEyeX", xr)
|
||||
client.send_message("/avatar/parameters/LeftEyeX", xr)
|
||||
|
||||
print('XR', xr)
|
||||
if xl > 0:
|
||||
if xl > 1:
|
||||
xl = 1.0
|
||||
client.send_message("/avatar/parameters/RightEyeX", -abs(xl))
|
||||
client.send_message("/avatar/parameters/LeftEyeX", -abs(xl))
|
||||
print('XL', xl)
|
||||
|
||||
if yd > 0:
|
||||
if yd > 1:
|
||||
yd = 1.0
|
||||
client.send_message("/avatar/parameters/EyesY", -abs(yd))
|
||||
# print('YD', yd)
|
||||
|
||||
if yu > 0:
|
||||
if yu > 1:
|
||||
yu = 1.0
|
||||
|
||||
client.send_message("/avatar/parameters/EyesY", yu)
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
# if xrl >= 0:
|
||||
# client.send_message("/avatar/parameters/RightEyeX", -abs(xrl))
|
||||
# client.send_message("/avatar/parameters/LeftEyeX", -abs(xrl))
|
||||
# if xrl <= 0:
|
||||
# client.send_message("/avatar/parameters/RightEyeX", abs(xrl))
|
||||
# client.send_message("/avatar/parameters/LeftEyeX", abs(xrl))
|
||||
# client.send_message("/avatar/parameters/RightEyeX", abs(xrl - vc.eyeoffx))
|
||||
#client.send_message("/avatar/parameters/LeftEyeX", abs(xrl - vc.eyeoffx))
|
||||
|
||||
# if eyey >= 0:
|
||||
# client.send_message("/avatar/parameters/EyesY", -abs(eyey))
|
||||
# client.send_message("/avatar/parameters/EyesY", -abs(eyey))
|
||||
|
||||
# if eyey <= 0:
|
||||
# client.send_message("/avatar/parameters/EyesY", abs(eyey))
|
||||
# client.send_message("/avatar/parameters/EyesY", abs(eyey))
|
||||
|
||||
# client.send_message("/avatar/parameters/LeftEyeLid", float(1))
|
||||
# client.send_message("/avatar/parameters/RightEyeLid", float(1))
|
||||
|
||||
# ellipse_3d["center"][0] x mid?
|
||||
# ellipse_3d["center"][1] y mid
|
||||
|
||||
|
||||
|
||||
|
||||
except:
|
||||
try:
|
||||
contours, _ = cv2.findContours(backupthresh, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE)
|
||||
contours = sorted(contours, key=lambda x: cv2.contourArea(x), reverse=True)
|
||||
rows, cols, _ = img.shape
|
||||
for cnt in contours:
|
||||
|
||||
(x, y, w, h) = cv2.boundingRect(cnt)
|
||||
xt = x + int(w/2)
|
||||
yt = y + int(h/2)
|
||||
if h > 8 and w > 8 and h < 30 and w < 30:
|
||||
|
||||
|
||||
xrlb = (xt - projected_sphere["center"][0]) / projected_sphere["axes"][0]
|
||||
eyeyb = (yt - projected_sphere["center"][1]) / projected_sphere["axes"][1]
|
||||
print(xrlb, eyeyb)
|
||||
cv2.line(image_gray, (x + int(w/2), 0), (x + int(w/2), rows), (255, 0, 0), 1) #visualizes eyetracking on thresh
|
||||
cv2.line(image_gray, (0, y + int(h/2)), (cols, y + int(h/2)), (255, 0, 0), 1)
|
||||
cv2.drawContours(image_gray, [cnt], -1, (255, 0, 0), 3)
|
||||
cv2.rectangle(image_gray, (x, y), (x + w, y + h), (255, 0, 0), 2)
|
||||
|
||||
if xrlb >= 0:
|
||||
client.send_message("/avatar/parameters/RightEyeX", -abs(xrl))
|
||||
client.send_message("/avatar/parameters/LeftEyeX", -abs(xrl))
|
||||
if eyeyb <= 0:
|
||||
client.send_message("/avatar/parameters/RightEyeX", -abs(xrl))
|
||||
client.send_message("/avatar/parameters/LeftEyeX", -abs(xrl))
|
||||
|
||||
else:
|
||||
client.send_message("/avatar/parameters/LeftEyeLid", float(0))
|
||||
client.send_message("/avatar/parameters/RightEyeLid", float(0))
|
||||
print('[INFO] BLINK Detected.')
|
||||
|
||||
except:
|
||||
print('[ERROR] Backup Tracking Failed.')
|
||||
#pass
|
||||
imgo = img
|
||||
|
||||
cv2.imshow("Ransac", image_gray)
|
||||
cv2.imshow("Original", thresh)
|
||||
cv2.imshow("eye", img)
|
||||
cv2.imshow("backupthresh", backupthresh)
|
||||
if cv2.waitKey(1) & 0xFF == ord("q"):
|
||||
cv2.destroyAllWindows()
|
||||
break
|
||||
#except:
|
||||
|
||||
# print('E1')
|
||||
@ -1,43 +0,0 @@
|
||||
|
||||
# use the app insted
|
||||
|
||||
## Installation & Setup
|
||||
#
|
||||
|
||||
Make sure you have Python 3.6.x installed as that is the only compatible version of Python. Be sure to uninstall all other versions or correctly setup a virtual enviroment.
|
||||
|
||||
Open the folder in a command prompt and exicute the command: `pip install -r requirements.txt`
|
||||
|
||||
Once everything has installed run `RANSAC3d.py`
|
||||
|
||||
You will need to enter the IP address of your camera or the port (if using wired cam)
|
||||
|
||||
Next you will need to set a threshold. A good starting point is 80-110.
|
||||
|
||||
After that a window will appear where you can set a ROI (Region of Interest)
|
||||
You will need to Draw a rectangle with your mouse that selects just the area where your eye is in.
|
||||
|
||||
Once drawn press enter.
|
||||
|
||||
Once tracking has started, run `threshGUI.py`
|
||||
|
||||
To adjust the threshold, move the slider.
|
||||
|
||||
Increase the number if a pupil is not detected.
|
||||
|
||||
Decrease the number if the pupil is too big.
|
||||
|
||||
Find the highest you can go with the best quality and close the program.
|
||||
|
||||
|
||||
|
||||
#
|
||||
### The main tracking part is adapted from https://github.com/SummerSigh/TheVrMLEyeToolbox/tree/main/Pupil3dDetector
|
||||
|
||||
*I DO NOT OWN THE LICENSE TO THE PUPIL LABS PART OF THE CODE. Distributed under the terms of the GNU Lesser General Public License (LGPL v3.0). See COPYING and COPYING. LESSER for license details.*
|
||||
|
||||
Copyright (C) 2018 Pupil Labs
|
||||
|
||||
All Rights Reserved.
|
||||
|
||||
This is the Pye3d system by Pupil Labs
|
||||
@ -1,33 +0,0 @@
|
||||
import cv2
|
||||
|
||||
|
||||
try:
|
||||
with open("camaddr.cfg", 'r') as camr:
|
||||
lines = camr.readlines()
|
||||
camaddr = lines[0].strip()
|
||||
camr.close()
|
||||
|
||||
|
||||
except:
|
||||
camaddr = input('Camera Address:> ')
|
||||
with open('camaddr.cfg', 'w+') as cam:
|
||||
cam.write(str(camaddr))
|
||||
|
||||
|
||||
cap = cv2.VideoCapture(camaddr)
|
||||
ret, img = cap.read()
|
||||
roibb = cv2.selectROI("image", img, fromCenter=False, showCrosshair=True)
|
||||
|
||||
print('X', roibb[0])
|
||||
print('Y', roibb[1])
|
||||
print('Width', roibb[2])
|
||||
print('Height', roibb[3])
|
||||
|
||||
with open('roi.cfg', 'w+') as rf:
|
||||
rf.write(str(roibb[0]))
|
||||
rf.write('\n')
|
||||
rf.write(str(roibb[1]))
|
||||
rf.write('\n')
|
||||
rf.write(str(roibb[2]))
|
||||
rf.write('\n')
|
||||
rf.write(str(roibb[3]))
|
||||
@ -1 +0,0 @@
|
||||
vcruntime140_1.dll
|
||||
Binary file not shown.
@ -1,34 +0,0 @@
|
||||
|
||||
|
||||
"""""" # start delvewheel patch
|
||||
def _delvewheel_init_patch_0_0_15():
|
||||
import os
|
||||
import sys
|
||||
libs_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir, 'pye3d.libs'))
|
||||
if sys.version_info[:2] >= (3, 8):
|
||||
if os.path.exists(os.path.join(sys.base_prefix, 'conda-meta')):
|
||||
# backup the state of the environment variable CONDA_DLL_SEARCH_MODIFICATION_ENABLE
|
||||
conda_dll_search_modification_enable = os.environ.get("CONDA_DLL_SEARCH_MODIFICATION_ENABLE")
|
||||
os.environ['CONDA_DLL_SEARCH_MODIFICATION_ENABLE']='1'
|
||||
|
||||
os.add_dll_directory(libs_dir)
|
||||
|
||||
if os.path.exists(os.path.join(sys.base_prefix, 'conda-meta')):
|
||||
# restore the state of the environment variable CONDA_DLL_SEARCH_MODIFICATION_ENABLE
|
||||
if conda_dll_search_modification_enable is None:
|
||||
os.environ.pop("CONDA_DLL_SEARCH_MODIFICATION_ENABLE", None)
|
||||
else:
|
||||
os.environ["CONDA_DLL_SEARCH_MODIFICATION_ENABLE"] = conda_dll_search_modification_enable
|
||||
else:
|
||||
from ctypes import WinDLL
|
||||
with open(os.path.join(libs_dir, '.load-order-pye3d-0.3.0')) as file:
|
||||
load_order = file.read().split()
|
||||
for lib in load_order:
|
||||
WinDLL(os.path.join(libs_dir, lib))
|
||||
|
||||
|
||||
_delvewheel_init_patch_0_0_15()
|
||||
del _delvewheel_init_patch_0_0_15
|
||||
# end delvewheel patch
|
||||
|
||||
__version__ = "0.3.0"
|
||||
@ -1,6 +0,0 @@
|
||||
from typing import Tuple, NamedTuple
|
||||
|
||||
|
||||
class CameraModel(NamedTuple):
|
||||
focal_length: float
|
||||
resolution: Tuple[float, float]
|
||||
@ -1,4 +0,0 @@
|
||||
import typing as T
|
||||
|
||||
_EYE_RADIUS_DEFAULT: float = 10.392304845413264
|
||||
DEFAULT_SPHERE_CENTER: T.Tuple[float, float, float] = (0.0, 0.0, 35.0)
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
@ -1,725 +0,0 @@
|
||||
"""
|
||||
(*)~---------------------------------------------------------------------------
|
||||
Pupil - eye tracking platform
|
||||
Copyright (C) 2012-2019 Pupil Labs
|
||||
|
||||
Distributed under the terms of the GNU
|
||||
Lesser General Public License (LGPL v3.0).
|
||||
See COPYING and COPYING.LESSER for license details.
|
||||
---------------------------------------------------------------------------~(*)
|
||||
"""
|
||||
import enum
|
||||
import logging
|
||||
import traceback
|
||||
from typing import Dict, NamedTuple, Type
|
||||
|
||||
import numpy as np
|
||||
import cv2 # Todo: DELETE
|
||||
from .geometry.projections import (
|
||||
unproject_edges_to_sphere,
|
||||
project_point_into_image_plane,
|
||||
) # Todo: DELETE
|
||||
|
||||
from .camera import CameraModel
|
||||
from .constants import _EYE_RADIUS_DEFAULT
|
||||
from .cpp.pupil_detection_3d import get_edges
|
||||
from .cpp.pupil_detection_3d import search_on_sphere as search_on_sphere
|
||||
from .geometry.primitives import Circle, Ellipse, Sphere
|
||||
from .geometry.projections import (
|
||||
project_circle_into_image_plane,
|
||||
project_sphere_into_image_plane,
|
||||
)
|
||||
from .geometry.utilities import cart2sph, sph2cart
|
||||
from .kalman import KalmanFilter
|
||||
from .observation import (
|
||||
BinBufferedObservationStorage,
|
||||
BufferedObservationStorage,
|
||||
Observation,
|
||||
)
|
||||
from .eye_model import (
|
||||
SphereCenterEstimates,
|
||||
TwoSphereModelAbstract,
|
||||
TwoSphereModel,
|
||||
TwoSphereModelAsync,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class DetectorMode(enum.Enum):
|
||||
blocking = TwoSphereModel
|
||||
asynchronous = TwoSphereModelAsync
|
||||
|
||||
@classmethod
|
||||
def from_name(cls, mode_name: str):
|
||||
return {mode.name: mode for mode in cls}[mode_name]
|
||||
|
||||
|
||||
def ellipse2dict(ellipse: Ellipse) -> Dict:
|
||||
return {
|
||||
"center": (
|
||||
ellipse.center[0],
|
||||
ellipse.center[1],
|
||||
),
|
||||
"axes": (
|
||||
ellipse.minor_radius,
|
||||
ellipse.major_radius,
|
||||
),
|
||||
"angle": ellipse.angle,
|
||||
}
|
||||
|
||||
|
||||
def circle2dict(circle: Circle) -> Dict:
|
||||
return {
|
||||
"center": (
|
||||
circle.center[0],
|
||||
circle.center[1],
|
||||
circle.center[2],
|
||||
),
|
||||
"normal": (
|
||||
circle.normal[0],
|
||||
circle.normal[1],
|
||||
circle.normal[2],
|
||||
),
|
||||
"radius": float(circle.radius),
|
||||
}
|
||||
|
||||
|
||||
class Prediction(NamedTuple):
|
||||
sphere_center: np.ndarray
|
||||
pupil_circle: Circle
|
||||
|
||||
|
||||
class Search3DResult(NamedTuple):
|
||||
circle: Circle
|
||||
confidence: float
|
||||
|
||||
|
||||
def sigmoid(x, baseline=0.1, amplitude=500.0, center=0.99, width=0.02):
|
||||
return baseline + amplitude * 1.0 / (1.0 + np.exp(-(x - center) / width))
|
||||
|
||||
|
||||
class Detector3D(object):
|
||||
def __init__(
|
||||
self,
|
||||
camera: CameraModel,
|
||||
threshold_swirski=0.7,
|
||||
threshold_kalman=0.98,
|
||||
threshold_short_term=0.8,
|
||||
threshold_long_term=0.98,
|
||||
long_term_buffer_size=30,
|
||||
long_term_forget_time=5,
|
||||
long_term_forget_observations=300,
|
||||
long_term_mode: DetectorMode = DetectorMode.blocking,
|
||||
model_update_interval_long_term=1.0,
|
||||
model_update_interval_ult_long_term=10.0,
|
||||
model_warmup_duration=5.0,
|
||||
calculate_rms_residual=False,
|
||||
):
|
||||
self._camera = camera
|
||||
self._long_term_mode = long_term_mode
|
||||
self._calculate_rms_residual = calculate_rms_residual
|
||||
# NOTE: changing settings after intialization can lead to inconsistent behavior
|
||||
# if .reset() is not called.
|
||||
self._settings = {
|
||||
"threshold_swirski": threshold_swirski,
|
||||
"threshold_kalman": threshold_kalman,
|
||||
"threshold_short_term": threshold_short_term,
|
||||
"threshold_long_term": threshold_long_term,
|
||||
"long_term_buffer_size": long_term_buffer_size,
|
||||
"long_term_forget_time": long_term_forget_time,
|
||||
"long_term_forget_observations": long_term_forget_observations,
|
||||
"model_update_interval_long_term": model_update_interval_long_term,
|
||||
"model_update_interval_ult_long_term": model_update_interval_ult_long_term,
|
||||
"model_warmup_duration": model_warmup_duration,
|
||||
}
|
||||
self.reset()
|
||||
logger.debug(
|
||||
f"{type(self)} initialized with "
|
||||
f"long_term_mode={long_term_mode} "
|
||||
f"calculate_rms_residual={calculate_rms_residual} "
|
||||
f"settings={self._settings}"
|
||||
)
|
||||
|
||||
@property
|
||||
def camera(self) -> CameraModel:
|
||||
return self._camera
|
||||
|
||||
@property
|
||||
def long_term_mode(self) -> DetectorMode:
|
||||
return self._long_term_mode
|
||||
|
||||
@long_term_mode.setter
|
||||
def long_term_mode(self, mode: DetectorMode):
|
||||
needs_reset = mode != self._long_term_mode
|
||||
self._long_term_mode = mode
|
||||
if needs_reset:
|
||||
self.reset()
|
||||
|
||||
@property
|
||||
def is_long_term_model_frozen(self) -> bool:
|
||||
# If _ult_long_term_schedule is paused or not does not actually matter. The
|
||||
# _ult_long_term_model is only used for fitting the _long_term_model. If the
|
||||
# _long_term_schedule is paused, the _long_term_model is not being fitted and
|
||||
# therefore the state of _ult_long_term_model will be ignored.
|
||||
return self._long_term_schedule.is_paused
|
||||
|
||||
@is_long_term_model_frozen.setter
|
||||
def is_long_term_model_frozen(self, should_be_frozen: bool) -> None:
|
||||
# We pause/resume _ult_long_term_schedule here as well to save CPU resources
|
||||
# while the _long_term_model is frozen.
|
||||
if should_be_frozen:
|
||||
self._long_term_schedule.pause()
|
||||
self._ult_long_term_schedule.pause()
|
||||
else:
|
||||
self._long_term_schedule.resume()
|
||||
self._ult_long_term_schedule.resume()
|
||||
|
||||
def reset_camera(self, camera: CameraModel):
|
||||
"""Change camera model and reset detector state."""
|
||||
self._camera = camera
|
||||
self.reset()
|
||||
|
||||
def reset(self):
|
||||
self._cleanup_models()
|
||||
self._initialize_models(
|
||||
long_term_model_cls=self._long_term_mode.value,
|
||||
ultra_long_term_model_cls=self._long_term_mode.value,
|
||||
)
|
||||
self._long_term_schedule = _ModelUpdateSchedule(
|
||||
update_interval=self._settings["model_update_interval_long_term"],
|
||||
warmup_duration=self._settings["model_warmup_duration"],
|
||||
)
|
||||
self._ult_long_term_schedule = _ModelUpdateSchedule(
|
||||
update_interval=self._settings["model_update_interval_ult_long_term"],
|
||||
warmup_duration=self._settings["model_warmup_duration"],
|
||||
)
|
||||
|
||||
self.kalman_filter = KalmanFilter()
|
||||
|
||||
def _initialize_models(
|
||||
self,
|
||||
short_term_model_cls: Type[TwoSphereModelAbstract] = TwoSphereModel,
|
||||
long_term_model_cls: Type[TwoSphereModelAbstract] = TwoSphereModel,
|
||||
ultra_long_term_model_cls: Type[TwoSphereModelAbstract] = TwoSphereModel,
|
||||
):
|
||||
# Recreate all models. This is required in case any of the settings (incl
|
||||
# camera) changed in the meantime.
|
||||
self.short_term_model = short_term_model_cls(
|
||||
camera=self.camera,
|
||||
storage_cls=BufferedObservationStorage,
|
||||
storage_kwargs=dict(
|
||||
confidence_threshold=self._settings["threshold_short_term"],
|
||||
buffer_length=10,
|
||||
),
|
||||
)
|
||||
self.long_term_model = long_term_model_cls(
|
||||
camera=self.camera,
|
||||
storage_cls=BinBufferedObservationStorage,
|
||||
storage_kwargs=dict(
|
||||
camera=self.camera,
|
||||
confidence_threshold=self._settings["threshold_long_term"],
|
||||
n_bins_horizontal=10,
|
||||
bin_buffer_length=self._settings["long_term_buffer_size"],
|
||||
forget_min_observations=self._settings["long_term_forget_observations"],
|
||||
forget_min_time=self._settings["long_term_forget_time"],
|
||||
),
|
||||
)
|
||||
self.ultra_long_term_model = ultra_long_term_model_cls(
|
||||
camera=self.camera,
|
||||
storage_cls=BinBufferedObservationStorage,
|
||||
storage_kwargs=dict(
|
||||
camera=self.camera,
|
||||
confidence_threshold=self._settings["threshold_long_term"],
|
||||
n_bins_horizontal=10,
|
||||
bin_buffer_length=self._settings["long_term_buffer_size"],
|
||||
forget_min_observations=(
|
||||
2 * self._settings["long_term_forget_observations"]
|
||||
),
|
||||
forget_min_time=60,
|
||||
),
|
||||
)
|
||||
|
||||
def _cleanup_models(self):
|
||||
try:
|
||||
self.short_term_model.cleanup()
|
||||
self.long_term_model.cleanup()
|
||||
self.ultra_long_term_model.cleanup()
|
||||
except AttributeError:
|
||||
pass # models have not been initialized yet
|
||||
|
||||
def update_and_detect(
|
||||
self,
|
||||
pupil_datum: Dict,
|
||||
frame: np.ndarray,
|
||||
apply_refraction_correction: bool = True,
|
||||
debug: bool = False,
|
||||
):
|
||||
# update models
|
||||
observation = self._extract_observation(pupil_datum)
|
||||
self.update_models(observation)
|
||||
|
||||
# predict target variables
|
||||
sphere_center = self.long_term_model.sphere_center
|
||||
pupil_circle = self._predict_pupil_circle(observation, frame)
|
||||
prediction_uncorrected = Prediction(sphere_center, pupil_circle)
|
||||
|
||||
# apply refraction correction
|
||||
if apply_refraction_correction:
|
||||
pupil_circle = self.long_term_model.apply_refraction_correction(
|
||||
pupil_circle
|
||||
)
|
||||
sphere_center = self.long_term_model.corrected_sphere_center
|
||||
# Falls back to uncorrected version if correction is disabled
|
||||
prediction_corrected = Prediction(sphere_center, pupil_circle)
|
||||
|
||||
result = self._prepare_result(
|
||||
observation,
|
||||
prediction_uncorrected=prediction_uncorrected,
|
||||
prediction_corrected=prediction_corrected,
|
||||
)
|
||||
|
||||
if debug:
|
||||
result["debug_info"] = self._collect_debug_info()
|
||||
|
||||
return result
|
||||
|
||||
def update_models(self, observation: Observation):
|
||||
self.short_term_model.add_observation(observation)
|
||||
self.long_term_model.add_observation(observation)
|
||||
self.ultra_long_term_model.add_observation(observation)
|
||||
|
||||
if (
|
||||
self.short_term_model.n_observations <= 0
|
||||
or self.long_term_model.n_observations <= 0
|
||||
or self.ultra_long_term_model.n_observations <= 0
|
||||
):
|
||||
return
|
||||
|
||||
try:
|
||||
if self._ult_long_term_schedule.is_update_due(observation.timestamp):
|
||||
self.ultra_long_term_model.estimate_sphere_center(
|
||||
calculate_rms_residual=self._calculate_rms_residual
|
||||
)
|
||||
|
||||
if self._long_term_schedule.is_update_due(observation.timestamp):
|
||||
# update long term model with ultra long term bias
|
||||
long_term_estimate = self.long_term_model.estimate_sphere_center(
|
||||
prior_3d=self.ultra_long_term_model.sphere_center,
|
||||
prior_strength=0.1,
|
||||
calculate_rms_residual=self._calculate_rms_residual,
|
||||
)
|
||||
else:
|
||||
# use existing sphere center estimates
|
||||
long_term_estimate = SphereCenterEstimates(
|
||||
projected=self.long_term_model.projected_sphere_center,
|
||||
three_dim=self.long_term_model.sphere_center,
|
||||
rms_residual=self.long_term_model.rms_residual,
|
||||
)
|
||||
|
||||
# update short term model with help of long-term model
|
||||
# using 2d center for disambiguation and 3d center as prior bias
|
||||
# prior strength is set as a funcition of circularity of the 2D pupil
|
||||
# when frozen: do not update
|
||||
if not self.is_long_term_model_frozen:
|
||||
circularity_mean = self.short_term_model.mean_observation_circularity()
|
||||
self.short_term_model.estimate_sphere_center(
|
||||
from_2d=long_term_estimate.projected,
|
||||
prior_3d=long_term_estimate.three_dim,
|
||||
prior_strength=sigmoid(circularity_mean),
|
||||
calculate_rms_residual=self._calculate_rms_residual,
|
||||
)
|
||||
except Exception:
|
||||
# Known issues:
|
||||
# - Can raise numpy.linalg.LinAlgError: SVD did not converge
|
||||
logger.error("Error updating models:")
|
||||
logger.debug(traceback.format_exc())
|
||||
|
||||
def _extract_observation(self, pupil_datum: Dict) -> Observation:
|
||||
width, height = self.camera.resolution
|
||||
center = (
|
||||
pupil_datum["ellipse"]["center"][0] - width / 2,
|
||||
pupil_datum["ellipse"]["center"][1] - height / 2,
|
||||
)
|
||||
minor_radius = pupil_datum["ellipse"]["axes"][0] / 2.0
|
||||
major_radius = pupil_datum["ellipse"]["axes"][1] / 2.0
|
||||
angle = (pupil_datum["ellipse"]["angle"] - 90.0) * np.pi / 180.0
|
||||
ellipse = Ellipse(center, minor_radius, major_radius, angle)
|
||||
|
||||
return Observation(
|
||||
ellipse,
|
||||
pupil_datum["confidence"],
|
||||
pupil_datum["timestamp"],
|
||||
self.camera.focal_length,
|
||||
)
|
||||
|
||||
def _predict_pupil_circle(
|
||||
self, observation: Observation, frame: np.ndarray
|
||||
) -> Circle:
|
||||
# NOTE: General idea: predict pupil circle from long and short term models based
|
||||
# on current observation. Filter results with a kalman filter.
|
||||
|
||||
# Kalman filter needs to be queried every timestamp to update it internally.
|
||||
pupil_circle_kalman = self._predict_from_kalman_filter(observation.timestamp)
|
||||
|
||||
if observation.confidence > self._settings["threshold_swirski"]:
|
||||
# high-confidence observation, use to construct pupil circle from models
|
||||
|
||||
# short-term-model is best for estimating gaze direction (circle normal) if
|
||||
# one needs to assume slippage. long-term-model ist more stable for
|
||||
# positions (center and radius)
|
||||
long_term = self.long_term_model.predict_pupil_circle(observation)
|
||||
if self.is_long_term_model_frozen:
|
||||
normal = long_term.normal
|
||||
else:
|
||||
short_term = self.short_term_model.predict_pupil_circle(observation)
|
||||
normal = short_term.normal
|
||||
pupil_circle = Circle(
|
||||
normal=normal,
|
||||
center=long_term.center,
|
||||
radius=long_term.radius,
|
||||
)
|
||||
|
||||
else:
|
||||
# low confidence: use kalman prediction to search for circles in image
|
||||
pupil_circle, confidence_3d_search = self._predict_from_3d_search(
|
||||
frame, best_guess=pupil_circle_kalman
|
||||
)
|
||||
observation.confidence = confidence_3d_search
|
||||
|
||||
if observation.confidence > self._settings["threshold_kalman"]:
|
||||
# very-high-confidence: correct kalman filter
|
||||
self._correct_kalman_filter(pupil_circle)
|
||||
|
||||
return pupil_circle
|
||||
|
||||
def _predict_from_kalman_filter(self, timestamp):
|
||||
phi, theta, pupil_radius_kalman = self.kalman_filter.predict(timestamp)
|
||||
gaze_vector_kalman = sph2cart(phi, theta)
|
||||
pupil_center_kalman = (
|
||||
self.short_term_model.sphere_center
|
||||
+ _EYE_RADIUS_DEFAULT * gaze_vector_kalman
|
||||
)
|
||||
pupil_circle_kalman = Circle(
|
||||
pupil_center_kalman, gaze_vector_kalman, pupil_radius_kalman
|
||||
)
|
||||
return pupil_circle_kalman
|
||||
|
||||
def _correct_kalman_filter(self, observed_pupil_circle: Circle):
|
||||
if observed_pupil_circle.is_null():
|
||||
return
|
||||
|
||||
phi, theta, r = observed_pupil_circle.spherical_representation()
|
||||
self.kalman_filter.correct(phi, theta, r)
|
||||
|
||||
def _predict_from_3d_search(
|
||||
# TODO: Remove debug code
|
||||
self,
|
||||
frame: np.ndarray,
|
||||
best_guess: Circle,
|
||||
debug=False,
|
||||
) -> Search3DResult:
|
||||
no_result = Search3DResult(Circle.null(), 0.0)
|
||||
|
||||
if best_guess.is_null():
|
||||
return no_result
|
||||
|
||||
frame, frame_roi, edge_frame, edges, roi = get_edges(
|
||||
frame,
|
||||
best_guess.normal,
|
||||
best_guess.radius,
|
||||
self.long_term_model.sphere_center,
|
||||
_EYE_RADIUS_DEFAULT,
|
||||
self.camera.focal_length,
|
||||
self.camera.resolution,
|
||||
major_axis_factor=2.5,
|
||||
)
|
||||
|
||||
if len(edges) <= 0:
|
||||
return no_result
|
||||
|
||||
(gaze_vector, pupil_radius, final_edges, edges_on_sphere) = search_on_sphere(
|
||||
edges,
|
||||
best_guess.normal,
|
||||
best_guess.radius,
|
||||
self.long_term_model.sphere_center,
|
||||
_EYE_RADIUS_DEFAULT,
|
||||
self.camera.focal_length,
|
||||
self.camera.resolution,
|
||||
)
|
||||
|
||||
if debug:
|
||||
frame_ = cv2.cvtColor(frame, cv2.COLOR_GRAY2BGR)
|
||||
try:
|
||||
for edge in edges_on_sphere:
|
||||
edge = project_point_into_image_plane(
|
||||
edge, self.camera.focal_length
|
||||
).astype(np.int)
|
||||
edge[0] += self.camera.resolution[0] / 2
|
||||
edge[1] += self.camera.resolution[1] / 2
|
||||
cv2.rectangle(
|
||||
frame_,
|
||||
(edge[0] - roi[2], edge[1] - roi[0]),
|
||||
(edge[0] + 1 - roi[2], edge[1] + 1 - roi[0]),
|
||||
(255, 0, 0),
|
||||
2,
|
||||
)
|
||||
|
||||
for edge in final_edges:
|
||||
edge = project_point_into_image_plane(
|
||||
edge, self.camera.focal_length
|
||||
).astype(np.int)
|
||||
edge[0] += self.camera.resolution[0] / 2
|
||||
edge[1] += self.camera.resolution[1] / 2
|
||||
cv2.rectangle(
|
||||
frame_,
|
||||
(edge[0] - roi[2], edge[1] - roi[0]),
|
||||
(edge[0] + 1 - roi[2], edge[1] + 1 - roi[0]),
|
||||
(255, 255, 255),
|
||||
1,
|
||||
)
|
||||
|
||||
cv2.imshow("", frame_)
|
||||
cv2.waitKey(1)
|
||||
except Exception as e:
|
||||
print(e)
|
||||
|
||||
pupil_center = (
|
||||
self.long_term_model.sphere_center + _EYE_RADIUS_DEFAULT * gaze_vector
|
||||
)
|
||||
pupil_circle = Circle(pupil_center, gaze_vector, pupil_radius)
|
||||
|
||||
if pupil_circle.is_null():
|
||||
confidence_3d_search = 0.0
|
||||
else:
|
||||
ellipse_2d = project_circle_into_image_plane(
|
||||
pupil_circle,
|
||||
focal_length=self.camera.focal_length,
|
||||
transform=False,
|
||||
width=self.camera.resolution[0],
|
||||
height=self.camera.resolution[1],
|
||||
)
|
||||
if ellipse_2d:
|
||||
circumference = ellipse_2d.circumference()
|
||||
confidence_3d_search = np.clip(
|
||||
len(final_edges) / circumference, 0.0, 1.0
|
||||
)
|
||||
else:
|
||||
confidence_3d_search = 0.0
|
||||
|
||||
return Search3DResult(pupil_circle, confidence_3d_search * 0.6)
|
||||
|
||||
def _prepare_result(
|
||||
self,
|
||||
observation: Observation,
|
||||
prediction_uncorrected: Prediction,
|
||||
prediction_corrected: Prediction,
|
||||
) -> Dict:
|
||||
"""[summary]
|
||||
|
||||
Args:
|
||||
observation (Observation): [description]
|
||||
prediction_uncorrected (Prediction): Used for 2d projections
|
||||
prediction_corrected (Prediction): Used for 3d data
|
||||
|
||||
Returns:
|
||||
Dict: pye3d pupil detection result
|
||||
"""
|
||||
|
||||
result = {
|
||||
"timestamp": observation.timestamp,
|
||||
"sphere": {
|
||||
"center": (
|
||||
prediction_corrected.sphere_center[0],
|
||||
prediction_corrected.sphere_center[1],
|
||||
prediction_corrected.sphere_center[2],
|
||||
),
|
||||
"radius": _EYE_RADIUS_DEFAULT,
|
||||
},
|
||||
}
|
||||
|
||||
eye_sphere_projected = project_sphere_into_image_plane(
|
||||
Sphere(prediction_uncorrected.sphere_center, _EYE_RADIUS_DEFAULT),
|
||||
transform=True,
|
||||
focal_length=self.camera.focal_length,
|
||||
width=self.camera.resolution[0],
|
||||
height=self.camera.resolution[1],
|
||||
)
|
||||
result["projected_sphere"] = ellipse2dict(eye_sphere_projected)
|
||||
|
||||
result["circle_3d"] = circle2dict(prediction_corrected.pupil_circle)
|
||||
|
||||
result["diameter_3d"] = prediction_corrected.pupil_circle.radius * 2
|
||||
|
||||
projected_pupil_circle = project_circle_into_image_plane(
|
||||
prediction_uncorrected.pupil_circle,
|
||||
focal_length=self.camera.focal_length,
|
||||
transform=True,
|
||||
width=self.camera.resolution[0],
|
||||
height=self.camera.resolution[1],
|
||||
)
|
||||
if not projected_pupil_circle:
|
||||
projected_pupil_circle = Ellipse(np.asarray([0.0, 0.0]), 0.0, 0.0, 0.0)
|
||||
|
||||
result["ellipse"] = ellipse2dict(projected_pupil_circle)
|
||||
result["location"] = result["ellipse"]["center"] # pupil center in pixels
|
||||
|
||||
# projected_pupil_circle is an OpenCV ellipse, i.e. major_radius is major diameter
|
||||
result["diameter"] = projected_pupil_circle.major_radius
|
||||
|
||||
result["confidence"] = observation.confidence
|
||||
|
||||
# Model confidence:
|
||||
# - Prior to version 0.1.0, model_confidence was fixed to 1.0 as there was no
|
||||
# way to estimate it
|
||||
# - Starting with version 0.1.0, model_confidence is 1.0 by default but set to
|
||||
# 0.1 if at least one model output exceeds its physiologically reasonable
|
||||
# range. These ranges also inform the input range for the refraction
|
||||
# correction function.
|
||||
# If the ranges are exceeded, it is likely that the model is either not fit
|
||||
# well or the 2d input ellipse was a false detection.
|
||||
model_confidence_default = 1.0
|
||||
model_confidence_out_of_range = 0.1
|
||||
model_confidence_phi_theta_nan = 0.0
|
||||
|
||||
result["model_confidence"] = model_confidence_default
|
||||
|
||||
phi, theta = cart2sph(prediction_corrected.pupil_circle.normal)
|
||||
if not np.any(np.isnan([phi, theta])):
|
||||
result["theta"] = theta
|
||||
result["phi"] = phi
|
||||
|
||||
is_phi_in_range = -80 <= np.rad2deg(phi) + 90.0 <= 80
|
||||
is_theta_in_range = -80 <= np.rad2deg(theta) - 90.0 <= 80
|
||||
if not is_phi_in_range or not is_theta_in_range:
|
||||
result["model_confidence"] = model_confidence_out_of_range
|
||||
else:
|
||||
result["theta"] = 0.0
|
||||
result["phi"] = 0.0
|
||||
result["model_confidence"] = model_confidence_phi_theta_nan
|
||||
|
||||
is_center_x_in_range = -10 <= prediction_corrected.sphere_center[0] <= 10
|
||||
is_center_y_in_range = -10 <= prediction_corrected.sphere_center[1] <= 10
|
||||
is_center_z_in_range = 20 <= prediction_corrected.sphere_center[2] <= 75
|
||||
is_diameter_in_range = 1.0 <= result["diameter_3d"] <= 9.0
|
||||
parameters_in_range = (
|
||||
is_center_x_in_range,
|
||||
is_center_y_in_range,
|
||||
is_center_z_in_range,
|
||||
is_diameter_in_range,
|
||||
)
|
||||
if not all(parameters_in_range):
|
||||
result["model_confidence"] = model_confidence_out_of_range
|
||||
|
||||
return result
|
||||
|
||||
def _collect_debug_info(self):
|
||||
debug_info = {}
|
||||
|
||||
projected_short_term = project_sphere_into_image_plane(
|
||||
Sphere(self.short_term_model.sphere_center, _EYE_RADIUS_DEFAULT),
|
||||
transform=True,
|
||||
focal_length=self.camera.focal_length,
|
||||
width=self.camera.resolution[0],
|
||||
height=self.camera.resolution[1],
|
||||
)
|
||||
projected_long_term = project_sphere_into_image_plane(
|
||||
Sphere(self.long_term_model.sphere_center, _EYE_RADIUS_DEFAULT),
|
||||
transform=True,
|
||||
focal_length=self.camera.focal_length,
|
||||
width=self.camera.resolution[0],
|
||||
height=self.camera.resolution[1],
|
||||
)
|
||||
projected_ultra_long_term = project_sphere_into_image_plane(
|
||||
Sphere(self.ultra_long_term_model.sphere_center, _EYE_RADIUS_DEFAULT),
|
||||
transform=True,
|
||||
focal_length=self.camera.focal_length,
|
||||
width=self.camera.resolution[0],
|
||||
height=self.camera.resolution[1],
|
||||
)
|
||||
debug_info["projected_short_term"] = ellipse2dict(projected_short_term)
|
||||
debug_info["projected_long_term"] = ellipse2dict(projected_long_term)
|
||||
debug_info["projected_ultra_long_term"] = ellipse2dict(
|
||||
projected_ultra_long_term
|
||||
)
|
||||
|
||||
try:
|
||||
bin_data = self.long_term_model.storage.get_bin_counts()
|
||||
max_bin_level = np.max(bin_data)
|
||||
if max_bin_level >= 0:
|
||||
bin_data = bin_data / max_bin_level
|
||||
bin_data = np.flip(bin_data, axis=0)
|
||||
debug_info["bin_data"] = bin_data.tolist()
|
||||
except AttributeError:
|
||||
debug_info["bin_data"] = []
|
||||
|
||||
# TODO: Pupil visualizer_pye3d.py attempts to draw Dierkes lines. Currently we
|
||||
# don't calculate them here, we could probably do that again. Based on which
|
||||
# model? Might be hard to do when things run in the background. We might have to
|
||||
# remove this from the visualizer_pye3d.py
|
||||
debug_info["Dierkes_lines"] = []
|
||||
|
||||
return debug_info
|
||||
|
||||
# pupil-detector interface: See base class implementation as reference:
|
||||
# https://github.com/pupil-labs/pupil-detectors/blob/master/src/pupil_detectors/detector_base.pyx
|
||||
|
||||
PUBLIC_PROPERTY_NAMES = ("is_long_term_model_frozen",)
|
||||
|
||||
def get_properties(self):
|
||||
return {
|
||||
property_name: getattr(self, property_name)
|
||||
for property_name in self.PUBLIC_PROPERTY_NAMES
|
||||
if hasattr(self, property_name)
|
||||
}
|
||||
|
||||
def update_properties(self, properties):
|
||||
keys_to_update = set(self.PUBLIC_PROPERTY_NAMES)
|
||||
keys_to_update.intersection_update(properties.keys())
|
||||
for key in keys_to_update:
|
||||
expected_type = type(getattr(self, key))
|
||||
value = properties[key]
|
||||
try:
|
||||
value = expected_type(value)
|
||||
except ValueError as e:
|
||||
raise ValueError(
|
||||
f"Value `{repr(value)}` for key `{key}` could not be converted to"
|
||||
f" expected type: {expected_type}"
|
||||
) from e
|
||||
setattr(self, key, value)
|
||||
|
||||
|
||||
class _ModelUpdateSchedule:
|
||||
def __init__(self, update_interval: float, warmup_duration: float = 5.0) -> None:
|
||||
self._update_interval = update_interval
|
||||
self._warmup_duration = warmup_duration
|
||||
self._warmup_start = None
|
||||
self._paused = False
|
||||
self._last_update = None
|
||||
|
||||
@property
|
||||
def is_paused(self) -> bool:
|
||||
return self._paused
|
||||
|
||||
def pause(self) -> None:
|
||||
self._paused = True
|
||||
|
||||
def resume(self) -> None:
|
||||
self._paused = False
|
||||
self._last_update = None
|
||||
|
||||
def is_update_due(self, current_time: float):
|
||||
if self._paused:
|
||||
return False
|
||||
if self._warmup_start is None:
|
||||
self._warmup_start = current_time
|
||||
return True
|
||||
if current_time - self._warmup_start < self._warmup_duration:
|
||||
return True
|
||||
if self._last_update is None:
|
||||
self._last_update = current_time
|
||||
return True
|
||||
if current_time - self._last_update > self._update_interval:
|
||||
self._last_update = current_time
|
||||
return True
|
||||
return False
|
||||
@ -1,22 +0,0 @@
|
||||
"""
|
||||
(*)~---------------------------------------------------------------------------
|
||||
Pupil - eye tracking platform
|
||||
Copyright (C) 2012-2019 Pupil Labs
|
||||
|
||||
Distributed under the terms of the GNU
|
||||
Lesser General Public License (LGPL v3.0).
|
||||
See COPYING and COPYING.LESSER for license details.
|
||||
---------------------------------------------------------------------------~(*)
|
||||
"""
|
||||
|
||||
from .abstract import TwoSphereModelAbstract, SphereCenterEstimates
|
||||
from .base import TwoSphereModel
|
||||
from .asynchronous import TwoSphereModelAsync
|
||||
|
||||
|
||||
__all__ = [
|
||||
"TwoSphereModelAbstract",
|
||||
"TwoSphereModel",
|
||||
"TwoSphereModelAsync",
|
||||
"SphereCenterEstimates",
|
||||
]
|
||||
@ -1,116 +0,0 @@
|
||||
"""
|
||||
(*)~---------------------------------------------------------------------------
|
||||
Pupil - eye tracking platform
|
||||
Copyright (C) 2012-2019 Pupil Labs
|
||||
|
||||
Distributed under the terms of the GNU
|
||||
Lesser General Public License (LGPL v3.0).
|
||||
See COPYING and COPYING.LESSER for license details.
|
||||
---------------------------------------------------------------------------~(*)
|
||||
"""
|
||||
import abc
|
||||
import typing as T
|
||||
|
||||
import numpy as np
|
||||
|
||||
from ..geometry.primitives import Circle
|
||||
from ..observation import Observation, ObservationStorage
|
||||
from ..camera import CameraModel
|
||||
|
||||
|
||||
class SphereCenterEstimates(T.NamedTuple):
|
||||
projected: np.ndarray
|
||||
three_dim: np.ndarray
|
||||
rms_residual: T.Optional[float] = None
|
||||
|
||||
|
||||
class TwoSphereModelAbstract(abc.ABC):
|
||||
@abc.abstractmethod
|
||||
def __init__(
|
||||
self,
|
||||
camera: CameraModel,
|
||||
storage_cls: T.Type[ObservationStorage] = None,
|
||||
storage_kwargs: T.Dict = None,
|
||||
):
|
||||
raise NotImplementedError
|
||||
|
||||
@abc.abstractmethod
|
||||
def add_observation(self, observation: Observation):
|
||||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
@abc.abstractmethod
|
||||
def n_observations(self) -> int:
|
||||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
@abc.abstractmethod
|
||||
def sphere_center(self) -> np.ndarray:
|
||||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
@abc.abstractmethod
|
||||
def corrected_sphere_center(self) -> np.ndarray:
|
||||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
@abc.abstractmethod
|
||||
def projected_sphere_center(self) -> np.ndarray:
|
||||
raise NotImplementedError
|
||||
|
||||
@abc.abstractmethod
|
||||
def set_sphere_center(self, new_sphere_center: np.ndarray):
|
||||
raise NotImplementedError
|
||||
|
||||
@abc.abstractmethod
|
||||
def estimate_sphere_center(
|
||||
self,
|
||||
from_2d: T.Optional[np.ndarray] = None,
|
||||
prior_3d: T.Optional[np.ndarray] = None,
|
||||
prior_strength: float = 0.0,
|
||||
calculate_rms_residual: bool = False,
|
||||
) -> SphereCenterEstimates:
|
||||
raise NotImplementedError
|
||||
|
||||
@abc.abstractmethod
|
||||
def estimate_sphere_center_2d(self) -> np.ndarray:
|
||||
raise NotImplementedError
|
||||
|
||||
@abc.abstractmethod
|
||||
def estimate_sphere_center_3d(
|
||||
self,
|
||||
sphere_center_2d: np.ndarray,
|
||||
prior_3d: T.Optional[np.ndarray] = None,
|
||||
prior_strength: float = 0.0,
|
||||
calculate_rms_residual: bool = False,
|
||||
) -> T.Tuple[np.array, T.Optional[float]]:
|
||||
raise NotImplementedError
|
||||
|
||||
# GAZE PREDICTION
|
||||
@abc.abstractmethod
|
||||
def _extract_unproject_disambiguate(self, pupil_datum: T.Dict) -> Circle:
|
||||
raise NotImplementedError
|
||||
|
||||
@abc.abstractmethod
|
||||
def _disambiguate_circle_3d_pair(
|
||||
self, circle_3d_pair: T.Tuple[Circle, Circle]
|
||||
) -> Circle:
|
||||
raise NotImplementedError
|
||||
|
||||
@abc.abstractmethod
|
||||
def predict_pupil_circle(
|
||||
self, observation: Observation, use_unprojection: bool = False
|
||||
) -> Circle:
|
||||
raise NotImplementedError
|
||||
|
||||
@abc.abstractmethod
|
||||
def apply_refraction_correction(self, pupil_circle: Circle) -> Circle:
|
||||
raise NotImplementedError
|
||||
|
||||
@abc.abstractmethod
|
||||
def mean_observation_circularity(self) -> float:
|
||||
raise NotImplementedError
|
||||
|
||||
@abc.abstractmethod
|
||||
def cleanup(self):
|
||||
raise NotImplementedError
|
||||
@ -1,320 +0,0 @@
|
||||
"""
|
||||
(*)~---------------------------------------------------------------------------
|
||||
Pupil - eye tracking platform
|
||||
Copyright (C) 2012-2019 Pupil Labs
|
||||
|
||||
Distributed under the terms of the GNU
|
||||
Lesser General Public License (LGPL v3.0).
|
||||
See COPYING and COPYING.LESSER for license details.
|
||||
---------------------------------------------------------------------------~(*)
|
||||
"""
|
||||
|
||||
import ctypes
|
||||
import logging
|
||||
import typing as T
|
||||
|
||||
import numpy as np
|
||||
|
||||
from ..constants import DEFAULT_SPHERE_CENTER
|
||||
from .abstract import (
|
||||
TwoSphereModelAbstract,
|
||||
CameraModel,
|
||||
Circle,
|
||||
Observation,
|
||||
ObservationStorage,
|
||||
SphereCenterEstimates,
|
||||
)
|
||||
from .background_helper import BackgroundProcess, mp
|
||||
from .base import TwoSphereModel
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TwoSphereModelAsync(TwoSphereModelAbstract):
|
||||
def __init__(
|
||||
self,
|
||||
camera: CameraModel,
|
||||
storage_cls: T.Type[ObservationStorage] = None,
|
||||
storage_kwargs: T.Dict = None,
|
||||
):
|
||||
synced_sphere_center = mp.Array(ctypes.c_double, 3)
|
||||
synced_corrected_sphere_center = mp.Array(ctypes.c_double, 3)
|
||||
synced_projected_sphere_center = mp.Array(ctypes.c_double, 2)
|
||||
synced_observation_count = mp.Value(ctypes.c_long)
|
||||
synced_rms_residual = mp.Value(ctypes.c_double)
|
||||
is_estimation_ongoing_flag = mp.Event()
|
||||
|
||||
self._frontend = _TwoSphereModelSyncedFrontend(
|
||||
synced_sphere_center,
|
||||
synced_corrected_sphere_center,
|
||||
synced_projected_sphere_center,
|
||||
synced_observation_count,
|
||||
synced_rms_residual,
|
||||
is_estimation_ongoing_flag,
|
||||
camera=camera,
|
||||
)
|
||||
self._backend_process = BackgroundProcess(
|
||||
function=self._process_relayed_commands,
|
||||
setup=self._setup_backend,
|
||||
setup_args=(
|
||||
synced_sphere_center,
|
||||
synced_corrected_sphere_center,
|
||||
synced_projected_sphere_center,
|
||||
synced_observation_count,
|
||||
synced_rms_residual,
|
||||
is_estimation_ongoing_flag,
|
||||
),
|
||||
setup_kwargs=dict(
|
||||
camera=camera,
|
||||
storage_cls=storage_cls,
|
||||
storage_kwargs=storage_kwargs,
|
||||
),
|
||||
cleanup=self._cleanup_backend,
|
||||
log_handlers=logging.getLogger().handlers,
|
||||
)
|
||||
|
||||
@property
|
||||
def sphere_center(self) -> np.ndarray:
|
||||
return self._frontend.sphere_center
|
||||
|
||||
@property
|
||||
def corrected_sphere_center(self) -> np.ndarray:
|
||||
return self._frontend.corrected_sphere_center
|
||||
|
||||
@property
|
||||
def projected_sphere_center(self) -> np.ndarray:
|
||||
return self._frontend.projected_sphere_center
|
||||
|
||||
@property
|
||||
def rms_residual(self) -> float:
|
||||
return self._frontend.rms_residual
|
||||
|
||||
def relay_command(self, function_name: str, *args, **kwargs):
|
||||
self._backend_process.send(function_name, *args, **kwargs)
|
||||
|
||||
@staticmethod
|
||||
def _process_relayed_commands(
|
||||
backend: "_TwoSphereModelSyncedBackend", function_name: str, *args, **kwargs
|
||||
):
|
||||
function = getattr(backend, function_name)
|
||||
return function(*args, **kwargs)
|
||||
|
||||
@staticmethod
|
||||
def _setup_backend(*args, **kwargs) -> "_TwoSphereModelSyncedBackend":
|
||||
logger = logging.getLogger(__name__)
|
||||
logger.debug(f"Setting up backend: {args}, {kwargs}")
|
||||
return _TwoSphereModelSyncedBackend(*args, **kwargs)
|
||||
|
||||
@staticmethod
|
||||
def _cleanup_backend(backend: "_TwoSphereModelSyncedBackend"):
|
||||
backend.cleanup()
|
||||
logger = logging.getLogger(__name__)
|
||||
logger.debug(f"Backend cleaned")
|
||||
|
||||
def add_observation(self, observation: Observation):
|
||||
self.relay_command("add_observation", observation)
|
||||
|
||||
@property
|
||||
def n_observations(self) -> int:
|
||||
return self._frontend.n_observations
|
||||
|
||||
def set_sphere_center(self, new_sphere_center: np.ndarray):
|
||||
raise NotImplementedError
|
||||
|
||||
def estimate_sphere_center(
|
||||
self,
|
||||
from_2d: T.Optional[np.ndarray] = None,
|
||||
prior_3d: T.Optional[np.ndarray] = None,
|
||||
prior_strength: float = 0.0,
|
||||
calculate_rms_residual=False,
|
||||
) -> SphereCenterEstimates:
|
||||
if not self._frontend._is_estimation_ongoing_flag.is_set():
|
||||
self.relay_command(
|
||||
"estimate_sphere_center",
|
||||
from_2d,
|
||||
prior_3d,
|
||||
prior_strength,
|
||||
calculate_rms_residual,
|
||||
)
|
||||
self._frontend._is_estimation_ongoing_flag.set()
|
||||
projected_sphere_center = self._frontend.projected_sphere_center
|
||||
sphere_center = self._frontend.sphere_center
|
||||
rms_residual = self._frontend.rms_residual
|
||||
return SphereCenterEstimates(
|
||||
projected_sphere_center, sphere_center, rms_residual
|
||||
)
|
||||
|
||||
def estimate_sphere_center_2d(self) -> np.ndarray:
|
||||
raise NotImplementedError
|
||||
|
||||
def estimate_sphere_center_3d(
|
||||
self,
|
||||
sphere_center_2d: np.ndarray,
|
||||
prior_3d: T.Optional[np.ndarray] = None,
|
||||
prior_strength: float = 0.0,
|
||||
calculate_rms_residual: bool = False,
|
||||
) -> T.Tuple[np.array, T.Optional[float]]:
|
||||
raise NotImplementedError
|
||||
|
||||
# GAZE PREDICTION
|
||||
def _extract_unproject_disambiguate(self, pupil_datum: T.Dict) -> Circle:
|
||||
return self._frontend._extract_unproject_disambiguate(pupil_datum)
|
||||
|
||||
def _disambiguate_circle_3d_pair(
|
||||
self, circle_3d_pair: T.Tuple[Circle, Circle]
|
||||
) -> Circle:
|
||||
return self._frontend._disambiguate_circle_3d_pair(circle_3d_pair)
|
||||
|
||||
def predict_pupil_circle(
|
||||
self, observation: Observation, use_unprojection: bool = False
|
||||
) -> Circle:
|
||||
return self._frontend.predict_pupil_circle(observation, use_unprojection)
|
||||
|
||||
def apply_refraction_correction(self, pupil_circle: Circle) -> Circle:
|
||||
return self._frontend.apply_refraction_correction(pupil_circle)
|
||||
|
||||
def cleanup(self):
|
||||
logger.debug("Cancelling backend process")
|
||||
self._backend_process.cancel()
|
||||
self._frontend.cleanup()
|
||||
|
||||
def mean_observation_circularity(self) -> float:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class _TwoSphereModelSyncedAbstract(TwoSphereModel):
|
||||
def __init__(
|
||||
self,
|
||||
synced_sphere_center: mp.Array, # c_double_Array_3
|
||||
synced_corrected_sphere_center: mp.Array, # c_double_Array_3
|
||||
synced_projected_sphere_center: mp.Array, # c_double_Array_2
|
||||
synced_observation_count: mp.Value, # c_long
|
||||
synced_rms_residual: mp.Value, # c_double
|
||||
flag_is_estimation_ongoing: mp.Event,
|
||||
**kwargs,
|
||||
):
|
||||
self._synced_sphere_center = synced_sphere_center
|
||||
self._synced_corrected_sphere_center = synced_corrected_sphere_center
|
||||
self._synced_projected_sphere_center = synced_projected_sphere_center
|
||||
self._synced_observation_count = synced_observation_count
|
||||
self._synced_rms_residual = synced_rms_residual
|
||||
self._is_estimation_ongoing_flag = flag_is_estimation_ongoing
|
||||
super().__init__(**kwargs)
|
||||
|
||||
@property
|
||||
def sphere_center(self):
|
||||
with self._synced_sphere_center:
|
||||
return np.array(self._synced_sphere_center.get_obj())
|
||||
|
||||
@sphere_center.setter
|
||||
def sphere_center(self, coordinates: np.array):
|
||||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
def corrected_sphere_center(self):
|
||||
with self._synced_corrected_sphere_center:
|
||||
return np.array(self._synced_corrected_sphere_center.get_obj())
|
||||
|
||||
@corrected_sphere_center.setter
|
||||
def corrected_sphere_center(self, coordinates: np.array):
|
||||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
def projected_sphere_center(self):
|
||||
with self._synced_projected_sphere_center:
|
||||
return np.array(self._synced_projected_sphere_center.get_obj())
|
||||
|
||||
@projected_sphere_center.setter
|
||||
def projected_sphere_center(self, coordinates: np.array):
|
||||
raise NotImplementedError
|
||||
|
||||
def mean_observation_circularity(self) -> float:
|
||||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
def rms_residual(self) -> float:
|
||||
with self._synced_rms_residual:
|
||||
return self._synced_rms_residual.value
|
||||
|
||||
@rms_residual.setter
|
||||
def rms_residual(self, residual: float):
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class _TwoSphereModelSyncedFrontend(_TwoSphereModelSyncedAbstract):
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
del self.storage # There is no storage in the frontend
|
||||
|
||||
def _set_default_model_params(self):
|
||||
with self._synced_sphere_center:
|
||||
self._synced_sphere_center[:] = DEFAULT_SPHERE_CENTER
|
||||
|
||||
corrected_sphere_center = self.refractionizer.correct_sphere_center(
|
||||
np.asarray([[*self.sphere_center]])
|
||||
)[0]
|
||||
with self._synced_corrected_sphere_center:
|
||||
self._synced_corrected_sphere_center[:] = corrected_sphere_center
|
||||
|
||||
@property
|
||||
def n_observations(self) -> int:
|
||||
return self._synced_observation_count.value
|
||||
|
||||
|
||||
class _TwoSphereModelSyncedBackend(_TwoSphereModelSyncedAbstract):
|
||||
@property
|
||||
def sphere_center(self):
|
||||
return super().sphere_center
|
||||
|
||||
@sphere_center.setter
|
||||
def sphere_center(self, coordinates: np.array):
|
||||
with self._synced_sphere_center:
|
||||
self._synced_sphere_center[:] = coordinates
|
||||
|
||||
@property
|
||||
def corrected_sphere_center(self):
|
||||
return super().corrected_sphere_center
|
||||
|
||||
@corrected_sphere_center.setter
|
||||
def corrected_sphere_center(self, coordinates: np.array):
|
||||
with self._synced_corrected_sphere_center:
|
||||
self._synced_corrected_sphere_center[:] = coordinates
|
||||
|
||||
@property
|
||||
def projected_sphere_center(self):
|
||||
return super().projected_sphere_center
|
||||
|
||||
@projected_sphere_center.setter
|
||||
def projected_sphere_center(self, coordinates: np.array):
|
||||
with self._synced_projected_sphere_center:
|
||||
self._synced_projected_sphere_center[:] = coordinates
|
||||
|
||||
def add_observation(self, observation: Observation):
|
||||
super().add_observation(observation=observation)
|
||||
n_observations = super().n_observations
|
||||
with self._synced_observation_count:
|
||||
self._synced_observation_count.value = n_observations
|
||||
|
||||
@property
|
||||
def n_observations(self) -> int:
|
||||
return self._synced_observation_count.value
|
||||
|
||||
def estimate_sphere_center(self, *args, **kwargs):
|
||||
result = super().estimate_sphere_center(*args, **kwargs)
|
||||
self._is_estimation_ongoing_flag.clear()
|
||||
return result
|
||||
|
||||
def estimate_sphere_center_2d(self) -> np.ndarray:
|
||||
estimated: np.ndarray = super().estimate_sphere_center_2d()
|
||||
self.projected_sphere_center = estimated
|
||||
return estimated
|
||||
|
||||
@property
|
||||
def rms_residual(self) -> float:
|
||||
with self._synced_rms_residual:
|
||||
return self._synced_rms_residual.value
|
||||
|
||||
@rms_residual.setter
|
||||
def rms_residual(self, residual: float):
|
||||
with self._synced_rms_residual:
|
||||
self._synced_rms_residual.value = residual
|
||||
@ -1,164 +0,0 @@
|
||||
"""
|
||||
(*)~---------------------------------------------------------------------------
|
||||
Pupil - eye tracking platform
|
||||
Copyright (C) 2012-2020 Pupil Labs
|
||||
|
||||
Distributed under the terms of the GNU
|
||||
Lesser General Public License (LGPL v3.0).
|
||||
See COPYING and COPYING.LESSER for license details.
|
||||
---------------------------------------------------------------------------~(*)
|
||||
"""
|
||||
|
||||
import logging
|
||||
import multiprocessing as mp
|
||||
import queue
|
||||
import signal
|
||||
import time
|
||||
from ctypes import c_bool
|
||||
from logging import Handler
|
||||
from logging.handlers import QueueHandler, QueueListener
|
||||
import traceback
|
||||
from typing import Any, Callable, Dict, Iterable, Optional, Tuple, TypeVar
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
WorkerSetupResult = TypeVar("WorkerSetupResult")
|
||||
WorkerFunctionResult = TypeVar("WorkerFunctionResult")
|
||||
|
||||
|
||||
class BackgroundProcess:
|
||||
class StoppedError(Exception):
|
||||
"""Interaction with a BackgroundProcess that was stopped."""
|
||||
|
||||
class NothingToReceiveError(Exception):
|
||||
"""Trying to receive data from BackgroundProcess without sending input first."""
|
||||
|
||||
class MultipleSendError(Exception):
|
||||
"""Trying to send data without first receiving previous output."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
setup: Callable[..., WorkerSetupResult],
|
||||
function: Callable[[WorkerSetupResult], WorkerFunctionResult],
|
||||
cleanup: Callable[[WorkerSetupResult], None],
|
||||
setup_args: Optional[Tuple] = None,
|
||||
setup_kwargs: Optional[Dict] = None,
|
||||
log_handlers: Iterable[Handler] = (),
|
||||
):
|
||||
self._running = True
|
||||
|
||||
self._task_queue = mp.Queue(maxsize=500) # TODO: figure out good value
|
||||
|
||||
logging_queue = mp.Queue()
|
||||
self._log_listener = QueueListener(logging_queue, *log_handlers)
|
||||
self._log_listener.start()
|
||||
|
||||
self._should_terminate_flag = mp.Value(c_bool, 0)
|
||||
|
||||
self._process = mp.Process(
|
||||
name="Pye3D Background Process",
|
||||
daemon=True,
|
||||
target=BackgroundProcess._worker,
|
||||
kwargs=dict(
|
||||
setup=setup,
|
||||
function=function,
|
||||
cleanup=cleanup,
|
||||
task_queue=self._task_queue,
|
||||
should_terminate_flag=self._should_terminate_flag,
|
||||
logging_queue=logging_queue,
|
||||
setup_args=setup_args if setup_args else (),
|
||||
setup_kwargs=setup_kwargs if setup_kwargs else {},
|
||||
),
|
||||
)
|
||||
self._process.start()
|
||||
|
||||
@property
|
||||
def running(self) -> bool:
|
||||
"""Whether background task is running (not necessarily doing work)."""
|
||||
return self._running and self._process.is_alive()
|
||||
|
||||
def send(self, *args: Tuple[Any], **kwargs: Dict[Any, Any]):
|
||||
"""Send data to background process for processing.
|
||||
Raises StoppedError when called on a stopped process.
|
||||
"""
|
||||
|
||||
if not self.running:
|
||||
logger.error("Background process was closed previously!")
|
||||
raise BackgroundProcess.StoppedError()
|
||||
|
||||
try:
|
||||
self._task_queue.put_nowait({"args": args, "kwargs": kwargs})
|
||||
except queue.Full:
|
||||
logger.debug(f"Dropping task! args: {args}, kwargs: {kwargs}")
|
||||
|
||||
def cancel(self, timeout=-1):
|
||||
"""Stop process as soon as current task is finished."""
|
||||
|
||||
self._should_terminate_flag.value = 1
|
||||
if self.running:
|
||||
self._task_queue.close()
|
||||
self._task_queue.cancel_join_thread()
|
||||
self._task_queue.join_thread()
|
||||
self._process.join(timeout)
|
||||
self._running = False
|
||||
self._log_listener.stop()
|
||||
|
||||
@staticmethod
|
||||
def _install_sigint_interception():
|
||||
def interrupt_handler(sig, frame):
|
||||
import traceback
|
||||
|
||||
trace = traceback.format_stack(f=frame)
|
||||
logger.debug(f"Caught (and dropping) signal {sig} in:\n" + "".join(trace))
|
||||
|
||||
signal.signal(signal.SIGINT, interrupt_handler)
|
||||
|
||||
@staticmethod
|
||||
def _worker(
|
||||
setup: Callable[..., WorkerSetupResult],
|
||||
function: Callable[[WorkerSetupResult], Any],
|
||||
cleanup: Callable[[WorkerSetupResult], None],
|
||||
task_queue: mp.Queue,
|
||||
should_terminate_flag: mp.Value,
|
||||
logging_queue: mp.Queue,
|
||||
setup_args: Tuple,
|
||||
setup_kwargs: Dict,
|
||||
):
|
||||
log_queue_handler = QueueHandler(logging_queue)
|
||||
logger = logging.getLogger()
|
||||
logger.setLevel(logging.NOTSET)
|
||||
logger.addHandler(log_queue_handler)
|
||||
|
||||
# Intercept SIGINT (ctrl-c), do required cleanup in foreground process!
|
||||
BackgroundProcess._install_sigint_interception()
|
||||
|
||||
setup_result: WorkerSetupResult = setup(*setup_args, **setup_kwargs)
|
||||
|
||||
while not should_terminate_flag.value:
|
||||
try:
|
||||
params = task_queue.get(block=True, timeout=0.1)
|
||||
args = params["args"]
|
||||
kwargs = params["kwargs"]
|
||||
except queue.Empty:
|
||||
continue
|
||||
# except EOFError:
|
||||
# logger.info("Pipe was closed from foreground process .")
|
||||
# break
|
||||
|
||||
try:
|
||||
t0 = time.perf_counter()
|
||||
function(setup_result, *args, **kwargs)
|
||||
t1 = time.perf_counter()
|
||||
# logger.debug(f"Finished background calculation in {(t1 - t0):.2}s")
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Error executing background process with parameters {params}:\n{e}"
|
||||
)
|
||||
logger.debug(traceback.format_exc())
|
||||
break
|
||||
else:
|
||||
logger.info("Background process received termination signal.")
|
||||
|
||||
cleanup(setup_result)
|
||||
|
||||
logger.info("Stopping background process.")
|
||||
logger.removeHandler(log_queue_handler)
|
||||
@ -1,297 +0,0 @@
|
||||
"""
|
||||
(*)~---------------------------------------------------------------------------
|
||||
Pupil - eye tracking platform
|
||||
Copyright (C) 2012-2019 Pupil Labs
|
||||
|
||||
Distributed under the terms of the GNU
|
||||
Lesser General Public License (LGPL v3.0).
|
||||
See COPYING and COPYING.LESSER for license details.
|
||||
---------------------------------------------------------------------------~(*)
|
||||
"""
|
||||
import logging
|
||||
import typing as T
|
||||
|
||||
import numpy as np
|
||||
|
||||
from .abstract import TwoSphereModelAbstract, SphereCenterEstimates
|
||||
from ..camera import CameraModel
|
||||
from ..constants import _EYE_RADIUS_DEFAULT, DEFAULT_SPHERE_CENTER
|
||||
from ..geometry.intersections import nearest_point_on_sphere_to_line
|
||||
from ..geometry.primitives import Circle, Line
|
||||
from ..geometry.projections import (
|
||||
project_line_into_image_plane,
|
||||
project_point_into_image_plane,
|
||||
unproject_ellipse,
|
||||
)
|
||||
from ..geometry.utilities import normalize
|
||||
from ..observation import BasicStorage, Observation, ObservationStorage
|
||||
from ..refraction import Refractionizer
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TwoSphereModel(TwoSphereModelAbstract):
|
||||
def __init__(
|
||||
self,
|
||||
camera: CameraModel,
|
||||
storage_cls: T.Type[ObservationStorage] = None,
|
||||
storage_kwargs: T.Dict = None,
|
||||
):
|
||||
if storage_cls:
|
||||
kwargs = storage_kwargs if storage_kwargs is not None else {}
|
||||
self.storage = storage_cls(**kwargs)
|
||||
else:
|
||||
self.storage = BasicStorage()
|
||||
self.camera = camera
|
||||
|
||||
self.refractionizer = Refractionizer()
|
||||
self._set_default_model_params()
|
||||
|
||||
@property
|
||||
def sphere_center(self) -> np.ndarray:
|
||||
return self._sphere_center
|
||||
|
||||
@sphere_center.setter
|
||||
def sphere_center(self, coordinates: np.ndarray):
|
||||
self._sphere_center = coordinates
|
||||
|
||||
@property
|
||||
def corrected_sphere_center(self) -> np.ndarray:
|
||||
return self._corrected_sphere_center
|
||||
|
||||
@corrected_sphere_center.setter
|
||||
def corrected_sphere_center(self, coordinates: np.ndarray):
|
||||
self._corrected_sphere_center = coordinates
|
||||
|
||||
@property
|
||||
def projected_sphere_center(self) -> np.ndarray:
|
||||
return self._projected_sphere_center
|
||||
|
||||
@projected_sphere_center.setter
|
||||
def projected_sphere_center(self, projected_sphere_center: np.ndarray):
|
||||
self._projected_sphere_center = projected_sphere_center
|
||||
|
||||
def _set_default_model_params(self):
|
||||
# Overwrite in subclasses that do not allow setting these attributes
|
||||
self._sphere_center = np.asarray(DEFAULT_SPHERE_CENTER)
|
||||
self._corrected_sphere_center = self.refractionizer.correct_sphere_center(
|
||||
np.asarray([[*self.sphere_center]])
|
||||
)[0]
|
||||
self.rms_residual = np.nan
|
||||
|
||||
def add_observation(self, observation: Observation):
|
||||
self.storage.add(observation)
|
||||
|
||||
@property
|
||||
def n_observations(self) -> int:
|
||||
return self.storage.count()
|
||||
|
||||
def set_sphere_center(self, new_sphere_center):
|
||||
self.sphere_center = new_sphere_center
|
||||
self.corrected_sphere_center = self.refractionizer.correct_sphere_center(
|
||||
np.asarray([[*self.sphere_center]])
|
||||
)[0]
|
||||
|
||||
def estimate_sphere_center(
|
||||
self,
|
||||
from_2d=None,
|
||||
prior_3d=None,
|
||||
prior_strength=0.0,
|
||||
calculate_rms_residual=False,
|
||||
):
|
||||
self.projected_sphere_center = (
|
||||
from_2d if from_2d is not None else self.estimate_sphere_center_2d()
|
||||
)
|
||||
sphere_center, rms_residual = self.estimate_sphere_center_3d(
|
||||
self.projected_sphere_center,
|
||||
prior_3d,
|
||||
prior_strength,
|
||||
calculate_rms_residual=calculate_rms_residual,
|
||||
)
|
||||
self.set_sphere_center(sphere_center)
|
||||
self.rms_residual = rms_residual if rms_residual is not None else float("nan")
|
||||
return SphereCenterEstimates(
|
||||
self.projected_sphere_center, sphere_center, rms_residual
|
||||
)
|
||||
|
||||
def estimate_sphere_center_2d(self):
|
||||
observations = self.storage.observations
|
||||
|
||||
# slightly faster than np.array
|
||||
aux_2d = np.concatenate([obs.aux_2d for obs in observations])
|
||||
aux_2d.shape = -1, 2, 3
|
||||
|
||||
# Estimate projected sphere center by nearest intersection of 2d gaze lines
|
||||
sum_aux_2d = aux_2d.sum(axis=0)
|
||||
projected_sphere_center = np.linalg.pinv(sum_aux_2d[:2, :2]) @ sum_aux_2d[:2, 2]
|
||||
|
||||
return projected_sphere_center
|
||||
|
||||
def estimate_sphere_center_3d(
|
||||
self,
|
||||
sphere_center_2d,
|
||||
prior_3d=None,
|
||||
prior_strength=0.0,
|
||||
calculate_rms_residual=False,
|
||||
) -> T.Tuple[np.array, T.Optional[float]]:
|
||||
observations, aux_3d, gaze_2d = self._prep_data()
|
||||
sum_aux_3d, disamb_indices, aux_3d_disamb = self._disambiguate_dierkes_lines(
|
||||
aux_3d, gaze_2d, sphere_center_2d
|
||||
)
|
||||
sphere_center = self._calc_sphere_center(sum_aux_3d, prior_3d, prior_strength)
|
||||
|
||||
rms_residual = (
|
||||
self._calc_rms_residual(
|
||||
observations, disamb_indices, sphere_center, aux_3d_disamb
|
||||
)
|
||||
if calculate_rms_residual
|
||||
else None
|
||||
)
|
||||
|
||||
return sphere_center, rms_residual
|
||||
|
||||
def _prep_data(self):
|
||||
observations = self.storage.observations
|
||||
aux_3d = np.concatenate([obs.aux_3d for obs in observations])
|
||||
aux_3d.shape = -1, 2, 3, 4
|
||||
gaze_2d = np.concatenate([obs.gaze_2d_line for obs in observations])
|
||||
gaze_2d.shape = -1, 4
|
||||
return observations, aux_3d, gaze_2d
|
||||
|
||||
def _disambiguate_dierkes_lines(self, aux_3d, gaze_2d, sphere_center_2d):
|
||||
# Disambiguate Dierkes lines
|
||||
# We want gaze_2d to points towards the sphere center. gaze_2d was collected
|
||||
# from Dierkes[0]. If it points into the correct direction, we know that
|
||||
# Dierkes[0] is the correct one to use, otherwise we need to use Dierkes[1]. We
|
||||
# can check that with the sign of the dot product.
|
||||
gaze_2d_origins = gaze_2d[:, :2]
|
||||
gaze_2d_directions = gaze_2d[:, 2:]
|
||||
gaze_2d_towards_center = gaze_2d_origins - sphere_center_2d
|
||||
|
||||
dot_products = np.sum(gaze_2d_towards_center * gaze_2d_directions, axis=1)
|
||||
disambiguation_indices = np.where(dot_products < 0, 1, 0)
|
||||
|
||||
obs_idc = np.arange(disambiguation_indices.shape[0])
|
||||
aux_3d_disambiguated = aux_3d[obs_idc, disambiguation_indices, :, :]
|
||||
|
||||
# Estimate sphere center by nearest intersection of Dierkes lines
|
||||
sum_aux_3d = aux_3d_disambiguated.sum(axis=0)
|
||||
return sum_aux_3d, disambiguation_indices, aux_3d_disambiguated
|
||||
|
||||
def _calc_sphere_center(self, sum_aux_3d, prior_3d=None, prior_strength=0.0):
|
||||
matrix = sum_aux_3d[:3, :3]
|
||||
try:
|
||||
if prior_3d is None:
|
||||
return np.linalg.pinv(matrix) @ sum_aux_3d[:3, 3]
|
||||
else:
|
||||
return np.linalg.pinv(matrix + prior_strength * np.eye(3)) @ (
|
||||
sum_aux_3d[:3, 3] + prior_strength * prior_3d
|
||||
)
|
||||
except np.linalg.LinAlgError:
|
||||
# happens if lines are parallel, very rare
|
||||
return DEFAULT_SPHERE_CENTER
|
||||
|
||||
def _calc_rms_residual(
|
||||
self, observations, disamb_indices, sphere_center, aux_3d_disamb
|
||||
):
|
||||
# Here we use eq. (10) in https://docplayer.net/21072949-Least-squares-intersection-of-lines.html.
|
||||
origins_dierkes_lines = np.array(
|
||||
[
|
||||
obs.get_Dierkes_line(idx).origin
|
||||
for obs, idx in zip(observations, disamb_indices)
|
||||
]
|
||||
)
|
||||
origins_dierkes_lines.shape = -1, 3, 1
|
||||
deltas = origins_dierkes_lines - sphere_center[:, np.newaxis]
|
||||
tmp = np.einsum("ijk,ikl->ijl", aux_3d_disamb[:, :3, :3], deltas)
|
||||
squared_residuals = np.einsum(
|
||||
"ikj,ijk->i", np.transpose(deltas, (0, 2, 1)), tmp
|
||||
)
|
||||
rms_residual = np.clip(squared_residuals, 0.0, None)
|
||||
rms_residual = np.mean(np.sqrt(rms_residual))
|
||||
return rms_residual
|
||||
|
||||
# GAZE PREDICTION
|
||||
def _extract_unproject_disambiguate(self, pupil_datum):
|
||||
ellipse = self._extract_ellipse(pupil_datum)
|
||||
circle_3d_pair = unproject_ellipse(ellipse, self.camera.focal_length)
|
||||
if circle_3d_pair:
|
||||
circle_3d = self._disambiguate_circle_3d_pair(circle_3d_pair)
|
||||
else:
|
||||
circle_3d = Circle([0.0, 0.0, 0.0], [0.0, 0.0, -1.0], 0.0)
|
||||
return circle_3d
|
||||
|
||||
def _disambiguate_circle_3d_pair(self, circle_3d_pair):
|
||||
circle_center_2d = project_point_into_image_plane(
|
||||
circle_3d_pair[0].center, self.camera.focal_length
|
||||
)
|
||||
circle_normal_2d = normalize(
|
||||
project_line_into_image_plane(
|
||||
Line(circle_3d_pair[0].center, circle_3d_pair[0].normal),
|
||||
self.camera.focal_length,
|
||||
).direction
|
||||
)
|
||||
sphere_center_2d = project_point_into_image_plane(
|
||||
self.sphere_center, self.camera.focal_length
|
||||
)
|
||||
|
||||
if np.dot(circle_center_2d - sphere_center_2d, circle_normal_2d) >= 0:
|
||||
return circle_3d_pair[0]
|
||||
else:
|
||||
return circle_3d_pair[1]
|
||||
|
||||
def predict_pupil_circle(
|
||||
self, observation: Observation, use_unprojection: bool = False
|
||||
) -> Circle:
|
||||
if observation.invalid:
|
||||
return Circle.null()
|
||||
|
||||
circle_3d = self._disambiguate_circle_3d_pair(observation.circle_3d_pair)
|
||||
unprojection_depth = np.linalg.norm(circle_3d.center)
|
||||
direction = circle_3d.center / unprojection_depth
|
||||
|
||||
nearest_point_on_sphere = nearest_point_on_sphere_to_line(
|
||||
self.sphere_center, _EYE_RADIUS_DEFAULT, [0.0, 0.0, 0.0], direction
|
||||
)
|
||||
|
||||
if use_unprojection:
|
||||
gaze_vector = circle_3d.normal
|
||||
else:
|
||||
gaze_vector = normalize(nearest_point_on_sphere - self.sphere_center)
|
||||
|
||||
radius = np.linalg.norm(nearest_point_on_sphere) / unprojection_depth
|
||||
pupil_circle = Circle(nearest_point_on_sphere, gaze_vector, radius)
|
||||
return pupil_circle
|
||||
|
||||
def apply_refraction_correction(self, pupil_circle):
|
||||
input_features = np.asarray(
|
||||
[[*self.sphere_center, *pupil_circle.normal, pupil_circle.radius]]
|
||||
)
|
||||
refraction_corrected_params = self.refractionizer.correct_pupil_circle(
|
||||
input_features
|
||||
)[0]
|
||||
|
||||
refraction_corrected_gaze_vector = normalize(refraction_corrected_params[:3])
|
||||
refraction_corrected_radius = refraction_corrected_params[-1]
|
||||
refraction_corrected_pupil_center = (
|
||||
self.corrected_sphere_center
|
||||
+ _EYE_RADIUS_DEFAULT * refraction_corrected_gaze_vector
|
||||
)
|
||||
|
||||
refraction_corrected_pupil_circle = Circle(
|
||||
refraction_corrected_pupil_center,
|
||||
refraction_corrected_gaze_vector,
|
||||
refraction_corrected_radius,
|
||||
)
|
||||
|
||||
return refraction_corrected_pupil_circle
|
||||
|
||||
def mean_observation_circularity(self):
|
||||
observation_circularities = [
|
||||
observation.ellipse.circularity()
|
||||
for observation in self.storage.observations
|
||||
]
|
||||
return np.mean(observation_circularities)
|
||||
|
||||
def cleanup(self):
|
||||
pass
|
||||
@ -1,161 +0,0 @@
|
||||
"""
|
||||
(*)~---------------------------------------------------------------------------
|
||||
Pupil - eye tracking platform
|
||||
Copyright (C) 2012-2019 Pupil Labs
|
||||
|
||||
Distributed under the terms of the GNU
|
||||
Lesser General Public License (LGPL v3.0).
|
||||
See COPYING and COPYING.LESSER for license details.
|
||||
---------------------------------------------------------------------------~(*)
|
||||
"""
|
||||
import numpy as np
|
||||
|
||||
|
||||
def intersect_line_line(p11, p12, p21, p22, internal=False):
|
||||
x1, y1 = p11
|
||||
x2, y2 = p12
|
||||
x3, y3 = p21
|
||||
x4, y4 = p22
|
||||
|
||||
if ((x1 - x2) * (y3 - y4) - (y1 - y2) * (x3 - x4)) != 0:
|
||||
Px = ((x1 * y2 - y1 * x2) * (x3 - x4) - (x1 - x2) * (x3 * y4 - y3 * x4)) / (
|
||||
(x1 - x2) * (y3 - y4) - (y1 - y2) * (x3 - x4)
|
||||
)
|
||||
Py = ((x1 * y2 - y1 * x2) * (y3 - y4) - (y1 - y2) * (x3 * y4 - y3 * x4)) / (
|
||||
(x1 - x2) * (y3 - y4) - (y1 - y2) * (x3 - x4)
|
||||
)
|
||||
if internal:
|
||||
if x1 != x2:
|
||||
lam = (Px - x2) / (x1 - x2)
|
||||
else:
|
||||
lam = (Py - y2) / (y1 - y2)
|
||||
if 0 <= lam <= 1:
|
||||
return [True, Px, Py]
|
||||
else:
|
||||
return [False]
|
||||
else:
|
||||
return [True, Px, Py]
|
||||
else:
|
||||
return [False]
|
||||
|
||||
|
||||
def intersect_sphere_multiple_lines(sphere_center, radius, points, directions):
|
||||
# Note: Directions need to be normalized!
|
||||
intermediate = np.einsum("ij,ij->i", directions, points - sphere_center)
|
||||
discriminant = (
|
||||
intermediate ** 2 - np.sum((points - sphere_center) ** 2, axis=1) + radius ** 2
|
||||
)
|
||||
idx = discriminant > 0
|
||||
sqr = np.sqrt(discriminant[idx])
|
||||
d1 = -intermediate[idx] + sqr
|
||||
d2 = -intermediate[idx] - sqr
|
||||
d_final = np.expand_dims(np.minimum(d1, d2), axis=1)
|
||||
intersections_on_sphere = points[idx] + d_final * directions[idx]
|
||||
|
||||
return intersections_on_sphere, idx
|
||||
|
||||
|
||||
def intersect_sphere_line(sphere_center, radius, point, direction):
|
||||
temp = np.dot(direction, point - sphere_center)
|
||||
discriminant = temp ** 2 - np.linalg.norm(point - sphere_center) ** 2 + radius ** 2
|
||||
if discriminant >= 0.0:
|
||||
sqr = np.sqrt(discriminant)
|
||||
d1 = -temp + sqr
|
||||
d2 = -temp - sqr
|
||||
return [True, d1, d2]
|
||||
else:
|
||||
return [False, 0.0, 0.0]
|
||||
|
||||
|
||||
def intersect_plane_line(p_plane, n_plane, p_line, l_line, radius=-1):
|
||||
if np.dot(n_plane, l_line) == 0 or np.dot(p_plane - p_line, n_plane) == 0:
|
||||
return [False]
|
||||
else:
|
||||
d = np.dot(p_plane - p_line, n_plane) / np.dot(l_line, n_plane)
|
||||
p_intersect = p_line + d * l_line
|
||||
if radius > 0:
|
||||
if np.linalg.norm(p_plane - p_intersect) <= radius[0]:
|
||||
return [True, p_intersect[0], p_intersect[1], p_intersect[2]]
|
||||
else:
|
||||
return [False, 0.0, 0.0, 0.0]
|
||||
else:
|
||||
return [True, p_intersect[0], p_intersect[1], p_intersect[2]]
|
||||
|
||||
|
||||
def nearest_point_on_sphere_to_line(center, radius, origin, direction):
|
||||
intersection = intersect_sphere_line(center, radius, origin, direction)
|
||||
if intersection[0]:
|
||||
d = np.min(intersection[1:])
|
||||
return origin + d * direction
|
||||
else:
|
||||
temp = np.dot(direction, center - origin)
|
||||
origin_prime = origin + temp * direction
|
||||
direction_prime = center - origin_prime
|
||||
direction_prime /= np.linalg.norm(direction_prime)
|
||||
success, d1, d2 = intersect_sphere_line(
|
||||
center, radius, origin_prime, direction_prime
|
||||
)
|
||||
if success:
|
||||
d = min(d1, d2)
|
||||
return origin_prime + d * direction_prime
|
||||
else:
|
||||
np.zeros(3)
|
||||
|
||||
|
||||
def nearest_intersection_points(p1, p2, p3, p4):
|
||||
"""Calculates the two nearest points, and their distance to each other on
|
||||
two lines defined by (p1,p2) respectively (p3,p4)
|
||||
"""
|
||||
|
||||
def mag(p):
|
||||
return np.sqrt(p.dot(p))
|
||||
|
||||
def normalise(p1, p2):
|
||||
p = p2 - p1
|
||||
m = mag(p)
|
||||
if m == 0:
|
||||
return [0.0, 0.0, 0.0]
|
||||
else:
|
||||
return p / m
|
||||
|
||||
d1 = normalise(p1, p2)
|
||||
d2 = normalise(p3, p4)
|
||||
|
||||
diff = p1 - p3
|
||||
a01 = -d1.dot(d2)
|
||||
b0 = diff.dot(d1)
|
||||
|
||||
if np.abs(a01) < 1.0:
|
||||
|
||||
# Lines are not parallel.
|
||||
det = 1.0 - a01 * a01
|
||||
b1 = -diff.dot(d2)
|
||||
s0 = (a01 * b1 - b0) / det
|
||||
s1 = (a01 * b0 - b1) / det
|
||||
|
||||
else:
|
||||
|
||||
# Lines are parallel, select any pair of closest points.
|
||||
s0 = -b0
|
||||
s1 = 0
|
||||
|
||||
closestPoint1 = p1 + s0 * d1
|
||||
closestPoint2 = p3 + s1 * d2
|
||||
dist = mag(closestPoint2 - closestPoint1)
|
||||
|
||||
return closestPoint1, closestPoint2, dist
|
||||
|
||||
|
||||
def nearest_intersection_lines(lines):
|
||||
dim = len(lines[0].origin)
|
||||
|
||||
R = np.zeros((dim, dim))
|
||||
q = np.zeros(dim)
|
||||
|
||||
for line in lines:
|
||||
v = np.reshape(line.direction, (dim, 1))
|
||||
A = np.eye(dim) - v @ v.T
|
||||
R += A
|
||||
q += A @ line.origin
|
||||
|
||||
return np.linalg.pinv(R) @ q
|
||||
@ -1,188 +0,0 @@
|
||||
"""
|
||||
(*)~---------------------------------------------------------------------------
|
||||
Pupil - eye tracking platform
|
||||
Copyright (C) 2012-2019 Pupil Labs
|
||||
|
||||
Distributed under the terms of the GNU
|
||||
Lesser General Public License (LGPL v3.0).
|
||||
See COPYING and COPYING.LESSER for license details.
|
||||
---------------------------------------------------------------------------~(*)
|
||||
"""
|
||||
import abc
|
||||
|
||||
import numpy as np
|
||||
|
||||
from .utilities import cart2sph, normalize
|
||||
|
||||
|
||||
class Primitive(abc.ABC):
|
||||
__slots__ = ()
|
||||
|
||||
def __repr__(self):
|
||||
klass = "{}.{}".format(self.__class__.__module__, self.__class__.__name__)
|
||||
attributes = " ".join(
|
||||
"{}={}".format(k, v.__repr__()) for k, v in self.__dict__.items()
|
||||
)
|
||||
return "<{klass} at {id}: {attributes}>".format(
|
||||
klass=klass, id=id(self), attributes=attributes
|
||||
)
|
||||
|
||||
def __str__(self):
|
||||
def to_str(obj, float_fmt="{:f}") -> str:
|
||||
if isinstance(obj, float) or isinstance(obj, int):
|
||||
return float_fmt.format(obj)
|
||||
if isinstance(obj, np.ndarray):
|
||||
if obj.dtype != np.object:
|
||||
return ", ".join(float_fmt.format(x) for x in obj)
|
||||
return str(obj)
|
||||
|
||||
klass = self.__class__.__name__
|
||||
attributes = " - ".join(
|
||||
"{}: {}".format(k, to_str(v)) for k, v in self.__dict__.items()
|
||||
)
|
||||
return "{klass} -> {attributes}".format(klass=klass, attributes=attributes)
|
||||
|
||||
|
||||
class Line(Primitive):
|
||||
__slots__ = ("origin", "direction", "dim")
|
||||
|
||||
def __init__(self, origin, direction):
|
||||
self.origin = np.asarray(origin)
|
||||
self.direction = normalize(np.asarray(direction))
|
||||
self.dim = self.origin.shape[0]
|
||||
|
||||
|
||||
class Circle(Primitive):
|
||||
__slots__ = ("center", "normal", "radius")
|
||||
|
||||
def __init__(self, center=[0.0, 0.0, 0.0], normal=[0.0, 0.0, -1.0], radius=0.0):
|
||||
self.center = np.asarray(center, dtype=float)
|
||||
self.normal = np.asarray(normal, dtype=float)
|
||||
self.radius = radius
|
||||
|
||||
def spherical_representation(self):
|
||||
phi, theta = cart2sph(self.normal)
|
||||
return phi, theta, self.radius
|
||||
|
||||
def is_null(self):
|
||||
return self.radius <= 0.0
|
||||
|
||||
@staticmethod
|
||||
def null() -> "Circle":
|
||||
return Circle(radius=0.0)
|
||||
|
||||
|
||||
class Ellipse(Primitive):
|
||||
__slots__ = ("center", "major_radius", "minor_radius", "angle")
|
||||
|
||||
def __init__(self, center, minor_radius, major_radius, angle):
|
||||
self.center = center
|
||||
self.major_radius = major_radius
|
||||
self.minor_radius = minor_radius
|
||||
self.angle = angle
|
||||
|
||||
if self.minor_radius > self.major_radius:
|
||||
current_minor_radius = self.minor_radius
|
||||
self.minor_radius = self.major_radius
|
||||
self.major_radius = current_minor_radius
|
||||
self.angle = self.angle + np.pi / 2
|
||||
|
||||
def circumference(self):
|
||||
a = self.minor_radius
|
||||
b = self.major_radius
|
||||
return np.pi * (3.0 * (a + b) - np.sqrt((3.0 * a + b) * (a + 3.0 * b)))
|
||||
|
||||
def area(self):
|
||||
return np.pi * self.minor_radius * self.major_radius
|
||||
|
||||
def circularity(self):
|
||||
return self.minor_radius / self.major_radius
|
||||
|
||||
def parameters(self):
|
||||
return (
|
||||
self.center[0],
|
||||
self.center[1],
|
||||
self.minor_radius,
|
||||
self.major_radius,
|
||||
self.angle,
|
||||
)
|
||||
|
||||
|
||||
class Sphere(Primitive):
|
||||
__slots__ = ("center", "radius")
|
||||
|
||||
def __init__(self, center, radius):
|
||||
self.center = center
|
||||
self.radius = radius
|
||||
|
||||
def __bool__(self):
|
||||
return self.radius > 0
|
||||
|
||||
|
||||
class Conicoid(Primitive):
|
||||
"""
|
||||
Coefficients of the general equation (implicit form) of a cone, given its vertex and base (ellipse/conic).
|
||||
Formulae follow equations (1)-(3) of:
|
||||
Safaee-Rad, R. et al.: "Three-Dimensional Location Estimation of Circular Features for Machine Vision",
|
||||
IEEE Transactions on Robotics and Automation, Vol.8(5), 1992, pp624-640.
|
||||
"""
|
||||
|
||||
__slots__ = tuple("ABCFGHUVWD")
|
||||
|
||||
def __init__(self, conic, vertex):
|
||||
alpha = vertex[0]
|
||||
beta = vertex[1]
|
||||
gamma = vertex[2]
|
||||
self.A = (gamma ** 2) * conic.A
|
||||
self.B = (gamma ** 2) * conic.C
|
||||
self.C = (
|
||||
conic.A * (alpha ** 2)
|
||||
+ conic.B * alpha * beta
|
||||
+ conic.C * (beta ** 2)
|
||||
+ conic.D * alpha
|
||||
+ conic.E * beta
|
||||
+ conic.F
|
||||
)
|
||||
self.F = -gamma * (conic.C * beta + conic.B / 2 * alpha + conic.E / 2)
|
||||
self.G = -gamma * (conic.B / 2 * beta + conic.A * alpha + conic.D / 2)
|
||||
self.H = (gamma ** 2) * conic.B / 2
|
||||
self.U = (gamma ** 2) * conic.D / 2
|
||||
self.V = (gamma ** 2) * conic.E / 2
|
||||
self.W = -gamma * (conic.E / 2 * beta + conic.D / 2 * alpha + conic.F)
|
||||
self.D = (gamma ** 2) * conic.F
|
||||
|
||||
|
||||
class Conic(Primitive):
|
||||
"""
|
||||
Coefficients A-F of the general equation (implicit form) of a conic
|
||||
Ax² + Bxy + Cy² + Dx + Ey + F = 0
|
||||
calculated from 5 ellipse parameters, see https://en.wikipedia.org/wiki/Ellipse#General_ellipse
|
||||
"""
|
||||
|
||||
__slots__ = tuple("ABCDEF")
|
||||
|
||||
def __init__(self, *args):
|
||||
if len(args) == 1:
|
||||
ellipse = args[0]
|
||||
ax = np.cos(ellipse.angle)
|
||||
ay = np.sin(ellipse.angle)
|
||||
a2 = ellipse.major_radius ** 2
|
||||
b2 = ellipse.minor_radius ** 2
|
||||
|
||||
self.A = a2 * ay * ay + b2 * ax * ax
|
||||
self.B = 2.0 * (b2 - a2) * ax * ay
|
||||
self.C = a2 * ax * ax + b2 * ay * ay
|
||||
self.D = -2.0 * self.A * ellipse.center[0] - self.B * ellipse.center[1]
|
||||
self.E = -self.B * ellipse.center[0] - 2.0 * self.C * ellipse.center[1]
|
||||
self.F = (
|
||||
self.A * ellipse.center[0] * ellipse.center[0]
|
||||
+ self.B * ellipse.center[0] * ellipse.center[1]
|
||||
+ self.C * ellipse.center[1] * ellipse.center[1]
|
||||
- a2 * b2
|
||||
)
|
||||
|
||||
if len(args) == 6:
|
||||
self.A, self.B, self.C, self.D, self.E, self.F = args
|
||||
|
||||
def discriminant(self):
|
||||
return self.B ** 2 - 4 * self.A * self.C
|
||||
@ -1,123 +0,0 @@
|
||||
"""
|
||||
(*)~---------------------------------------------------------------------------
|
||||
Pupil - eye tracking platform
|
||||
Copyright (C) 2012-2019 Pupil Labs
|
||||
|
||||
Distributed under the terms of the GNU
|
||||
Lesser General Public License (LGPL v3.0).
|
||||
See COPYING and COPYING.LESSER for license details.
|
||||
---------------------------------------------------------------------------~(*)
|
||||
"""
|
||||
import logging
|
||||
import warnings
|
||||
|
||||
import numpy as np
|
||||
|
||||
from .intersections import intersect_sphere_multiple_lines
|
||||
from .primitives import Circle, Conic, Conicoid, Ellipse, Line
|
||||
from .utilities import normalize
|
||||
from ..cpp.projections import unproject_ellipse
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def unproject_edges_to_sphere(
|
||||
edges, focal_length, sphere_center, sphere_radius, width=640, height=480
|
||||
):
|
||||
n_edges = edges.shape[0]
|
||||
|
||||
directions = edges - np.asarray([width / 2.0, height / 2.0])
|
||||
directions = np.hstack((directions, focal_length * np.ones((n_edges, 1))))
|
||||
directions = directions / np.linalg.norm(directions, axis=1, keepdims=1)
|
||||
|
||||
origins = np.zeros((n_edges, 3))
|
||||
|
||||
edges_on_sphere, idxs = intersect_sphere_multiple_lines(
|
||||
sphere_center, sphere_radius, origins, directions
|
||||
)
|
||||
|
||||
return edges_on_sphere, idxs
|
||||
|
||||
|
||||
def project_point_into_image_plane(point, focal_length):
|
||||
scale = focal_length / point[2]
|
||||
point_projected = scale * np.asarray(point)
|
||||
return point_projected[:2]
|
||||
|
||||
|
||||
def project_line_into_image_plane(line, focal_length):
|
||||
p1 = line.origin
|
||||
p2 = line.origin + line.direction
|
||||
|
||||
p1_projected = project_point_into_image_plane(p1, focal_length)
|
||||
p2_projected = project_point_into_image_plane(p2, focal_length)
|
||||
|
||||
return Line(p1_projected, p2_projected - p1_projected)
|
||||
|
||||
|
||||
def project_circle_into_image_plane(
|
||||
circle, focal_length, transform=True, width=0, height=0
|
||||
):
|
||||
c = circle.center
|
||||
n = circle.normal
|
||||
r = circle.radius
|
||||
f = focal_length
|
||||
|
||||
cn = np.dot(c, n)
|
||||
c2r2 = np.dot(c, c) - r ** 2
|
||||
ABC = cn ** 2 - 2.0 * cn * (c * n) + c2r2 * (n ** 2)
|
||||
F = 2.0 * (c2r2 * n[1] * n[2] - cn * (n[1] * c[2] + n[2] * c[1]))
|
||||
G = 2.0 * (c2r2 * n[2] * n[0] - cn * (n[2] * c[0] + n[0] * c[2]))
|
||||
H = 2.0 * (c2r2 * n[0] * n[1] - cn * (n[0] * c[1] + n[1] * c[0]))
|
||||
conic = Conic(ABC[0], H, ABC[1], G * f, F * f, ABC[2] * f ** 2)
|
||||
|
||||
disc_ = conic.discriminant()
|
||||
|
||||
if disc_ < 0:
|
||||
|
||||
A, B, C, D, E, F = conic.A, conic.B, conic.C, conic.D, conic.E, conic.F
|
||||
center_x = (2 * C * D - B * E) / disc_
|
||||
center_y = (2 * A * E - B * D) / disc_
|
||||
temp_ = 2 * (A * E ** 2 + C * D ** 2 - B * D * E + disc_ * F)
|
||||
minor_axis = (
|
||||
-np.sqrt(np.abs(temp_ * (A + C - np.sqrt((A - C) ** 2 + B ** 2)))) / disc_
|
||||
) # Todo: Absolute value???
|
||||
major_axis = (
|
||||
-np.sqrt(np.abs(temp_ * (A + C + np.sqrt((A - C) ** 2 + B ** 2)))) / disc_
|
||||
)
|
||||
|
||||
if B == 0 and A < C:
|
||||
angle = 0
|
||||
elif B == 0 and A >= C:
|
||||
angle = np.pi / 2.0
|
||||
else:
|
||||
angle = np.arctan((C - A - np.sqrt((A - C) ** 2 + B ** 2)) / B)
|
||||
|
||||
# TO BE CONSISTENT WITH PUPIL
|
||||
if transform:
|
||||
center_x = center_x + width / 2.0
|
||||
center_y = center_y + height / 2.0
|
||||
minor_axis, major_axis = 2.0 * minor_axis, 2.0 * major_axis
|
||||
angle = angle * 180.0 / np.pi + 90.0
|
||||
|
||||
return Ellipse(np.asarray([center_x, center_y]), minor_axis, major_axis, angle)
|
||||
|
||||
else:
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def project_sphere_into_image_plane(
|
||||
sphere, focal_length, transform=True, width=0, height=0
|
||||
):
|
||||
scale = focal_length / sphere.center[2]
|
||||
|
||||
projected_sphere_center = scale * sphere.center
|
||||
projected_radius = scale * sphere.radius
|
||||
|
||||
if transform:
|
||||
projected_sphere_center[0] += width / 2.0
|
||||
projected_sphere_center[1] += height / 2
|
||||
projected_radius *= 2.0
|
||||
|
||||
return Ellipse(projected_sphere_center[:2], projected_radius, projected_radius, 0.0)
|
||||
@ -1,92 +0,0 @@
|
||||
"""
|
||||
(*)~---------------------------------------------------------------------------
|
||||
Pupil - eye tracking platform
|
||||
Copyright (C) 2012-2019 Pupil Labs
|
||||
|
||||
Distributed under the terms of the GNU
|
||||
Lesser General Public License (LGPL v3.0).
|
||||
See COPYING and COPYING.LESSER for license details.
|
||||
---------------------------------------------------------------------------~(*)
|
||||
"""
|
||||
import numpy as np
|
||||
|
||||
|
||||
def cart2sph(x):
|
||||
|
||||
phi = np.arctan2(x[2], x[0])
|
||||
theta = np.arccos(x[1] / np.linalg.norm(x))
|
||||
|
||||
return phi, theta
|
||||
|
||||
|
||||
def sph2cart(phi, theta):
|
||||
|
||||
result = np.empty(3)
|
||||
|
||||
result[0] = np.sin(theta) * np.cos(phi)
|
||||
result[1] = np.cos(theta)
|
||||
result[2] = np.sin(theta) * np.sin(phi)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def normalize(v, axis=-1):
|
||||
|
||||
return v / np.linalg.norm(v, axis=axis)
|
||||
|
||||
|
||||
def enclosed_angle(v1, v2, unit="deg", axis=-1):
|
||||
|
||||
v1 = normalize(v1, axis=axis)
|
||||
v2 = normalize(v2, axis=axis)
|
||||
|
||||
alpha = np.arccos(np.clip(np.dot(v1.T, v2), -1, 1))
|
||||
|
||||
if unit == "deg":
|
||||
return 180.0 / np.pi * alpha
|
||||
else:
|
||||
return alpha
|
||||
|
||||
|
||||
def make_homogeneous_vector(v):
|
||||
|
||||
return np.hstack((v, [0.0]))
|
||||
|
||||
|
||||
def make_homogeneous_point(p):
|
||||
return np.hstack((p, [1.0]))
|
||||
|
||||
|
||||
def transform_as_homogeneous_point(p, trafo):
|
||||
p = make_homogeneous_point(p)
|
||||
return (trafo @ p)[:3]
|
||||
|
||||
|
||||
def transform_as_homogeneous_vector(v, trafo):
|
||||
v = make_homogeneous_vector(v)
|
||||
return (trafo @ v)[:3]
|
||||
|
||||
|
||||
def rotate_v1_on_v2(v1, v2):
|
||||
|
||||
v1 = normalize(v1)
|
||||
v2 = normalize(v2)
|
||||
cos_angle = np.dot(v1, v2)
|
||||
|
||||
if not np.allclose(np.abs(cos_angle), 1):
|
||||
u = np.cross(v1, v2)
|
||||
s = np.linalg.norm(u)
|
||||
c = np.dot(v1, v2)
|
||||
|
||||
I = np.eye(3)
|
||||
ux = np.asarray([[0, -u[2], u[1]], [u[2], 0, -u[0]], [-u[1], u[0], 0]])
|
||||
|
||||
R = I + ux + np.dot(ux, ux) * (1 - c) / s ** 2
|
||||
|
||||
elif np.allclose(cos_angle, 1):
|
||||
R = np.eye(3)
|
||||
|
||||
elif np.allclose(cos_angle, -1):
|
||||
R = -np.eye(3)
|
||||
|
||||
return R
|
||||
@ -1,58 +0,0 @@
|
||||
"""
|
||||
(*)~---------------------------------------------------------------------------
|
||||
Pupil - eye tracking platform
|
||||
Copyright (C) 2012-2019 Pupil Labs
|
||||
|
||||
Distributed under the terms of the GNU
|
||||
Lesser General Public License (LGPL v3.0).
|
||||
See COPYING and COPYING.LESSER for license details.
|
||||
---------------------------------------------------------------------------~(*)
|
||||
"""
|
||||
import cv2
|
||||
import numpy as np
|
||||
|
||||
|
||||
class KalmanFilter(object):
|
||||
def __init__(self):
|
||||
self.filter = cv2.KalmanFilter(7, 3, 0, cv2.CV_32F)
|
||||
self.filter.measurementMatrix = np.asarray(
|
||||
[[1, 0, 0, 0, 0, 0, 0], [0, 1, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, 0, 1]],
|
||||
dtype=np.float32,
|
||||
)
|
||||
self.filter.processNoiseCov = 1e-4 * np.eye(7, dtype=np.float32)
|
||||
self.filter.measurementNoiseCov = 1e-5 * np.eye(3, dtype=np.float32)
|
||||
self.filter.measurementNoiseCov[2][2] = 0.1
|
||||
self.filter.statePost = np.asarray([0, 0, 0, 0, 0, 0, 2.0], dtype=np.float32)
|
||||
self.filter.errorCovPost = np.eye(7, dtype=np.float32)
|
||||
self.last_call = -1
|
||||
|
||||
def predict(self, t):
|
||||
if self.last_call != -1 and t > self.last_call:
|
||||
dt = t - self.last_call
|
||||
self.filter.transitionMatrix = np.asarray(
|
||||
[
|
||||
[1, 0, dt, 0, 0.5 * dt * dt, 0, 0],
|
||||
[0, 1, 0, dt, 0, 0.5 * dt * dt, 0],
|
||||
[0, 0, 1, 0, dt, 0, 0],
|
||||
[0, 0, 0, 1, 0, dt, 0],
|
||||
[0, 0, 0, 0, 1, 0, 0],
|
||||
[0, 0, 0, 0, 0, 1, 0],
|
||||
[0, 0, 0, 0, 0, 0, 1],
|
||||
],
|
||||
dtype=np.float32,
|
||||
)
|
||||
prediction = self.filter.predict()
|
||||
phi, theta, pupil_radius = (
|
||||
prediction[0][0],
|
||||
prediction[1][0],
|
||||
prediction[6][0],
|
||||
)
|
||||
else:
|
||||
phi, theta, pupil_radius = -np.pi / 2, np.pi / 2, 0
|
||||
|
||||
self.last_call = t
|
||||
|
||||
return phi, theta, pupil_radius
|
||||
|
||||
def correct(self, phi, theta, radius):
|
||||
self.filter.correct(np.asarray([phi, theta, radius], dtype=np.float32))
|
||||
@ -1,232 +0,0 @@
|
||||
"""
|
||||
(*)~---------------------------------------------------------------------------
|
||||
Pupil - eye tracking platform
|
||||
Copyright (C) 2012-2019 Pupil Labs
|
||||
|
||||
Distributed under the terms of the GNU
|
||||
Lesser General Public License (LGPL v3.0).
|
||||
See COPYING and COPYING.LESSER for license details.
|
||||
---------------------------------------------------------------------------~(*)
|
||||
"""
|
||||
from abc import abstractmethod, abstractproperty
|
||||
from collections import deque
|
||||
from math import floor
|
||||
from typing import Sequence, Optional
|
||||
|
||||
import numpy as np
|
||||
from sortedcontainers import SortedList
|
||||
|
||||
from .camera import CameraModel
|
||||
from .constants import _EYE_RADIUS_DEFAULT
|
||||
from .geometry.primitives import Ellipse, Line
|
||||
from .geometry.projections import project_line_into_image_plane, unproject_ellipse
|
||||
|
||||
|
||||
class Observation(object):
|
||||
def __init__(
|
||||
self, ellipse: Ellipse, confidence: float, timestamp: float, focal_length: float
|
||||
):
|
||||
self.ellipse = ellipse
|
||||
self.confidence_2d = confidence
|
||||
self.confidence = 0.0
|
||||
self.timestamp = timestamp
|
||||
|
||||
self.circle_3d_pair = None
|
||||
self.gaze_3d_pair = None
|
||||
self.gaze_2d = None
|
||||
self.aux_2d = None
|
||||
self.aux_3d = None
|
||||
self.invalid = True
|
||||
|
||||
circle_3d_pair = unproject_ellipse(ellipse, focal_length)
|
||||
if not circle_3d_pair:
|
||||
# unprojecting ellipse failed, invalid observation!
|
||||
return
|
||||
|
||||
self.invalid = False
|
||||
self.confidence = self.confidence_2d
|
||||
self.circle_3d_pair = circle_3d_pair
|
||||
|
||||
self.gaze_3d_pair = [
|
||||
Line(
|
||||
circle_3d_pair[i].center,
|
||||
circle_3d_pair[i].center + circle_3d_pair[i].normal,
|
||||
)
|
||||
for i in [0, 1]
|
||||
]
|
||||
self.gaze_2d = project_line_into_image_plane(self.gaze_3d_pair[0], focal_length)
|
||||
self.gaze_2d_line = np.array([*self.gaze_2d.origin, *self.gaze_2d.direction])
|
||||
|
||||
self.aux_2d = np.empty((2, 3))
|
||||
v = np.reshape(self.gaze_2d.direction, (2, 1))
|
||||
self.aux_2d[:, :2] = np.eye(2) - v @ v.T
|
||||
self.aux_2d[:, 2] = (np.eye(2) - v @ v.T) @ self.gaze_2d.origin
|
||||
|
||||
self.aux_3d = np.empty((2, 3, 4))
|
||||
for i in range(2):
|
||||
Dierkes_line = self.get_Dierkes_line(i)
|
||||
v = np.reshape(Dierkes_line.direction, (3, 1))
|
||||
self.aux_3d[i, :3, :3] = np.eye(3) - v @ v.T
|
||||
self.aux_3d[i, :3, 3] = (np.eye(3) - v @ v.T) @ Dierkes_line.origin
|
||||
|
||||
def get_Dierkes_line(self, i):
|
||||
origin = (
|
||||
self.circle_3d_pair[i].center
|
||||
- _EYE_RADIUS_DEFAULT * self.circle_3d_pair[i].normal
|
||||
)
|
||||
direction = self.circle_3d_pair[i].center
|
||||
return Line(origin, direction)
|
||||
|
||||
|
||||
class ObservationStorage:
|
||||
@abstractmethod
|
||||
def add(self, observation: Observation):
|
||||
pass
|
||||
|
||||
@abstractproperty
|
||||
def observations(self) -> Sequence[Observation]:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def clear(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def count(self) -> int:
|
||||
pass
|
||||
|
||||
|
||||
class BasicStorage(ObservationStorage):
|
||||
def __init__(self):
|
||||
self._storage = []
|
||||
|
||||
def add(self, observation: Observation):
|
||||
if observation.invalid:
|
||||
return
|
||||
self._storage.append(observation)
|
||||
|
||||
@property
|
||||
def observations(self) -> Sequence[Observation]:
|
||||
return self._storage
|
||||
|
||||
def clear(self):
|
||||
self._storage.clear()
|
||||
|
||||
def count(self) -> int:
|
||||
return len(self._storage)
|
||||
|
||||
|
||||
class BufferedObservationStorage(ObservationStorage):
|
||||
def __init__(self, confidence_threshold: float, buffer_length: int):
|
||||
self.confidence_threshold = confidence_threshold
|
||||
self._storage = deque(maxlen=buffer_length)
|
||||
|
||||
def add(self, observation: Observation):
|
||||
if observation.invalid:
|
||||
return
|
||||
if observation.confidence < self.confidence_threshold:
|
||||
return
|
||||
|
||||
self._storage.append(observation)
|
||||
|
||||
@property
|
||||
def observations(self) -> Sequence[Observation]:
|
||||
return list(self._storage)
|
||||
|
||||
def clear(self):
|
||||
self._storage.clear()
|
||||
|
||||
def count(self) -> int:
|
||||
return len(self._storage)
|
||||
|
||||
|
||||
class BinBufferedObservationStorage(ObservationStorage):
|
||||
def __init__(
|
||||
self,
|
||||
camera: CameraModel,
|
||||
confidence_threshold: float,
|
||||
n_bins_horizontal: int,
|
||||
bin_buffer_length: int,
|
||||
forget_min_observations: Optional[int] = None,
|
||||
forget_min_time: Optional[float] = None,
|
||||
):
|
||||
self.camera = camera
|
||||
self.confidence_threshold = confidence_threshold
|
||||
self.bin_buffer_length = bin_buffer_length
|
||||
self.forget_min_observations = forget_min_observations
|
||||
self.forget_min_time = forget_min_time
|
||||
self.pixels_per_bin = self.camera.resolution[0] / n_bins_horizontal
|
||||
self.w = n_bins_horizontal
|
||||
self.h = int(round(self.camera.resolution[1] / self.pixels_per_bin))
|
||||
|
||||
self._by_time = SortedList(key=lambda obs: obs.timestamp)
|
||||
self._by_bin = dict()
|
||||
|
||||
def add(self, observation: Observation):
|
||||
if observation.invalid:
|
||||
return
|
||||
if observation.confidence < self.confidence_threshold:
|
||||
return
|
||||
|
||||
idx = self._get_bin(observation)
|
||||
if idx < 0 or idx >= self.w * self.h:
|
||||
print(f"INDEX OUT OF BOUNDS: {idx}")
|
||||
return
|
||||
|
||||
if idx not in self._by_bin:
|
||||
self._by_bin[idx] = SortedList(key=lambda obs: obs.timestamp)
|
||||
|
||||
# add to both lookup structures
|
||||
_bin: SortedList = self._by_bin[idx]
|
||||
_bin.add(observation)
|
||||
self._by_time.add(observation)
|
||||
|
||||
# manage within-bin forgetting
|
||||
while len(_bin) > self.bin_buffer_length:
|
||||
old = _bin.pop(0)
|
||||
self._by_time.remove(old)
|
||||
|
||||
# manage across-bin forgetting
|
||||
if self.forget_min_observations is None or self.forget_min_time is None:
|
||||
return
|
||||
|
||||
while self.count() > self.forget_min_observations:
|
||||
oldest_age = observation.timestamp - self._by_time[0].timestamp
|
||||
if oldest_age < self.forget_min_time:
|
||||
break
|
||||
|
||||
# forget oldest entry
|
||||
old = self._by_time.pop(0)
|
||||
idx = self._get_bin(old)
|
||||
_bin = self._by_bin[idx]
|
||||
_bin.remove(old)
|
||||
# make sure to remove bin if empty for bin-counting to work
|
||||
if len(_bin) == 0:
|
||||
self._by_bin.pop(idx)
|
||||
|
||||
@property
|
||||
def observations(self) -> Sequence[Observation]:
|
||||
return list(self._by_time)
|
||||
|
||||
def clear(self):
|
||||
self._by_time.clear()
|
||||
self._by_bin.clear()
|
||||
|
||||
def count(self) -> int:
|
||||
return len(self._by_time)
|
||||
|
||||
def get_bin_counts(self) -> np.ndarray:
|
||||
dense_1d = np.zeros((self.w * self.h,))
|
||||
for idx, _bin in self._by_bin.items():
|
||||
dense_1d[idx] = len(_bin)
|
||||
return np.reshape(dense_1d, (self.w, self.h))
|
||||
|
||||
def _get_bin(self, observation: Observation) -> int:
|
||||
x, y = (
|
||||
floor((ellipse_center + resolution / 2) / self.pixels_per_bin)
|
||||
for ellipse_center, resolution in zip(
|
||||
observation.ellipse.center, self.camera.resolution
|
||||
)
|
||||
)
|
||||
# convert to 1D bin index
|
||||
return x + y * self.h
|
||||
@ -1,141 +0,0 @@
|
||||
import itertools
|
||||
from pathlib import Path
|
||||
from .cpp.refraction_correction import apply_correction_pipeline
|
||||
|
||||
import numpy as np
|
||||
import msgpack
|
||||
|
||||
LOAD_DIR = Path(__file__).parent / "refraction_models"
|
||||
LOAD_VERSION = 1
|
||||
|
||||
|
||||
class ModelDeserializationError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class Refractionizer:
|
||||
def __init__(self, degree=3, type_="default", custom_load_dir=None):
|
||||
self.pipeline_radius_as_list = self.load_config_from_msgpack(
|
||||
"radius", type_, degree, custom_load_dir
|
||||
)
|
||||
|
||||
self.pipeline_gaze_vector_as_list = self.load_config_from_msgpack(
|
||||
"gaze_vector", type_, degree, custom_load_dir
|
||||
)
|
||||
|
||||
self.pipeline_sphere_center_as_list = self.load_config_from_msgpack(
|
||||
"sphere_center", type_, degree, custom_load_dir
|
||||
)
|
||||
|
||||
self.pipeline_pupil_circle_as_list = self.load_config_from_msgpack(
|
||||
"pupil_circle", type_, degree, custom_load_dir
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def load_config_from_msgpack(feature, type_, degree, custom_load_dir=None):
|
||||
load_dir = Path(custom_load_dir or LOAD_DIR).resolve()
|
||||
name = f"{type_}_refraction_model_{feature}_degree_{degree}.msgpack"
|
||||
path = load_dir / name
|
||||
with path.open("rb") as file:
|
||||
config_model = msgpack.unpack(file)
|
||||
Refractionizer._validate_loaded_model_config(config_model)
|
||||
try:
|
||||
return list(
|
||||
itertools.chain(
|
||||
Refractionizer._polynomial_features_from_config(config_model),
|
||||
Refractionizer._standard_scaler_from_config(config_model),
|
||||
Refractionizer._linear_regression_from_config(config_model),
|
||||
)
|
||||
)
|
||||
except KeyError as err:
|
||||
raise ModelDeserializationError from err
|
||||
|
||||
@staticmethod
|
||||
def _validate_loaded_model_config(config_model):
|
||||
if not isinstance(config_model, dict) or "version" not in config_model:
|
||||
raise ModelDeserializationError("Unrecognized format")
|
||||
if config_model["version"] != LOAD_VERSION:
|
||||
raise ModelDeserializationError(
|
||||
f"Unexpected version `{config_model['version']}` "
|
||||
f"(expected `{LOAD_VERSION}``)"
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _polynomial_features_from_config(config_model):
|
||||
yield np.array(config_model["steps"]["PolynomialFeatures"]["powers"])
|
||||
|
||||
@staticmethod
|
||||
def _standard_scaler_from_config(config_model):
|
||||
config_scaler = config_model["steps"]["StandardScaler"]
|
||||
yield np.array(config_scaler["mean"])
|
||||
yield np.array(config_scaler["var"])
|
||||
|
||||
@staticmethod
|
||||
def _linear_regression_from_config(config_model):
|
||||
config_lin_reg = config_model["steps"]["LinearRegression"]
|
||||
yield np.array(config_lin_reg["coef"])
|
||||
yield np.array(config_lin_reg["intercept"])
|
||||
|
||||
@staticmethod
|
||||
def _apply_correction_pipeline(X, pipeline_arrays):
|
||||
return apply_correction_pipeline(np.asarray(X).T, *pipeline_arrays)
|
||||
|
||||
def correct_radius(self, X):
|
||||
return self._apply_correction_pipeline(X, self.pipeline_radius_as_list)
|
||||
|
||||
def correct_gaze_vector(self, X):
|
||||
return self._apply_correction_pipeline(X, self.pipeline_gaze_vector_as_list)
|
||||
|
||||
def correct_sphere_center(self, X):
|
||||
return self._apply_correction_pipeline(X, self.pipeline_sphere_center_as_list)
|
||||
|
||||
def correct_pupil_circle(self, X):
|
||||
return self._apply_correction_pipeline(X, self.pipeline_pupil_circle_as_list)
|
||||
|
||||
|
||||
class SklearnRefractionizer(Refractionizer):
|
||||
def __init__(self, degree=3, type_="default", custom_load_dir=None):
|
||||
self.correct_radius = self.load_predict_fn_from_joblib_pickle(
|
||||
"radius", type_, degree, custom_load_dir
|
||||
)
|
||||
|
||||
self.correct_gaze_vector = self.load_predict_fn_from_joblib_pickle(
|
||||
"gaze_vector", type_, degree, custom_load_dir
|
||||
)
|
||||
|
||||
self.correct_sphere_center = self.load_predict_fn_from_joblib_pickle(
|
||||
"sphere_center", type_, degree, custom_load_dir
|
||||
)
|
||||
|
||||
self.correct_pupil_circle = self.load_predict_fn_from_joblib_pickle(
|
||||
"pupil_circle", type_, degree, custom_load_dir
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def load_predict_fn_from_joblib_pickle(
|
||||
feature, type_, degree, custom_load_dir=None
|
||||
):
|
||||
import joblib
|
||||
|
||||
load_dir = Path(custom_load_dir or LOAD_DIR).resolve()
|
||||
name = f"{type_}_refraction_model_{feature}_degree_{degree}.save"
|
||||
path = load_dir / name
|
||||
try:
|
||||
pipeline = joblib.load(path)
|
||||
except FileNotFoundError as err:
|
||||
raise
|
||||
except Exception as exc:
|
||||
raise ModelDeserializationError(
|
||||
f"Failed to load pickled model from {path}"
|
||||
) from exc
|
||||
return pipeline.predict
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
refractionizer = Refractionizer()
|
||||
|
||||
print(refractionizer.correct_sphere_center([[0.0, 0.0, 35.0]]))
|
||||
print(refractionizer.correct_radius([[0.0, 0.0, 35.0, 0.0, 0.0, -1.0, 2.0]]))
|
||||
print(refractionizer.correct_gaze_vector([[0.0, 0.0, 35.0, 0.0, 0.0, -1.0, 2.0]]))
|
||||
print(refractionizer.correct_pupil_circle([[0.0, 0.0, 35.0, 0.0, 0.0, -1.0, 2.0]]))
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@ -1,8 +0,0 @@
|
||||
cmake
|
||||
python-osc
|
||||
opencv-python
|
||||
numpy
|
||||
pillow
|
||||
kivy
|
||||
scipy
|
||||
pyttsx3
|
||||
@ -1,103 +0,0 @@
|
||||
|
||||
import kivy
|
||||
from multiprocessing import Process,Queue,Pipe
|
||||
kivy.require("1.9.1")
|
||||
from kivy.app import App
|
||||
from kivy.uix.gridlayout import GridLayout
|
||||
from kivy.uix.slider import Slider
|
||||
from kivy.uix.label import Label
|
||||
from kivy.uix.floatlayout import FloatLayout
|
||||
from kivy.properties import NumericProperty
|
||||
from kivy.uix.scatter import Scatter
|
||||
from kivy.uix.textinput import TextInput
|
||||
from kivy.uix.boxlayout import BoxLayout
|
||||
from kivy.core.window import Window
|
||||
import time
|
||||
import cv2
|
||||
|
||||
|
||||
|
||||
###############################################################################
|
||||
|
||||
Window.size = (600, 60)
|
||||
|
||||
class WidgetContainer(GridLayout):
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
|
||||
super(WidgetContainer, self).__init__(**kwargs)
|
||||
|
||||
|
||||
|
||||
|
||||
############################################################################### right
|
||||
|
||||
self.cols = 3
|
||||
self.xcc = Slider(min = 20, max = 200,
|
||||
value_track = True,
|
||||
value_track_color =[1, 1, 1, 1])
|
||||
self.add_widget(Label(text ='Threshold'))
|
||||
self.add_widget(self.xcc)
|
||||
self.xValue = Label(text ='Select')
|
||||
self.add_widget(self.xValue)
|
||||
self.xcc.bind(value = self.on_value)
|
||||
|
||||
|
||||
|
||||
|
||||
self.rota = Slider(min = 0, max = 360,
|
||||
value_track = True,
|
||||
value_track_color =[1, 1, 1, 1])
|
||||
self.add_widget(Label(text ='Rotation'))
|
||||
self.add_widget(self.rota)
|
||||
self.rotav= Label(text ='Select')
|
||||
self.add_widget(self.rotav)
|
||||
self.rota.bind(value = self.on_value1)
|
||||
|
||||
|
||||
|
||||
def on_value(self, instance, brightness):
|
||||
self.xValue.text = "% d"% brightness
|
||||
confg.fx = self.xValue.text
|
||||
configsave()
|
||||
time.sleep(0.1)
|
||||
|
||||
|
||||
def on_value1(self, instance, brightness):
|
||||
self.rotav.text = "% d"% brightness
|
||||
confg.rv = self.rotav.text
|
||||
configsave()
|
||||
time.sleep(0.1)
|
||||
|
||||
|
||||
|
||||
class EyetrackGUI(App):
|
||||
def build(self):
|
||||
widgetContainer = WidgetContainer()
|
||||
print()
|
||||
|
||||
return widgetContainer
|
||||
|
||||
|
||||
|
||||
def confg():
|
||||
|
||||
confg.fx = 128
|
||||
confg.rv = 0
|
||||
|
||||
|
||||
|
||||
def configsave():
|
||||
with open('settings.cfg', 'w+') as rf:
|
||||
rf.write(str(confg.fx))
|
||||
rf.write('\n')
|
||||
rf.write(str(confg.rv))
|
||||
|
||||
|
||||
|
||||
confg()
|
||||
|
||||
rootGUI = EyetrackGUI()
|
||||
|
||||
|
||||
rootGUI.run()
|
||||
Loading…
Reference in New Issue
Block a user