Add CPS raw2Sv realtime processing flow

src/echodataflow/deployment/flow_registry.py

@@ -18,6 +18,10 @@ class FlowRegistration:
         entrypoint="echodataflow/flows/flows_acoustics.py:flow_raw2Sv",
         description="Incrementally convert newly available raw sonar files to Sv.",
     ),
+    "raw2Sv_CPS": FlowRegistration(
+        entrypoint="echodataflow/flows/flows_acoustics.py:flow_raw2Sv_CPS",
+        description="Convert newly registered RAW files to Sv using the CPS processing database.",
+    ),
     "create_MVBS": FlowRegistration(
         entrypoint="echodataflow/flows/flows_acoustics.py:flow_create_MVBS",
     ),
@@ -63,9 +67,6 @@ class FlowRegistration:
     "copy_trawl": FlowRegistration(
         entrypoint="echodataflow/flows/flows_simulation.py:flow_copy_trawl",
     ),
-    "simulate_transects": FlowRegistration(
-        entrypoint="echodataflow/flows/flows_simulation.py:flow_simulate_transects",
-    ),
     "update_cache_MVBS": FlowRegistration(
         entrypoint="echodataflow/flows/flows_viz_cloud.py:flow_update_cache_MVBS",
     ),

src/echodataflow/flows/flows_CPS.py

@@ -1,4 +1,3 @@
-import asyncio
 from pathlib import Path
 
 import dask_image.ndfilters
@@ -7,9 +6,7 @@
 import numpy as np
 import pandas as pd
 import xarray as xr
-from prefect import flow, get_client, runtime
-from prefect.states import Cancelled
-from echodataflow.flows.flows_helper import deployment_already_running
+from prefect import flow
 from prefect_dask import DaskTaskRunner
 
 from echodataflow.utils.processing_ledger import get_completed_sv_files, resolve_database
@@ -257,28 +254,6 @@ def flow_process_CPS(
     nasc_process_id: int = 1928,
 ):
 
-    # Prevent overlapping runs of this deployment
-    already_running = asyncio.run(
-        deployment_already_running()
-    )
-
-    if already_running:
-
-        async def cancel_run():
-            async with get_client() as client:
-                await client.set_flow_run_state(
-                    flow_run_id=runtime.flow_run.id,
-                    state=Cancelled(
-                        message=(
-                            "Another instance of this "
-                            "flow is already running"
-                        )
-                    ),
-                )
-
-        asyncio.run(cancel_run())
-        return
-
     path_main = Path(path_main)
 
     path_transect = Path(

src/echodataflow/flows/flows_acoustics.py

@@ -37,6 +37,15 @@
     plan_mvbs_slices,
     read_or_create_ledger,
 )
