import streamlit as st from google.cloud import vision import os from PIL import Image, ImageDraw, ImageFont import io import numpy as np from streamlit_option_menu import option_menu import json from google.oauth2 import service_account import google.auth import av from streamlit_webrtc import webrtc_streamer, VideoProcessorBase, RTCConfiguration import cv2 from typing import List, Union from google.cloud import documentai import pandas as pd from google.cloud import bigquery from google.cloud.exceptions import NotFound import tempfile import time import matplotlib.pyplot as plt from pathlib import Path import plotly.express as px from groq import Groq import streamlit.components.v1 as components import html from streamlit_chat import message import uuid from dotenv import load_dotenv from langchain_community.embeddings import OpenAIEmbeddings from langchain_groq import ChatGroq from langchain_community.vectorstores import FAISS from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.chains import ConversationalRetrievalChain from langchain.memory import ConversationBufferMemory from langchain_community.document_loaders import TextLoader import re import base64 # Set page config st.set_page_config( page_title="Vision AI Analyzer", page_icon="👁️", layout="wide" ) # Custom CSS st.markdown(""" """, unsafe_allow_html=True) def analyze_image(image, analysis_types, confidence_threshold=0.5): """Analyze image with selected analysis types and confidence filtering""" # Convert uploaded image to bytes if image is None: return None, {}, {}, "", {} img_byte_arr = io.BytesIO() image.save(img_byte_arr, format='PNG') content = img_byte_arr.getvalue() # Create vision image object vision_image = vision.Image(content=content) # Perform detection based on selected types labels_data = {} objects_data = {} text_content = "" colors_data = {} # New: store dominant colors text_language = "" # New: store detected language img_with_boxes = image.copy() draw = ImageDraw.Draw(img_with_boxes) # Extract color information regardless of analysis types if "Visual Attributes" in analysis_types: image_properties = client.image_properties(image=vision_image).image_properties_annotation # Get top 5 dominant colors with scores colors_data = { f"Color #{i+1}": { "rgb": (int(color.color.red), int(color.color.green), int(color.color.blue)), "score": round(color.score * 100, 2), "pixel_fraction": round(color.pixel_fraction * 100, 2) } for i, color in enumerate(image_properties.dominant_colors.colors[:5]) } if "Labels" in analysis_types: labels = client.label_detection(image=vision_image) # Apply confidence threshold labels_data = {label.description: round(label.score * 100) for label in labels.label_annotations if label.score >= confidence_threshold} if "Objects" in analysis_types: objects = client.object_localization(image=vision_image) # Apply confidence threshold filtered_objects = [obj for obj in objects.localized_object_annotations if obj.score >= confidence_threshold] objects_data = {obj.name: round(obj.score * 100) for obj in filtered_objects} # Draw object boundaries for obj in filtered_objects: box = [(vertex.x * image.width, vertex.y * image.height) for vertex in obj.bounding_poly.normalized_vertices] draw.polygon(box, outline='red', width=2) draw.text((box[0][0], box[0][1] - 10), f"{obj.name}: {int(obj.score * 100)}%", fill='red') if "Text" in analysis_types: text = client.text_detection(image=vision_image) if text.text_annotations: text_content = text.text_annotations[0].description # New: Detect language if text is found if text_content: try: # Get language of text document = vision.types.Document( content=content, type_=vision.types.Document.Type.GENERAL_DOCUMENT ) response = client.document_text_detection(image=vision_image) if response.text_annotations: # Get the language code from the first page if response.pages and response.pages[0].property.detected_languages: lang = response.pages[0].property.detected_languages[0] text_language = f"{lang.language_code} ({round(lang.confidence * 100)}%)" except Exception as e: text_language = "Detection failed" # Draw text boundaries for text_annot in text.text_annotations[1:]: # Skip the first one (full text) box = [(vertex.x, vertex.y) for vertex in text_annot.bounding_poly.vertices] draw.polygon(box, outline='blue', width=1) if "Face Detection" in analysis_types: faces = client.face_detection(image=vision_image) # Apply confidence threshold - filter by detection confidence filtered_faces = [face for face in faces.face_annotations if face.detection_confidence >= confidence_threshold] for face in filtered_faces: vertices = face.bounding_poly.vertices box = [(vertex.x, vertex.y) for vertex in vertices] draw.polygon(box, outline='green', width=2) # Draw facial landmarks for landmark in face.landmarks: px = landmark.position.x py = landmark.position.y draw.ellipse((px-2, py-2, px+2, py+2), fill='yellow') # Return extended results return img_with_boxes, labels_data, objects_data, text_content, colors_data, text_language def display_results(annotated_img, labels, objects, text, colors=None, text_language=None): """Display analysis results in a clean format with enhanced features""" # Store results in session state for chatbot context st.session_state.analysis_results = { "labels": labels, "objects": objects, "text": text, "colors": colors if colors else {}, "text_language": text_language if text_language else "", "timestamp": time.strftime("%Y-%m-%d %H:%M:%S") } # Update vectorstore with new results update_vectorstore_with_results(st.session_state.analysis_results) col1, col2 = st.columns([3, 2]) with col1: st.markdown('

Analyzed Image

', unsafe_allow_html=True) st.image(annotated_img, use_container_width=True) with col2: st.markdown('

Analysis Results

', unsafe_allow_html=True) # Labels tab if labels: st.markdown("##### 🏷️ Labels Detected") st.markdown('

', unsafe_allow_html=True) for label, confidence in labels.items(): st.markdown(f'

{label}: {confidence}%

', unsafe_allow_html=True) st.markdown('

', unsafe_allow_html=True) # Objects tab if objects: st.markdown("##### 📦 Objects Detected") st.markdown('

', unsafe_allow_html=True) for obj, confidence in objects.items(): st.markdown(f'

{obj}: {confidence}%

', unsafe_allow_html=True) st.markdown('

', unsafe_allow_html=True) # Text tab if text: st.markdown("##### 📝 Text Detected") if text_language: st.markdown(f"**Detected Language:** {text_language}") st.markdown('

', unsafe_allow_html=True) st.markdown(f'

{text}

', unsafe_allow_html=True) st.markdown('

', unsafe_allow_html=True) # Color analysis tab (new) if colors: st.markdown("##### 🎨 Dominant Colors") st.markdown('

', unsafe_allow_html=True) # Create color swatches for color_name, color_data in colors.items(): rgb = color_data["rgb"] hex_color = f"#{rgb[0]:02x}{rgb[1]:02x}{rgb[2]:02x}" # Display color swatch with info st.markdown(f"""

{color_name}: {color_data["score"]}% coverage
RGB: {rgb}

