"""GPT-4o Vision analyser for RICS section photos. Extracts professional survey observations from uploaded images. Has a hard timeout and retry logic to handle transient API failures gracefully. """ from __future__ import annotations import base64 import json import logging from pathlib import Path from typing import TYPE_CHECKING if TYPE_CHECKING: pass logger = logging.getLogger(__name__) _VISION_SYSTEM_PROMPT = """You are a UK Chartered Building Surveyor (MRICS) examining a photograph taken during a residential property inspection. Analyse the image and extract factual, professional observations suitable for a RICS survey report. Focus on: - Visible condition (good, fair, poor, defective) - Materials and construction type - Defects (cracks, damp stains, decay, displacement) - Maintenance items or recommendations Do NOT invent or assume defects that are not clearly visible. State only what can be observed. Output a JSON object with a single key "observations" containing a list of concise bullet strings (max 8 items).""" _VISION_USER_TEMPLATE = "This photo is from report section: {section_label}. Describe the relevant observations." _VISION_TIMEOUT_SECONDS: float = 45.0 _VISION_MAX_ATTEMPTS: int = 2 class VisionAnalysisResult: """Outcome of a vision-analysis attempt. Attributes: observations: Extracted bullets (may be empty). ok: ``True`` if the call succeeded. error: Short, human-readable failure reason (``""`` if ok). """ __slots__ = ("observations", "ok", "error") def __init__(self, observations: list[str], ok: bool, error: str = "") -> None: self.observations = observations self.ok = ok self.error = error def analyze_section_photo( image_path: str | Path, mime_type: str, section_label: str, openai_api_key: str, vision_model: str = "gpt-4o", ) -> VisionAnalysisResult: """Analyse a section photo and return observation bullets. Has a hard timeout of ~45 seconds and retries once on transient errors (timeout / connection / 5xx). The return type is a small structured object so the caller can surface a useful error message to the UI when vision fails — avoiding silent skips on a paid API call. Args: image_path: Absolute path to the stored image file. mime_type: MIME type of the image (``image/jpeg``, ``image/png``, etc.). section_label: Human-readable section name (e.g. "Main Roof"). openai_api_key: OpenAI API key. vision_model: Model that supports vision (default ``gpt-4o``). Returns: VisionAnalysisResult with observations list, ok flag, and error message. """ if not openai_api_key: logger.debug("No OpenAI key — skipping vision analysis") return VisionAnalysisResult([], ok=False, error="OpenAI API key not configured.") try: image_bytes = Path(image_path).read_bytes() except OSError as exc: logger.warning("Could not read photo %s: %s", image_path, exc) return VisionAnalysisResult([], ok=False, error=f"Could not read image file: {exc}") b64 = base64.standard_b64encode(image_bytes).decode("ascii") data_url = f"data:{mime_type};base64,{b64}" from app.config import settings as app_settings from openai import APIConnectionError, APIStatusError, APITimeoutError, OpenAI last_error: str = "" for attempt in range(_VISION_MAX_ATTEMPTS): try: messages = [ {"role": "system", "content": _VISION_SYSTEM_PROMPT}, { "role": "user", "content": [ { "type": "image_url", "image_url": {"url": data_url, "detail": "high"}, }, { "type": "text", "text": _VISION_USER_TEMPLATE.format(section_label=section_label), }, ], }, ] from app.llm.openai_chat import chat_completions_create_raw async def _vision_call() -> str: return await chat_completions_create_raw( messages=messages, model=vision_model, max_tokens=400, temperature=0.0, response_format={"type": "json_object"}, phase="section_photo_vision", section_id=section_label, api_key=openai_api_key, ) import asyncio import concurrent.futures def _run_in_fresh_loop() -> str: return asyncio.run(_vision_call()) try: asyncio.get_running_loop() with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: raw = pool.submit(_run_in_fresh_loop).result( timeout=_VISION_TIMEOUT_SECONDS + 10 ) except RuntimeError: raw = _run_in_fresh_loop() data = json.loads(raw) # Distinguish three cases that the previous .get(key, []) collapsed: # 1. key absent → contract violation, surface as ok=False # 2. key present, !list → contract violation, surface as ok=False # 3. key present, list → genuine result (empty list is "model saw nothing", ok=True) if "observations" not in data: logger.warning( "Vision response missing 'observations' key: %s", raw[:200], ) return VisionAnalysisResult( [], ok=False, error="Vision response missing 'observations' key.", ) obs = data["observations"] if not isinstance(obs, list): logger.warning( "Vision response 'observations' is not a list (got %s): %s", type(obs).__name__, raw[:200], ) return VisionAnalysisResult( [], ok=False, error=f"Invalid 'observations' field type: {type(obs).__name__}.", ) clean = [str(o).strip() for o in obs if isinstance(o, str) and o.strip()][:8] logger.info("Vision analysis OK: %d observations from %s", len(clean), image_path) return VisionAnalysisResult(clean, ok=True, error="") except (APITimeoutError, APIConnectionError) as exc: last_error = f"Network error: {exc}" logger.warning("Vision API transient error (attempt %d/%d): %s", attempt + 1, _VISION_MAX_ATTEMPTS, exc) continue except APIStatusError as exc: if exc.status_code >= 500: last_error = f"Server error ({exc.status_code})" logger.warning("Vision API 5xx (attempt %d/%d): %s", attempt + 1, _VISION_MAX_ATTEMPTS, exc) continue else: last_error = f"API error ({exc.status_code}): {exc.message}" logger.warning("Vision API client error: %s", exc) break except json.JSONDecodeError as exc: last_error = f"Invalid JSON response: {exc}" logger.warning("Vision response not valid JSON: %s", exc) break except Exception as exc: last_error = f"Unexpected error: {exc}" logger.exception("Vision analysis unexpected failure") break return VisionAnalysisResult([], ok=False, error=last_error or "Vision analysis failed after retries.")