+
+from echodataflow.utils.processing_ledger import (
+    get_raw_files_to_process,
+    initialize_ledger,
+    mark_raw_completed,
+    mark_raw_failed,
+    mark_raw_processing,
+    resolve_database,
+)
 from echodataflow.operations.operations_storage import S3CopySettings, S3CopyWorkItem
 from echodataflow.tasks.tasks_acoustics import (
     task_create_MVBS,
@@ -142,7 +151,7 @@ def flow_raw2Sv(
             f"Limiting to first {new_file_num_limit} files."
         )
         new_files = new_files[:new_file_num_limit]
-    print(f"Files to process: \n" + "".join([f"- {nf}\n" for nf in new_files]))
+    print("Files to process: \n" + "".join([f"- {nf}\n" for nf in new_files]))
 
     settings = RawToSvSettings(
         output_directory=path_Sv_zarr,
@@ -227,6 +236,179 @@ async def set_failed_state():
         asyncio.run(set_failed_state())
         raise Exception(error_msg)
 
+
+@flow(log_prints=True, task_runner=dask_task_runner_from_environment())
+def flow_raw2Sv_CPS(
+    exclude_before: str | None = None,
+    exclude_raw_file: list[str] = [],
+    parallel: bool = False,
+    encode_mode: str = "power",
+    waveform_mode: str = "CW",
+    depth_offset: float = 9.5,
+    sonar_model: str = "EK80",
+    datagram_type: str | None = None,
+    nmea_sentence: str | None = None,
+    path_main: str = "",
+    processing_db: str = "processing.db",
+    new_file_num_limit: int = 50,
+    add_depth: bool = True,
+    add_location: bool = True,
+    add_splitbeam_angle: bool = False,
+):
+    """Convert newly registered RAW files to Sv using the CPS processing database."""
+
+    path_Sv_zarr = Path(path_main) / "Sv"
+    path_Sv_zarr.mkdir(parents=True, exist_ok=True)
+
+    db_path = resolve_database(path_main, processing_db)
+    initialize_ledger(db_path)
+
+    raw_files = get_raw_files_to_process(
+        db_path,
+        limit=new_file_num_limit,
+    )
+
+    if exclude_before is not None:
+        exclude_before_dt = pd.to_datetime(
+            exclude_before,
+            utc=True,
+        ).to_pydatetime()
+
+        raw_files = [
+            raw_path
+            for raw_path in raw_files
+            if extract_datetime_from_filename(raw_path.name)
+            >= exclude_before_dt
+        ]
+
+    if exclude_raw_file:
+        excluded = set(exclude_raw_file)
+        raw_files = [
+            raw_path
+            for raw_path in raw_files
+            if raw_path.name not in excluded
+        ]
+
+    print(f"Found {len(raw_files)} RAW files to process")
+    print(
+        "Files to process:\n"
+        + "".join(f"- {raw_path.name}\n" for raw_path in raw_files)
+    )
+
+    if not raw_files:
+        return
+
+    settings = RawToSvSettings(
+        output_directory=str(path_Sv_zarr),
+        encode_mode=encode_mode,
+        waveform_mode=waveform_mode,
+        depth_offset=depth_offset,
+        sonar_model=sonar_model,
+        datagram_type=datagram_type,
+        nmea_sentence=nmea_sentence,
+        add_depth=add_depth,
+        add_location=add_location,
+        add_splitbeam_angle=add_splitbeam_angle,
+    )
+
+    errors = []
+
+    if parallel:
+        print("Processing RAW files in parallel")
+        futures = {}
+
+        for raw_path in raw_files:
+            mark_raw_processing(db_path, raw_path)
+
+            future = task_raw2Sv.with_options(
+                task_run_name=raw_path.name,
+                name=raw_path.name,
+                retries=3,
+            ).submit(
+                RawToSvWorkItem(raw_path=str(raw_path)),
+                settings,
+            )
+
+            futures[future] = raw_path
+
+        for future in as_completed(futures):
+            raw_path = futures[future]
+
+            try:
+                result = future.result()
+
+                mark_raw_completed(
+                    db_path,
+                    raw_path,
+                    result.filename_Sv,
+                    result.first_ping_time,
+                    result.last_ping_time,
+                )
+
+            except Exception as exc:
+                mark_raw_failed(
+                    db_path,
+                    raw_path,
+                    str(exc),
+                )
+                errors.append(exc)
+                print(f"Error converting {raw_path.name}: {exc}")
+
+    else:
+        print("Processing RAW files sequentially")
+
+        for raw_path in raw_files:
+            try:
+                print(f"Converting {raw_path.name}")
+
+                mark_raw_processing(
+                    db_path,
+                    raw_path,
+                )
+
+                result = task_raw2Sv.with_options(
+                    task_run_name=raw_path.name,
+                    name=raw_path.name,
+                    retries=3,
+                )(
+                    RawToSvWorkItem(raw_path=str(raw_path)),
+                    settings,
+                )
+
+                mark_raw_completed(
+                    db_path,
+                    raw_path,
+                    result.filename_Sv,
+                    result.first_ping_time,
+                    result.last_ping_time,
+                )
+
+            except Exception as exc:
+                mark_raw_failed(
+                    db_path,
+                    raw_path,
+                    str(exc),
+                )
+                errors.append(exc)
+                print(f"Error converting {raw_path.name}: {exc}")
+
+    if errors:
+        error_msg = (
+            f"{len(errors)} errors during raw to Sv conversion "
+            f"out of {len(raw_files)} files"
+        )
+
+        async def set_failed_state():
+            async with get_client() as client:
+                await client.set_flow_run_state(
+                    flow_run_id=runtime.flow_run.id,
+                    state=Failed(message=error_msg),
+                )
+
+        asyncio.run(set_failed_state())
+        raise RuntimeError(error_msg)
+
+
 @flow(log_prints=True)
 async def flow_create_MVBS(
     time_offset_seconds: float = 0.0,

src/echodataflow/flows/flows_simulation.py

@@ -246,11 +246,12 @@ def flow_simulate_transects(
     transect_state_key = _var_key(prefix="transect_state")
     state = Variable.get(transect_state_key, default=None)
 
-    transect_num_curr = (
-        start_transect_num
-        if state is None
-        else int(state)
-    )
+    if state is None:
+        transect_num_curr = start_transect_num
+        action = "open"
+    else:
+        transect_num_str, action = str(state).split(":", maxsplit=1)
+        transect_num_curr = int(transect_num_str)
 
     if transect_num_curr > max_transects:
         print("All simulated transects have been generated.")

src/echodataflow/flows/flows_viz_cloud.py

@@ -9,7 +9,7 @@
 
 from prefect import flow, get_run_logger
 
-from echodataflow.utils.utils import round_up_mins, get_slice_start_end_times
+from echodataflow.utils.utils import get_slice_start_end_times
 
 
 @flow()
@@ -256,9 +256,17 @@ def flow_update_cache_CPS(
     # Find latest completed CPS transect
     # -----------------------------------------------------
 
+    def _transect_number(path: Path) -> int:
+        return int(
+            path.name
+            .replace("transect_", "")
+            .replace("_CPS.zarr", "")
+        )
+
+
     cps_files = sorted(
         path_CPS.glob("transect_*_CPS.zarr"),
-        key=lambda path: path.stat().st_mtime,
+        key=_transect_number,
     )
 
     if not cps_files:

src/echodataflow/operations/operations_watchdog.py

@@ -1,6 +1,7 @@
 from pathlib import Path
 
 from prefect.events import emit_event
+from prefect.events.worker import EventsWorker
 
 from echodataflow.utils.file_watcher import watch_directory, watch_file
 from echodataflow.utils.processing_ledger import (
@@ -17,9 +18,15 @@
 TRANSECT_RELATED_RESOURCE_ID = "transect-monitor"
 
 
+def _flush_events() -> None:
+    """Wait until queued Prefect events have been sent."""
+    EventsWorker.instance().wait_until_empty()
+
+
 def emit_raw_update_event(path: Path) -> None:
     """Emit a Prefect event when a RAW file arrives."""
-    emit_event(
+
+    event = emit_event(
         event=RAW_UPDATE_EVENT,
         resource={
             "prefect.resource.id": RAW_RESOURCE_ID,
@@ -28,8 +35,17 @@ def emit_raw_update_event(path: Path) -> None:
         },
     )
 
+    print(f"RAW event emitted for {path}: {event}")
+
+    if event is not None:
+        _flush_events()
+        print("RAW event queue flushed")
 
-def register_and_emit_raw_update(path: Path, db_path: str | Path) -> None:
+
+def register_and_emit_raw_update(
+    path: Path,
+    db_path: str | Path,
+) -> None:
     """Register a RAW file in the ledger, then emit its Prefect event."""
     register_raw_file(db_path, path)
     emit_raw_update_event(path)
@@ -57,15 +73,18 @@ def watch_raw_directory(
 
     return watch_directory(
         directory=raw_directory,
-        callback=lambda raw_path: register_and_emit_raw_update(raw_path, db_path),
+        callback=lambda raw_path: register_and_emit_raw_update(
+            raw_path,
+            db_path,
+        ),
         pattern="*.raw",
     )
 
 
 def emit_transect_update_event(path: Path) -> None:
     """Emit a Prefect event when the transect CSV is updated."""
 
-    emit_event(
+    event = emit_event(
         event=TRANSECT_UPDATE_EVENT,
         resource={
             "prefect.resource.id": TRANSECT_RESOURCE_ID,
@@ -80,11 +99,14 @@ def emit_transect_update_event(path: Path) -> None:
         ],
     )
 
+    if event is not None:
+        _flush_events()
+
 
 def watch_transect_file(path: str | Path):
     """Watch the transect start/end CSV and emit a Prefect event on update."""
 
     return watch_file(
         target_file=path,
         callback=emit_transect_update_event,
-    )
+    )