From 2b3748da4b0e8de0d399990e9d1583bc51f997f3 Mon Sep 17 00:00:00 2001 From: klpoland Date: Fri, 20 Feb 2026 14:16:47 -0500 Subject: [PATCH 1/7] adding time based filtering and slider ui --- .../api_methods/helpers/temporal_filtering.py | 168 ++++++++++++++++++ .../serializers/capture_serializers.py | 56 +++++- gateway/sds_gateway/api_methods/tasks.py | 62 ++++++- .../js/actions/DownloadActionManager.js | 13 +- gateway/sds_gateway/static/js/file-list.js | 6 +- gateway/sds_gateway/templates/base.html | 3 + .../templates/users/file_list.html | 2 + .../users/partials/captures_page_table.html | 4 +- .../users/partials/web_download_modal.html | 164 +++++++++++++++++ gateway/sds_gateway/users/views.py | 11 ++ 10 files changed, 476 insertions(+), 13 deletions(-) create mode 100644 gateway/sds_gateway/api_methods/helpers/temporal_filtering.py diff --git a/gateway/sds_gateway/api_methods/helpers/temporal_filtering.py b/gateway/sds_gateway/api_methods/helpers/temporal_filtering.py new file mode 100644 index 000000000..16b648d91 --- /dev/null +++ b/gateway/sds_gateway/api_methods/helpers/temporal_filtering.py @@ -0,0 +1,168 @@ +import re + +from django.db.models import QuerySet + +from sds_gateway.api_methods.models import CaptureType, Capture, File +from sds_gateway.api_methods.utils.opensearch_client import get_opensearch_client +from sds_gateway.api_methods.utils.relationship_utils import get_capture_files +from loguru import logger as log + +# Digital RF spec: rf@SECONDS.MILLISECONDS.h5 (e.g. rf@1396379502.000.h5) +# https://github.com/MITHaystack/digital_rf +DRF_RF_FILENAME_PATTERN = re.compile( + r"^rf@(\d+)\.(\d+)\.h5$", + re.IGNORECASE, +) +DRF_RF_FILENAME_REGEX_STR = r"^rf@\d+\.\d+\.h5$" + + +def drf_rf_filename_from_ms(ms: int) -> str: + """Format ms as DRF rf data filename (canonical for range queries).""" + return f"rf@{ms // 1000}.{ms % 1000:03d}.h5" + + +def drf_rf_filename_to_ms(file_name: str) -> int | None: + """ + Parse DRF rf data filename to milliseconds. + Handles rf@SECONDS.MILLISECONDS.h5; fractional part padded to 3 digits. + """ + name = file_name.strip() + match = DRF_RF_FILENAME_PATTERN.match(name) + if not match: + return None + try: + seconds = int(match.group(1)) + frac = match.group(2).ljust(3, "0")[:3] + return seconds * 1000 + int(frac) + except (ValueError, TypeError): + return None + + +def _catch_capture_type_error(capture_type: CaptureType) -> None: + if capture_type != CaptureType.DigitalRF: + msg = "Only DigitalRF captures are supported for temporal filtering." + log.error(msg) + raise ValueError(msg) + + +def _parse_drf_rf_timestamp(file_name: str) -> int | None: + """Extract timestamp in ms from a Digital RF data filename (alias for drf_rf_filename_to_ms).""" + return drf_rf_filename_to_ms(file_name) + + +def get_capture_bounds(capture_type: CaptureType, capture_uuid: str) -> tuple[int, int]: + """Get start and end bounds for capture from opensearch.""" + + _catch_capture_type_error(capture_type) + + client = get_opensearch_client() + index = f"captures-{capture_type}" + + try: + response = client.get(index=index, id=capture_uuid) + except Exception as e: + if getattr(e, "status_code", None) == 404 or (hasattr(e, "info") and e.info.get("status") == 404): + raise ValueError( + f"Capture {capture_uuid} not found in OpenSearch index {index}" + ) from e + raise + + if not response.get("found"): + raise ValueError( + f"Capture {capture_uuid} not found in OpenSearch index {index}" + ) + + source = response["_source"] + search_props = source["search_props"] + start_time = search_props["start_time"] + end_time = search_props["end_time"] + print(f"start_time: {start_time}, end_time: {end_time}") + return start_time, end_time + + +def get_data_files(capture_type: CaptureType, capture: Capture) -> QuerySet[File]: + """Get the data files in the capture.""" + _catch_capture_type_error(capture_type) + + return get_capture_files(capture).filter(name__regex=DRF_RF_FILENAME_REGEX_STR) + + +def get_file_cadence(capture_type: CaptureType, capture: Capture) -> int: + """Get the file cadence in milliseconds. OpenSearch bounds are in seconds.""" + _catch_capture_type_error(capture_type) + + capture_uuid = str(capture.uuid) + try: + start_time, end_time = get_capture_bounds(capture_type, capture_uuid) + except ValueError as e: + log.error(e) + raise e + + data_files = get_data_files(capture_type, capture) + count = data_files.count() + if count == 0: + return 0 + duration_sec = end_time - start_time + print(f"duration_sec: {duration_sec}") + duration_ms = duration_sec * 1000 + print(f"duration_ms: {duration_ms}") + return max(1, int(duration_ms / count)) + + +def get_duration_bounds(capture_type: CaptureType, capture_uuid: str, relative_time: int) -> tuple[int, int]: + """Return (0, length_of_capture_ms). OpenSearch bounds are in seconds.""" + try: + start_time, end_time = get_capture_bounds(capture_type, capture_uuid) + except ValueError as e: + log.error(e) + raise e + + length_of_capture_ms = (end_time - start_time) * 1000 + return 0, length_of_capture_ms + + +def filter_capture_data_files_selection_bounds( + capture_type: CaptureType, + capture: Capture, + start_time: int, # relative ms from start of capture (from UI) + end_time: int, # relative ms from start of capture (from UI) +) -> QuerySet[File]: + """Filter the capture file selection bounds to the given start and end times.""" + _catch_capture_type_error(capture_type) + epoch_start_sec, _ = get_capture_bounds(capture_type, capture.uuid) + epoch_start_ms = epoch_start_sec * 1000 + start_ms = epoch_start_ms + start_time + end_ms = epoch_start_ms + end_time + + start_file_name = drf_rf_filename_from_ms(start_ms) + end_file_name = drf_rf_filename_from_ms(end_ms) + + data_files = get_data_files(capture_type, capture) + return data_files.filter( + name__gte=start_file_name, + name__lte=end_file_name, + ).order_by("name") + +def get_capture_files_with_temporal_filter( + capture_type: CaptureType, + capture: Capture, + start_time: int | None = None, # milliseconds since epoch (start of capture) + end_time: int | None = None, # milliseconds since epoch +) -> QuerySet[File]: + """Get the capture files with temporal filtering.""" + _catch_capture_type_error(capture_type) + + if start_time is None or end_time is None: + log.warning("Start or end time is None, returning all capture files without temporal filtering") + return get_capture_files(capture) + + # get non-data files + non_data_files = get_capture_files(capture).exclude(name__regex=DRF_RF_FILENAME_REGEX_STR) + + # get data files with temporal filtering + data_files = filter_capture_data_files_selection_bounds( + capture_type, capture, start_time, end_time + ) + + # return all files + return non_data_files.union(data_files) \ No newline at end of file diff --git a/gateway/sds_gateway/api_methods/serializers/capture_serializers.py b/gateway/sds_gateway/api_methods/serializers/capture_serializers.py index 037ebafd3..692628483 100644 --- a/gateway/sds_gateway/api_methods/serializers/capture_serializers.py +++ b/gateway/sds_gateway/api_methods/serializers/capture_serializers.py @@ -9,6 +9,8 @@ from rest_framework.utils.serializer_helpers import ReturnList from sds_gateway.api_methods.helpers.index_handling import retrieve_indexed_metadata +from sds_gateway.api_methods.helpers.temporal_filtering import get_capture_bounds +from sds_gateway.api_methods.helpers.temporal_filtering import get_file_cadence from sds_gateway.api_methods.models import Capture from sds_gateway.api_methods.models import CaptureType from sds_gateway.api_methods.models import DEPRECATEDPostProcessedData @@ -70,6 +72,8 @@ class CaptureGetSerializer(serializers.ModelSerializer[Capture]): files = serializers.SerializerMethodField() center_frequency_ghz = serializers.SerializerMethodField() sample_rate_mhz = serializers.SerializerMethodField() + length_of_capture_ms = serializers.SerializerMethodField() + file_cadence_ms = serializers.SerializerMethodField() files_count = serializers.SerializerMethodField() total_file_size = serializers.SerializerMethodField() formatted_created_at = serializers.SerializerMethodField() @@ -94,12 +98,29 @@ def get_files(self, capture: Capture) -> ReturnList[File]: def get_center_frequency_ghz(self, capture: Capture) -> float | None: """Get the center frequency in GHz from the capture model property.""" return capture.center_frequency_ghz - - @extend_schema_field(serializers.FloatField) + + @extend_schema_field(serializers.FloatField(allow_null=True)) def get_sample_rate_mhz(self, capture: Capture) -> float | None: - """Get the sample rate in MHz from the capture model property.""" + """Get the sample rate in MHz from the capture model property. None if not indexed in OpenSearch.""" return capture.sample_rate_mhz + @extend_schema_field(serializers.IntegerField(allow_null=True)) + def get_length_of_capture_ms(self, capture: Capture) -> int | None: + """Get the length of the capture in milliseconds. OpenSearch bounds are in seconds.""" + try: + start_time, end_time = get_capture_bounds(capture.capture_type, str(capture.uuid)) + return (end_time - start_time) * 1000 + except (ValueError, IndexError, KeyError): + return None + + @extend_schema_field(serializers.IntegerField(allow_null=True)) + def get_file_cadence_ms(self, capture: Capture) -> int | None: + """Get the file cadence in milliseconds. None if not indexed in OpenSearch.""" + try: + return get_file_cadence(capture.capture_type, capture) + except (ValueError, IndexError, KeyError): + return None + @extend_schema_field(serializers.IntegerField) def get_files_count(self, capture: Capture) -> int: """Get the count of files associated with this capture.""" @@ -304,6 +325,8 @@ class CompositeCaptureSerializer(serializers.Serializer): files_count = serializers.SerializerMethodField() total_file_size = serializers.SerializerMethodField() formatted_created_at = serializers.SerializerMethodField() + length_of_capture_ms = serializers.SerializerMethodField() + file_cadence_ms = serializers.SerializerMethodField() def get_files(self, obj: dict[str, Any]) -> ReturnList[File]: """Get all files from all channels in the composite capture.""" @@ -350,6 +373,33 @@ def get_formatted_created_at(self, obj: dict[str, Any]) -> str: return created_at.strftime("%m/%d/%Y %I:%M:%S %p") return "" + @extend_schema_field(serializers.IntegerField(allow_null=True)) + def get_length_of_capture_ms(self, obj: dict[str, Any]) -> int | None: + """Use first channel's bounds for composite capture duration.""" + channels = obj.get("channels") or [] + if not channels: + return None + try: + capture = Capture.objects.get(uuid=channels[0]["uuid"]) + start_time, end_time = get_capture_bounds( + capture.capture_type, str(capture.uuid) + ) + return (end_time - start_time) * 1000 + except (ValueError, IndexError, KeyError): + return None + + @extend_schema_field(serializers.IntegerField(allow_null=True)) + def get_file_cadence_ms(self, obj: dict[str, Any]) -> int | None: + """Use first channel's file cadence for composite capture.""" + channels = obj.get("channels") or [] + if not channels: + return None + try: + capture = Capture.objects.get(uuid=channels[0]["uuid"]) + return get_file_cadence(capture.capture_type, capture) + except (ValueError, IndexError, KeyError): + return None + def build_composite_capture_data(captures: list[Capture]) -> dict[str, Any]: """Build composite capture data from a list of captures with the same top_level_dir. diff --git a/gateway/sds_gateway/api_methods/tasks.py b/gateway/sds_gateway/api_methods/tasks.py index e4aed2651..c7dff0b31 100644 --- a/gateway/sds_gateway/api_methods/tasks.py +++ b/gateway/sds_gateway/api_methods/tasks.py @@ -26,6 +26,7 @@ from sds_gateway.api_methods.models import TemporaryZipFile from sds_gateway.api_methods.models import ZipFileStatus from sds_gateway.api_methods.models import user_has_access_to_item +from sds_gateway.api_methods.models import CaptureType from sds_gateway.api_methods.utils.disk_utils import DISK_SPACE_BUFFER from sds_gateway.api_methods.utils.disk_utils import check_disk_space_available from sds_gateway.api_methods.utils.disk_utils import estimate_disk_size @@ -676,15 +677,26 @@ def _process_item_files( item_type: ItemType, item_uuid: UUID, temp_zip: TemporaryZipFile, + start_time: int | None = None, + end_time: int | None = None, ) -> tuple[Mapping[str, UUID | int | str] | None, str | None, int | None, int | None]: # pyright: ignore[reportMissingTypeArgument] """ Process files for an item and create a zip file. + Args: + user: The user requesting the files + item: The item object (Dataset or Capture) + item_type: Type of item (dataset or capture) + item_uuid: UUID of the item to download + temp_zip: The temporary zip file to create + start_time: Optional start time for temporal filtering + end_time: Optional end time for temporal filtering + Returns: tuple: (error_response, zip_file_path, total_size, files_processed) If error_response is not None, the other values are None """ - files = _get_item_files(user, item, item_type) + files = _get_item_files(user, item, item_type, start_time, end_time) if not files: log.warning(f"No files found for {item_type} {item_uuid}") error_message = f"No files found in {item_type}" @@ -979,7 +991,11 @@ def _handle_timeout_exception( time_limit=30 * 60, soft_time_limit=25 * 60 ) # 30 min hard limit, 25 min soft limit def send_item_files_email( # noqa: C901, PLR0911, PLR0912, PLR0915 - item_uuid: UUID, user_id: str, item_type: str | ItemType + item_uuid: UUID, + user_id: str, + item_type: str | ItemType, + start_time: int | None = None, + end_time: int | None = None, ) -> Mapping[str, UUID | str | int]: """ Unified Celery task to create a zip file of item files and send it via email. @@ -990,6 +1006,8 @@ def send_item_files_email( # noqa: C901, PLR0911, PLR0912, PLR0915 item_uuid: UUID of the item to process user_id: ID of the user requesting the download item_type: Type of item (dataset or capture) + start_time: Optional start time for temporal filtering + end_time: Optional end time for temporal filtering Returns: dict: Task result with status and details """ @@ -1053,6 +1071,8 @@ def send_item_files_email( # noqa: C901, PLR0911, PLR0912, PLR0915 item_type=item_type_enum, item_uuid=item_uuid, temp_zip=temp_zip, + start_time=start_time, + end_time=end_time, ) ) if error_response: @@ -1251,7 +1271,13 @@ def _validate_item_download_request( return None, user, item -def _get_item_files(user: User, item: Any, item_type: ItemType) -> list[File]: +def _get_item_files( + user: User, + item: Any, + item_type: ItemType, + start_time: int | None = None, + end_time: int | None = None, +) -> list[File]: """ Get all files for an item based on its type. @@ -1259,14 +1285,16 @@ def _get_item_files(user: User, item: Any, item_type: ItemType) -> list[File]: user: The user requesting the files item: The item object (Dataset or Capture) item_type: Type of item (dataset or capture) - + start_time: Optional start time for temporal filtering + end_time: Optional end time for temporal filtering Returns: List of files associated with the item """ - from sds_gateway.api_methods.utils.relationship_utils import get_capture_files + from sds_gateway.api_methods.helpers.temporal_filtering import get_capture_files_with_temporal_filter from sds_gateway.api_methods.utils.relationship_utils import ( get_dataset_files_including_captures, ) + from sds_gateway.api_methods.utils.relationship_utils import get_capture_files if item_type == ItemType.DATASET: files_queryset = get_dataset_files_including_captures( @@ -1277,8 +1305,28 @@ def _get_item_files(user: User, item: Any, item_type: ItemType) -> list[File]: return files if item_type == ItemType.CAPTURE: - files = get_capture_files(item, include_deleted=False) - log.info(f"Found {len(files)} files for capture {item.uuid}") + capture_type = item.capture_type + # temporal filtering is only supported for DigitalRF captures + if capture_type is CaptureType.DigitalRF: + files = get_capture_files_with_temporal_filter( + capture_type=capture_type, + capture=item, + start_time=start_time, + end_time=end_time, + ) + else: + if start_time is not None or end_time is not None: + logger.warning( + "Temporal filtering is only supported for DigitalRF captures, " + "ignoring start_time and end_time" + ) + + files = get_capture_files( + capture=item, + include_deleted=False, + ) + + logger.info(f"Found {len(files)} files for capture {item.uuid}") return list(files) log.warning(f"Unknown item type: {item_type}") diff --git a/gateway/sds_gateway/static/js/actions/DownloadActionManager.js b/gateway/sds_gateway/static/js/actions/DownloadActionManager.js index ca33ca45d..233cb9b43 100644 --- a/gateway/sds_gateway/static/js/actions/DownloadActionManager.js +++ b/gateway/sds_gateway/static/js/actions/DownloadActionManager.js @@ -274,7 +274,18 @@ class DownloadActionManager { } }; - window.DOMUtils.openModal(modalId); + // Initialize temporal slider from button data attributes (clears or builds slider) + const durationMs = parseInt(button.getAttribute("data-length-of-capture-ms"), 10); + const fileCadenceMs = parseInt(button.getAttribute("data-file-cadence-ms"), 10); + if (typeof window.initCaptureDownloadSlider === "function") { + window.initCaptureDownloadSlider( + Number.isNaN(durationMs) ? 0 : durationMs, + Number.isNaN(fileCadenceMs) ? 1000 : fileCadenceMs, + ); + } + + // Show the modal + window.showWebDownloadModal(captureUuid, captureName); } /** diff --git a/gateway/sds_gateway/static/js/file-list.js b/gateway/sds_gateway/static/js/file-list.js index aba3069c9..609c19f17 100644 --- a/gateway/sds_gateway/static/js/file-list.js +++ b/gateway/sds_gateway/static/js/file-list.js @@ -711,6 +711,8 @@ class FileListCapturesTableManager extends CapturesTableManager { centerFrequencyGhz: ComponentUtils.escapeHtml( capture.center_frequency_ghz || "", ), + lengthOfCaptureMs: capture.length_of_capture_ms ?? 0, + fileCadenceMs: capture.file_cadence_ms ?? 1000, }; let typeDisplay = safeData.captureTypeDisplay || safeData.captureType; @@ -835,7 +837,9 @@ class FileListCapturesTableManager extends CapturesTableManager { diff --git a/gateway/sds_gateway/templates/base.html b/gateway/sds_gateway/templates/base.html index 2e14c130a..fdebfe804 100644 --- a/gateway/sds_gateway/templates/base.html +++ b/gateway/sds_gateway/templates/base.html @@ -19,6 +19,8 @@ href="https://cdn.jsdelivr.net/npm/bootstrap-icons@1.11.3/font/bootstrap-icons.min.css" /> + {% block css %} @@ -226,6 +228,7 @@ {# Removed JS that was hiding/showing the body #} {% endblock inline_javascript %} + diff --git a/gateway/sds_gateway/users/views.py b/gateway/sds_gateway/users/views.py index 41f2a70eb..85e340f91 100644 --- a/gateway/sds_gateway/users/views.py +++ b/gateway/sds_gateway/users/views.py @@ -3333,6 +3333,15 @@ def post( Returns: A JSON response containing the download status """ + # optional start and end times for temporal filtering + start_time = request.POST.get("start_time", None) + end_time = request.POST.get("end_time", None) + + if start_time: + start_time = int(start_time) + if end_time: + end_time = int(end_time) + # Validate item type if item_type not in self.ITEM_MODELS: return JsonResponse( @@ -3399,6 +3408,8 @@ def post( str(item.uuid), str(request.user.id), item_type, + start_time=start_time, + end_time=end_time, ) return JsonResponse( From 39c488502ee8a5d71b0c8dace50fcd84b1c097ef Mon Sep 17 00:00:00 2001 From: klpoland Date: Mon, 23 Feb 2026 09:17:22 -0500 Subject: [PATCH 2/7] fix label updates --- gateway/sds_gateway/api_methods/tasks.py | 12 ++-- .../users/partials/web_download_modal.html | 55 +++++++++++-------- 2 files changed, 40 insertions(+), 27 deletions(-) diff --git a/gateway/sds_gateway/api_methods/tasks.py b/gateway/sds_gateway/api_methods/tasks.py index c7dff0b31..7a13f4390 100644 --- a/gateway/sds_gateway/api_methods/tasks.py +++ b/gateway/sds_gateway/api_methods/tasks.py @@ -20,13 +20,13 @@ from redis import Redis from sds_gateway.api_methods.models import Capture +from sds_gateway.api_methods.models import CaptureType from sds_gateway.api_methods.models import Dataset from sds_gateway.api_methods.models import File from sds_gateway.api_methods.models import ItemType from sds_gateway.api_methods.models import TemporaryZipFile from sds_gateway.api_methods.models import ZipFileStatus from sds_gateway.api_methods.models import user_has_access_to_item -from sds_gateway.api_methods.models import CaptureType from sds_gateway.api_methods.utils.disk_utils import DISK_SPACE_BUFFER from sds_gateway.api_methods.utils.disk_utils import check_disk_space_available from sds_gateway.api_methods.utils.disk_utils import estimate_disk_size @@ -1290,11 +1290,13 @@ def _get_item_files( Returns: List of files associated with the item """ - from sds_gateway.api_methods.helpers.temporal_filtering import get_capture_files_with_temporal_filter + from sds_gateway.api_methods.helpers.temporal_filtering import ( + get_capture_files_with_temporal_filter, + ) + from sds_gateway.api_methods.utils.relationship_utils import get_capture_files from sds_gateway.api_methods.utils.relationship_utils import ( get_dataset_files_including_captures, ) - from sds_gateway.api_methods.utils.relationship_utils import get_capture_files if item_type == ItemType.DATASET: files_queryset = get_dataset_files_including_captures( @@ -1316,7 +1318,7 @@ def _get_item_files( ) else: if start_time is not None or end_time is not None: - logger.warning( + log.warning( "Temporal filtering is only supported for DigitalRF captures, " "ignoring start_time and end_time" ) @@ -1326,7 +1328,7 @@ def _get_item_files( include_deleted=False, ) - logger.info(f"Found {len(files)} files for capture {item.uuid}") + log.info(f"Found {len(files)} files for capture {item.uuid}") return list(files) log.warning(f"Unknown item type: {item_type}") diff --git a/gateway/sds_gateway/templates/users/partials/web_download_modal.html b/gateway/sds_gateway/templates/users/partials/web_download_modal.html index 5275a8972..8fb67d878 100644 --- a/gateway/sds_gateway/templates/users/partials/web_download_modal.html +++ b/gateway/sds_gateway/templates/users/partials/web_download_modal.html @@ -38,13 +38,14 @@