feat: stage one of new output formats

This commit is contained in:
Prohurtz 2023-10-03 14:51:36 -05:00
parent 62fba485ba
commit 8e28092601
5 changed files with 417 additions and 250 deletions

View File

@ -63,7 +63,6 @@ class EyeTrackSettingsConfig(BaseModel):
gui_thresh_add: int = 11
gui_update_check: bool = False
gui_ROSC: bool = False
gui_vrc_native: bool = True
gui_circular_crop_right: bool = False
gui_circular_crop_left: bool = False
ibo_filter_samples: int = 400
@ -81,6 +80,9 @@ class EyeTrackSettingsConfig(BaseModel):
gui_legacy_ransac_thresh_right: int = 80
gui_legacy_ransac_thresh_left: int = 80
gui_LEAP_lid: bool = False
gui_osc_vrcft_v1: bool = False
gui_osc_vrcft_v2: bool = False
gui_vrc_native: bool = True
class EyeTrackConfig(BaseModel):

View File

@ -5,8 +5,11 @@ from osc import EyeId
from queue import Queue
from threading import Event
class SettingsWidget:
def __init__(self, widget_id: EyeId, main_config: EyeTrackSettingsConfig, osc_queue: Queue):
def __init__(
self, widget_id: EyeId, main_config: EyeTrackSettingsConfig, osc_queue: Queue
):
self.gui_flip_x_axis_left = f"-FLIPXAXISLEFT{widget_id}-"
self.gui_flip_x_axis_right = f"-FLIPXAXISRIGHT{widget_id}-"
@ -43,160 +46,179 @@ class SettingsWidget:
self.gui_circular_crop_right = f"-CIRCLECROPRIGHT{widget_id}-"
self.gui_update_check = f"-UPDATECHECK{widget_id}-"
self.gui_threshold_slider = f"-BLOBTHRESHOLD{widget_id}-"
self.gui_osc_vrcft_v1 = f"-OSCVRCFTV1{widget_id}-"
self.gui_osc_vrcft_v2 = f"-OSCVRCFTV2{widget_id}-"
self.main_config = main_config
self.config = main_config.settings
self.osc_queue = osc_queue
# Define the window's contents
self.general_settings_layout = [
[
sg.Checkbox(
"Flip Left Eye X Axis",
default=self.config.gui_flip_x_axis_left,
key=self.gui_flip_x_axis_left,
background_color='#424042',
tooltip = "Flips the left eye's X axis.",
background_color="#424042",
tooltip="Flips the left eye's X axis.",
),
sg.Checkbox(
"Flip Right Eye X Axis",
default=self.config.gui_flip_x_axis_right,
key=self.gui_flip_x_axis_right,
background_color='#424042',
tooltip = "Flips the right eye's X axis.",
background_color="#424042",
tooltip="Flips the right eye's X axis.",
),
# ],
#[
sg.Checkbox(
# ],
# [
sg.Checkbox(
"Flip Y Axis",
default=self.config.gui_flip_y_axis,
key=self.gui_flip_y_axis,
background_color='#424042',
tooltip = "Flips the eye's Y axis.",
background_color="#424042",
tooltip="Flips the eye's Y axis.",
),
],
[sg.Checkbox(
"VRC Native Eyetracking",
default=self.config.gui_vrc_native,
key=self.gui_vrc_native,
background_color='#424042',
tooltip = "Toggle VRCFT output or VRC native",
),
[
sg.Checkbox(
"Dual Eye Falloff",
default=self.config.gui_eye_falloff,
key=self.gui_eye_falloff,
background_color='#424042',
tooltip = "If one eye stops tracking, we send tracking data from your other eye.",
background_color="#424042",
tooltip="If one eye stops tracking, we send tracking data from your other eye.",
),
],
[sg.Checkbox(
[
sg.Checkbox(
"Check For Updates",
default=self.config.gui_update_check,
key=self.gui_update_check,
background_color='#424042',
tooltip = "Toggle update check on launch.",
background_color="#424042",
tooltip="Toggle update check on launch.",
),
],
[
sg.Text("One Euro Filter Paramaters:", background_color='#242224'),
sg.Text("One Euro Filter Paramaters:", background_color="#242224"),
],
[
sg.Text("Min Frequency Cutoff", background_color='#424042'),
sg.Text("Min Frequency Cutoff", background_color="#424042"),
sg.InputText(
self.config.gui_min_cutoff,
key=self.gui_min_cutoff,
size=(0,10),
size=(0, 10),
),
#],
#[
sg.Text("Speed Coefficient", background_color='#424042'),
# ],
# [
sg.Text("Speed Coefficient", background_color="#424042"),
sg.InputText(
self.config.gui_speed_coefficient,
self.config.gui_speed_coefficient,
key=self.gui_speed_coefficient,
size=(0,10),
size=(0, 10),
),
],
[
sg.Text("OSC Settings:", background_color='#242224'),
[
sg.Text("OSC Settings:", background_color="#242224"),
],
[
sg.Text("Address:", background_color='#424042'),
sg.Checkbox(
"VRC Native",
default=self.config.gui_vrc_native,
key=self.gui_vrc_native,
background_color="#424042",
tooltip="Toggle VRC's Native Eyetracking format.",
),
sg.Checkbox(
"VRCFT v1",
default=self.config.gui_osc_vrcft_v1,
key=self.gui_osc_vrcft_v1,
background_color="#424042",
tooltip="Toggle VRCFT's v1 Eyetracking format.",
),
sg.Checkbox(
"VRCFT v2 (UE)",
default=self.config.gui_osc_vrcft_v2,
key=self.gui_osc_vrcft_v2,
background_color="#424042",
tooltip="Toggle VRCFT's v2 (UE) Eyetracking format.",
),
],
[
sg.Text("Address:", background_color="#424042"),
sg.InputText(
self.config.gui_osc_address,
self.config.gui_osc_address,
key=self.gui_osc_address,
size=(0,20),
tooltip = "IP address we send OSC data to.",
size=(0, 20),
tooltip="IP address we send OSC data to.",
),
# ],
# [
sg.Text("Port:", background_color='#424042'),
# ],
# [
sg.Text("Port:", background_color="#424042"),
sg.InputText(
self.config.gui_osc_port,
self.config.gui_osc_port,
key=self.gui_osc_port,
size=(0,10),
tooltip = "OSC port we send data to.",
size=(0, 10),
tooltip="OSC port we send data to.",
),
],
[
sg.Text("Receive functions", background_color='#424042'),
sg.Text("Receive functions", background_color="#424042"),
sg.Checkbox(
"",
default=self.config.gui_ROSC,
key=self.gui_ROSC,
background_color='#424042',
size=(0,10),
tooltip = "Toggle OSC receive functions.",
background_color="#424042",
size=(0, 10),
tooltip="Toggle OSC receive functions.",
),
],
[
sg.Text("Receiver Port:", background_color='#424042'),
sg.Text("Receiver Port:", background_color="#424042"),
sg.InputText(
self.config.gui_osc_receiver_port,
self.config.gui_osc_receiver_port,
key=self.gui_osc_receiver_port,
size=(0,10),
tooltip = "Port we receive OSC data from (used to recalibrate or recenter app from within VRChat.",
size=(0, 10),
tooltip="Port we receive OSC data from (used to recalibrate or recenter app from within VRChat.",
),
#],
# [
sg.Text("Recenter Address:", background_color='#424042'),
# ],
# [
sg.Text("Recenter Address:", background_color="#424042"),
sg.InputText(
self.config.gui_osc_recenter_address,
self.config.gui_osc_recenter_address,
key=self.gui_osc_recenter_address,
size=(0,10),
tooltip = "OSC Address used for recentering your eye.",
),
size=(0, 10),
tooltip="OSC Address used for recentering your eye.",
),
],
[
sg.Text("Recalibrate Address:", background_color='#424042'),
sg.Text("Recalibrate Address:", background_color="#424042"),
sg.InputText(
self.config.gui_osc_recalibrate_address,
self.config.gui_osc_recalibrate_address,
key=self.gui_osc_recalibrate_address,
size=(0,10),
tooltip = "OSC address we use for recalibrating your eye",
),
]
size=(0, 10),
tooltip="OSC address we use for recalibrating your eye",
),
],
]
self.widget_layout = [
[
sg.Text("General Settings:", background_color='#242224'),
[
sg.Text("General Settings:", background_color="#242224"),
],
[
sg.Column(self.general_settings_layout, key=self.gui_general_settings_layout, background_color='#424042' ),
sg.Column(
self.general_settings_layout,
key=self.gui_general_settings_layout,
background_color="#424042",
),
],
]
self.cancellation_event = Event() # Set the event until start is called, otherwise we can block if shutdown is called.
self.cancellation_event = (
Event()
) # Set the event until start is called, otherwise we can block if shutdown is called.
self.cancellation_event.set()
self.image_queue = Queue()
def started(self):
return not self.cancellation_event.is_set()
@ -218,43 +240,61 @@ class SettingsWidget:
if self.config.gui_osc_port != int(values[self.gui_osc_port]):
print(self.config.gui_osc_port, values[self.gui_osc_port])
try:
try:
int(values[self.gui_osc_port])
if len(values[self.gui_osc_port]) <= 5:
self.config.gui_osc_port = int(values[self.gui_osc_port])
changed = True
else:
print("\033[91m[ERROR] OSC port value must be an integer 0-65535\033[0m")
print(
"\033[91m[ERROR] OSC port value must be an integer 0-65535\033[0m"
)
except:
print("\033[91m[ERROR] OSC port value must be an integer 0-65535\033[0m")
print(
"\033[91m[ERROR] OSC port value must be an integer 0-65535\033[0m"
)
if self.config.gui_osc_receiver_port != int(values[self.gui_osc_receiver_port]):
try:
try:
int(values[self.gui_osc_receiver_port])
if len(values[self.gui_osc_receiver_port]) <= 5:
self.config.gui_osc_receiver_port = int(values[self.gui_osc_receiver_port])
self.config.gui_osc_receiver_port = int(
values[self.gui_osc_receiver_port]
)
changed = True
else:
print("\033[91m[ERROR] OSC receive port value must be an integer 0-65535\033[0m")
print(
"\033[91m[ERROR] OSC receive port value must be an integer 0-65535\033[0m"
)
except:
print("\033[91m[ERROR] OSC receive port value must be an integer 0-65535\033[0m")
print(
"\033[91m[ERROR] OSC receive port value must be an integer 0-65535\033[0m"
)
if self.config.gui_osc_address != values[self.gui_osc_address]:
self.config.gui_osc_address = values[self.gui_osc_address]
changed = True
if self.config.gui_osc_recenter_address != values[self.gui_osc_recenter_address]:
if (
self.config.gui_osc_recenter_address
!= values[self.gui_osc_recenter_address]
):
self.config.gui_osc_recenter_address = values[self.gui_osc_recenter_address]
changed = True
if self.config.gui_osc_recalibrate_address != values[self.gui_osc_recalibrate_address]:
self.config.gui_osc_recalibrate_address = values[self.gui_osc_recalibrate_address]
if (
self.config.gui_osc_recalibrate_address
!= values[self.gui_osc_recalibrate_address]
):
self.config.gui_osc_recalibrate_address = values[
self.gui_osc_recalibrate_address
]
changed = True
if self.config.gui_min_cutoff != values[self.gui_min_cutoff]:
self.config.gui_min_cutoff = values[self.gui_min_cutoff]
changed = True
if self.config.gui_speed_coefficient != values[self.gui_speed_coefficient]:
self.config.gui_speed_coefficient = values[self.gui_speed_coefficient]
changed = True
@ -267,10 +307,18 @@ class SettingsWidget:
self.config.gui_flip_x_axis_left = values[self.gui_flip_x_axis_left]
changed = True
if self.config.gui_osc_vrcft_v1 != values[self.gui_osc_vrcft_v1]:
self.config.gui_osc_vrcft_v1 = values[self.gui_osc_vrcft_v1]
changed = True
if self.config.gui_osc_vrcft_v2 != values[self.gui_osc_vrcft_v2]:
self.config.gui_osc_vrcft_v2 = values[self.gui_osc_vrcft_v2]
changed = True
if self.config.gui_vrc_native != values[self.gui_vrc_native]:
self.config.gui_vrc_native = values[self.gui_vrc_native]
changed = True
if self.config.gui_update_check != values[self.gui_update_check]:
self.config.gui_update_check = values[self.gui_update_check]
changed = True
@ -278,7 +326,7 @@ class SettingsWidget:
if self.config.gui_flip_y_axis != values[self.gui_flip_y_axis]:
self.config.gui_flip_y_axis = values[self.gui_flip_y_axis]
changed = True
if self.config.gui_eye_falloff != values[self.gui_eye_falloff]:
self.config.gui_eye_falloff = values[self.gui_eye_falloff]
changed = True

View File

@ -37,7 +37,7 @@ import psutil
import sys
process = psutil.Process(os.getpid()) # set process priority to low
try: # medium chance this does absolutely nothing but eh
try: # medium chance this does absolutely nothing but eh
sys.getwindowsversion()
except AttributeError:
process.nice(0) # UNIX: 0 low 10 high
@ -46,6 +46,7 @@ else:
process.nice(psutil.BELOW_NORMAL_PRIORITY_CLASS) # Windows
process.nice()
class EyeId(IntEnum):
RIGHT = 0
LEFT = 1
@ -247,7 +248,7 @@ class IntensityBasedOpeness:
upper_y = min(int_y + 25, frame.shape[0] - 1)
lower_y = max(int_y - 25, 0)
# frame_crop = frame[lower_y:upper_y, lower_x:upper_x]
# frame_crop = frame[lower_y:upper_y, lower_x:upper_x]
# frame = safe_crop(frame, lower_x, lower_y, upper_x, upper_y, False)
# ret_, th = cv2.threshold(frame_crop, 80, 1.0, cv2.THRESH_BINARY_INV, dst=frame_crop)
frame_crop = frame
@ -274,10 +275,10 @@ class IntensityBasedOpeness:
intensity = self.maxval
# if intensity <= np.percentile( # TODO test this
# self.filterlist, 0.3
# ): # filter abnormally low values
# print('filter, assume blink')
# intensity = self.data[int_y, int_x]
# self.filterlist, 0.3
# ): # filter abnormally low values
# print('filter, assume blink')
# intensity = self.data[int_y, int_x]
except:
pass
# self.tri_filter.append(intensity)
@ -341,10 +342,9 @@ class IntensityBasedOpeness:
intensitya = max(
data_val + 5000, 1
) # if current intensity value is not less use this is an agressive adjust, test
self.data[int_y, int_x] = intensitya # set value
self.data[int_y, int_x] = intensitya # set value
changed = True
# min pupil global
if self.maxval == 0: # that value is not yet saved
self.maxval = intensity # set value at 0 index
@ -371,7 +371,6 @@ class IntensityBasedOpeness:
) # for whatever reason when input and maxp are too close it outputs high
eyeopen = 1 - eyeopen
if outputSamples > 0:
if len(self.averageList) < outputSamples:
self.averageList.append(eyeopen)
@ -386,8 +385,6 @@ class IntensityBasedOpeness:
if eyeopen < 0:
eyeopen = 0.0
if changed and (
(time.time() - self.lct) > 5
): # save every 5 seconds if something changed to save disk usage
@ -412,4 +409,4 @@ class IntensityBasedOpeness:
# if eyevec > 0.4:
# print("BLINK LCOK")
return eyeopen
return eyeopen

View File

@ -85,7 +85,7 @@ class LEAP_C(object):
else:
self.model_path = resource_path("Models/mommy072623.onnx")
self.interval = 1 # FPS print update rate
self.low_priority = True # set process priority to low
self.low_priority = True # set process priority to low may cause issues
self.print_fps = True
# Init variables
self.frames = 0
@ -106,9 +106,6 @@ class LEAP_C(object):
onnxruntime.GraphOptimizationLevel.ORT_ENABLE_ALL
)
opts.optimized_model_filepath = ""
self.ort_session = onnxruntime.InferenceSession(
self.model_path, opts, providers=["CPUExecutionProvider"]
)
if self.low_priority:
process = psutil.Process(os.getpid()) # set process priority to low
@ -128,9 +125,9 @@ class LEAP_C(object):
self.one_euro_filter = OneEuroFilter(
np.random.rand(7, 2), min_cutoff=min_cutoff, beta=beta
)
#self.one_euro_filter_open = OneEuroFilter(
# np.random.rand(1, 2), min_cutoff=0.01, beta=0.04
#)
# self.one_euro_filter_open = OneEuroFilter(
# np.random.rand(1, 2), min_cutoff=0.01, beta=0.04
# )
self.dmax = 0
self.dmin = 0
self.openlist = []
@ -227,11 +224,11 @@ class LEAP_C(object):
if len(self.openlist) < 5000: # TODO expose as setting?
self.openlist.append(d)
else:
# if d >= np.percentile(self.openlist, 99) or d <= np.percentile(
# if d >= np.percentile(self.openlist, 99) or d <= np.percentile(
# self.openlist, 1
# ):
# ):
# pass
#else:
# else:
self.openlist.pop(0)
self.openlist.append(d)
@ -243,18 +240,18 @@ class LEAP_C(object):
except:
per = 0.7
pass
# print(d, per)
# print(d, per)
x = pre_landmark[6][0]
y = pre_landmark[6][1]
frame = cv2.cvtColor(img, cv2.COLOR_RGB2GRAY)
# per = d - 0.1
# per = d - 0.1
self.last_lid = per
# pera = np.array([per, per])
#self.one_euro_filter_open(pera)
if per <= 0.2: #TODO: EXPOSE AS SETTING
# pera = np.array([per, per])
# self.one_euro_filter_open(pera)
if per <= 0.2: # TODO: EXPOSE AS SETTING
per == 0.0
# print(per)
# print(per)
return frame, float(x), float(y), per
frame = cv2.cvtColor(img, cv2.COLOR_RGB2GRAY)

View File

@ -1,22 +1,25 @@
from pythonosc import udp_client
from pythonosc import osc_server
from pythonosc import dispatcher
from utils.misc_utils import PlaySound,SND_FILENAME,SND_ASYNC
from utils.misc_utils import PlaySound, SND_FILENAME, SND_ASYNC
import queue
import threading
from enum import IntEnum
import time
class EyeId(IntEnum):
RIGHT = 0
LEFT = 1
BOTH = 2
SETTINGS = 3
ALGOSETTINGS = 4
from config import EyeTrackConfig
def eyelid_transformer(self,eye_blink):
def eyelid_transformer(self, eye_blink):
if self.config.osc_invert_eye_close:
return float(1 - eye_blink)
else:
@ -24,158 +27,260 @@ def eyelid_transformer(self,eye_blink):
se = False
def output_osc(eye_x, eye_y, eye_blink, last_blink, self):
global se
global se
# self.config.gui_osc_vrcft_v2
# self.config.gui_osc_vrcft_v1
# self.config.gui_vrc_native
if not self.config.gui_vrc_native:
if self.config.gui_osc_vrcft_v1:
if self.main_config.eye_display_id in [EyeId.RIGHT, EyeId.LEFT]: #we are in single eye mode
se = True
if self.main_config.eye_display_id in [
EyeId.RIGHT,
EyeId.LEFT,
]: # we are in single eye mode
se = True
self.client.send_message(self.config.osc_left_eye_x_address, eye_x)
self.client.send_message(self.config.osc_right_eye_x_address, eye_x)
self.client.send_message(self.config.osc_eyes_y_address, eye_y)
self.config.osc_left_eye_close_address
self.client.send_message(self.config.osc_right_eye_close_address, eyelid_transformer(self,eye_blink))
self.client.send_message(self.config.osc_left_eye_close_address, eyelid_transformer(self,eye_blink))
else:
se = False
self.client.send_message(self.config.osc_left_eye_x_address, eye_x)
self.client.send_message(self.config.osc_right_eye_x_address, eye_x)
self.client.send_message(self.config.osc_eyes_y_address, eye_y)
if self.eye_id in [EyeId.LEFT] and not se: #left eye, send data to left
self.l_eye_x = eye_x
self.l_eye_blink = eye_blink
self.config.osc_left_eye_close_address
if self.l_eye_blink == 0.0:
if last_blink > 0.15: #when binary blink is on, blinks may be too fast for OSC so we repeat them.
for i in range(4):
self.client.send_message(self.config.osc_left_eye_close_address, eyelid_transformer(self,self.l_eye_blink))
last_blink = time.time() - last_blink
if self.config.gui_eye_falloff:
if self.r_eye_blink == 0.0: #if both eyes closed and DEF is enables, blink
self.client.send_message(self.config.osc_left_eye_close_address, eyelid_transformer(self,self.l_eye_blink))
self.client.send_message(self.config.osc_right_eye_close_address, eyelid_transformer(self,self.l_eye_blink))
self.l_eye_x = self.r_eye_x
self.client.send_message(
self.config.osc_right_eye_close_address,
eyelid_transformer(self, eye_blink),
)
self.client.send_message(
self.config.osc_left_eye_close_address,
eyelid_transformer(self, eye_blink),
)
else:
se = False
self.client.send_message(self.config.osc_left_eye_x_address, self.l_eye_x)
self.left_y = eye_y
if self.eye_id in [EyeId.LEFT] and not se: # left eye, send data to left
self.l_eye_x = eye_x
self.l_eye_blink = eye_blink
self.client.send_message(self.config.osc_left_eye_close_address, eyelid_transformer(self,self.l_eye_blink))
if self.l_eye_blink == 0.0:
if (
last_blink > 0.15
): # when binary blink is on, blinks may be too fast for OSC so we repeat them.
for i in range(4):
self.client.send_message(
self.config.osc_left_eye_close_address,
eyelid_transformer(self, self.l_eye_blink),
)
last_blink = time.time() - last_blink
if self.config.gui_eye_falloff:
if (
self.r_eye_blink == 0.0
): # if both eyes closed and DEF is enables, blink
self.client.send_message(
self.config.osc_left_eye_close_address,
eyelid_transformer(self, self.l_eye_blink),
)
self.client.send_message(
self.config.osc_right_eye_close_address,
eyelid_transformer(self, self.l_eye_blink),
)
self.l_eye_x = self.r_eye_x
self.client.send_message(self.config.osc_left_eye_x_address, self.l_eye_x)
self.left_y = eye_y
elif self.eye_id in [EyeId.RIGHT] and not se: #Right eye, send data to right
self.r_eye_x = eye_x
self.r_eye_blink = eye_blink
self.client.send_message(
self.config.osc_left_eye_close_address,
eyelid_transformer(self, self.l_eye_blink),
)
if self.r_eye_blink == 0.0:
if last_blink > 0.15: #when binary blink is on, blinks may be too fast for OSC so we repeat them.
print("REPEATING R BLINK")
for i in range(4):
self.client.send_message(self.config.osc_right_eye_close_address, eyelid_transformer(self,self.r_eye_blink))
last_blink = time.time() - last_blink
if self.config.gui_eye_falloff:
if self.l_eye_blink == 0.0: #if both eyes closed and DEF is enables, blink
self.client.send_message(self.config.osc_left_eye_close_address, eyelid_transformer(self,self.r_eye_blink))
self.client.send_message(self.config.osc_right_eye_close_address, eyelid_transformer(self,self.r_eye_blink))
elif self.eye_id in [EyeId.RIGHT] and not se: # Right eye, send data to right
self.r_eye_x = eye_x
self.r_eye_blink = eye_blink
self.r_eye_x = self.l_eye_x
self.client.send_message(self.config.osc_right_eye_x_address, eye_x)
self.right_y = eye_y
if self.r_eye_blink == 0.0:
if (
last_blink > 0.15
): # when binary blink is on, blinks may be too fast for OSC so we repeat them.
print("REPEATING R BLINK")
for i in range(4):
self.client.send_message(
self.config.osc_right_eye_close_address,
eyelid_transformer(self, self.r_eye_blink),
)
last_blink = time.time() - last_blink
if self.config.gui_eye_falloff:
if (
self.l_eye_blink == 0.0
): # if both eyes closed and DEF is enables, blink
self.client.send_message(
self.config.osc_left_eye_close_address,
eyelid_transformer(self, self.r_eye_blink),
)
self.client.send_message(
self.config.osc_right_eye_close_address,
eyelid_transformer(self, self.r_eye_blink),
)
self.client.send_message(self.config.osc_right_eye_close_address, eyelid_transformer(self,self.r_eye_blink))
self.r_eye_x = self.l_eye_x
if self.main_config.eye_display_id in [EyeId.BOTH] and self.right_y != 621 and self.left_y != 621:
y = (self.right_y + self.left_y) / 2
self.client.send_message(self.config.osc_eyes_y_address, y)
self.client.send_message(self.config.osc_right_eye_x_address, eye_x)
self.right_y = eye_y
self.client.send_message(
self.config.osc_right_eye_close_address,
eyelid_transformer(self, self.r_eye_blink),
)
if (
self.main_config.eye_display_id in [EyeId.BOTH]
and self.right_y != 621
and self.left_y != 621
):
y = (self.right_y + self.left_y) / 2
self.client.send_message(self.config.osc_eyes_y_address, y)
else: # VRC NATIVE
if self.config.gui_vrc_native: # VRC NATIVE
if self.main_config.eye_display_id in [EyeId.RIGHT, EyeId.LEFT]: # we are in single eye mode
se = True
if eye_blink == 0.0:
if last_blink > 0.2: # when binary blink is on, blinks may be too fast for OSC so we repeat them.
for i in range(5):
self.client.send_message("/tracking/eye/EyesClosedAmount",
float(1 - eye_blink))
eye_blink += 0.02 #TODO finish tuning value
last_blink = time.time() - last_blink
else:
self.client.send_message("/tracking/eye/EyesClosedAmount", float(1 - eye_blink))
self.client.send_message("/tracking/eye/LeftRightVec", [float(eye_x), float(eye_y), 1.0, float(eye_x), float(eye_y), 1.0]) # vrc native ET
if self.main_config.eye_display_id in [
EyeId.RIGHT,
EyeId.LEFT,
]: # we are in single eye mode
se = True
if eye_blink == 0.0:
if (
last_blink > 0.2
): # when binary blink is on, blinks may be too fast for OSC so we repeat them.
for i in range(5):
self.client.send_message(
"/tracking/eye/EyesClosedAmount", float(1 - eye_blink)
)
eye_blink += 0.02 # TODO finish tuning value
last_blink = time.time() - last_blink
else:
se = False
self.client.send_message(
"/tracking/eye/EyesClosedAmount", float(1 - eye_blink)
)
self.client.send_message(
"/tracking/eye/LeftRightVec",
[float(eye_x), float(eye_y), 1.0, float(eye_x), float(eye_y), 1.0],
) # vrc native ET
if self.eye_id in [EyeId.LEFT] and not se: # left eye, send data to left
self.l_eye_x = eye_x
self.l_eye_blink = eye_blink
self.left_y = eye_y
self.client.send_message(self.config.osc_left_eye_close_address,eyelid_transformer(self,eye_blink))
else:
se = False
if self.l_eye_blink == 0.0:
if last_blink > 0.2: # when binary blink is on, blinks may be too fast for OSC so we repeat them.
for i in range(5):
self.client.send_message("/tracking/eye/EyesClosedAmount",
float(1 - eye_blink))
last_blink = time.time() - last_blink
if self.config.gui_eye_falloff:
if self.r_eye_blink == 0.0: # if both eyes closed and DEF is enables, blink
self.client.send_message("/tracking/eye/EyesClosedAmount",
float(1 - eye_blink))
self.l_eye_x = self.r_eye_x
if self.eye_id in [EyeId.LEFT] and not se: # left eye, send data to left
self.l_eye_x = eye_x
self.l_eye_blink = eye_blink
self.left_y = eye_y
self.client.send_message(
self.config.osc_left_eye_close_address,
eyelid_transformer(self, eye_blink),
)
if self.l_eye_blink == 0.0:
if (
last_blink > 0.2
): # when binary blink is on, blinks may be too fast for OSC so we repeat them.
for i in range(5):
self.client.send_message(
"/tracking/eye/EyesClosedAmount", float(1 - eye_blink)
)
last_blink = time.time() - last_blink
if self.config.gui_eye_falloff:
if (
self.r_eye_blink == 0.0
): # if both eyes closed and DEF is enables, blink
self.client.send_message(
"/tracking/eye/EyesClosedAmount", float(1 - eye_blink)
)
self.l_eye_x = self.r_eye_x
elif self.eye_id in [EyeId.RIGHT] and not se: # Right eye, send data to right
self.r_eye_x = eye_x
self.r_eye_blink = eye_blink
self.right_y = eye_y
self.client.send_message(self.config.osc_right_eye_close_address,eyelid_transformer(self,eye_blink))
elif self.eye_id in [EyeId.RIGHT] and not se: # Right eye, send data to right
self.r_eye_x = eye_x
self.r_eye_blink = eye_blink
self.right_y = eye_y
self.client.send_message(
self.config.osc_right_eye_close_address,
eyelid_transformer(self, eye_blink),
)
if self.r_eye_blink == 0.0:
if last_blink > 0.2: # when binary blink is on, blinks may be too fast for OSC so we repeat them.
for i in range(5):
self.client.send_message("/tracking/eye/EyesClosedAmount",
float(1 - eye_blink))
last_blink = time.time() - last_blink
if self.config.gui_eye_falloff:
if self.l_eye_blink == 0.0: # if both eyes closed and DEF is enables, blink
self.client.send_message("/tracking/eye/EyesClosedAmount",
float(0))
if self.r_eye_blink == 0.0:
if (
last_blink > 0.2
): # when binary blink is on, blinks may be too fast for OSC so we repeat them.
for i in range(5):
self.client.send_message(
"/tracking/eye/EyesClosedAmount", float(1 - eye_blink)
)
last_blink = time.time() - last_blink
if self.config.gui_eye_falloff:
if (
self.l_eye_blink == 0.0
): # if both eyes closed and DEF is enables, blink
self.client.send_message(
"/tracking/eye/EyesClosedAmount", float(0)
)
self.r_eye_x = self.l_eye_x
self.r_eye_x = self.l_eye_x
if (
self.main_config.eye_display_id in [EyeId.BOTH]
and self.r_eye_blink != 621
and self.r_eye_blink != 621
):
if self.r_eye_blink == 0.0 or self.l_eye_blink == 0.0:
if (
last_blink > 0.2
): # when binary blink is on, blinks may be too fast for OSC so we repeat them.
for i in range(5):
self.client.send_message(
"/tracking/eye/EyesClosedAmount", float(1)
)
last_blink = time.time() - last_blink
eye_blink = (self.r_eye_blink + self.l_eye_blink) / 2
self.client.send_message(
"/tracking/eye/EyesClosedAmount", float(1 - eye_blink)
)
if self.main_config.eye_display_id in [EyeId.BOTH] and self.r_eye_blink != 621 and self.r_eye_blink != 621:
if self.r_eye_blink == 0.0 or self.l_eye_blink == 0.0:
if last_blink > 0.2: # when binary blink is on, blinks may be too fast for OSC so we repeat them.
for i in range(5):
self.client.send_message("/tracking/eye/EyesClosedAmount",
float(1))
last_blink = time.time() - last_blink
eye_blink = (self.r_eye_blink + self.l_eye_blink) / 2
self.client.send_message("/tracking/eye/EyesClosedAmount",
float(1 - eye_blink))
if (
self.main_config.eye_display_id in [EyeId.BOTH]
and self.right_y != 621
and self.left_y != 621
):
eye_y = (self.right_y + self.left_y) / 2
if self.main_config.eye_display_id in [EyeId.BOTH] and self.right_y != 621 and self.left_y != 621:
eye_y = (self.right_y + self.left_y) / 2
if not se:
# vrc native ET (z values may need tweaking, they act like a scalar)
self.client.send_message("/tracking/eye/LeftRightVec",[float(self.l_eye_x), float(self.left_y), 1.0, float(self.r_eye_x), float(self.right_y), 1.0])
if not se:
# vrc native ET (z values may need tweaking, they act like a scalar)
self.client.send_message(
"/tracking/eye/LeftRightVec",
[
float(self.l_eye_x),
float(self.left_y),
1.0,
float(self.r_eye_x),
float(self.right_y),
1.0,
],
)
class VRChatOSC:
# Use a tuple of blink (true, blinking, false, not), x, y for now.
def __init__(self, cancellation_event: threading.Event, msg_queue: queue.Queue[tuple[bool, int, int]], main_config: EyeTrackConfig,):
# Use a tuple of blink (true, blinking, false, not), x, y for now.
def __init__(
self,
cancellation_event: threading.Event,
msg_queue: queue.Queue[tuple[bool, int, int]],
main_config: EyeTrackConfig,
):
self.main_config = main_config
self.config = main_config.settings
self.client = udp_client.SimpleUDPClient(self.config.gui_osc_address, int(self.config.gui_osc_port)) # use OSC port and address that was set in the config
self.client = udp_client.SimpleUDPClient(
self.config.gui_osc_address, int(self.config.gui_osc_port)
) # use OSC port and address that was set in the config
self.cancellation_event = cancellation_event
self.msg_queue = msg_queue
self.eye_id = EyeId.RIGHT
@ -186,7 +291,6 @@ class VRChatOSC:
self.r_eye_blink = 0.7
self.l_eye_blink = 0.7
def run(self):
start = time.time()
last_blink = time.time()
@ -203,15 +307,22 @@ class VRChatOSC:
class VRChatOSCReceiver:
def __init__(self, cancellation_event: threading.Event, main_config: EyeTrackConfig, eyes: []):
def __init__(
self, cancellation_event: threading.Event, main_config: EyeTrackConfig, eyes: []
):
self.config = main_config.settings
self.cancellation_event = cancellation_event
self.dispatcher = dispatcher.Dispatcher()
self.eyes = eyes # we cant import CameraWidget so any type it is
try:
self.server = osc_server.OSCUDPServer((self.config.gui_osc_address, int(self.config.gui_osc_receiver_port)), self.dispatcher)
self.server = osc_server.OSCUDPServer(
(self.config.gui_osc_address, int(self.config.gui_osc_receiver_port)),
self.dispatcher,
)
except:
print(f"\033[91m[ERROR] OSC Receive port: {self.config.gui_osc_receiver_port} occupied.\033[0m")
print(
f"\033[91m[ERROR] OSC Receive port: {self.config.gui_osc_receiver_port} occupied.\033[0m"
)
def shutdown(self):
print("\033[94m[INFO] Exiting OSC Receiver\033[0m")
@ -221,28 +332,40 @@ class VRChatOSCReceiver:
pass
def recenter_eyes(self, address, osc_value):
if type(osc_value) != bool: return # just incase we get anything other than bool
if type(osc_value) != bool:
return # just incase we get anything other than bool
if osc_value:
for eye in self.eyes:
eye.settings.gui_recenter_eyes = True
def recalibrate_eyes(self, address, osc_value):
if type(osc_value) != bool: return # just incase we get anything other than bool
if type(osc_value) != bool:
return # just incase we get anything other than bool
if osc_value:
for eye in self.eyes:
eye.ransac.ibo.clear_filter()
eye.ransac.calibration_frame_counter = self.config.calibration_samples
PlaySound('Audio/start.wav', SND_FILENAME | SND_ASYNC)
PlaySound("Audio/start.wav", SND_FILENAME | SND_ASYNC)
def run(self):
# bind what function to run when specified OSC message is received
try:
self.dispatcher.map(self.config.gui_osc_recalibrate_address, self.recalibrate_eyes)
self.dispatcher.map(self.config.gui_osc_recenter_address, self.recenter_eyes)
self.dispatcher.map(
self.config.gui_osc_recalibrate_address, self.recalibrate_eyes
)
self.dispatcher.map(
self.config.gui_osc_recenter_address, self.recenter_eyes
)
# start the server
print("\033[92m[INFO] VRChatOSCReceiver serving on {}\033[0m".format(self.server.server_address))
print(
"\033[92m[INFO] VRChatOSCReceiver serving on {}\033[0m".format(
self.server.server_address
)
)
self.server.serve_forever()
except:
print(f"\033[91m[ERROR] OSC Receive port: {self.config.gui_osc_receiver_port} occupied.\033[0m")
print(
f"\033[91m[ERROR] OSC Receive port: {self.config.gui_osc_receiver_port} occupied.\033[0m"
)