""", unsafe_allow_html=True) st.markdown('

', unsafe_allow_html=True) # Add Download Summary Image button summary_img = create_summary_image(annotated_img, labels, objects, text, colors) buf = io.BytesIO() summary_img.save(buf, format="JPEG", quality=90) byte_im = buf.getvalue() st.download_button( label="📥 Download Complete Results Summary", data=byte_im, file_name="analysis_summary.jpg", mime="image/jpeg", help="Download a complete image showing the analyzed image and all detected features" ) def create_summary_image(annotated_img, labels, objects, text, colors=None): """Create a downloadable summary image with analysis results""" # Create a new image with space for results img_width, img_height = annotated_img.size # Make room for text results (adjust height based on content) result_height = 400 # Space for results summary_img = Image.new('RGB', (img_width, img_height + result_height), color=(255, 255, 255)) # Paste the annotated image at the top summary_img.paste(annotated_img, (0, 0)) # Create a drawing object draw = ImageDraw.Draw(summary_img) # Try to get a font - use default if not available try: font = ImageFont.truetype("arial.ttf", 16) title_font = ImageFont.truetype("arial.ttf", 20) except IOError: font = ImageFont.load_default() title_font = ImageFont.load_default() # Draw title - using dark blue color draw.text((20, img_height + 20), "Cosmick Cloud AI Analyzer Results", fill=(0, 0, 139), font=title_font) # Draw divider line draw.line([(0, img_height + 50), (img_width, img_height + 50)], fill=(200, 200, 200), width=2) # Current Y position for drawing text y_pos = img_height + 60 # Draw labels if labels: draw.text((20, y_pos), "🏷️ Labels Detected:", fill=(0, 0, 0), font=title_font) y_pos += 30 for i, (label, confidence) in enumerate(sorted(labels.items(), key=lambda x: x[1], reverse=True)): if i < 8: # Limit to top 8 labels to avoid overcrowding draw.text((40, y_pos), f"{label}: {confidence}%", fill=(0, 100, 0), font=font) y_pos += 25 # Draw a column divider mid_point = img_width // 2 draw.line([(mid_point - 20, img_height + 60), (mid_point - 20, img_height + result_height - 20)], fill=(200, 200, 200), width=1) # Reset Y position for second column y_pos = img_height + 60 # Draw objects in second column if objects: draw.text((mid_point, y_pos), "📦 Objects Detected:", fill=(0, 0, 0), font=title_font) y_pos += 30 for i, (obj, confidence) in enumerate(sorted(objects.items(), key=lambda x: x[1], reverse=True)): if i < 8: # Limit to top 8 objects draw.text((mid_point + 20, y_pos), f"{obj}: {confidence}%", fill=(0, 0, 128), font=font) y_pos += 25 # Add text detection summary at the bottom with improved visibility if text: bottom_y = img_height + result_height - 80 draw.text((20, bottom_y), "📝 Text Detected:", fill=(0, 0, 0), font=title_font) # Truncate text if too long display_text = text if len(text) < 100 else text[:97] + "..." # Change text color to dark red for better visibility draw.text((20, bottom_y + 30), display_text, fill=(139, 0, 0), font=font) # Add timestamp with darker color timestamp = time.strftime("%Y-%m-%d %H:%M:%S") draw.text((img_width - 200, img_height + result_height - 30), f"Generated: {timestamp}", fill=(50, 50, 50), font=font) return summary_img class VideoProcessor(VideoProcessorBase): """Process video frames for real-time analysis with enhanced OpenCV processing""" def __init__(self, analysis_types: List[str], processing_mode: str = "Hybrid (Google Vision + OpenCV)", track_update_frames: int = 5, confidence_threshold: float = 0.5): self.analysis_types = analysis_types self.processing_mode = processing_mode self.frame_counter = 0 self.process_every_n_frames = track_update_frames # Process every N frames self.confidence_threshold = confidence_threshold self.vision_client = client # Store client reference self.last_results = {} # Cache results between processed frames self.last_processed_time = time.time() self.processing_active = True # Enhanced tracking self.object_trackers = {} self.tracking_points = None self.prev_gray = None # Motion history for better activity detection self.motion_history = np.zeros((480, 640), np.float32) self.motion_threshold = 32 self.max_time_delta = 0.5 self.min_time_delta = 0.05 # For OpenCV-only detection mode self.opencv_detector = None self.init_opencv_detector() def init_opencv_detector(self): """Initialize OpenCV-based object detector if needed""" if self.processing_mode == "OpenCV Only" or self.processing_mode == "Hybrid (Google Vision + OpenCV)": try: # Initialize YOLO or other available models # This is a placeholder - you might need to adjust based on available OpenCV DNN models weights_path = os.path.join(os.path.dirname(__file__), "models/yolov3.weights") config_path = os.path.join(os.path.dirname(__file__), "models/yolov3.cfg") # Check if files exist, otherwise use a simpler fallback detector if os.path.exists(weights_path) and os.path.exists(config_path): self.opencv_detector = cv2.dnn.readNetFromDarknet(config_path, weights_path) else: # Fallback to HOG detector for people self.opencv_detector = cv2.HOGDescriptor() self.opencv_detector.setSVMDetector(cv2.HOGDescriptor_getDefaultPeopleDetector()) st.info("Using basic OpenCV HOG detector. For better results, install YOLO model files.") except Exception as e: st.warning(f"Could not initialize OpenCV detector: {str(e)}. Falling back to basic detection.") self.opencv_detector = None def transform(self, frame: av.VideoFrame) -> av.VideoFrame: img = frame.to_ndarray(format="bgr24") self.frame_counter += 1 # Resize for consistent processing if needed if img.shape[0] != 480 or img.shape[1] != 640: img = cv2.resize(img, (640, 480)) # Add status display on all frames cv2.putText(img, f"Vision AI: {'Active' if self.processing_active else 'Paused'} - Mode: {self.processing_mode}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2) # Convert to grayscale for motion detection gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) # Apply motion detection for all frames if enabled if "Motion" in self.analysis_types and self.prev_gray is not None: # Calculate frame difference for smoother motion detection frame_diff = cv2.absdiff(gray, self.prev_gray) _, motion_mask = cv2.threshold(frame_diff, self.motion_threshold, 1, cv2.THRESH_BINARY) timestamp = time.time() # Update motion history cv2.motempl.updateMotionHistory(motion_mask, self.motion_history, timestamp, self.max_time_delta) # Calculate motion gradient mg_mask = cv2.motempl.calcMotionGradient( self.motion_history, self.min_time_delta, self.max_time_delta, apertureSize=5) # Visualize motion segments seg_mask, segments = cv2.motempl.segmentMotion( self.motion_history, timestamp, self.max_time_delta) # Visualize motion segments motion_img = np.zeros_like(img) for i, segment in enumerate(segments): if segment[1] < 50: # Filter out small segments continue # Draw motion regions with random colors color = np.random.randint(0, 255, 3).tolist() motion_img = cv2.drawContours(motion_img, [np.array(segment[2])], -1, color, -1) # Overlay motion visualization alpha = 0.3 cv2.addWeighted(motion_img, alpha, img, 1 - alpha, 0, img) # Process with Vision API at regular intervals if using Google Vision current_time = time.time() if (self.processing_mode == "Google Vision API Only" or self.processing_mode == "Hybrid (Google Vision + OpenCV)") and \ (current_time - self.last_processed_time > 1.0) and self.processing_active and \ self.vision_client is not None: self.last_processed_time = current_time # Convert frame to JPEG for Vision API success, jpeg_frame = cv2.imencode('.jpg', img) if success: image_content = jpeg_frame.tobytes() # Create vision image vision_image = vision.Image(content=image_content) try: # Perform detection based on selected types if "Objects" in self.analysis_types: objects = self.vision_client.object_localization(image=vision_image) # Filter objects by confidence threshold filtered_objects = [obj for obj in objects.localized_object_annotations if obj.score >= self.confidence_threshold] self.last_results["objects"] = filtered_objects # Log detection for tracking for obj in filtered_objects: # Draw object boundaries box = [(vertex.x * img.shape[1], vertex.y * img.shape[0]) for vertex in obj.bounding_poly.normalized_vertices] points = np.array([[int(p[0]), int(p[1])] for p in box]) cv2.polylines(img, [points], True, (0, 255, 0), 2) # Add label with confidence cv2.putText(img, f"{obj.name}: {int(obj.score * 100)}%", (int(box[0][0]), int(box[0][1] - 10)), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2) # Create unique object ID for tracking obj_id = f"{obj.name}_{self.frame_counter}" # Calculate bounding box for tracker x_values = [p[0] for p in box] y_values = [p[1] for p in box] x_min, x_max = min(x_values), max(x_values) y_min, y_max = min(y_values), max(y_values) # Create or update tracker if obj.name not in self.object_trackers: self.object_trackers[obj.name] = { "bbox": (int(x_min), int(y_min), int(x_max - x_min), int(y_max - y_min)), "last_seen": self.frame_counter, "score": obj.score } else: # Update existing tracker self.object_trackers[obj.name] = { "bbox": (int(x_min), int(y_min), int(x_max - x_min), int(y_max - y_min)), "last_seen": self.frame_counter, "score": obj.score } # Face detection if selected if "Face Detection" in self.analysis_types: faces = self.vision_client.face_detection(image=vision_image) self.last_results["faces"] = faces.face_annotations # Draw face boundaries for face in faces.face_annotations: if face.detection_confidence >= self.confidence_threshold: vertices = face.bounding_poly.vertices points = [(vertex.x, vertex.y) for vertex in vertices] points = np.array([[p[0], p[1]] for p in points]) cv2.polylines(img, [points], True, (0, 0, 255), 2) # Add confidence score cv2.putText(img, f"Face: {int(face.detection_confidence * 100)}%", (points[0][0], points[0][1] - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 2) # Draw facial landmarks for landmark in face.landmarks: px = landmark.position.x py = landmark.position.y cv2.circle(img, (int(px), int(py)), 2, (255, 255, 0), -1) # Text detection if selected if "Text" in self.analysis_types: text = self.vision_client.text_detection(image=vision_image) if text.text_annotations: self.last_results["text"] = text.text_annotations # Draw text bounding boxes for text_annot in text.text_annotations[1:]: # Skip the first one (full text) box = [(vertex.x, vertex.y) for vertex in text_annot.bounding_poly.vertices] points = np.array([[int(p[0]), int(p[1])] for p in box]) cv2.polylines(img, [points], True, (255, 0, 0), 2) # Add recognized text cv2.putText(img, text_annot.description, (points[0][0], points[0][1] - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 0, 0), 2) except Exception as e: # Handle API errors gracefully error_msg = f"API Error: {str(e)}" cv2.putText(img, error_msg, (10, 70), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 0, 255), 2) # Process with OpenCV object detection if enabled if (self.processing_mode == "OpenCV Only" or self.processing_mode == "Hybrid (Google Vision + OpenCV)") and \ self.opencv_detector is not None and \ (self.frame_counter % self.process_every_n_frames == 0 or not self.object_trackers): try: # If using HOG detector (the fallback) if isinstance(self.opencv_detector, cv2.HOGDescriptor): # Detect people boxes, weights = self.opencv_detector.detectMultiScale( img, winStride=(8, 8), padding=(4, 4), scale=1.05 ) # Draw bounding boxes for i, (x, y, w, h) in enumerate(boxes): if weights[i] > 0.3: # Confidence threshold cv2.rectangle(img, (x, y), (x+w, y+h), (255, 0, 0), 2) cv2.putText(img, f"Person: {int(weights[i] * 100)}%", (x, y-10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 0, 0), 2) # Add to trackers self.object_trackers[f"person_{i}"] = { "bbox": (x, y, w, h), "last_seen": self.frame_counter, "score": weights[i] } else: # Using YOLO or another DNN-based detector blob = cv2.dnn.blobFromImage(img, 1/255.0, (416, 416), swapRB=True, crop=False) self.opencv_detector.setInput(blob) layer_names = self.opencv_detector.getLayerNames() output_layers = [layer_names[i - 1] for i in self.opencv_detector.getUnconnectedOutLayers()] outputs = self.opencv_detector.forward(output_layers) # Process detections class_ids = [] confidences = [] boxes = [] for output in outputs: for detection in output: scores = detection[5:] class_id = np.argmax(scores) confidence = scores[class_id] if confidence > self.confidence_threshold: # Object detected center_x = int(detection[0] * img.shape[1]) center_y = int(detection[1] * img.shape[0]) w = int(detection[2] * img.shape[1]) h = int(detection[3] * img.shape[0]) # Rectangle coordinates x = int(center_x - w / 2) y = int(center_y - h / 2) boxes.append([x, y, w, h]) confidences.append(float(confidence)) class_ids.append(class_id) # Apply non-maximum suppression indices = cv2.dnn.NMSBoxes(boxes, confidences, self.confidence_threshold, 0.4) # Define COCO class names class_names = ["person", "bicycle", "car", "motorcycle", "airplane", "bus", "train", "truck", "boat", "traffic light", "fire hydrant", "stop sign", "parking meter", "bench", "bird", "cat", "dog", "horse", "sheep", "cow", "elephant", "bear", "zebra", "giraffe", "backpack", "umbrella", "handbag", "tie", "suitcase", "frisbee", "skis", "snowboard", "sports ball", "kite", "baseball bat", "baseball glove", "skateboard", "surfboard", "tennis racket", "bottle", "wine glass", "cup", "fork", "knife", "spoon", "bowl", "banana", "apple", "sandwich", "orange", "broccoli", "carrot", "hot dog", "pizza", "donut", "cake", "chair", "couch", "potted plant", "bed", "dining table", "toilet", "tv", "laptop", "mouse", "remote", "keyboard", "cell phone", "microwave", "oven", "toaster", "sink", "refrigerator", "book", "clock", "vase", "scissors", "teddy bear", "hair drier", "toothbrush"] for i in indices: if isinstance(i, (list, tuple)): # Handle different OpenCV versions i = i[0] box = boxes[i] x, y, w, h = box # Get class label and draw bounding box class_id = class_ids[i] label = f"{class_names[class_id]}: {int(confidences[i] * 100)}%" cv2.rectangle(img, (x, y), (x + w, y + h), (0, 255, 0), 2) cv2.putText(img, label, (x, y - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2) # Add to trackers object_name = class_names[class_id] self.object_trackers[f"{object_name}_{i}"] = { "bbox": (x, y, w, h), "last_seen": self.frame_counter, "score": confidences[i], "class": object_name } except Exception as e: cv2.putText(img, f"OpenCV Error: {str(e)}", (10, 110), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 2) # Update object tracking for existing objects (every frame) objects_to_remove = [] for obj_id, tracker_info in self.object_trackers.items(): # Remove old trackers if self.frame_counter - tracker_info["last_seen"] > 30: # Remove after 30 frames objects_to_remove.append(obj_id) continue # Draw tracking box (for objects not updated this frame) if self.frame_counter - tracker_info["last_seen"] <= 5: # Only show recent tracked objects x, y, w, h = tracker_info["bbox"] # Use different color for tracked vs detected objects if self.frame_counter == tracker_info["last_seen"]: color = (0, 255, 0) # Green for newly detected else: color = (255, 165, 0) # Orange for tracked cv2.rectangle(img, (x, y), (x + w, y + h), color, 2) # Add label with confidence and tracking status tracking_age = self.frame_counter - tracker_info["last_seen"] label = f"{obj_id.split('_')[0]}: {int(tracker_info['score'] * 100)}%" if tracking_age > 0: label += f" (tracked {tracking_age}f)" cv2.putText(img, label, (x, y - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 2) # Remove expired trackers for obj_id in objects_to_remove: del self.object_trackers[obj_id] # Save current frame for next iteration self.prev_gray = gray # Add processing mode indicator cv2.putText(img, f"Mode: {self.processing_mode}", (img.shape[1] - 300, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2) # Add frame counter cv2.putText(img, f"Frame: {self.frame_counter}", (img.shape[1] - 150, img.shape[0] - 20), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 255), 1) return av.VideoFrame.from_ndarray(img, format="bgr24") def analyze_document(file_content, processor_id, location="us"): """Analyze document using Document AI""" # Create Document AI client client = documentai.DocumentProcessorServiceClient(credentials=credentials) # The full resource name of the processor processor_name = f"projects/{credentials.project_id}/locations/{location}/processors/{processor_id}" # Determine the mime type based on input file type if file_content[:4] == b'%PDF': mime_type = "application/pdf" else: # Default to image for other types mime_type = "image/jpeg" # Create the request raw_document = documentai.RawDocument(content=file_content, mime_type=mime_type) # Updated API request format request = documentai.ProcessRequest( name=processor_name, raw_document=raw_document ) # Process the document result = client.process_document(request=request) document = result.document # Extract text, entities, etc. text = document.text entities = {} # Extract entities and their values for entity in document.entities: entities[entity.type_] = entity.mention_text # Extract table data if available tables = [] for page in document.pages: for table in page.tables: table_data = [] # Get header row if available headers = [] if hasattr(table, 'header_rows') and table.header_rows: for cell in table.header_rows[0].cells: if cell.layout.text_anchor.text_segments: segment = cell.layout.text_anchor.text_segments[0] headers.append(text[segment.start_index:segment.end_index]) else: headers.append("") # Get data rows for row in table.body_rows: row_data = [] for cell in row.cells: if cell.layout.text_anchor.text_segments: segment = cell.layout.text_anchor.text_segments[0] cell_text = text[segment.start_index:segment.end_index] row_data.append(cell_text) else: row_data.append("") table_data.append(row_data) # If no header found, create generic column names if not headers and table_data: headers = [f"Column_{i+1}" for i in range(len(table_data[0]))] tables.append({"headers": headers, "data": table_data}) # Store results in session state for chatbot context results = (text, entities, tables) st.session_state.analysis_results = { "text": text, "entities": entities, "tables": tables, "timestamp": time.strftime("%Y-%m-%d %H:%M:%S") } # Update vectorstore with new results update_vectorstore_with_results(results) return text, entities, tables def create_bigquery_table(dataset_id, table_id, schema=None): """Create a BigQuery table if it doesn't exist""" # Create client bq_client = bigquery.Client(credentials=credentials, project=credentials.project_id) # Create dataset if it doesn't exist dataset_ref = bq_client.dataset(dataset_id) try: bq_client.get_dataset(dataset_ref) except NotFound: dataset = bigquery.Dataset(dataset_ref) dataset.location = "US" bq_client.create_dataset(dataset) st.info(f"Dataset '{dataset_id}' created.") # Create table reference table_ref = dataset_ref.table(table_id) # Check if table exists try: bq_client.get_table(table_ref) st.info(f"Table '{table_id}' already exists.") return table_ref except NotFound: # Create the table with schema if provided if schema: table = bigquery.Table(table_ref, schema=schema) else: table = bigquery.Table(table_ref) bq_client.create_table(table) st.info(f"Table '{table_id}' created.") return table_ref def upload_csv_to_bigquery(file, dataset_id, table_id, append=False): """Upload a CSV file to BigQuery""" # Create client bq_client = bigquery.Client(credentials=credentials, project=credentials.project_id) # First, ensure dataset and table exist table_ref = create_bigquery_table(dataset_id, table_id) # Create a temporary file with tempfile.NamedTemporaryFile(delete=False, suffix='.csv') as temp_file: temp_file.write(file.getvalue()) temp_file_path = temp_file.name # Configure the load job job_config = bigquery.LoadJobConfig( source_format=bigquery.SourceFormat.CSV, skip_leading_rows=1, # Skip header row autodetect=True, # Auto-detect schema ) if append: job_config.write_disposition = bigquery.WriteDisposition.WRITE_APPEND else: job_config.write_disposition = bigquery.WriteDisposition.WRITE_TRUNCATE # Load the file with open(temp_file_path, "rb") as source_file: job = bq_client.load_table_from_file( source_file, table_ref, job_config=job_config ) # Wait for the job to complete job.result() # Clean up the temp file os.unlink(temp_file_path) # Get the table table = bq_client.get_table(table_ref) result = { "num_rows": table.num_rows, "size_bytes": table.num_bytes, "schema": [field.name for field in table.schema] } # Store results in session state for chatbot context st.session_state.analysis_results = { "data_source": f"{dataset_id}.{table_id}", "num_rows": table.num_rows, "schema": [field.name for field in table.schema], "timestamp": time.strftime("%Y-%m-%d %H:%M:%S") } # Update vectorstore with new results update_vectorstore_with_results(result) return result def run_bigquery(query): """Run a BigQuery query and return results""" # Create client bq_client = bigquery.Client(credentials=credentials, project=credentials.project_id) # Run the query query_job = bq_client.query(query) # Wait for the query to finish results = query_job.result() # Convert to dataframe df = results.to_dataframe() # Store results in session state for chatbot context st.session_state.analysis_results = { "query": query, "results": df, "timestamp": time.strftime("%Y-%m-%d %H:%M:%S") } # Update vectorstore with new results update_vectorstore_with_results(df) return df def list_bigquery_resources(): """List all datasets and tables in the project""" # Create client bq_client = bigquery.Client(credentials=credentials, project=credentials.project_id) # Get datasets datasets = list(bq_client.list_datasets()) # Create a dictionary to store dataset -> tables mapping resources = {} if datasets: for dataset in datasets: dataset_id = dataset.dataset_id # Get tables for this dataset tables = list(bq_client.list_tables(dataset_id)) # Store table names resources[dataset_id] = [table.table_id for table in tables] return resources def process_video_file(video_file, analysis_types, processing_mode="Hybrid (Google Vision + OpenCV)", track_update_frames=5, confidence_threshold=0.5, vision_update_interval=1.0, max_results=10, enable_face_landmarks=True, tracking_algorithm="KCF", motion_sensitivity=32, prioritize_vision=True, blend_results=True, yolo_confidence=0.5, enabled_classes=None): """Process an uploaded video file with enhanced Vision AI detection and analytics""" # Create a temporary file to save the uploaded video with tempfile.NamedTemporaryFile(delete=False, suffix='.mp4') as temp_file: temp_file.write(video_file.read()) temp_video_path = temp_file.name # Create a temp file for the output video output_path = f"{temp_video_path}_processed.mp4" # Open the video file cap = cv2.VideoCapture(temp_video_path) if not cap.isOpened(): st.error("Error opening video file") os.unlink(temp_video_path) return None, None # Get video properties width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) fps = cap.get(cv2.CAP_PROP_FPS) # Calculate max frames for 10-second limit max_frames = int(fps * 10) total_frames = min(int(cap.get(cv2.CAP_PROP_FRAME_COUNT)), max_frames) # Define all configuration values at the beginning of the function # ----------------- Key Parameters ----------------- # Scene change detection threshold scene_change_threshold = 40.0 # Adjust as needed: lower = more sensitive # Process every Nth frame to reduce API calls process_every_n_frames = track_update_frames # Initialize object trackers dictionary for continuous tracking object_trackers = {} # Motion history parameters motion_threshold = motion_sensitivity max_time_delta = 0.5 min_time_delta = 0.05 # Check OpenCV version for compatibility with advanced features opencv_version = cv2.__version__ use_advanced_tracking = True # Initialize the optical flow parameters conditionally based on OpenCV version try: # Optical flow parameters lk_params = dict(winSize=(15, 15), maxLevel=2, criteria=(cv2.TERM_CRITERIA_EPS | cv2.TERM_CRITERIA_COUNT, 10, 0.03)) # Feature detection parameters feature_params = dict(maxCorners=100, qualityLevel=0.3, minDistance=7, blockSize=7) except Exception as e: st.warning(f"Advanced tracking features unavailable: {str(e)}") use_advanced_tracking = False # ----------------- End Parameters ----------------- # Initialize OpenCV detector if needed opencv_detector = None if processing_mode == "OpenCV Only" or processing_mode == "Hybrid (Google Vision + OpenCV)": try: # Check if YOLO model files exist weights_path = os.path.join(os.path.dirname(__file__), "models/yolov3.weights") config_path = os.path.join(os.path.dirname(__file__), "models/yolov3.cfg") if os.path.exists(weights_path) and os.path.exists(config_path): opencv_detector = cv2.dnn.readNetFromDarknet(config_path, weights_path) st.info("Using YOLO model for OpenCV detection") else: # Fallback to HOG detector for people opencv_detector = cv2.HOGDescriptor() opencv_detector.setSVMDetector(cv2.HOGDescriptor_getDefaultPeopleDetector()) st.info("Using basic OpenCV HOG detector. For better results, install YOLO model files.") except Exception as e: st.warning(f"Could not initialize OpenCV detector: {str(e)}. Falling back to basic detection.") # Initialize the selected tracking algorithm if tracking_algorithm == "CSRT": tracker_create_func = cv2.legacy.TrackerCSRT_create elif tracking_algorithm == "KCF": tracker_create_func = cv2.legacy.TrackerKCF_create elif tracking_algorithm == "MOSSE": tracker_create_func = cv2.legacy.TrackerMOSSE_create elif tracking_algorithm == "MedianFlow": tracker_create_func = cv2.legacy.TrackerMedianFlow_create else: # Default to KCF if specified algorithm not available tracker_create_func = cv2.legacy.TrackerKCF_create # Inform user if video is being truncated if int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) > max_frames: st.info("⚠️ Video is longer than 10 seconds. Only the first 10 seconds will be processed.") # Slow down the output video by reducing the fps (60% of original speed) output_fps = fps * 0.6 st.info(f"Output video will be slowed down to {output_fps:.1f} FPS (60% of original speed) for better visualization.") # Create video writer with higher quality settings try: # Try XVID first (widely available) fourcc = cv2.VideoWriter_fourcc(*'XVID') except Exception: # If that fails, try Motion JPEG try: fourcc = cv2.VideoWriter_fourcc(*'MJPG') except Exception: # Last resort - use uncompressed fourcc = cv2.VideoWriter_fourcc(*'DIB ') # Uncompressed RGB out = cv2.VideoWriter(output_path, fourcc, output_fps, (width, height), isColor=True) # Create a progress bar progress_bar = st.progress(0) status_text = st.empty() # Enhanced statistics tracking detection_stats = { "objects": {}, "faces": 0, "text_blocks": 0, "labels": {}, # New advanced tracking "object_tracking": {}, # Track object appearances by frame "activity_metrics": [], # Track frame-to-frame differences "scene_changes": [] # Track major scene transitions } # For scene change detection and motion tracking previous_frame_gray = None prev_points = None # Display mode being used st.info(f"Processing with {processing_mode} mode") try: frame_count = 0 while frame_count < max_frames: # Limit to 10 seconds ret, frame = cap.read() if not ret: break frame_count += 1 # Update progress progress = int(frame_count / total_frames * 100) progress_bar.progress(progress) status_text.text(f"Processing frame {frame_count}/{total_frames} ({progress}%) - {frame_count/fps:.1f}s of 10s") # Add timestamp to frame cv2.putText(frame, f"Time: {frame_count/fps:.2f}s", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2) # Add processing mode indicator cv2.putText(frame, f"Mode: {processing_mode}", (10, 60), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2) # Convert frame to grayscale for motion detection current_frame_gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) current_frame_gray = cv2.GaussianBlur(current_frame_gray, (21, 21), 0) if previous_frame_gray is not None: # Calculate frame difference for activity detection frame_diff = cv2.absdiff(current_frame_gray, previous_frame_gray) activity_level = np.mean(frame_diff) detection_stats["activity_metrics"].append((frame_count/fps, activity_level)) # Scene change detection if activity_level > scene_change_threshold: detection_stats["scene_changes"].append(frame_count/fps) # Mark scene change on frame cv2.putText(frame, "SCENE CHANGE", (width // 2 - 100, 50), cv2.FONT_HERSHEY_SIMPLEX, 1.0, (0, 255, 255), 2) # Add optical flow tracking if enabled if use_advanced_tracking and prev_points is not None: try: # Calculate optical flow next_points, status, _ = cv2.calcOpticalFlowPyrLK(previous_frame_gray, current_frame_gray, prev_points, None, **lk_params) # Select good points if next_points is not None: good_new = next_points[status==1] good_old = prev_points[status==1] # Draw motion tracks for i, (new, old) in enumerate(zip(good_new, good_old)): a, b = new.ravel() c, d = old.ravel() # Draw motion lines cv2.line(frame, (int(c), int(d)), (int(a), int(b)), (0, 255, 255), 2) cv2.circle(frame, (int(a), int(b)), 3, (0, 255, 0), -1) except Exception as e: # If optical flow fails, just continue without it pass # Update tracking points periodically if enabled if use_advanced_tracking and (frame_count % 5 == 0 or prev_points is None or (prev_points is not None and len(prev_points) < 10)): try: prev_points = cv2.goodFeaturesToTrack(current_frame_gray, **feature_params) except Exception: # If feature tracking fails, just continue without it prev_points = None previous_frame_gray = current_frame_gray # Process frames with Vision API if using Google Vision if (processing_mode == "Google Vision API Only" or processing_mode == "Hybrid (Google Vision + OpenCV)") and \ frame_count % process_every_n_frames == 0 and client is not None: # Convert frame to JPEG for Vision API success, jpeg_frame = cv2.imencode('.jpg', frame) if success: image_content = jpeg_frame.tobytes() # Create vision image vision_image = vision.Image(content=image_content) try: # Perform detection based on selected types if "Objects" in analysis_types: objects = client.object_localization(image=vision_image) # Filter objects by confidence threshold filtered_objects = [obj for obj in objects.localized_object_annotations if obj.score >= confidence_threshold] # Update object counts in stats for obj in filtered_objects: if obj.name in detection_stats["objects"]: detection_stats["objects"][obj.name] += 1 else: detection_stats["objects"][obj.name] = 1 # Draw object boundaries box = [(vertex.x * frame.shape[1], vertex.y * frame.shape[0]) for vertex in obj.bounding_poly.normalized_vertices] points = np.array([[int(p[0]), int(p[1])] for p in box]) cv2.polylines(frame, [points], True, (0, 255, 0), 2) # Add label with confidence cv2.putText(frame, f"{obj.name}: {int(obj.score * 100)}%", (int(box[0][0]), int(box[0][1] - 10)), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2) # Add to trackers for future frames # Calculate bounding box x_values = [p[0] for p in box] y_values = [p[1] for p in box] x_min, x_max = min(x_values), max(x_values) y_min, y_max = min(y_values), max(y_values) object_trackers[obj.name] = { "bbox": (int(x_min), int(y_min), int(x_max - x_min), int(y_max - y_min)), "last_seen": frame_count, "score": obj.score } # Process faces if selected if "Face Detection" in analysis_types: faces = client.face_detection(image=vision_image) # Count faces and draw boundaries face_count = 0 for face in faces.face_annotations: if face.detection_confidence >= confidence_threshold: face_count += 1 # Draw face boundary vertices = face.bounding_poly.vertices points = [(vertex.x, vertex.y) for vertex in vertices] points = np.array([[p[0], p[1]] for p in points]) cv2.polylines(frame, [points], True, (0, 0, 255), 2) # Add confidence score cv2.putText(frame, f"Face: {int(face.detection_confidence * 100)}%", (points[0][0], points[0][1] - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 2) # Draw facial landmarks if enabled if enable_face_landmarks: for landmark in face.landmarks: px = landmark.position.x py = landmark.position.y cv2.circle(frame, (int(px), int(py)), 2, (255, 255, 0), -1) # Update face count detection_stats["faces"] += face_count # Process text if selected if "Text" in analysis_types: text = client.text_detection(image=vision_image) if text.text_annotations: # Count text blocks text_blocks = len(text.text_annotations) - 1 # Subtract 1 for the full text annotation detection_stats["text_blocks"] += text_blocks # Draw text bounding boxes for text_annot in text.text_annotations[1:]: # Skip the first one (full text) box = [(vertex.x, vertex.y) for vertex in text_annot.bounding_poly.vertices] points = np.array([[int(p[0]), int(p[1])] for p in box]) cv2.polylines(frame, [points], True, (255, 0, 0), 2) # Add recognized text cv2.putText(frame, text_annot.description, (points[0][0], points[0][1] - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 0, 0), 2) except Exception as e: # Handle API errors gracefully error_msg = f"API Error: {str(e)}" cv2.putText(frame, error_msg, (10, 70), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 0, 255), 2) # Process with OpenCV object detection if enabled if (processing_mode == "OpenCV Only" or processing_mode == "Hybrid (Google Vision + OpenCV)") and \ opencv_detector is not None and \ (frame_count % process_every_n_frames == 0): # The OpenCV detection code goes here... # This would be similar to what's in the VideoProcessor.transform method try: # If using HOG detector (the fallback) if isinstance(opencv_detector, cv2.HOGDescriptor): # Detect people boxes, weights = opencv_detector.detectMultiScale( frame, winStride=(8, 8), padding=(4, 4), scale=1.05 ) # Draw bounding boxes for i, (x, y, w, h) in enumerate(boxes): if weights[i] > 0.3: # Confidence threshold cv2.rectangle(frame, (x, y), (x+w, y+h), (255, 0, 0), 2) cv2.putText(frame, f"Person: {int(weights[i] * 100)}%", (x, y-10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 0, 0), 2) # Add to trackers object_trackers["person"] = { "bbox": (x, y, w, h), "last_seen": frame_count, "score": weights[i] } # Update count in stats if "person" in detection_stats["objects"]: detection_stats["objects"]["person"] += 1 else: detection_stats["objects"]["person"] = 1 else: # Using YOLO or another DNN-based detector blob = cv2.dnn.blobFromImage(frame, 1/255.0, (416, 416), swapRB=True, crop=False) opencv_detector.setInput(blob) # Get output layer names layer_names = opencv_detector.getLayerNames() output_layers = [] # Handle different OpenCV versions try: if cv2.__version__.startswith('4'): # OpenCV 4.x output_layers = [layer_names[i - 1] for i in opencv_detector.getUnconnectedOutLayers()] else: # OpenCV 3.x output_layers = [layer_names[i[0] - 1] for i in opencv_detector.getUnconnectedOutLayers()] except: # Fallback method unconnected_layers = opencv_detector.getUnconnectedOutLayers() if isinstance(unconnected_layers[0], list) or isinstance(unconnected_layers[0], tuple): output_layers = [layer_names[i[0] - 1] for i in unconnected_layers] else: output_layers = [layer_names[i - 1] for i in unconnected_layers] outputs = opencv_detector.forward(output_layers) # Process detections class_ids = [] confidences = [] boxes = [] # Define COCO class names class_names = ["person", "bicycle", "car", "motorcycle", "airplane", "bus", "train", "truck", "boat", "traffic light", "fire hydrant", "stop sign", "parking meter", "bench", "bird", "cat", "dog", "horse", "sheep", "cow", "elephant", "bear", "zebra", "giraffe", "backpack", "umbrella", "handbag", "tie", "suitcase", "frisbee", "skis", "snowboard", "sports ball", "kite", "baseball bat", "baseball glove", "skateboard", "surfboard", "tennis racket", "bottle", "wine glass", "cup", "fork", "knife", "spoon", "bowl", "banana", "apple", "sandwich", "orange", "broccoli", "carrot", "hot dog", "pizza", "donut", "cake", "chair", "couch", "potted plant", "bed", "dining table", "toilet", "tv", "laptop", "mouse", "remote", "keyboard", "cell phone", "microwave", "oven", "toaster", "sink", "refrigerator", "book", "clock", "vase", "scissors", "teddy bear", "hair drier", "toothbrush"] # Process each detection for output in outputs: for detection in output: scores = detection[5:] class_id = np.argmax(scores) confidence = scores[class_id] if confidence > confidence_threshold: # Object detected center_x = int(detection[0] * frame.shape[1]) center_y = int(detection[1] * frame.shape[0]) w = int(detection[2] * frame.shape[1]) h = int(detection[3] * frame.shape[0]) # Rectangle coordinates x = int(center_x - w / 2) y = int(center_y - h / 2) boxes.append([x, y, w, h]) confidences.append(float(confidence)) class_ids.append(class_id) # Apply non-maximum suppression indices = cv2.dnn.NMSBoxes(boxes, confidences, confidence_threshold, 0.4) # Draw the detections if len(indices) > 0: for i in indices: if isinstance(i, (list, tuple)): # Handle different OpenCV versions i = i[0] box = boxes[i] x, y, w, h = box # Get class name class_id = class_ids[i] label = f"{class_names[class_id]}: {int(confidences[i] * 100)}%" # Different colors for different classes color = (0, 255, 0) # Default color # Draw rectangle and label cv2.rectangle(frame, (x, y), (x + w, y + h), color, 2) cv2.putText(frame, label, (x, y - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 2) # Add to object trackers for future frames object_name = class_names[class_id] object_trackers[f"{object_name}_{i}"] = { "bbox": (x, y, w, h), "last_seen": frame_count, "score": confidences[i] } # Update detection stats if object_name in detection_stats["objects"]: detection_stats["objects"][object_name] += 1 else: detection_stats["objects"][object_name] = 1 except Exception as e: cv2.putText(frame, f"OpenCV Error: {str(e)}", (10, 110), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 2) # Add hint about slowed down speed cv2.putText(frame, "Playback: 60% speed for better visualization", (width - 400, height - 30), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 200, 0), 2) # Write the frame to output video out.write(frame) # Release resources cap.release() out.release() # Clear progress indicators progress_bar.empty() status_text.empty() # Read the processed video as bytes for download with open(output_path, 'rb') as file: processed_video_bytes = file.read() # Clean up temporary files os.unlink(temp_video_path) os.unlink(output_path) # Return results results = {"detection_stats": detection_stats} # Store results in session state for chatbot context st.session_state.analysis_results = results # Update vectorstore with new results update_vectorstore_with_results(results) return processed_video_bytes, results except Exception as e: # Clean up on error cap.release() if 'out' in locals(): out.release() os.unlink(temp_video_path) if os.path.exists(output_path): os.unlink(output_path) # Return error information st.error(f"Error processing video: {str(e)}") return None, None def load_bigquery_table(dataset_id, table_id, limit=1000): """Load data directly from an existing BigQuery table""" # Create client bq_client = bigquery.Client(credentials=credentials, project=credentials.project_id) # Build query to get data from the table query = f""" SELECT * FROM `{credentials.project_id}.{dataset_id}.{table_id}` LIMIT {limit} """ # Run the query query_job = bq_client.query(query) results = query_job.result() # Convert to dataframe df = results.to_dataframe() # Get table schema for metadata table_ref = bq_client.dataset(dataset_id).table(table_id) table = bq_client.get_table(table_ref) return { "data": df, "num_rows": table.num_rows, "size_bytes": table.num_bytes, "schema": [field.name for field in table.schema] } def setup_groq_client(): """Setup GROQ client with API key from environment variables""" # Load environment variables from .env file load_dotenv() # Get API key from environment variable api_key = os.environ.get("GROQ_API") if api_key: return Groq(api_key=api_key) else: st.sidebar.warning("GROQ_API environment variable not found. Chatbot functionality will be limited.") return None def process_documents(): """Process documentation and past analysis results to create a knowledge base""" # Create a directory for storing app documentation if it doesn't exist os.makedirs("app_docs", exist_ok=True) # Create basic app documentation if it doesn't exist app_doc_path = "app_docs/app_info.txt" if not os.path.exists(app_doc_path): with open(app_doc_path, "w") as f: f.write(""" Cosmick Cloud AI Analyzer Features: 1. Image Analysis - Analyze images for labels, objects, text, and faces using Google Cloud Vision AI 2. Video Analysis - Process videos to detect objects, faces, and text 3. Document Analysis - Extract text and structure from documents 4. Data Analysis - Upload, query, and visualize data using Google BigQuery Usage instructions: - Select a tool from the navigation bar - Follow the instructions for each tool - Ask questions using the chat assistant for help """) # Load documents documents = [] # Load app documentation try: loader = TextLoader(app_doc_path) documents.extend(loader.load()) except Exception as e: st.warning(f"Could not load app documentation: {str(e)}") # Process documents if documents: text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200 ) return text_splitter.split_documents(documents) return [] def create_vectorstore(documents): """Create or update a vectorstore with embeddings""" try: # Check if an API key is available for embeddings api_key = os.environ.get("OPENAI_API_KEY") if not api_key: st.warning("OpenAI API Key not found. Using default embeddings.") return None # Initialize embeddings embeddings = OpenAIEmbeddings(openai_api_key=api_key) # Create or load the vectorstore if os.path.exists("vectorstore") and os.path.isdir("vectorstore"): try: vectorstore = FAISS.load_local("vectorstore", embeddings) # Add new documents to existing vectorstore if documents: vectorstore.add_documents(documents) except Exception as e: st.warning(f"Error loading existing vectorstore: {str(e)}") # Create a new vectorstore vectorstore = FAISS.from_documents(documents, embeddings) else: # Create a new vectorstore vectorstore = FAISS.from_documents(documents, embeddings) # Save the updated vectorstore vectorstore.save_local("vectorstore") return vectorstore except Exception as e: st.warning(f"Error creating vectorstore: {str(e)}") return None def update_vectorstore_with_results(results): """Update the vectorstore with new analysis results""" if not results: return try: # Convert results to document format based on type results_text = "" timestamp = time.strftime("%Y-%m-%d %H:%M:%S") # Check what type of results we have if isinstance(results, dict): if "labels" in results: # Image analysis results results_text = f""" Image Analysis Results at {results.get('timestamp', timestamp)}: Labels detected: {', '.join(results.get('labels', {}).keys())} Objects detected: {', '.join(results.get('objects', {}).keys())} Text detected: {results.get('text', 'None')} """ elif "detection_stats" in results: # Video analysis results detection_stats = results.get("detection_stats", {}) results_text = f""" Video Analysis Results at {timestamp}: Objects detected: {', '.join(detection_stats.get('objects', {}).keys())} Faces detected: {detection_stats.get('faces', 0)} Text blocks detected: {detection_stats.get('text_blocks', 0)} Labels detected: {', '.join(detection_stats.get('labels', {}).keys())} """ elif "data" in results: # Data analysis results results_text = f""" Data Analysis Results at {timestamp}: Dataset loaded with {results.get('num_rows', 0)} rows Columns: {', '.join(results.get('schema', []))} """ elif isinstance(results, tuple) and len(results) == 3: # Document analysis results text, entities, tables = results entities_text = ", ".join(f"{k}: {v}" for k, v in entities.items()) tables_info = f"{len(tables)} tables extracted" if tables else "No tables extracted" results_text = f""" Document Analysis Results at {timestamp}: Extracted text length: {len(text)} characters Entities detected: {entities_text} Tables: {tables_info} """ elif isinstance(results, pd.DataFrame): # Query results results_text = f""" Query Results at {timestamp}: Retrieved {len(results)} rows of data Columns: {', '.join(results.columns)} """ # Create a document if results_text: text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200 ) docs = text_splitter.create_documents([results_text]) # Initialize vectorstore if it doesn't exist in session state if "vectorstore" not in st.session_state: docs_data = process_documents() st.session_state.vectorstore = create_vectorstore(docs_data) # Add the new documents to the vectorstore if st.session_state.vectorstore and docs: api_key = os.environ.get("OPENAI_API_KEY") if api_key: embeddings = OpenAIEmbeddings(openai_api_key=api_key) st.session_state.vectorstore.add_documents(docs) st.session_state.vectorstore.save_local("vectorstore") except Exception as e: st.warning(f"Error updating vectorstore: {str(e)}") def setup_rag_chain(): """Set up a RAG chain with Groq LLM and vectorstore""" if "vectorstore" not in st.session_state: # Initialize vectorstore with documentation docs = process_documents() st.session_state.vectorstore = create_vectorstore(docs) if st.session_state.vectorstore is None: return None try: # Set up Groq client if "groq_client" not in st.session_state: st.session_state.groq_client = setup_groq_client() if not st.session_state.groq_client: return None # Initialize conversational memory memory = ConversationBufferMemory( memory_key="chat_history", return_messages=True ) # Create the RAG chain retriever = st.session_state.vectorstore.as_retriever( search_type="similarity", search_kwargs={"k": 5} ) # Initialize chat model api_key = os.environ.get("GROQ_API") if not api_key: return None llm = ChatGroq(api_key=api_key, model_name="llama3-70b-8192") # Create the chain chain = ConversationalRetrievalChain.from_llm( llm=llm, retriever=retriever, memory=memory, return_source_documents=True ) return chain except Exception as e: st.warning(f"Error setting up RAG chain: {str(e)}") return None def parse_command(command, analysis_types=None): """Parse user command and return function to execute""" command = command.lower().strip() # Image analysis commands if "analyze image" in command and analysis_types: return "analyze_image", analysis_types # Video analysis commands elif "process video" in command and analysis_types: return "process_video", analysis_types # Data analysis commands elif "run query" in command: query = re.search(r"run query\s*:\s*(.*)", command, re.IGNORECASE) if query: return "run_query", query.group(1) # Document analysis commands elif "process document" in command: return "process_document", None # Help commands elif any(x in command for x in ["what can you do", "help", "capabilities"]): return "help", None # No command detected return None, None def execute_command(command_type, params): """Execute a command based on the parsed command""" if command_type == "analyze_image": st.write("To analyze an image, please upload an image in the Image Analysis section and select your desired analysis types.") return "I can help you analyze images. Please upload an image in the Image Analysis section and select which features you want to detect." elif command_type == "process_video": st.write("To process a video, please upload a video in the Video Analysis section and select your desired analysis types.") return "I can help you process videos. Please upload a video in the Video Analysis section and select which features you want to detect." elif command_type == "run_query": st.write("To run a BigQuery query, please go to the Data Analysis section.") return f"I can help you run the query: {params}. Please go to the Data Analysis section to execute it." elif command_type == "process_document": st.write("To process a document, please upload a document in the Document Analysis section.") return "I can help you analyze documents. Please upload a document in the Document Analysis section." elif command_type == "help": capabilities = """ I can help you with several tasks in the Cosmick Cloud AI Analyzer: 1. **Image Analysis** - I can identify objects, text, faces, and labels in images 2. **Video Analysis** - I can process videos to detect objects, faces, and text 3. **Document Analysis** - I can extract text and structure from documents 4. **Data Analysis** - I can help query and visualize data in BigQuery Try asking me specific questions about your analysis results or how to use the app! """ return capabilities return None def get_assistant_response(client, prompt, context=None, model="llama3-70b-8192"): """Get response from GROQ assistant with context awareness""" if client is None: return "I'm unable to connect to my knowledge base right now. Please check the API configuration." # Check if the input is a command command_type, params = parse_command(prompt) if command_type: command_response = execute_command(command_type, params) if command_response: return command_response # Set up RAG chain if available rag_chain = None try: if "vectorstore" in st.session_state and st.session_state.vectorstore: rag_chain = setup_rag_chain() except Exception as e: st.warning(f"Error setting up RAG: {str(e)}") # If RAG is available, use it for enhanced responses if rag_chain: try: result = rag_chain({"question": prompt}) return result["answer"] except Exception as e: st.warning(f"RAG error: {str(e)}, falling back to standard response") # Build a context-aware prompt as fallback if context: full_prompt = f"""You are an AI assistant for the Cosmick Cloud AI Analyzer application. Current application context: {context} User question: {prompt} Please provide a helpful, accurate response based on the current application context. If you need more specific information to answer correctly, please ask for it.""" else: full_prompt = f"""You are an AI assistant for the Cosmick Cloud AI Analyzer application. The application has the following tools: 1. Image Analysis: Analyzes images for labels, objects, text, and faces 2. Video Analysis: Processes videos to detect objects, faces, and text 3. Document Analysis: Extracts text and structure from documents 4. Data Analysis: Uploads, queries, and visualizes data in BigQuery User question: {prompt} Please provide a helpful response to guide the user on using these tools.""" # Call GROQ API try: chat_completion = client.chat.completions.create( messages=[ { "role": "system", "content": "You are a helpful, knowledgeable assistant for the Cosmick Cloud AI Analyzer application." }, { "role": "user", "content": full_prompt } ], model=model, temperature=0.5, max_tokens=1024, top_p=1, stream=False, ) return chat_completion.choices[0].message.content except Exception as e: return f"I encountered an error: {str(e)}. Please try again or check the API configuration." def chatbot_interface(): """Create a chatbot interface at the bottom of the app""" # Initialize chat history if "messages" not in st.session_state: st.session_state.messages = [] # Initialize GROQ client if "groq_client" not in st.session_state: st.session_state.groq_client = setup_groq_client() # Get current context current_context = "" if "current_tool" in st.session_state: current_context += f"Current tool: {st.session_state.current_tool}\n" if "analysis_results" in st.session_state: current_context += f"Analysis results: {st.session_state.analysis_results}\n" # Create chatbot container with improved styling st.markdown('

