Update REACH Download to use Date-Based API

README.rst

@@ -93,27 +93,39 @@ Generate lines of code report and upload
 Generates a lines-of-code report and uploads the output artifact.
 
 
-download_UDL_REACH_to_file
-^^^^^^^^^^^^^^^^^^^^^^^^^^
+import_UDL_REACH_to_s3
+^^^^^^^^^^^^^^^^^^^^^^^
 
 Downloads REACH data from UDL in chunked requests, combines all records,
 and writes a single output file to Lambda storage for upload.
 
+The query window is snapped to UTC midnight boundaries so each run always
+covers a whole midnight-to-midnight day, independent of the exact second
+the Lambda executes. The window is driven by two day-based offsets
+measured from UTC midnight of the current run day:
+
+- ``REACH_WINDOW_END_DAYS_AGO``: number of days before today's UTC
+  midnight at which the window ends.
+- ``REACH_WINDOW_DAYS``: length of the window in whole days.
+
 Suggested daily pattern:
 
-- Set ``REACH_WINDOW_SECONDS=86400`` for one day per run.
-- Schedule one daily EventBridge trigger.
+- Schedule one daily EventBridge trigger, e.g. ``cron(0 6 * * ? *)`` UTC.
+- Keep ``REACH_WINDOW_END_DAYS_AGO=1`` and ``REACH_WINDOW_DAYS=1`` to pull
+  the full day before yesterday (54 hours ago to 30 hours ago at run
+  time), e.g. the window ``[today 00:00 - 2 days, today 00:00 - 1 day)``.
 - Upload the single combined artifact produced by each run.
 
 Recommended environment variables:
 
 - ``REACH_SENSOR_ID`` (default: ``ALL``): sensors to query from UDL.
 - ``REACH_DESCRIPTOR`` (default: ``QUICKLOOK``): data product to query from UDL.
 - ``REACH_FILE_FORMAT`` (default: ``json``): output format (``json`` or ``csv``).
-- ``REACH_DELAY_SECONDS`` (default: ``7200``): offset from ``datetime.now(timezone.utc)`` to the end time.
-- ``REACH_WINDOW_SECONDS`` (default: ``600``): window size before end time.
+- ``REACH_WINDOW_END_DAYS_AGO`` (default: ``1``): days before today's UTC midnight at which the window ends.
+- ``REACH_WINDOW_DAYS`` (default: ``1``): length of the query window in whole days.
 - ``REACH_OUTPUT_DIR`` (default: ``/tmp``): Lambda container directory for output files.
-- ``REACH_DESTINATION_BUCKET`` (default: ``dev-swxsoc-pipeline-incoming``): bucket to upload output files.
+- ``REACH_DESTINATION_BUCKET_DEV`` (default: ``dev-swxsoc-pipeline-incoming``): bucket to upload output files.
+- ``REACH_DESTINATION_BUCKET_PROD`` (default: ``swxsoc-pipeline-incoming``): bucket to upload output files.
 - ``REACH_UDL_MAX_CONCURRENT_REQUESTS`` (default: ``8``): max concurrent workers for UDL pulls.
 - ``REACH_UDL_INITIAL_RATE`` (default: ``5.0``): AIMD starting request rate (requests/second).
 - ``REACH_UDL_ADDITIVE_INCREASE`` (default: ``1.0``): AIMD additive increase after successful requests.
@@ -137,8 +149,7 @@ Building and Running Locally
 ----------------------------
 
 The container image can be built and run locally. You can specify the base image at runtime.
-At the time of writing, the base image defaults to
-``padre-swsoc-docker-lambda-base:latest`` in AWS.
+At the time of writing, the base image defaults to ``padre-swsoc-docker-lambda-base:latest`` in AWS.
 
 .. code-block:: sh
 
@@ -158,7 +169,8 @@ You can retrieve the Grafana and UDL ARNs from AWS.
 .. code-block:: sh
 
     docker run -p 9000:8080 \
