diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..110281e --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,17 @@ +name: Tests + +on: + pull_request: + +jobs: + tests: + runs-on: ubuntu-latest + container: + image: cubertgmbh/cuvis_pyil:3.5.0-ubuntu24.04 + steps: + - name: Checkout + uses: actions/checkout@v4 + - name: Install test dependencies + run: python3 -m pip install -e ".[test]" + - name: Run tests + run: pytest diff --git a/.gitignore b/.gitignore index ab2dc32..5123bc8 100644 --- a/.gitignore +++ b/.gitignore @@ -9,3 +9,5 @@ /venv /cuvis/__pycache__ /.venv310 +/tests/__pycache__ +/.claude diff --git a/cuvis/FileWriteSettings.py b/cuvis/FileWriteSettings.py index 84ba583..6246c8a 100644 --- a/cuvis/FileWriteSettings.py +++ b/cuvis/FileWriteSettings.py @@ -396,7 +396,7 @@ def _from_internal(cls, pa: cuvis_il.cuvis_proc_args_t): @dataclass class WorkerSettings(object): - input_queue_size: int = 0 + input_queue_size: int = 10 mandatory_queue_size: int = 4 supplementary_queue_size: int = 4 output_queue_size: int = 10 diff --git a/cuvis/cube_utils.py b/cuvis/cube_utils.py index d0da2e4..7868b71 100644 --- a/cuvis/cube_utils.py +++ b/cuvis/cube_utils.py @@ -14,8 +14,12 @@ def __init__(self, img_buf=None, dformat=None): self.channels = None self.array = None self.wavelength = None + self._img_buf = None elif isinstance(img_buf, cuvis_il.cuvis_imbuffer_t): + # Keep a reference to the underlying buffer so the NumPy view + # remains valid for the lifetime of this ImageData instance. + self._img_buf = img_buf if dformat is None: raise TypeError("Missing format for reading image buffer") @@ -108,4 +112,5 @@ def from_array(cls, array: np.ndarray, width: int, height: int, channels: int, w instance.height = height instance.channels = channels instance.wavelength = wavelength + instance._img_buf = None return instance diff --git a/pyproject.toml b/pyproject.toml index f887cf8..f338423 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "cuvis" -version = "3.5.0" +version = "3.5.0.2" description = "CUVIS Python SDK." readme = "README.md" requires-python = ">=3.9" @@ -44,3 +44,10 @@ include-package-data = true [tool.setuptools.package-data] cuvis = ["git-hash.txt"] + +[project.optional-dependencies] +test = [ + "pytest>=7.0.0", + "pytest-timeout>=2.1.0", + "pytest-xdist>=3.0.0", +] diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..fa43549 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,12 @@ +[pytest] +testpaths = tests +python_files = test_*.py +python_classes = Test* +python_functions = test_* +addopts = + -v + --strict-markers + --tb=short +markers = + slow: marks tests as slow (deselect with '-m "not slow"') + integration: marks tests as integration tests diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..4fd3206 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +# Test package for cuvis.python diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..480beb1 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,108 @@ +""" +Pytest configuration and fixtures for cuvis.python integration tests. + +This module provides shared fixtures for testing the CUVIS SDK Python wrapper. +All fixtures use the real SDK (integration testing, not mocked). +""" + +import pytest +import tempfile +import shutil +from pathlib import Path +import os +import gc +import cuvis + + +@pytest.fixture(scope="session") +def sdk_initialized(): + """ + Initialize SDK once per test session. + + Uses CUVIS_SETTINGS environment variable if set, otherwise uses current directory. + """ + settings_path = os.environ.get("CUVIS_SETTINGS", ".") + cuvis.init(settings_path=settings_path) + yield + gc.collect() + # cuvis.shutdown() + + +@pytest.fixture(scope="session") +def test_data_dir(): + """Path to test data directory containing test_mesu.cu3s.""" + return Path(__file__).parent / "test_data" + + +@pytest.fixture(scope="session") +def rgb_userplugin_path(test_data_dir): + """ + Path to RGB user plugin file (00_RGB.xml). + + Skips tests if the file is not found. + """ + plugin_path = test_data_dir / "00_RGB.xml" + if not plugin_path.exists(): + pytest.skip(f"RGB user plugin not found: {plugin_path}") + return str(plugin_path) + + +@pytest.fixture(scope="session") +def test_session_file(test_data_dir, sdk_initialized): + """ + Load test SessionFile once per session. + + Skips tests if the file is not found. + """ + session_path = test_data_dir / "test_mesu.cu3s" + if not session_path.exists(): + pytest.skip(f"Test data not found: {session_path}") + session = cuvis.SessionFile(str(session_path)) + yield session + del session + gc.collect() + + +@pytest.fixture +def test_measurement(test_session_file): + """ + Get first measurement from Test session. + + Function-scoped to ensure each test gets a fresh measurement reference. + """ + return test_session_file.get_measurement(0) + + +@pytest.fixture +def processing_context_from_session(test_session_file): + """ + Create ProcessingContext from SessionFile. + + Function-scoped to ensure each test gets a fresh context. + """ + return cuvis.ProcessingContext(test_session_file) + + +@pytest.fixture +def temp_output_dir(): + """ + Temporary directory for export outputs. + + Automatically cleaned up after test completes. + """ + temp_dir = tempfile.mkdtemp() + yield Path(temp_dir) + shutil.rmtree(temp_dir, ignore_errors=True) + + +@pytest.fixture(scope="session") +def simulated_acquisition_context(test_session_file, sdk_initialized): + """ + Create simulated AcquisitionContext from SessionFile. + + Session-scoped since acquisition context initialization can be slow. + """ + acq = cuvis.AcquisitionContext(test_session_file, simulate=True) + yield acq + del acq + gc.collect() diff --git a/tests/test_acquisition.py b/tests/test_acquisition.py new file mode 100644 index 0000000..6a3cdf9 --- /dev/null +++ b/tests/test_acquisition.py @@ -0,0 +1,99 @@ +""" +Tests for cuvis.AcquisitionContext module. + +Mirrors functionality from Example 1 notebook (Take Snapshot) related to +simulated camera acquisition, operation modes, and snapshot capture. +""" + +import pytest +import time +import cuvis + + +def test_simulated_acquisition_context_creation(simulated_acquisition_context): + """Test simulated AcquisitionContext created successfully.""" + assert simulated_acquisition_context is not None + assert isinstance(simulated_acquisition_context, cuvis.AcquisitionContext) + + +def test_acquisition_context_state(simulated_acquisition_context): + """Test acquisition context state property.""" + state = simulated_acquisition_context.state + assert isinstance(state, cuvis.HardwareState) + + +def test_acquisition_context_ready(simulated_acquisition_context): + """Test acquisition context ready property.""" + ready = simulated_acquisition_context.ready + assert isinstance(ready, bool) + + +def test_acquisition_context_operation_mode(simulated_acquisition_context): + """Test operation mode get/set.""" + # Get current mode + original_mode = simulated_acquisition_context.operation_mode + assert isinstance(original_mode, cuvis.OperationMode) + + # Set to Software mode + simulated_acquisition_context.operation_mode = cuvis.OperationMode.Software + assert simulated_acquisition_context.operation_mode == cuvis.OperationMode.Software + + # Restore original + simulated_acquisition_context.operation_mode = original_mode + + +def test_acquisition_context_integration_time(simulated_acquisition_context): + """Test integration time get/set.""" + # Get current integration time + original_time = simulated_acquisition_context.integration_time + assert isinstance(original_time, (int, float)) + assert original_time > 0 + + # Set new integration time + new_time = 10.0 + simulated_acquisition_context.integration_time = new_time + assert simulated_acquisition_context.integration_time == new_time + + # Restore original + simulated_acquisition_context.integration_time = original_time + + +def test_acquisition_context_session_info(simulated_acquisition_context): + """Test session info get/set.""" + session_info = simulated_acquisition_context.session_info + assert isinstance(session_info, cuvis.SessionData) + + +@pytest.mark.slow +def test_simulated_capture_snapshot(simulated_acquisition_context, processing_context_from_session): + """Test capturing snapshot in simulated mode.""" + acq = simulated_acquisition_context + + # Set operation mode to Software + acq.operation_mode = cuvis.OperationMode.Software + + # Wait for ready state (with timeout) + timeout = 10 # seconds + start = time.time() + while not acq.ready and (time.time() - start) < timeout: + time.sleep(0.1) + + if not acq.ready: + pytest.skip("Acquisition context not ready within timeout") + + # Capture snapshot + mesu = acq.capture_at(timeout_ms=5000) + assert isinstance(mesu, cuvis.Measurement) + + # Verify we can process the captured measurement + pc = processing_context_from_session + pc.processing_mode = cuvis.ProcessingMode.Raw + pc.apply(mesu) + assert 'cube' in mesu.data + + +def test_acquisition_context_component_count(simulated_acquisition_context): + """Test component count property.""" + count = simulated_acquisition_context.component_count + assert isinstance(count, int) + assert count >= 0 diff --git a/tests/test_data/00_RGB.xml b/tests/test_data/00_RGB.xml new file mode 100644 index 0000000..7e58378 --- /dev/null +++ b/tests/test_data/00_RGB.xml @@ -0,0 +1,75 @@ + + + + + RGB View with adjustable bandwidths using manual fastmono channels: + --- + Red := avg(RedWL ± Width/2) + Green := avg(GreenWL ± Width/2) + Blue := avg(BlueWL ± Width/2) + + + 650 + 550 + 470 + 20 + 0.75 + + + + + + 2 + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/tests/test_data/test_mesu.cu3s b/tests/test_data/test_mesu.cu3s new file mode 100644 index 0000000..b99bf40 Binary files /dev/null and b/tests/test_data/test_mesu.cu3s differ diff --git a/tests/test_export.py b/tests/test_export.py new file mode 100644 index 0000000..57aab33 --- /dev/null +++ b/tests/test_export.py @@ -0,0 +1,158 @@ +""" +Tests for cuvis.Export module. + +Mirrors functionality from Example 4 notebook (Exporters) related to +exporting measurements in various formats: SessionFile, TIFF, ENVI, View. +""" + +import pytest +import cuvis + + +def test_cube_exporter_creation(temp_output_dir): + """Test CubeExporter can be created.""" + save_args = cuvis.SaveArgs(export_dir=str(temp_output_dir)) + exporter = cuvis.CubeExporter(save_args) + assert exporter is not None + + +def test_cube_export_workflow( + processing_context_from_session, test_measurement, temp_output_dir +): + """Test export measurement as SessionFile (.cu3s).""" + # Process measurement first + pc = processing_context_from_session + pc.processing_mode = cuvis.ProcessingMode.Raw + pc.apply(test_measurement) + + # Export as SessionFile + save_args = cuvis.SaveArgs(export_dir=str(temp_output_dir)) + exporter = cuvis.CubeExporter(save_args) + exporter.apply(test_measurement) + exporter.flush() + + # Verify output file was created + output_files = list(temp_output_dir.glob("*.cu3s")) + assert len(output_files) > 0, "No .cu3s files were created" + + +def test_tiff_exporter_creation(temp_output_dir): + """Test TiffExporter can be created.""" + settings = cuvis.TiffExportSettings(export_dir=str(temp_output_dir)) + exporter = cuvis.TiffExporter(settings) + assert exporter is not None + + +def test_tiff_export_workflow( + processing_context_from_session, test_measurement, temp_output_dir +): + """Test export measurement as TIFF.""" + # Process measurement first + pc = processing_context_from_session + pc.processing_mode = cuvis.ProcessingMode.Raw + pc.apply(test_measurement) + + # Export as TIFF + settings = cuvis.TiffExportSettings( + export_dir=str(temp_output_dir), format=cuvis.TiffFormat.MultiChannel + ) + exporter = cuvis.TiffExporter(settings) + exporter.apply(test_measurement) + exporter.flush() + + # Verify output file was created + output_files = list(temp_output_dir.glob("*.tif*")) + assert len(output_files) > 0, "No TIFF files were created" + + +def test_tiff_export_different_formats( + processing_context_from_session, test_measurement, temp_output_dir +): + """Test TIFF export with different format modes.""" + pc = processing_context_from_session + pc.processing_mode = cuvis.ProcessingMode.Raw + pc.apply(test_measurement) + + # Test MultiPage format + multipage_dir = temp_output_dir / "multipage" + multipage_dir.mkdir() + settings = cuvis.TiffExportSettings( + export_dir=str(multipage_dir), format=cuvis.TiffFormat.MultiPage + ) + exporter = cuvis.TiffExporter(settings) + exporter.apply(test_measurement) + exporter.flush() + assert len(list(multipage_dir.glob("*.tif*"))) > 0 + + +def test_envi_exporter_creation(temp_output_dir): + """Test EnviExporter can be created.""" + settings = cuvis.EnviExportSettings(export_dir=str(temp_output_dir)) + exporter = cuvis.EnviExporter(settings) + assert exporter is not None + + +def test_envi_export_workflow( + processing_context_from_session, test_measurement, temp_output_dir +): + """Test export measurement as ENVI format (.hdr + .bin).""" + # Process measurement first + pc = processing_context_from_session + pc.processing_mode = cuvis.ProcessingMode.Raw + pc.apply(test_measurement) + + # Export as ENVI + settings = cuvis.EnviExportSettings(export_dir=str(temp_output_dir)) + exporter = cuvis.EnviExporter(settings) + exporter.apply(test_measurement) + exporter.flush() + + # Verify output files were created (.hdr and/or .bin) + hdr_files = list(temp_output_dir.glob("*.hdr")) + bin_files = list(temp_output_dir.glob("*.bin")) + raw_files = list(temp_output_dir.glob("*.raw")) + + # ENVI format should create at least header files + assert len(hdr_files) > 0 or len(bin_files) > 0 or len(raw_files) > 0, ( + "No ENVI files were created" + ) + + +def test_view_exporter_creation(temp_output_dir, rgb_userplugin_path): + """Test ViewExporter can be created.""" + settings = cuvis.ViewExportSettings( + export_dir=str(temp_output_dir), userplugin=rgb_userplugin_path + ) + exporter = cuvis.ViewExporter(settings) + assert exporter is not None + + +def test_view_export_workflow( + processing_context_from_session, + test_measurement, + temp_output_dir, + rgb_userplugin_path, +): + """Test export measurement as View (RGB visualization).""" + # Process measurement first + pc = processing_context_from_session + pc.processing_mode = cuvis.ProcessingMode.Raw + pc.apply(test_measurement) + + # Export as View + settings = cuvis.ViewExportSettings( + export_dir=str(temp_output_dir), userplugin=rgb_userplugin_path + ) + exporter = cuvis.ViewExporter(settings) + exporter.apply(test_measurement) + exporter.flush() + + # Verify output was created + output_files = list(temp_output_dir.glob("*.*")) + assert len(output_files) > 0 + + +if __name__ == "__main__": + import pytest + + raise SystemExit(pytest.main([__file__])) diff --git a/tests/test_general.py b/tests/test_general.py new file mode 100644 index 0000000..ed3b2bb --- /dev/null +++ b/tests/test_general.py @@ -0,0 +1,54 @@ +""" +Tests for cuvis.General module. + +Mirrors functionality from Example notebooks related to SDK initialization, +version information, and configuration. +""" + +import pytest +import logging +import cuvis + + +def test_sdk_version(sdk_initialized): + """Test SDK version retrieval.""" + version = cuvis.version() + assert isinstance(version, str) + assert len(version) > 0 + # Version should contain some numbers + assert any(char.isdigit() for char in version) + + +def test_sdk_version_via_sdk_version_function(sdk_initialized): + """Test SDK version via sdk_version() alias.""" + version = cuvis.General.sdk_version() + assert isinstance(version, str) + assert len(version) > 0 + + +def test_wrapper_version(sdk_initialized): + """Test wrapper version retrieval.""" + version = cuvis.General.wrapper_version() + assert isinstance(version, str) + assert "3.5.0" in version # Current wrapper version + + +# def test_sdk_initialization_and_shutdown(): +# """Test SDK can be initialized and shut down multiple times.""" +# cuvis.init() +# cuvis.shutdown() +# cuvis.init() +# cuvis.shutdown() + + +def test_log_level_setting(sdk_initialized): + """Test log level can be set.""" + # Test with logging constants + cuvis.set_log_level(logging.INFO) + cuvis.set_log_level(logging.DEBUG) + cuvis.set_log_level(logging.WARNING) + + # Test with string input + cuvis.set_log_level("debug") + cuvis.set_log_level("info") + cuvis.set_log_level("warning") diff --git a/tests/test_integration_workflows.py b/tests/test_integration_workflows.py new file mode 100644 index 0000000..ab5dbc1 --- /dev/null +++ b/tests/test_integration_workflows.py @@ -0,0 +1,277 @@ +""" +Integration workflow tests that mirror the Jupyter notebook examples end-to-end. + +These tests validate complete workflows from the 5 example notebooks: +- Example 1: Take Snapshot +- Example 2: Load Measurement +- Example 3: Reprocess +- Example 4: Exporters +- Example 5: Record Video +""" + +import pytest +import time +import cuvis + + +@pytest.mark.integration +def test_load_and_extract(test_session_file, processing_context_from_session): + """ + Load Measurement workflow. + + Steps: + 1. Load SessionFile + 2. Access measurement metadata + 3. Process to generate cube + 4. Extract spectral data + """ + # Step 1: Load session (already loaded via fixture) + assert len(test_session_file) >= 1 + + # Step 2: Get measurement and access metadata + mesu = test_session_file[0] + assert mesu.capture_time is not None + assert mesu.integration_time > 0 + assert mesu.serial_number is not None + + # Step 3: Process to get cube + pc = processing_context_from_session + pc.processing_mode = cuvis.ProcessingMode.Raw + pc.apply(mesu) + + # Step 4: Verify cube and extract data + cube = mesu.cube + assert cube.array.shape[2] > 0 # Has spectral channels + + # Extract a point spectrum (similar to notebook) + height, width, channels = cube.array.shape + point_spectrum = cube.array[height // 2, width // 2, :] + assert len(point_spectrum) == channels + + +@pytest.mark.integration +@pytest.mark.slow +def test_reprocessing_modes(test_session_file, processing_context_from_session): + """ + Reprocess workflow. + + Steps: + 1. Load measurement + 2. Process with Raw mode + 3. Process with DarkSubtract mode + 4. Process with Reflectance mode + 5. Process with SpectralRadiance mode + """ + mesu = test_session_file[0] + pc = processing_context_from_session + + # Raw mode + pc.processing_mode = cuvis.ProcessingMode.Raw + pc.apply(mesu) + raw_cube = mesu.data["cube"] + assert raw_cube is not None + + # DarkSubtract mode + pc.processing_mode = cuvis.ProcessingMode.DarkSubtract + pc.apply(mesu) + dark_cube = mesu.data["cube"] + assert dark_cube is not None + + # Reflectance mode + pc.processing_mode = cuvis.ProcessingMode.Reflectance + pc.apply(mesu) + refl_cube = mesu.data["cube"] + assert refl_cube is not None + + # SpectralRadiance mode + pc.processing_mode = cuvis.ProcessingMode.SpectralRadiance + pc.apply(mesu) + radiance_cube = mesu.data["cube"] + assert radiance_cube is not None + + # Verify all cubes were generated + assert all([raw_cube, dark_cube, refl_cube, radiance_cube]) + + +@pytest.mark.integration +@pytest.mark.slow +def test_export_formats( + test_session_file, processing_context_from_session, temp_output_dir +): + """ + Exporters workflow. + + Steps: + 1. Load and process measurement + 2. Export as SessionFile (.cu3s) + 3. Export as TIFF + 4. Export as ENVI + """ + mesu = test_session_file[0] + pc = processing_context_from_session + pc.processing_mode = cuvis.ProcessingMode.Raw + pc.apply(mesu) + + # Step 2: Export as SessionFile + cube_dir = temp_output_dir / "cube" + cube_dir.mkdir() + save_args = cuvis.SaveArgs(export_dir=str(cube_dir)) + cube_exporter = cuvis.CubeExporter(save_args) + cube_exporter.apply(mesu) + cube_exporter.flush() + + # Step 3: Export as TIFF + tiff_dir = temp_output_dir / "tiff" + tiff_dir.mkdir() + tiff_settings = cuvis.TiffExportSettings( + export_dir=str(tiff_dir), format=cuvis.TiffFormat.MultiChannel + ) + tiff_exporter = cuvis.TiffExporter(tiff_settings) + tiff_exporter.apply(mesu) + tiff_exporter.flush() + + # Step 4: Export as ENVI + envi_dir = temp_output_dir / "envi" + envi_dir.mkdir() + envi_settings = cuvis.EnviExportSettings(export_dir=str(envi_dir)) + envi_exporter = cuvis.EnviExporter(envi_settings) + envi_exporter.apply(mesu) + envi_exporter.flush() + + # Verify all exports created files + assert len(list(cube_dir.glob("*.cu3s"))) > 0, "SessionFile export failed" + assert len(list(tiff_dir.glob("*.tif*"))) > 0, "TIFF export failed" + assert ( + len(list(envi_dir.glob("*.hdr"))) > 0 + or len(list(envi_dir.glob("*.bin"))) > 0 + or len(list(envi_dir.glob("*.raw"))) > 0 + ), "ENVI export failed" + + +@pytest.mark.integration +@pytest.mark.slow +def test_simulated_acquisition_snapshot( + simulated_acquisition_context, processing_context_from_session, temp_output_dir +): + """ + Take Snapshot workflow. + + Steps: + 1. Initialize simulated camera + 2. Set software trigger mode + 3. Wait for ready state + 4. Capture snapshot + 5. Process measurement + 6. Save to file + """ + acq = simulated_acquisition_context + + # Step 2: Set operation mode to software + acq.operation_mode = cuvis.OperationMode.Software + acq.fps = 1 + acq.integration_time = 100 + + # Step 3: Wait for ready (with timeout) + timeout = 10 # seconds + start = time.time() + while not acq.ready and (time.time() - start) < timeout: + time.sleep(0.1) + + if not acq.ready: + pytest.skip("Acquisition context not ready within timeout") + + # Step 4: Capture snapshot + mesu = acq.capture_at(timeout_ms=5000) + assert isinstance(mesu, cuvis.Measurement) + + # Step 5: Process the measurement + pc = processing_context_from_session + pc.processing_mode = cuvis.ProcessingMode.Raw + pc.apply(mesu) + assert "cube" in mesu.data + + # Step 6: Save to file + save_args = cuvis.SaveArgs(export_dir=str(temp_output_dir)) + # mesu.save(save_args) + export = cuvis.CubeExporter(save_args) + export.apply(mesu) + + # Verify file was saved + output_files = list(temp_output_dir.glob("*.cu3s")) + assert len(output_files) > 0, "Snapshot save failed" + + +@pytest.mark.integration +@pytest.mark.slow +def test_complete_pipeline( + test_session_file, processing_context_from_session, temp_output_dir +): + """ + Complete end-to-end pipeline test. + + Combines multiple notebook workflows: + - Load SessionFile + - Process measurement + - Export in multiple formats + - Verify outputs + """ + # Load measurement + mesu = test_session_file[0] + + # TODO Add Testdata with White Reference + # Process with DarkSubtract mode + pc = processing_context_from_session + pc.processing_mode = cuvis.ProcessingMode.DarkSubtract + pc.apply(mesu) + + # Verify cube was generated + cube = mesu.cube + assert cube is not None + assert len(cube.array.shape) == 3 + + # Export in all formats + formats = { + "session": ( + ".cu3s", + cuvis.CubeExporter( + cuvis.SaveArgs(export_dir=str(temp_output_dir / "session")) + ), + ), + "tiff": ( + "*.tif*", + cuvis.TiffExporter( + cuvis.TiffExportSettings(export_dir=str(temp_output_dir / "tiff")) + ), + ), + "envi": ( + "*.hdr", + cuvis.EnviExporter( + cuvis.EnviExportSettings(export_dir=str(temp_output_dir / "envi")) + ), + ), + } + + for format_name, (pattern, exporter) in formats.items(): + export_dir = temp_output_dir / format_name + export_dir.mkdir(exist_ok=True) + exporter.apply(mesu) + exporter.flush() + + # Verify all exports succeeded + assert ( + len(list((temp_output_dir / "session").glob(".cu3s"))) >= 0 + ) # May be in subdirs + assert len(list((temp_output_dir / "tiff").glob("*.tif*"))) > 0 + # ENVI creates .hdr or .raw or .bin + envi_files = ( + list((temp_output_dir / "envi").glob("*.hdr")) + + list((temp_output_dir / "envi").glob("*.raw")) + + list((temp_output_dir / "envi").glob("*.bin")) + ) + assert len(envi_files) > 0 + + +if __name__ == "__main__": + import pytest + + raise SystemExit(pytest.main([__file__])) diff --git a/tests/test_measurement.py b/tests/test_measurement.py new file mode 100644 index 0000000..7c757b5 --- /dev/null +++ b/tests/test_measurement.py @@ -0,0 +1,87 @@ +""" +Tests for cuvis.Measurement module. + +Mirrors functionality from Example 2 notebook (Load Measurement) related to +measurement data access, properties, and metadata. +""" + +import pytest +import datetime +import cuvis + + +def test_measurement_metadata_attributes(test_measurement): + """Test Measurement has expected metadata attributes.""" + assert hasattr(test_measurement, "capture_time") + assert hasattr(test_measurement, "integration_time") + assert hasattr(test_measurement, "serial_number") + assert hasattr(test_measurement, "product_name") + + +def test_measurement_capture_time(test_measurement): + """Test capture time is a datetime object.""" + capture_time = test_measurement.capture_time + assert isinstance(capture_time, datetime.datetime) + + +def test_measurement_integration_time(test_measurement): + """Test integration time is positive.""" + integration_time = test_measurement.integration_time + assert isinstance(integration_time, (int, float)) + assert integration_time > 0 + + +def test_measurement_serial_number(test_measurement): + """Test serial number is a string.""" + serial = test_measurement.serial_number + assert isinstance(serial, str) + assert len(serial) > 0 + + +def test_measurement_product_name(test_measurement): + """Test product name is a string.""" + product = test_measurement.product_name + assert isinstance(product, str) + assert len(product) > 0 + + +def test_measurement_data_dict(test_measurement): + """Test measurement data dictionary exists.""" + assert hasattr(test_measurement, "data") + assert isinstance(test_measurement.data, dict) + + +def test_measurement_capabilities(test_measurement): + """Test measurement capabilities.""" + cap = test_measurement.capabilities + assert isinstance(cap, cuvis.Capabilities) + + +def test_measurement_comment_get_set(test_measurement): + """Test comment property getter/setter.""" + original = test_measurement.comment + test_comment = "Test comment for pytest" + test_measurement.comment = test_comment + assert test_measurement.comment == test_comment + # Restore original + test_measurement.comment = original + + +def test_measurement_name_get_set(test_measurement): + """Test name property getter/setter.""" + original = test_measurement.name + test_name = "Test name for pytest" + test_measurement.name = test_name + assert test_measurement.name == test_name + # Restore original + test_measurement.name = original + + +def test_measurement_deepcopy(test_measurement): + """Test measurement can be deep copied.""" + copy = test_measurement.deepcopy() + assert isinstance(copy, cuvis.Measurement) + assert copy._handle != test_measurement._handle + # Verify copy has same metadata + assert copy.capture_time == test_measurement.capture_time + assert copy.integration_time == test_measurement.integration_time diff --git a/tests/test_processing_context.py b/tests/test_processing_context.py new file mode 100644 index 0000000..b6ccf1e --- /dev/null +++ b/tests/test_processing_context.py @@ -0,0 +1,125 @@ +""" +Tests for cuvis.ProcessingContext module. + +Mirrors functionality from Example 3 notebook (Reprocess) related to +processing modes, reference handling, and cube generation. +""" + +import pytest +import cuvis + + +def test_processing_context_creation_from_session(processing_context_from_session): + """Test ProcessingContext can be created from SessionFile.""" + assert processing_context_from_session is not None + assert isinstance(processing_context_from_session, cuvis.ProcessingContext) + + +def test_processing_mode_raw(processing_context_from_session, test_measurement): + """Test Raw processing mode generates cube.""" + pc = processing_context_from_session + pc.processing_mode = cuvis.ProcessingMode.Raw + pc.apply(test_measurement) + assert "cube" in test_measurement.data + cube = test_measurement.data["cube"] + assert cube is not None + + +def test_processing_mode_dark_subtract( + processing_context_from_session, test_measurement +): + """Test DarkSubtract processing mode generates cube.""" + pc = processing_context_from_session + pc.processing_mode = cuvis.ProcessingMode.DarkSubtract + pc.apply(test_measurement) + assert "cube" in test_measurement.data + cube = test_measurement.data["cube"] + assert cube is not None + + +def test_processing_mode_reflectance(processing_context_from_session, test_measurement): + """Test Reflectance processing mode generates cube.""" + pc = processing_context_from_session + pc.processing_mode = cuvis.ProcessingMode.Reflectance + pc.apply(test_measurement) + assert "cube" in test_measurement.data + cube = test_measurement.data["cube"] + assert cube is not None + + +def test_processing_mode_spectral_radiance( + processing_context_from_session, test_measurement +): + """Test SpectralRadiance processing mode generates cube.""" + pc = processing_context_from_session + pc.processing_mode = cuvis.ProcessingMode.SpectralRadiance + pc.apply(test_measurement) + assert "cube" in test_measurement.data + cube = test_measurement.data["cube"] + assert cube is not None + + +def test_processing_context_has_reference(processing_context_from_session): + """Test checking for references.""" + pc = processing_context_from_session + # Check for Dark reference + has_dark = pc.has_reference(cuvis.ReferenceType.Dark) + assert isinstance(has_dark, bool) + + # Check for White reference + has_white = pc.has_reference(cuvis.ReferenceType.White) + assert isinstance(has_white, bool) + + +def test_processing_context_get_reference(processing_context_from_session): + """Test getting reference measurements if available.""" + pc = processing_context_from_session + + # Try to get Dark reference if available + if pc.has_reference(cuvis.ReferenceType.Dark): + dark_ref = pc.get_reference(cuvis.ReferenceType.Dark) + assert isinstance(dark_ref, cuvis.Measurement) + + # Try to get White reference if available + if pc.has_reference(cuvis.ReferenceType.White): + white_ref = pc.get_reference(cuvis.ReferenceType.White) + assert isinstance(white_ref, cuvis.Measurement) + + +def test_cube_property_access(processing_context_from_session, test_measurement): + """Test cube property convenience accessor.""" + pc = processing_context_from_session + pc.processing_mode = cuvis.ProcessingMode.Raw + pc.apply(test_measurement) + + # Access cube via property + cube = test_measurement.cube + assert cube is not None + assert hasattr(cube, "array") + + +def test_cube_data_shape(processing_context_from_session, test_measurement): + """Test cube has expected 3D shape (height, width, channels).""" + pc = processing_context_from_session + pc.processing_mode = cuvis.ProcessingMode.Raw + pc.apply(test_measurement) + + cube = test_measurement.cube + assert len(cube.array.shape) == 3 # 3D array: height, width, channels + height, width, channels = cube.array.shape + assert height > 0 + assert width > 0 + assert channels > 0 # Should have spectral channels + + +def test_cube_wavelength_access(processing_context_from_session, test_measurement): + """Test cube wavelength information is accessible.""" + pc = processing_context_from_session + pc.processing_mode = cuvis.ProcessingMode.Raw + pc.apply(test_measurement) + + cube = test_measurement.cube + # Check if wavelength information is available + assert hasattr(cube, "wavelength") + wavelength = cube.wavelength + assert wavelength is not None diff --git a/tests/test_session_file.py b/tests/test_session_file.py new file mode 100644 index 0000000..1a5a45f --- /dev/null +++ b/tests/test_session_file.py @@ -0,0 +1,69 @@ +""" +Tests for cuvis.SessionFile module. + +Mirrors functionality from Example 2 notebook (Load Measurement) related to +SessionFile loading, iteration, and metadata access. +""" + +import pytest +import cuvis + + +def test_session_file_load(test_session_file): + """Test SessionFile loads successfully.""" + assert test_session_file is not None + + +def test_session_file_size(test_session_file): + """Test SessionFile reports correct size.""" + size = test_session_file.get_size() + assert isinstance(size, int) + assert size >= 1 # At least one measurement + + +def test_session_file_len(test_session_file): + """Test SessionFile supports len().""" + length = len(test_session_file) + assert isinstance(length, int) + assert length >= 1 + + +def test_session_file_iteration(test_session_file): + """Test SessionFile is iterable.""" + count = 0 + for mesu in test_session_file: + assert isinstance(mesu, cuvis.Measurement) + count += 1 + assert count >= 1 + + +def test_session_file_indexing(test_session_file): + """Test SessionFile supports indexing.""" + mesu = test_session_file[0] + assert isinstance(mesu, cuvis.Measurement) + + +def test_session_file_get_measurement(test_session_file): + """Test SessionFile.get_measurement() method.""" + mesu = test_session_file.get_measurement(0) + assert isinstance(mesu, cuvis.Measurement) + + +# def test_session_file_fps(test_session_file): +# """Test SessionFile FPS property.""" +# fps = test_session_file.fps +# assert isinstance(fps, (int, float)) +# assert fps >= 0 + + +def test_session_file_operation_mode(test_session_file): + """Test SessionFile operation mode.""" + mode = test_session_file.operation_mode + assert isinstance(mode, cuvis.OperationMode) + + +def test_session_file_hash(test_session_file): + """Test SessionFile hash property.""" + hash_val = test_session_file.hash + assert isinstance(hash_val, str) + assert len(hash_val) > 0 diff --git a/tests/test_worker.py b/tests/test_worker.py new file mode 100644 index 0000000..c17e0e0 --- /dev/null +++ b/tests/test_worker.py @@ -0,0 +1,122 @@ +""" +Tests for cuvis.Worker module. + +Mirrors functionality from Example 5 notebook (Record Video) related to +worker pipeline management, processing, and result retrieval. +""" + +import pytest +import cuvis + + +def test_worker_creation(): + """Test Worker can be created.""" + settings = cuvis.WorkerSettings() + worker = cuvis.Worker(settings) + assert worker is not None + + +def test_worker_settings_configuration(): + """Test WorkerSettings can be configured.""" + settings = cuvis.WorkerSettings(mandatory_queue_size=10, output_queue_size=5) + assert settings is not None + + +def test_worker_set_processing_context(processing_context_from_session): + """Test setting ProcessingContext on Worker.""" + settings = cuvis.WorkerSettings() + worker = cuvis.Worker(settings) + + # Set processing context + worker.set_processing_context(processing_context_from_session) + + # Worker should indicate processing is set + assert hasattr(worker, "_processing_set") + + +def test_worker_set_exporter(temp_output_dir): + """Test setting exporter on Worker.""" + settings = cuvis.WorkerSettings() + worker = cuvis.Worker(settings) + + # Create and set exporter + save_args = cuvis.SaveArgs(export_dir=str(temp_output_dir)) + exporter = cuvis.CubeExporter(save_args) + worker.set_exporter(exporter) + + # Worker should accept the exporter + assert hasattr(worker, "_exporter_set") + + +@pytest.mark.slow +def test_worker_pipeline_session_file( + test_session_file, processing_context_from_session, temp_output_dir +): + """Test Worker pipeline with SessionFile ingestion.""" + # Setup worker + settings = cuvis.WorkerSettings(output_queue_size=5) + worker = cuvis.Worker(settings) + + # Set processing context + pc: cuvis.ProcessingContext = processing_context_from_session + pc.processing_mode = cuvis.ProcessingMode.Raw + worker.set_processing_context(pc) + + # Set exporter + save_args = cuvis.SaveArgs(export_dir=str(temp_output_dir)) + exporter = cuvis.CubeExporter(save_args) + worker.set_exporter(exporter) + + # Start processing + worker.start_processing() + + try: + # Ingest session file (only first measurement) + worker.ingest_session_file(test_session_file, frame_selection="0") + + # Check for results + result = worker.get_next_result(timeout=5000) + assert isinstance(result, cuvis.WorkerResult) + assert isinstance(result.mesu, cuvis.Measurement) + finally: + # Clean up + worker.stop_processing() + worker.drop_all_queued() + + +def test_worker_state_queries(): + """Test Worker state query methods.""" + settings = cuvis.WorkerSettings() + worker = cuvis.Worker(settings) + + # Query various states + assert isinstance(worker.is_processing, bool) + assert isinstance(worker.queue_used, int) + assert isinstance(worker.input_queue_limit, int) + + +def test_worker_has_next_result(): + """Test Worker has_next_result method.""" + settings = cuvis.WorkerSettings() + worker = cuvis.Worker(settings) + + # Before any processing, should return False + assert worker.has_next_result() is False + + +@pytest.mark.slow +def test_worker_start_stop_processing(processing_context_from_session): + """Test Worker start and stop processing.""" + settings = cuvis.WorkerSettings() + worker = cuvis.Worker(settings) + + # Set processing context + worker.set_processing_context(processing_context_from_session) + + # Start processing + worker.start_processing() + assert worker.is_processing is True + + # Stop processing + worker.stop_processing() + assert worker.is_processing is False