Add transect update monitoring recipe

.github/workflows/test_monitoring.yaml

@@ -0,0 +1,154 @@
+name: Test monitoring recipes
+
+on:
+  pull_request:
+  push:
+    branches:
+      - main
+
+jobs:
+  test-monitoring:
+    runs-on: ubuntu-latest
+
+    env:
+      PREFECT_API_URL: http://127.0.0.1:4200/api
+
+    steps:
+      - name: Checkout recipes
+        uses: actions/checkout@v4
+
+      - name: Checkout echodataflow monitoring PR
+        uses: actions/checkout@v4
+        with:
+          repository: echostack-org/echodataflow
+          ref: refs/pull/174/head
+          path: echodataflow
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+
+      - name: Install echodataflow
+        run: pip install -e ./echodataflow
+
+      - name: Start Prefect server
+        run: |
+          prefect server start --host 127.0.0.1 > /tmp/prefect-server.log 2>&1 &
+          echo $! > /tmp/prefect-server.pid
+
+      - name: Wait for Prefect server
+        run: |
+          for i in {1..30}; do
+            if curl -fsS http://127.0.0.1:4200/api/health >/dev/null; then
+              exit 0
+            fi
+            sleep 2
+          done
+
+          cat /tmp/prefect-server.log
+          exit 1
+
+      - name: Test raw and transect event monitoring
+        shell: python
+        run: |
+          import asyncio
+          import tempfile
+          import time
+          from datetime import datetime, timedelta, timezone
+          from pathlib import Path
+
+          from prefect import get_client
+          from prefect.events.filters import EventFilter, EventOccurredFilter
+
+          from echodataflow.utils.raw_monitor import watch_raw_directory
+          from echodataflow.utils.transect_monitor import watch_transect_file
+
+
+          async def has_event(event_name):
+              async with get_client() as client:
+                  now = datetime.now(timezone.utc)
+                  page = await client.read_events(
+                      filter=EventFilter(
+                          occurred=EventOccurredFilter(
+                              since=now - timedelta(minutes=5),
+                              until=now,
+                          )
+                      ),
+                      limit=100,
+                  )
+
+                  return any(
+                      event.event == event_name
+                      for event in page.events
+                  )
+
+
+          async def wait_for_event(event_name, timeout=10):
+              deadline = time.time() + timeout
+
+              while time.time() < deadline:
+                  if await has_event(event_name):
+                      return
+                  await asyncio.sleep(0.5)
+
+              raise AssertionError(
+                  f"Did not receive expected Prefect event: {event_name}"
+              )
+
+
+          async def main():
+              with tempfile.TemporaryDirectory() as tmp:
+                  root = Path(tmp)
+
+                  # RAW monitoring
+                  raw_dir = root / "raw"
+                  raw_dir.mkdir()
+
+                  observer = watch_raw_directory(raw_dir)
+
+                  try:
+                      (raw_dir / "test.raw").write_bytes(b"dummy raw data")
+                      await wait_for_event("echodataflow.raw.updated")
+                  finally:
+                      observer.stop()
+                      observer.join()
+
+                  # Transect monitoring
+                  transect_file = root / "transects.csv"
+                  transect_file.write_text(
+                      "transectPart,transectNumber,transectStart,transectEnd\n"
+                      "1,1,2024-06-25T21:45:00+00:00,\n"
+                  )
+
+                  observer = watch_transect_file(transect_file)
+
+                  try:
+                      transect_file.write_text(
+                          "transectPart,transectNumber,transectStart,transectEnd\n"
+                          "1,1,2024-06-25T21:45:00+00:00,"
+                          "2024-06-25T21:50:00+00:00\n"
+                      )
+
+                      await wait_for_event(
+                          "echodataflow.transect.updated"
+                      )
+                  finally:
+                      observer.stop()
+                      observer.join()
+
+
+          asyncio.run(main())
+
+          print("RAW and transect Prefect events received successfully.")
+
+      - name: Show Prefect server log on failure
+        if: failure()
+        run: cat /tmp/prefect-server.log
+
+      - name: Upload monitoring test artifacts
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: monitoring-test-artifacts
+          path: /tmp/prefect-server.log

scripts/watch_raw_updates.py

@@ -0,0 +1,40 @@
+import argparse
+import time
+from pathlib import Path
+
+from echodataflow.utils.raw_monitor import watch_raw_directory
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Watch a RAW directory and emit Prefect update events."
+    )
+    parser.add_argument(
+        "path",
+        type=Path,
+        help="Path to the RAW directory to watch.",
+    )
+    args = parser.parse_args()
+
+    target_dir = args.path.resolve()
+
+    if not target_dir.exists():
+        raise FileNotFoundError(
+            f"RAW directory does not exist: {target_dir}"
+        )
+
+    observer = watch_raw_directory(target_dir)
+
+    print(f"Watching {target_dir}")
+
+    try:
+        while True:
+            time.sleep(1)
+    except KeyboardInterrupt:
+        observer.stop()
+
+    observer.join()
+
+
+if __name__ == "__main__":
+    main()

scripts/watch_transect_updates.py

@@ -0,0 +1,38 @@
+import argparse
+import time
+from pathlib import Path
+
+from echodataflow.utils.transect_monitor import watch_transect_file
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Watch a transect CSV file and emit Prefect update events."
+    )
+    parser.add_argument(
+        "path",
+        type=Path,
+        help="Path to the transect CSV file to watch.",
+    )
+    args = parser.parse_args()
+
+    target_file = args.path.resolve()
+
+    if not target_file.exists():
+        raise FileNotFoundError(f"Transect file does not exist: {target_file}")
+
+    observer = watch_transect_file(target_file)
+
+    print(f"Watching {target_file}")
+
+    try:
+        while True:
+            time.sleep(1)
+    except KeyboardInterrupt:
+        observer.stop()
+
+    observer.join()
+
+
+if __name__ == "__main__":
+    main()

tests/monitoring/deploy.yaml

@@ -0,0 +1,19 @@
+source:
+  mode: local
+
+default_work_pool_name: local
+
+flows:
+  raw2Sv:
+    deployment_name: raw2Sv-monitor-test
+    triggers:
+      - expect: "echodataflow.raw.updated"
+        resource_name: "raw-monitor"
+        resource_scope: primary
+
+  transect_update:
+    deployment_name: transect-update-monitor-test
+    triggers:
+      - expect: "echodataflow.transect.updated"
+        resource_name: "transect-monitor"
+        resource_scope: related

tests/monitoring/params.yaml

@@ -0,0 +1,24 @@
+flows:
+  raw2Sv:
+    exclude_before:
+    exclude_raw_file: []
+    parallel: false
+    new_file_num_limit: 1
+
+    encode_mode: power
+    waveform_mode: CW
+    depth_offset: 0
+    sonar_model: EK80
+    datagram_type:
+    nmea_sentence: GGA
+    filename_pattern: "*.raw"
+
+    path_raw: "/path/to/raw"
+    path_main: "/path/to/echodataflow/output"
+    file_Sv_csv: Sv_files.csv
+
+  transect_update:
+    path_transect_csv: "/path/to/plotSurvey_Survey_Data_Visualizer.csv"
+    path_snapshot_csv: "/path/to/plotSurvey_Survey_Data_Visualizer_snapshot.csv"
+    path_main: "/path/to/echodataflow/output"
+    file_Sv_csv: Sv_files.csv