feat: more readable logging with colorama

This commit is contained in:
Sebastian Fitt 2023-04-29 14:40:44 +02:00
parent 6d7630cf60
commit 2efa3c3589

View File

@ -1,13 +1,14 @@
from config import EyeTrackConfig
from enum import Enum
import cv2 import cv2
import numpy as np
import queue import queue
import serial import serial
import serial.tools.list_ports import serial.tools.list_ports
import threading import threading
import time import time
import numpy as np from colorama import Fore
from config import EyeTrackConfig
from enum import Enum
WAIT_TIME = 0.1 WAIT_TIME = 0.1
@ -55,7 +56,7 @@ class Camera:
self.start = True self.start = True
self.buffer = b'' self.buffer = b''
self.error_message = "\033[93m[WARN] Capture source {} not found, retrying...\033[0m" self.error_message = f"{Fore.YELLOW}[WARN] Capture source {{}} not found, retrying...{Fore.RESET}"
def __del__(self): def __del__(self):
if self.serial_connection is not None: if self.serial_connection is not None:
@ -67,7 +68,7 @@ class Camera:
def run(self): def run(self):
while True: while True:
if self.cancellation_event.is_set(): if self.cancellation_event.is_set():
print("\033[94m[INFO] Exiting Capture thread\033[0m") print(f"{Fore.CYAN}[INFO] Exiting Capture thread{Fore.RESET}")
return return
should_push = True should_push = True
# If things aren't open, retry until they are. Don't let read requests come in any earlier # If things aren't open, retry until they are. Don't let read requests come in any earlier
@ -130,7 +131,7 @@ class Camera:
if should_push: if should_push:
self.push_image_to_queue(image, frame_number, self.fps) self.push_image_to_queue(image, frame_number, self.fps)
except: except:
print("\033[93m[WARN] Capture source problem, assuming camera disconnected, waiting for reconnect.\033[0m") print(f"{Fore.YELLOW}[WARN] Capture source problem, assuming camera disconnected, waiting for reconnect.{Fore.RESET}")
self.camera_status = CameraState.DISCONNECTED self.camera_status = CameraState.DISCONNECTED
pass pass
@ -164,12 +165,12 @@ class Camera:
# Create jpeg frame from byte string # Create jpeg frame from byte string
image = cv2.imdecode(np.fromstring(jpeg, dtype=np.uint8), cv2.IMREAD_UNCHANGED) image = cv2.imdecode(np.fromstring(jpeg, dtype=np.uint8), cv2.IMREAD_UNCHANGED)
if image is None: if image is None:
print("\033[93m[WARN] Frame drop. Corrupted JPEG.\033[0m") print(f"{Fore.YELLOW}[WARN] Frame drop. Corrupted JPEG.{Fore.RESET}")
return return
# Discard the serial buffer. This is due to the fact that it # Discard the serial buffer. This is due to the fact that it
# may build up some outdated frames. A bit of a workaround here tbh. # may build up some outdated frames. A bit of a workaround here tbh.
if self.serial_connection.in_waiting > 32768: if self.serial_connection.in_waiting > 32768:
print(f"\033[94m[INFO] Discarding the serial buffer ({self.serial_connection.in_waiting} bytes\033[0m") print(f"{Fore.CYAN}[INFO] Discarding the serial buffer ({self.serial_connection.in_waiting} bytes{Fore.RESET}")
self.serial_connection.reset_input_buffer() self.serial_connection.reset_input_buffer()
self.buffer = b'' self.buffer = b''
# Calculate the fps. # Calculate the fps.
@ -183,7 +184,7 @@ class Camera:
if should_push: if should_push:
self.push_image_to_queue(image, self.frame_number, self.fps) self.push_image_to_queue(image, self.frame_number, self.fps)
except Exception: except Exception:
print("\033[93m[WARN]Serial capture source problem, assuming camera disconnected, waiting for reconnect.\033[0m") print(f"{Fore.YELLOW}[WARN] Serial capture source problem, assuming camera disconnected, waiting for reconnect.{Fore.RESET}")
self.serial_connection.close() self.serial_connection.close()
self.camera_status = CameraState.DISCONNECTED self.camera_status = CameraState.DISCONNECTED
pass pass
@ -209,11 +210,11 @@ class Camera:
# Set explicit buffer size for serial. # Set explicit buffer size for serial.
conn.set_buffer_size(rx_size = 65536, tx_size = 65536) conn.set_buffer_size(rx_size = 65536, tx_size = 65536)
print(f"\033[94m[INFO] ETVR Serial Tracker device connected on {port}\033[0m") print(f"{Fore.CYAN}[INFO] ETVR Serial Tracker device connected on {port}{Fore.RESET}")
self.serial_connection = conn self.serial_connection = conn
self.camera_status = CameraState.CONNECTED self.camera_status = CameraState.CONNECTED
except Exception: except Exception:
print(f"\033[94m[INFO] Failed to connect on {port}\033[0m") print(f"{Fore.CYAN}[INFO] Failed to connect on {port}{Fore.RESET}")
self.camera_status = CameraState.DISCONNECTED self.camera_status = CameraState.DISCONNECTED
def push_image_to_queue(self, image, frame_number, fps): def push_image_to_queue(self, image, frame_number, fps):
@ -222,6 +223,6 @@ class Camera:
qsize = self.camera_output_outgoing.qsize() qsize = self.camera_output_outgoing.qsize()
if qsize > 1: if qsize > 1:
print( print(
f"\033[91m[WARN] CAPTURE QUEUE BACKPRESSURE OF {qsize}. CHECK FOR CRASH OR TIMING ISSUES IN ALGORITHM.\033[0m") 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.camera_output_outgoing.put((image, frame_number, fps))
self.capture_event.clear() self.capture_event.clear()