from time import perf_counter from typing import Any from .config import Settings from .embedder import ImageEmbedder from .manifest import Manifest from .qdrant_store import QdrantStore, inspector_results, match_filter, payload_filter from .schemas import Inspector, QueryFilters class AfterimageService: def __init__(self, settings: Settings): self.settings = settings self.manifest = Manifest(settings) self.embedder = ImageEmbedder(settings) self.store = QdrantStore(settings) def health(self) -> dict[str, Any]: collections = self.store.client.get_collections() return { "ok": True, "collection": self.settings.collection_name, "point_count": self.store.count() if self.store.client.collection_exists(self.settings.collection_name) else 0, "embedding_mode": self.embedder.mode, "collections": [item.name for item in collections.collections], "default_scenario": self.manifest.default_scenario, "scenarios": self.manifest.scenario_summaries(), } def scenarios(self) -> dict[str, Any]: return {"default_scenario": self.manifest.default_scenario, "scenarios": self.manifest.scenario_summaries()} def memories(self, scenario: str, zone: str | None, region_id: str | None, is_baseline: bool | None) -> dict[str, Any]: self.manifest.scenario(scenario) query_filter = match_filter(scenario=scenario, zone=zone, region_id=region_id, is_baseline=is_baseline) points = self.store.scroll(query_filter=query_filter, limit=200) return { "points": [self._point_view(point) for point in points], "floorplan": self.manifest.floorplan(scenario), "scenario": self.manifest.scenario(scenario), } def scan_anomalies(self, scenario: str) -> dict[str, Any]: self.manifest.scenario(scenario) regions = [] inspector_blocks = [] for region in self.manifest.regions(scenario): crop = self.manifest.asset_path(region["incident_crop"]) vector = self.embedder.embed_path(crop) query_filter = match_filter(scenario=scenario, region_id=region["region_id"], is_baseline=True) t0 = perf_counter() points = self.store.search(vector, query_filter=query_filter, limit=8) took_ms = round((perf_counter() - t0) * 1000, 2) top_score = points[0].score if points else 0.0 normal_band = self._normal_band(scenario, region["region_id"]) threshold = normal_band["floor"] if normal_band["source"] == "baseline_variants" else float(region.get("threshold", 0.78)) alarm_enabled = bool(region.get("alarm_enabled", True)) status = "anomalous" if alarm_enabled and top_score < threshold else "normal" result = { "region_id": region["region_id"], "region_label": region["region_label"], "camera_id": region["camera_id"], "zone": region["zone"], "score": top_score, "threshold": threshold, "normal_band": normal_band, "alarm_enabled": alarm_enabled, "status": status, "incident_crop_url": self.manifest.asset_url(region["incident_crop"]), "nearest": [self._point_view(point) for point in points], } regions.append(result) inspector_blocks.append( self._inspector( api="query_points", params={ "limit": 8, "query": "incident_region_crop", "region_id": region["region_id"], "alarm_enabled": alarm_enabled, "normal_band": normal_band, }, query_filter={"scenario": scenario, "region_id": region["region_id"], "is_baseline": True}, points=points, took_ms=took_ms, ) ) return {"regions": regions, "inspector": inspector_blocks} def object_search(self, scenario: str, asset_id: str | None, image_ref: str | None, filters: QueryFilters, limit: int = 8): self.manifest.scenario(scenario) crop = self.manifest.crop_for_ref(scenario, asset_id, image_ref) vector = self.embedder.embed_path(crop) filter_dict = filters.model_dump() filter_dict["scenario"] = scenario if filter_dict.get("memory_type") is None: filter_dict["memory_type"] = "object_sighting" query_filter = payload_filter(filter_dict) points = self.store.search(vector, query_filter=query_filter, limit=limit) inspector = self._inspector( api="query_points", params={"limit": limit, "query_crop": str(crop.relative_to(self.settings.resolved_asset_root))}, query_filter=filter_dict, points=points, ) return {"results": [self._point_view(point) for point in points], "inspector": inspector} def trail(self, scenario: str, asset_id: str | None, image_ref: str | None, score_cutoff: float): search = self.object_search( scenario=scenario, asset_id=asset_id, image_ref=image_ref, filters=QueryFilters(scenario=scenario, memory_type="object_sighting"), limit=12, ) by_zone: dict[str, dict[str, Any]] = {} for result in search["results"]: payload = result["payload"] if result["score"] < score_cutoff: continue zone = payload["zone"] current = by_zone.get(zone) if current is None or result["score"] > current["score"]: by_zone[zone] = result sightings = sorted(by_zone.values(), key=lambda item: item["payload"]["timestamp"]) path = [] floor_coords = self._floor_coords(scenario) for item in sightings: payload = item["payload"] path.append({**item, "floorplan_xy": floor_coords[payload["zone"]]}) return {"trail": path, "ranked_results": search["results"], "inspector": search["inspector"]} def outliers(self, scenario: str): self.manifest.scenario(scenario) negative_filter = match_filter(scenario=scenario, memory_type="region_baseline", alarm_enabled=True) negative_points = self.store.scroll(query_filter=negative_filter, limit=50) negative_ids = [str(point.id) for point in negative_points] candidate_filter = match_filter(scenario=scenario, memory_type="region_incident", alarm_enabled=True) t0 = perf_counter() points = self.store.recommend_best_score( negative_ids=negative_ids, query_filter=candidate_filter, limit=8, ) took_ms = round((perf_counter() - t0) * 1000, 2) return { "strategy": "RecommendQuery best_score: alarm-zone normal memories as negatives, alarm-zone incidents as candidates", "results": [self._point_view(point) for point in points], "inspector": self._inspector( api="RecommendQuery(best_score)", params={"candidate_scope": "alarm_enabled regions only", "negative_example_count": len(negative_ids), "limit": 8}, query_filter={"scenario": scenario, "memory_type": "region_incident", "alarm_enabled": True}, points=points, took_ms=took_ms, ), } def text_search(self, query: str, scenario: str | None = None, limit: int = 8): query = (query or "").strip() if not query: raise ValueError("Empty query.") vector = self.embedder.embed_text(query) query_filter = match_filter(scenario=scenario) if scenario else None t0 = perf_counter() points = self.store.search(vector, query_filter=query_filter, limit=limit) took_ms = round((perf_counter() - t0) * 1000, 2) return { "query": query, "results": [self._point_view(point) for point in points], "inspector": self._inspector( api="query_points · text", params={"limit": limit, "query": query, "encoder": "clip-ViT-B-32-text"}, query_filter={"scenario": scenario} if scenario else None, points=points, took_ms=took_ms, ), } def matrix(self, scenario: str, sample: int = 24): self.manifest.scenario(scenario) query_filter = match_filter(scenario=scenario) points = self.store.scroll(query_filter=query_filter, limit=sample) matrix = self.store.search_matrix_pairs(query_filter=query_filter, limit=3, sample=min(sample, len(points) or 1)) nodes = [] floor_coords = self._floor_coords(scenario) for index, point in enumerate(points): payload = point.payload or {} zone = payload.get("zone", "main_hall") base = floor_coords.get(zone, [120, 120]) nodes.append( { "id": point.id, "x": base[0] + (index % 4) * 9, "y": base[1] + (index // 4) * 9, "payload": payload, } ) pairs = [{"a": str(pair.a), "b": str(pair.b), "score": pair.score} for pair in matrix.pairs] return {"nodes": nodes, "pairs": pairs} def _inspector( self, api: str, params: dict[str, Any], query_filter: dict[str, Any] | None, points, took_ms: float | None = None, ) -> dict[str, Any]: return Inspector( api=api, collection=self.settings.collection_name, filter=query_filter, params=params, results=inspector_results(points), took_ms=took_ms, ).model_dump() def _floor_coords(self, scenario: str) -> dict[str, list[int]]: zones = self.manifest.floorplan(scenario).get("zones", []) return {zone["id"]: zone.get("floorplan_xy", [120, 120]) for zone in zones} def _normal_band(self, scenario: str, region_id: str) -> dict[str, Any]: query_filter = match_filter(scenario=scenario, region_id=region_id, is_baseline=True) baseline_points = self.store.scroll(query_filter=query_filter, limit=40) peer_scores = [] for point in baseline_points: neighbors = self.store.search(str(point.id), query_filter=query_filter, limit=min(len(baseline_points), 8)) for neighbor in neighbors: if str(neighbor.id) != str(point.id) and neighbor.score is not None: peer_scores.append(float(neighbor.score)) break if len(peer_scores) < 2: return { "source": "manifest_threshold", "count": len(peer_scores), "mean": None, "min": None, "std": None, "floor": None, } mean = sum(peer_scores) / len(peer_scores) variance = sum((score - mean) ** 2 for score in peer_scores) / len(peer_scores) std = variance ** 0.5 # Data-derived control limit: three sigma below the mean baseline self-similarity. # No hand-tuned constant — the band widens or tightens with the footage itself. floor = max(0.0, mean - 3.0 * std) return { "source": "baseline_variants", "count": len(peer_scores), "mean": mean, "min": min(peer_scores), "std": std, "floor": floor, } def _point_view(self, point) -> dict[str, Any]: payload = dict(point.payload or {}) return { "id": point.id, "score": getattr(point, "score", None), "payload": payload, "crop_url": payload.get("crop_url"), "frame_url": payload.get("frame_url"), }