Reformat the files to use 4 spaces

This commit is contained in:
lorow 2022-06-14 00:03:17 +02:00 committed by qDot
parent d30b6fb37f
commit 1359bff868
4 changed files with 250 additions and 237 deletions

View File

@ -4,32 +4,34 @@ from dacite import from_dict
import os.path
import json
# TODO Who even needs synchronization? (We do.)
@dataclass
class RansacConfig:
threshold: "int" = 0
rotation_angle: "int" = 0
roi_window_x: "int" = 0
roi_window_y: "int" = 0
roi_window_w: "int" = 640
roi_window_h: "int" = 480
focal_length: "int" = 30
capture_source: "Union[int, str, None]" = None
vrc_eye_position_scalar: "int" = 3000
show_color_image: "bool" = False
threshold: "int" = 0
rotation_angle: "int" = 0
roi_window_x: "int" = 0
roi_window_y: "int" = 0
roi_window_w: "int" = 640
roi_window_h: "int" = 480
focal_length: "int" = 30
capture_source: "Union[int, str, None]" = None
vrc_eye_position_scalar: "int" = 3000
show_color_image: "bool" = False
def load():
if not os.path.exists("ransac_settings.json"):
print("No settings file, using base settings")
return RansacConfig()
with open("ransac_settings.json", 'r') as settings_file:
try:
return from_dict(data_class = RansacConfig, data = json.load(settings_file))
except:
print("Configuration invalid, creating new config")
return RansacConfig()
@staticmethod
def load():
if not os.path.exists("ransac_settings.json"):
print("No settings file, using base settings")
return RansacConfig()
with open("ransac_settings.json", 'r') as settings_file:
try:
return from_dict(data_class=RansacConfig, data=json.load(settings_file))
except:
print("Configuration invalid, creating new config")
return RansacConfig()
def save(self):
with open("ransac_settings.json", 'w+') as settings_file:
json.dump(self.__dict__, settings_file)
def save(self):
with open("ransac_settings.json", 'w+') as settings_file:
json.dump(self.__dict__, settings_file)

View File

