diff --git a/gateway/sds_gateway/visualizations/api_views.py b/gateway/sds_gateway/visualizations/api_views.py
index b9f447d6..62be4906 100644
--- a/gateway/sds_gateway/visualizations/api_views.py
+++ b/gateway/sds_gateway/visualizations/api_views.py
@@ -1,5 +1,7 @@
"""API views for the visualizations app."""
+import json
+
from django.http import FileResponse
from django.shortcuts import get_object_or_404
from drf_spectacular.utils import OpenApiExample
@@ -593,6 +595,75 @@ def create_waterfall(self, request: Request, pk: str | None = None) -> Response:
status=status.HTTP_500_INTERNAL_SERVER_ERROR,
)
+ @extend_schema(
+ summary="Get waterfall metadata",
+ description="Get metadata for a waterfall visualization",
+ parameters=[
+ OpenApiParameter(
+ name="capture_uuid",
+ type=str,
+ location=OpenApiParameter.PATH,
+ description="UUID of the capture",
+ ),
+ OpenApiParameter(
+ name="job_id",
+ type=str,
+ location=OpenApiParameter.QUERY,
+ description="UUID of the processing job",
+ ),
+ ],
+ responses={
+ 200: OpenApiResponse(
+ description="Waterfall metadata",
+ ),
+ 404: OpenApiResponse(description="Job not found"),
+ },
+ )
+ @action(detail=True, methods=["get"], url_path="waterfall_metadata")
+ def get_waterfall_metadata(
+ self, request: Request, pk: str | None = None
+ ) -> Response:
+ """
+ Get metadata for a waterfall visualization.
+ """
+ job_id = self.get_request_param(request, "job_id", from_query=True)
+ if not job_id:
+ return Response(
+ {"error": "job_id parameter is required"},
+ status=status.HTTP_400_BAD_REQUEST,
+ )
+
+ # Get the post-processed data object
+ processed_data = get_object_or_404(
+ PostProcessedData,
+ uuid=job_id,
+ capture__uuid=pk,
+ processing_type=ProcessingType.Waterfall.value,
+ )
+
+ if processed_data.processing_status != ProcessingStatus.Completed.value:
+ return Response(
+ {"error": "Waterfall processing not completed"},
+ status=status.HTTP_400_BAD_REQUEST,
+ )
+
+ # Get metadata, with defaults if not present
+ metadata = processed_data.metadata or {}
+
+ if not metadata.get("total_slices"):
+ # Count the number of slices in the JSON file
+ with open(processed_data.data_file) as file:
+ num_slices = len(json.load(file))
+
+ metadata["slices_processed"] = num_slices
+ processed_data.metadata = metadata
+ processed_data.save()
+
+ return Response(
+ metadata,
+ status=status.HTTP_200_OK,
+ )
+
@extend_schema(
summary="Get waterfall status",
description="Get the status of a waterfall generation job",
@@ -651,7 +722,10 @@ def get_waterfall_status(self, request: Request, pk: str | None = None) -> Respo
@extend_schema(
summary="Download waterfall result",
- description="Download the generated waterfall data",
+ description=(
+ "Download the generated waterfall data. Supports range queries "
+ "with start_index and end_index parameters."
+ ),
parameters=[
OpenApiParameter(
name="capture_uuid",
@@ -665,11 +739,35 @@ def get_waterfall_status(self, request: Request, pk: str | None = None) -> Respo
location=OpenApiParameter.QUERY,
description="UUID of the processing job",
),
+ OpenApiParameter(
+ name="start_index",
+ type=int,
+ location=OpenApiParameter.QUERY,
+ required=False,
+ description=(
+ "Start index for range query (0-based). "
+ "If omitted, returns all data."
+ ),
+ ),
+ OpenApiParameter(
+ name="end_index",
+ type=int,
+ location=OpenApiParameter.QUERY,
+ required=False,
+ description=(
+ "End index for range query (exclusive). "
+ "If omitted, returns all data from start_index."
+ ),
+ ),
],
responses={
- 200: OpenApiResponse(description="Waterfall data file"),
+ 200: OpenApiResponse(
+ description="Waterfall data file or JSON response with range"
+ ),
404: OpenApiResponse(description="Result not found"),
- 400: OpenApiResponse(description="Processing not completed"),
+ 400: OpenApiResponse(
+ description="Processing not completed or invalid range"
+ ),
},
)
@action(detail=True, methods=["get"], url_path="download_waterfall")
@@ -678,6 +776,11 @@ def download_waterfall(
) -> Response | FileResponse:
"""
Download the generated waterfall data.
+
+ Supports range queries:
+ - If start_index and/or end_index are provided, returns only the
+ requested slice range
+ - Otherwise, returns the full dataset (for backward compatibility)
"""
job_id = self.get_request_param(request, "job_id", from_query=True)
if not job_id:
@@ -686,8 +789,8 @@ def download_waterfall(
status=status.HTTP_400_BAD_REQUEST,
)
- # Get the processing job
- processing_job = get_object_or_404(
+ # Get the post-processed data object
+ processed_data = get_object_or_404(
PostProcessedData,
uuid=job_id,
capture__uuid=pk,
@@ -695,27 +798,113 @@ def download_waterfall(
)
try:
- if processing_job.processing_status != ProcessingStatus.Completed.value:
+ if processed_data.processing_status != ProcessingStatus.Completed.value:
return Response(
{"error": "Waterfall processing not completed"},
status=status.HTTP_400_BAD_REQUEST,
)
- if not processing_job.data_file:
+ if not processed_data.data_file:
return Response(
{"error": "No waterfall file found"},
status=status.HTTP_404_NOT_FOUND,
)
- # Return the file
- file_response = FileResponse(
- processing_job.data_file, content_type="application/json"
+ # Check for range query parameters
+ start_index = self.get_request_param(
+ request, "start_index", from_query=True
)
- file_response["Content-Disposition"] = (
- f'attachment; filename="waterfall_{pk}.json"'
- )
- return file_response # noqa: TRY300
+ end_index = self.get_request_param(request, "end_index", from_query=True)
+ # If no range parameters specified, return full file
+ # (backward compatibility)
+ if start_index is None and end_index is None:
+ file_response = FileResponse(
+ processed_data.data_file, content_type="application/json"
+ )
+ file_response["Content-Disposition"] = (
+ f'attachment; filename="waterfall_{pk}.json"'
+ )
+ return file_response
+
+ # Read the JSON file for range queries
+ processed_data.data_file.seek(0)
+ waterfall_data = json.load(processed_data.data_file)
+
+ # Get total slices from metadata if available, otherwise from data length
+ total_slices = len(waterfall_data)
+ if processed_data.metadata and "total_slices" in processed_data.metadata:
+ total_slices = processed_data.metadata["total_slices"]
+ elif (
+ processed_data.metadata
+ and "slices_processed" in processed_data.metadata
+ ):
+ total_slices = processed_data.metadata["slices_processed"]
+
+ # Handle range query
+ if start_index is not None or end_index is not None:
+ try:
+ start_idx = int(start_index) if start_index is not None else 0
+ end_idx = int(end_index) if end_index is not None else total_slices
+ except (ValueError, TypeError):
+ return Response(
+ {
+ "error": (
+ "Invalid start_index or end_index. Must be integers."
+ )
+ },
+ status=status.HTTP_400_BAD_REQUEST,
+ )
+
+ # Validate range
+ if start_idx < 0 or end_idx < 0:
+ return Response(
+ {"error": ("start_index and end_index must be non-negative")},
+ status=status.HTTP_400_BAD_REQUEST,
+ )
+ if start_idx > end_idx:
+ return Response(
+ {
+ "error": (
+ "start_index must be less than or equal to end_index"
+ )
+ },
+ status=status.HTTP_400_BAD_REQUEST,
+ )
+ if start_idx >= total_slices:
+ error_msg = (
+ f"start_index {start_idx} is out of range. "
+ f"Total slices: {total_slices}"
+ )
+ return Response(
+ {"error": error_msg},
+ status=status.HTTP_400_BAD_REQUEST,
+ )
+
+ # Clamp end_index to total_slices
+ end_idx = min(end_idx, total_slices)
+
+ # Slice the data
+ sliced_data = waterfall_data[start_idx:end_idx]
+
+ # Return JSON response with range info
+ return Response(
+ {
+ "data": sliced_data,
+ "start_index": start_idx,
+ "end_index": end_idx,
+ "total_slices": total_slices,
+ "count": len(sliced_data),
+ },
+ status=status.HTTP_200_OK,
+ )
+
+ except json.JSONDecodeError as e:
+ log.error(f"Error parsing waterfall JSON: {e}")
+ return Response(
+ {"error": "Invalid JSON data in waterfall file"},
+ status=status.HTTP_500_INTERNAL_SERVER_ERROR,
+ )
except Exception as e: # noqa: BLE001
log.error(f"Error downloading waterfall: {e}")
return Response(
diff --git a/gateway/sds_gateway/visualizations/processing/waterfall.py b/gateway/sds_gateway/visualizations/processing/waterfall.py
index cfe2365d..2ea3483e 100644
--- a/gateway/sds_gateway/visualizations/processing/waterfall.py
+++ b/gateway/sds_gateway/visualizations/processing/waterfall.py
@@ -185,10 +185,29 @@ def convert_drf_to_waterfall_json(
msg = "No valid waterfall slices found"
raise SourceDataError(msg)
+ # Calculate power bounds from all slices
+ global_min = float("inf")
+ global_max = float("-inf")
+
+ for slice_data in waterfall_data:
+ # Decode the base64 data to calculate bounds
+ data_bytes = base64.b64decode(slice_data["data"])
+ power_spectrum_db = np.frombuffer(data_bytes, dtype=np.float32)
+
+ slice_min = float(np.min(power_spectrum_db))
+ slice_max = float(np.max(power_spectrum_db))
+
+ global_min = min(global_min, slice_min)
+ global_max = max(global_max, slice_max)
+
+ power_scale_min = global_min if global_min != float("inf") else None
+ power_scale_max = global_max if global_max != float("-inf") else None
+
# Log final summary
logger.info(
f"Waterfall processing complete: {len(waterfall_data)} slices processed, "
- f"{skipped_slices} slices skipped due to data issues"
+ f"{skipped_slices} slices skipped due to data issues. "
+ f"Power bounds: [{power_scale_min:.2f}, {power_scale_max:.2f}] dB"
)
metadata = {
@@ -202,6 +221,8 @@ def convert_drf_to_waterfall_json(
"fft_size": base_params.fft_size,
"samples_per_slice": SAMPLES_PER_SLICE,
"channel": channel,
+ "power_scale_min": power_scale_min,
+ "power_scale_max": power_scale_max,
}
return {
diff --git a/gateway/sds_gateway/visualizations/tests/test_views.py b/gateway/sds_gateway/visualizations/tests/test_views.py
index ea63d6f3..ec22c034 100644
--- a/gateway/sds_gateway/visualizations/tests/test_views.py
+++ b/gateway/sds_gateway/visualizations/tests/test_views.py
@@ -157,6 +157,11 @@ def setUp(self) -> None:
kwargs={"pk": self.capture.uuid},
)
+ self.get_waterfall_metadata_url = reverse(
+ "api:visualizations-get-waterfall-metadata",
+ kwargs={"pk": self.capture.uuid},
+ )
+
def test_create_waterfall_api_requires_authentication(self) -> None:
"""Test that create_waterfall API requires authentication."""
response = self.client.post(self.create_waterfall_url)
@@ -328,6 +333,510 @@ def test_download_waterfall_api_success(self) -> None:
content = b"".join(response.streaming_content).decode()
assert content == test_content
+ def test_get_waterfall_metadata_api_requires_authentication(self) -> None:
+ """Test that get_waterfall_metadata API requires authentication."""
+ response = self.client.get(self.get_waterfall_metadata_url)
+ assert response.status_code == status.HTTP_403_FORBIDDEN
+
+ def test_get_waterfall_metadata_api_missing_job_id(self) -> None:
+ """Test that missing job_id parameter returns 400 for API."""
+ self.client.force_login(self.user)
+
+ response = self.client.get(self.get_waterfall_metadata_url)
+ assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+ data = response.json()
+ assert "error" in data
+ assert "job_id parameter is required" in data["error"]
+
+ def test_get_waterfall_metadata_api_success_with_existing_metadata(self) -> None:
+ """Test successful waterfall metadata retrieval with existing metadata."""
+ self.client.force_login(self.user)
+
+ # Create test file content (array of slices)
+ test_slices = [{"freq": i, "data": [1, 2, 3]} for i in range(10)]
+ test_content = json.dumps(test_slices)
+ test_file = SimpleUploadedFile(
+ "waterfall_test.json",
+ test_content.encode(),
+ content_type="application/json",
+ )
+
+ # Create completed waterfall with metadata already set
+ waterfall_data = PostProcessedData.objects.create(
+ capture=self.capture,
+ processing_type=ProcessingType.Waterfall.value,
+ processing_parameters={},
+ processing_status=ProcessingStatus.Completed.value,
+ metadata={"total_slices": 10, "sample_rate": 1000000},
+ data_file=test_file,
+ )
+
+ response = self.client.get(
+ self.get_waterfall_metadata_url,
+ {"job_id": str(waterfall_data.uuid)},
+ )
+
+ assert response.status_code == status.HTTP_200_OK
+ data = response.json()
+ assert data["total_slices"] == 10
+ assert data["sample_rate"] == 1000000
+
+ def test_get_waterfall_metadata_api_calculates_slices_from_file(self) -> None:
+ """Test that metadata slices are calculated from file if not present."""
+ self.client.force_login(self.user)
+
+ # Create test file content (array of slices)
+ test_slices = [{"freq": i, "data": [1, 2, 3]} for i in range(15)]
+ test_content = json.dumps(test_slices)
+ test_file = SimpleUploadedFile(
+ "waterfall_test.json",
+ test_content.encode(),
+ content_type="application/json",
+ )
+
+ # Create completed waterfall without total_slices in metadata
+ waterfall_data = PostProcessedData.objects.create(
+ capture=self.capture,
+ processing_type=ProcessingType.Waterfall.value,
+ processing_parameters={},
+ processing_status=ProcessingStatus.Completed.value,
+ metadata={"sample_rate": 1000000}, # No total_slices
+ data_file=test_file,
+ )
+
+ response = self.client.get(
+ self.get_waterfall_metadata_url,
+ {"job_id": str(waterfall_data.uuid)},
+ )
+
+ assert response.status_code == status.HTTP_200_OK
+ data = response.json()
+ assert data["slices_processed"] == 15
+ assert data["sample_rate"] == 1000000
+
+ # Verify that the metadata was saved back to the object
+ waterfall_data.refresh_from_db()
+ assert waterfall_data.metadata["slices_processed"] == 15
+
+ def test_get_waterfall_metadata_api_job_not_found(self) -> None:
+ """Test that non-existent job returns 404."""
+ self.client.force_login(self.user)
+
+ fake_job_id = "00000000-0000-0000-0000-000000000000"
+ response = self.client.get(
+ self.get_waterfall_metadata_url, {"job_id": fake_job_id}
+ )
+ assert response.status_code == status.HTTP_404_NOT_FOUND
+
+ def test_get_waterfall_metadata_api_processing_not_completed(self) -> None:
+ """Test that incomplete processing returns 400."""
+ self.client.force_login(self.user)
+
+ # Create waterfall processing job that's still processing
+ waterfall_data = PostProcessedData.objects.create(
+ capture=self.capture,
+ processing_type=ProcessingType.Waterfall.value,
+ processing_parameters={},
+ processing_status=ProcessingStatus.Processing.value,
+ metadata={},
+ )
+
+ response = self.client.get(
+ self.get_waterfall_metadata_url,
+ {"job_id": str(waterfall_data.uuid)},
+ )
+
+ assert response.status_code == status.HTTP_400_BAD_REQUEST
+ data = response.json()
+ assert "error" in data
+ assert "Waterfall processing not completed" in data["error"]
+
+ def test_get_waterfall_metadata_api_capture_not_owned(self) -> None:
+ """Test that users cannot get metadata for others' captures."""
+ other_user = User.objects.create(
+ email="otheruser@example.com",
+ password="testpassword", # noqa: S106
+ is_approved=True,
+ )
+
+ self.client.force_login(self.user)
+
+ # Create waterfall data for other user's capture
+ other_capture = Capture.objects.create(
+ capture_type=CaptureType.DigitalRF,
+ channel="ch0",
+ index_name="other-index",
+ owner=other_user,
+ top_level_dir="/other/dir",
+ )
+
+ waterfall_data = PostProcessedData.objects.create(
+ capture=other_capture,
+ processing_type=ProcessingType.Waterfall.value,
+ processing_parameters={},
+ processing_status=ProcessingStatus.Completed.value,
+ metadata={"total_slices": 10},
+ )
+
+ other_capture_url = reverse(
+ "api:visualizations-get-waterfall-metadata",
+ kwargs={"pk": other_capture.uuid},
+ )
+
+ response = self.client.get(
+ other_capture_url, {"job_id": str(waterfall_data.uuid)}
+ )
+ assert response.status_code == status.HTTP_404_NOT_FOUND
+
+ def test_download_waterfall_api_with_start_index(self) -> None:
+ """Test download_waterfall with start_index parameter."""
+ self.client.force_login(self.user)
+
+ # Create test file content with multiple slices
+ test_slices = [
+ {"index": i, "data": [i * 10 + j for j in range(5)]} for i in range(20)
+ ]
+ test_content = json.dumps(test_slices)
+ test_file = SimpleUploadedFile(
+ "waterfall_test.json",
+ test_content.encode(),
+ content_type="application/json",
+ )
+
+ waterfall_data = PostProcessedData.objects.create(
+ capture=self.capture,
+ processing_type=ProcessingType.Waterfall.value,
+ processing_parameters={},
+ processing_status=ProcessingStatus.Completed.value,
+ metadata={"total_slices": 20},
+ data_file=test_file,
+ )
+
+ # Request slices from index 5 to the end
+ response = self.client.get(
+ self.download_waterfall_url,
+ {"job_id": str(waterfall_data.uuid), "start_index": 5},
+ )
+
+ assert response.status_code == status.HTTP_200_OK
+ data = response.json()
+ assert "data" in data
+ assert "start_index" in data
+ assert "end_index" in data
+ assert "total_slices" in data
+ assert "count" in data
+
+ assert data["start_index"] == 5
+ assert data["end_index"] == 20
+ assert data["total_slices"] == 20
+ assert data["count"] == 15
+ assert len(data["data"]) == 15
+ assert data["data"][0]["index"] == 5
+ assert data["data"][-1]["index"] == 19
+
+ def test_download_waterfall_api_with_end_index(self) -> None:
+ """Test download_waterfall with end_index parameter."""
+ self.client.force_login(self.user)
+
+ # Create test file content with multiple slices
+ test_slices = [
+ {"index": i, "data": [i * 10 + j for j in range(5)]} for i in range(20)
+ ]
+ test_content = json.dumps(test_slices)
+ test_file = SimpleUploadedFile(
+ "waterfall_test.json",
+ test_content.encode(),
+ content_type="application/json",
+ )
+
+ waterfall_data = PostProcessedData.objects.create(
+ capture=self.capture,
+ processing_type=ProcessingType.Waterfall.value,
+ processing_parameters={},
+ processing_status=ProcessingStatus.Completed.value,
+ metadata={"total_slices": 20},
+ data_file=test_file,
+ )
+
+ # Request slices from start to index 10
+ response = self.client.get(
+ self.download_waterfall_url,
+ {"job_id": str(waterfall_data.uuid), "end_index": 10},
+ )
+
+ assert response.status_code == status.HTTP_200_OK
+ data = response.json()
+ assert data["start_index"] == 0
+ assert data["end_index"] == 10
+ assert data["total_slices"] == 20
+ assert data["count"] == 10
+ assert len(data["data"]) == 10
+ assert data["data"][0]["index"] == 0
+ assert data["data"][-1]["index"] == 9
+
+ def test_download_waterfall_api_with_range(self) -> None:
+ """Test download_waterfall with both start_index and end_index."""
+ self.client.force_login(self.user)
+
+ # Create test file content with multiple slices
+ test_slices = [
+ {"index": i, "data": [i * 10 + j for j in range(5)]} for i in range(20)
+ ]
+ test_content = json.dumps(test_slices)
+ test_file = SimpleUploadedFile(
+ "waterfall_test.json",
+ test_content.encode(),
+ content_type="application/json",
+ )
+
+ waterfall_data = PostProcessedData.objects.create(
+ capture=self.capture,
+ processing_type=ProcessingType.Waterfall.value,
+ processing_parameters={},
+ processing_status=ProcessingStatus.Completed.value,
+ metadata={"total_slices": 20},
+ data_file=test_file,
+ )
+
+ # Request slices from index 5 to 15
+ response = self.client.get(
+ self.download_waterfall_url,
+ {
+ "job_id": str(waterfall_data.uuid),
+ "start_index": 5,
+ "end_index": 15,
+ },
+ )
+
+ assert response.status_code == status.HTTP_200_OK
+ data = response.json()
+ assert data["start_index"] == 5
+ assert data["end_index"] == 15
+ assert data["total_slices"] == 20
+ assert data["count"] == 10
+ assert len(data["data"]) == 10
+ assert data["data"][0]["index"] == 5
+ assert data["data"][-1]["index"] == 14
+
+ def test_download_waterfall_api_invalid_range_start_greater_than_end(self) -> None:
+ """Test that start_index > end_index returns 400."""
+ self.client.force_login(self.user)
+
+ # Create test file content
+ test_slices = [{"index": i} for i in range(10)]
+ test_content = json.dumps(test_slices)
+ test_file = SimpleUploadedFile(
+ "waterfall_test.json",
+ test_content.encode(),
+ content_type="application/json",
+ )
+
+ waterfall_data = PostProcessedData.objects.create(
+ capture=self.capture,
+ processing_type=ProcessingType.Waterfall.value,
+ processing_parameters={},
+ processing_status=ProcessingStatus.Completed.value,
+ metadata={"total_slices": 10},
+ data_file=test_file,
+ )
+
+ response = self.client.get(
+ self.download_waterfall_url,
+ {
+ "job_id": str(waterfall_data.uuid),
+ "start_index": 7,
+ "end_index": 5,
+ },
+ )
+
+ assert response.status_code == status.HTTP_400_BAD_REQUEST
+ data = response.json()
+ assert "error" in data
+ assert "start_index must be less than or equal to end_index" in data["error"]
+
+ def test_download_waterfall_api_start_index_out_of_bounds(self) -> None:
+ """Test that start_index >= total_slices returns 400."""
+ self.client.force_login(self.user)
+
+ # Create test file content
+ test_slices = [{"index": i} for i in range(10)]
+ test_content = json.dumps(test_slices)
+ test_file = SimpleUploadedFile(
+ "waterfall_test.json",
+ test_content.encode(),
+ content_type="application/json",
+ )
+
+ waterfall_data = PostProcessedData.objects.create(
+ capture=self.capture,
+ processing_type=ProcessingType.Waterfall.value,
+ processing_parameters={},
+ processing_status=ProcessingStatus.Completed.value,
+ metadata={"total_slices": 10},
+ data_file=test_file,
+ )
+
+ response = self.client.get(
+ self.download_waterfall_url,
+ {
+ "job_id": str(waterfall_data.uuid),
+ "start_index": 10,
+ },
+ )
+
+ assert response.status_code == status.HTTP_400_BAD_REQUEST
+ data = response.json()
+ assert "error" in data
+ assert "start_index 10 is out of range" in data["error"]
+ assert "Total slices: 10" in data["error"]
+
+ def test_download_waterfall_api_negative_indices(self) -> None:
+ """Test that negative indices return 400."""
+ self.client.force_login(self.user)
+
+ # Create test file content
+ test_slices = [{"index": i} for i in range(10)]
+ test_content = json.dumps(test_slices)
+ test_file = SimpleUploadedFile(
+ "waterfall_test.json",
+ test_content.encode(),
+ content_type="application/json",
+ )
+
+ waterfall_data = PostProcessedData.objects.create(
+ capture=self.capture,
+ processing_type=ProcessingType.Waterfall.value,
+ processing_parameters={},
+ processing_status=ProcessingStatus.Completed.value,
+ metadata={"total_slices": 10},
+ data_file=test_file,
+ )
+
+ response = self.client.get(
+ self.download_waterfall_url,
+ {
+ "job_id": str(waterfall_data.uuid),
+ "start_index": -1,
+ },
+ )
+
+ assert response.status_code == status.HTTP_400_BAD_REQUEST
+ data = response.json()
+ assert "error" in data
+ assert "must be non-negative" in data["error"]
+
+ def test_download_waterfall_api_invalid_index_type(self) -> None:
+ """Test that non-integer indices return 400."""
+ self.client.force_login(self.user)
+
+ # Create test file content
+ test_slices = [{"index": i} for i in range(10)]
+ test_content = json.dumps(test_slices)
+ test_file = SimpleUploadedFile(
+ "waterfall_test.json",
+ test_content.encode(),
+ content_type="application/json",
+ )
+
+ waterfall_data = PostProcessedData.objects.create(
+ capture=self.capture,
+ processing_type=ProcessingType.Waterfall.value,
+ processing_parameters={},
+ processing_status=ProcessingStatus.Completed.value,
+ metadata={"total_slices": 10},
+ data_file=test_file,
+ )
+
+ response = self.client.get(
+ self.download_waterfall_url,
+ {
+ "job_id": str(waterfall_data.uuid),
+ "start_index": "not_a_number",
+ },
+ )
+
+ assert response.status_code == status.HTTP_400_BAD_REQUEST
+ data = response.json()
+ assert "error" in data
+ assert "Must be integers" in data["error"]
+
+ def test_download_waterfall_api_end_index_clamped_to_total(self) -> None:
+ """Test that end_index is clamped to total_slices."""
+ self.client.force_login(self.user)
+
+ # Create test file content
+ test_slices = [{"index": i} for i in range(10)]
+ test_content = json.dumps(test_slices)
+ test_file = SimpleUploadedFile(
+ "waterfall_test.json",
+ test_content.encode(),
+ content_type="application/json",
+ )
+
+ waterfall_data = PostProcessedData.objects.create(
+ capture=self.capture,
+ processing_type=ProcessingType.Waterfall.value,
+ processing_parameters={},
+ processing_status=ProcessingStatus.Completed.value,
+ metadata={"total_slices": 10},
+ data_file=test_file,
+ )
+
+ # Request end_index beyond total_slices
+ response = self.client.get(
+ self.download_waterfall_url,
+ {
+ "job_id": str(waterfall_data.uuid),
+ "start_index": 5,
+ "end_index": 20,
+ },
+ )
+
+ assert response.status_code == status.HTTP_200_OK
+ data = response.json()
+ assert data["start_index"] == 5
+ assert data["end_index"] == 10 # Clamped to total_slices
+ assert data["total_slices"] == 10
+ assert data["count"] == 5
+ assert len(data["data"]) == 5
+
+ def test_download_waterfall_api_backward_compatibility_no_params(self) -> None:
+ """Test that download_waterfall without range params returns full file."""
+ self.client.force_login(self.user)
+
+ # Create test file content
+ test_slices = [{"index": i} for i in range(10)]
+ test_content = json.dumps(test_slices)
+ test_file = SimpleUploadedFile(
+ "waterfall_test.json",
+ test_content.encode(),
+ content_type="application/json",
+ )
+
+ waterfall_data = PostProcessedData.objects.create(
+ capture=self.capture,
+ processing_type=ProcessingType.Waterfall.value,
+ processing_parameters={},
+ processing_status=ProcessingStatus.Completed.value,
+ metadata={},
+ data_file=test_file,
+ )
+
+ # Request without range parameters (backward compatibility)
+ response = self.client.get(
+ self.download_waterfall_url,
+ {"job_id": str(waterfall_data.uuid)},
+ )
+
+ assert response.status_code == status.HTTP_200_OK
+ assert response.get("Content-Type") == "application/json"
+ assert "attachment" in response.get("Content-Disposition", "")
+
+ # Verify it returns the full file content (not JSON response)
+ content = b"".join(response.streaming_content).decode()
+ assert content == test_content
+
class SpectrogramVisualizationViewTestCases(TestCase):
"""Test cases for SpectrogramVisualizationView."""