File size: 9,854 Bytes
1da285f | 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 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 | import os
import base64
import requests
import openai
import random
import time
import json
import logging
from tqdm import tqdm
from bs4 import BeautifulSoup
import json_repair
DATASET_PATH = '../../evaluation/gts/det/semantics/union3_test.json'
IMG_DIR = '../../dataset/data/coco_det/images/semantics/'
OUTPUT_PATH = './output.json'
# from approach.llm.gpt_polling import API_KEY
API_KEYS = [os.environ.get("ARK_API_KEY", "[ARK_API_KEY]")]
openai.base_url = "https://ark.cn-beijing.volces.com/api/v3/"
key_idx = 0
# OpenAI API Key
openai.api_key = API_KEYS[0]
MAX_RETRY = 10
def get_steam_app_data(app_id, image_name):
# TODO:
if app_id == '1718650':
app_name = 'Nuremberg: VRdict of Nations'
app_description = 'A VR investigation of the crimes against humanity committed by the Nazi leaders. This is a detective story in VR, a documentary investigation, brought to you by Rossiya Segodnya. Your goal is to find and collect evidence to prove that the key Nazi criminals and the leaders of the Third Reich are guilty. Today, less than 50% of young people are familiar with the historic Nuremberg trial, its process and results (according to a survey conducted by the “Nuremberg: Casus Pacis” project). The charges and verdicts brought forward by the Nuremberg tribunal, its fairness or even necessity are often questioned. This happens out of ignorance or under pressure from those who benefit from distorting the historical facts. Your virtual journey begins in the Spandau Prison dining area recreated from one of the legendary post-war photos. Here, you are met by serenely dining Nuremberg defendants – Göring, Dönitz, Rosenberg, von Ribbentrop and von Schirach. By touching each of them, you can travel back to these Nazi criminals’ past. Your task is to find evidence of their terrible crimes against humanity in – seemingly – an ordinary and peaceful environment. As you gather more pieces of evidence, you put together a convincing dossier, akin to those ones that the 1946 Nuremberg verdict was based on. Thus you will restore the historical truth.'
return app_name, app_description
url = f"https://store.steampowered.com/app/{app_id}"
# proxies = {
# "http": f"http://{proxy}",
# "https": f"http://{proxy}"
# }
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36',
}
try:
response = requests.get(url, headers=headers)
response.raise_for_status()
soup = BeautifulSoup(response.content, "html.parser")
app_name = soup.find("div", {"id": "appHubAppName"}).get_text(strip=True)
short_desc = soup.find("div", class_="game_description_snippet").get_text(strip=True)
long_desc = soup.find("div", {"id": "game_area_description"}).get_text(strip=True)
return app_name, short_desc + " " + long_desc
except Exception as e:
# print(f"Error fetching app data for app_id {app_id} of image {image_name}. Error: {e}")
logging.error(f"Error fetching app data for app_id {app_id} of image {image_name}. Error: {e}")
return "", ""
# Function to encode the image
def encode_image(image_path):
with open(image_path, "rb") as image_file:
return base64.b64encode(image_file.read()).decode('utf-8')
def process_image(image_path, key_idx):
# Getting the base64 string
base64_image = encode_image(image_path)
gpt4v_prompt = '''--- Task ---
This is a screenshot of a VR game with a size of 960*540. Please identify all interactable objects on the screenshot, describe what they are, locate them with bounding box in the image and give how confident you are about the result, ranging from 0 to 1.
--- Bounding Box Format ---
[x, y, width, height], where x, y are the coordinates of the top-left corner of the box, and width, height are the width and height of the box. The unit of x, y, width, height are pixel.
x coordinate is the horizontal distance from the left edge of the image, and y coordinate is the vertical distance from the top edge of the image.
--- Output Format ---
Output in JSON format: [{"name": object_name1, "bbox": bounding_box1, "confidence": confidence1}, {"name": object_name2, "bbox": bounding_box2, "confidence": confidence2}].
DO NOT OUTPUT any other content besides JSON.'''
# payload = {
# # "model": "gpt-4-vision-preview",
# "model": "o4-mini",
# "messages": [
# {
# "role": "user",
# "content": [
# {
# "type": "text",
# "text": gpt4v_prompt
# },
# {
# "type": "image_url",
# "image_url": {
# "url": f"data:image/jpeg;base64,{base64_image}"
# }
# }
# ]
# }
# ],
# "reasoning": {"effort": "minimal"},
# }
res = ''
try:
response = ''
idx = 0
delay = 1 # Starting delay
try_cnt = 0
while (idx == 0) or ('choices' not in resp_json.keys()) and try_cnt < MAX_RETRY:
try_cnt += 1
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {openai.api_key}"
}
idx = 1
response = openai.chat.completions.create(
model="doubao-1-5-vision-pro-32k-250115",
messages=[
{
"role": "user",
"content": [
{
"type": "text",
"text": gpt4v_prompt
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}"
}
}
]
}
],
reasoning_effort="minimal",
timeout=300
)
# response = requests.post(f"{openai.api_base}/chat/completions", headers=headers, json=payload, timeout=300)
# response = requests.post(f"{openai.api_base}/chat/completions", headers=headers, json=payload, timeout=300)
key_idx = (key_idx + 1) % len(API_KEYS)
openai.api_key = API_KEYS[key_idx]
resp_json = response.model_dump()
# print(resp_json)
logging.info(f"Response: {resp_json}")
# If rate limited or need to retry, wait for an exponentially increasing delay
if 'error' in resp_json:
error_code = resp_json['error']['code']
if error_code == 429 or 'retry' in resp_json['error']['message'].lower():
# time.sleep(delay)
delay = min(delay * 2, 60) # Increase delay, max out at 60 seconds
else:
break # Break loop on other errors
else:
delay = 1 # Reset delay on successful attempt
if 'error' in resp_json:
logging.error(f"Error generating completion for image {image_path}. Error: {resp_json['error']['message']}, skipped")
return ''
res = resp_json['choices'][0]['message']['content'].strip()
if res.startswith('```'):
return json_repair.loads(res.strip('`')[4:].strip())
else:
return json_repair.loads(res)
except Exception as e:
# print(response)
# print(f"Error generating completion for image {image_path}. Error: {e}")
# print('#Res ', res)
logging.error(f"Error generating completion for image {image_path}. Error: {e}")
logging.error(f"Response: {res}")
time.sleep(5)
return ''
def main():
logging.basicConfig(format='%(asctime)s - %(levelname)s - %(name)s - %(message)s',
datefmt='%m/%d/%Y %H:%M:%S',level=logging.INFO )
start_time = time.time()
logging.info('Started at %s', time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()))
with open(DATASET_PATH, 'r') as f:
dataset = json.load(f)
if os.path.exists(OUTPUT_PATH):
with open(OUTPUT_PATH, 'r') as f:
output = json.load(f)
else:
output = {}
progress = tqdm(dataset['images'], total=len(dataset['images']), leave=True, position=0)
try:
for img in progress:
# for img in dataset['images']:
img_id = img['id']
img_name = img['file_name']
if str(img_id) in output.keys() and not type(output[str(img_id)]) == str:
# logging.info(f"Skipping {img_name}")
continue
# logging.info(f"Processing {img_name}")
img_path = os.path.join(IMG_DIR, img_name)
progress.set_description(f"Processing {img_name}")
objects = process_image(img_path, key_idx)
output[str(img_id)] = objects
with open(OUTPUT_PATH, 'w') as f:
json.dump(output, f, indent=4)
except KeyboardInterrupt:
logging.info('Keyboard Interrupted')
except Exception as e:
logging.error(f"Error: {e}")
finally:
with open(OUTPUT_PATH, 'w') as f:
json.dump(output, f, indent=4)
logging.info('Completed at %s', time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()))
logging.info(f"Total time taken: {time.time() - start_time} seconds")
if __name__ == "__main__":
# process_image('This is a screenshot of a VR game. Please describe all objects on the screenshot.', 'examples/vr_screenshot.jpg')
main() |