gps-denied-onboard/test_f13_result_manager.py

import pytest
pytest.skip("Obsolete test file replaced by test_f14_result_manager", allow_module_level=True)
import os
import json
import csv

from f13_result_manager import ResultManager, ResultData, GPSPoint

@pytest.fixture
def temp_result_manager(tmp_path):
    """Creates a ResultManager with a temporary isolated SQLite database."""
    db_path = tmp_path / "test_results.db"
    return ResultManager(db_path=str(db_path))

def test_store_initial_result(temp_result_manager):
    """Test Case 1: Store Initial Result from the Vision Pipeline."""
    res = ResultData(
        flight_id="Test_Flight_001",
        image_id="AD000001.jpg",
        sequence_number=1,
        estimated_gps=GPSPoint(lat=48.275290, lon=37.385218),
        ground_truth_gps=GPSPoint(lat=48.275292, lon=37.385220),
        confidence=0.92,
        source="L3",
        processing_time_ms=450,
        metadata={"layer": "L3", "inlier_ratio": 0.85}
    )
    
    stored = temp_result_manager.store_result(res)
    
    assert stored.version == 1
    assert stored.error_m is not None
    assert stored.error_m > 0.0  # Haversine distance should be calculated
    
    retrieved = temp_result_manager.get_result("Test_Flight_001", "AD000001.jpg")
    assert retrieved is not None
    assert retrieved.version == 1
    assert retrieved.confidence == 0.92
    assert retrieved.metadata["layer"] == "L3"

def test_store_refined_result_versions(temp_result_manager):
    """Test Case 2 & 12: Store Refined Result and verify versioning history (AC-8)."""
    # 1. Initial Estimate
    res1 = ResultData(
        flight_id="Test_Flight_001",
        image_id="AD000001.jpg",
        sequence_number=1,
        estimated_gps=GPSPoint(lat=48.275290, lon=37.385218),
        confidence=0.5,
        source="vo_frontend"
    )
    temp_result_manager.store_result(res1)
    
    # 2. Refined Estimate (e.g., after Factor Graph convergence)
    res2 = ResultData(
        flight_id="Test_Flight_001",
        image_id="AD000001.jpg",
        sequence_number=1,
        estimated_gps=GPSPoint(lat=48.275291, lon=37.385219),
        confidence=0.95,
        source="factor_graph",
        refinement_reason="new_anchor"
    )
    stored2 = temp_result_manager.store_result(res2)
    
    assert stored2.version == 2
    
    # Verify fetching latest defaults to version 2
    latest = temp_result_manager.get_result("Test_Flight_001", "AD000001.jpg")
    assert latest.version == 2
    assert latest.source == "factor_graph"
    
    # Verify history contains both
    history = temp_result_manager.get_result_history("Test_Flight_001", "AD000001.jpg")
    assert len(history) == 2
    assert history[0].version == 1
    assert history[1].version == 2

def test_batch_store_and_retrieve_flight(temp_result_manager):
    """Test Case 3 & 5: Batch Store Results and Retrieve Flight Results."""
    batch = []
    for i in range(1, 11):
        batch.append(ResultData(
            flight_id="Test_Flight_002",
            image_id=f"AD{i:06d}.jpg",
            sequence_number=i,
            estimated_gps=GPSPoint(lat=48.0 + (i*0.001), lon=37.0),
            confidence=0.9
        ))
        
    temp_result_manager.store_results_batch(batch)
    
    results = temp_result_manager.get_flight_results("Test_Flight_002")
    assert len(results) == 10
    assert results[0].sequence_number == 1
    assert results[-1].sequence_number == 10