-      -e REACH_DESTINATION_BUCKET="dev-swxsoc-pipeline-incoming" \
+      -e REACH_DESTINATION_BUCKET_DEV="dev-swxsoc-pipeline-incoming" \
+      -e REACH_DESTINATION_BUCKET_PROD="swxsoc-pipeline-incoming" \
       -e SECRET_ARN_GRAFANA=$SECRET_ARN_GRAFANA \
       -e SECRET_ARN_UDL=$SECRET_ARN_UDL \
       -e AWS_ACCESS_KEY_ID=$AWS_ACCESS_KEY_ID \

buildspec.yml

@@ -1,47 +1,44 @@
 version: 0.2
 
 phases:
-  pre_build:
-    commands:
-      - echo Installing CI/CD Dependencies...
-      - pip3 install flake8 black pytest pytest-astropy pytest-cov
-      - pip3 install pytest pytest-astropy pytest-cov
-      - pip3 install -r requirements.txt
-      - echo ________________________________
-
-      - echo Linting with Black...
-      - black --check --diff lambda_function
-      - echo ________________________________
-
-      - echo Linting with Flake...
-      - flake8 --count --max-line-length 88 lambda_function --exclude lambda_function_test_script.py
-      - echo ________________________________
-
   build:
     commands:
-      - REGION=us-east-1
-      - echo Login to Private ECR $REGION
-      - aws ecr get-login-password --region $REGION | docker login --username AWS --password-stdin 351967858401.dkr.ecr.$REGION.amazonaws.com
-      - echo ________________________________
-
-      - echo Create Date Tag
-      - export TAG=$(date +%Y-%m-%d-%H-%M-%S)
-      - echo ________________________________
+    - ACCOUNT_ID=$(aws sts get-caller-identity --query 'Account' --output text)
+    - REGION=us-east-1
+    - ECR_REPO="swxsoc_sdc_aws_executor_lambda"
+    - BASE_IMAGE=padre-swsoc-docker-lambda-base:latest
+    - echo Login to Private ECR $REGION
+    - aws ecr get-login-password --region $REGION | docker login --username AWS --password-stdin $ACCOUNT_ID.dkr.ecr.$REGION.amazonaws.com
+    - echo ________________________________
+    - |
+      if git describe --tags --exact-match > /dev/null 2>&1; then
+        echo "This is a tag push event"
+        CDK_ENVIRONMENT=PRODUCTION
+        BASE_IMAGE="public.ecr.aws/w5r9l1c8/$BASE_IMAGE"
+        VERSION=`git describe --tags --exact-match`
+      else
+        echo "This is a production environment"
+        ECR_REPO="$ECR_REPO"
+        BASE_IMAGE="public.ecr.aws/w5r9l1c8/$BASE_IMAGE"
+        CDK_ENVIRONMENT=PRODUCTION
+        VERSION=`date -u +"%Y%m%d%H%M%S"`
+      fi
+    - echo ________________________________
+    - echo Build Docker Image
+    - FULL_ECR_REPO="$ACCOUNT_ID.dkr.ecr.$REGION.amazonaws.com/$ECR_REPO"
+
+    - docker build --build-arg BASE_IMAGE=$BASE_IMAGE -t $FULL_ECR_REPO:latest lambda_function/.
+    - echo Tagging Docker Image...
+    - docker tag $FULL_ECR_REPO:latest $FULL_ECR_REPO:$VERSION
 
-      - echo Build Docker Image
-      - docker build -t 351967858401.dkr.ecr.$REGION.amazonaws.com/sdc_aws_executor_lambda:latest lambda_function/.
-
-      - echo Tagging Docker Image...
-      - docker tag 351967858401.dkr.ecr.$REGION.amazonaws.com/sdc_aws_executor_lambda:latest 351967858401.dkr.ecr.$REGION.amazonaws.com/sdc_aws_executor_lambda:$TAG
+    - echo Pushing the Docker image with Tags...
+    - docker push $FULL_ECR_REPO:latest
+    - docker push $FULL_ECR_REPO:$VERSION
+    - echo ________________________________
 
