| """ |
| Test script for the Depth Pro Distance Estimation FastAPI app. |
| """ |
|
|
| import requests |
| import numpy as np |
| from PIL import Image |
| import io |
| import tempfile |
|
|
| def create_test_image(): |
| """Create a simple test image""" |
| |
| width, height = 512, 384 |
| image = np.zeros((height, width, 3), dtype=np.uint8) |
| |
| |
| for y in range(height): |
| intensity = int(255 * (1 - y / height)) |
| image[y, :, :] = [intensity, intensity//2, intensity//3] |
| |
| |
| image[50:60, :, :] = [255, 255, 255] |
| image[height-60:height-50, :, :] = [255, 255, 255] |
| |
| return Image.fromarray(image) |
|
|
| def test_web_interface(): |
| """Test the web interface (HTML page)""" |
| try: |
| |
| response = requests.get('http://localhost:7860/', timeout=10) |
| |
| if response.status_code == 200: |
| content = response.text |
| if "Depth Pro Distance Estimation" in content and "upload" in content.lower(): |
| print("Web Interface Test:") |
| print("Status Code:", response.status_code) |
| print("Content-Type:", response.headers.get('content-type', 'N/A')) |
| print("Page Title Found: β
") |
| print("Upload Form Found: β
") |
| print("β
Web interface test passed!") |
| return True |
| else: |
| print("β Web interface content validation failed") |
| return False |
| else: |
| print(f"β Web interface test failed with status code: {response.status_code}") |
| return False |
| |
| except requests.ConnectionError: |
| print("β οΈ FastAPI server not running. Start the server with: python app.py") |
| return False |
| except Exception as e: |
| print(f"β Web interface test failed: {e}") |
| return False |
|
|
| def test_fastapi_endpoint(): |
| """Test the FastAPI endpoint (requires running server)""" |
| try: |
| |
| test_image = create_test_image() |
| |
| |
| img_byte_arr = io.BytesIO() |
| test_image.save(img_byte_arr, format='JPEG') |
| img_byte_arr.seek(0) |
| |
| |
| files = {'file': ('test_image.jpg', img_byte_arr, 'image/jpeg')} |
| response = requests.post('http://localhost:7860/estimate-depth', files=files, timeout=30) |
| |
| if response.status_code == 200: |
| result = response.json() |
| print("FastAPI Endpoint Test:") |
| print("Status Code:", response.status_code) |
| print("Response:", result) |
| print("β
FastAPI endpoint test passed!") |
| return True |
| else: |
| print(f"β FastAPI endpoint test failed with status code: {response.status_code}") |
| print("Response:", response.text) |
| return False |
| |
| except requests.ConnectionError: |
| print("β οΈ FastAPI server not running. Start the server with: python app.py") |
| return False |
| except Exception as e: |
| print(f"β FastAPI endpoint test failed: {e}") |
| return False |
|
|
| def test_depth_estimator(): |
| """Test the DepthEstimator class directly""" |
| try: |
| from app import DepthEstimator, DummyDepthPipeline |
| |
| |
| dummy_pipeline = DummyDepthPipeline() |
| estimator = DepthEstimator(dummy_pipeline) |
| |
| |
| test_image = create_test_image() |
| with tempfile.NamedTemporaryFile(suffix='.jpg', delete=False) as temp_file: |
| test_image.save(temp_file, format='JPEG') |
| temp_file_path = temp_file.name |
| |
| |
| depth_map, new_size, focal_length = estimator.estimate_depth(temp_file_path) |
| |
| print("Depth Estimator Test:") |
| print("Depth map shape:", depth_map.shape if depth_map is not None else "None") |
| print("New size:", new_size) |
| print("Focal length:", focal_length) |
| |
| if depth_map is not None: |
| print("Depth stats:", { |
| "min": np.min(depth_map), |
| "max": np.max(depth_map), |
| "mean": np.mean(depth_map) |
| }) |
| print("β
Depth estimator test passed!") |
| return True |
| else: |
| print("β Depth estimator returned None") |
| return False |
| |
| except Exception as e: |
| print(f"β Depth estimator test failed: {e}") |
| return False |
|
|
| if __name__ == "__main__": |
| print("π§ͺ Testing Depth Pro Distance Estimation App\n") |
| |
| |
| tests = [ |
| ("Depth Estimator", test_depth_estimator), |
| ("Web Interface", test_web_interface), |
| ("FastAPI Endpoint", test_fastapi_endpoint), |
| ] |
| |
| results = [] |
| for test_name, test_func in tests: |
| print(f"\n--- {test_name} Test ---") |
| try: |
| success = test_func() |
| results.append((test_name, success)) |
| except Exception as e: |
| print(f"β {test_name} test crashed: {e}") |
| results.append((test_name, False)) |
| |
| |
| print("\n" + "="*50) |
| print("π Test Summary:") |
| print("="*50) |
| |
| passed = 0 |
| for test_name, success in results: |
| status = "β
PASSED" if success else "β FAILED" |
| print(f"{test_name}: {status}") |
| if success: |
| passed += 1 |
| |
| print(f"\nTests passed: {passed}/{len(results)}") |
| |
| if passed == len(results): |
| print("π All tests passed! The app is ready to deploy.") |
| else: |
| print("β οΈ Some tests failed. Please check the errors above.") |
|
|