Orginised camera sources. (UDP vs serial + http + cv2)

This commit is contained in:
BOTAlex 2025-03-07 14:08:01 +01:00
parent 1010427e3f
commit 6498a60da6
6 changed files with 232 additions and 66 deletions

View File

@ -0,0 +1,6 @@
from enum import Enum
class CameraState(Enum):
CONNECTING = 0
CONNECTED = 1
DISCONNECTED = 2

View File

@ -0,0 +1,75 @@
import cv2
import numpy as np
import queue
import serial
import serial.tools.list_ports
import threading
import time
from colorama import Fore
from config import EyeTrackCameraConfig
from enum import Enum
import psutil, os
import sys
from Camera.CameraState import CameraState
from abc import ABC, abstractmethod
# This is when C# dev does Python dev
class ICameraSource:
def __init__(
self,
config: EyeTrackCameraConfig,
camera_index: int,
cancellation_event: "threading.Event",
capture_event: "threading.Event",
camera_status_outgoing: "queue.Queue[CameraState]",
camera_output_outgoing: "queue.Queue(maxsize=20)",
):
self.camera_status = CameraState.CONNECTING
self.config = config
self.camera_index = camera_index
self.camera_address = config.capture_source
self.camera_status_outgoing = camera_status_outgoing
self.camera_output_outgoing = camera_output_outgoing
self.capture_event = capture_event
self.cancellation_event = cancellation_event
self.current_capture_source = config.capture_source
self.cv2_camera: "cv2.VideoCapture" = None
self.serial_connection = None
self.last_frame_time = time.time()
self.frame_number = 0
self.fps = 0
self.bps = 0
self.start = True
self.buffer = b""
self.pf_fps = 0
self.prevft = 0
self.newft = 0
self.fl = [0]
self.extraInit()
self.error_message = f"{Fore.YELLOW}[WARN] Capture source {{}} not found, retrying...{Fore.RESET}"
def __del__(self):
pass
def push_image_to_queue(self, image, frame_number, fps):
# If there's backpressure, just yell. We really shouldn't have this unless we start getting
# some sort of capture event conflict though.
qsize = self.camera_output_outgoing.qsize()
if qsize > 1:
print(
f"{Fore.YELLOW}[WARN] CAPTURE QUEUE BACKPRESSURE OF {qsize}. CHECK FOR CRASH OR TIMING ISSUES IN ALGORITHM.{Fore.RESET}"
)
self.camera_output_outgoing.put((image, frame_number, fps))
self.capture_event.clear()
@abstractmethod
def run(self):
pass
def extraInit(self):
pass

View File

@ -0,0 +1,20 @@
import struct
class DataPacket:
def __init__(self, buffer, offset, size):
self.frame_num = None
self.id = None
self.buf_size = None
self.data = None
if len(buffer) < offset + 12: # Ensure buffer is large enough for metadata
return
self.frame_num, self.id, self.buf_size = struct.unpack_from("iii", buffer, offset)
current_index = offset + 12
message_length = size - current_index
if message_length < 0 or current_index + message_length > len(buffer):
return
self.data = buffer[current_index:current_index + message_length]

View File

@ -0,0 +1,97 @@
import cv2
import numpy as np
import queue
import serial
import serial.tools.list_ports
import threading
import time
from colorama import Fore
from config import EyeTrackCameraConfig
from enum import Enum
import psutil, os
import sys
from Camera.CameraState import CameraState
from Camera.ICameraSource import ICameraSource
import socket
from .DataPacket import DataPacket # Python imports stupid. missed a single "."
WAIT_TIME = 0.1
# I HATE PYTHON!!!! slow fucking...
# Never know when value passed by value or ref
# EVERYTHING is lowercase, EXCEPT True and False. ok
# const is UPPERCASE, but still modifiable. ok
# Unreadable sudo code language
# forced upon me by every industry. VR, AI, Websites? bruh
# Rant over
class UDP_Camera(ICameraSource):
def extraInit(self):
process = psutil.Process(os.getpid()) # set process priority to low
try:
sys.getwindowsversion()
except AttributeError:
process.nice(10) # UNIX: 0 low 10 high
process.nice()
else:
process.nice(psutil.HIGH_PRIORITY_CLASS) # Windows
process.nice()
# See https://learn.microsoft.com/en-us/windows/win32/api/processthreadsapi/nf-processthreadsapi-getpriorityclass#return-value for values
self.host = "0.0.0.0"
self.port = 3333
self.num_fragments = 48
self.num_loaded = 0
self.packets = [None] * self.num_fragments
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.sock.bind((self.host, self.port))
def run(self):
while True:
if self.cancellation_event.is_set():
print(f"{Fore.CYAN}[INFO] Exiting Capture thread{Fore.RESET}")
# openCV won't switch to a new source if provided with one
# so, we have to manually release the camera on exit
return
else:
# We don't have a capture source to try yet, wait for one to show up in the GUI.
if self.cancellation_event.wait(WAIT_TIME):
self.camera_status = CameraState.DISCONNECTED
return
data, addr = self.sock.recvfrom(1024)
self.handle_packet(data, addr)
def handle_packet(self, data, addr):
packet = DataPacket(data, 0, len(data))
if packet.data is None or packet.id < 0 or packet.id >= self.num_fragments:
return
if packet.id == 0:
self.num_loaded = 1
self.packets = [None] * self.num_fragments
self.packets[packet.id] = packet
elif self.packets[packet.id] is None:
self.packets[packet.id] = packet
self.num_loaded += 1
if self.num_loaded >= self.num_fragments:
if self.packets[0] is None:
return
current_frame = self.packets[0].frame_num
if any(p is None or p.frame_num != current_frame for p in self.packets):
print("Packet loss detected.")
else:
print("Whole data received. Processing...")
data_stream = [b for p in self.packets for b in p.data]
print(data_stream)
self.num_loaded = 0
# Send acknowledgment
self.sock.sendto(f"{packet.id}:ACK".encode(), addr)

