import os
import random
import json
import uuid
import time
from datetime import datetime, timedelta, timezone
from typing import Dict, Any, List, Optional
import spaces
import requests
from dotenv import load_dotenv
import gradio as gr
from gradio.components import LoginButton
import data_manager
from huggingface_hub import HfApi, hf_hub_download, whoami
from transformers import Mistral3ForConditionalGeneration, AutoTokenizer, TextIteratorStreamer
import threading
import torch
load_dotenv()
APP_SECRET = os.urandom(24)
ZONES_FILE = 'zones.json'
zones = {
"easy": [],
"medium": [],
"hard": []
}
user_sessions: Dict[str, Dict[str, Any]] = {}
DEFAULT_USERNAME = "player"
def save_zones_to_file():
with open(ZONES_FILE, 'w') as f:
json.dump(zones, f, indent=4)
def load_zones_from_file():
global zones
if os.path.exists(ZONES_FILE):
try:
with open(ZONES_FILE, 'r') as f:
loaded_zones = json.load(f)
if not (isinstance(loaded_zones, dict) and all(k in loaded_zones for k in ["easy", "medium", "hard"])):
raise ValueError("Invalid format")
migrated = False
for difficulty in loaded_zones:
for zone in loaded_zones[difficulty]:
if 'id' not in zone:
zone['id'] = uuid.uuid4().hex
migrated = True
zones = loaded_zones
print(zones)
if migrated:
print("Info: Migrated old zone data by adding unique IDs.")
save_zones_to_file()
except (json.JSONDecodeError, IOError, ValueError):
print(f"Warning: '{ZONES_FILE}' is corrupted or invalid. Recreating with empty zones.")
save_zones_to_file()
else:
save_zones_to_file()
LOCATIONS = [
{'lat': 48.85824, 'lng': 2.2945},
{'lat': 40.748440, 'lng': -73.985664},
{'lat': 35.689487, 'lng': 139.691711},
{'lat': -33.856784, 'lng': 151.215297}
]
def generate_id():
return ''.join(random.choices('abcdefghijklmnopqrstuvwxyz0123456789', k=10))
HF_DATASET_REPO = 'Jofthomas/geoguessr_game_of_the_day'
GOOGLE_MAPS_API_KEY = os.getenv('GOOGLE_MAPS_API_KEY')
SERVER_HF_TOKEN = os.getenv('HF_TOKEN', '')
model_id = "mistralai/Magistral-Small-2509"
tokenizer = AutoTokenizer.from_pretrained(model_id, tokenizer_type="mistral", use_fast=False)
model = Mistral3ForConditionalGeneration.from_pretrained(
model_id, torch_dtype=torch.bfloat16, device_map="auto"
).eval()
# SYSTEM_PROMPT_TEXT = (
# "You are a world-class geolocation expert. Given a street-view style image, "
# "think step by step about visual clues and infer approximate coordinates. "
# "When you conclude, output your answer inside [ANSWER]lat,lng[/ANSWER]."
# )
SYSTEM_PROMPT_TEXT = """First draft your thinking process (inner monologue) until you arrive at a response. Format your response using Markdown, and use LaTeX for any mathematical equations. Write both your thoughts and the response in the same language as the input.
Your thinking process must follow the template below:[THINK]Your thoughts or/and draft, like working through an exercise on scratch paper. Be as casual and as long as you want until you are confident to generate the response to the user.[/THINK]Here, provide a self-contained response."""
USER_INSTRUCTION = """You are a world-class geolocation expert. Given a street-view style image, think step by step about visual clues and infer approximate coordinates.
When you conclude, output your final answer inside [ANSWER]lat,lng[/ANSWER].
Please analyze this image and provide coordinates in the required format."""
@spaces.GPU(duration=120)
def llm_decode_image_return_text(image_bytes: bytes) -> str:
print(f"[llm] decode start. image_bytes={len(image_bytes)} bytes")
import base64, mimetypes
try:
encoded_image = base64.b64encode(image_bytes).decode('utf-8')
mime_type = 'image/jpeg'
data_url = f"data:{mime_type};base64,{encoded_image}"
messages = [
{"role": "system", "content": [{"type": "text", "text": SYSTEM_PROMPT_TEXT}]},
{"role": "user", "content": [
{"type": "text", "text": USER_INSTRUCTION},
{"type": "image_url", "image_url": {"url": data_url}},
]},
]
print(f"[llm] messages prepared. system+user with image_url length={len(data_url)}")
tokenized = tokenizer.apply_chat_template(messages, return_dict=True)
print(f"[llm] tokenized keys={list(tokenized.keys())}")
import torch
input_ids = torch.tensor(tokenized.input_ids).unsqueeze(0)
attention_mask = torch.tensor(tokenized.attention_mask).unsqueeze(0)
print(f"[llm] input_ids shape={tuple(input_ids.shape)} attn_mask shape={tuple(attention_mask.shape)} device={model.device}")
kwargs = {
'input_ids': input_ids.to(model.device),
'attention_mask': attention_mask.to(model.device),
}
if 'pixel_values' in tokenized and len(tokenized.pixel_values) > 0:
pixel_values = torch.tensor(tokenized.pixel_values[0], dtype=model.dtype).unsqueeze(0).to(model.device)
image_sizes = torch.tensor(pixel_values.shape[-2:]).unsqueeze(0).to(model.device)
kwargs.update({'pixel_values': pixel_values, 'image_sizes': image_sizes})
print(f"[llm] pixel_values shape={tuple(pixel_values.shape)} image_sizes={tuple(image_sizes.shape)}")
output = model.generate(**kwargs)[0]
print(f"[llm] generate done. output length={len(output)}")
decoded = tokenizer.decode(output[len(tokenized.input_ids): ( -1 if output[-1] == tokenizer.eos_token_id else len(output) )])
print(f"[llm] decode done. text length={len(decoded)}")
return decoded
except Exception as e:
print(f"[llm] decode failed: {e}")
return f"[Error] {e}"
@spaces.GPU(duration=120)
def llm_stream_image_text(image_bytes: bytes):
print(f"[llm-stream] start. image_bytes={len(image_bytes)} bytes")
import base64
try:
encoded_image = base64.b64encode(image_bytes).decode('utf-8')
data_url = f"data:image/jpeg;base64,{encoded_image}"
messages = [
{"role": "system", "content": [{"type": "text", "text": SYSTEM_PROMPT_TEXT}]},
{"role": "user", "content": [
{"type": "text", "text": USER_INSTRUCTION},
{"type": "image_url", "image_url": {"url": data_url}},
]},
]
tokenized = tokenizer.apply_chat_template(messages, return_dict=True)
input_ids = torch.tensor(tokenized.input_ids).unsqueeze(0)
attention_mask = torch.tensor(tokenized.attention_mask).unsqueeze(0)
kwargs = {
'input_ids': input_ids.to(model.device),
'attention_mask': attention_mask.to(model.device),
'max_new_tokens': 80_000,
}
if 'pixel_values' in tokenized and len(tokenized.pixel_values) > 0:
pixel_values = torch.tensor(tokenized.pixel_values[0], dtype=model.dtype).unsqueeze(0).to(model.device)
image_sizes = torch.tensor(pixel_values.shape[-2:]).unsqueeze(0).to(model.device)
kwargs.update({'pixel_values': pixel_values, 'image_sizes': image_sizes})
streamer = TextIteratorStreamer(tokenizer, skip_prompt=True, skip_special_tokens=False)
kwargs['streamer'] = streamer
thread = threading.Thread(target=model.generate, kwargs=kwargs)
thread.start()
acc = ""
for new_text in streamer:
acc += new_text
yield acc
except Exception as e:
yield f"[Error] {e}"
def pick_random_location(difficulty: str) -> Dict[str, float]:
candidates = zones.get(difficulty, [])
if candidates:
selected_zone = random.choice(candidates)
if selected_zone.get('type') == 'rectangle':
b = selected_zone['bounds']
north, south, east, west = b['north'], b['south'], b['east'], b['west']
if west > east:
east += 360
lng = random.uniform(west, east)
if lng > 180:
lng -= 360
lat = random.uniform(south, north)
ensured = _ensure_street_view_location(lat, lng)
if ensured:
return ensured
fallback = random.choice(LOCATIONS)
ensured_fallback = _ensure_street_view_location(fallback['lat'], fallback['lng'])
return ensured_fallback or fallback
def street_view_image_url(lat: float, lng: float) -> str:
if not GOOGLE_MAPS_API_KEY:
# Fallback placeholder to avoid blank image when key is missing
return "https://picsum.photos/1200/800"
return (
f"https://maps.googleapis.com/maps/api/streetview?size=1200x800&location={lat},{lng}&fov=60&pitch=0&source=outdoor&key={GOOGLE_MAPS_API_KEY}"
)
def _has_street_view(lat: float, lng: float) -> bool:
if not GOOGLE_MAPS_API_KEY:
return True
try:
resp = requests.get(
"https://maps.googleapis.com/maps/api/streetview/metadata",
params={"location": f"{lat},{lng}", "source": "outdoor", "key": GOOGLE_MAPS_API_KEY},
timeout=5,
)
resp.raise_for_status()
data = resp.json()
# Check if it's OK and preferably outdoor (not inside buildings)
if data.get("status") == "OK":
# Prefer locations that are explicitly outdoor
location_type = data.get("location_type")
# If location_type is available, check it's not indoors
if location_type and location_type == "INDOOR":
return False
return True
return False
except Exception:
return False
def _snap_to_nearest_road(lat: float, lng: float) -> Optional[Dict[str, float]]:
if not GOOGLE_MAPS_API_KEY:
return None
try:
resp = requests.get(
"https://roads.googleapis.com/v1/nearestRoads",
params={"points": f"{lat},{lng}", "key": GOOGLE_MAPS_API_KEY},
timeout=5,
)
resp.raise_for_status()
data = resp.json()
points = data.get("snappedPoints") or []
if not points:
return None
loc = points[0].get("location") or {}
if "latitude" in loc and "longitude" in loc:
return {"lat": float(loc["latitude"]), "lng": float(loc["longitude"])}
except Exception:
pass
return None
def _ensure_street_view_location(lat: float, lng: float) -> Optional[Dict[str, float]]:
"""Return a coordinate with confirmed Street View coverage, snapped near a road when possible."""
if not GOOGLE_MAPS_API_KEY:
return {"lat": lat, "lng": lng}
checked: set[tuple] = set()
snapped = _snap_to_nearest_road(lat, lng)
candidates: List[Dict[str, float]] = []
if snapped:
candidates.append(snapped)
candidates.append({"lat": lat, "lng": lng})
# Explore a few jittered points if needed
if not snapped:
increments = [0.0005, -0.0005, 0.001, -0.001]
for d_lat in increments:
for d_lng in increments:
if d_lat == 0 and d_lng == 0:
continue
candidates.append({"lat": lat + d_lat, "lng": lng + d_lng})
for candidate in candidates:
key = (round(candidate["lat"], 6), round(candidate["lng"], 6))
if key in checked:
continue
checked.add(key)
if _has_street_view(candidate["lat"], candidate["lng"]):
return candidate
return None
def haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
from math import radians, cos, sin, asin, sqrt
lon1, lat1, lon2, lat2 = map(radians, [lon1, lat1, lon2, lat2])
dlon = lon2 - lon1
dlat = lat2 - lat1
a = sin(dlat/2)**2 + cos(lat1) * cos(lat2) * sin(dlon/2)**2
c = 2 * asin(sqrt(a))
r = 6371
return c * r
def score_from_distance_km(distance_km: float) -> float:
max_score = 5000.0
return max(0.0, max_score - distance_km)
def build_street_html(image_url: str) -> str:
base = """
"""
return base.replace('__IMG_URL__', image_url)
def gr_start_game(difficulty: str, username: str, request: gr.Request):
rounds: List[Dict[str, Any]] = []
date_str = datetime.now(timezone.utc).date().isoformat()
game_id = str(uuid.uuid4()) # Generate unique game ID
for _ in range(3):
loc = pick_random_location(difficulty)
round_id = generate_id()
rounds.append({
'id': round_id,
'lat': loc['lat'],
'lng': loc['lng'],
'image_url': street_view_image_url(loc['lat'], loc['lng']),
'human_guess': None,
'ai_guess': None,
'human_score': 0.0,
'ai_score': 0.0,
})
user_sessions[username] = {
'game_id': game_id,
'difficulty': difficulty,
'rounds': rounds,
'total_score': 0.0,
'completed': False,
'date': date_str,
}
r0 = rounds[0]
street_html = build_street_html(r0['image_url'])
return rounds, 0, r0['id'], street_html, "", ""
def get_round(username: str, round_id: str) -> Optional[Dict[str, Any]]:
session_data = user_sessions.get(username)
if not session_data:
return None
for r in session_data['rounds']:
if r['id'] == round_id:
return r
return None
def gr_submit_guess(round_id: str, lat: float, lng: float, username: str, request: gr.Request):
rnd = get_round(username, round_id)
if not rnd:
return "", "Round not found", gr.update(), gr.update(), gr.update()
distance_km = haversine_km(rnd['lat'], rnd['lng'], float(lat), float(lng))
score = score_from_distance_km(distance_km)
rnd['human_guess'] = {'lat': float(lat), 'lng': float(lng)}
rnd['human_score'] = score
rnd['human_distance_km'] = float(distance_km)
result_text = f"Your guess was {distance_km:.2f} km away. You scored {score:.0f} points."
scoreboard_html = (
f""
f"
Human Guess Recorded
"
f"
{result_text}
"
f"
"
f" Distance: {distance_km:.2f} km"
f" Score: {score:.0f} pts"
f"
"
f"
AI analysis will be added once the model finishes.
"
f"
"
f"
Round {round_idx}
"
f"
{summary_safe}
"
f"
"
f" Human: {rnd.get('human_score',0):.0f} pts ( {human_guess_str} ) ({rnd.get('human_distance_km',0.0):.1f} km)"
f" AI: {rnd.get('ai_score',0):.0f} pts ( {ai_guess_str} ) ({rnd.get('ai_distance_km',0.0):.1f} km)"
f"
"
f"
Totals — Human {total_human:.0f} / AI {total_ai:.0f}
"
f"
"
f"
AI Analysis
"
f"
" + ai_text_safe + "
"
f"
"
f"
Game Over!
Here are the final scores:
Your Score: {total_human:.0f}
AI's Score: {total_ai:.0f}
{winner_message}
"""
return idx, gr.update(value=summary_html), gr.update(value=""), gr.update(value=""), gr.update(value="")
r = sess['rounds'][idx]
s_html = build_street_html(r['image_url'])
return idx, gr.update(value=s_html), gr.update(value=r['id']), gr.update(value=""), gr.update(value="")
next_btn.click(on_next, inputs=[rounds_state, idx_state, user_profile], outputs=[idx_state, street_html, round_id_box, popup_html, ai_chat])
# Inject boot JS using load(js=callable) compatible format
demo.load(fn=lambda: None, inputs=None, outputs=None, js=APP_BOOT_JS)
if __name__ == "__main__":
demo.queue().launch(server_name="0.0.0.0", server_port=7860)