wehub-resource-sync
1165 lines
50 KiB
Plaintext
1165 lines
50 KiB
Plaintext
{
|
|
"cells": [
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 1,
|
|
"metadata": {
|
|
"id": "f705f4be70e9"
|
|
},
|
|
"outputs": [],
|
|
"source": [
|
|
"# Copyright 2025 Google LLC\n",
|
|
"#\n",
|
|
"# Licensed under the Apache License, Version 2.0 (the \"License\");\n",
|
|
"# you may not use this file except in compliance with the License.\n",
|
|
"# You may obtain a copy of the License at\n",
|
|
"#\n",
|
|
"# https://www.apache.org/licenses/LICENSE-2.0\n",
|
|
"#\n",
|
|
"# Unless required by applicable law or agreed to in writing, software\n",
|
|
"# distributed under the License is distributed on an \"AS IS\" BASIS,\n",
|
|
"# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
|
|
"# See the License for the specific language governing permissions and\n",
|
|
"# limitations under the License."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {
|
|
"id": "e47ccf0ace61"
|
|
},
|
|
"source": [
|
|
"# Enhanced Vision Assistant with Gemini\n",
|
|
"\n",
|
|
"<table align=\"left\">\n",
|
|
" <td style=\"text-align: center\">\n",
|
|
" <a href=\"https://colab.research.google.com/github/GoogleCloudPlatform/generative-ai/vision/use-cases/vision-assistant/enhanced_vision_assistant.ipynb\">\n",
|
|
" <img width=\"32px\" src=\"https://www.gstatic.com/pantheon/images/bigquery/welcome_page/colab-logo.svg\" alt=\"Google Colaboratory logo\"><br> Open in Colab\n",
|
|
" </a>\n",
|
|
" </td>\n",
|
|
"\n",
|
|
" <td style=\"text-align: center\">\n",
|
|
" <a href=\"https://github.com/GoogleCloudPlatform/generative-ai/vision/use-cases/vision-assistant/enhanced_vision_assistant.ipynb\">\n",
|
|
" <img width=\"32px\" src=\"https://raw.githubusercontent.com/primer/octicons/refs/heads/main/icons/mark-github-24.svg\" alt=\"GitHub logo\"><br> View on GitHub\n",
|
|
" </a>\n",
|
|
" </td>\n",
|
|
"</table>\n",
|
|
"\n",
|
|
"<div style=\"clear: both;\"></div>\n",
|
|
"\n",
|
|
"<b>Share to:</b>\n",
|
|
"\n",
|
|
"<a href=\"https://www.linkedin.com/sharing/share-offsite/?url=https%3A//github.com/GoogleCloudPlatform/generative-ai/blob/main/gemini/use-cases/vision-assistant/enhanced_vision_assistant.ipynb\" target=\"_blank\">\n",
|
|
" <img width=\"20px\" src=\"https://upload.wikimedia.org/wikipedia/commons/8/81/LinkedIn_icon.svg\" alt=\"LinkedIn logo\">\n",
|
|
"</a>\n",
|
|
"\n",
|
|
"<a href=\"https://bsky.app/intent/compose?text=https%3A//github.com/GoogleCloudPlatform/generative-ai/blob/main/gemini/use-cases/vision-assistant/enhanced_vision_assistant.ipynb\" target=\"_blank\">\n",
|
|
" <img width=\"20px\" src=\"https://upload.wikimedia.org/wikipedia/commons/7/7a/Bluesky_Logo.svg\" alt=\"Bluesky logo\">\n",
|
|
"</a>\n",
|
|
"\n",
|
|
"<a href=\"https://twitter.com/intent/tweet?url=https%3A//github.com/GoogleCloudPlatform/generative-ai/blob/main/gemini/use-cases/vision-assistant/enhanced_vision_assistant.ipynb\" target=\"_blank\">\n",
|
|
" <img width=\"20px\" src=\"https://upload.wikimedia.org/wikipedia/commons/5/5a/X_icon_2.svg\" alt=\"X logo\">\n",
|
|
"</a>\n",
|
|
"\n",
|
|
"<a href=\"https://reddit.com/submit?url=https%3A//github.com/GoogleCloudPlatform/generative-ai/blob/main/gemini/use-cases/vision-assistant/enhanced_vision_assistant.ipynb\" target=\"_blank\">\n",
|
|
" <img width=\"20px\" src=\"https://redditinc.com/hubfs/Reddit%20Inc/Brand/Reddit_Logo.png\" alt=\"Reddit logo\">\n",
|
|
"</a>\n",
|
|
"\n",
|
|
"<a href=\"https://www.facebook.com/sharer/sharer.php?u=https%3A//github.com/GoogleCloudPlatform/generative-ai/blob/main/gemini/use-cases/vision-assistant/enhanced_vision_assistant.ipynb\" target=\"_blank\">\n",
|
|
" <img width=\"20px\" src=\"https://upload.wikimedia.org/wikipedia/commons/5/51/Facebook_f_logo_%282019%29.svg\" alt=\"Facebook logo\">\n",
|
|
"</a>"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {
|
|
"id": "5a832eb3d892"
|
|
},
|
|
"source": [
|
|
"| | |\n",
|
|
"|-|-|\n",
|
|
"|Author(s) | [Prajwal](https://github.com/iprajwaal) |"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {
|
|
"id": "79329fa09037"
|
|
},
|
|
"source": [
|
|
"### Overview\n",
|
|
"\n",
|
|
"\n",
|
|
"\n",
|
|
"The Enhanced Vision Assistant shows how Google Cloud services can be combined to create assistive technology for visually impaired users. This solution delivers real-time environmental awareness through computer vision and AI-powered guidance. Using Gen AI with Gemini alongside Cloud Vision and Text-to-Speech APIs, the system transforms visual information into helpful audio navigation cues.\n",
|
|
"\n",
|
|
"### Demo Video \n",
|
|
"\n",
|
|
"[YouTube Video](https://youtu.be/Jpili5kx3hA)\n",
|
|
"\n",
|
|
"### Objectives\n",
|
|
"\n",
|
|
"In this notebook, you will build a complete vision assistant system that:\n",
|
|
"\n",
|
|
"* Detects objects in real-time using Google Cloud Vision API\n",
|
|
"* Analyzes scenes with Gemini to identify hazards and safe paths\n",
|
|
"* Generates natural language navigation guidance prioritized by urgency\n",
|
|
"* Delivers audio instructions through Google Text-to-Speech\n",
|
|
"* Implements a full navigation pipeline with object tracking and context awareness\n",
|
|
"\n",
|
|
"This cookbook demonstrates practical AI application development for accessibility technology using Google Cloud's API ecosystem.\n",
|
|
"\n",
|
|
""
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {
|
|
"id": "862632423257"
|
|
},
|
|
"source": [
|
|
"### Features\n",
|
|
"\n",
|
|
"- **Real-time object detection** using Google Cloud Vision API\n",
|
|
"- **Intelligent scene analysis** with Google Gen AI and Gemini\n",
|
|
"- **Natural language navigation guidance** prioritized by urgency\n",
|
|
"- **Audio feedback** through Google Text-to-Speech\n",
|
|
"- **Obstacle tracking** and movement prediction\n",
|
|
"- **Dynamic priority system** for focusing on the most relevant hazards\n",
|
|
"\n",
|
|
"### How It Works\n",
|
|
"\n",
|
|
"1. The camera captures the user's environment in real-time.\n",
|
|
"2. Cloud Vision API detects objects and their positions.\n",
|
|
"3. The system estimates depth and analyzes potential hazards.\n",
|
|
"4. Gemini processes the scene and generates natural language guidance.\n",
|
|
"5. Audio instructions are delivered through text-to-speech, prioritized by urgency."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {
|
|
"id": "f0316df526f8"
|
|
},
|
|
"source": [
|
|
"## Get started"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {
|
|
"id": "ba5a5d175314"
|
|
},
|
|
"source": [
|
|
"#### Install Google Gen AI SDK for Python, and required dependencies for the project"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 2,
|
|
"metadata": {
|
|
"id": "018ac3f38d64"
|
|
},
|
|
"outputs": [],
|
|
"source": [
|
|
"%pip install --upgrade --quiet google-genai opencv-python pygame numpy scipy google-cloud-vision google-cloud-texttospeech"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {
|
|
"id": "94c93de4621b"
|
|
},
|
|
"source": [
|
|
"#### Authenticate your notebook environment (Colab only)\n",
|
|
"##### If you are running this notebook on Google Colab, run the following cell to authenticate your environment."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 3,
|
|
"metadata": {
|
|
"id": "2082bbcb771c"
|
|
},
|
|
"outputs": [],
|
|
"source": [
|
|
"import sys\n",
|
|
"\n",
|
|
"if \"google.colab\" in sys.modules:\n",
|
|
" from google.colab import auth\n",
|
|
"\n",
|
|
" auth.authenticate_user()"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {
|
|
"id": "b1e6301eae57"
|
|
},
|
|
"source": [
|
|
"### Importing the necessary libraries"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 19,
|
|
"metadata": {
|
|
"id": "788feebbbf09"
|
|
},
|
|
"outputs": [],
|
|
"source": [
|
|
"import os\n",
|
|
"import queue\n",
|
|
"import threading\n",
|
|
"import time\n",
|
|
"from dataclasses import dataclass\n",
|
|
"from enum import Enum\n",
|
|
"\n",
|
|
"import cv2\n",
|
|
"import numpy as np\n",
|
|
"import pygame\n",
|
|
"from google import genai\n",
|
|
"from google.cloud import texttospeech, vision_v1"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {
|
|
"id": "97b83b55360b"
|
|
},
|
|
"source": [
|
|
"### Set Google Cloud Project and Create Client\n",
|
|
"\n",
|
|
"Before running this notebook, you'll need a Google Cloud project with the necessary APIs enabled. If you haven't already set up your environment:\n",
|
|
"\n",
|
|
"1. [Create or select a Google Cloud project](https://console.cloud.google.com/cloud-resource-manager)\n",
|
|
"2. Enable the required APIs:\n",
|
|
" - [Vertex AI API](https://console.cloud.google.com/flows/enableapi?apiid=aiplatform.googleapis.com) \n",
|
|
" - [Cloud Vision API](https://console.cloud.google.com/flows/enableapi?apiid=vision.googleapis.com)\n",
|
|
" - [Cloud Text-to-Speech API](https://console.cloud.google.com/flows/enableapi?apiid=texttospeech.googleapis.com)\n",
|
|
"3. [Set up authentication](https://cloud.google.com/docs/authentication/getting-started)\n",
|
|
"\n",
|
|
"The code below configures your project settings and initializes the Gen AI client that will be used throughout this notebook. Simply update the project ID or use environment variables if running in a [managed environment](https://cloud.google.com/vertex-ai/docs/workbench/managed/introduction).\n",
|
|
"\n",
|
|
"For more detailed setup instructions, see [Google Cloud's documentation on setting up a development environment](https://cloud.google.com/vertex-ai/docs/start/cloud-environment)."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {
|
|
"id": "714c768d628a"
|
|
},
|
|
"source": [
|
|
"#### Set your Google Cloud project details"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 6,
|
|
"metadata": {
|
|
"id": "d37a589ae408"
|
|
},
|
|
"outputs": [],
|
|
"source": [
|
|
"# fmt: off\n",
|
|
"PROJECT_ID = \"[your-project-id]\" # @param {type: \"string\", placeholder: \"[your-project-id]\", isTemplate: true}\n",
|
|
"# fmt: on\n",
|
|
"if not PROJECT_ID or PROJECT_ID == \"[your-project-id]\":\n",
|
|
" PROJECT_ID = str(os.environ.get(\"GOOGLE_CLOUD_PROJECT\"))\n",
|
|
"\n",
|
|
"LOCATION = os.environ.get(\"GOOGLE_CLOUD_REGION\", \"us-central1\")"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {
|
|
"id": "e85a4c270774"
|
|
},
|
|
"source": [
|
|
"#### Set up credentials (you can also use application default credentials)"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 7,
|
|
"metadata": {
|
|
"id": "f2ea5b356262"
|
|
},
|
|
"outputs": [],
|
|
"source": [
|
|
"# fmt: off\n",
|
|
"CREDENTIALS_PATH = \"[path-to-your-credentials]\" # @param {type: \"string\", placeholder: \"[path-to-your-credentials]\", isTemplate: true}\n",
|
|
"# fmt: on\n",
|
|
"if CREDENTIALS_PATH and CREDENTIALS_PATH != \"[path-to-your-credentials]\":\n",
|
|
" os.environ[\"GOOGLE_APPLICATION_CREDENTIALS\"] = CREDENTIALS_PATH"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {
|
|
"id": "ec8a2b80cfc4"
|
|
},
|
|
"source": [
|
|
"#### Initialize the Gen AI client"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 8,
|
|
"metadata": {
|
|
"id": "149ebb514c54"
|
|
},
|
|
"outputs": [],
|
|
"source": [
|
|
"genai_client = genai.Client(project=PROJECT_ID, location=LOCATION, vertexai=True)"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {
|
|
"id": "b55d9449233a"
|
|
},
|
|
"source": [
|
|
"### Define Navigation and Object Detection Classes"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 9,
|
|
"metadata": {
|
|
"id": "2b4859bc7c93"
|
|
},
|
|
"outputs": [],
|
|
"source": [
|
|
"# Define priority levels for navigation\n",
|
|
"\n",
|
|
"\n",
|
|
"class NavigationPriority(Enum):\n",
|
|
" CRITICAL = 1 # Immediate collision risk\n",
|
|
" HIGH = 2 # Hazardous objects or close obstacles\n",
|
|
" MEDIUM = 3 # Notable obstacles at moderate distance\n",
|
|
" LOW = 4 # Background information\n",
|
|
" IGNORE = 5 # Not worth mentioning"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 10,
|
|
"metadata": {
|
|
"id": "99b1c793d19c"
|
|
},
|
|
"outputs": [],
|
|
"source": [
|
|
"@dataclass\n",
|
|
"class DetectedObject:\n",
|
|
" \"\"\"Class for detected objects with navigation-relevant attributes\"\"\"\n",
|
|
"\n",
|
|
" name: str\n",
|
|
" bbox: list[vision_v1.NormalizedVertex]\n",
|
|
" confidence: float\n",
|
|
" depth_estimate: float\n",
|
|
" priority: NavigationPriority\n",
|
|
" trajectory: tuple[float, float] = (0, 0) # x, y movement\n",
|
|
" last_seen: float = time.time()"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {
|
|
"id": "d343df995d83"
|
|
},
|
|
"source": [
|
|
"### Movement Tracking"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 11,
|
|
"metadata": {
|
|
"id": "8517b9dfb4fa"
|
|
},
|
|
"outputs": [],
|
|
"source": [
|
|
"class MovementTracker:\n",
|
|
" \"\"\"Tracks movement history to provide context for navigation decisions\"\"\"\n",
|
|
"\n",
|
|
" def __init__(self):\n",
|
|
" self.movement_history = []\n",
|
|
" self.last_position = None\n",
|
|
"\n",
|
|
" def update_position(self, position):\n",
|
|
" if self.last_position:\n",
|
|
" movement = {\n",
|
|
" \"from\": self.last_position,\n",
|
|
" \"to\": position,\n",
|
|
" \"timestamp\": time.time(),\n",
|
|
" }\n",
|
|
" self.movement_history.append(movement)\n",
|
|
" self.last_position = position\n",
|
|
"\n",
|
|
" def get_recent_movement(self):\n",
|
|
" # Return last 5 movements\n",
|
|
" return self.movement_history[-5:] if self.movement_history else []"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {
|
|
"id": "f838da6be93c"
|
|
},
|
|
"source": [
|
|
"### Navigation Context"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 12,
|
|
"metadata": {
|
|
"id": "51e16167d4d2"
|
|
},
|
|
"outputs": [],
|
|
"source": [
|
|
"class NavigationContext:\n",
|
|
" \"\"\"Maintains context about navigation environment and obstacles\"\"\"\n",
|
|
"\n",
|
|
" def __init__(self):\n",
|
|
" self.known_obstacles: dict[str, DetectedObject] = {}\n",
|
|
" self.danger_zones: list[tuple[float, float, float]] = [] # x, y, radius\n",
|
|
" self.safe_paths: list[tuple[float, float]] = []\n",
|
|
" self.user_movement = {\"speed\": 0, \"direction\": 0}\n",
|
|
" self.last_guidance_time = 0\n",
|
|
" self.environment_type = \"unknown\"\n",
|
|
"\n",
|
|
" def update(self, new_objects: list[DetectedObject]):\n",
|
|
" current_time = time.time()\n",
|
|
" # Update known obstacles with new information\n",
|
|
" for obj in new_objects:\n",
|
|
" if obj.name in self.known_obstacles:\n",
|
|
" old_obj = self.known_obstacles[obj.name]\n",
|
|
" # Calculate object movement\n",
|
|
" old_center = self.get_object_center(old_obj.bbox)\n",
|
|
" new_center = self.get_object_center(obj.bbox)\n",
|
|
" trajectory = (\n",
|
|
" new_center[0] - old_center[0],\n",
|
|
" new_center[1] - old_center[1],\n",
|
|
" )\n",
|
|
" obj.trajectory = trajectory\n",
|
|
" self.known_obstacles[obj.name] = obj\n",
|
|
"\n",
|
|
" # Clean up old obstacles\n",
|
|
" self.known_obstacles = {\n",
|
|
" k: v\n",
|
|
" for k, v in self.known_obstacles.items()\n",
|
|
" if current_time - v.last_seen < 10\n",
|
|
" } # 10 second timeout\n",
|
|
"\n",
|
|
" # Update danger zones based on current obstacles\n",
|
|
" self.update_danger_zones()\n",
|
|
"\n",
|
|
" def get_object_center(self, bbox):\n",
|
|
" return ((bbox[0].x + bbox[2].x) / 2, (bbox[0].y + bbox[2].y) / 2)\n",
|
|
"\n",
|
|
" def update_danger_zones(self):\n",
|
|
" self.danger_zones = []\n",
|
|
" for obj in self.known_obstacles.values():\n",
|
|
" center = self.get_object_center(obj.bbox)\n",
|
|
" # Calculate danger zone radius based on object size and type\n",
|
|
" size = (obj.bbox[2].x - obj.bbox[0].x) * (obj.bbox[2].y - obj.bbox[0].y)\n",
|
|
" radius = size * (\n",
|
|
" 1.5 if obj.priority == NavigationPriority.CRITICAL else 1.0\n",
|
|
" )\n",
|
|
" self.danger_zones.append((center[0], center[1], radius))"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {
|
|
"id": "118df7f082e5"
|
|
},
|
|
"source": [
|
|
"### Agent Mind - Scene Analysis Logic"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 13,
|
|
"metadata": {
|
|
"id": "b96ce7de52d7"
|
|
},
|
|
"outputs": [],
|
|
"source": [
|
|
"class AgentMind:\n",
|
|
" \"\"\"Core scene analysis and decision-making engine\"\"\"\n",
|
|
"\n",
|
|
" def __init__(self):\n",
|
|
" self.context = NavigationContext()\n",
|
|
" self.hazard_weights = {\n",
|
|
" \"stairs\": 1.0,\n",
|
|
" \"hole\": 1.0,\n",
|
|
" \"edge\": 1.0,\n",
|
|
" \"glass\": 0.9,\n",
|
|
" \"knife\": 0.9,\n",
|
|
" \"fire\": 1.0,\n",
|
|
" \"chair\": 0.7,\n",
|
|
" \"table\": 0.7,\n",
|
|
" \"person\": 0.6,\n",
|
|
" \"wall\": 0.8,\n",
|
|
" \"door\": 0.6,\n",
|
|
" \"furniture\": 0.7,\n",
|
|
" }\n",
|
|
"\n",
|
|
" def analyze_scene(self, objects: list[DetectedObject]) -> dict:\n",
|
|
" \"\"\"Analyze the scene and make intelligent decisions about navigation\"\"\"\n",
|
|
" analysis = {\n",
|
|
" \"immediate_threats\": [],\n",
|
|
" \"potential_hazards\": [],\n",
|
|
" \"safe_paths\": [],\n",
|
|
" \"guidance_priority\": NavigationPriority.LOW,\n",
|
|
" \"recommended_action\": None,\n",
|
|
" }\n",
|
|
"\n",
|
|
" # Update navigation context\n",
|
|
" self.context.update(objects)\n",
|
|
"\n",
|
|
" # Identify immediate threats\n",
|
|
" for obj in objects:\n",
|
|
" if self.is_immediate_threat(obj):\n",
|
|
" analysis[\"immediate_threats\"].append(obj)\n",
|
|
" analysis[\"guidance_priority\"] = NavigationPriority.CRITICAL\n",
|
|
"\n",
|
|
" # Find safe paths\n",
|
|
" safe_paths = self.identify_safe_paths(objects)\n",
|
|
" if safe_paths:\n",
|
|
" analysis[\"safe_paths\"] = safe_paths\n",
|
|
"\n",
|
|
" # Determine recommended action\n",
|
|
" analysis[\"recommended_action\"] = self.determine_best_action(analysis)\n",
|
|
"\n",
|
|
" return analysis\n",
|
|
"\n",
|
|
" def is_immediate_threat(self, obj: DetectedObject) -> bool:\n",
|
|
" \"\"\"Determine if an object poses an immediate threat\"\"\"\n",
|
|
" bbox = obj.bbox\n",
|
|
" height = bbox[2].y - bbox[0].y\n",
|
|
" width = bbox[2].x - bbox[0].x\n",
|
|
" center_x = (bbox[0].x + bbox[2].x) / 2\n",
|
|
"\n",
|
|
" conditions = [\n",
|
|
" height > 0.4 and 0.3 < center_x < 0.7, # Large object directly ahead\n",
|
|
" obj.name.lower() in self.hazard_weights\n",
|
|
" and self.hazard_weights[obj.name.lower()] > 0.8,\n",
|
|
" obj.depth_estimate < 1.5 and width > 0.3, # Very close wide object\n",
|
|
" any(\n",
|
|
" self.is_moving_towards_user(o) for o in [obj]\n",
|
|
" ), # Object moving towards user\n",
|
|
" ]\n",
|
|
"\n",
|
|
" return any(conditions)\n",
|
|
"\n",
|
|
" def is_moving_towards_user(self, obj: DetectedObject) -> bool:\n",
|
|
" \"\"\"Check if object is moving towards the user\"\"\"\n",
|
|
" if not obj.trajectory:\n",
|
|
" return False\n",
|
|
" return obj.trajectory[1] > 0.1 # Threshold for significant movement\n",
|
|
"\n",
|
|
" def identify_safe_paths(self, objects: list[DetectedObject]) -> list[str]:\n",
|
|
" \"\"\"Identify safe navigation paths\"\"\"\n",
|
|
" safe_paths = []\n",
|
|
" sectors = {\n",
|
|
" \"left\": {\"clear\": True, \"score\": 0},\n",
|
|
" \"center\": {\"clear\": True, \"score\": 0},\n",
|
|
" \"right\": {\"clear\": True, \"score\": 0},\n",
|
|
" }\n",
|
|
"\n",
|
|
" for obj in objects:\n",
|
|
" center_x = (obj.bbox[0].x + obj.bbox[2].x) / 2\n",
|
|
" if center_x < 0.33:\n",
|
|
" sectors[\"left\"][\"clear\"] = False\n",
|
|
" sectors[\"left\"][\"score\"] += self.calculate_obstacle_score(obj)\n",
|
|
" elif center_x < 0.66:\n",
|
|
" sectors[\"center\"][\"clear\"] = False\n",
|
|
" sectors[\"center\"][\"score\"] += self.calculate_obstacle_score(obj)\n",
|
|
" else:\n",
|
|
" sectors[\"right\"][\"clear\"] = False\n",
|
|
" sectors[\"right\"][\"score\"] += self.calculate_obstacle_score(obj)\n",
|
|
"\n",
|
|
" for sector, data in sectors.items():\n",
|
|
" if data[\"clear\"] or data[\"score\"] < 0.5:\n",
|
|
" safe_paths.append(sector)\n",
|
|
"\n",
|
|
" return safe_paths\n",
|
|
"\n",
|
|
" def calculate_obstacle_score(self, obj: DetectedObject) -> float:\n",
|
|
" \"\"\"Calculate how much an obstacle blocks a path\"\"\"\n",
|
|
" size = (obj.bbox[2].y - obj.bbox[0].y) * (obj.bbox[2].x - obj.bbox[0].x)\n",
|
|
" hazard_weight = self.hazard_weights.get(obj.name.lower(), 0.5)\n",
|
|
" depth_factor = 1 / max(obj.depth_estimate, 0.1)\n",
|
|
" return size * hazard_weight * depth_factor\n",
|
|
"\n",
|
|
" def determine_best_action(self, analysis: dict) -> str:\n",
|
|
" \"\"\"Determine the best action based on scene analysis\"\"\"\n",
|
|
" if analysis[\"immediate_threats\"]:\n",
|
|
" threat = analysis[\"immediate_threats\"][0]\n",
|
|
" center_x = (threat.bbox[0].x + threat.bbox[2].x) / 2\n",
|
|
" return \"Stop and step right\" if center_x < 0.5 else \"Stop and step left\"\n",
|
|
"\n",
|
|
" if analysis[\"safe_paths\"]:\n",
|
|
" if \"center\" in analysis[\"safe_paths\"]:\n",
|
|
" return \"Proceed straight ahead carefully\"\n",
|
|
" if \"left\" in analysis[\"safe_paths\"]:\n",
|
|
" return \"Turn slightly left and proceed\"\n",
|
|
" if \"right\" in analysis[\"safe_paths\"]:\n",
|
|
" return \"Turn slightly right and proceed\"\n",
|
|
"\n",
|
|
" return \"Stop and wait for assistance\""
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {
|
|
"id": "4b601d9f3bad"
|
|
},
|
|
"source": [
|
|
"### Enhanced Vision Assistant - Main Application"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {
|
|
"id": "b739686ae411"
|
|
},
|
|
"outputs": [],
|
|
"source": [
|
|
"class EnhancedVisionAssistant:\n",
|
|
" \"\"\"Main application class combining vision, AI, and audio guidance\"\"\"\n",
|
|
"\n",
|
|
" def __init__(self, project_id, location, credentials_path=None):\n",
|
|
" # Initialize pygame mixer for audio\n",
|
|
" pygame.mixer.init(buffer=512)\n",
|
|
"\n",
|
|
" # Configuration\n",
|
|
" self.PROJECT_ID = project_id\n",
|
|
" self.REGION = location\n",
|
|
" self.CREDENTIALS_PATH = credentials_path\n",
|
|
"\n",
|
|
" # Initialize state variables\n",
|
|
" self.running = False\n",
|
|
" self.is_speaking = False\n",
|
|
" self.cap = None\n",
|
|
" self.audio_thread = None\n",
|
|
" self.previous_guidance = None\n",
|
|
" self.movement_tracker = MovementTracker()\n",
|
|
"\n",
|
|
" # Initialize queues\n",
|
|
" self.audio_queue = queue.PriorityQueue()\n",
|
|
"\n",
|
|
" # Set credentials if provided\n",
|
|
" if self.CREDENTIALS_PATH:\n",
|
|
" os.environ[\"GOOGLE_APPLICATION_CREDENTIALS\"] = self.CREDENTIALS_PATH\n",
|
|
"\n",
|
|
" # Initialize Google clients\n",
|
|
" self.vision_client = vision_v1.ImageAnnotatorClient()\n",
|
|
" self.speech_client = texttospeech.TextToSpeechClient()\n",
|
|
"\n",
|
|
" # Initialize Gen AI client\n",
|
|
" self.genai_client = genai.Client(\n",
|
|
" project=self.PROJECT_ID, location=self.REGION, vertexai=True\n",
|
|
" )\n",
|
|
"\n",
|
|
" # Initialize the agent mind\n",
|
|
" self.agent = AgentMind()\n",
|
|
"\n",
|
|
" # Configuration parameters\n",
|
|
" self.DETECTION_INTERVAL = 3 # seconds\n",
|
|
" self.MIN_OBJECT_SIZE = 0.1\n",
|
|
"\n",
|
|
" print(\"Enhanced Vision Assistant initialized with Gemini model\")\n",
|
|
"\n",
|
|
" def detect_objects_with_depth(self, frame):\n",
|
|
" \"\"\"Detect objects and estimate their depth\"\"\"\n",
|
|
" try:\n",
|
|
" success, buffer = cv2.imencode(\".jpg\", frame)\n",
|
|
" content = buffer.tobytes()\n",
|
|
"\n",
|
|
" image = vision_v1.Image(content=content)\n",
|
|
" features = [\n",
|
|
" vision_v1.Feature(type=vision_v1.Feature.Type.OBJECT_LOCALIZATION),\n",
|
|
" vision_v1.Feature(type=vision_v1.Feature.Type.LABEL_DETECTION),\n",
|
|
" ]\n",
|
|
"\n",
|
|
" request = vision_v1.AnnotateImageRequest(image=image, features=features)\n",
|
|
" response = self.vision_client.annotate_image(request=request)\n",
|
|
"\n",
|
|
" detected_objects = []\n",
|
|
" for obj in response.localized_object_annotations:\n",
|
|
" bbox = obj.bounding_poly.normalized_vertices\n",
|
|
" height = bbox[2].y - bbox[0].y\n",
|
|
" width = bbox[2].x - bbox[0].x\n",
|
|
"\n",
|
|
" if height * width > self.MIN_OBJECT_SIZE:\n",
|
|
" depth_estimate = 1 / (height * width)\n",
|
|
" priority = self.calculate_priority(obj, bbox, depth_estimate)\n",
|
|
"\n",
|
|
" detected_obj = DetectedObject(\n",
|
|
" name=obj.name,\n",
|
|
" bbox=bbox,\n",
|
|
" confidence=obj.score,\n",
|
|
" depth_estimate=depth_estimate,\n",
|
|
" priority=priority,\n",
|
|
" )\n",
|
|
" detected_objects.append(detected_obj)\n",
|
|
"\n",
|
|
" return detected_objects\n",
|
|
"\n",
|
|
" except Exception as e:\n",
|
|
" print(f\"Object detection error: {e}\")\n",
|
|
" return []\n",
|
|
"\n",
|
|
" def calculate_priority(self, obj, bbox, depth_estimate):\n",
|
|
" \"\"\"Calculate priority based on object properties\"\"\"\n",
|
|
" if self.agent.is_immediate_threat(\n",
|
|
" DetectedObject(\n",
|
|
" name=obj.name,\n",
|
|
" bbox=bbox,\n",
|
|
" confidence=obj.score,\n",
|
|
" depth_estimate=depth_estimate,\n",
|
|
" )\n",
|
|
" ):\n",
|
|
" return NavigationPriority.CRITICAL\n",
|
|
"\n",
|
|
" height = bbox[2].y - bbox[0].y\n",
|
|
" center_x = (bbox[0].x + bbox[2].x) / 2\n",
|
|
"\n",
|
|
" if height > 0.4 or obj.name.lower() in self.agent.hazard_weights:\n",
|
|
" return NavigationPriority.HIGH\n",
|
|
" if 0.3 < center_x < 0.7 and height > 0.2:\n",
|
|
" return NavigationPriority.MEDIUM\n",
|
|
" if height > 0.1:\n",
|
|
" return NavigationPriority.LOW\n",
|
|
" return NavigationPriority.IGNORE\n",
|
|
"\n",
|
|
" def generate_smart_guidance(self, objects):\n",
|
|
" \"\"\"Generate intelligent guidance using scene analysis\"\"\"\n",
|
|
" try:\n",
|
|
" if not objects:\n",
|
|
" return None\n",
|
|
"\n",
|
|
" # Generate scene description\n",
|
|
" scene_description = []\n",
|
|
" for obj in objects:\n",
|
|
" description = self.generate_enhanced_object_description(obj, obj.bbox)\n",
|
|
" scene_description.append(description)\n",
|
|
"\n",
|
|
" # Update context\n",
|
|
" context = {\n",
|
|
" \"previous_guidance\": self.previous_guidance,\n",
|
|
" \"movement_history\": self.movement_tracker.get_recent_movement(),\n",
|
|
" \"environment_type\": \"indoor\", # This could be detected\n",
|
|
" \"current_speed\": \"normal\",\n",
|
|
" \"recent_obstacles\": [obj.name for obj in objects],\n",
|
|
" \"safe_zones\": [],\n",
|
|
" \"last_guidance_time\": time.time(),\n",
|
|
" }\n",
|
|
"\n",
|
|
" # Generate and send prompt to Gen AI model\n",
|
|
" prompt = self.generate_agent_prompt(scene_description, context)\n",
|
|
"\n",
|
|
" # Use the client's generate_content method directly\n",
|
|
" response = self.genai_client.generate_content(\n",
|
|
" model=\"gemini-2.5-pro\", contents=prompt\n",
|
|
" )\n",
|
|
"\n",
|
|
" # Store guidance for future context\n",
|
|
" self.previous_guidance = response.text\n",
|
|
" return self.previous_guidance\n",
|
|
"\n",
|
|
" except Exception as e:\n",
|
|
" print(f\"Guidance generation error: {e}\")\n",
|
|
" return self.generate_fallback_guidance(objects)\n",
|
|
"\n",
|
|
" def generate_enhanced_object_description(self, obj, bbox):\n",
|
|
" \"\"\"Generate detailed object description\"\"\"\n",
|
|
" center_x = (bbox[0].x + bbox[2].x) / 2\n",
|
|
" position = self.calculate_relative_position(center_x)\n",
|
|
" distance = self.estimate_distance(bbox[2].y - bbox[0].y, bbox[2].x - bbox[0].x)\n",
|
|
"\n",
|
|
" description = f\"{obj.name} {position} at {distance}\"\n",
|
|
" return description\n",
|
|
"\n",
|
|
" def calculate_relative_position(self, center_x):\n",
|
|
" \"\"\"Calculate relative position of object\"\"\"\n",
|
|
" if center_x < 0.2:\n",
|
|
" return \"far left\"\n",
|
|
" if center_x < 0.4:\n",
|
|
" return \"to your left\"\n",
|
|
" if center_x < 0.6:\n",
|
|
" return \"directly ahead\"\n",
|
|
" if center_x < 0.8:\n",
|
|
" return \"to your right\"\n",
|
|
" return \"far right\"\n",
|
|
"\n",
|
|
" def estimate_distance(self, height, width):\n",
|
|
" \"\"\"Estimate distance using object dimensions\"\"\"\n",
|
|
" # Note: These thresholds should be calibrated for your specific camera\n",
|
|
" # and environment setup. Consider implementing a more sophisticated\n",
|
|
" # depth estimation technique for better accuracy.\n",
|
|
" area = height * width\n",
|
|
"\n",
|
|
" # Using the camera-specific calibrated thresholds from configuration\n",
|
|
" if area > self.config.DISTANCE_THRESHOLD_VERY_CLOSE:\n",
|
|
" return \"very close\"\n",
|
|
" if area > self.config.DISTANCE_THRESHOLD_CLOSE:\n",
|
|
" return \"close\"\n",
|
|
" if area > self.config.DISTANCE_THRESHOLD_MODERATE:\n",
|
|
" return \"moderate distance\"\n",
|
|
" return \"far ahead\"\n",
|
|
"\n",
|
|
" def generate_agent_prompt(self, scene_description, context):\n",
|
|
" \"\"\"Generate a sophisticated prompt for the AI agent\"\"\"\n",
|
|
" prompt = f\"\"\"You are an intelligent navigation assistant for a visually impaired person. Your role is to be their eyes and ensure their safety while helping them navigate their environment. Think carefully through each step before providing guidance.\n",
|
|
"\n",
|
|
" Current Scene Analysis:\n",
|
|
" 1. Detected Objects: {\", \".join(scene_description)}\n",
|
|
" 2. Previous Context: {context.get(\"previous_guidance\", \"No previous guidance\")}\n",
|
|
" 3. Movement History: {context.get(\"movement_history\", \"Starting navigation\")}\n",
|
|
" 4. Environment Type: {context.get(\"environment_type\", \"Unknown\")}\n",
|
|
"\n",
|
|
" Step-by-step Thinking Process:\n",
|
|
" 1. First, analyze immediate safety threats:\n",
|
|
" - Are there any obstacles in imminent collision path?\n",
|
|
" - Are there any hazardous objects nearby?\n",
|
|
" - Is there any moving object approaching?\n",
|
|
"\n",
|
|
" 2. Then, evaluate navigation options:\n",
|
|
" - Which paths are completely clear?\n",
|
|
" - What is the safest direction to move?\n",
|
|
" - Are there any stable objects that could serve as landmarks?\n",
|
|
"\n",
|
|
" 3. Consider environmental context:\n",
|
|
" - Is this an indoor or outdoor space?\n",
|
|
" - Are there any recognizable features for orientation?\n",
|
|
" - Are there any potential changes in elevation (steps, curbs)?\n",
|
|
"\n",
|
|
" 4. Think about user comfort:\n",
|
|
" - How can they move most confidently?\n",
|
|
" - What landmarks can help them maintain orientation?\n",
|
|
" - How urgent is the guidance needed?\n",
|
|
"\n",
|
|
" Based on this analysis, provide:\n",
|
|
" 1. A primary safety alert if needed (most urgent threats)\n",
|
|
" 2. Clear directional guidance (where to move)\n",
|
|
" 3. Contextual information (what to expect in that direction)\n",
|
|
"\n",
|
|
" Requirements for your response:\n",
|
|
" - Use natural, conversational language suitable for text-to-speech\n",
|
|
" - Be concise but thorough (maximum 3 sentences)\n",
|
|
" - Start with any urgent warnings\n",
|
|
" - Use clear spatial references (left, right, ahead, behind)\n",
|
|
" - Mention distances in practical terms (very close, nearby, ahead)\n",
|
|
" - If relevant, include time-sensitive information (approaching object, changing conditions)\n",
|
|
"\n",
|
|
" Additional Context:\n",
|
|
" - Person's current speed: {context.get(\"current_speed\", \"unknown\")}\n",
|
|
" - Recent obstacles: {context.get(\"recent_obstacles\", [])}\n",
|
|
" - Known safe zones: {context.get(\"safe_zones\", [])}\n",
|
|
" - Last guidance timestamp: {context.get(\"last_guidance_time\", \"initial\")}\n",
|
|
"\n",
|
|
" Give your response in a clear, calming voice that prioritizes safety while maintaining user confidence.\n",
|
|
" \"\"\"\n",
|
|
" return prompt\n",
|
|
"\n",
|
|
" def generate_fallback_guidance(self, objects):\n",
|
|
" \"\"\"Generate basic guidance when AI model fails\"\"\"\n",
|
|
" if not objects:\n",
|
|
" return \"Path appears clear, proceed with caution.\"\n",
|
|
"\n",
|
|
" warnings = []\n",
|
|
" for obj in objects:\n",
|
|
" description = self.generate_enhanced_object_description(obj, obj.bbox)\n",
|
|
" warnings.append(description)\n",
|
|
"\n",
|
|
" return \". \".join(warnings) + \". Proceed with caution.\"\n",
|
|
"\n",
|
|
" def process_audio_queue(self):\n",
|
|
" \"\"\"Process audio queue with priority handling\"\"\"\n",
|
|
" while self.running:\n",
|
|
" try:\n",
|
|
" if not self.is_speaking and not self.audio_queue.empty():\n",
|
|
" priority, text = self.audio_queue.get(timeout=1)\n",
|
|
" if text: # Only speak if there's actual guidance\n",
|
|
" self.speak(text, priority)\n",
|
|
" time.sleep(0.1)\n",
|
|
" except queue.Empty:\n",
|
|
" continue\n",
|
|
" except Exception as e:\n",
|
|
" print(f\"Audio processing error: {e}\")\n",
|
|
"\n",
|
|
" def speak(self, text, priority=5):\n",
|
|
" \"\"\"Enhanced text-to-speech with priority handling\"\"\"\n",
|
|
" try:\n",
|
|
" self.is_speaking = True\n",
|
|
"\n",
|
|
" # Configure voice based on priority\n",
|
|
" speaking_rate = 1.4 if priority <= 2 else 1.2\n",
|
|
"\n",
|
|
" input_text = texttospeech.SynthesisInput(text=text)\n",
|
|
" voice = texttospeech.VoiceSelectionParams(\n",
|
|
" language_code=\"en-US\",\n",
|
|
" name=\"en-US-Standard-F\",\n",
|
|
" ssml_gender=texttospeech.SsmlVoiceGender.FEMALE,\n",
|
|
" )\n",
|
|
" audio_config = texttospeech.AudioConfig(\n",
|
|
" audio_encoding=texttospeech.AudioEncoding.MP3,\n",
|
|
" speaking_rate=speaking_rate,\n",
|
|
" pitch=1 if priority > 2 else 1.2,\n",
|
|
" )\n",
|
|
"\n",
|
|
" response = self.speech_client.synthesize_speech(\n",
|
|
" input=input_text, voice=voice, audio_config=audio_config\n",
|
|
" )\n",
|
|
"\n",
|
|
" # Use a temporary file with priority in name\n",
|
|
" temp_file = f\"temp_audio_p{priority}_{time.time()}.mp3\"\n",
|
|
" with open(temp_file, \"wb\") as out:\n",
|
|
" out.write(response.audio_content)\n",
|
|
"\n",
|
|
" pygame.mixer.music.load(temp_file)\n",
|
|
" pygame.mixer.music.play()\n",
|
|
"\n",
|
|
" while pygame.mixer.music.get_busy() and self.running:\n",
|
|
" time.sleep(0.1)\n",
|
|
"\n",
|
|
" if os.path.exists(temp_file):\n",
|
|
" os.remove(temp_file)\n",
|
|
"\n",
|
|
" except Exception as e:\n",
|
|
" print(f\"Speech synthesis error: {e}\")\n",
|
|
" finally:\n",
|
|
" self.is_speaking = False\n",
|
|
"\n",
|
|
" def display_debug_frame(self, frame, objects):\n",
|
|
" \"\"\"Display frame with object annotations for debugging\"\"\"\n",
|
|
" if objects:\n",
|
|
" for obj in objects:\n",
|
|
" bbox = obj.bbox\n",
|
|
" h, w = frame.shape[:2]\n",
|
|
" pts = np.array([[int(v.x * w), int(v.y * h)] for v in bbox], np.int32)\n",
|
|
"\n",
|
|
" # Color based on priority\n",
|
|
" color = (0, 255, 0) # Default color\n",
|
|
" if obj.priority == NavigationPriority.CRITICAL:\n",
|
|
" color = (0, 0, 255) # Red for critical\n",
|
|
" elif obj.priority == NavigationPriority.HIGH:\n",
|
|
" color = (0, 165, 255) # Orange for high priority\n",
|
|
"\n",
|
|
" cv2.polylines(frame, [pts], True, color, 2)\n",
|
|
" cv2.putText(\n",
|
|
" frame,\n",
|
|
" f\"{obj.name} (P:{obj.priority.value})\",\n",
|
|
" (int(bbox[0].x * w), int(bbox[0].y * h) - 10),\n",
|
|
" cv2.FONT_HERSHEY_SIMPLEX,\n",
|
|
" 0.5,\n",
|
|
" color,\n",
|
|
" 2,\n",
|
|
" )\n",
|
|
"\n",
|
|
" cv2.imshow(\"Enhanced Vision Assistant View\", frame)\n",
|
|
"\n",
|
|
" def start(self):\n",
|
|
" \"\"\"Start the enhanced vision assistant\"\"\"\n",
|
|
" try:\n",
|
|
" print(\"Starting Enhanced Vision Assistant...\")\n",
|
|
" self.running = True\n",
|
|
"\n",
|
|
" # Start audio processing thread\n",
|
|
" self.audio_thread = threading.Thread(target=self.process_audio_queue)\n",
|
|
" self.audio_thread.daemon = True\n",
|
|
" self.audio_thread.start()\n",
|
|
"\n",
|
|
" # Initialize camera with proper error handling\n",
|
|
" try:\n",
|
|
" self.cap = cv2.VideoCapture(0)\n",
|
|
" if not self.cap.isOpened():\n",
|
|
" raise Exception(\"Could not open camera\")\n",
|
|
" except Exception as e:\n",
|
|
" print(f\"Error opening camera: {e}\")\n",
|
|
" self.running = False # Stop the assistant if camera fails\n",
|
|
" return # Exit the method early\n",
|
|
"\n",
|
|
" last_detection_time = 0\n",
|
|
"\n",
|
|
" print(\"Press 'q' to quit\")\n",
|
|
"\n",
|
|
" while self.running:\n",
|
|
" try:\n",
|
|
" ret, frame = self.cap.read()\n",
|
|
" if not ret:\n",
|
|
" break\n",
|
|
"\n",
|
|
" current_time = time.time()\n",
|
|
"\n",
|
|
" # Perform object detection at intervals\n",
|
|
" if current_time - last_detection_time >= self.DETECTION_INTERVAL:\n",
|
|
" objects = self.detect_objects_with_depth(frame)\n",
|
|
"\n",
|
|
" if objects:\n",
|
|
" guidance = self.generate_smart_guidance(objects)\n",
|
|
" if guidance: # Only queue if there's meaningful guidance\n",
|
|
" # Determine priority based on scene analysis\n",
|
|
" analysis = self.agent.analyze_scene(objects)\n",
|
|
" priority = analysis[\"guidance_priority\"].value\n",
|
|
" self.audio_queue.put((priority, guidance))\n",
|
|
"\n",
|
|
" last_detection_time = current_time\n",
|
|
"\n",
|
|
" # Display frame with annotations (debug view)\n",
|
|
" self.display_debug_frame(frame, objects)\n",
|
|
"\n",
|
|
" if cv2.waitKey(1) & 0xFF == ord(\"q\"):\n",
|
|
" break\n",
|
|
" except Exception as e:\n",
|
|
" print(f\"Error in frame processing: {e}\")\n",
|
|
" break # Exit loop on error\n",
|
|
"\n",
|
|
" except Exception as e:\n",
|
|
" print(f\"Error in main loop: {e}\")\n",
|
|
" finally:\n",
|
|
" self.stop()\n",
|
|
"\n",
|
|
" def stop(self):\n",
|
|
" \"\"\"Stop the assistant and cleanup\"\"\"\n",
|
|
" print(\"Stopping Enhanced Vision Assistant...\")\n",
|
|
" self.running = False\n",
|
|
"\n",
|
|
" # Clear audio queue\n",
|
|
" while not self.audio_queue.empty():\n",
|
|
" try:\n",
|
|
" self.audio_queue.get_nowait()\n",
|
|
" except queue.Empty:\n",
|
|
" break\n",
|
|
"\n",
|
|
" # Cleanup resources\n",
|
|
" pygame.mixer.music.stop()\n",
|
|
" pygame.mixer.quit()\n",
|
|
"\n",
|
|
" if self.cap is not None:\n",
|
|
" self.cap.release()\n",
|
|
"\n",
|
|
" cv2.destroyAllWindows()\n",
|
|
"\n",
|
|
" if self.audio_thread is not None and self.audio_thread.is_alive():\n",
|
|
" self.audio_thread.join(timeout=2)\n",
|
|
"\n",
|
|
" # Clean up temporary files\n",
|
|
" for file in os.listdir():\n",
|
|
" if file.startswith(\"temp_audio_\") and file.endswith(\".mp3\"):\n",
|
|
" try:\n",
|
|
" os.remove(file)\n",
|
|
" except:\n",
|
|
" pass\n",
|
|
"\n",
|
|
" print(\"Enhanced Vision Assistant stopped.\")"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {
|
|
"id": "a75803098d6f"
|
|
},
|
|
"source": [
|
|
"### Run the Enhanced Vision Assistant\n",
|
|
"\n",
|
|
"The following code will initialize and run the Enhanced Vision Assistant. \n",
|
|
"\n",
|
|
"> **Note**: You will need a camera connected to your system and appropriate permissions to access it.\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 30,
|
|
"metadata": {
|
|
"id": "c2a458836f49"
|
|
},
|
|
"outputs": [],
|
|
"source": [
|
|
"def main():\n",
|
|
" # Initialize the assistant with your project details\n",
|
|
" assistant = EnhancedVisionAssistant(\n",
|
|
" project_id=PROJECT_ID, location=LOCATION, credentials_path=CREDENTIALS_PATH\n",
|
|
" )\n",
|
|
"\n",
|
|
" try:\n",
|
|
" # Start the assistant\n",
|
|
" assistant.start()\n",
|
|
" except KeyboardInterrupt:\n",
|
|
" print(\"\\nKeyboard interrupt received\")\n",
|
|
" finally:\n",
|
|
" # Ensure proper cleanup\n",
|
|
" assistant.stop()"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {
|
|
"id": "4c8d35b1a06b"
|
|
},
|
|
"source": [
|
|
"### Execute the Application\n",
|
|
"\n",
|
|
"Run the following cell to start the Enhanced Vision Assistant. Press 'q' to quit the application."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 32,
|
|
"metadata": {
|
|
"id": "afd7d74043d1"
|
|
},
|
|
"outputs": [],
|
|
"source": [
|
|
"if __name__ == \"__main__\":\n",
|
|
" main()"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {
|
|
"id": "271834603285"
|
|
},
|
|
"source": [
|
|
"## Conclusion\n",
|
|
"\n",
|
|
"Congratulations! You've successfully built a complete Enhanced Vision Assistant using Google Cloud's AI capabilities!\n",
|
|
"\n",
|
|
"This notebook demonstrated how to build an enhanced vision assistant that uses Google Cloud APIs:\n",
|
|
"* Google Cloud Vision API for object detection\n",
|
|
"* Google Gen AI SDK with Gemini for intelligent scene guidance\n",
|
|
"* Google Cloud Text-to-Speech for audio feedback\n",
|
|
"\n",
|
|
"The system provides real-time navigation assistance for visually impaired users by:\n",
|
|
"1. Detecting objects in the environment\n",
|
|
"2. Analyzing their position, size, and potential danger\n",
|
|
"3. Using Gemini to generate natural language guidance\n",
|
|
"4. Providing audio feedback in a prioritized manner\n",
|
|
"\n",
|
|
"This approach showcases how multiple Google Cloud services can be combined to create assistive technology applications with generative AI at their core.\n",
|
|
"\n",
|
|
"To explore more applications of Google's generative AI technologies, visit the [Google AI documentation](https://ai.google.dev/) and [Google Cloud Vertex AI resources](https://cloud.google.com/vertex-ai)."
|
|
]
|
|
}
|
|
],
|
|
"metadata": {
|
|
"colab": {
|
|
"name": "enhanced_vision_assistant.ipynb",
|
|
"toc_visible": true
|
|
},
|
|
"kernelspec": {
|
|
"display_name": "Python 3",
|
|
"name": "python3"
|
|
}
|
|
},
|
|
"nbformat": 4,
|
|
"nbformat_minor": 0
|
|
}
|