', unsafe_allow_html=True) st.markdown('

💬 Cosmick AI Assistant

', unsafe_allow_html=True) # Clear conversation button st.markdown('

', unsafe_allow_html=True) if st.button("Clear conversation"): st.session_state.messages = [] st.rerun() st.markdown('

', unsafe_allow_html=True) # Display chat message history with improved styling st.markdown('

', unsafe_allow_html=True) for message in st.session_state.messages: if message["role"] == "user": st.markdown(f'

{message["content"]}

', unsafe_allow_html=True) else: st.markdown(f'

{message["content"]}

', unsafe_allow_html=True) st.markdown('

', unsafe_allow_html=True) # Chat input user_input = st.chat_input("Ask me anything about image, video, document or data analysis...") if user_input: # Add user message to history st.session_state.messages.append({"role": "user", "content": user_input}) # Check if it's a command command_type, params = parse_command(user_input) if command_type: # Execute command response = execute_command(command_type, params) else: # Get assistant response response = get_assistant_response( st.session_state.groq_client, user_input, context=current_context ) # Add assistant response to history st.session_state.messages.append({"role": "assistant", "content": response}) # Rerun to update chat display st.rerun() st.markdown('

', unsafe_allow_html=True) def main(): # Header - Updated title st.markdown('

Cosmick Cloud AI Analyzer

