diff --git a/mini-nav/tests/__init__.py b/mini-nav/tests/__init__.py deleted file mode 100644 index b7a1c46..0000000 --- a/mini-nav/tests/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Test suite for DINOv2 Feature Compressor.""" diff --git a/mini-nav/tests/test_compressors.py b/mini-nav/tests/test_compressors.py deleted file mode 100644 index 7ee60c3..0000000 --- a/mini-nav/tests/test_compressors.py +++ /dev/null @@ -1,341 +0,0 @@ -"""Tests for compressor modules (HashCompressor, Pipeline).""" - -import pytest -import torch -from unittest.mock import Mock, patch -from compressors import ( - BinarySign, - HashCompressor, - HashPipeline, - VideoPositiveMask, - bits_to_hash, - create_pipeline_from_config, - hamming_distance, - hamming_similarity, - hash_to_bits, -) -from configs import cfg_manager -from PIL import Image - - -class TestHashCompressor: - """Test suite for HashCompressor.""" - - def test_hash_compressor_init(self): - """Verify HashCompressor initializes with correct dimensions.""" - compressor = HashCompressor(input_dim=1024, hash_bits=512) - assert compressor.input_dim == 1024 - assert compressor.hash_bits == 512 - - def test_hash_compressor_forward(self): - """Verify forward pass produces correct output shapes.""" - compressor = HashCompressor(input_dim=1024, hash_bits=512) - tokens = torch.randn(4, 197, 1024) # [B, N, input_dim] - - logits, hash_codes, bits = compressor(tokens) - - assert logits.shape == (4, 512) - assert hash_codes.shape == (4, 512) - assert bits.shape == (4, 512) - # Verify bits are binary (0 or 1) - assert torch.all((bits == 0) | (bits == 1)) - - def test_hash_compressor_encode(self): - """Verify encode method returns binary bits.""" - compressor = HashCompressor(input_dim=1024, hash_bits=512) - tokens = torch.randn(2, 197, 1024) - - bits = compressor.encode(tokens) - - assert bits.shape == (2, 512) - assert bits.dtype == torch.int32 - assert torch.all((bits == 0) | (bits == 1)) - - def test_hash_compressor_similarity(self): - """Verify compute_similarity returns correct shape.""" - compressor = HashCompressor(input_dim=1024, hash_bits=512) - - # Create random bits - bits1 = torch.randint(0, 2, (3, 512)) - bits2 = torch.randint(0, 2, (5, 512)) - - sim = compressor.compute_similarity(bits1, bits2) - - assert sim.shape == (3, 5) - - -class TestBinarySign: - """Test suite for BinarySign function.""" - - def test_binary_sign_forward(self): - """Verify BinarySign produces {-1, +1} outputs.""" - x = torch.randn(4, 512) - result = BinarySign.apply(x) - - assert torch.all((result == 1) | (result == -1)) - - def test_binary_sign_round_trip(self): - """Verify bits -> hash -> bits preserves values.""" - bits = torch.randint(0, 2, (4, 512)) - hash_codes = bits_to_hash(bits) - bits_recovered = hash_to_bits(hash_codes) - - assert torch.equal(bits, bits_recovered) - - -class TestHammingMetrics: - """Test suite for Hamming distance and similarity.""" - - def test_hamming_distance_same_codes(self): - """Verify hamming distance is 0 for identical single codes.""" - bits1 = torch.randint(0, 2, (512,)) - bits2 = bits1.clone() - - dist = hamming_distance(bits1, bits2) - - assert dist.item() == 0 - - def test_hamming_distance_self_comparison(self): - """Verify hamming distance diagonal is 0 (each code compared to itself).""" - bits = torch.randint(0, 2, (10, 512)) - - dist = hamming_distance(bits, bits) - - # Diagonal should be 0 (distance to self) - diagonal = torch.diag(dist) - assert torch.all(diagonal == 0) - - def test_hamming_distance_different(self): - """Verify hamming distance is correct for different codes.""" - bits1 = torch.zeros(1, 512, dtype=torch.int32) - bits2 = torch.ones(1, 512, dtype=torch.int32) - - dist = hamming_distance(bits1, bits2) - - assert dist.item() == 512 - - def test_hamming_similarity(self): - """Verify hamming similarity is positive for similar codes.""" - hash1 = torch.ones(1, 512) - hash2 = torch.ones(1, 512) - - sim = hamming_similarity(hash1, hash2) - - assert sim.item() == 512 # Max similarity - - -class TestHashLoss: - """Test suite for HashLoss.""" - - def test_hash_loss_init(self): - """Verify HashLoss initializes with correct parameters.""" - from compressors import HashLoss - - loss_fn = HashLoss( - contrastive_weight=1.0, - distill_weight=0.5, - quant_weight=0.01, - temperature=0.2, - ) - - assert loss_fn.contrastive_weight == 1.0 - assert loss_fn.distill_weight == 0.5 - assert loss_fn.quant_weight == 0.01 - assert loss_fn.temperature == 0.2 - - def test_hash_loss_forward(self): - """Verify HashLoss computes loss correctly.""" - from compressors import HashLoss - - loss_fn = HashLoss() - - batch_size = 4 - hash_bits = 512 - logits = torch.randn(batch_size, hash_bits) - hash_codes = torch.sign(logits) - teacher_embed = torch.randn(batch_size, 1024) - positive_mask = torch.eye(batch_size, dtype=torch.bool) - - total_loss, components = loss_fn( - logits=logits, - hash_codes=hash_codes, - teacher_embed=teacher_embed, - positive_mask=positive_mask, - ) - - assert "contrastive" in components - assert "distill" in components - assert "quantization" in components - assert "total" in components - - -class TestVideoPositiveMask: - """Test suite for VideoPositiveMask.""" - - def test_from_frame_indices(self): - """Verify positive mask generation from frame indices.""" - mask_gen = VideoPositiveMask(temporal_window=2) - - frame_indices = torch.tensor([0, 1, 3, 5]) - - mask = mask_gen.from_frame_indices(frame_indices) - - assert mask.shape == (4, 4) - # Frame 0 and 1 should be positive (distance 1 <= 2) - assert mask[0, 1] == True - # Frame 0 and 3 should be negative (distance 3 > 2) - assert mask[0, 3] == False - - def test_from_video_ids(self): - """Verify positive mask generation from video IDs and frame indices.""" - mask_gen = VideoPositiveMask(temporal_window=2) - - video_ids = torch.tensor([0, 0, 1, 1]) - frame_indices = torch.tensor([0, 1, 0, 1]) - - mask = mask_gen.from_video_ids(video_ids, frame_indices) - - assert mask.shape == (4, 4) - # Same video and temporally close - assert mask[0, 1] == True # video 0, frames 0,1 - # Different video - assert mask[0, 2] == False # video 0 vs 1 - - -class TestHashPipeline: - """Test suite for HashPipeline.""" - - @patch("compressors.pipeline.load_sam_model") - @patch("compressors.pipeline.AutoModel.from_pretrained") - @patch("compressors.pipeline.AutoImageProcessor.from_pretrained") - def test_pipeline_init( - self, - mock_processor_from_pretrained, - mock_model_from_pretrained, - mock_load_sam_model, - ): - """Verify pipeline initializes all components.""" - mock_processor_from_pretrained.return_value = Mock() - - mock_model = Mock() - mock_model.to.return_value = mock_model - mock_model.eval.return_value = None - mock_model_from_pretrained.return_value = mock_model - - mock_load_sam_model.return_value = (Mock(), Mock()) - - pipeline = HashPipeline( - dino_model="facebook/dinov2-large", - hash_bits=512, - ) - - assert pipeline.dino_model == "facebook/dinov2-large" - assert pipeline.sam_model_name == "facebook/sam2.1-hiera-large" - assert pipeline.dino_dim == 1024 - mock_load_sam_model.assert_called_once() - - @patch("compressors.pipeline.load_sam_model") - @patch("compressors.pipeline.AutoModel.from_pretrained") - @patch("compressors.pipeline.AutoImageProcessor.from_pretrained") - def test_pipeline_hash_bits( - self, - mock_processor_from_pretrained, - mock_model_from_pretrained, - mock_load_sam_model, - ): - """Verify pipeline uses correct hash bits.""" - mock_processor_from_pretrained.return_value = Mock() - - mock_model = Mock() - mock_model.to.return_value = mock_model - mock_model.eval.return_value = None - mock_model_from_pretrained.return_value = mock_model - - mock_load_sam_model.return_value = (Mock(), Mock()) - - pipeline = HashPipeline(hash_bits=256) - assert pipeline.hash_bits == 256 - - -class TestConfigIntegration: - """Test suite for config integration with pipeline.""" - - @patch("compressors.pipeline.load_sam_model") - @patch("compressors.pipeline.AutoModel.from_pretrained") - @patch("compressors.pipeline.AutoImageProcessor.from_pretrained") - def test_create_pipeline_from_config( - self, - mock_processor_from_pretrained, - mock_model_from_pretrained, - mock_load_sam_model, - ): - """Verify pipeline can be created from config.""" - mock_processor_from_pretrained.return_value = Mock() - - mock_model = Mock() - mock_model.to.return_value = mock_model - mock_model.eval.return_value = None - mock_model_from_pretrained.return_value = mock_model - - mock_load_sam_model.return_value = (Mock(), Mock()) - - config = cfg_manager.load() - - pipeline = create_pipeline_from_config(config) - - assert isinstance(pipeline, HashPipeline) - assert pipeline.hash_bits == config.model.compression_dim - assert pipeline.sam_max_masks == config.model.sam_max_masks - assert pipeline.sam_min_mask_area == config.model.sam_min_mask_area - - def test_config_settings(self): - """Verify config contains required settings.""" - config = cfg_manager.load() - - assert hasattr(config.model, "dino_model") - assert hasattr(config.model, "compression_dim") - - -@pytest.mark.slow -class TestPipelineIntegration: - """Integration tests for full pipeline (slow, requires model downloads).""" - - def test_pipeline_end_to_end(self): - """Test full pipeline with actual models (slow test).""" - # Skip if no GPU - if not torch.cuda.is_available(): - pytest.skip("Requires CUDA") - - # Create a simple test image - image = Image.new("RGB", (640, 480), color=(128, 128, 128)) - - # Initialize pipeline (will download models on first run) - pipeline = HashPipeline( - dino_model="facebook/dinov2-large", - hash_bits=512, - ) - - # Run pipeline - hash_bits = pipeline(image) - - # Verify output shape - assert hash_bits.dim() == 2 - assert hash_bits.shape[1] == 512 - assert torch.all((hash_bits == 0) | (hash_bits == 1)) - - def test_extract_features(self): - """Test feature extraction.""" - if not torch.cuda.is_available(): - pytest.skip("Requires CUDA") - - image = Image.new("RGB", (640, 480), color=(128, 128, 128)) - - pipeline = HashPipeline( - dino_model="facebook/dinov2-large", - ) - - features = pipeline.extract_features(image) - - # Should return DINO features (1024 for large) - assert features.dim() == 2 - assert features.shape[1] == 1024 diff --git a/mini-nav/tests/test_config.py b/mini-nav/tests/test_config.py deleted file mode 100644 index 0e4019a..0000000 --- a/mini-nav/tests/test_config.py +++ /dev/null @@ -1,167 +0,0 @@ -"""Tests for configuration system using Pydantic models.""" - -import tempfile -from pathlib import Path - -import pytest -import yaml -from configs import ( - Config, - ConfigError, - ConfigManager, - ModelConfig, - OutputConfig, - PoolingType, - cfg_manager, - load_yaml, - save_yaml, -) -from pydantic import ValidationError - - -class TestConfigModels: - """Test suite for Pydantic configuration models.""" - - def test_model_config_defaults(self): - """Verify ModelConfig creates with correct defaults.""" - config = ModelConfig() - assert config.dino_model == "facebook/dinov2-large" - assert config.compression_dim == 512 - assert config.device == "auto" - - def test_model_config_validation(self): - """Test validation constraints for ModelConfig.""" - # Test compression_dim > 0 - with pytest.raises(ValidationError, match="greater than 0"): - ModelConfig(compression_dim=0) - - with pytest.raises(ValidationError, match="greater than 0"): - ModelConfig(compression_dim=-1) - - def test_output_config_defaults(self): - """Verify OutputConfig creates with correct defaults.""" - config = OutputConfig() - output_dir = Path(__file__).parent.parent.parent / "outputs" - - assert config.directory == output_dir - - def test_pooling_type_enum(self): - """Verify PoolingType enum values.""" - assert PoolingType.ATTENTION.value == "attention" - assert PoolingType.ATTENTION == PoolingType("attention") - - def test_feature_compressor_config(self): - """Verify FeatureCompressorConfig nests all models correctly.""" - model_cfg = ModelConfig(compression_dim=512) - out_cfg = OutputConfig(directory="/tmp/outputs") - - config = Config( - model=model_cfg, - output=out_cfg, - ) - - assert config.model.compression_dim == 512 - assert config.output.directory == Path("/tmp/outputs") - - -class TestYamlLoader: - """Test suite for YAML loading and saving.""" - - def test_load_existing_yaml(self): - """Load config.yaml and verify values.""" - config_path = Path(__file__).parent.parent / "configs" / "config.yaml" - config = load_yaml(config_path, Config) - - # Verify model config - assert config.model.dino_model == "facebook/dinov2-large" - assert config.model.compression_dim == 256 - - # Verify output config - output_dir = Path(__file__).parent.parent.parent / "outputs" - - assert config.output.directory == output_dir - - def test_load_yaml_validation(self): - """Test that invalid data raises ConfigError.""" - with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f: - # Write invalid config (missing required fields) - yaml.dump({"invalid": "data"}, f) - temp_path = f.name - - try: - with pytest.raises(ConfigError, match="validation failed"): - load_yaml(Path(temp_path), Config) - finally: - Path(temp_path).unlink() - - def test_save_yaml_roundtrip(self): - """Create config, save to temp, verify file exists with content.""" - original = cfg_manager.load() - - with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f: - temp_path = Path(f.name) - - try: - save_yaml(temp_path, original) - - # Verify file exists and has content - assert Path(temp_path).exists() - with open(temp_path, "r") as f: - content = f.read() - assert len(content) > 0 - assert "model" in content - assert "visualization" in content - assert "output" in content - finally: - Path(temp_path).unlink() - - def test_load_yaml_file_not_found(self): - """Verify FileNotFoundError raises ConfigError.""" - with pytest.raises(ConfigError, match="not found"): - load_yaml(Path("/nonexistent/path/config.yaml"), Config) - - -class TestConfigManager: - """Test suite for ConfigManager singleton with multi-config support.""" - - def test_singleton_pattern(self): - """Verify ConfigManager() returns same instance.""" - manager1 = ConfigManager() - manager2 = ConfigManager() - assert manager1 is manager2 - - def test_load_config(self): - """Test loading default config.""" - config = cfg_manager.load() - - assert config is not None - assert config.model.compression_dim == 256 - - def test_get_without_load(self): - """Test that get() auto-loads config if not loaded.""" - # Reset the singleton's cached config - cfg_manager._config = None - - # get() should auto-load - config = cfg_manager.get() - assert config is not None - assert config.model.compression_dim == 256 - - def test_save_config(self): - """Test saving configuration to file.""" - config = Config( - model=ModelConfig(compression_dim=512), - output=OutputConfig(), - ) - - with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f: - temp_path = Path(f.name) - - try: - cfg_manager.save(config, path=temp_path) - loaded_config = load_yaml(temp_path, Config) - - assert loaded_config.model.compression_dim == 512 - finally: - if temp_path.exists(): - temp_path.unlink() diff --git a/mini-nav/tests/test_feature_extractor.py b/mini-nav/tests/test_feature_extractor.py deleted file mode 100644 index 2a3274c..0000000 --- a/mini-nav/tests/test_feature_extractor.py +++ /dev/null @@ -1,50 +0,0 @@ -"""Tests for feature extraction utilities.""" - -import pytest -import torch -from PIL import Image -from transformers import AutoImageProcessor, AutoModel - -from utils.feature_extractor import ( - extract_batch_features, - extract_single_image_feature, - infer_vector_dim, -) - -TEST_MODEL_NAME = "facebook/dinov2-base" - - -@pytest.fixture -def model_and_processor(): - processor = AutoImageProcessor.from_pretrained(TEST_MODEL_NAME) - model = AutoModel.from_pretrained(TEST_MODEL_NAME) - model.eval() - yield processor, model - del model - del processor - - -def test_infer_vector_dim(model_and_processor): - """Verify infer_vector_dim returns correct dimension.""" - processor, model = model_and_processor - sample_image = Image.new("RGB", (224, 224), color="blue") - dim = infer_vector_dim(processor, model, sample_image) - assert dim == 768 - - -def test_extract_single_image_feature(model_and_processor): - """Verify single image feature extraction.""" - processor, model = model_and_processor - sample_image = Image.new("RGB", (224, 224), color="red") - features = extract_single_image_feature(processor, model, sample_image) - assert isinstance(features, list) - assert len(features) == 768 - - -def test_extract_batch_features(model_and_processor): - """Verify batch feature extraction.""" - processor, model = model_and_processor - images = [Image.new("RGB", (224, 224), color="red") for _ in range(3)] - features = extract_batch_features(processor, model, images) - assert isinstance(features, torch.Tensor) - assert features.shape == (3, 768) diff --git a/mini-nav/tests/test_habitat_simulator.py b/mini-nav/tests/test_habitat_simulator.py deleted file mode 100644 index 46c99b4..0000000 --- a/mini-nav/tests/test_habitat_simulator.py +++ /dev/null @@ -1,124 +0,0 @@ -from types import SimpleNamespace - -import pytest - -from simulator import HabitatSimulatorConfig, create_habitat_simulator - - -class _FakeSimulatorConfiguration: - def __init__(self): - self.scene_id = "" - self.enable_physics = True - - -class _FakeAgentConfiguration: - def __init__(self): - self.sensor_specifications = [] - self.action_space = {} - - -class _FakeCameraSensorSpec: - def __init__(self): - self.uuid = "" - self.sensor_type = None - self.resolution = [] - self.position = [] - - -class _FakeActuationSpec: - def __init__(self, amount): - self.amount = amount - - -class _FakeActionSpec: - def __init__(self, name, actuation): - self.name = name - self.actuation = actuation - - -class _FakeConfiguration: - def __init__(self, sim_cfg, agent_cfgs): - self.sim_cfg = sim_cfg - self.agent_cfgs = agent_cfgs - - -class _FakeSimulator: - def __init__(self, cfg): - self.cfg = cfg - self.initialized_agent_id = None - - def initialize_agent(self, agent_id): - self.initialized_agent_id = agent_id - return {"agent_id": agent_id} - - -def _create_fake_habitat_module(): - return SimpleNamespace( - SimulatorConfiguration=_FakeSimulatorConfiguration, - CameraSensorSpec=_FakeCameraSensorSpec, - SensorType=SimpleNamespace(COLOR="color"), - Configuration=_FakeConfiguration, - Simulator=_FakeSimulator, - agent=SimpleNamespace( - AgentConfiguration=_FakeAgentConfiguration, - ActionSpec=_FakeActionSpec, - ActuationSpec=_FakeActuationSpec, - ), - ) - - -def test_create_habitat_simulator_builds_expected_configuration(): - fake_habitat = _create_fake_habitat_module() - config = HabitatSimulatorConfig( - scene_path="scene.glb", - views_per_room=8, - image_size=128, - sensor_height=1.25, - move_forward_step=0.5, - enable_physics=False, - sensor_uuid="rgb", - agent_id=2, - ) - - simulator, agent = create_habitat_simulator(config, habitat_sim_module=fake_habitat) - - assert simulator.cfg.sim_cfg.scene_id == "scene.glb" - assert simulator.cfg.sim_cfg.enable_physics is False - - created_agent_cfg = simulator.cfg.agent_cfgs[0] - sensor = created_agent_cfg.sensor_specifications[0] - assert sensor.uuid == "rgb" - assert sensor.sensor_type == "color" - assert sensor.resolution == [128, 128] - assert sensor.position == [0.0, 1.25, 0.0] - - assert created_agent_cfg.action_space["move_forward"].actuation.amount == 0.5 - assert created_agent_cfg.action_space["turn_left"].actuation.amount == 45.0 - assert created_agent_cfg.action_space["turn_right"].actuation.amount == 45.0 - - assert simulator.initialized_agent_id == 2 - assert agent == {"agent_id": 2} - - -def test_create_habitat_simulator_validates_views_per_room(): - fake_habitat = _create_fake_habitat_module() - config = HabitatSimulatorConfig(scene_path="scene.glb", views_per_room=0) - - with pytest.raises(ValueError, match="views_per_room"): - create_habitat_simulator(config, habitat_sim_module=fake_habitat) - - -def test_create_habitat_simulator_validates_image_size(): - fake_habitat = _create_fake_habitat_module() - config = HabitatSimulatorConfig(scene_path="scene.glb", image_size=0) - - with pytest.raises(ValueError, match="image_size"): - create_habitat_simulator(config, habitat_sim_module=fake_habitat) - - -def test_create_habitat_simulator_validates_move_forward_step(): - fake_habitat = _create_fake_habitat_module() - config = HabitatSimulatorConfig(scene_path="scene.glb", move_forward_step=0) - - with pytest.raises(ValueError, match="move_forward_step"): - create_habitat_simulator(config, habitat_sim_module=fake_habitat) diff --git a/mini-nav/tests/test_image_utils.py b/mini-nav/tests/test_image_utils.py deleted file mode 100644 index 8e3093d..0000000 --- a/mini-nav/tests/test_image_utils.py +++ /dev/null @@ -1,156 +0,0 @@ -from unittest.mock import Mock - -import torch -from PIL import Image - -from utils.image import segment_image, segment_image_dataset - - -def test_segment_image_passes_pil_image_to_mask_generator() -> None: - mock_generator = Mock(return_value={"masks": []}) - - segment_image( - mock_generator, - Image.new("RGBA", (16, 16), color=(255, 0, 0, 255)), - points_per_batch=32, - ) - - image_arg = mock_generator.call_args.args[0] - assert isinstance(image_arg, Image.Image) - assert image_arg.mode == "RGB" - assert mock_generator.call_args.kwargs["points_per_batch"] == 32 - - -def test_segment_image_supports_tensor_masks_output() -> None: - masks_tensor = torch.tensor( - [ - [ - [1, 1, 0], - [1, 1, 0], - [0, 0, 0], - ], - [ - [1, 1, 1], - [1, 1, 1], - [1, 1, 1], - ], - ], - dtype=torch.float32, - ) - mock_generator = Mock(return_value={"masks": masks_tensor}) - - result = segment_image( - mock_generator, - Image.new("RGB", (3, 3), color=(0, 0, 0)), - min_area=3, - max_masks=5, - ) - - assert len(result) == 2 - assert result[0]["area"] == 9 - assert result[0]["bbox"] == [0, 0, 3, 3] - assert result[1]["area"] == 4 - assert result[1]["bbox"] == [0, 0, 2, 2] - - -def test_segment_image_filters_tensor_masks_by_min_area() -> None: - masks_tensor = torch.tensor( - [ - [ - [1, 0, 0], - [0, 0, 0], - [0, 0, 0], - ], - [ - [1, 1, 0], - [1, 1, 0], - [0, 0, 0], - ], - ], - dtype=torch.float32, - ) - mock_generator = Mock(return_value={"masks": masks_tensor}) - - result = segment_image( - mock_generator, - Image.new("RGB", (3, 3), color=(0, 0, 0)), - min_area=2, - max_masks=5, - ) - - assert len(result) == 1 - assert result[0]["area"] == 4 - - -def test_segment_image_dataset_returns_per_image_masks_in_order() -> None: - first_masks = torch.tensor( - [[[1, 1, 0], [1, 1, 0], [0, 0, 0]]], - dtype=torch.float32, - ) - second_masks = torch.tensor( - [[[1, 1, 1], [1, 1, 1], [1, 1, 1]]], - dtype=torch.float32, - ) - mock_generator = Mock( - return_value=[{"masks": first_masks}, {"masks": second_masks}] - ) - images = [ - Image.new("RGB", (3, 3), color=(0, 0, 0)), - Image.new("RGB", (3, 3), color=(0, 0, 0)), - ] - - result = segment_image_dataset( - mock_generator, - images, - min_area=2, - max_masks=5, - points_per_batch=16, - ) - - assert len(result) == 2 - assert result[0][0]["area"] == 4 - assert result[1][0]["area"] == 9 - assert mock_generator.call_count == 1 - - -def test_segment_image_dataset_falls_back_to_single_image_calls() -> None: - call_index = {"value": 0} - - def fake_generator(images, points_per_batch): - if isinstance(images, list): - raise TypeError("Batch input unsupported") - - result_options = [ - { - "masks": torch.tensor( - [[[1, 1, 0], [1, 1, 0], [0, 0, 0]]], - dtype=torch.float32, - ) - }, - { - "masks": torch.tensor( - [[[1, 1, 1], [1, 1, 1], [1, 1, 1]]], - dtype=torch.float32, - ) - }, - ] - out = result_options[call_index["value"]] - call_index["value"] += 1 - return out - - images = [ - Image.new("RGB", (3, 3), color=(0, 0, 0)), - Image.new("RGB", (3, 3), color=(0, 0, 0)), - ] - - result = segment_image_dataset( - fake_generator, - images, - min_area=2, - max_masks=5, - points_per_batch=16, - ) - - assert len(result) == 2 - assert result[0][0]["area"] == 4 - assert result[1][0]["area"] == 9 diff --git a/mini-nav/tests/test_multi_object_retrieval.py b/mini-nav/tests/test_multi_object_retrieval.py deleted file mode 100644 index 7d71ac2..0000000 --- a/mini-nav/tests/test_multi_object_retrieval.py +++ /dev/null @@ -1,238 +0,0 @@ -"""Integration tests for multi-object retrieval benchmark pipeline. - -These tests verify the end-to-end functionality of the multi-object retrieval -benchmark, including schema building, database population, and evaluation. -""" - -import numpy as np -import pytest -from unittest.mock import Mock, patch, MagicMock -from PIL import Image - - -class TestMultiObjectRetrievalIntegration: - """Integration tests for multi-object retrieval benchmark.""" - - @pytest.fixture - def mock_model_processor(self): - """Create mock model and processor.""" - mock_model = Mock() - mock_processor = Mock() - - # Mock the feature extraction to return a fixed-size vector - def mock_extract(processor, model, image): - return [0.1] * 256 # 256-dim vector - mock_processor.images = mock_extract - - return mock_model, mock_processor - - @pytest.fixture - def mock_dataset(self): - """Create a mock dataset with images and annotations.""" - # Create mock items - items = [] - for i in range(3): - item = { - "image": Image.new("RGB", (224, 224), color=(i * 50, 100, 150)), - "image_id": f"scene_{i}", - "objects": { - "bbox": [[10, 10, 50, 50], [60, 60, 40, 40]], - "category": ["object_a", "object_b"], - "area": [2500, 1600], - "id": [0, 1], - }, - } - items.append(item) - - mock_dataset = Mock() - mock_dataset.__len__ = Mock(return_value=len(items)) - mock_dataset.__getitem__ = lambda self, idx: items[idx] - mock_dataset.with_format = lambda fmt: mock_dataset - - return mock_dataset - - def test_build_object_schema(self): - """Test that object schema is built correctly.""" - from benchmarks.tasks.multi_object_retrieval import _build_object_schema - import pyarrow as pa - - vector_dim = 256 - schema = _build_object_schema(vector_dim) - - assert isinstance(schema, pa.Schema) - assert "id" in schema.names - assert "image_id" in schema.names - assert "object_id" in schema.names - assert "category" in schema.names - assert "vector" in schema.names - - # Check vector field has correct dimension - vector_field = schema.field("vector") - assert isinstance(vector_field.type, pa.List) - assert vector_field.type.value_type == pa.float32() - - @patch("benchmarks.tasks.multi_object_retrieval.load_sam_model") - @patch("benchmarks.tasks.multi_object_retrieval.segment_image") - def test_build_database_with_mocked_sam( - self, - mock_segment, - mock_load_sam, - mock_model_processor, - mock_dataset, - ): - """Test database building with mocked SAM segmentation.""" - from benchmarks.tasks.multi_object_retrieval import ( - MultiObjectRetrievalTask, - _build_object_schema, - ) - - mock_model, mock_processor = mock_model_processor - - # Mock SAM - mock_load_sam.return_value = (Mock(), Mock()) - mock_segment.return_value = [ - { - "segment": np.ones((224, 224), dtype=bool), - "area": 50000, - "bbox": [0, 0, 224, 224], - } - ] - - # Create task with config - task = MultiObjectRetrievalTask( - sam_model="facebook/sam2.1-hiera-large", - min_mask_area=1024, - max_masks_per_image=5, - gamma=1.0, - top_k_per_object=50, - num_query_objects=3, - ) - - # Create mock table - mock_table = Mock() - mock_table.schema = _build_object_schema(256) - - # Build database (this should not raise) - task.build_database(mock_model, mock_processor, mock_dataset, mock_table, batch_size=1) - - # Verify table.add was called - assert mock_table.add.called - - @patch("benchmarks.tasks.multi_object_retrieval.load_sam_model") - @patch("benchmarks.tasks.multi_object_retrieval.segment_image") - def test_evaluate_with_mocked_sam( - self, - mock_segment, - mock_load_sam, - mock_model_processor, - mock_dataset, - ): - """Test evaluation with mocked SAM segmentation.""" - from benchmarks.tasks.multi_object_retrieval import ( - MultiObjectRetrievalTask, - _build_object_schema, - ) - - mock_model, mock_processor = mock_model_processor - - # Mock SAM - mock_load_sam.return_value = (Mock(), Mock()) - mock_segment.return_value = [ - { - "segment": np.ones((224, 224), dtype=bool), - "area": 50000, - "bbox": [0, 0, 224, 224], - "object_id": "query_obj_0", - } - ] - - # Create mock table with search results - mock_table = Mock() - mock_table.schema = _build_object_schema(256) - - # Mock search to return matching result - mock_result = Mock() - mock_result.to_polars.return_value = { - "image_id": ["scene_0"], - "object_id": ["scene_0_obj_0"], - "_distance": [0.1], - } - - mock_table.search.return_value.select.return_value.limit.return_value = mock_result - - # Create task - task = MultiObjectRetrievalTask( - sam_model="facebook/sam2.1-hiera-large", - min_mask_area=1024, - max_masks_per_image=5, - gamma=1.0, - top_k_per_object=50, - num_query_objects=1, - ) - - # Evaluate - results = task.evaluate(mock_model, mock_processor, mock_dataset, mock_table, batch_size=1) - - # Verify results structure - assert "accuracy" in results - assert "correct" in results - assert "total" in results - assert "top_k" in results - assert results["top_k"] == 50 - - def test_task_initialization_with_config(self): - """Test task initialization with custom config.""" - from benchmarks.tasks.multi_object_retrieval import MultiObjectRetrievalTask - - task = MultiObjectRetrievalTask( - sam_model="facebook/sam2.1-hiera-small", - min_mask_area=500, - max_masks_per_image=3, - gamma=0.5, - top_k_per_object=100, - num_query_objects=5, - ) - - assert task.sam_model == "facebook/sam2.1-hiera-small" - assert task.min_mask_area == 500 - assert task.max_masks_per_image == 3 - assert task.config.gamma == 0.5 - assert task.config.top_k_per_object == 100 - assert task.config.num_query_objects == 5 - - def test_task_initialization_defaults(self): - """Test task initialization with default config.""" - from benchmarks.tasks.multi_object_retrieval import MultiObjectRetrievalTask - - task = MultiObjectRetrievalTask() - - # Check defaults from BenchmarkTaskConfig - assert task.config.gamma == 1.0 - assert task.config.top_k_per_object == 50 - assert task.config.num_query_objects == 3 - # SAM settings from ModelConfig defaults - assert task.sam_model == "facebook/sam2.1-hiera-large" - assert task.min_mask_area == 1024 - assert task.max_masks_per_image == 5 - - -class TestInsDetScenesDataset: - """Tests for InsDetScenesDataset class.""" - - def test_dataset_class_exists(self): - """Test that InsDetScenesDataset can be imported.""" - from data_loading.insdet_scenes import InsDetScenesDataset - - assert InsDetScenesDataset is not None - - @patch("data_loading.insdet_scenes.load_val_dataset") - def test_dataset_loads_correct_split(self, mock_load): - """Test dataset loads correct split.""" - from data_loading.insdet_scenes import InsDetScenesDataset - - mock_load.return_value = Mock() - - dataset = InsDetScenesDataset("/path/to/scenes", split="easy") - - mock_load.assert_called_once_with("/path/to/scenes", "easy") - assert dataset.split == "easy" diff --git a/mini-nav/tests/test_object_score.py b/mini-nav/tests/test_object_score.py deleted file mode 100644 index 1c9121e..0000000 --- a/mini-nav/tests/test_object_score.py +++ /dev/null @@ -1,113 +0,0 @@ -import numpy as np - -from compressors.object_score import ( - MaskScoringConfig, - compute_mask_features, - rank_masks, - score_mask, - select_best_mask, -) - - -def _rect_mask(height: int, width: int, x: int, y: int, w: int, h: int) -> np.ndarray: - mask = np.zeros((height, width), dtype=bool) - mask[y : y + h, x : x + w] = True - return mask - - -def test_compute_mask_features_core_metrics() -> None: - mask = _rect_mask(height=20, width=20, x=5, y=4, w=6, h=5) - mask_dict = { - "segment": mask, - "area": int(mask.sum()), - "bbox": [5, 4, 6, 5], - "predicted_iou": 0.8, - "stability_score": 0.9, - } - - features = compute_mask_features(mask_dict, image_shape=(20, 20)) - - assert features.area_ratio == 30 / 400 - assert features.fill_ratio == 1.0 - assert features.aspect_ratio == 6 / 5 - assert features.touch_top is False - assert features.touch_left is False - assert features.num_components == 1 - assert features.largest_component_ratio == 1.0 - assert features.num_holes == 0 - - -def test_rank_masks_rejects_extreme_small_and_fragmented_masks() -> None: - cfg = MaskScoringConfig(min_area_ratio=0.02) - good_mask = _rect_mask(height=30, width=30, x=6, y=6, w=10, h=10) - - fragmented = np.zeros((30, 30), dtype=bool) - fragmented[2, 2] = True - fragmented[4, 7] = True - fragmented[8, 12] = True - fragmented[12, 16] = True - fragmented[16, 20] = True - fragmented[20, 24] = True - fragmented[24, 26] = True - - masks = [ - {"segment": np.zeros((30, 30), dtype=bool), "area": 1, "bbox": [0, 0, 1, 1]}, - { - "segment": fragmented, - "area": int(fragmented.sum()), - "bbox": [2, 2, 25, 25], - }, - { - "segment": good_mask, - "area": int(good_mask.sum()), - "bbox": [6, 6, 10, 10], - "predicted_iou": 0.9, - "stability_score": 0.9, - }, - ] - - ranked = rank_masks(masks=masks, image_shape=(30, 30), config=cfg, max_masks=3) - - assert len(ranked) == 1 - assert ranked[0]["area"] == int(good_mask.sum()) - assert "mask_score" in ranked[0] - - -def test_score_mask_prefers_stable_reasonable_object() -> None: - cfg = MaskScoringConfig() - - candidate = { - "segment": _rect_mask(height=100, width=100, x=30, y=20, w=24, h=25), - "area": 24 * 25, - "bbox": [30, 20, 24, 25], - "predicted_iou": 0.92, - "stability_score": 0.91, - } - weak = { - "segment": _rect_mask(height=100, width=100, x=0, y=0, w=4, h=60), - "area": 4 * 60, - "bbox": [0, 0, 4, 60], - "predicted_iou": 0.4, - "stability_score": 0.3, - } - - score_candidate = score_mask(candidate, image_shape=(100, 100), config=cfg) - score_weak = score_mask(weak, image_shape=(100, 100), config=cfg) - - assert score_candidate > score_weak - - -def test_select_best_mask_falls_back_to_largest_area_when_all_rejected() -> None: - cfg = MaskScoringConfig(min_area_ratio=0.2) - tiny = _rect_mask(height=20, width=20, x=1, y=1, w=2, h=2) - larger = _rect_mask(height=20, width=20, x=5, y=5, w=4, h=4) - - masks = [ - {"segment": tiny, "area": int(tiny.sum()), "bbox": [1, 1, 2, 2]}, - {"segment": larger, "area": int(larger.sum()), "bbox": [5, 5, 4, 4]}, - ] - - best = select_best_mask(masks=masks, image_shape=(20, 20), config=cfg) - - assert best is not None - assert best["area"] == int(larger.sum()) diff --git a/mini-nav/tests/test_room_views.py b/mini-nav/tests/test_room_views.py deleted file mode 100644 index f15a129..0000000 --- a/mini-nav/tests/test_room_views.py +++ /dev/null @@ -1,108 +0,0 @@ -from types import SimpleNamespace - -import pytest - -from simulator import collect_room_views_by_room - - -class _FakeAgent: - def __init__(self): - self.positions = [] - - def set_state(self, state): - self.positions.append(state.position) - - -class _FakeSimulator: - def __init__(self): - self._frame_index = 0 - self.actions = [] - - def get_sensor_observations(self): - observations = { - "color_sensor": f"frame_{self._frame_index}", - "depth_sensor": f"depth_{self._frame_index}", - } - self._frame_index += 1 - return observations - - def step(self, action_name): - self.actions.append(action_name) - - -class _FakeAgentState: - def __init__(self): - self.position = None - - -def test_collect_room_views_by_room_collects_grouped_frames_with_single_outer_progress(): - track_calls = [] - - def fake_track(iterable, description): - track_calls.append(description) - return iterable - - agent = _FakeAgent() - sim = _FakeSimulator() - room_nodes = [ - SimpleNamespace(room_id="room_00", center=[1.0, 2.0, 3.0]), - SimpleNamespace(room_id="room_01", center=[4.0, 5.0, 6.0]), - ] - fake_habitat = SimpleNamespace(AgentState=_FakeAgentState) - - room_views = collect_room_views_by_room( - agent=agent, - sim=sim, - room_nodes=room_nodes, - views_per_room=3, - habitat_sim_module=fake_habitat, - progress_track=fake_track, - ) - - assert track_calls == ["Collecting room views"] - assert room_views == { - "room_00": ["frame_0", "frame_1", "frame_2"], - "room_01": ["frame_3", "frame_4", "frame_5"], - } - assert sim.actions == [ - "turn_left", - "turn_left", - "turn_left", - "turn_left", - "turn_left", - "turn_left", - ] - assert agent.positions == [[1.0, 2.0, 3.0], [4.0, 5.0, 6.0]] - - -def test_collect_room_views_by_room_uses_custom_sensor_and_turn_action(): - agent = _FakeAgent() - sim = _FakeSimulator() - room_nodes = [SimpleNamespace(room_id="room_00", center=[0.0, 1.0, 0.0])] - fake_habitat = SimpleNamespace(AgentState=_FakeAgentState) - - room_views = collect_room_views_by_room( - agent=agent, - sim=sim, - room_nodes=room_nodes, - views_per_room=2, - habitat_sim_module=fake_habitat, - sensor_uuid="depth_sensor", - turn_action="turn_right", - progress_track=lambda iterable, description: iterable, - ) - - assert room_views == {"room_00": ["depth_0", "depth_1"]} - assert sim.actions == ["turn_right", "turn_right"] - - -def test_collect_room_views_by_room_validates_views_per_room(): - with pytest.raises(ValueError, match="views_per_room"): - collect_room_views_by_room( - agent=_FakeAgent(), - sim=_FakeSimulator(), - room_nodes=[SimpleNamespace(room_id="room_00", center=[0.0, 1.0, 0.0])], - views_per_room=0, - habitat_sim_module=SimpleNamespace(AgentState=_FakeAgentState), - progress_track=lambda iterable, description: iterable, - ) diff --git a/mini-nav/tests/test_sam.py b/mini-nav/tests/test_sam.py deleted file mode 100644 index 918c0a4..0000000 --- a/mini-nav/tests/test_sam.py +++ /dev/null @@ -1,205 +0,0 @@ -"""Tests for SAM segmentation utilities. - -Note: These tests mock the SAM model loading since SAM requires -heavy model weights. The actual SAM integration should be tested -separately in integration tests. -""" - -import numpy as np -import pytest -from unittest.mock import Mock, patch -from PIL import Image - - -class TestSAMSegmentation: - """Test suite for SAM segmentation utilities.""" - - def test_segment_image_empty_masks(self): - """Test segment_image returns empty list when no masks generated.""" - from utils.sam import segment_image - - # Create mock mask generator that returns empty list - mock_generator = Mock() - mock_generator.generate.return_value = [] - - result = segment_image(mock_generator, Image.new("RGB", (100, 100))) - - assert result == [] - - def test_segment_image_filters_small_masks(self): - """Test segment_image filters masks below min_area threshold.""" - from utils.sam import segment_image - - # Create mock masks with different areas - small_mask = { - "segment": np.zeros((10, 10), dtype=bool), - "area": 50, # Below 32*32 = 1024 - "bbox": [0, 0, 10, 10], - "predicted_iou": 0.9, - "stability_score": 0.8, - } - large_mask = { - "segment": np.ones((100, 100), dtype=bool), - "area": 10000, # Above threshold - "bbox": [0, 0, 100, 100], - "predicted_iou": 0.95, - "stability_score": 0.9, - } - - mock_generator = Mock() - mock_generator.generate.return_value = [small_mask, large_mask] - - result = segment_image( - mock_generator, - Image.new("RGB", (100, 100)), - min_area=32 * 32, - max_masks=5, - ) - - # Should only return the large mask - assert len(result) == 1 - assert result[0]["area"] == 10000 - - def test_segment_image_limits_max_masks(self): - """Test segment_image limits to max_masks largest masks.""" - from utils.sam import segment_image - - # Create 10 masks with different areas - masks = [ - { - "segment": np.ones((i + 1, i + 1), dtype=bool), - "area": (i + 1) * (i + 1), - "bbox": [0, 0, i + 1, i + 1], - "predicted_iou": 0.9, - "stability_score": 0.8, - } - for i in range(10) - ] - - mock_generator = Mock() - mock_generator.generate.return_value = masks - - result = segment_image( - mock_generator, - Image.new("RGB", (100, 100)), - min_area=1, - max_masks=3, - ) - - # Should only return top 3 largest masks - assert len(result) == 3 - # Check they are sorted by area (largest first) - areas = [m["area"] for m in result] - assert areas == sorted(areas, reverse=True) - - def test_segment_image_sorted_by_area(self): - """Test segment_image returns masks sorted by area descending.""" - from utils.sam import segment_image - - # Create masks with known areas (unordered) - mask1 = { - "segment": np.ones((5, 5), dtype=bool), - "area": 25, - "bbox": [0, 0, 5, 5], - } - mask2 = { - "segment": np.ones((10, 10), dtype=bool), - "area": 100, - "bbox": [0, 0, 10, 10], - } - mask3 = { - "segment": np.ones((3, 3), dtype=bool), - "area": 9, - "bbox": [0, 0, 3, 3], - } - - mock_generator = Mock() - mock_generator.generate.return_value = [mask1, mask2, mask3] - - result = segment_image( - mock_generator, - Image.new("RGB", (100, 100)), - min_area=1, - max_masks=10, - ) - - # Should be sorted by area descending - assert result[0]["area"] == 100 - assert result[1]["area"] == 25 - assert result[2]["area"] == 9 - - -class TestSAMLoading: - @patch("utils.sam.pipeline") - def test_load_sam_model_uses_transformers_pipeline(self, mock_pipeline): - from utils.sam import Sam2MaskGenerator, load_sam_model - - mock_pipe_obj = Mock() - mock_pipe_obj.model = Mock() - mock_pipeline.return_value = mock_pipe_obj - - sam_model, mask_generator = load_sam_model( - model_name="facebook/sam2.1-hiera-large", - device="cpu", - points_per_batch=16, - ) - - assert sam_model is mock_pipe_obj.model - assert isinstance(mask_generator, Sam2MaskGenerator) - assert mask_generator.points_per_batch == 16 - - _, kwargs = mock_pipeline.call_args - assert kwargs["task"] == "mask-generation" - assert kwargs["model"] == "facebook/sam2.1-hiera-large" - assert kwargs["device"] == -1 - - -class TestExtractMaskedRegion: - """Test suite for extracting masked regions from images.""" - - def test_extract_masked_region_binary(self): - """Test extracting masked region with binary mask.""" - from utils.sam import extract_masked_region - - # Create a simple image - image = Image.new("RGB", (10, 10), color=(255, 0, 0)) - - # Create a binary mask (half kept, half masked) - mask = np.zeros((10, 10), dtype=bool) - mask[:, :5] = True - - result = extract_masked_region(image, mask) - - # Check that left half is red, right half is black - result_np = np.array(result) - left_half = result_np[:, :5, :] - right_half = result_np[:, 5:, :] - - assert np.all(left_half == [255, 0, 0]) - assert np.all(right_half == [0, 0, 0]) - - def test_extract_masked_region_all_masked(self): - """Test extracting when entire image is masked.""" - from utils.sam import extract_masked_region - - image = Image.new("RGB", (10, 10), color=(255, 0, 0)) - mask = np.ones((10, 10), dtype=bool) - - result = extract_masked_region(image, mask) - result_np = np.array(result) - - # Entire image should be preserved - assert np.all(result_np == [255, 0, 0]) - - def test_extract_masked_region_all_zero_mask(self): - """Test extracting when mask is all zeros.""" - from utils.sam import extract_masked_region - - image = Image.new("RGB", (10, 10), color=(255, 0, 0)) - mask = np.zeros((10, 10), dtype=bool) - - result = extract_masked_region(image, mask) - result_np = np.array(result) - - # Entire image should be black - assert np.all(result_np == [0, 0, 0]) diff --git a/mini-nav/tests/test_scene_scoring.py b/mini-nav/tests/test_scene_scoring.py deleted file mode 100644 index de12468..0000000 --- a/mini-nav/tests/test_scene_scoring.py +++ /dev/null @@ -1,121 +0,0 @@ -"""Tests for scene scoring algorithm in multi-object retrieval.""" - -import pytest -from benchmarks.tasks.multi_object_retrieval import _compute_scene_score - - -class TestSceneScoringAlgorithm: - """Test suite for scene scoring with co-occurrence penalty.""" - - def test_scene_score_basic(self): - """Test basic scene scoring with single match.""" - query_object_ids = ["obj_1", "obj_2", "obj_3"] - - # Scene A has obj_1 - retrieved_results = { - "scene_A": [("distance_1", "obj_1")], - } - - scores = _compute_scene_score(query_object_ids, retrieved_results, gamma=1.0) - - # Hit rate = 1/3, similarity = 1/(1+distance_1) - assert "scene_A" in scores - assert scores["scene_A"] > 0 - - def test_scene_score_no_match(self): - """Test scene scoring when no objects match.""" - query_object_ids = ["obj_1", "obj_2", "obj_3"] - - retrieved_results = { - "scene_A": [("distance_1", "other_obj")], - } - - scores = _compute_scene_score(query_object_ids, retrieved_results, gamma=1.0) - - assert scores["scene_A"] == 0.0 - - def test_scene_score_multiple_scenes(self): - """Test scoring across multiple scenes.""" - query_object_ids = ["obj_1", "obj_2"] - - retrieved_results = { - "scene_A": [("0.1", "obj_1")], - "scene_B": [("0.1", "obj_2")], - "scene_C": [("0.1", "other")], - } - - scores = _compute_scene_score(query_object_ids, retrieved_results, gamma=1.0) - - # Scenes with matches should have positive scores - assert scores["scene_A"] > 0 - assert scores["scene_B"] > 0 - # Scene C has no match, score should be 0 - assert scores["scene_C"] == 0.0 - - def test_scene_score_gamma_zero(self): - """Test scoring with gamma=0 (no penalty).""" - query_object_ids = ["obj_1", "obj_2", "obj_3", "obj_4", "obj_5"] - - retrieved_results = { - "scene_A": [("0.1", "obj_1")], - } - - scores_gamma_0 = _compute_scene_score(query_object_ids, retrieved_results, gamma=0.0) - scores_gamma_1 = _compute_scene_score(query_object_ids, retrieved_results, gamma=1.0) - - # With gamma=0, hit_rate^0 = 1, so score = similarity - # With gamma=1, hit_rate^1 = 1/5, so score = similarity * 1/5 - # scores_gamma_0 should be larger - assert scores_gamma_0["scene_A"] > scores_gamma_1["scene_A"] - - def test_scene_score_multiple_matches(self): - """Test scoring when scene has multiple matching objects.""" - query_object_ids = ["obj_1", "obj_2"] - - retrieved_results = { - "scene_A": [("0.1", "obj_1"), ("0.2", "obj_2")], - } - - scores = _compute_scene_score(query_object_ids, retrieved_results, gamma=1.0) - - # Both objects match, hit_rate = 2/2 = 1.0 - # Score = (1/(1+0.1) + 1/(1+0.2)) * 1.0 - expected_similarity = 1.0 / 1.1 + 1.0 / 1.2 - assert abs(scores["scene_A"] - expected_similarity) < 0.01 - - def test_scene_score_distance_to_similarity(self): - """Test that smaller distance yields higher score.""" - query_object_ids = ["obj_1"] - - retrieved_results = { - "scene_close": [("0.01", "obj_1")], - "scene_far": [("10.0", "obj_1")], - } - - scores = _compute_scene_score(query_object_ids, retrieved_results, gamma=1.0) - - # Closer scene should have higher score - assert scores["scene_close"] > scores["scene_far"] - - def test_scene_score_empty_results(self): - """Test scoring with empty retrieved results.""" - query_object_ids = ["obj_1", "obj_2"] - - retrieved_results = {} - - scores = _compute_scene_score(query_object_ids, retrieved_results, gamma=1.0) - - assert scores == {} - - def test_scene_score_empty_query(self): - """Test scoring with empty query objects.""" - query_object_ids = [] - - retrieved_results = { - "scene_A": [("0.1", "obj_1")], - } - - scores = _compute_scene_score(query_object_ids, retrieved_results, gamma=1.0) - - # With empty query, no scenes should have positive score - assert all(score == 0.0 for score in scores.values()) diff --git a/mini-nav/tests/test_topdown_map.py b/mini-nav/tests/test_topdown_map.py deleted file mode 100644 index 931b6e9..0000000 --- a/mini-nav/tests/test_topdown_map.py +++ /dev/null @@ -1,126 +0,0 @@ -from types import SimpleNamespace - -import pytest - -from simulator import ( - TopDownRenderStyle, - TopDownSceneElements, - render_topdown_scene_map, -) - - -class _FakeMaps: - def __init__(self): - self.to_grid_calls: list[tuple[float, float]] = [] - - def get_topdown_map(self, pathfinder, height, meters_per_pixel): - return [[0, 0], [0, 0]] - - def to_grid(self, z, x, shape, pathfinder): - self.to_grid_calls.append((z, x)) - return (int(z), int(x)) - - -class _FakePlt: - def __init__(self): - self.scatter_calls: list[tuple[int, int]] = [] - self.text_calls: list[str] = [] - self.shown = False - - def figure(self, figsize): - return None - - def imshow(self, image, cmap): - return None - - def scatter(self, x, y, c, s): - self.scatter_calls.append((x, y)) - - def text(self, x, y, text, color, fontsize): - self.text_calls.append(text) - - def title(self, title): - return None - - def axis(self, mode): - return None - - def show(self): - self.shown = True - - -def test_render_topdown_scene_map_renders_room_nodes_only(): - fake_maps = _FakeMaps() - fake_plt = _FakePlt() - room_nodes = [ - SimpleNamespace(room_id="room_00", center=[1.0, 2.0, 3.0]), - SimpleNamespace(room_id="room_01", center=[4.0, 2.0, 5.0]), - ] - elements = TopDownSceneElements(room_nodes=room_nodes) - - top_down_map = render_topdown_scene_map( - pathfinder=SimpleNamespace(), - elements=elements, - meters_per_pixel=0.05, - style=TopDownRenderStyle(), - maps_module=fake_maps, - plt_module=fake_plt, - ) - - assert top_down_map == [[0, 0], [0, 0]] - assert fake_maps.to_grid_calls == [(3.0, 1.0), (5.0, 4.0)] - assert fake_plt.scatter_calls == [(1, 3), (4, 5)] - assert fake_plt.text_calls == ["room_00", "room_01"] - assert fake_plt.shown is True - - -def test_render_topdown_scene_map_validates_room_nodes(): - with pytest.raises(ValueError, match="room_nodes"): - render_topdown_scene_map( - pathfinder=SimpleNamespace(), - elements=TopDownSceneElements(room_nodes=[]), - meters_per_pixel=0.05, - maps_module=_FakeMaps(), - plt_module=_FakePlt(), - ) - - -def test_render_topdown_scene_map_validates_meters_per_pixel(): - with pytest.raises(ValueError, match="meters_per_pixel"): - render_topdown_scene_map( - pathfinder=SimpleNamespace(), - elements=TopDownSceneElements( - room_nodes=[SimpleNamespace(room_id="room_00", center=[0.0, 1.0, 0.0])] - ), - meters_per_pixel=0, - maps_module=_FakeMaps(), - plt_module=_FakePlt(), - ) - - -def test_render_topdown_scene_map_rejects_object_nodes_before_implementation(): - with pytest.raises(NotImplementedError, match="object_nodes"): - render_topdown_scene_map( - pathfinder=SimpleNamespace(), - elements=TopDownSceneElements( - room_nodes=[SimpleNamespace(room_id="room_00", center=[0.0, 1.0, 0.0])], - object_nodes=[SimpleNamespace(obj_id="obj_00")], - ), - meters_per_pixel=0.05, - maps_module=_FakeMaps(), - plt_module=_FakePlt(), - ) - - -def test_render_topdown_scene_map_rejects_edges_before_implementation(): - with pytest.raises(NotImplementedError, match="edge"): - render_topdown_scene_map( - pathfinder=SimpleNamespace(), - elements=TopDownSceneElements( - room_nodes=[SimpleNamespace(room_id="room_00", center=[0.0, 1.0, 0.0])], - edges=[("room_00", "obj_00")], - ), - meters_per_pixel=0.05, - maps_module=_FakeMaps(), - plt_module=_FakePlt(), - ) diff --git a/mini-nav/tests/test_visualizer.py b/mini-nav/tests/test_visualizer.py deleted file mode 100644 index 8b63dd2..0000000 --- a/mini-nav/tests/test_visualizer.py +++ /dev/null @@ -1,77 +0,0 @@ -"""Tests for visualizer app image upload similarity search.""" - -import base64 -import io - -import numpy as np -from PIL import Image -from sklearn.metrics.pairwise import cosine_similarity - - -class TestImageUploadSimilaritySearch: - """Test suite for image upload similarity search functionality.""" - - def test_base64_to_pil_image(self): - """Test conversion from base64 string to PIL Image.""" - # Create a test image - img_array = np.random.randint(0, 255, (224, 224, 3), dtype=np.uint8) - img = Image.fromarray(img_array) - - # Convert to base64 - buffer = io.BytesIO() - img.save(buffer, format="PNG") - img_base64 = base64.b64encode(buffer.getvalue()).decode("utf-8") - - # Add data URI prefix (as Dash provides) - img_base64_with_prefix = f"data:image/png;base64,{img_base64}" - - # Parse base64 to PIL Image - # Remove prefix - base64_str = img_base64_with_prefix.split(",")[1] - img_bytes = base64.b64decode(base64_str) - parsed_img = Image.open(io.BytesIO(img_bytes)) - - # Verify the image is valid - assert parsed_img.size == (224, 224) - assert parsed_img.mode == "RGB" - - -class TestCosineSimilarity: - """Test suite for cosine similarity computation between feature vectors.""" - - def test_identical_vectors_return_one(self): - """Identical vectors should have cosine similarity of 1.0.""" - vec = np.random.randn(1024).tolist() - similarity = cosine_similarity([vec], [vec])[0][0] - assert np.isclose(similarity, 1.0) - - def test_orthogonal_vectors_return_zero(self): - """Orthogonal vectors should have cosine similarity of 0.0.""" - vec_a = [1.0, 0.0] - vec_b = [0.0, 1.0] - similarity = cosine_similarity([vec_a], [vec_b])[0][0] - assert np.isclose(similarity, 0.0) - - def test_opposite_vectors_return_negative_one(self): - """Opposite vectors should have cosine similarity of -1.0.""" - vec_a = [1.0, 0.0, 0.0] - vec_b = [-1.0, 0.0, 0.0] - similarity = cosine_similarity([vec_a], [vec_b])[0][0] - assert np.isclose(similarity, -1.0) - - def test_similarity_range(self): - """Cosine similarity should always be within [-1, 1].""" - # Random vectors - for _ in range(10): - vec_a = np.random.randn(1024).tolist() - vec_b = np.random.randn(1024).tolist() - similarity = cosine_similarity([vec_a], [vec_b])[0][0] - assert -1.0 <= similarity <= 1.0 - - def test_similarity_with_list_input(self): - """Cosine similarity should work with Python list inputs (as stored in dcc.Store).""" - # Simulate feature vectors stored as Python lists in dcc.Store - vec_a = [0.1, 0.2, 0.3, 0.4, 0.5] - vec_b = [0.1, 0.2, 0.3, 0.4, 0.5] - similarity = cosine_similarity([vec_a], [vec_b])[0][0] - assert np.isclose(similarity, 1.0)