View File

@ -36,6 +36,10 @@ from config import EyeTrackCameraConfig
from enum import Enum
import psutil, os
import sys
from Camera.CameraState import CameraState
from Camera.ICameraSource import ICameraSource
process = psutil.Process(os.getpid()) # set process priority to low
@ -60,11 +64,6 @@ ETVR_HEADER_FRAME = b"\xff\xa1"
ETVR_HEADER_LEN = 6
class CameraState(Enum):
CONNECTING = 0
CONNECTED = 1
DISCONNECTED = 2
def is_serial_capture_source(addr: str) -> bool:
"""
@ -74,49 +73,11 @@ def is_serial_capture_source(addr: str) -> bool:
addr.startswith("COM") or addr.startswith("/dev/cu") or addr.startswith("/dev/tty") # Windows # macOS # Linux
)
class Camera:
def __init__(
self,
config: EyeTrackCameraConfig,
camera_index: int,
cancellation_event: "threading.Event",
capture_event: "threading.Event",
camera_status_outgoing: "queue.Queue[CameraState]",
camera_output_outgoing: "queue.Queue(maxsize=20)",
):
self.camera_status = CameraState.CONNECTING
self.config = config
self.camera_index = camera_index
self.camera_address = config.capture_source
self.camera_status_outgoing = camera_status_outgoing
self.camera_output_outgoing = camera_output_outgoing
self.capture_event = capture_event
self.cancellation_event = cancellation_event
self.current_capture_source = config.capture_source
self.cv2_camera: "cv2.VideoCapture" = None
self.serial_connection = None
self.last_frame_time = time.time()
self.frame_number = 0
self.fps = 0
self.bps = 0
self.start = True
self.buffer = b""
self.pf_fps = 0
self.prevft = 0
self.newft = 0
self.fl = [0]
def is_udp(addr: str) -> bool:
return addr.find("UDP") >= 0
self.error_message = f"{Fore.YELLOW}[WARN] Capture source {{}} not found, retrying...{Fore.RESET}"
def __del__(self):
if self.serial_connection is not None:
self.serial_connection.close()
class Camera(ICameraSource):
def set_output_queue(self, camera_output_outgoing: "queue.Queue"):
self.camera_output_outgoing = camera_output_outgoing
@ -147,6 +108,7 @@ class Camera:
if self.config.capture_source != None and self.config.capture_source != "":
self.current_capture_source = self.config.capture_source
addr = str(self.current_capture_source)
if is_serial_capture_source(addr):
if (
self.serial_connection is None
@ -156,6 +118,9 @@ class Camera:
port = self.config.capture_source
self.current_capture_source = port
self.start_serial_connection(port)
elif (is_udp(addr)):
# print("UDP selected!")
setup_udp(self)
else:
if (
self.cv2_camera is None
@ -190,6 +155,8 @@ class Camera:
addr = str(self.current_capture_source)
if is_serial_capture_source(addr):
self.get_serial_camera_picture(should_push)
elif is_udp(addr):
udp_get_image(self)
else:
self.get_cv2_camera_picture(should_push)
if not should_push:
@ -328,14 +295,3 @@ class Camera:
except Exception:
print(f"{Fore.CYAN}[INFO] Failed to connect on {port}{Fore.RESET}")
self.camera_status = CameraState.DISCONNECTED
def push_image_to_queue(self, image, frame_number, fps):
# If there's backpressure, just yell. We really shouldn't have this unless we start getting
# some sort of capture event conflict though.
qsize = self.camera_output_outgoing.qsize()
if qsize > 1:
print(
f"{Fore.YELLOW}[WARN] CAPTURE QUEUE BACKPRESSURE OF {qsize}. CHECK FOR CRASH OR TIMING ISSUES IN ALGORITHM.{Fore.RESET}"
)
self.camera_output_outgoing.put((image, frame_number, fps))
self.capture_event.clear()

View File

@ -32,7 +32,8 @@ import math
from eye import EyeId
from eye_processor import EyeProcessor, EyeInfoOrigin
from queue import Queue, Empty
from camera import Camera, CameraState
from Camera.camera import Camera, CameraState
from Camera.UDP_Camera.UDP_Camera import UDP_Camera
import cv2
from osc.OSCMessage import OSCMessageType, OSCMessage
from utils.misc_utils import PlaySound, SND_FILENAME, SND_ASYNC, resource_path
@ -104,14 +105,25 @@ class CameraWidget:
)
self.camera_status_queue = Queue()
self.camera = Camera(
self.config,
0,
self.cancellation_event,
self.capture_event,
self.camera_status_queue,
self.capture_queue,
)
if not self.config.capture_source is None and "UDP" in self.config.capture_source: # Not the best code but I don't care. I want to be finished
print("UDP selected. Prepare for bugs from BOTAlex")
self.camera = UDP_Camera(
self.config,
0,
self.cancellation_event,
self.capture_event,
self.camera_status_queue,
self.capture_queue,
)
else:
self.camera = Camera(
self.config,
0,
self.cancellation_event,
self.capture_event,
self.camera_status_queue,
self.capture_queue,
)
self.hover = None