', unsafe_allow_html=True) # Navigation selected = option_menu( menu_title=None, options=["Image Analysis", "Video Analysis", "Document Analysis", "Data Analysis", "About"], icons=["image", "camera-video", "file-text", "bar-chart", "info-circle"], menu_icon="cast", default_index=0, orientation="horizontal", ) # Store current tool in session state for context st.session_state.current_tool = selected if selected == "Image Analysis": # Sidebar controls with st.sidebar: st.markdown("### Analysis Settings") # Add mode selection processing_mode = st.radio("Processing Mode", ["Single Image", "Batch Processing (up to 5 images)"]) # Analysis types selection st.write("Choose analysis types:") analysis_types = [] if st.checkbox("Label Detection", value=True): analysis_types.append("Labels") if st.checkbox("Object Detection", value=True): analysis_types.append("Objects") if st.checkbox("Text Recognition", value=True): analysis_types.append("Text") if st.checkbox("Face Detection"): analysis_types.append("Face Detection") # New enhanced analysis options if st.checkbox("Visual Attributes (Colors)", value=False): analysis_types.append("Visual Attributes") st.markdown("---") # Confidence threshold control confidence_threshold = st.slider("Detection Confidence Threshold", min_value=0.0, max_value=1.0, value=0.5, help="Filter results based on confidence level") # Image quality settings st.write("Image settings:") quality = st.slider("Image Quality", min_value=0, max_value=100, value=100) st.markdown("---") st.info("This application analyzes images using Google Cloud Vision AI. Upload an image to get started.") # Main content if processing_mode == "Single Image": st.markdown("## Single Image Analysis") uploaded_file = st.file_uploader("Choose an image...", type=["jpg", "jpeg", "png"]) if uploaded_file is not None: # Convert uploaded file to image image = Image.open(uploaded_file) # Apply quality adjustment if needed if quality < 100: img_byte_arr = io.BytesIO() image.save(img_byte_arr, format='JPEG', quality=quality) image = Image.open(img_byte_arr) # Show original image st.markdown('

