import cv2
import numpy as np
import sqlite3
from datetime import datetime
import soundfile as sf
import yamnet
import tensorflow as tf
from scipy.spatial.transform import Rotation as R
# Initialize YAMNet model for sound event detection
model = yamnet.yamnet_model.YAMNet()
class_names = yamnet.class_names
# Function to detect face direction as a vector using yaw and pitch
def calculate_face_direction(yaw, pitch):
# Convert yaw and pitch to radians
yaw_rad = np.radians(yaw)
pitch_rad = np.radians(pitch)
# Face direction vector (unit vector representation)
direction_vector = np.array([np.sin(yaw_rad) * np.cos(pitch_rad), np.sin(pitch_rad), np.cos(yaw_rad) * np.cos(pitch_rad)])
# Normalize the direction vector
direction_vector = direction_vector / np.linalg.norm(direction_vector)
return direction_vector
# Initialize OpenCV face detection (using MTCNN or Haar cascades)
face_cascade = cv2.CascadeClassifier(cv2.data.haarcascades + 'haarcascade_frontalface_default.xml')
# Initialize database
def create_database():
conn = sqlite3.connect('event_tracking.db')
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS face_events (
event_id INTEGER PRIMARY KEY,
face_id INTEGER,
bounding_box TEXT,
timestamp TEXT,
face_direction TEXT,
angle REAL,
holding_phone BOOLEAN,
phone_proximity REAL,
attention_needed BOOLEAN
)''')
conn.commit()
conn.close()
# Function to insert face event into database
def insert_face_event(face_id, bounding_box, timestamp, face_direction, angle, holding_phone, phone_proximity, attention_needed):
conn = sqlite3.connect('event_tracking.db')
cursor = conn.cursor()
cursor.execute('''INSERT INTO face_events (face_id, bounding_box, timestamp, face_direction, angle, holding_phone, phone_proximity, attention_needed)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)''',
(face_id, str(bounding_box), timestamp, str(face_direction), angle, holding_phone, phone_proximity, attention_needed))
conn.commit()
conn.close()
# Function to process sound and classify it using YAMNet
def classify_sound(frame, sound_event_data):
# Process audio data for classification (dummy example)
audio_data, samplerate = sf.read('audio_file.wav') # This would be a live microphone stream
prediction = model.predict(audio_data)
# Detect sounds in frame and return event information
for i, score in enumerate(prediction[0]):
if score > 0.5: # If score is above threshold, classify as a detected sound
event_name = class_names
sound_event_data.append({'event': event_name, 'confidence': score})
return sound_event_data
# Detect faces and their direction, then store in the database
def process_frame(frame):
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
faces = face_cascade.detectMultiScale(gray, 1.1, 4)
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# Dummy sound detection data for now
sound_event_data = []
# Loop through detected faces and get direction and face properties
for face_id, (x, y, w, h) in enumerate(faces):
# Example: face direction calculated from pose or landmarks
yaw, pitch = 10, 5 # These should come from face landmark detection
face_direction = calculate_face_direction(yaw, pitch)
# Simulate detecting phone near head (this would come from another sensor or model)
holding_phone = False
# Simulate proximity of phone (distance in meters, 0 indicates no phone detected)
phone_proximity = 0.5 # Example value for proximity, replace with actual sensor data
# Check for attention conditions
angle = np.degrees(np.arccos(np.dot(face_direction, [0, 0, 1]))) # Angle relative to camera's forward vector
attention_needed = (angle < 30) and ("speech" in [e['event'] for e in sound_event_data]) and not holding_phone
# Insert the event into the database
insert_face_event(face_id, (x, y, w, h), timestamp, face_direction, angle, holding_phone, phone_proximity, attention_needed)
return sound_event_data
# Query database for face attention events
def query_attention():
conn = sqlite3.connect('event_tracking.db')
cursor = conn.cursor()
cursor.execute('''SELECT * FROM face_events WHERE attention_needed = 1''')
rows = cursor.fetchall()
conn.close()
return rows
# Create the database
create_database()
# Process a sample frame from the camera feed
cap = cv2.VideoCapture(0)
while True:
ret, frame = cap.read()
if not ret:
break
# Classify sound in the frame
sound_event_data = classify_sound(frame, [])
# Process face detection and store events
process_frame(frame)
# Optionally, query the database for faces needing attention
attention_events = query_attention()
print(attention_events)
# Display the resulting frame
cv2.imshow('Frame', frame)
if cv2.waitKey(1) & 0xFF == ord('q'):
break
cap.release()
cv2.destroyAllWindows()
that the IMU processing/decoding system itself might work better as an Ai to learn the best way to process the data; after that it can just be a "hardcoded" set of code that can be optimized to just do that job and not have to be learning anymore, and take less ahrdwaer to rune it. but i am going on intuition based on a very limited and fragile understanding of what's going on iside these ais, and what is needed to do what i am trying to do.