-      - echo Pushing the Docker image with Tags...
-      - docker push 351967858401.dkr.ecr.$REGION.amazonaws.com/sdc_aws_executor_lambda:latest
-      - docker push 351967858401.dkr.ecr.$REGION.amazonaws.com/sdc_aws_executor_lambda:$TAG
-      - echo ________________________________
-
-      - echo Updating Deployment
-      - aws codebuild start-build --project-name build_sdc_aws_pipeline_architecture --region us-east-2 --environment-variables-override name=LAMBDA_PIPELINE,value=executor,type=PLAINTEXT name=TAG,value=$TAG,type=PLAINTEXT
-      - echo ________________________________
+    - echo Updating Deployment
+    - echo ________________________________
+    # - aws codebuild start-build --project-name arn:aws:codebuild:us-east-1:$ACCOUNT_ID:project/build_padre_sdc_aws_pipeline_architecture --environment-variables-override name=CDK_ENVIRONMENT,value=$CDK_ENVIRONMENT,type=PLAINTEXT
 
   post_build:
     commands:

lambda_function/src/executor/executor.py

@@ -22,7 +22,7 @@
 from sdc_aws_utils.config import parser as science_filename_parser
 from swxsoc import log
 from swxsoc.util import util
-from swxsoc_reach.net.udl import download_UDL_reach_to_file
+from swxsoc_reach.net.udl import download_UDL_reach_window
 
 
 def handle_event(event: dict[str, Any], context: Any) -> dict[str, Any]:
@@ -261,13 +261,29 @@ def _upload_reach_file_to_s3(filepath: str) -> list[str]:
     def import_UDL_REACH_to_s3() -> None:
         """
         Download REACH data from UDL and upload the resulting file to S3.
+
+        The query window is snapped to UTC midnight boundaries so the
+        downloaded file always covers a whole midnight-to-midnight day,
+        independent of the exact second the Lambda happens to run. The
+        window is driven by two day-based offsets measured from UTC
+        midnight of the current run day:
+
+        - ``REACH_WINDOW_END_DAYS_AGO``: number of days before today's
+          UTC midnight at which the window ends (default ``1`` -> the
+          window ends at the start of yesterday).
+        - ``REACH_WINDOW_DAYS``: length of the window in whole days
+          (default ``1``).
+
+        With the defaults and the ``cron(0 6 * * ? *)`` UTC schedule this
+        yields ``[today 00:00 - 2 days, today 00:00 - 1 day)``: all of the
+        day before yesterday (54 hours ago to 30 hours ago at run time).
         """
         basic_auth = os.environ["basicauth".upper()]
         sensor_id = os.environ.get("REACH_SENSOR_ID", "ALL")
         descriptor = os.environ.get("REACH_DESCRIPTOR", "QUICKLOOK")
         output_format = os.environ.get("REACH_FILE_FORMAT", "csv").lower()