Original Image

', unsafe_allow_html=True) st.image(image, use_container_width=True) # Add analyze button if st.button("Analyze Image"): if not analysis_types: st.warning("Please select at least one analysis type.") else: with st.spinner("Analyzing image..."): # Call analyze function annotated_img, labels, objects, text, colors, text_language = analyze_image(image, analysis_types) # Display results display_results(annotated_img, labels, objects, text, colors, text_language) # Add download button for the annotated image buf = io.BytesIO() annotated_img.save(buf, format="PNG") byte_im = buf.getvalue() st.download_button( label="Download Annotated Image", data=byte_im, file_name="annotated_image.png", mime="image/png" ) else: # Batch Processing mode st.markdown("## Batch Image Analysis") st.info("Upload up to 5 images for batch processing.") uploaded_files = st.file_uploader("Choose images...", type=["jpg", "jpeg", "png"], accept_multiple_files=True) if uploaded_files and len(uploaded_files) > 0: if len(uploaded_files) > 5: st.warning("You've uploaded more than 5 images. Only the first 5 will be processed.") uploaded_files = uploaded_files[:5] if st.button("Process Batch"): st.write(f"Processing {len(uploaded_files)} images...") # Process each image with a unique key for each download button for i, uploaded_file in enumerate(uploaded_files): st.markdown(f"### Image {i+1}: {uploaded_file.name}") # Open and process the image try: image = Image.open(uploaded_file) annotated_img, labels, objects, text, colors, text_language = analyze_image( image, analysis_types, confidence_threshold ) # Create a unique identifier for this image image_id = f"{i}_{uploaded_file.name.replace(' ', '_')}" # Display results with unique download button keys col1, col2 = st.columns([3, 2]) with col1: st.image(annotated_img, use_container_width=True) with col2: # Display analysis results if labels: st.markdown("##### Labels Detected") for label, confidence in labels.items(): st.write(f"{label}: {confidence}%") if objects: st.markdown("##### Objects Detected") for obj, confidence in objects.items(): st.write(f"{obj}: {confidence}%") if text: st.markdown("##### Text Detected") if text_language: st.markdown(f"**Language:** {text_language}") st.text(text) if colors: st.markdown("##### Dominant Colors") for color_name, color_data in colors.items(): rgb = color_data["rgb"] hex_color = f"#{rgb[0]:02x}{rgb[1]:02x}{rgb[2]:02x}" st.markdown(f"

