import streamlit as st import cv2 import numpy as np import tempfile from PIL import Image import torch import torch.nn as nn from torchvision import transforms, models import time from collections import deque import yaml from ultralytics import YOLO # Set page config st.set_page_config( page_title="Advanced Dog Language Understanding", page_icon="🐕", layout="wide" ) class DogBehaviorAnalyzer: def __init__(self): self.behaviors = { 'tail_wagging': {'threshold': 0.15, 'description': 'Your dog is displaying happiness and excitement!'}, 'movement': {'threshold': 0.02, 'description': 'Your dog is active and moving around.'}, 'stationary': {'threshold': 0.01, 'description': 'Your dog is calm and still.'}, 'high_activity': {'threshold': 0.05, 'description': 'Your dog is very energetic!'} } # Motion detection parameters self.history = [] self.max_history = 10 self.prev_frame = None self.motion_history = deque(maxlen=5) # Store recent motion scores def detect_motion(self, frame): """Detect motion in frame with improved sensitivity""" # Resize frame for consistent motion detection frame = cv2.resize(frame, (300, 300)) gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) gray = cv2.GaussianBlur(gray, (21, 21), 0) if self.prev_frame is None: self.prev_frame = gray return 0.0 frame_delta = cv2.absdiff(self.prev_frame, gray) thresh = cv2.threshold(frame_delta, 20, 255, cv2.THRESH_BINARY)[1] # Lower threshold for better sensitivity thresh = cv2.dilate(thresh, None, iterations=2) # Calculate motion score motion_score = np.sum(thresh > 0) / thresh.size self.prev_frame = gray # Add to motion history self.motion_history.append(motion_score) # Return average of recent motion scores for stability return np.mean(self.motion_history) if len(self.motion_history) > 0 else motion_score def detect_color_changes(self, frame): """Detect significant color changes with improved sensitivity""" frame = cv2.resize(frame, (300, 300)) hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV) # Define range for common dog colors color_ranges = [ (np.array([0, 30, 30]), np.array([30, 255, 255])), # Brown/Red (np.array([0, 0, 0]), np.array([180, 50, 255])), # White/Gray/Black ] total_change_score = 0 for lower, upper in color_ranges: mask = cv2.inRange(hsv, lower, upper) if len(self.history) > 0: prev_mask = self.history[-1] diff = cv2.absdiff(mask, prev_mask) change_score = np.sum(diff > 0) / diff.size total_change_score = max(total_change_score, change_score) self.history.append(mask) if len(self.history) > self.max_history: self.history.pop(0) return total_change_score def analyze_frame(self, frame): """Analyze frame with improved behavior detection logic""" motion_score = self.detect_motion(frame) color_change_score = self.detect_color_changes(frame) detected_behaviors = [] # High activity detection (running, jumping) if motion_score > self.behaviors['high_activity']['threshold']: detected_behaviors.append(('high_activity', motion_score)) # Regular movement detection elif motion_score > self.behaviors['movement']['threshold']: detected_behaviors.append(('movement', motion_score)) # Stationary detection - only if very little motion elif motion_score < self.behaviors['stationary']['threshold']: detected_behaviors.append(('stationary', 1.0 - motion_score)) # Tail wagging detection - based on localized color changes if color_change_score > self.behaviors['tail_wagging']['threshold']: detected_behaviors.append(('tail_wagging', color_change_score)) # Debug information if not detected_behaviors: st.sidebar.write(f"Debug - Motion Score: {motion_score:.4f}") st.sidebar.write(f"Debug - Color Change Score: {color_change_score:.4f}") return detected_behaviors def main(): st.title("🐕 Dog Behavior Analyzer") st.write("Upload a video of your dog for behavior analysis!") analyzer = DogBehaviorAnalyzer() video_file = st.file_uploader("Upload Video", type=['mp4', 'avi', 'mov']) if video_file is not None: tfile = tempfile.NamedTemporaryFile(delete=False) tfile.write(video_file.read()) cap = cv2.VideoCapture(tfile.name) col1, col2 = st.columns(2) with col1: st.subheader("Video Analysis") video_placeholder = st.empty() with col2: st.subheader("Real-time Behavior Detection") analysis_placeholder = st.empty() progress_bar = st.progress(0) behavior_counts = {behavior: 0 for behavior in analyzer.behaviors.keys()} confidence_history = {behavior: [] for behavior in analyzer.behaviors.keys()} frame_count = 0 total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) while cap.isOpened(): ret, frame = cap.read() if not ret: break frame_count += 1 progress = frame_count / total_frames progress_bar.progress(progress) # Analyze frame detected_behaviors = analyzer.analyze_frame(frame) # Draw behavior labels on frame y_pos = 30 for behavior, conf in detected_behaviors: behavior_counts[behavior] += 1 confidence_history[behavior].append(conf) cv2.putText(frame, f"{behavior.replace('_', ' ').title()}: {conf:.2f}", (10, y_pos), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2) y_pos += 30 video_placeholder.image( cv2.cvtColor(frame, cv2.COLOR_BGR2RGB), channels="RGB", use_container_width=True ) # Update analysis display analysis_text = "Detected Behaviors:\n\n" for behavior, count in behavior_counts.items(): if count > 0: avg_conf = sum(confidence_history[behavior]) / len(confidence_history[behavior]) analysis_text += f"• {behavior.replace('_', ' ').title()}:\n" analysis_text += f" Count: {count} | Confidence: {avg_conf:.2f}\n" analysis_text += f" {analyzer.behaviors[behavior]['description']}\n\n" analysis_placeholder.text_area( "Analysis Results", analysis_text, height=300, key=f"analysis_text_{frame_count}" ) time.sleep(0.05) cap.release() # Final analysis st.subheader("Comprehensive Analysis") # Create metrics col1, col2, col3 = st.columns(3) with col1: most_common = max(behavior_counts.items(), key=lambda x: x[1])[0] if any(behavior_counts.values()) else "None" st.metric("Primary Behavior", most_common.replace('_', ' ').title()) with col2: total_behaviors = sum(behavior_counts.values()) st.metric("Total Behaviors", total_behaviors) with col3: valid_confidences = [conf for confs in confidence_history.values() if confs for conf in confs] avg_confidence = np.mean(valid_confidences) if valid_confidences else 0 st.metric("Average Confidence", f"{avg_confidence:.2%}") # Behavior distribution chart if any(behavior_counts.values()): st.subheader("Behavior Distribution") behavior_data = {k.replace('_', ' ').title(): v for k, v in behavior_counts.items() if v > 0} st.bar_chart(behavior_data) # Recommendations st.subheader("Behavior Insights") recommendations = [] if behavior_counts['tail_wagging'] > total_behaviors * 0.3: recommendations.append("• Your dog shows frequent happiness - great time for positive reinforcement!") if behavior_counts['high_activity'] > total_behaviors * 0.4: recommendations.append("• High energy levels detected - consider more physical exercise") if behavior_counts['stationary'] > total_behaviors * 0.5: recommendations.append("• Your dog appears calm - good for training sessions") for rec in recommendations: st.write(rec) else: st.write("Upload a video to see behavior analysis!") if __name__ == "__main__": main()