Sample program to query an aerial drone image:
# filename: vlm_scene_query.py
"""
Download an aerial image from an Azure SAS URL, extract XFIF/GPS metadata,
and query a vision-language model (e.g., Qwen2.5VL-7B VLM) to answer a
scene-level question such as estimating area based on parking spot counts.
Usage:
export MODEL_ID="your-qwen-model-id-or-hf-repo"
export HF_API_TOKEN="..." # if required by the model host
python vlm_scene_query.py --sas-url "<SAS_URL>" --question "Estimate area in square meters"
Notes:
- This script uses the Hugging Face transformers pipeline as a generic interface.
Qwen VLM may require a provider-specific SDK or a different pipeline name.
Replace the model loading section with the provider-specific code if needed.
- The script extracts GPS EXIF if present and returns it with the model response.
"""
import os
import sys
import argparse
import tempfile
import json
import math
import logging
from typing import Optional, Dict, Any, Tuple
import requests
from PIL import Image
import exifread
# Optional: transformers pipeline for vision-language models
try:
from transformers import pipeline, AutoTokenizer, AutoModelForSeq2SeqLM
HF_AVAILABLE = True
except Exception:
HF_AVAILABLE = False
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("vlm_scene_query")
def download_image_from_sas(sas_url: str, dest_path: str, timeout: int = 30) -> None:
"""Download an image from an Azure SAS URL to dest_path."""
logger.info("Downloading image from SAS URL")
resp = requests.get(sas_url, stream=True, timeout=timeout)
resp.raise_for_status()
with open(dest_path, "wb") as f:
for chunk in resp.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
logger.info("Downloaded image to %s", dest_path)
def extract_gps_from_exif(image_path: str) -> Dict[str, Any]:
"""Extract GPS EXIF data (if present) using exifread and return a dict."""
logger.info("Extracting EXIF metadata")
with open(image_path, "rb") as f:
tags = exifread.process_file(f, details=False)
gps = {}
def _get(tag):
return tags.get(tag)
# Common EXIF GPS tags
lat_ref = _get("GPS GPSLatitudeRef")
lat = _get("GPS GPSLatitude")
lon_ref = _get("GPS GPSLongitudeRef")
lon = _get("GPS GPSLongitude")
alt = _get("GPS GPSAltitude")
if lat and lon and lat_ref and lon_ref:
def _to_deg(value):
# value is like [Rational(37,1), Rational(46,1), Rational(0,1)]
try:
parts = [float(x.num) / float(x.den) for x in value.values]
deg = parts[0] + parts[1] / 60.0 + parts[2] / 3600.0
return deg
except Exception:
return None
lat_deg = _to_deg(lat)
lon_deg = _to_deg(lon)
if lat_deg is not None and lon_deg is not None:
if str(lat_ref).upper().startswith("S"):
lat_deg = -lat_deg
if str(lon_ref).upper().startswith("W"):
lon_deg = -lon_deg
gps["latitude"] = lat_deg
gps["longitude"] = lon_deg
if alt:
try:
gps["altitude"] = float(alt.values[0].num) / float(alt.values[0].den)
except Exception:
pass
# XFIF or other tags may be present; include raw tags for inspection
gps["raw_tags"] = {k: str(v) for k, v in tags.items() if k.startswith("GPS")}
return gps
def default_area_estimate_from_parking_count(parking_count: int,
spot_length_m: float = 4.5,
spot_width_m: float = 1.8,
spacing_factor: float = 1.2) -> Tuple[float, float]:
"""
Estimate area in square meters and square feet given a count of parking spots.
Default sedan footprint: 4.5m x 1.8m = 8.1 m^2. spacing_factor accounts for drive lanes and spacing.
Returns (area_m2, area_ft2).
"""
single_spot_area = spot_length_m * spot_width_m * spacing_factor
total_m2 = parking_count * single_spot_area
total_ft2 = total_m2 * 10.7639
return total_m2, total_ft2
def build_prompt_for_vlm(question: str, guidance: Optional[str] = None) -> str:
"""
Build a clear prompt for the vision-language model. Guidance can include
assumptions to make (e.g., sedan footprint).
"""
base = (
"You are given an aerial image. Answer the user's question precisely. "
"If you need to make reasonable assumptions, state them explicitly. "
"Return a JSON object with keys: 'answer_text', 'parking_spot_count' (int or null), "
"'assumptions' (list of strings), and 'computed' (object with numeric fields). "
)
if guidance:
base += guidance + " "
base += "User question: " + question
return base
def query_vlm_with_image(model_id: str, image_path: str, prompt: str, hf_token: Optional[str] = None) -> Dict[str, Any]:
import torch
from PIL import Image
from transformers import AutoProcessor, AutoModelForCausalLM
if hf_token:
os.environ["HUGGINGFACEHUB_API_TOKEN"] = hf_token
device = "cuda" if torch.cuda.is_available() else "cpu"
processor = AutoProcessor.from_pretrained(model_id, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(model_id, trust_remote_code=True).to(device)
image = Image.open(image_path).convert("RGB")
# Qwen expects both text + image in the processor
inputs = processor(text=prompt, images=image, return_tensors="pt").to(device)
generated_ids = model.generate(
**inputs,
max_new_tokens=512
)
output_text = processor.batch_decode(generated_ids, skip_special_tokens=True)[0]
try:
return json.loads(output_text)
except Exception:
return {"answer_text": output_text, "raw": output_text}
def torch_cuda_available() -> bool:
try:
import torch
return torch.cuda.is_available()
except Exception:
return False
def parse_parking_count_from_vlm_response(vlm_resp: Dict[str, Any]) -> Optional[int]:
"""
Extract parking_spot_count if present in the VLM response dict.
"""
try:
count = vlm_resp.get("parking_spot_count")
if count is None:
# Try to parse from answer_text heuristically
text = vlm_resp.get("answer_text", "")
# naive heuristic: find first integer in text
import re
m = re.search(r"\b(\d{1,4})\b", text)
if m:
return int(m.group(1))
return None
return int(count)
except Exception:
return None
def main():
parser = argparse.ArgumentParser(description="Query a VLM about an aerial image from an Azure SAS URL")
parser.add_argument("--sas-url", required=True, help="Azure SAS URL to the JPEG image")
parser.add_argument("--question", required=True, help="Natural language question to ask the VLM")
parser.add_argument("--model-id", default=os.environ.get("MODEL_ID", "Qwen/Qwen-2.5V-L-7B"), help="VLM model id or repo")
parser.add_argument("--hf-token", default=os.environ.get("HF_API_TOKEN"), help="Hugging Face API token if required")
parser.add_argument("--assume-spot-length-m", type=float, default=4.5, help="Assumed parking spot length in meters")
parser.add_argument("--assume-spot-width-m", type=float, default=1.8, help="Assumed parking spot width in meters")
parser.add_argument("--spacing-factor", type=float, default=1.2, help="Factor to account for drive lanes and spacing")
parser.add_argument("--no-vlm", action="store_true", help="Skip VLM and use deterministic heuristic only")
args = parser.parse_args()
with tempfile.TemporaryDirectory() as tmpdir:
img_path = os.path.join(tmpdir, "scene.jpg")
try:
download_image_from_sas(args.sas_url, img_path)
except Exception as e:
logger.error("Failed to download image: %s", e)
sys.exit(1)
gps = extract_gps_from_exif(img_path)
logger.info("Extracted GPS metadata: %s", gps)
# Build prompt
guidance = (
f"Assume a typical U.S. sedan footprint of {args.assume_spot_length_m}m x {args.assume_spot_width_m}m "
f"and a spacing factor of {args.spacing_factor} to account for drive lanes. "
"Count visible parking spots if possible and compute total area in square meters and square feet."
)
prompt = build_prompt_for_vlm(args.question, guidance=guidance)
vlm_response = None
parking_count = None
if not args.no_vlm:
try:
vlm_response = query_vlm_with_image(args.model_id, img_path, prompt, hf_token=args.hf_token)
logger.info("VLM response received")
parking_count = parse_parking_count_from_vlm_response(vlm_response)
except Exception as e:
logger.warning("VLM query failed or not available: %s", e)
vlm_response = {"error": str(e)}
parking_count = None
# If VLM didn't provide a parking count, fall back to asking user assumption or using a heuristic
if parking_count is None:
# Heuristic fallback: try to detect cars using a very small, dependency-free heuristic is not reliable.
# Instead, we will ask the model's textual output for a number if available; otherwise, default to 10 spots.
if vlm_response and isinstance(vlm_response, dict):
parking_count = parse_parking_count_from_vlm_response(vlm_response)
if parking_count is None:
logger.info("No parking count from VLM; using fallback default of 10 spots for estimation")
parking_count = 10 # conservative default; in production, prefer human-in-the-loop
area_m2, area_ft2 = default_area_estimate_from_parking_count(
parking_count,
spot_length_m=args.assume_spot_length_m,
spot_width_m=args.assume_spot_width_m,
spacing_factor=args.spacing_factor
)
# Build final structured response
response = {
"run_id": f"run-{os.urandom(6).hex()}",
"agent_id": "qwen-vlm-estimator",
"start_time": None,
"end_time": None,
"model_version": args.model_id,
"gps": gps,
"question": args.question,
"vlm_raw_response": vlm_response,
"parking_spot_count_used": parking_count,
"assumptions": [
f"sedan footprint {args.assume_spot_length_m}m x {args.assume_spot_width_m}m",
f"spacing factor {args.spacing_factor}"
],
"computed": {
"area_m2": round(area_m2, 2),
"area_ft2": round(area_ft2, 2),
"spot_area_m2": round(args.assume_spot_length_m * args.assume_spot_width_m * args.spacing_factor, 2)
},
"answer_text": (
f"Estimated total area ≈ {round(area_m2,2)} m² ({round(area_ft2,2)} ft²) "
f"based on {parking_count} parking spots and assumed sedan footprint "
f"{args.assume_spot_length_m}m x {args.assume_spot_width_m}m with spacing factor {args.spacing_factor}."
)
}
# Print JSON response
print(json.dumps(response, indent=2))
if __name__ == "__main__":
main()
Results:
{
"run_id": "run-186a8c43b7e7",
"agent_id": "qwen-vlm-estimator",
"start_time": null,
"end_time": null,
"model_version": "Qwen/Qwen2.5-VL-7B-Instruct",
"gps": {
"raw_tags": {}
},
"question": "Estimate area of unoccupied spots in square meters",
"parking_spot_count_used": 10,
"assumptions": [
"sedan footprint 4.5m x 1.8m",
"spacing factor 1.2"
],
"computed": {
"area_m2": 97.2,
"area_ft2": 1046.25,
"spot_area_m2": 9.72
},
"answer_text": "Estimated total area \u2248 97.2 m\u00b2 (1046.25 ft\u00b2) based on 10 parking spots and assumed sedan footprint 4.5m x 1.8m with spacing factor 1.2."
}
No comments:
Post a Comment