{color_name}: {color_data['score']}%", unsafe_allow_html=True) # Create summary image for download summary_img = create_summary_image(annotated_img, labels, objects, text, colors) buf = io.BytesIO() summary_img.save(buf, format="JPEG", quality=90) byte_im = buf.getvalue() # Use unique key for each download button st.download_button( label=f"📥 Download Results for {uploaded_file.name}", data=byte_im, file_name=f"analysis_{image_id}.jpg", mime="image/jpeg", key=f"download_batch_{image_id}" # Unique key for each image ) st.markdown("---") # Add separator between images except Exception as e: st.error(f"Error processing {uploaded_file.name}: {str(e)}") elif selected == "Video Analysis": st.markdown('

Video Analysis

', unsafe_allow_html=True) # Analysis settings st.sidebar.markdown("### Video Analysis Settings") # Add processing mode selection processing_mode = st.sidebar.radio( "Processing Engine", ["Hybrid (Google Vision + OpenCV)", "Google Vision API Only", "OpenCV Only"], help="Select which technology to use for video analysis" ) # Common analysis types selection st.sidebar.markdown("### Detection Types") analysis_types = [] if st.sidebar.checkbox("Object Detection", value=True): analysis_types.append("Objects") if st.sidebar.checkbox("Face Detection"): analysis_types.append("Face Detection") if st.sidebar.checkbox("Text Recognition"): analysis_types.append("Text") # Add motion tracking option if st.sidebar.checkbox("Motion Tracking", value=True): analysis_types.append("Motion") # Settings specific to the selected processing mode st.sidebar.markdown("---") st.sidebar.markdown(f"### {processing_mode} Settings") # Parameters for all modes track_update_frames = 5 confidence_threshold = 0.5 # Initialize variables with default values vision_update_interval = 1.0 max_results = 10 enable_face_landmarks = True tracking_algorithm = "KCF" motion_sensitivity = 32 prioritize_vision = "Google Vision (more accurate)" blend_results = True # Mode-specific parameters if processing_mode == "Google Vision API Only" or processing_mode == "Hybrid (Google Vision + OpenCV)": # Google Vision parameters st.sidebar.markdown("#### Google Vision Parameters") vision_update_interval = st.sidebar.slider( "Vision API update interval (seconds)", min_value=0.5, max_value=5.0, value=1.0, step=0.5, help="How often to call the Vision API (longer intervals save API quota)" ) confidence_threshold = st.sidebar.slider( "Google Vision Confidence Threshold", min_value=0.0, max_value=1.0, value=0.5, help="Minimum confidence score for Google Vision detections" ) # Detailed API options (using an expander for advanced settings) with st.sidebar.expander("Advanced Vision API Settings"): max_results = st.slider( "Max objects per frame", min_value=1, max_value=20, value=10, help="Maximum number of objects to detect per frame" ) enable_face_landmarks = st.checkbox( "Enable Face Landmarks", value=True, help="Detect facial features (eyes, nose, etc.)" ) if processing_mode == "OpenCV Only" or processing_mode == "Hybrid (Google Vision + OpenCV)": # OpenCV parameters st.sidebar.markdown("#### OpenCV Parameters") # Add YOLO model download option if models aren't found models_dir = os.path.join(os.path.dirname(__file__), "models") weights_path = os.path.join(models_dir, "yolov3.weights") config_path = os.path.join(models_dir, "yolov3.cfg") if not os.path.exists(models_dir): os.makedirs(models_dir, exist_ok=True) if not (os.path.exists(weights_path) and os.path.exists(config_path)): st.sidebar.warning("⚠️ YOLO models not found. Using basic people detector.") # Create a download button if st.sidebar.button("Download YOLO Models"): # Use a placeholder in the sidebar to show status download_status = st.sidebar.empty() download_status.info("Downloading YOLO models... Please wait.") # Ensure models directory exists os.makedirs(models_dir, exist_ok=True) # Download YOLOv3 config try: import urllib.request # Download config file if not os.path.exists(config_path): download_status.info("Downloading configuration file...") urllib.request.urlretrieve( "https://raw.githubusercontent.com/pjreddie/darknet/master/cfg/yolov3.cfg", config_path ) # Download weights file (this is large - about 240MB) if not os.path.exists(weights_path): download_status.info("Downloading weights file (large, ~240MB)...") urllib.request.urlretrieve( "https://pjreddie.com/media/files/yolov3.weights", weights_path ) download_status.success("✅ YOLO models downloaded successfully! Please refresh the page.") except Exception as e: download_status.error(f"Error downloading YOLO models: {str(e)}") download_status.info("You can manually download the models from: https://pjreddie.com/darknet/yolo/") else: st.sidebar.success("✅ YOLO models found. Using advanced object detection.") track_update_frames = st.sidebar.slider( "Update OpenCV tracking every N frames", min_value=1, max_value=15, value=5, help="Lower values = more accurate tracking but higher processing load" ) if processing_mode == "OpenCV Only": # Only show this in OpenCV-only mode confidence_threshold = st.sidebar.slider( "OpenCV Detector Confidence Threshold", min_value=0.0, max_value=1.0, value=0.4, help="Minimum confidence score for OpenCV detections" ) # OpenCV tracking options with st.sidebar.expander("OpenCV Tracking Options"): tracking_algorithm = st.selectbox( "Tracking Algorithm", ["KCF", "CSRT", "MOSSE", "MedianFlow"], index=0, help="Different algorithms have different speed/accuracy tradeoffs" ) motion_sensitivity = st.slider( "Motion Sensitivity", min_value=10, max_value=100, value=32, help="Lower values detect more subtle motion" ) # Hybrid-specific settings if processing_mode == "Hybrid (Google Vision + OpenCV)": # Hybrid specific parameters st.sidebar.markdown("#### Hybrid Mode Settings") prioritize_vision = st.sidebar.radio( "When results conflict, prioritize:", ["Google Vision (more accurate)", "OpenCV (faster)"], index=0, help="Which detection source to prioritize when there are conflicting results" ) blend_results = st.sidebar.checkbox( "Blend detection results", value=True, help="Combine detections from both systems for better accuracy" ) # Display warning about API usage st.sidebar.markdown("---") if processing_mode != "OpenCV Only": st.sidebar.warning("⚠️ Google Vision API usage may incur costs. Use responsibly.") # Upload Video mode only - removed real-time camera option st.markdown(""" #### 📤 Video Analysis Upload a video file to analyze it using the selected processing engine. **Instructions:** 1. Select the processing mode and parameters in the sidebar 2. Upload a video file (MP4, MOV, AVI) 3. Click "Process Video" to begin analysis 4. Download the processed video when complete **Note:** Videos are limited to 10 seconds of processing to manage API usage. """) # File uploader for videos uploaded_file = st.file_uploader("Choose a video file", type=["mp4", "mov", "avi"]) if uploaded_file is not None: # Display file info file_details = {"Filename": uploaded_file.name, "Size": f"{uploaded_file.size / (1024*1024):.2f} MB"} st.write("### File Details") st.json(file_details) # Process video button if st.button("Process Video"): if not analysis_types: st.warning("Please select at least one analysis type.") else: with st.spinner(f"Processing video with {processing_mode} mode (max 10 seconds)..."): try: # Create a base dict with common parameters processing_params = { "processing_mode": processing_mode, "track_update_frames": track_update_frames, "confidence_threshold": confidence_threshold, } # Add mode-specific parameters if processing_mode == "Google Vision API Only" or processing_mode == "Hybrid (Google Vision + OpenCV)": processing_params.update({ "vision_update_interval": vision_update_interval, "max_results": max_results, "enable_face_landmarks": enable_face_landmarks }) if processing_mode == "OpenCV Only" or processing_mode == "Hybrid (Google Vision + OpenCV)": processing_params.update({ "tracking_algorithm": tracking_algorithm, "motion_sensitivity": motion_sensitivity }) if processing_mode == "Hybrid (Google Vision + OpenCV)": processing_params.update({ "prioritize_vision": prioritize_vision == "Google Vision (more accurate)", "blend_results": blend_results }) # Add to the OpenCV parameters section: with st.sidebar.expander("YOLO Class Filters"): # Allow users to select which object classes to detect st.markdown("Select which objects to detect:") # Create a multiselect with common categories selected_categories = st.multiselect( "Object Categories", ["People", "Vehicles", "Animals", "Indoor Objects", "Sports Equipment", "Food", "All"], default=["People", "Vehicles"] ) # Map categories to actual YOLO classes yolo_classes = [] if "People" in selected_categories: yolo_classes.extend(["person"]) if "Vehicles" in selected_categories: yolo_classes.extend(["bicycle", "car", "motorcycle", "airplane", "bus", "train", "truck"]) if "Animals" in selected_categories: yolo_classes.extend(["bird", "cat", "dog", "horse", "sheep", "cow", "elephant", "bear", "zebra", "giraffe"]) if "All" in selected_categories: yolo_classes = None # Detect all classes # Pass this to your processing function in the processing_params processing_params["enabled_classes"] = yolo_classes # Process the video with the parameters processed_video, results = process_video_file(uploaded_file, analysis_types, **processing_params) if processed_video: # Offer download of processed video st.success("Video processing complete!") st.download_button( label="⬇️ Download Processed Video", data=processed_video, file_name=f"processed_{uploaded_file.name}", mime="video/mp4" ) # Show detailed analysis results st.markdown("### Detailed Analysis Results") # Display object detection summary if "Objects" in analysis_types and results["detection_stats"]["objects"]: st.markdown("#### 📦 Objects Detected") # Sort objects by frequency sorted_objects = dict(sorted(results["detection_stats"]["objects"].items(), key=lambda x: x[1], reverse=True)) # Create bar chart for objects if sorted_objects: fig, ax = plt.subplots(figsize=(10, 5)) objects = list(sorted_objects.keys()) counts = list(sorted_objects.values()) ax.barh(objects, counts, color='skyblue') ax.set_xlabel('Number of Detections') ax.set_title('Objects Detected in Video') st.pyplot(fig) # List with counts col1, col2 = st.columns(2) with col1: st.markdown("**Top Objects:**") for obj, count in list(sorted_objects.items())[:10]: st.markdown(f"- {obj}: {count} occurrences") else: st.info("No objects were detected in the video.") # Display face detection summary if "Face Detection" in analysis_types: st.markdown("#### 👤 Face Analysis") if results["detection_stats"]["faces"] > 0: st.markdown(f"Total faces detected: {results['detection_stats']['faces']}") else: st.info("No faces were detected in the video.") # Display text detection summary if "Text" in analysis_types: st.markdown("#### 📝 Text Analysis") if results["detection_stats"]["text_blocks"] > 0: st.markdown(f"Total text blocks detected: {results['detection_stats']['text_blocks']}") else: st.info("No text was detected in the video.") # Display scene analysis if "Motion" in analysis_types: st.markdown("#### 🎬 Scene Analysis") # Display scene changes if results["detection_stats"]["scene_changes"]: st.markdown(f"**Scene Changes:** {len(results['detection_stats']['scene_changes'])} detected") st.markdown("Scene changes at time points (seconds):") scene_times = [f"{t:.2f}s" for t in results["detection_stats"]["scene_changes"]] st.write(", ".join(scene_times)) # Activity metrics visualization if results["detection_stats"]["activity_metrics"]: st.markdown("**Activity Level Over Time:**") activity_data = results["detection_stats"]["activity_metrics"] times = [point[0] for point in activity_data] levels = [point[1] for point in activity_data] fig, ax = plt.subplots(figsize=(10, 4)) ax.plot(times, levels, 'r-') ax.set_xlabel('Time (seconds)') ax.set_ylabel('Activity Level') ax.set_title('Motion Activity Throughout Video') ax.grid(True, alpha=0.3) st.pyplot(fig) except Exception as e: st.error(f"Error processing video: {str(e)}") elif selected == "Document Analysis": st.markdown('

