diff --git a/.gitignore b/.gitignore index 91060e5..71b6543 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,6 @@ +# macOS .DS_Store files +.DS_Store + # Byte-compiled / optimized / DLL files __pycache__/ *.py[cod] @@ -22,7 +25,7 @@ var/ wheels/ share/python-wheels/ *.egg-info/ -notebooks/ +notebooks .installed.cfg *.egg MANIFEST diff --git a/oceanstream/L2_calibrated_data/compute_sv.py b/oceanstream/L2_calibrated_data/compute_sv.py new file mode 100644 index 0000000..af76b43 --- /dev/null +++ b/oceanstream/L2_calibrated_data/compute_sv.py @@ -0,0 +1,136 @@ +""" +compute_sv.py +------------- + +Module for computing the volume backscattering strength (Sv) from raw data. +Supported Sonar Models: +- EK60 +- AZFP +- EK80 +Functions and Classes: +- `SupportedSonarModelsForSv`: An Enum containing the sonar models supported +for Sv computation. +- `WaveformMode`: Enum specifying the waveform mode (CW or BB). +- `EncodeMode`: Enum indicating the encoding mode (complex or power). +- `ComputeSVParams`: Class to validate and structure the parameters passed +to the Sv computation function. +- `compute_sv`: Main function to calculate Sv given an EchoData object +and other optional parameters. + +Usage: + +To compute Sv for a given EchoData object, `ed`, simply call: +`compute_sv(ed)` +""" + +from enum import Enum +from typing import Any, Optional + +import echopype as ep +import xarray as xr +from echopype.echodata.echodata import EchoData +from pydantic import BaseModel, ValidationError, field_validator +from pydantic_core.core_schema import FieldValidationInfo + + +class SupportedSonarModelsForSv(str, Enum): + EK60 = "EK60" + AZFP = "AZFP" + EK80 = "EK80" + + +class WaveformMode(str, Enum): + CW = "CW" + BB = "BB" + + +class EncodeMode(str, Enum): + COMPLEX = "complex" + POWER = "power" + + +class ComputeSVParams(BaseModel): + echodata: Any + env_params: Optional[dict] = None + cal_params: Optional[dict] = None + waveform_mode: Optional[WaveformMode] = None + encode_mode: Optional[EncodeMode] = None + + @field_validator("echodata") + def check_echodata_type(cls, value): + if not isinstance(value, EchoData): + raise ValueError( + "Invalid type for echodata. Expected an instance of EchoData." + ) + return value + + @field_validator("waveform_mode") + def check_waveform_mode(cls, waveform_mode, info: FieldValidationInfo): + echodata = info.data.get("echodata") + is_not_ek80 = echodata and echodata.sonar_model != "EK80" + if is_not_ek80 and waveform_mode is not None: + raise ValueError( + f"waveform_mode is only valid for EK80. \ + Got sonar_model='{echodata.sonar_model}'" + ) + return waveform_mode + + @field_validator("encode_mode") + def check_encode_mode(cls, encode_mode, info: FieldValidationInfo): + echodata = info.data.get("echodata") + is_not_ek80 = echodata and echodata.sonar_model != "EK80" + if is_not_ek80 and encode_mode is not None: + raise ValueError( + f"encode_mode is only valid for EK80. \ + Got sonar_model='{echodata.sonar_model}'" + ) + return encode_mode + + +def compute_sv(echodata: EchoData, **kwargs) -> xr.Dataset: + """ + Computes the volume backscattering strength (Sv) from the given echodata. + + Parameters: + - echodata (EchoData): The EchoData object containing + sonar data for computation. + - **kwargs: Additional keyword arguments passed to the Sv computation. + + Returns: + - xr.Dataset: A Dataset containing the computed Sv values. + + Example: + >>> sv_results = compute_sv(echodata_object) + >>> print(sv_results) + + Notes: + This function: + - Validates the `echodata`'s sonar model against supported models. + - Uses the `ComputeSVParams` pydantic model to validate parameters. + - Checks if the computed Sv is empty. + - Returns Sv only if it is not empty. + + """ + sonar_model = echodata.sonar_model + # Check if the sonar model is supported + try: + SupportedSonarModelsForSv(sonar_model) + except ValueError: + raise ValueError( + f"Sonar model '{sonar_model}'\ + is not supported for Sv computation.\ + Supported models are \ + {list(SupportedSonarModelsForSv)}." + ) + + # Validate parameters using the pydantic model + try: + ComputeSVParams(echodata=echodata, **kwargs) + except ValidationError as e: + raise ValueError(str(e)) + # Compute Sv + Sv = ep.calibrate.compute_Sv(echodata, **kwargs) + # Check if the computed Sv is empty + if Sv["Sv"].values.size == 0: + raise ValueError("Computed Sv is empty!") + return Sv diff --git a/requirements.txt b/requirements.txt index e157c6f..5a73281 100644 --- a/requirements.txt +++ b/requirements.txt @@ -41,4 +41,5 @@ sphinx_rtd_theme sphinxcontrib-mermaid twine wheel +pydantic echopype diff --git a/tests/conftest.py b/tests/conftest.py index 3fe3d4c..0d4f412 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,6 +1,8 @@ import os from ftplib import FTP +from pathlib import Path +import echopype as ep import pytest @@ -50,4 +52,37 @@ def ftp_data(): download_ftp_directory(ftp, FTP_PARTIAL_PATH, TEST_DATA_FOLDER) yield TEST_DATA_FOLDER # Optional: Cleanup after tests are done - # shutil.rmtree(TEST_DATA_FOLDER) \ No newline at end of file + # shutil.rmtree(TEST_DATA_FOLDER) + + +@pytest.fixture(scope="module") +def ed_ek_60_for_Sv(): + bucket = "ncei-wcsd-archive" + base_path = "data/raw/Bell_M._Shimada/SH1707/EK60/" + filename = "Summer2017-D20170620-T011027.raw" + rawdirpath = base_path + filename + + s3raw_fpath = f"s3://{bucket}/{rawdirpath}" + storage_opts = {"anon": True} + ed = ep.open_raw( + s3raw_fpath, + sonar_model="EK60", + storage_options=storage_opts + ) + return ed + + +# Read test raw data EK80 +@pytest.fixture(scope="module") +def ed_ek_80_for_Sv(): + base_url = "noaa-wcsd-pds.s3.amazonaws.com/" + path = "data/raw/Sally_Ride/SR1611/EK80/" + file_name = "D20161109-T163350.raw" + raw_file_address = base_url + path + file_name + + rf = Path(raw_file_address) + ed_EK80 = ep.open_raw( + f"https://{rf}", + sonar_model="EK80", + ) + return ed_EK80 diff --git a/tests/test_compute_sv.py b/tests/test_compute_sv.py new file mode 100644 index 0000000..166c116 --- /dev/null +++ b/tests/test_compute_sv.py @@ -0,0 +1,83 @@ +import pytest +from pydantic import ValidationError + +from oceanstream.L2_calibrated_data.compute_sv import ( + ComputeSVParams, SupportedSonarModelsForSv) + + +def test_valid_sonar_models(): + assert SupportedSonarModelsForSv("EK60") == SupportedSonarModelsForSv.EK60 + assert SupportedSonarModelsForSv("AZFP") == SupportedSonarModelsForSv.AZFP + assert SupportedSonarModelsForSv("EK80") == SupportedSonarModelsForSv.EK80 + + +def test_invalid_sonar_model(): + with pytest.raises(ValueError): + SupportedSonarModelsForSv("INVALID_MODEL") + + with pytest.raises(ValueError): + SupportedSonarModelsForSv("EK90") + + +def test_invalid_echodata_type(): + # Test with invalid echodata + with pytest.raises(ValidationError): + ComputeSVParams(echodata={"sonar_model": "dummy"}) + + +def test_waveform_mode_validity_for_ek80(ed_ek_60_for_Sv, ed_ek_80_for_Sv): + for ed in [ed_ek_60_for_Sv, ed_ek_80_for_Sv]: + # Using sonar model from real echodata + if ed.sonar_model == "EK80": + # This should pass + ComputeSVParams(echodata=ed, waveform_mode="CW") + else: + # This should raise ValidationError + # since waveform_mode is only valid for EK80 + with pytest.raises(ValidationError): + ComputeSVParams(echodata=ed, waveform_mode="CW") + + +def test_encode_mode_validator(ed_ek_60_for_Sv, ed_ek_80_for_Sv): + for ed in [ed_ek_60_for_Sv, ed_ek_80_for_Sv]: + # Using sonar model from real echodata + if ed.sonar_model == "EK80": + # This should pass + ComputeSVParams(echodata=ed, encode_mode="complex") + else: + # This should raise ValidationError since encode_mode + # is only valid for EK80 + with pytest.raises(ValidationError): + ComputeSVParams(echodata=ed, encode_mode="complex") + + +def test_env_params(ed_ek_60_for_Sv, ed_ek_80_for_Sv): + for ed in [ed_ek_60_for_Sv, ed_ek_80_for_Sv]: + # Test with typical env_params + env_params = {"temperature": "20"} + model = ComputeSVParams(echodata=ed, env_params=env_params) + assert model.env_params == env_params + + # Test with missing or None env_params + model = ComputeSVParams(echodata=ed, env_params=None) + assert model.env_params is None + + # Test with incorrect env_params + with pytest.raises(ValueError): + ComputeSVParams(echodata=ed, env_params="incorrect_value") + + +def test_cal_params(ed_ek_60_for_Sv, ed_ek_80_for_Sv): + for ed in [ed_ek_60_for_Sv, ed_ek_80_for_Sv]: + # Test with typical cal_params + cal_params = {"gain_correction": "0.5"} + model = ComputeSVParams(echodata=ed, cal_params=cal_params) + assert model.cal_params == cal_params + + # Test with missing or None cal_params + model = ComputeSVParams(echodata=ed, cal_params=None) + assert model.cal_params is None + + # Test with incorrect cal_params + with pytest.raises(ValueError): + ComputeSVParams(echodata=ed, cal_params="incorrect_value")