Fixed test and added a test data download utility

Author: beatfactor

echopype/consolidate/split_beam_angle.py

@@ -2,9 +2,9 @@
 Contains functions necessary to compute the split-beam (alongship/athwartship)
 angles and add them to a Dataset.
 """
-
 from typing import List, Tuple
 
+import dask.array as da
 import numpy as np
 import xarray as xr
 
@@ -111,8 +111,12 @@ def _compute_angle_from_complex(
     else:
         raise ValueError("beam_type not recognized!")
 
-    theta = theta / sens[0] - offset[0]
-    phi = phi / sens[1] - offset[1]
+    if isinstance(sens[0].data, da.Array):
+        theta = theta / float(sens[0].values) - float(offset[0].values)
+        phi = phi / float(sens[1].values) - float(offset[1].values)
+    else:
+        theta = theta / sens[0] - offset[0]
+        phi = phi / sens[1] - offset[1]
 
     return theta, phi
 
@@ -234,39 +238,46 @@ def get_angle_complex_samples(
     else:
         # beam_type different for some channels, process each channel separately
         theta_list, phi_list, valid_channels = [], [], []
-        beam_types = ds_beam["beam_type"].values
-
-        for idx, ch_id in enumerate(bs["channel"].data):
-            beam_type_ch = beam_types[idx]
-
-            if beam_type_ch not in SUPPORTED_BEAM_TYPES:
-                logger.warning(f"Skipping channel {ch_id}: unsupported beam_type {beam_type_ch}")
+        for ch_id in bs["channel"].data:
+            beam_type = ds_beam["beam_type"].sel(channel=ch_id)
+            beam_type = int(beam_type)
+            if beam_type not in SUPPORTED_BEAM_TYPES:
+                logger.warning(f"Skipping channel {ch_id}: unsupported beam_type {beam_type}")
                 continue
 
             theta_ch, phi_ch = _compute_angle_from_complex(
-                bs=bs.isel(channel=idx),
-                beam_type=beam_type_ch,
+                bs=bs.sel(channel=ch_id),
+                # beam_type is not time-varying
+                beam_type=beam_type,
                 sens=[
-                    float(angle_params["angle_sensitivity_alongship"].isel(channel=idx).values),
-                    float(angle_params["angle_sensitivity_athwartship"].isel(channel=idx).values),
+                    angle_params["angle_sensitivity_alongship"].sel(channel=ch_id),
+                    angle_params["angle_sensitivity_athwartship"].sel(channel=ch_id),
                 ],
                 offset=[
-                    float(angle_params["angle_offset_alongship"].isel(channel=idx).values),
-                    float(angle_params["angle_offset_athwartship"].isel(channel=idx).values),
+                    angle_params["angle_offset_alongship"].sel(channel=ch_id),
+                    angle_params["angle_offset_athwartship"].sel(channel=ch_id),
                 ],
             )
             theta_list.append(theta_ch)
             phi_list.append(phi_ch)
             valid_channels.append(ch_id)
 
-        if not theta_list:
-            raise ValueError("No valid channels found for angle computation.")
-
         # Combine angles from all channels
-        theta = xr.concat(theta_list, dim="channel")
-        theta = theta.assign_coords(channel=("channel", valid_channels))
-
-        phi = xr.concat(phi_list, dim="channel")
-        phi = phi.assign_coords(channel=("channel", valid_channels))
+        theta = xr.DataArray(
+            data=theta_list,
+            coords={
+                "channel": valid_channels,
+                "ping_time": bs["ping_time"],
+                "range_sample": bs["range_sample"],
+            },
+        )
+        phi = xr.DataArray(
+            data=phi_list,
+            coords={
+                "channel": valid_channels,
+                "ping_time": bs["ping_time"],
+                "range_sample": bs["range_sample"],
+            },
+        )
 
     return theta, phi

echopype/tests/consolidate/test_consolidate_integration.py

@@ -341,16 +341,12 @@ def test_add_splitbeam_angle_partial_valid_channels(test_path):
 
     # Manually override beam_type for one channel to simulate single-beam
     beam_types = ed["Sonar/Beam_group1"]["beam_type"].values
-    total_channels = beam_types.shape[0]
-    valid_channels = 0
     channel_to_invalidate = None
 
     # Force the first channel to an unsupported beam_type
     for idx, bt in enumerate(beam_types):
-        if bt != 0:
-            valid_channels += 1
-            if channel_to_invalidate is None:
-                channel_to_invalidate = idx
+        if bt != 0 and channel_to_invalidate is None:
+            channel_to_invalidate = idx
 
     # Simulate disabling one valid channel
     beam_types[channel_to_invalidate] = 0
@@ -363,7 +359,7 @@ def test_add_splitbeam_angle_partial_valid_channels(test_path):
     ds_Sv = ep.consolidate.add_splitbeam_angle(source_Sv=ds_Sv, echodata=ed, waveform_mode="CW",
                                                encode_mode="complex", pulse_compression=False, to_disk=False)
 
-    expected_valid_channels = valid_channels - 1
+    expected_valid_channels = len(ds_Sv.channel.values) - 1
 
     valid_angle_channels = [
         ch for ch in ds_Sv.channel.values

echopype/tests/download_test_data.py

@@ -0,0 +1,51 @@
+import subprocess
+import sys
+from pathlib import Path
+
+CONTAINER_NAME = "echopype-test-http-server"
+DOCKER_IMAGE = "cormorack/http"
+
+
+def check_docker():
+    try:
+        subprocess.run(["docker", "--version"], check=True, stdout=subprocess.DEVNULL)
+        subprocess.run(["docker", "info"], check=True, stdout=subprocess.DEVNULL)
+    except subprocess.CalledProcessError:
+        print("Error: Docker is not installed or the daemon is not running.")
+        sys.exit(1)
+
+
+def run_container():
+    subprocess.run(["docker", "run", "-d", "--name", CONTAINER_NAME, DOCKER_IMAGE], check=True)
+
+
+def copy_test_data(destination: Path):
+    destination.mkdir(parents=True, exist_ok=True)
+    subprocess.run([
+        "docker", "cp", "-L",
+        f"{CONTAINER_NAME}:/usr/local/apache2/htdocs/data/.",
+        str(destination)
+    ], check=True)
+
+
+def cleanup_container():
+    subprocess.run(["docker", "rm", "-f", CONTAINER_NAME], check=True)
+
+
+def main():
+    script_dir = Path(__file__).resolve().parent
+    destination_dir = script_dir.parent / "test_data"
+    check_docker()
+    run_container()
+
+    print(f"Copying test data to {destination_dir} ...")
+    copy_test_data(destination_dir)
+
+    print("Cleaning up container...")
+    cleanup_container()
+
+    print(f"Test data downloaded successfully to {destination_dir}")
+
+
+if __name__ == "__main__":
+    main()