Document Processing & Analysis

', unsafe_allow_html=True) # Sidebar controls for document analysis with st.sidebar: st.markdown("### Document Analysis Settings") # Select document processor type processor_type = st.selectbox( "Select Document Type", ["Document OCR", "Form Parser", "Layout Parser", "Invoice Parser", "ID Document"] ) # Mapping of processor types to processor IDs with all your actual IDs processor_mapping = { "Document OCR": "4c80189b1a7863b0", # ocr-process "Form Parser": "47542cfc343edcac", # Form-Process "Layout Parser": "54e2616441b939e5", # layout-process "Invoice Parser": "e6efe8aa1d3afa61", # invoice-parser "ID Document": "894a011b810ebfee" # ID-parser } st.markdown("---") st.info("Upload a document to extract information using Google Document AI.") # Main content uploaded_file = st.file_uploader( "Upload a document (PDF, TIFF, JPG, PNG)", type=["pdf", "tiff", "jpg", "jpeg", "png"] ) if uploaded_file is not None: # Display file details file_details = { "Filename": uploaded_file.name, "File size": f"{uploaded_file.size / 1024:.2f} KB", "File type": uploaded_file.type } st.write("### File Details") for key, value in file_details.items(): st.write(f"**{key}:** {value}") # If it's an image file, display it if uploaded_file.type.startswith('image/'): st.image(uploaded_file, caption="Uploaded Document", use_container_width=True) else: st.info("PDF document uploaded (preview not available)") # Process button if st.button("Process Document"): with st.spinner("Processing document..."): # Get processor ID based on selection processor_id = processor_mapping[processor_type] # Get file content file_content = uploaded_file.getvalue() # Process document try: text, entities, tables = analyze_document(file_content, processor_id) # Display results st.markdown("### Document Analysis Results") # Show extracted information in tabs tab1, tab2, tab3 = st.tabs(["Text", "Extracted Fields", "Tables"]) with tab1: st.markdown("#### Extracted Text") st.markdown('

', unsafe_allow_html=True) st.write(text) st.markdown('

', unsafe_allow_html=True) with tab2: st.markdown("#### Extracted Fields") st.markdown('

', unsafe_allow_html=True) if entities: for entity_type, value in entities.items(): st.markdown(f"**{entity_type}:** {value}") else: st.info("No fields extracted from this document.") st.markdown('

', unsafe_allow_html=True) with tab3: st.markdown("#### Extracted Tables") if tables: for i, table in enumerate(tables): st.markdown(f"**Table {i+1}**") df = pd.DataFrame(table["data"], columns=table["headers"]) st.dataframe(df) else: st.info("No tables found in this document.") except Exception as e: st.error(f"Error processing document: {str(e)}") elif selected == "Data Analysis": st.markdown('

BigQuery Data Analysis

