""" Object Intelligence service — combines object detection (YOLO) + QR codes + barcodes into a unified object intelligence report. For each detected object, produces: - class label + confidence + bounding box - cropped image (base64) - classification flags (is_vehicle, is_screen, is_document, etc.) - searchable metadata (decoded text for QR/barcodes, etc.) Categorizes objects into: - vehicles (car, truck, bus, motorcycle, etc.) - license_plates (detected via OCR on vehicle regions — heuristic) - qr_codes (from QR provider) - barcodes (from barcode provider) - documents (book, document-like objects) - screens (tv, laptop, monitor) - phones (cell phone) - laptops (laptop) - watches (clock, watch) - logos (TODO — would need a logo detector) - text_regions (from OCR provider) Pure computation — relies on existing providers via the orchestrator. """ from __future__ import annotations import base64 import time from typing import List import cv2 import numpy as np from cores.vision import BBox, crop_region, numpy_to_base64 from models.jobs import JobRequest from models.reports import ( DetectedObjectIntelligence, ObjectIntelligenceResult, ) from models.providers import ProviderCapability from orchestrator.runner import Orchestrator from pipeline import InputValidator, ImagePreprocessor, ImageHasher, FeatureExtractor from utils.logging import execution_context, new_execution_id # COCO class -> category mapping VEHICLE_CLASSES = {"car", "truck", "bus", "motorcycle", "bicycle", "boat", "airplane", "train"} SCREEN_CLASSES = {"tv", "laptop", "mouse", "remote", "keyboard", "cell phone"} DOCUMENT_CLASSES = {"book"} PHONE_CLASSES = {"cell phone"} LAPTOP_CLASSES = {"laptop"} WATCH_CLASSES = {"clock"} # text-region is detected by OCR, not YOLO # QR/barcode objects come with is_qr_code / is_barcode flags class ObjectIntelligenceService: """Combines object detection + QR + barcode into unified intelligence.""" def __init__( self, orchestrator: Orchestrator, validator: InputValidator, preprocessor: ImagePreprocessor, hasher: ImageHasher, feature_extractor: FeatureExtractor, ) -> None: self._orchestrator = orchestrator self._validator = validator self._preprocessor = preprocessor self._hasher = hasher self._feature_extractor = feature_extractor async def analyze(self, request: JobRequest) -> dict: """Run object intelligence: YOLO + QR + barcode + classification.""" eid = new_execution_id() with execution_context(execution_id=eid, provider_id="object_intelligence_service"): t0 = time.perf_counter() vr = self._validator.validate( image_url=request.image_url, image_base64=request.image_base64, ) if not vr.valid: return {"success": False, "error": vr.error, "error_type": "ValidationError"} if vr.source == "url": pre = self._preprocessor.from_url(request.image_url) else: pre = self._preprocessor.from_bytes(vr.image_bytes, vr.source) img_hash = self._hasher.hash(pre.image) pipeline_output = self._feature_extractor.extract( pre.image, img_hash, pre.width, pre.height, pre.source, original_bytes=pre.original_bytes, original_format=pre.original_format, ) # Run object detection + QR + barcode providers results = await self._orchestrator.run( pipeline_output=pipeline_output, capabilities=[ProviderCapability.OBJECT_DETECTION], provider_whitelist=request.providers or None, execution_id=eid, ) # Collect all detected objects all_objects: list[dict] = [] for name, result in results.items(): if not result.success: continue for obj in result.normalized.get("objects", []): obj["_provider"] = name all_objects.append(obj) # Build DetectedObjectIntelligence for each img = pre.image intelligent_objects: list[DetectedObjectIntelligence] = [] for obj in all_objects: label = obj.get("label", "unknown") confidence = float(obj.get("confidence", 0.0)) box_dict = obj.get("box", {"x": 0, "y": 0, "w": 0, "h": 0}) # Crop the object bbox = BBox(box_dict["x"], box_dict["y"], box_dict["w"], box_dict["h"]) crop = crop_region(img, bbox, margin=0.05) crop_b64 = None if crop.size > 0: try: crop_b64 = numpy_to_base64(crop, ".jpg", quality=85) except Exception: crop_b64 = None # Classify is_vehicle = label in VEHICLE_CLASSES is_screen = label in SCREEN_CLASSES is_document = label in DOCUMENT_CLASSES is_phone = label in PHONE_CLASSES is_laptop = label in LAPTOP_CLASSES is_watch = label in WATCH_CLASSES is_qr = obj.get("is_qr_code", False) is_barcode = obj.get("is_barcode", False) # license plate heuristic: text near a vehicle is_license_plate = False # would need OCR + vehicle overlap check is_logo = False # no logo detector yet is_text_region = False # would need OCR region detection # Searchable metadata searchable: dict = {"provider": obj.get("_provider", "")} if is_qr or is_barcode: decoded = obj.get("decoded_text", "") if decoded: searchable["decoded_text"] = decoded searchable["search_url"] = f"https://www.google.com/search?q={decoded}" if is_barcode: btype = obj.get("barcode_type", "") if btype: searchable["barcode_type"] = btype intelligent_objects.append(DetectedObjectIntelligence( class_label=label, confidence=round(confidence, 4), box=box_dict, crop_base64=crop_b64, is_vehicle=is_vehicle, is_screen=is_screen, is_document=is_document, is_phone=is_phone, is_laptop=is_laptop, is_watch=is_watch, is_logo=is_logo, is_text_region=is_text_region, is_license_plate=is_license_plate, is_qr_code=is_qr, is_barcode=is_barcode, searchable_metadata=searchable, )) # Categorize vehicles = [o for o in intelligent_objects if o.is_vehicle] license_plates = [o for o in intelligent_objects if o.is_license_plate] qr_codes = [o for o in intelligent_objects if o.is_qr_code] barcodes = [o for o in intelligent_objects if o.is_barcode] documents = [o for o in intelligent_objects if o.is_document] screens = [o for o in intelligent_objects if o.is_screen] phones = [o for o in intelligent_objects if o.is_phone] laptops = [o for o in intelligent_objects if o.is_laptop] watches = [o for o in intelligent_objects if o.is_watch] logos = [o for o in intelligent_objects if o.is_logo] text_regions = [o for o in intelligent_objects if o.is_text_region] elapsed = (time.perf_counter() - t0) * 1000.0 result = ObjectIntelligenceResult( total_objects=len(intelligent_objects), objects=intelligent_objects, vehicles=vehicles, license_plates=license_plates, qr_codes=qr_codes, barcodes=barcodes, documents=documents, screens=screens, phones=phones, laptops=laptops, watches=watches, logos=logos, text_regions=text_regions, elapsed_ms=round(elapsed, 3), ) return { "success": True, "object_intelligence": result.model_dump(), "elapsed_ms": round(elapsed, 3), }