feat: stage6 — Image Pipeline (F05) and Rotation Manager (F06)

tests/test_pipeline.py

@@ -0,0 +1,82 @@
+"""Tests for Image Input Pipeline (F05)."""
+
+import asyncio
+
+import cv2
+import numpy as np
+import pytest
+
+from gps_denied.core.pipeline import ImageInputPipeline, QueueFullError, ValidationError
+from gps_denied.schemas.image import ImageBatch
+
+
+@pytest.fixture
+def pipeline(tmp_path):
+    storage = str(tmp_path / "images")
+    return ImageInputPipeline(storage_dir=storage, max_queue_size=2)
+
+
+def test_batch_validation(pipeline):
+    # Too few images
+    b1 = ImageBatch(images=[b"1", b"2"], filenames=["1.jpg", "2.jpg"], start_sequence=1, end_sequence=2, batch_number=1)
+    val = pipeline.validate_batch(b1)
+    assert not val.valid
+    assert "Batch is empty" in val.errors
+
+    # Let's mock a valid batch of 10 images
+    fake_imgs = [b"fake"] * 10
+    fake_names = [f"AD{i:06d}.jpg" for i in range(1, 11)]
+    b2 = ImageBatch(images=fake_imgs, filenames=fake_names, start_sequence=1, end_sequence=10, batch_number=1)
+    val2 = pipeline.validate_batch(b2)
+    assert val2.valid
+
+
+@pytest.mark.asyncio
+async def test_queue_and_process(pipeline):
+    flight_id = "test_f1"
+    
+    # Create valid fake images
+    fake_img_np = np.zeros((10, 10, 3), dtype=np.uint8)
+    _, encoded = cv2.imencode(".jpg", fake_img_np)
+    fake_bytes = encoded.tobytes()
+    
+    fake_imgs = [fake_bytes] * 10
+    fake_names = [f"AD{i:06d}.jpg" for i in range(1, 11)]
+    b = ImageBatch(images=fake_imgs, filenames=fake_names, start_sequence=1, end_sequence=10, batch_number=1)
+    
+    pipeline.queue_batch(flight_id, b)
+    
+    # Process
+    processed = await pipeline.process_next_batch(flight_id)
+    assert processed is not None
+    assert len(processed.images) == 10
+    assert processed.images[0].sequence == 1
+    assert processed.images[-1].sequence == 10
+    
+    # Status
+    st = pipeline.get_processing_status(flight_id)
+    assert st.total_images == 10
+    assert st.processed_images == 10
+    
+    # Sequential get
+    next_img = pipeline.get_next_image(flight_id)
+    assert next_img is not None
+    assert next_img.sequence == 1
+    
+    # Second get
+    next_img2 = pipeline.get_next_image(flight_id)
+    assert next_img2 is not None
+    assert next_img2.sequence == 2
+
+
+def test_queue_full(pipeline):
+    flight_id = "test_full"
+    fake_imgs = [b"fake"] * 10
+    fake_names = [f"AD{i:06d}.jpg" for i in range(1, 11)]
+    b = ImageBatch(images=fake_imgs, filenames=fake_names, start_sequence=1, end_sequence=10, batch_number=1)
+    
+    pipeline.queue_batch(flight_id, b)
+    pipeline.queue_batch(flight_id, b)
+    
+    with pytest.raises(QueueFullError):
+        pipeline.queue_batch(flight_id, b)

tests/test_rotation.py

@@ -0,0 +1,88 @@
+"""Tests for Image Rotation Manager (F06)."""
+
+from datetime import datetime, timezone
+
+import numpy as np
+import pytest
+
+from gps_denied.core.rotation import IImageMatcher, ImageRotationManager
+from gps_denied.schemas.rotation import RotationResult
+from gps_denied.schemas.satellite import TileBounds
+from gps_denied.schemas import GPSPoint
+
+
+@pytest.fixture
+def rotation_manager():
+    return ImageRotationManager()
+
+
+def test_rotate_image_360(rotation_manager):
+    img = np.zeros((100, 100, 3), dtype=np.uint8)
+    
+    # Just draw a white rectangle to test rotation
+    img[10:40, 10:40] = [255, 255, 255]
+    
+    r90 = rotation_manager.rotate_image_360(img, 90.0)
+    assert r90.shape == (100, 100, 3)
+    
+    # Top left corner should move
+    assert not np.array_equal(img, r90)
+
+
+def test_heading_management(rotation_manager):
+    fid = "flight_1"
+    now = datetime.now(timezone.utc)
+    
+    assert rotation_manager.get_current_heading(fid) is None
+    
+    rotation_manager.update_heading(fid, 1, 370.0, now)  # should modulo to 10
+    assert rotation_manager.get_current_heading(fid) == 10.0
+    
+    rotation_manager.update_heading(fid, 2, 90.0, now)
+    assert rotation_manager.get_current_heading(fid) == 90.0
+
+
+def test_detect_sharp_turn(rotation_manager):
+    fid = "flight_2"
+    now = datetime.now(timezone.utc)
+    
+    assert rotation_manager.detect_sharp_turn(fid, 90.0) is False  # no history
+    
+    rotation_manager.update_heading(fid, 1, 90.0, now)
+    
+    assert rotation_manager.detect_sharp_turn(fid, 100.0) is False  # delta 10
+    assert rotation_manager.detect_sharp_turn(fid, 180.0) is True   # delta 90
+    assert rotation_manager.detect_sharp_turn(fid, 350.0) is True   # delta 100
+    assert rotation_manager.detect_sharp_turn(fid, 80.0) is False   # delta 10 (wraparound)
+    
+    # Wraparound test explicitly
+    rotation_manager.update_heading(fid, 2, 350.0, now)
+    assert rotation_manager.detect_sharp_turn(fid, 10.0) is False   # delta 20
+
+
+class MockMatcher(IImageMatcher):
+    def align_to_satellite(self, uav_image: np.ndarray, satellite_tile: np.ndarray, tile_bounds: TileBounds) -> RotationResult:
+        # Mock that only matches when angle was originally 90
+        # By checking internal state or just returning generic true for test
+        return RotationResult(matched=True, initial_angle=90.0, precise_angle=90.0, confidence=0.99)
+
+
+def test_try_rotation_steps(rotation_manager):
+    fid = "flight_3"
+    img = np.zeros((10, 10, 3), dtype=np.uint8)
+    sat = np.zeros((10, 10, 3), dtype=np.uint8)
+    
+    nw = GPSPoint(lat=10.0, lon=10.0)
+    se = GPSPoint(lat=9.0, lon=11.0)
+    tb = TileBounds(nw=nw, ne=nw, sw=se, se=se, center=nw, gsd=0.5)
+    
+    matcher = MockMatcher()
+    
+    # Should perform sweep and mock matcher says matched=True immediately in the loop
+    res = rotation_manager.try_rotation_steps(fid, 1, img, sat, tb, datetime.now(timezone.utc), matcher)
+    
+    assert res is not None
+    assert res.matched is True
+    # The first step is 0 degrees, the mock matcher returns matched=True.
+    # Therefore the first matched angle is 0.
+    assert res.initial_angle == 0.0