feat(scenegraph): add SoftwareCamIndex for visual hash similarity queries

- Add SoftwareCamIndex class with xnor_popcount_score for CAM-style matching
- Add CamMatch and SceneGraphMatch dataclasses for query results
- Add query_by_visual_hash method to SimpleSceneGraph
- Add comprehensive tests for SoftwareCamIndex and xnor_popcount_score
This commit is contained in:
2026-05-17 21:00:54 +08:00
parent ddb8cff6a9
commit 583b2156ea
4 changed files with 314 additions and 2 deletions

View File

@@ -14,15 +14,20 @@ from .hash_codec import (
)
from .objectnode import ObjectNode
from .roomnode import RoomNode
from .scenegraph import SimpleSceneGraph
from .scenegraph import SceneGraphMatch, SimpleSceneGraph
from .software_cam import CamMatch, SoftwareCamIndex, xnor_popcount_score
__all__ = [
"CamMatch",
"ObjectNode",
"RoomNode",
"SceneGraphMatch",
"SimpleSceneGraph",
"SoftwareCamIndex",
"bits_tensor_to_cam_row",
"bits_tensor_to_hash_bytes",
"cam_row_to_hash_bytes",
"hash_bytes_to_bits_array",
"hash_bytes_to_cam_row",
"xnor_popcount_score",
]

View File

@@ -1,10 +1,42 @@
from dataclasses import dataclass, field
from .roomnode import RoomNode
from .objectnode import ObjectNode
from .roomnode import RoomNode
@dataclass(frozen=True)
class SceneGraphMatch:
obj_id: str
node: ObjectNode
row_index: int
score: int
similarity: float
hash_bytes: bytes
@dataclass
class SimpleSceneGraph:
rooms: dict[str, RoomNode] = field(default_factory=dict)
objects: dict[str, ObjectNode] = field(default_factory=dict)
def query_by_visual_hash(
self,
query_hash_bytes: bytes,
*,
top_k: int = 1,
) -> list[SceneGraphMatch]:
from .software_cam import SoftwareCamIndex
index = SoftwareCamIndex.from_scene_graph(self)
cam_matches = index.query(query_hash_bytes, top_k=top_k)
return [
SceneGraphMatch(
obj_id=match.obj_id,
node=self.objects[match.obj_id],
row_index=match.row_index,
score=match.score,
similarity=match.similarity,
hash_bytes=match.hash_bytes,
)
for match in cam_matches
]

View File

@@ -0,0 +1,84 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING
from .hash_codec import DEFAULT_HASH_WIDTH, hash_bytes_to_cam_row
if TYPE_CHECKING:
from .scenegraph import SimpleSceneGraph
def xnor_popcount_score(
query_row: int,
stored_row: int,
*,
width: int = DEFAULT_HASH_WIDTH,
) -> int:
"""Compute CAM-style same-bit score for two integer hash rows."""
if width <= 0:
raise ValueError("width must be greater than 0")
mask = (1 << width) - 1
return int((~(int(query_row) ^ int(stored_row)) & mask).bit_count())
@dataclass(frozen=True)
class CamMatch:
row_index: int
obj_id: str
score: int
similarity: float
hash_bytes: bytes
@dataclass(frozen=True)
class SoftwareCamIndex:
obj_ids: tuple[str, ...]
rows: tuple[int, ...]
hashes: tuple[bytes, ...]
width: int = DEFAULT_HASH_WIDTH
@classmethod
def from_scene_graph(
cls,
scene_graph: "SimpleSceneGraph",
*,
width: int = DEFAULT_HASH_WIDTH,
) -> "SoftwareCamIndex":
obj_ids: list[str] = []
rows: list[int] = []
hashes: list[bytes] = []
for obj_id, node in scene_graph.objects.items():
hash_bytes = node.visual_hash
obj_ids.append(obj_id)
hashes.append(hash_bytes)
rows.append(hash_bytes_to_cam_row(hash_bytes, width=width))
return cls(
obj_ids=tuple(obj_ids),
rows=tuple(rows),
hashes=tuple(hashes),
width=width,
)
def query(self, query_hash_bytes: bytes, *, top_k: int = 1) -> list[CamMatch]:
if top_k <= 0:
raise ValueError("top_k must be greater than 0")
if not self.rows:
raise ValueError("cannot query an empty SoftwareCamIndex")
query_row = hash_bytes_to_cam_row(query_hash_bytes, width=self.width)
matches = [
CamMatch(
row_index=row_index,
obj_id=self.obj_ids[row_index],
score=score,
similarity=score / float(self.width),
hash_bytes=self.hashes[row_index],
)
for row_index, row in enumerate(self.rows)
for score in [xnor_popcount_score(query_row, row, width=self.width)]
]
matches.sort(key=lambda match: (-match.score, match.row_index))
return matches[: min(top_k, len(matches))]