', unsafe_allow_html=True) # Sidebar controls for BigQuery with st.sidebar: st.markdown("### BigQuery Settings") # List existing resources try: resources = list_bigquery_resources() # Direct data selection option st.markdown("### Select Existing Data") if resources: # Dataset selection dataset_options = list(resources.keys()) dataset_options.insert(0, "-- Select a dataset --") selected_dataset = st.selectbox("Dataset", dataset_options) # Table selection (dependent on dataset) table_options = [] if selected_dataset and selected_dataset != "-- Select a dataset --": table_options = resources[selected_dataset] if not table_options: st.info("No tables in this dataset") table_options.insert(0, "-- Select a table --") selected_table = st.selectbox("Table", table_options) # Load button if selected_dataset != "-- Select a dataset --" and selected_table != "-- Select a table --": if st.button("Load Selected Table"): with st.spinner(f"Loading data from {selected_dataset}.{selected_table}..."): try: # Load the data result = load_bigquery_table(selected_dataset, selected_table) # Store in session state for use in other tabs st.session_state["table_info"] = { "dataset_id": selected_dataset, "table_id": selected_table, "schema": result["schema"] } st.session_state["query_results"] = result["data"] # Show success message st.success(f"Loaded {result['data'].shape[0]} rows from {selected_dataset}.{selected_table}") except Exception as e: st.error(f"Error loading table: {str(e)}") else: st.info("No datasets found in this project") except Exception as e: st.error(f"Error listing resources: {str(e)}") st.markdown("---") # Manual dataset and table settings for upload st.markdown("### Upload New Data") dataset_id = st.text_input("Dataset ID", "my_dataset") table_id = st.text_input("Table ID", "my_table") # Upload options replace_data = st.radio( "Upload Mode:", ["Replace existing data", "Append to existing data"] ) # Tabs for different actions upload_tab, explore_tab, query_tab, visualization_tab = st.tabs(["Upload Data", "Explore Data", "Query Data", "Visualize Data"]) # New Explore Data tab with explore_tab: st.markdown("### Explore BigQuery Data") if "query_results" in st.session_state and not st.session_state["query_results"].empty: df = st.session_state["query_results"] # Show summary of the data st.write("### Data Summary") st.write(f"**Rows:** {df.shape[0]}") st.write(f"**Columns:** {df.shape[1]}") # Display the data st.write("### Data Preview") st.dataframe(df.head(100)) # Display column information st.write("### Column Information") col_info = pd.DataFrame({ "Column": df.columns, "Type": df.dtypes, "Non-Null Count": df.count(), "Null Count": df.isnull().sum(), "Unique Values": [df[col].nunique() for col in df.columns] }) st.dataframe(col_info) # Quick statistics for numeric columns num_cols = df.select_dtypes(include=['int64', 'float64']).columns if not num_cols.empty: st.write("### Numeric Column Statistics") st.dataframe(df[num_cols].describe()) else: st.info("Select an existing dataset and table from the sidebar and click 'Load Selected Table', or upload a CSV file in the 'Upload Data' tab.") with upload_tab: st.markdown("### Upload Data to BigQuery") # File uploader for CSV files uploaded_file = st.file_uploader("Upload a CSV file", type=["csv"]) if uploaded_file is not None: # Display file details file_details = { "Filename": uploaded_file.name, "File size": f"{uploaded_file.size / 1024:.2f} KB" } # Show file preview try: df_preview = pd.read_csv(uploaded_file) st.write("### File Preview") st.dataframe(df_preview.head(5)) # Store dataframe in session state for other tabs st.session_state["query_results"] = df_preview # Upload button if st.button("Upload to BigQuery"): with st.spinner("Uploading to BigQuery..."): try: # Upload the file append = replace_data == "Append to existing data" result = upload_csv_to_bigquery(uploaded_file, dataset_id, table_id, append=append) # Show success message st.success(f"Successfully uploaded to {dataset_id}.{table_id}") st.write(f"Rows: {result['num_rows']}") st.write(f"Size: {result['size_bytes'] / 1024:.2f} KB") st.write(f"Schema: {', '.join(result['schema'])}") # Store table info in session state st.session_state["table_info"] = { "dataset_id": dataset_id, "table_id": table_id, "schema": result["schema"] } except Exception as e: st.error(f"Error uploading to BigQuery: {str(e)}") except Exception as e: st.error(f"Error reading CSV file: {str(e)}") else: st.info("Upload a CSV file to load data into BigQuery") with query_tab: st.markdown("### Query BigQuery Data") if "query_results" in st.session_state and "table_info" in st.session_state: # Display info about the loaded data table_info = st.session_state["table_info"] st.write(f"Working with table: **{table_info['dataset_id']}.{table_info['table_id']}**") # Query input default_query = f"SELECT * FROM `{credentials.project_id}.{table_info['dataset_id']}.{table_info['table_id']}` LIMIT 100" query = st.text_area("SQL Query", default_query, height=100) # Execute query button if st.button("Run Query"): with st.spinner("Executing query..."): try: # Run the query results = run_bigquery(query) # Store results in session state st.session_state["query_results"] = results # Display results st.write("### Query Results") st.dataframe(results) # Download button for results csv = results.to_csv(index=False) st.download_button( label="Download Results as CSV", data=csv, file_name="query_results.csv", mime="text/csv" ) except Exception as e: st.error(f"Error executing query: {str(e)}") else: st.info("Load a table from BigQuery or upload a CSV file first") with visualization_tab: st.markdown("### Visualize BigQuery Data") if "query_results" in st.session_state and not st.session_state["query_results"].empty: df = st.session_state["query_results"] # Chart type selection chart_type = st.selectbox( "Select Chart Type", ["Bar Chart", "Line Chart", "Scatter Plot", "Histogram", "Pie Chart"] ) # Column selection based on data types numeric_cols = df.select_dtypes(include=['int64', 'float64']).columns.tolist() all_cols = df.columns.tolist() if len(numeric_cols) < 1: st.warning("No numeric columns available for visualization") else: if chart_type in ["Bar Chart", "Line Chart", "Scatter Plot"]: col1, col2 = st.columns(2) with col1: x_axis = st.selectbox("X-axis", all_cols) with col2: y_axis = st.selectbox("Y-axis", numeric_cols) # Optional: Grouping/color dimension color_dim = st.selectbox("Color Dimension (Optional)", ["None"] + all_cols) # Generate the visualization based on selection if st.button("Generate Visualization"): st.write(f"### {chart_type}: {y_axis} by {x_axis}") if chart_type == "Bar Chart": if color_dim != "None": fig = px.bar(df, x=x_axis, y=y_axis, color=color_dim, title=f"{y_axis} by {x_axis}") else: fig = px.bar(df, x=x_axis, y=y_axis, title=f"{y_axis} by {x_axis}") st.plotly_chart(fig) elif chart_type == "Line Chart": if color_dim != "None": fig = px.line(df, x=x_axis, y=y_axis, color=color_dim, title=f"{y_axis} by {x_axis}") else: fig = px.line(df, x=x_axis, y=y_axis, title=f"{y_axis} by {x_axis}") st.plotly_chart(fig) elif chart_type == "Scatter Plot": if color_dim != "None": fig = px.scatter(df, x=x_axis, y=y_axis, color=color_dim, title=f"{y_axis} vs {x_axis}") else: fig = px.scatter(df, x=x_axis, y=y_axis, title=f"{y_axis} vs {x_axis}") st.plotly_chart(fig) elif chart_type == "Histogram": column = st.selectbox("Select Column", numeric_cols) bins = st.slider("Number of Bins", min_value=5, max_value=100, value=20) if st.button("Generate Visualization"): st.write(f"### Histogram of {column}") fig = px.histogram(df, x=column, nbins=bins, title=f"Distribution of {column}") st.plotly_chart(fig) elif chart_type == "Pie Chart": column = st.selectbox("Category Column", all_cols) value_col = st.selectbox("Value Column", numeric_cols) if st.button("Generate Visualization"): # Aggregate the data if needed pie_data = df.groupby(column)[value_col].sum().reset_index() st.write(f"### Pie Chart: {value_col} by {column}") fig = px.pie(pie_data, names=column, values=value_col, title=f"{value_col} by {column}") st.plotly_chart(fig) else: st.info("Load a table from BigQuery or upload a CSV file first") elif selected == "About": st.markdown("## About This App") st.write(""" This application uses Google Cloud Vision AI to analyze images and video streams. It can: - **Detect labels** in images - **Identify objects** and their locations - **Extract text** from images - **Detect faces** and facial landmarks - **Analyze real-time video** from your camera To use this app, you need to: 1. Set up Google Cloud Vision API credentials 2. Upload an image or use your camera 3. Select the types of analysis you want to perform 4. Click "Analyze Image" or start the video stream The app is built with Streamlit and Google Cloud Vision API. """) st.info("Note: Make sure your Google Cloud credentials are properly set up to use this application.") # Add the chatbot interface at the bottom of the page chatbot_interface() if __name__ == "__main__": # Use GOOGLE_CREDENTIALS directly - no need for file or GOOGLE_APPLICATION_CREDENTIALS try: if 'GOOGLE_CREDENTIALS' in os.environ: # Create credentials object directly from JSON string credentials_info = json.loads(os.environ['GOOGLE_CREDENTIALS']) credentials = service_account.Credentials.from_service_account_info(credentials_info) # Initialize client with these credentials directly client = vision.ImageAnnotatorClient(credentials=credentials) else: st.sidebar.error("GOOGLE_CREDENTIALS environment variable not found") client = None except Exception as e: st.sidebar.error(f"Error with credentials: {str(e)}") client = None main() # Add this function to your app def extract_video_frames(video_bytes, num_frames=5): """Extract frames from video bytes for thumbnail display with improved key frame selection""" import cv2 import numpy as np import tempfile from PIL import Image import io # Save video bytes to a temporary file with tempfile.NamedTemporaryFile(delete=False, suffix='.mp4') as temp_file: temp_file.write(video_bytes) temp_video_path = temp_file.name # Open the video file cap = cv2.VideoCapture(temp_video_path) # Get video properties frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) fps = cap.get(cv2.CAP_PROP_FPS) # Use more sophisticated frame selection based on content analysis frames = [] frame_scores = [] sample_interval = max(1, frame_count // (num_frames * 3)) # Sample more frames than needed # First pass: collect frame scores prev_frame = None frame_index = 0 while len(frame_scores) < num_frames * 3 and frame_index < frame_count: cap.set(cv2.CAP_PROP_POS_FRAMES, frame_index) ret, frame = cap.read() if not ret: break # Convert to grayscale for analysis gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) gray = cv2.GaussianBlur(gray, (21, 21), 0) # Calculate frame score based on Laplacian variance (focus measure) focus_score = cv2.Laplacian(gray, cv2.CV_64F).var() # Calculate frame difference if we have a previous frame diff_score = 0 if prev_frame is not None: frame_diff = cv2.absdiff(gray, prev_frame) diff_score = np.mean(frame_diff) # Combined score: favor sharp frames with significant changes combined_score = focus_score * 0.6 + diff_score * 0.4 frame_scores.append((frame_index, combined_score)) # Store frame for next comparison prev_frame = gray frame_index += sample_interval # Second pass: select the best frames based on scores # Sort by score and get top N frames sorted_frames = sorted(frame_scores, key=lambda x: x[1], reverse=True) best_frames = sorted_frames[:num_frames] # Sort back by frame index to maintain chronological order selected_frames = sorted(best_frames, key=lambda x: x[0]) # Extract the selected frames for idx, _ in selected_frames: cap.set(cv2.CAP_PROP_POS_FRAMES, idx) ret, frame = cap.read() if ret: # Apply subtle enhancement to frames enhanced_frame = frame.copy() # Auto color balance lab = cv2.cvtColor(enhanced_frame, cv2.COLOR_BGR2LAB) l, a, b = cv2.split(lab) clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8)) cl = clahe.apply(l) enhanced_lab = cv2.merge((cl, a, b)) enhanced_frame = cv2.cvtColor(enhanced_lab, cv2.COLOR_LAB2BGR) # Convert to RGB (from BGR) frame_rgb = cv2.cvtColor(enhanced_frame, cv2.COLOR_BGR2RGB) # Convert to PIL Image pil_img = Image.fromarray(frame_rgb) # Save to bytes img_byte_arr = io.BytesIO() pil_img.save(img_byte_arr, format='JPEG', quality=90) frames.append(img_byte_arr.getvalue()) # Clean up cap.release() import os os.unlink(temp_video_path) return frames