@ -27,198 +27,207 @@ MODE_READOUT_NAME = "-APPMODE-"
SHOW_COLOR_IMAGE_NAME = "-SHOWCOLORIMAGE-"
def main():
in_roi_mode = False
in_roi_mode = False
# Get Configuration
config: RansacConfig = RansacConfig.load()
config.save()
# Get Configuration
config: RansacConfig = RansacConfig.load()
config.save()
roi_layout = [
[sg.Graph((640, 480), (0, 480), (640, 0), key=ROI_SELECTION_NAME,drag_submits=True, enable_events=True)]
]
roi_layout = [
[sg.Graph((640, 480), (0, 480), (640, 0), key=ROI_SELECTION_NAME, drag_submits=True, enable_events=True)]
]
# Define the window's contents
tracking_layout = [
[sg.Text("Threshold"), sg.Slider(range=(0, 100), default_value=config.threshold, orientation = 'h', key=THRESHOLD_SLIDER_NAME)],
[sg.Text("Rotation"), sg.Slider(range=(0, 360), default_value=config.rotation_angle, orientation = 'h', key=ROTATION_SLIDER_NAME)],
[sg.Text("Eye Position Scalar"), sg.Slider(range=(0, 5000), default_value=config.vrc_eye_position_scalar, orientation = 'h', key=SCALAR_SLIDER_NAME)],
[sg.Button("Restart Calibration", key=RESTART_CALIBRATION_NAME), sg.Button("Recenter Eye", key=RECENTER_EYE_NAME), sg.Checkbox('Show Color Image:', default=config.show_color_image, key=SHOW_COLOR_IMAGE_NAME)],
[sg.Text("Mode:"), sg.Text("Calibrating", key=MODE_READOUT_NAME)],
[sg.Image(filename="", key=TRACKING_IMAGE_NAME)],
[sg.Graph((200,200), (-100, 100), (100, -100), background_color='white', key=OUTPUT_GRAPH_NAME,drag_submits=True, enable_events=True)]
]
# Define the window's contents
tracking_layout = [
[sg.Text("Threshold"),
sg.Slider(range=(0, 100), default_value=config.threshold, orientation='h', key=THRESHOLD_SLIDER_NAME)],
[sg.Text("Rotation"),
sg.Slider(range=(0, 360), default_value=config.rotation_angle, orientation='h', key=ROTATION_SLIDER_NAME)],
[sg.Text("Eye Position Scalar"),
sg.Slider(range=(0, 5000), default_value=config.vrc_eye_position_scalar, orientation='h',
key=SCALAR_SLIDER_NAME)],
[sg.Button("Restart Calibration", key=RESTART_CALIBRATION_NAME),
sg.Button("Recenter Eye", key=RECENTER_EYE_NAME),
sg.Checkbox('Show Color Image:', default=config.show_color_image, key=SHOW_COLOR_IMAGE_NAME)],
[sg.Text("Mode:"), sg.Text("Calibrating", key=MODE_READOUT_NAME)],
[sg.Image(filename="", key=TRACKING_IMAGE_NAME)],
[sg.Graph((200, 200), (-100, 100), (100, -100), background_color='white', key=OUTPUT_GRAPH_NAME,
drag_submits=True, enable_events=True)]
]
layout = [[[sg.Text("Camera Address"), sg.InputText(config.capture_source, key=CAMERA_ADDR_NAME), sg.Button("Save and Restart Tracking", key=SAVE_TRACKING_BUTTON_NAME)]],
[sg.Button("Tracking Mode", key=TRACKING_BUTTON_NAME), sg.Button("ROI Mode", key=ROI_BUTTON_NAME)],
[sg.Column(tracking_layout, key=TRACKING_LAYOUT_NAME), sg.Column(roi_layout, key=ROI_LAYOUT_NAME, visible=False)]]
layout = [[[sg.Text("Camera Address"), sg.InputText(config.capture_source, key=CAMERA_ADDR_NAME),
sg.Button("Save and Restart Tracking", key=SAVE_TRACKING_BUTTON_NAME)]],
[sg.Button("Tracking Mode", key=TRACKING_BUTTON_NAME), sg.Button("ROI Mode", key=ROI_BUTTON_NAME)],
[sg.Column(tracking_layout, key=TRACKING_LAYOUT_NAME),
sg.Column(roi_layout, key=ROI_LAYOUT_NAME, visible=False)]]
# Create the window
window = sg.Window('Eye Tracking', layout)
# Create the window
window = sg.Window('Eye Tracking', layout)
cancellation_event = threading.Event()
cancellation_event = threading.Event()
# Check to see if we can connect to our video source first. If not, bring up camera finding
# dialog.
# Check to see if we can connect to our video source first. If not, bring up camera finding
# dialog.
# Check to see if we have an ROI. If not, bring up ROI finder GUI.
# Check to see if we have an ROI. If not, bring up ROI finder GUI.
# Spawn worker threads
osc_queue: "queue.Queue[tuple[bool, int, int]]" = queue.Queue()
osc = VRChatOSC(cancellation_event, osc_queue)
osc_thread = threading.Thread(target=osc.run)
osc_thread.start()
# Spawn worker threads
osc_queue: "queue.Queue[tuple[bool, int, int]]" = queue.Queue()
osc = VRChatOSC(cancellation_event, osc_queue)
osc_thread = threading.Thread(target=osc.run)
osc_thread.start()
# t2s_queue: "queue.Queue[str | None]" = queue.Queue()
# t2s_engine = SpeechEngine(t2s_queue)
# t2s_thread = threading.Thread(target=t2s_engine.run)
# t2s_thread.start()
# t2s_queue.put("App Starting")
# t2s_queue: "queue.Queue[str | None]" = queue.Queue()
# t2s_engine = SpeechEngine(t2s_queue)
# t2s_thread = threading.Thread(target=t2s_engine.run)
# t2s_thread.start()
# t2s_queue.put("App Starting")
capture_event = threading.Event()
capture_queue = queue.Queue()
roi_queue = queue.Queue()
capture_event = threading.Event()
capture_queue = queue.Queue()
roi_queue = queue.Queue()
image_queue: queue.Queue = queue.Queue()
ransac = Ransac(config, cancellation_event, capture_event, capture_queue, image_queue)
ransac_thread = threading.Thread(target=ransac.run)
ransac_thread.start()
image_queue: queue.Queue = queue.Queue()
ransac = Ransac(config, cancellation_event, capture_event, capture_queue, image_queue)
ransac_thread = threading.Thread(target=ransac.run)
ransac_thread.start()
# Only start our camera AFTER we've brought up the RANSAC thread, otherwise we'll have no consumer
camera_status_queue = queue.Queue()
camera_0 = camera.Camera(config, 0, cancellation_event, capture_event, camera_status_queue, capture_queue)
camera_0_thread = threading.Thread(target=camera_0.run)
camera_0_thread.start()
# Only start our camera AFTER we've brought up the RANSAC thread, otherwise we'll have no consumer
camera_status_queue = queue.Queue()
camera_0 = camera.Camera(config, 0, cancellation_event, capture_event, camera_status_queue, capture_queue)
camera_0_thread = threading.Thread(target=camera_0.run)
camera_0_thread.start()
x0, y0 = None, None
x1, y1 = None, None
figure = None
is_mouse_up = True
x0, y0 = None, None
x1, y1 = None, None
figure = None
is_mouse_up = True
# GUI Render loop
# GUI Render loop
while True:
# First off, check for any events from the GUI
event, values = window.read(timeout=1)
while True:
# First off, check for any events from the GUI
event, values = window.read(timeout=1)
# If we're in either mode and someone hits q, quit immediately
if event == "Exit" or event == sg.WIN_CLOSED:
cancellation_event.set()
osc_thread.join()
ransac_thread.join()
# t2s_engine.force_stop()
# t2s_queue.put(None)
# t2s_thread.join()
print("Exiting RANSAC App")
return
# If we're in either mode and someone hits q, quit immediately
if event == "Exit" or event == sg.WIN_CLOSED:
cancellation_event.set()
osc_thread.join()
ransac_thread.join()
# t2s_engine.force_stop()
# t2s_queue.put(None)
# t2s_thread.join()
print("Exiting RANSAC App")
return
changed = False
# If anything has changed in our configuration settings, change/update those.
if event == SAVE_TRACKING_BUTTON_NAME and values[CAMERA_ADDR_NAME] != config.capture_source:
try:
# Try storing ints as ints, for those using wired cameras.
config.capture_source = int(values[CAMERA_ADDR_NAME])
except:
config.capture_source = values[CAMERA_ADDR_NAME]
changed = True
changed = False
# If anything has changed in our configuration settings, change/update those.
if event == SAVE_TRACKING_BUTTON_NAME and values[CAMERA_ADDR_NAME] != config.capture_source:
try:
# Try storing ints as ints, for those using wired cameras.
config.capture_source = int(values[CAMERA_ADDR_NAME])
except ValueError:
config.capture_source = values[CAMERA_ADDR_NAME]
changed = True
if config.threshold != values[THRESHOLD_SLIDER_NAME]:
config.threshold = int(values[THRESHOLD_SLIDER_NAME])
changed = True
if config.threshold != values[THRESHOLD_SLIDER_NAME]:
config.threshold = int(values[THRESHOLD_SLIDER_NAME])
changed = True
if config.rotation_angle != values[ROTATION_SLIDER_NAME]:
config.rotation_angle = int(values[ROTATION_SLIDER_NAME])
changed = True
if config.rotation_angle != values[ROTATION_SLIDER_NAME]:
config.rotation_angle = int(values[ROTATION_SLIDER_NAME])
changed = True
if config.vrc_eye_position_scalar != values[SCALAR_SLIDER_NAME]:
config.vrc_eye_position_scalar = int(values[SCALAR_SLIDER_NAME])
changed = True
if config.vrc_eye_position_scalar != values[SCALAR_SLIDER_NAME]:
config.vrc_eye_position_scalar = int(values[SCALAR_SLIDER_NAME])
changed = True
if config.show_color_image != values[SHOW_COLOR_IMAGE_NAME]:
config.show_color_image = values[SHOW_COLOR_IMAGE_NAME]
changed = True
if config.show_color_image != values[SHOW_COLOR_IMAGE_NAME]:
config.show_color_image = values[SHOW_COLOR_IMAGE_NAME]
changed = True
if changed:
config.save()
if changed:
config.save()
if event == TRACKING_BUTTON_NAME:
print("Moving to tracking mode")
in_roi_mode = False
camera_0.set_output_queue(capture_queue)
window[ROI_LAYOUT_NAME].update(visible=False)
window[TRACKING_LAYOUT_NAME].update(visible=True)
elif event == ROI_BUTTON_NAME:
print("move to roi mode")
in_roi_mode = True
camera_0.set_output_queue(roi_queue)
window[ROI_LAYOUT_NAME].update(visible=True)
window[TRACKING_LAYOUT_NAME].update(visible=False)
elif event == '-GRAPH-+UP':
# Event for mouse button up in ROI mode
is_mouse_up = True
if abs(x0-x1) != 0 and abs(y0-y1) != 0:
config.roi_window_x = min([x0, x1])
config.roi_window_y = min([y0, y1])
config.roi_window_w = abs(x0-x1)
config.roi_window_h = abs(y0-y1)
config.save()
elif event == '-GRAPH-':
# Event for mouse button down or mouse drag in ROI mode
if is_mouse_up:
is_mouse_up = False
x0, y0 = values['-GRAPH-']
x1, y1 = values['-GRAPH-']
elif event == RESTART_CALIBRATION_NAME:
ransac.calibration_frame_counter = 300
elif event == RECENTER_EYE_NAME:
ransac.recenter_eye = True
if event == TRACKING_BUTTON_NAME:
print("Moving to tracking mode")
in_roi_mode = False
camera_0.set_output_queue(capture_queue)
window[ROI_LAYOUT_NAME].update(visible=False)
window[TRACKING_LAYOUT_NAME].update(visible=True)
elif event == ROI_BUTTON_NAME:
print("move to roi mode")
in_roi_mode = True
camera_0.set_output_queue(roi_queue)
window[ROI_LAYOUT_NAME].update(visible=True)
window[TRACKING_LAYOUT_NAME].update(visible=False)
elif event == '-GRAPH-+UP':
# Event for mouse button up in ROI mode
is_mouse_up = True
if abs(x0 - x1) != 0 and abs(y0 - y1) != 0:
config.roi_window_x = min([x0, x1])
config.roi_window_y = min([y0, y1])
config.roi_window_w = abs(x0 - x1)
config.roi_window_h = abs(y0 - y1)
config.save()
elif event == '-GRAPH-':
# Event for mouse button down or mouse drag in ROI mode
if is_mouse_up:
is_mouse_up = False
x0, y0 = values['-GRAPH-']
x1, y1 = values['-GRAPH-']
elif event == RESTART_CALIBRATION_NAME:
ransac.calibration_frame_counter = 300
elif event == RECENTER_EYE_NAME:
ransac.recenter_eye = True
if ransac.calibration_frame_counter != None:
window[MODE_READOUT_NAME].update("Calibration")
else:
window[MODE_READOUT_NAME].update("Tracking")
if ransac.calibration_frame_counter != None:
window[MODE_READOUT_NAME].update("Calibration")
else:
window[MODE_READOUT_NAME].update("Tracking")
if in_roi_mode:
try:
if roi_queue.empty():
capture_event.set()
maybe_image = roi_queue.get(block = False)
imgbytes = cv2.imencode(".ppm", maybe_image[0])[1].tobytes()
graph = window[ROI_SELECTION_NAME]
if figure:
graph.delete_figure(figure)
# INCREDIBLY IMPORTANT ERASE. Drawing images does NOT overwrite the buffer, the fucking
# graph keeps every image fed in until you call this. Therefore we have to make sure we
# erase before we redraw, otherwise we'll leak memory *very* quickly.
graph.erase()
graph.draw_image(data=imgbytes, location=(0, 0))
if None not in (x0, y0, x1, y1):
figure = graph.draw_rectangle((x0, y0), (x1, y1), line_color='blue')
except queue.Empty:
pass
else:
try:
(maybe_image, eye_info) = image_queue.get(block = False)
imgbytes = cv2.imencode(".ppm", maybe_image)[1].tobytes()
window[TRACKING_IMAGE_NAME].update(data=imgbytes)
if in_roi_mode:
try:
if roi_queue.empty():
capture_event.set()
maybe_image = roi_queue.get(block=False)
imgbytes = cv2.imencode(".ppm", maybe_image[0])[1].tobytes()
graph = window[ROI_SELECTION_NAME]
if figure:
graph.delete_figure(figure)
# INCREDIBLY IMPORTANT ERASE. Drawing images does NOT overwrite the buffer, the fucking
# graph keeps every image fed in until you call this. Therefore we have to make sure we
# erase before we redraw, otherwise we'll leak memory *very* quickly.
graph.erase()
graph.draw_image(data=imgbytes, location=(0, 0))
if None not in (x0, y0, x1, y1):
figure = graph.draw_rectangle((x0, y0), (x1, y1), line_color='blue')
except queue.Empty:
pass
else:
try:
(maybe_image, eye_info) = image_queue.get(block=False)
imgbytes = cv2.imencode(".ppm", maybe_image)[1].tobytes()
window[TRACKING_IMAGE_NAME].update(data=imgbytes)
# Update the GUI
graph = window[OUTPUT_GRAPH_NAME]
graph.erase()
# Update the GUI
graph = window[OUTPUT_GRAPH_NAME]
graph.erase()
if eye_info.info_type != InformationOrigin.FAILURE and not eye_info.blink:
graph.update(background_color = 'white')
graph.draw_circle((eye_info.x * -100, eye_info.y * -100), 25, fill_color='black',line_color='white')
elif eye_info.blink:
graph.update(background_color = 'blue')
elif eye_info.info_type == InformationOrigin.FAILURE:
graph.update(background_color = 'red')
# Relay information to OSC
if eye_info.info_type != InformationOrigin.FAILURE:
osc_queue.put(eye_info)
except queue.Empty:
pass
if eye_info.info_type != InformationOrigin.FAILURE and not eye_info.blink:
graph.update(background_color='white')
graph.draw_circle((eye_info.x * -100, eye_info.y * -100), 25, fill_color='black',line_color='white')
elif eye_info.blink:
graph.update(background_color='blue')
elif eye_info.info_type == InformationOrigin.FAILURE:
graph.update(background_color='red')
# Relay information to OSC
if eye_info.info_type != InformationOrigin.FAILURE:
osc_queue.put(eye_info)
except queue.Empty:
pass
if __name__ == "__main__":
main()
main()