def test_retrieve_with_filtering(temp_result_manager):
    """Test Case 6: Retrieve with Filtering by confidence and error."""
    temp_result_manager.store_result(ResultData(
        flight_id="Test_Flight_003", image_id="1.jpg", sequence_number=1,
        estimated_gps=GPSPoint(lat=48.0, lon=37.0), ground_truth_gps=GPSPoint(lat=48.0, lon=37.0),
        confidence=0.95
    )) # Perfect match, high confidence
    
    temp_result_manager.store_result(ResultData(
        flight_id="Test_Flight_003", image_id="2.jpg", sequence_number=2,
        estimated_gps=GPSPoint(lat=48.0, lon=37.0), ground_truth_gps=GPSPoint(lat=48.1, lon=37.1),
        confidence=0.4
    )) # High error, low confidence
    
    # Filter by confidence > 0.9
    high_conf = temp_result_manager.get_flight_results("Test_Flight_003", min_confidence=0.9)
    assert len(high_conf) == 1
    assert high_conf[0].image_id == "1.jpg"
    
    # Filter by max error < 50m
    low_err = temp_result_manager.get_flight_results("Test_Flight_003", max_error=50.0)
    assert len(low_err) == 1
    assert low_err[0].image_id == "1.jpg"

def test_store_user_fix(temp_result_manager):
    """Test Case 10: Store User Fix (AC-6)."""
    fix = temp_result_manager.store_user_fix(
        flight_id="Test_Flight_001",
        image_id="AD000005.jpg",
        sequence_number=5,
        gps=GPSPoint(lat=48.273997, lon=37.379828)
    )
    
    assert fix.source == "user"
    assert fix.confidence == 1.0
    assert fix.refinement_reason == "Manual User Fix"
    
    retrieved = temp_result_manager.get_result("Test_Flight_001", "AD000005.jpg")
    assert retrieved.source == "user"
    assert retrieved.confidence == 1.0

def test_calculate_statistics(temp_result_manager):
    """Test Case 11: Calculate Statistics for AC-1 / AC-2 validation."""
    # Add simulated results with varying errors
    for i in range(10):
        # Shift estimated gps slightly to create an error
        offset = i * 0.0001
        temp_result_manager.store_result(ResultData(
            flight_id="Test_Flight_Stats",
            image_id=f"{i}.jpg",
            sequence_number=i,
            estimated_gps=GPSPoint(lat=48.0 + offset, lon=37.0),
            ground_truth_gps=GPSPoint(lat=48.0, lon=37.0),
            confidence=0.9
        ))
        
    stats = temp_result_manager.calculate_statistics("Test_Flight_Stats", total_flight_images=12)
    
    assert stats is not None
    assert stats.total_images == 12
    assert stats.processed_images == 10
    assert stats.registration_rate == pytest.approx((10/12)*100.0)
    assert stats.mean_error_m > 0
    assert stats.rmse_m > 0
    assert 0.0 <= stats.percent_under_20m <= 100.0
    assert 0.0 <= stats.percent_under_50m <= 100.0

def test_export_formats(temp_result_manager, tmp_path):
    """Test Case 7, 8, 9: Export to JSON, CSV, KML."""
    flight_id = "Test_Export"
    temp_result_manager.store_result(ResultData(
        flight_id=flight_id, image_id="1.jpg", sequence_number=1,
        estimated_gps=GPSPoint(lat=48.1, lon=37.1), confidence=0.9
    ))
    
    # JSON Export
    json_path = str(tmp_path / "export.json")
    temp_result_manager.export_results(flight_id, format="json", filepath=json_path)
    assert os.path.exists(json_path)
    with open(json_path, 'r') as f:
        data = json.load(f)
        assert data["flight_id"] == flight_id
        assert len(data["results"]) == 1
        
    # CSV Export
    csv_path = str(tmp_path / "export.csv")
    temp_result_manager.export_results(flight_id, format="csv", filepath=csv_path)
    assert os.path.exists(csv_path)
    with open(csv_path, 'r') as f:
        reader = csv.reader(f)
        assert "image" in next(reader) # header
        
    # KML Export
    kml_path = str(tmp_path / "export.kml")
    temp_result_manager.export_results(flight_id, format="kml", filepath=kml_path)
    assert os.path.exists(kml_path)