# Project export: CyberDrive This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small. ## Project metadata - Hackathon: TreeHacks 2025 - Tagline: CyberDrive - Transforming driving footage into intelligent insights. - Devpost: https://devpost.com/software/cyberdrive - GitHub: https://github.com/KoaChang/TeslaMCQTreehacks - Video: https://www.youtube.com/embed/BLKCyMwn-xY?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Result: winner (Tesla: Excellence Prize ($2k Tesla Store Gift Card [1st], $1k Gift Card [2nd], $1k Gift Card [3rd])) - Team: 0 GitHub contributor(s) — ## Devpost submission (written by the team) No Devpost description available. ## README (from the GitHub repository) No README available. ## Detected evidence (automated analysis) Indexed codebase: 16 recognized source files, 95 KB. - Python (language) — detected in the code - Google Gemini (technology) — claimed on Devpost, not found in the code - OpenAI (technology) — claimed on Devpost, not found in the code - PyTorch (technology) — claimed on Devpost, not found in the code ## Codebase structure (from repository index) ### Files (120 of 1353) ``` .DS_Store .env 8_frames_testing/.DS_Store 8_frames_testing/gpt4v_results_8frames/00001_result.json 8_frames_testing/gpt4v_results_8frames/00002_result.json 8_frames_testing/gpt4v_results_8frames/00003_result.json 8_frames_testing/gpt4v_results_8frames/00004_result.json 8_frames_testing/gpt4v_results_8frames/00005_result.json 8_frames_testing/gpt4v_results_8frames/00006_result.json 8_frames_testing/gpt4v_results_8frames/00007_result.json 8_frames_testing/gpt4v_results_8frames/00008_result.json 8_frames_testing/gpt4v_results_8frames/00009_result.json 8_frames_testing/gpt4v_results_8frames/00010_result.json 8_frames_testing/gpt4v_results_8frames/00011_result.json 8_frames_testing/gpt4v_results_8frames/00012_result.json 8_frames_testing/gpt4v_results_8frames/00013_result.json 8_frames_testing/gpt4v_results_8frames/00014_result.json 8_frames_testing/gpt4v_results_8frames/00015_result.json 8_frames_testing/gpt4v_results_8frames/00016_result.json 8_frames_testing/gpt4v_results_8frames/00017_result.json 8_frames_testing/gpt4v_results_8frames/00018_result.json 8_frames_testing/gpt4v_results_8frames/00019_result.json 8_frames_testing/gpt4v_results_8frames/00020_result.json 8_frames_testing/gpt4v_results_8frames/00021_result.json 8_frames_testing/gpt4v_results_8frames/00022_result.json 8_frames_testing/gpt4v_results_8frames/00023_result.json 8_frames_testing/gpt4v_results_8frames/00024_result.json 8_frames_testing/gpt4v_results_8frames/00025_result.json 8_frames_testing/gpt4v_results_8frames/00026_result.json 8_frames_testing/gpt4v_results_8frames/00027_result.json 8_frames_testing/gpt4v_results_8frames/00028_result.json 8_frames_testing/gpt4v_results_8frames/00029_result.json 8_frames_testing/gpt4v_results_8frames/00030_result.json 8_frames_testing/gpt4v_results_8frames/00031_result.json 8_frames_testing/gpt4v_results_8frames/00032_result.json 8_frames_testing/gpt4v_results_8frames/00033_result.json 8_frames_testing/gpt4v_results_8frames/00034_result.json 8_frames_testing/gpt4v_results_8frames/00035_result.json 8_frames_testing/gpt4v_results_8frames/00036_result.json 8_frames_testing/gpt4v_results_8frames/00037_result.json 8_frames_testing/gpt4v_results_8frames/00038_result.json 8_frames_testing/gpt4v_results_8frames/00039_result.json 8_frames_testing/gpt4v_results_8frames/00040_result.json 8_frames_testing/gpt4v_results_8frames/00041_result.json 8_frames_testing/gpt4v_results_8frames/00042_result.json 8_frames_testing/gpt4v_results_8frames/00043_result.json 8_frames_testing/gpt4v_results_8frames/00044_result.json 8_frames_testing/gpt4v_results_8frames/00045_result.json 8_frames_testing/gpt4v_results_8frames/00046_result.json 8_frames_testing/gpt4v_results_8frames/00047_result.json 8_frames_testing/gpt4v_results_8frames/00048_result.json 8_frames_testing/gpt4v_results_8frames/00049_result.json 8_frames_testing/gpt4v_results_8frames/00050_result.json 8_frames_testing/inference4o_8frames.py all_answers_1.csv all_answers_2.csv all_questions.csv choose_majority.py extract_answers.py extract_letter.py fill_gaps.py final_answers/00051_result.json final_answers/00052_result.json final_answers/00053_result.json final_answers/00054_result.json final_answers/00055_result.json final_answers/00056_result.json final_answers/00057_result.json final_answers/00058_result.json final_answers/00059_result.json final_answers/00060_result.json final_answers/00061_result.json final_answers/00062_result.json final_answers/00063_result.json final_answers/00064_result.json final_answers/00065_result.json final_answers/00066_result.json final_answers/00067_result.json final_answers/00068_result.json final_answers/00069_result.json final_answers/00070_result.json final_answers/00071_result.json final_answers/00072_result.json final_answers/00073_result.json final_answers/00074_result.json final_answers/00075_result.json final_answers/00076_result.json final_answers/00077_result.json final_answers/00078_result.json final_answers/00079_result.json final_answers/00080_result.json final_answers/00081_result.json final_answers/00082_result.json final_answers/00083_result.json final_answers/00084_result.json final_answers/00085_result.json final_answers/00086_result.json final_answers/00087_result.json final_answers/00088_result.json final_answers/00089_result.json final_answers/00090_result.json final_answers/00091_result.json final_answers/00092_result.json final_answers/00093_result.json final_answers/00094_result.json final_answers/00095_result.json final_answers/00096_result.json final_answers/00097_result.json final_answers/00098_result.json final_answers/00099_result.json final_answers/00100_result.json final_answers/00101_result.json final_answers/00102_result.json final_answers/00103_result.json final_answers/00104_result.json final_answers/00105_result.json final_answers/00106_result.json final_answers/00107_result.json final_answers/00108_result.json final_answers/00109_result.json [1233 more files omitted for size] ``` ### Dependencies No dependency index available. ### Recent commits (newest first) - upload extra files - upload files extra ## Key source files (fetched from GitHub, selected and truncated for size) ### choose_majority.py ```python import pandas as pd from collections import Counter def find_majority_answers(csv_files): """ Find the majority answer for each ID across multiple CSV files. Args: csv_files (list): List of CSV filenames to process Returns: pandas.DataFrame: DataFrame with ID and majority answer """ # Dictionary to store answers for each ID all_answers = {} # Process each CSV file for file in csv_files: try: # Read the CSV file df = pd.read_csv(file) # Ensure required columns exist if 'id' not in df.columns or 'answer' not in df.columns: print(f"Error: Required columns missing in {file}") continue # Process each row for _, row in df.iterrows(): id_num = row['id'] answer = row['answer'] # Initialize list for this ID if it doesn't exist if id_num not in all_answers: all_answers[id_num] = [] # Add this answer to the list for this ID all_answers[id_num].append(answer) except Exception as e: print(f"Error processing file {file}: {str(e)}") continue # Calculate majority answer for each ID majority_results = [] for id_num, answers in all_answers.items(): # Find the most common answer if answers: # Count occurrences of each answer counter = Counter(answers) max_count = max(counter.values()) # Get all answers that appear the maximum number of times max_answers = [ans for ans, count in counter.items() if count == max_count] # If there's a tie, use the alphabetically first answer majority_answer = min(max_answers) majority_results.append({ 'id': id_num, 'answer': majority_answer }) # Convert results to DataFrame and sort by ID result_df = pd.DataFrame(majority_results) result_df = result_df.sort_values('id') return result_df def main(): # List of CSV files to process csv_files = [ 'gemini_pro1.csv', 'gemini_pro2.csv', 'open_ai1.csv', 'open_ai2.csv', 'open_ai3.csv' ] # Find majority answers result_df = find_majority_answers(csv_files) # Save results to CSV output_file = 'majority_results.csv' result_df.to_csv(output_file, index=False) print(f"Results saved to {output_file}") if __name__ == "__main__": main() ``` ### extract_answers.py ```python import json import os import re import csv from pathlib import Path def extract_answer_from_json(json_content): """Extract the letter from tags, handling both simple letter and letter with additional content.""" answer_text = json_content.get('answer', '') # First try to match the pattern with additional content (e.g., "C. 27") match = re.search(r'([A-E])[.\s].*?', answer_text) if match: return match.group(1) # If no match, try to match just the letter match = re.search(r'([A-E])', answer_text) if match: return match.group(1) return None def process_files(folder_path): """Process all JSON files in the folder and create a CSV with results.""" # Create a list to store results results = [] # First, add 'X' for IDs 1-50 for id_num in range(1, 51): file_id = str(id_num).zfill(5) results.append([file_id, 'X']) # Get all JSON files in the folder folder = Path(folder_path) json_files = sorted(folder.glob('*.json')) # Keep track of processed IDs processed_ids = set() # Process each file (only for IDs 51-251) for json_file in json_files: # Extract ID from filename (00001 from 00001_result.json) file_id = json_file.name.split('_')[0] # Check if ID is within our range (51 to 251) try: id_num = int(file_id) if 51 <= id_num <= 251: # Read and parse JSON file try: with open(json_file, 'r', encoding='utf-8') as f: json_content = json.load(f) # Extract answer answer = extract_answer_from_json(json_content) if answer: results.append([file_id, answer]) processed_ids.add(id_num) else: print(f"Warning: No answer found in {json_file}") except Exception as e: print(f"Error processing {json_file}: {str(e)}") except ValueError: print(f"Warning: Invalid ID format in filename {json_file}") # Sort all results by ID results.sort(key=lambda x: int(x[0])) # Report any missing IDs in the range 51-251 expected_ids = set(range(51, 252)) missing_ids = expected_ids - processed_ids if missing_ids: print(f"Warning: Missing answers for IDs 51-251: {sorted(missing_ids)}") # Write results to CSV output_file = 'all_answers.csv' with open(output_file, 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f) writer.writerow(['id', 'answer']) # Write header writer.writerows(results) print(f"Results written to {output_file}") print(f"Processed {len(results)} files") print(f"First 50 IDs automatically set to 'X'") if __name__ == "__main__": # Specify the folder path containing JSON files folder_path = "o1_final_answers" # Process the files process_files(folder_path) ``` ### split_frames.py ```python import cv2 import os from pathlib import Path def extract_frames(video_dir): """ Extract 5 equally spaced frames from each video in the specified directory. Args: video_dir (str): Path to directory containing the videos """ # Create output directory if it doesn't exist output_base_dir = Path('extracted_frames') output_base_dir.mkdir(exist_ok=True) # Get all mp4 files in the directory video_files = list(Path(video_dir).glob('*.mp4')) for video_path in video_files: try: # Open the video cap = cv2.VideoCapture(str(video_path)) # Get video properties total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) fps = cap.get(cv2.CAP_PROP_FPS) if total_frames <= 0: print(f"Error: Could not read frames from {video_path.name}") continue # Calculate frame indices to extract (5 equally spaced frames) frame_indices = [int(i * (total_frames - 1) / 4) for i in range(5)] # Create directory for this video's frames video_name = video_path.stem output_dir = output_base_dir / video_name output_dir.mkdir(exist_ok=True) # Extract and save frames for frame_num, frame_idx in enumerate(frame_indices, 1): # Set frame position cap.set(cv2.CAP_PROP_POS_FRAMES, frame_idx) ret, frame = cap.read() if not ret: print(f"Error: Could not read frame {frame_idx} from {video_path.name}") continue # Add text to frame font = cv2.FONT_HERSHEY_DUPLEX # Changed to DUPLEX for bolder font text = f"frame {frame_num}" font_scale = 1.5 # Increased font size thickness = 3 # Get text size for background rectangle (text_width, text_height), baseline = cv2.getTextSize(text, font, font_scale, thickness) # Position text in bottom left with padding padding = 20 x = padding y = frame.shape[0] - padding # padding pixels from bottom # Draw semi-transparent background rectangle bg_rect_pts = [ (x - padding//2, y + padding//2), (x + text_width + padding//2, y - text_height - padding//2) ] overlay = frame.copy() cv2.rectangle(overlay, bg_rect_pts[0], bg_rect_pts[1], (0, 0, 0), -1) alpha = 0.6 # Transparency factor cv2.addWeighted(overlay, alpha, frame, 1 - alpha, 0, frame) # Add text with white outline for better visibility cv2.putText(frame, text, (x, y), font, font_scale, (255, 255, 255), thickness + 2) # outline cv2.putText(frame, text, (x, y), font, font_scale, (0, 0, 0), thickness) # main text # Save frame output_path = output_dir / f"frame_{frame_num}.jpg" cv2.imwrite(str(output_path), frame) print(f"Processed {video_path.name}: Extracted {len(frame_indices)} frames") except Exception as e: print(f"Error processing {video_path.name}: {str(e)}") finally: cap.release() if __name__ == "__main__": video_directory = "videos" # Change this if your videos are in a different directory extract_frames(video_directory) ``` ### extract_letter.py ```python import json import os import asyncio import aiohttp import aiofiles from typing import Optional from dotenv import load_dotenv async def process_single_file(file_path: str, session: aiohttp.ClientSession) -> Optional[tuple[str, str]]: """ Process a single JSON file and extract the answer choice using OpenAI API. Args: file_path: Path to the JSON file session: aiohttp ClientSession for making API calls Returns: Tuple of (filename, answer) if successful, None if failed """ try: # Read the JSON file asynchronously async with aiofiles.open(file_path, 'r') as f: content = await f.read() data = json.loads(content) # Extract the answer text answer_text = data.get('answer', '') if not answer_text: print(f"Warning: No answer found in {file_path}") return None # Create the API call to GPT-4 headers = { "Authorization": f"Bearer {os.getenv('OPENAI_API_KEY_KOA_4o')}", "Content-Type": "application/json" } payload = { "model": "gpt-4o-mini", "messages": [ { "role": "system", "content": "Extract the letter choice (A, B, C, D, or E) that is indicated as the answer in the text. Output the answer in tags like this: B" }, { "role": "user", "content": answer_text } ], "temperature": 0, "max_tokens": 50 } async with session.post( "https://api.openai.com/v1/chat/completions", headers=headers, json=payload ) as response: if response.status != 200: error_text = await response.text() print(f"API error for {file_path}: {error_text}") return None result = await response.json() answer = result['choices'][0]['message']['content'].strip() # Get video ID from filename video_id = os.path.splitext(os.path.basename(file_path))[0] # Create output directory if it doesn't exist output_dir = "gemini_pro_final_answers" os.makedirs(output_dir, exist_ok=True) # Write result to output file output_path = os.path.join(output_dir, f"{video_id}_result.json") async with aiofiles.open(output_path, 'w') as f: await f.write(json.dumps({"answer": answer}, indent=2)) print(f"Processed {video_id}: {answer}") return (video_id, answer) except Exception as e: print(f"Error processing {file_path}: {str(e)}") return None async def process_all_files(directory: str): """ Process all JSON files in the directory concurrently. Args: directory: Directory containing JSON files """ # Get all JSON files in the directory json_files = [ os.path.join(directory, f) for f in os.listdir(directory) if f.endswith('.json') ] # Configure rate limiting semaphore = asyncio.Semaphore(10) # Limit concurrent API calls async def process_with_semaphore(file_path: str, session: aiohttp.ClientSession): async with semaphore: return await process_single_file(file_path, session) # Process files concurrently async with aiohttp.ClientSession() as session: tasks = [ process_with_semaphore(file_path, session) for file_path in json_files ] results = await asyncio.gather(*tasks) # Count successful processes successful = len([r for r in results if r is not None]) print(f"\nProcessed {successful} files successfully") print(f"Results saved to {os.path.abspath('gemini_video_final_answers')} directory") async def main(): # Directory containing the JSON files directory = "gemini_pro_answers" # Process all files await process_all_files(directory) if __name__ == "__main__": load_dotenv() # Ensure you have set OPENAI_API_KEY environment variable if not os.getenv('OPENAI_API_KEY_KOA_4o'): print("Error: OPENAI_API_KEY environment variable not set") exit(1) # Run the async main function asyncio.run(main()) ``` ### gemini_video.py ```python import os import csv import asyncio import logging from pathlib import Path import json from asyncio import Semaphore from typing import List, Dict from google.cloud import aiplatform from vertexai.generative_models import GenerativeModel, Part # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class GeminiProcessor: def __init__(self, project_id: str, location: str = "us-central1", max_concurrent_requests: int = 5): """Initialize the Gemini processor""" # Initialize Vertex AI aiplatform.init( project=project_id, location=location ) self.model = GenerativeModel("gemini-2.0-flash-001") # self.model = GenerativeModel("gemini-1.5-pro") self.semaphore = Semaphore(max_concurrent_requests) self.request_count = 0 self.batch_size = 5 # Number of requests before sleeping self.sleep_duration = 0 # Sleep duration in seconds async def process_question(self, video_id: str, question: str) -> Dict: """Process a single question using Gemini API""" async with self.semaphore: try: # Direct GCS path to video video_path = f"gs://tesla_videos/videos/{video_id.zfill(5)}.mp4" # Create video part using GCS path video_part = Part.from_uri( uri=video_path, mime_type="video/mp4" ) # Prepare prompt prompt = f"""You are analyzing a dashcam video taken from the driver's forward-facing perspective. Using this video, answer the following multiple-choice question by choosing the single best answer. Incorporate any relevant details observed in the video (for example, lanes, signage, vehicles, pedestrians, traffic signals, road markings, obstructions) that might help in selecting the correct answer. Explain your reasoning in detail. Relate your findings to each of the multiple-choice options. Eliminate those that are inconsistent with the visual evidence or standard traffic rules, and select the most appropriate remaining choice. Conclude with the final choice that best matches the situation. Output that choice in `` tags. Question: {question} """ # Make API request contents = [video_part, prompt] response = self.model.generate_content(contents) # Increment request counter self.request_count += 1 # Check if we need to sleep if self.request_count % self.batch_size == 0: logger.info(f"Processed {self.request_count} requests. Sleeping for {self.sleep_duration} seconds...") await asyncio.sleep(self.sleep_duration) return { "video_id": video_id, "answer": response.text } except Exception as e: logger.error(f"Error processing video {video_id}: {str(e)}") return { "video_id": video_id, "error": str(e) } async def process_batch(self, questions: List[Dict]) -> List[Dict]: """Process multiple questions in parallel""" tasks = [] for q in questions: task = self.process_question(q['id'], q['question']) tasks.append(task) return await asyncio.gather(*tasks) async def main(): # Your Google Cloud project ID project_id = "tesla-451102" # Initialize processor processor = GeminiProcessor(project_id) # Create output directory output_dir = Path('gemini_video_answers') output_dir.mkdir(exist_ok=True) # Read questions from CSV questions = [] with open('all_questions.csv', 'r') as csvfile: reader = csv.DictReader(csvfile) for row in reader: # Convert ID to integer for comparison id_num = int(row['id']) row['id'] = str(id_num).zfill(5) questions.append(row) # Process questions logger.info(f"Processing {len(questions)} questions...") results = await processor.process_batch(questions) # Save results for result in results: video_id = result.pop('video_id') output_path = output_dir / f"{video_id}_result.json" with open(output_path, 'w') as f: json.dump(result, f, indent=2) logger.info(f"Saved result for video {video_id}") if __name__ == "__main__": asyncio.run(main()) ``` ### gemini_pro.py ```python import os import csv import asyncio import logging from pathlib import Path import json from asyncio import Semaphore from typing import List, Dict from google.cloud import aiplatform from vertexai.generative_models import GenerativeModel, Part import PIL.Image from io import BytesIO # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class GeminiProcessor: def __init__(self, project_id: str, location: str = "us-central1", max_concurrent_requests: int = 5): """Initialize the Gemini processor""" # Initialize Vertex AI aiplatform.init( project=project_id, location=location ) self.model = GenerativeModel("gemini-2.0-pro-exp-02-05") self.semaphore = Semaphore(max_concurrent_requests) def get_frame_paths(self, video_id: str) -> List[str]: """Get paths for all frames of a specific video""" frame_dir = Path('extracted_frames') / str(video_id) frame_paths = sorted(list(frame_dir.glob('frame_*.jpg'))) return [str(path) for path in frame_paths] def image_to_bytes(self, image: PIL.Image.Image) -> bytes: """Convert PIL Image to bytes""" img_byte_arr = BytesIO() image.save(img_byte_arr, format='JPEG') return img_byte_arr.getvalue() def create_prompt(self, question: str) -> str: """Create the prompt for Gemini""" return f"""You have 5 equally spaced frames (Frame 1 through Frame 5) captured from a 5-second dashcam video, taken from the driver’s forward-facing perspective. Using these frames, answer the following multiple-choice question by choosing the single best answer. Incorporate any relevant details observed in the frames (for example, lanes, signage, vehicles, pedestrians, traffic signals, road markings, obstructions) that might help in selecting the correct answer. Consider how details may change across the frames and note that some frames may be more crucial than others. Explain your reasoning in detail. Steps to follow: 1. **Frame-by-Frame Analysis:** Describe the significant elements you notice in each of the 5 frames (e.g., signs, road markings, obstructions, other vehicles, potential hazards). Make sure you particularly pay attention to road markings or signs with directional arrows whenever the question asks about the possible directions a given lane can go. 2. **Contextual Reasoning:** Integrate the observations from each frame. Think about what is happening over time, which elements are most relevant, and how they connect to the question. 3. **Match to Answer Choices:** Relate your findings to each of the multiple-choice options. Eliminate those that are inconsistent with the visual evidence or standard traffic rules, and select the most appropriate remaining choice. 4. **Provide the Best Answer:** Conclude with the final choice that best matches the situation. Output that choice in `` tags. Now, here is the question and its multiple-choice options: {question} """ async def process_question(self, video_id: str, question: str) -> Dict: """Process a single question using Gemini API""" async with self.semaphore: try: # Get frame paths frame_paths = self.get_frame_paths(video_id) if not frame_paths: raise FileNotFoundError(f"No frames found for video ID {video_id}") # Create image parts for each frame image_parts = [] for path in frame_paths: with PIL.Image.open(path) as image: # Convert image to bytes image_bytes = self.image_to_bytes(image) # Create part from bytes image_part = Part.from_data(data=image_bytes, mime_type="image/jpeg") image_parts.append(image_part) # Create prompt prompt = self.create_prompt(question) # Prepare contents list with prompt and frames contents = [prompt, *image_parts] # Make API request with temperature and top_p set to 0 response = self.model.generate_content( contents, generation_config={ "temperature": 0.0, "top_p": 0.0 } ) return { "video_id": video_id, "answer": response.text } except Exception as e: logger.error(f"Error processing video {video_id}: {str(e)}") return { "video_id": video_id, "error": str(e) } async def process_batch(self, questions: List[Dict]) -> List[Dict]: """Process multiple questions in parallel""" tasks = [] for q in questions: task = self.process_question(q['id'], q['question']) tasks.append(task) return await asyncio.gather(*tasks) async def main(): # Your Google Cloud project ID project_id = "tesla-451102" # Initialize processor processor = GeminiProcessor(project_id) # Create output directory output_dir = Path('gemini_pro_answers') output_dir.mkdir(exist_ok=True) # Read questions from CSV questions = [] with open('all_questions.csv', 'r') as csvfile: reader = csv.DictReader(csvfile) for row in reader: # Convert ID to integer for comparison id_num = int(row['id']) row['id'] = str(id_num).zfill(5) questions.append(row) # Process questions logger.info(f"Processing {len(questions)} questions...") results = await processo [truncated — 375 more characters] ``` ### fill_gaps.py ```python import os import csv import base64 from openai import OpenAI from pathlib import Path import logging from typing import List, Dict import json from typing import Optional, Set # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class GPT4VProcessor: def __init__(self, api_key: str = "", video_ids: Optional[Set[str]] = None): """ Initialize the GPT-4V processor with API credentials and optional video IDs filter Args: api_key (str): OpenAI API key video_ids (Optional[Set[str]]): Set of video IDs to process. If None, processes all allowed videos """ self.client = OpenAI(api_key=api_key) self.video_ids = video_ids def encode_image(self, image_path: str) -> str: """ Encode an image file to base64 Args: image_path (str): Path to the image file Returns: str: Base64 encoded image """ with open(image_path, "rb") as image_file: return base64.b64encode(image_file.read()).decode("utf-8") def get_frame_paths(self, video_id: str) -> List[str]: """ Get paths for all frames of a specific video Args: video_id (str): ID of the video/question Returns: List[str]: List of frame image paths """ frame_dir = Path('extracted_frames') / str(video_id) frame_paths = sorted(list(frame_dir.glob('frame_*.jpg'))) return [str(path) for path in frame_paths] def create_prompt(self, question: str, frame_count: int = 5) -> str: """ Create the prompt for GPT-4V Args: question (str): The question to answer frame_count (int): Number of frames Returns: str: Formatted prompt """ return f"""You have 5 equally spaced frames (Frame 1 through Frame 5) captured from a 5-second dashcam video, taken from the driver's forward-facing perspective. Using these frames, answer the following multiple-choice question. Incorporate any relevant details observed in the frames (for example, lanes, signage, vehicles, pedestrians, traffic signals, road markings, obstructions) that might help in selecting the correct answer. Consider how details may change across the frames and note that some frames may be more crucial than others. Steps to follow: 1. **Frame-by-Frame Analysis:** Briefly describe the significant elements you notice in each of the 5 frames (e.g., signs, road markings, obstructions, other vehicles, potential hazards). 2. **Contextual Reasoning:** Integrate the observations from each frame. Think about what is happening over time, which elements are most relevant, and how they connect to the question. 3. **Match to Answer Choices:** Relate your findings to each of the multiple-choice options. Eliminate those that are inconsistent with the visual evidence or standard traffic rules, and select the most appropriate remaining choice. 4. **Provide the Best Answer:** Conclude with the final choice that best matches the situation. Output that choice in `` tags. Now, here is the question and its multiple-choice options: {question} """ def process_question(self, video_id: str, question: str) -> Dict: """ Process a single question with its associated frames Args: video_id (str): ID of the video/question question (str): The question to answer Returns: Dict: API response """ try: # Skip if video_id is not in the specified set (if a set was provided) if self.video_ids is not None and video_id not in self.video_ids: logger.info(f"Skipping video ID: {video_id} as it is not in the specified list.") return {} frame_paths = self.get_frame_paths(video_id) if not frame_paths: raise FileNotFoundError(f"No frames found for video ID {video_id}") content = [{"type": "text", "text": self.create_prompt(question)}] for path in frame_paths: base64_image = self.encode_image(path) content.append({ "type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"} }) completion = self.client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": content}], max_tokens=4096, temperature=0, top_p=0 ) return { "answer": completion.choices[0].message.content, "finish_reason": completion.choices[0].finish_reason, } except Exception as e: logger.error(f"Error processing video ID {video_id}: {str(e)}") return {"error": str(e)} def main(): """ Main function to process questions for specified video IDs """ # Load environment variables from dotenv import load_dotenv load_dotenv() # Get API key from environment api_key = os.getenv('OPENAI_API_KEY_KOA_4o') if not api_key: raise ValueError("OPENAI_API_KEY_KOA_4o environment variable not set") # Specify the video IDs you want to process # Comment out or modify this line to process different video IDs video_ids_to_process = {"00023"} # Example: only process these IDs # Initialize processor with specific video IDs processor = GPT4VProcessor(api_key, video_ids=video_ids_to_process) # Create output directory output_dir = Path('gpt4v_results_new_prompt') output_dir.mkdir(exist_ok=True) # Process questions from CSV with open('questions.csv', 'r') as csvfile: reader = csv.DictReader(csvfile) for row in reade [truncated — 772 more characters] ``` [9 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]