""" Detection service — runs detection-only jobs through the pipeline + orchestrator + normalization + confidence, returns a UnifiedFaceReport. """ from __future__ import annotations import time from typing import Optional from confidence.engine import ConfidenceEngine from confidence.conflicts import ConflictDetector from metrics.collector import MetricsCollector from models.jobs import JobRequest from models.providers import ProviderCapability from normalization.merger import ReportMerger from orchestrator.runner import Orchestrator from pipeline import ( InputValidator, ImagePreprocessor, ImageHasher, FeatureExtractor, ) from providers.registry import ProviderRegistry from storage.cache import Cache from utils.logging import execution_context, new_execution_id class DetectionService: """Handles detection-only jobs.""" def __init__( self, registry: ProviderRegistry, orchestrator: Orchestrator, cache: Cache, metrics: MetricsCollector, validator: InputValidator, preprocessor: ImagePreprocessor, hasher: ImageHasher, feature_extractor: FeatureExtractor, confidence_engine: ConfidenceEngine, conflict_detector: ConflictDetector, ) -> None: self._registry = registry self._orchestrator = orchestrator self._cache = cache self._metrics = metrics self._validator = validator self._preprocessor = preprocessor self._hasher = hasher self._feature_extractor = feature_extractor self._merger = ReportMerger(confidence_engine, conflict_detector) async def detect(self, request: JobRequest) -> dict: eid = new_execution_id() with execution_context(execution_id=eid, provider_id="detection_service"): t0 = time.perf_counter() # 1. Validate 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"} # 2. Preprocess if vr.source == "url": pre = self._preprocessor.from_url(request.image_url) else: pre = self._preprocessor.from_bytes(vr.image_bytes, vr.source) # 3. Hash img_hash = self._hasher.hash(pre.image) # 4. Feature extraction (uses default detector) 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, ) # 5. Orchestrate detection providers results = await self._orchestrator.run( pipeline_output=pipeline_output, capabilities=[ProviderCapability.DETECTION], provider_whitelist=request.providers or None, execution_id=eid, ) # 6. Merge + confidence elapsed = (time.perf_counter() - t0) * 1000.0 report = self._merger.merge( results=results, image_hash=img_hash, job_id=eid, total_elapsed_ms=elapsed, kind="detection", ) self._metrics.timings.record("job.detection", elapsed) self._metrics.counters.inc("jobs.detection.completed") return { "success": True, "report": report.model_dump(), "elapsed_ms": round(elapsed, 3), }