Sample code to query Qwen2.5VL-7B VLM model in chat mode
#! /usr/bin/python
# 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]:
"""
Query a vision-language model with the image and prompt.
This uses the Hugging Face pipeline interface as a generic example.
Replace with provider-specific SDK if required by the model.
"""
if not HF_AVAILABLE:
raise RuntimeError("transformers not available in this environment; cannot load VLM pipeline")
# If the model requires authentication, set the token in the environment for HF
if hf_token:
os.environ["HUGGINGFACEHUB_API_TOKEN"] = hf_token
# Attempt to use a generic image-to-text pipeline. Some VLMs require custom loading.
logger.info("Loading VLM model pipeline (model_id=%s)", model_id)
try:
# Many VLMs use a "vision-text" or "image-to-text" pipeline; this is a best-effort example.
vlm = pipeline(task="image-to-text", model=model_id, device=0 if torch_cuda_available() else -1)
except Exception:
# Fallback: try seq2seq with processor + model (provider-specific)
# Provide a helpful error message to the user
raise RuntimeError(
"Failed to instantiate a generic image-to-text pipeline for model_id=%s. "
"Qwen VLMs often require provider-specific SDKs or a custom pipeline. "
"Replace this function with the provider's recommended loading code." % model_id
)
logger.info("Running VLM inference")
# Many pipelines accept a list of images and optional prompt; adapt as needed
with open(image_path, "rb") as f:
image_bytes = f.read()
# The pipeline may accept a PIL Image or bytes; try PIL first
pil_img = Image.open(image_path).convert("RGB")
# Some pipelines accept a 'prompt' kwarg; others require concatenating prompt to input.
try:
result = vlm(pil_img, prompt=prompt, max_new_tokens=512)
except TypeError:
# If pipeline doesn't accept prompt kwarg, pass prompt as first argument
result = vlm(prompt + "\n", pil_img, max_new_tokens=512)
# result is often a list of dicts or a string
logger.debug("Raw VLM result: %s", result)
# Normalize to dict
if isinstance(result, list) and len(result) > 0:
raw_text = result[0].get("generated_text") or result[0].get("text") or str(result[0])
elif isinstance(result, dict):
raw_text = result.get("generated_text") or result.get("text") or json.dumps(result)
else:
raw_text = str(result)
# Try to parse JSON from the model output; if not JSON, return text in answer_text
try:
parsed = json.loads(raw_text)
return parsed
except Exception:
return {"answer_text": raw_text, "raw": raw_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()
"""
Sample output:
python Qwen.py --sas-url "https://tinyurl.com/carlot01" --question "what is the size of the parking lot shown in terms of square feet assuming usual size of a sedan in the united states and counting at most 1 occupancy per parking spot" --hf-token "<your-token>"
2026-07-12 19:16:35,444 INFO Downloading image from SAS URL
2026-07-12 19:16:36,468 INFO Downloaded image to C:\Users\ravib\AppData\Local\Temp\tmp6sle4o5d\scene.jpg
2026-07-12 19:16:36,470 INFO Extracting EXIF metadata
2026-07-12 19:16:36,491 INFO Extracted GPS metadata: {'raw_tags': {}}
2026-07-12 19:16:36,492 INFO Loading VLM model pipeline (model_id=Qwen/Qwen-2.5V-L-7B)
2026-07-12 19:16:37,065 INFO HTTP Request: GET https://huggingface.co/api/agent-harnesses "HTTP/1.1 200 OK"
2026-07-12 19:16:37,176 INFO HTTP Request: HEAD https://huggingface.co/Qwen/Qwen-2.5V-L-7B/resolve/main/config.json "HTTP/1.1 401 Unauthorized"
2026-07-12 19:16:37,178 WARNING VLM query failed or not available: Failed to instantiate a generic image-to-text pipeline for model_id=Qwen/Qwen-2.5V-L-7B. Qwen VLMs often require provider-specific SDKs or a custom pipeline. Replace this function with the provider's recommended loading code.
2026-07-12 19:16:37,179 INFO No parking count from VLM; using fallback default of 10 spots for estimation
{
"run_id": "run-70391241b7e3",
"agent_id": "qwen-vlm-estimator",
"start_time": null,
"end_time": null,
"model_version": "Qwen/Qwen-2.5V-L-7B",
"gps": {
"raw_tags": {}
},
"question": "what is the size of the parking lot shown in terms of square feet assuming usual size of a sedan in the united states and counting at most 1 occupancy per parking spot",
"vlm_raw_response": {
"error": "Failed to instantiate a generic image-to-text pipeline for model_id=Qwen/Qwen-2.5V-L-7B. Qwen VLMs often require provider-specific SDKs or a custom pipeline. Replace this function with the provider's recommended loading code."
},
"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