| """Count visible objects with schema-constrained vision output.""" |
|
|
| from __future__ import annotations |
|
|
| import math |
| import os |
| import tempfile |
| from dataclasses import dataclass, field |
| from typing import Any, Dict, Optional |
|
|
| from eviscene.integrations.gemini import RUNTIME_METRICS_KEY, exception_runtime_metrics |
| from eviscene.tools.base import BaseTool |
|
|
|
|
| COUNT_PROMPT_TEMPLATE = """\ |
| You are the object-counting module of an image verification system. |
| Count only the visible instances of the requested target. |
| |
| Target: {target_object} |
| |
| Rules: |
| 1. Inspect the full supplied image or crop systematically. |
| 2. Count a partially occluded instance only when its presence is visually supported. |
| 3. Return zero when no target instance is visible. |
| 4. Do not infer instances outside the frame or hidden behind objects. |
| 5. Keep the explanation literal and concise. |
| 6. Return exactly one JSON object matching the response schema. |
| """ |
|
|
|
|
| COUNT_RESPONSE_SCHEMA: Dict[str, Any] = { |
| "type": "object", |
| "properties": { |
| "target": {"type": "string", "maxLength": 200}, |
| "count": {"type": "integer", "minimum": 0}, |
| "confidence": {"type": "number", "minimum": 0.0, "maximum": 1.0}, |
| "details": {"type": "string", "maxLength": 800}, |
| "locations": { |
| "type": "array", |
| "items": {"type": "string", "maxLength": 240}, |
| "maxItems": 50, |
| }, |
| }, |
| } |
|
|
|
|
| @dataclass |
| class CountObjectsTool(BaseTool): |
| """Count a requested visible object in the image or a normalized crop.""" |
|
|
| name: str = "count_objects" |
| description: str = ( |
| "Count a specific visible object in the image or in a normalized crop. " |
| "Provide a precise target such as 'fingers on the left hand' or " |
| "'people in the background'. Returns a count, confidence, and locations." |
| ) |
| parameters: dict = field( |
| default_factory=lambda: { |
| "type": "object", |
| "properties": { |
| "image_input": { |
| "type": "string", |
| "description": "Local path to the image file.", |
| }, |
| "target_object": { |
| "type": "string", |
| "description": "A precise description of what to count.", |
| }, |
| "bbox": { |
| "type": "array", |
| "items": {"type": "number"}, |
| "minItems": 4, |
| "maxItems": 4, |
| "description": ( |
| "Optional normalized crop [x1, y1, x2, y2], with every " |
| "coordinate in [0, 1]." |
| ), |
| }, |
| "visual_question_id": {"type": "string"}, |
| "source_evidence_id": {"type": "string"}, |
| "source_discovery_id": {"type": "string"}, |
| "expected_property": {"type": "string"}, |
| }, |
| "required": ["image_input", "target_object"], |
| } |
| ) |
|
|
| client: Optional[Any] = field(default=None, repr=False) |
| provider: str = "gemini" |
| model_name: str = "gemini-3.7-flash" |
|
|
| def _get_client(self) -> Any: |
| if self.client is None: |
| from eviscene.integrations.vlm.factory import build_vlm_client |
|
|
| self.client = build_vlm_client( |
| provider=self.provider, |
| model_name=self.model_name, |
| ) |
| return self.client |
|
|
| def call(self, params: Dict[str, Any]) -> Dict[str, Any]: |
| image_input = str(params.get("image_input", "")).strip() |
| target_object = str(params.get("target_object", "")).strip() |
| if not image_input: |
| return {"status": "error", "error": "image_input is required."} |
| if not target_object: |
| return {"status": "error", "error": "target_object is required."} |
|
|
| actual_image = image_input |
| temporary_path: Optional[str] = None |
| bbox = params.get("bbox") |
| if bbox is not None: |
| try: |
| coordinates = self._validate_bbox(bbox) |
| actual_image, temporary_path = self._crop(image_input, coordinates) |
| except Exception as exc: |
| return { |
| "status": "error", |
| "error": f"Requested count region could not be cropped: {exc}", |
| } |
|
|
| try: |
| parsed = self._get_client().create_image_json( |
| system_prompt=COUNT_PROMPT_TEMPLATE.format( |
| target_object=target_object, |
| ), |
| user_text=f"Count the visible instances of: {target_object}", |
| image_input=actual_image, |
| max_tokens=1000, |
| model_name=self.model_name, |
| response_schema=COUNT_RESPONSE_SCHEMA, |
| ) |
| except Exception as exc: |
| error = { |
| "status": "error", |
| "error": f"Counting failed: {type(exc).__name__}: {exc or '<no message>'}", |
| } |
| metrics = exception_runtime_metrics(exc) |
| if metrics: |
| error[RUNTIME_METRICS_KEY] = metrics |
| return error |
| finally: |
| if temporary_path: |
| try: |
| os.remove(temporary_path) |
| except OSError: |
| pass |
|
|
| result = self._validate_response(parsed, target_object) |
| runtime_metrics = parsed.get(RUNTIME_METRICS_KEY) |
| if runtime_metrics: |
| result[RUNTIME_METRICS_KEY] = runtime_metrics |
| return result |
|
|
| @staticmethod |
| def _validate_bbox(value: Any) -> tuple[float, float, float, float]: |
| if not isinstance(value, (list, tuple)) or len(value) != 4: |
| raise ValueError("bbox must contain exactly four normalized coordinates") |
| if any(isinstance(item, bool) or not isinstance(item, (int, float)) for item in value): |
| raise ValueError("bbox coordinates must be numeric") |
| coordinates = tuple(float(item) for item in value) |
| if any(not math.isfinite(item) or item < 0.0 or item > 1.0 for item in coordinates): |
| raise ValueError("bbox coordinates must be finite values in [0, 1]") |
| x1, y1, x2, y2 = coordinates |
| if x1 >= x2 or y1 >= y2: |
| raise ValueError("bbox must satisfy x1 < x2 and y1 < y2") |
| return x1, y1, x2, y2 |
|
|
| @staticmethod |
| def _crop( |
| image_input: str, |
| bbox: tuple[float, float, float, float], |
| ) -> tuple[str, str]: |
| from PIL import Image |
|
|
| with Image.open(image_input) as image: |
| width, height = image.size |
| x1, y1, x2, y2 = bbox |
| pixel_box = ( |
| max(0, min(int(x1 * width), width - 1)), |
| max(0, min(int(y1 * height), height - 1)), |
| max(1, min(int(math.ceil(x2 * width)), width)), |
| max(1, min(int(math.ceil(y2 * height)), height)), |
| ) |
| if pixel_box[0] >= pixel_box[2] or pixel_box[1] >= pixel_box[3]: |
| raise ValueError("bbox resolves to an empty crop") |
| cropped = image.crop(pixel_box) |
| handle, temporary_path = tempfile.mkstemp(prefix="count_objects_", suffix=".png") |
| os.close(handle) |
| try: |
| cropped.save(temporary_path) |
| except Exception: |
| try: |
| os.remove(temporary_path) |
| except OSError: |
| pass |
| raise |
| return temporary_path, temporary_path |
|
|
| @staticmethod |
| def _validate_response(parsed: Any, target_object: str) -> Dict[str, Any]: |
| if not isinstance(parsed, dict): |
| return {"status": "error", "error": "Counting response must be a JSON object."} |
|
|
| count = parsed.get("count") |
| if isinstance(count, bool) or not isinstance(count, int) or count < 0: |
| return { |
| "status": "error", |
| "error": "Counting response must contain a non-negative integer 'count'.", |
| } |
| confidence = parsed.get("confidence") |
| if isinstance(confidence, bool) or not isinstance(confidence, (int, float)): |
| return { |
| "status": "error", |
| "error": "Counting response must contain numeric 'confidence'.", |
| } |
| confidence = float(confidence) |
| if not math.isfinite(confidence) or not 0.0 <= confidence <= 1.0: |
| return { |
| "status": "error", |
| "error": "Counting confidence must be a finite value in [0, 1].", |
| } |
|
|
| details = parsed.get("details") |
| locations = parsed.get("locations") |
| if not isinstance(details, str): |
| return {"status": "error", "error": "Counting details must be a string."} |
| if not isinstance(locations, list) or any( |
| not isinstance(location, str) for location in locations |
| ): |
| return {"status": "error", "error": "Counting locations must be an array of strings."} |
|
|
| return { |
| "status": "success", |
| "target": target_object, |
| "count": count, |
| "confidence": confidence, |
| "details": details.strip(), |
| "locations": [location.strip() for location in locations], |
| } |
|
|
