Spaces:
Sleeping
Sleeping
File size: 4,742 Bytes
2b811d8 0508b3e 2b811d8 efc0118 0508b3e 8630bef 0508b3e 8630bef 2b811d8 efc0118 2b811d8 efc0118 2b811d8 efc0118 82aae5c efc0118 2b811d8 efc0118 2b811d8 efc0118 2b811d8 efc0118 2b811d8 efc0118 2b811d8 0508b3e 2b811d8 efc0118 2b811d8 0508b3e 2b811d8 f0c65a3 0508b3e efc0118 2b811d8 0508b3e 2b811d8 efc0118 2b811d8 efc0118 2b811d8 8630bef 0508b3e 2b811d8 0508b3e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 |
import os
import requests
import json
from datetime import datetime
from typing import Tuple
from data_models.park_manager import ParkManager
from data_models.image_manager import ImageManager
from data_models.openai_manager import OpenAIManager
from openai_predictions import process_agent_predictions
from cloud_storage.google_cloud_storage import GoogleCloudStorage
park_manager = ParkManager()
image_manager = ImageManager()
prediction_manager = OpenAIManager()
gcs = GoogleCloudStorage("ialab-fr-7047e56bef0b.json")
def nearby_search(coordinates: tuple) -> Tuple[str, str, dict]:
"""Search nearby places
Args:
coordinates (tuple): Coordinates
Raises:
Exception: Failed to fetch places
Returns:
str: Place ID
"""
url = "https://places.googleapis.com/v1/places:searchNearby"
payload = json.dumps(
{
"includedTypes": ["park"],
"maxResultCount": 1,
"locationRestriction": {
"circle": {
"center": {"longitude": coordinates[0], "latitude": coordinates[1]},
"radius": 500,
}
},
}
)
headers = {
"Content-Type": "application/json",
"X-Goog-Api-Key": "AIzaSyAPmP_sVOZMmJvyXFNPrWJY11BzhKRRBh4",
"X-Goog-FieldMask": "places.id,places.name,places.photos",
}
response = requests.request("POST", url, headers=headers, data=payload)
if response.status_code != 200:
raise Exception(f"Failed to fetch places: {response.text}")
response = response.json()
if "places" not in response:
return None, None, None
return (
response["places"][0]["id"],
response["places"][0]["name"],
response["places"][0]["photos"],
)
def place_details(place_id: str) -> dict:
"""Get place details
Args:
place_id (str): Place ID
Raises:
Exception: Failed to fetch place details
Returns:
dict: Place details
"""
url = f"https://places.googleapis.com/v1/places/{place_id}"
payload = {}
headers = {
"Content-Type": "application/json",
"X-Goog-Api-Key": "AIzaSyAPmP_sVOZMmJvyXFNPrWJY11BzhKRRBh4",
"X-Goog-FieldMask": "photos",
}
response = requests.request("GET", url, headers=headers, data=payload)
if response.status_code != 200:
raise Exception(f"Failed to fetch place details: {response.text}")
return response.json()
def fetch_photo(place_id: str, photo_id: str, max_width: int, max_height: int) -> bytes:
"""Fetch photo from place
Args:
place_id (str): Place ID
photo_id (str): Photo ID
max_width (int): Max width
max_height (int): Max height
Raises:
Exception: Failed to fetch photo
Returns:
str: Photo binary
"""
url = f"https://places.googleapis.com/v1/places/{place_id}/photos/{photo_id}/media?key=AIzaSyAPmP_sVOZMmJvyXFNPrWJY11BzhKRRBh4&maxHeightPx={max_width}&maxWidthPx={max_height}"
payload = {}
headers = {
"Content-Type": "application/json",
"X-Goog-Api-Key": "AIzaSyAPmP_sVOZMmJvyXFNPrWJY11BzhKRRBh4",
}
response = requests.request("GET", url, headers=headers, data=payload)
if response.status_code != 200:
raise Exception(f"Failed to fetch photo: {response.text}")
return response.content
def fetch_place_photos(place_id: str, place_name: str, photos: dict) -> int:
"""Fetch all photos from place datas
Args:
place_id (str): Place ID
place_name (str): Place name
photos (dict): Photos
folder (str): Folder to save photos
Returns:
str: Place ID
"""
folder_path = f"images/{place_name}"
os.makedirs(folder_path, exist_ok=True)
if park_manager.get_park_id(place_name):
print(f"Place: {place_name} already exists")
return 0
park_id = park_manager.add_park(place_name)
for photo in photos:
photo_id = photo["name"].replace(f"places/{place_id}/photos/", "")
photo_id = photo_id.replace("/", "_")
max_width = photo["widthPx"]
max_height = photo["heightPx"]
photo_binary = fetch_photo(place_id, photo_id, max_width, max_height)
file_name = f"{place_name}/{photo_id[:150]}.jpg"
bucket_path = gcs.upload_file_from_content("suad_park", file_name, photo_binary)
image_id = image_manager.add_image(bucket_path, datetime.now(), park_id=park_id)
predictions = process_agent_predictions(file_name, photo_binary)
prediction_manager.add_predictions(image_id, predictions)
print(f"{len(photos)} photos fetched for place: {place_name}")
return len(photos)
|