-        delay_seconds = int(os.environ.get("REACH_DELAY_SECONDS", "7200"))
-        window_seconds = int(os.environ.get("REACH_WINDOW_SECONDS", "600"))
+        end_days_ago = int(os.environ.get("REACH_WINDOW_END_DAYS_AGO", "1"))
+        window_days = int(os.environ.get("REACH_WINDOW_DAYS", "1"))
         output_dir = os.environ.get("REACH_OUTPUT_DIR", "/tmp")
         max_concurrent_requests = int(
             os.environ.get("REACH_UDL_MAX_CONCURRENT_REQUESTS", "8")
@@ -280,14 +296,22 @@ def import_UDL_REACH_to_s3() -> None:
         min_rate = float(os.environ.get("REACH_UDL_MIN_RATE", "5.0"))
         max_rate = float(os.environ.get("REACH_UDL_MAX_RATE", "25.0"))
 
+        # Snap to UTC midnight of the current run day so the window edges
+        # land exactly on midnight instead of drifting with the run second.
+        midnight_today = Time(Time.now().iso[0:10])
+        end_time = midnight_today - TimeDelta(end_days_ago * u.day)
+        start_time = end_time - TimeDelta(window_days * u.day)
+
         log.info(
             "Starting REACH import_UDL_REACH_to_s3",
             extra={
                 "sensor_id": sensor_id,
                 "descriptor": descriptor,
                 "output_format": output_format,
-                "delay_seconds": delay_seconds,
-                "window_seconds": window_seconds,
+                "end_days_ago": end_days_ago,
+                "window_days": window_days,
+                "start_time": start_time.isot,
+                "end_time": end_time.isot,
                 "output_dir": output_dir,
                 "max_concurrent_requests": max_concurrent_requests,
                 "initial_rate": initial_rate,
@@ -299,13 +323,13 @@ def import_UDL_REACH_to_s3() -> None:
         )
 
         # Download REACH Data to the Specific Folder
-        downloaded_path = download_UDL_reach_to_file(
+        downloaded_path = download_UDL_reach_window(
             auth_token=basic_auth,
             sensor_id=sensor_id,
             descriptor=descriptor,
             output_format=output_format,
-            delay_seconds=delay_seconds,
-            window_seconds=window_seconds,
+            start_time=start_time,
+            end_time=end_time,
             output_dir=output_dir,
             max_concurrent_requests=max_concurrent_requests,
             initial_rate=initial_rate,

lambda_function/tests/test_executor.py

@@ -6,7 +6,7 @@
 import boto3
 import numpy as np
 import pytest
-from astropy.time import Time
+from astropy.time import Time, TimeDelta
 from src.executor.executor import Executor
 
 
@@ -111,11 +111,11 @@ def fake_record_orbit(timeseries):
 
 
 def test_import_UDL_REACH_to_s3(monkeypatch, tmp_path) -> None:
-    """REACH import uses configured small window and uploads via mocked push function."""
+    """REACH import snaps to a midnight-aligned day window and uploads via mocked push."""
     download_call = {}
     upload_calls = []
 
-    def fake_download_udl_reach_to_file(**kwargs):
+    def fake_download_UDL_reach_window(**kwargs):
         download_call.update(kwargs)
         output_file = tmp_path / "REACH-ALL_20250603T000000_20250603T002000.csv"
         output_file.write_text("time,value\n2025-06-03T00:00:00Z,1\n")
@@ -134,25 +134,35 @@ def fake_push_science_file(
         return f"mock/{calibrated_filename}"
 
     monkeypatch.setenv("BASICAUTH", "mock-auth-token")
-    monkeypatch.setenv("REACH_DELAY_SECONDS", "86400")
-    monkeypatch.setenv("REACH_WINDOW_SECONDS", "1200")
+    monkeypatch.setenv("REACH_WINDOW_END_DAYS_AGO", "1")
+    monkeypatch.setenv("REACH_WINDOW_DAYS", "1")
     monkeypatch.setenv("REACH_UDL_MAX_CONCURRENT_REQUESTS", "4")
     monkeypatch.setenv("REACH_OUTPUT_DIR", str(tmp_path))
     monkeypatch.setenv("REACH_DESTINATION_BUCKET_PROD", "unit-test-bucket-prod")
     monkeypatch.setenv("REACH_DESTINATION_BUCKET_DEV", "unit-test-bucket-dev")
 
     monkeypatch.setattr(
-        "src.executor.executor.download_UDL_reach_to_file",
-        fake_download_udl_reach_to_file,
+        "src.executor.executor.download_UDL_reach_window",
+        fake_download_UDL_reach_window,
     )
     monkeypatch.setattr(
         "src.executor.executor.push_science_file", fake_push_science_file
     )
 
     Executor.import_UDL_REACH_to_s3()
 
-    assert download_call["delay_seconds"] == 86400
-    assert download_call["window_seconds"] == 1200
+    start_time = download_call["start_time"]
+    end_time = download_call["end_time"]
+
+    # Window edges are snapped to UTC midnight.
+    assert start_time.iso.endswith("00:00:00.000")
+    assert end_time.iso.endswith("00:00:00.000")
+
+    # End is one day before today's UTC midnight; window is one day long.
+    expected_end = Time(Time.now().iso[0:10]) - TimeDelta(1, format="jd")
+    assert (end_time - expected_end).to_value("sec") == pytest.approx(0.0, abs=1.0)
+    assert (end_time - start_time).to_value("jd") == pytest.approx(1.0)
+
     assert download_call["max_concurrent_requests"] == 4
     assert download_call["output_dir"] == str(tmp_path)
     assert len(upload_calls) == 2