View File

@ -2,40 +2,41 @@ from pythonosc import udp_client
import queue
import threading
class VRChatOSC:
# VRChat OSC Networking Info. For now, we'll assume it's always local.
OSC_IP="127.0.0.1"
OSC_PORT=9000 #VR Chat OSC port
# VRChat OSC Networking Info. For now, we'll assume it's always local.
OSC_IP = "127.0.0.1"
OSC_PORT = 9000 # VR Chat OSC port
# Use a tuple of blink (true, blinking, false, not), x, y for now. Probably clearer as a class but
# we're stuck in python 3.6 so still no dataclasses. God I hate python.
def __init__(self, cancellation_event: "threading.Event", msg_queue: "queue.Queue[tuple[bool, int, int]]"):
self.client = udp_client.SimpleUDPClient(VRChatOSC.OSC_IP, VRChatOSC.OSC_PORT)
self.cancellation_event = cancellation_event
self.msg_queue = msg_queue
# Use a tuple of blink (true, blinking, false, not), x, y for now. Probably clearer as a class but
# we're stuck in python 3.6 so still no dataclasses. God I hate python.
def __init__(self, cancellation_event: "threading.Event", msg_queue: "queue.Queue[tuple[bool, int, int]]"):
self.client = udp_client.SimpleUDPClient(VRChatOSC.OSC_IP, VRChatOSC.OSC_PORT)
self.cancellation_event = cancellation_event
self.msg_queue = msg_queue
def run(self):
# Set blinking status to true when we start, just so we make sure we get to an eyelid open state
# no matter what.
was_blinking = True
while True:
if self.cancellation_event.is_set():
print("Exiting OSC Queue")
return
try:
eye_info = self.msg_queue.get(block=True, timeout=0.1)
except queue.Empty:
continue
# If we're not blinking, set position
if not eye_info.blink:
self.client.send_message("/avatar/parameters/RightEyeX", eye_info.x)
self.client.send_message("/avatar/parameters/LeftEyeX", eye_info.x)
self.client.send_message("/avatar/parameters/EyesY", eye_info.y)
if was_blinking:
self.client.send_message("/avatar/parameters/LeftEyeLid", float(1))
self.client.send_message("/avatar/parameters/RightEyeLid", float(1))
was_blinking = False
else:
self.client.send_message("/avatar/parameters/LeftEyeLid", float(0))
self.client.send_message("/avatar/parameters/RightEyeLid", float(0))
def run(self):
# Set blinking status to true when we start, just so we make sure we get to an eyelid open state
# no matter what.
was_blinking = True
while True:
if self.cancellation_event.is_set():
print("Exiting OSC Queue")
return
try:
eye_info = self.msg_queue.get(block=True, timeout=0.1)
except queue.Empty:
continue
# If we're not blinking, set position
if not eye_info.blink:
self.client.send_message("/avatar/parameters/RightEyeX", eye_info.x)
self.client.send_message("/avatar/parameters/LeftEyeX", eye_info.x)
self.client.send_message("/avatar/parameters/EyesY", eye_info.y)
if was_blinking:
self.client.send_message("/avatar/parameters/LeftEyeLid", float(1))
self.client.send_message("/avatar/parameters/RightEyeLid", float(1))
was_blinking = False
else:
self.client.send_message("/avatar/parameters/LeftEyeLid", float(0))
self.client.send_message("/avatar/parameters/RightEyeLid", float(0))
was_blinking = True

View File

@ -2,24 +2,25 @@ import pyttsx3
import queue
import threading
class SpeechEngine:
def __init__(self, queue: "queue.Queue[str | None]"):
self.engine = pyttsx3.init()
self.queue = queue
def __init__(self, queue: "queue.Queue[str | None]"):
self.engine = pyttsx3.init()
self.queue = queue
def say(self, item):
self.engine.say(item)
def say(self, item):
self.engine.say(item)
def force_stop(self):
self.engine.stop()
def run(self):
while True:
print("Waiting for speech item")
item = self.queue.get()
if item is None:
print("Stopping speech engine")
def force_stop(self):
self.engine.stop()
return
self.engine.say(item)
self.engine.runAndWait()
def run(self):
while True:
print("Waiting for speech item")
item = self.queue.get()
if item is None:
print("Stopping speech engine")
self.engine.stop()
return
self.engine.say(item)
self.engine.runAndWait()