From bdcef4e79fc27e3e6a7699ccad7353dc24f00d47 Mon Sep 17 00:00:00 2001 From: Elias Capriles Date: Mon, 24 Aug 2026 06:52:40 -0700 Subject: [PATCH 1/8] Add CPS Flow last additions --- src/echodataflow/flows/flows_CPS.py | 329 +++++++++++++++++++--------- 1 file changed, 229 insertions(+), 100 deletions(-) diff --git a/src/echodataflow/flows/flows_CPS.py b/src/echodataflow/flows/flows_CPS.py index 4c8b9846..034f8506 100644 --- a/src/echodataflow/flows/flows_CPS.py +++ b/src/echodataflow/flows/flows_CPS.py @@ -7,6 +7,7 @@ import numpy as np import pandas as pd import xarray as xr +from scipy.signal import convolve2d from prefect import flow, get_client, runtime from prefect.states import Cancelled from echodataflow.flows.flows_helper import deployment_already_running @@ -42,6 +43,29 @@ def _dilate_7x7( coords=da.coords, ) +def _apply_2d_convolution(da): + """Applies a 2D convolution to an xarray DataArray using apply_ufunc.""" + def conv_wrapper(image, kern): + return convolve2d( + image, kern, mode="same", boundary="symm" + ) + kernel_data = np.ones((5, 5)) / 25.0 + kernel = xr.DataArray(kernel_data, dims=["ky", "kx"]) + + convolved = xr.apply_ufunc( + conv_wrapper, + da, + kernel, + input_core_dims=[ + ["y", "x"], + ["ky", "kx"], + ], + output_core_dims=[["y", "x"]], + vectorize=True, + ) + + return convolved + def _mask_above_seafloor( ds: xr.Dataset, @@ -135,6 +159,99 @@ def _mask_above_seafloor( fill_value=False, ) +def _mask_below_seafloor( + ds: xr.Dataset, + bottom_line_path: Path, + channel: str, +) -> xr.DataArray: + lines = er.read_lines_csv( + str(bottom_line_path) + ) + + depth_da = ds["depth"].sel( + channel=channel + ) + + deepest_ping_idx = int( + depth_da.max( + dim="range_sample", + skipna=True, + ) + .argmax(dim="ping_time") + .values + ) + + depth = ( + depth_da + .isel(ping_time=deepest_ping_idx) + .dropna( + dim="range_sample", + how="all", + ) + ) + + valid_length = depth.sizes[ + "range_sample" + ] + + ds_trimmed = ds.isel( + range_sample=slice( + 0, + valid_length, + ) + ) + + sv_target = ds_trimmed["Sv"].sel( + channel=[channel] + ) + + sv_for_regions = xr.DataArray( + sv_target.values, + dims=[ + "channel", + "ping_time", + "depth", + ], + coords={ + "channel": [channel], + "ping_time": sv_target["ping_time"], + "depth": depth.values, + }, + name="Sv", + ) + + bottom_mask, _ = lines.seafloor_mask( + sv_for_regions, + operation="above_below", + method="slinear", + limit_area=None, + limit_direction="both", + ) + + # Removed the ~ operator so True represents below the seafloor + below_mask = bottom_mask.astype(bool) + + if "channel" in below_mask.dims: + below_mask = below_mask.squeeze( + "channel", + drop=True, + ) + + below_mask = below_mask.rename( + {"depth": "range_sample"} + ) + + below_mask = below_mask.assign_coords( + range_sample=ds_trimmed[ + "range_sample" + ] + ) + + # Changed fill_value to True because truncated deep samples are below the seafloor + return below_mask.reindex( + range_sample=ds["range_sample"], + fill_value=True, + ) def _export_nasc_to_echoview_csv( ds_nasc: xr.Dataset, @@ -543,71 +660,25 @@ async def cancel_run(): # ----------------------------------------- # Common geometry # ----------------------------------------- - - aligned = ( - ep.commongrid - .resample_to_geometry( - chunked, - target_variable="Sv", - target_channel=target_channel, - ) + aligned_ds_Sv = ep.commongrid.resample_to_geometry( + ds, + target_variable="Sv", + target_channel=target_channel, ) + aligned_ds_Sv["sound_absorption"] = ds["sound_absorption"] + aligned_ds_Sv = ep.consolidate.add_depth(aligned_ds_Sv) - if "sound_absorption" in ds: - aligned[ - "sound_absorption" - ] = ds[ - "sound_absorption" - ] + ds[["Sv", "echo_range"]] = aligned_ds_Sv[["Sv", "echo_range"]] - aligned = ( - ep.consolidate.add_depth( - aligned - ) - ) - - ds[ - ["Sv", "echo_range"] - ] = aligned[ - ["Sv", "echo_range"] - ] - - ds = ( - ep.consolidate.add_depth( - ds - ) - ) - - # ----------------------------------------- - # Background noise - # ----------------------------------------- - - try: - ds = ( - ep.clean - .remove_background_noise( + for angle_var in ["angle_athwartship", "angle_alongship"]: + if angle_var in ds: + ds[angle_var] = ep.commongrid.resample_to_geometry( ds, - ping_num=20, - range_sample_num=5, - SNR_threshold="5.0dB", - ) - ) + target_variable=angle_var, + target_channel=target_channel, + )[angle_var] - except Exception as exc: - print( - f"{name}: background-noise " - f"removal failed: {exc}" - ) - ds["Sv_corrected"] = ( - ds["Sv"] - ) - - sv_var = ( - "Sv_corrected" - if "Sv_corrected" in ds - else "Sv" - ) # ----------------------------------------- # Detect seafloor with Blackwell @@ -616,35 +687,27 @@ async def cancel_run(): bottom_path = None try: - bottom = ( - ep.mask.detect_seafloor( - ds=ds, - method="blackwell", - params={ - "channel": ( - target_channel - ), - "var_name": "Sv", - "threshold": ( - seafloor_threshold - ), - "offset": ( - seafloor_offset - ), - "r0": ( - seafloor_r0 - ), - "r1": ( - seafloor_r1 - ), - "wtheta": ( - seafloor_wtheta - ), - "wphi": ( - seafloor_wphi - ), - }, - ) + q = 0.95 + + angle_alonghsip_threshold = ds["angle_alongship"].isel(channel=1).rolling(ping_time=7, range_sample=7).mean().quantile([q]).values[0] + angle_athwartship_threshold = ds["angle_athwartship"].isel(channel=1).rolling(ping_time=7, range_sample=7).mean().quantile([q]).values[0] + + seafloor_threshold = [-50, angle_alonghsip_threshold, angle_athwartship_threshold] + seafloor_params = { + "channel": target_channel, + "var_name": "Sv", + "threshold": seafloor_threshold, + "offset": seafloor_offset, + "r0": seafloor_r0, + "r1": seafloor_r1, + "wtheta": 7, + "wphi": 7, + } + + bottom = ep.mask.detect_seafloor( + ds=ds_Sv, + method="blackwell", + params=seafloor_params, ) bottom_df = pd.DataFrame( @@ -693,6 +756,13 @@ async def cancel_run(): # This happens BEFORE CPS classification. # ----------------------------------------- + + sv_var = ( + "Sv_corrected" + if "Sv_corrected" in ds + else "Sv" + ) + target_depth = ( ds["depth"].sel( channel=target_channel @@ -712,6 +782,12 @@ async def cancel_run(): dtype=bool, ) ) + below_seafloor_mask = ( + xr.ones_like( + surface_mask, + dtype=bool, + ) + ) if bottom_path is not None: try: @@ -722,6 +798,13 @@ async def cancel_run(): target_channel, ) ) + below_seafloor_mask = ( + _mask_below_seafloor( + ds, + bottom_path, + target_channel, + ) + ) except Exception as exc: print( @@ -737,23 +820,69 @@ async def cancel_run(): # Save intermediate masks/products for diagnostics ds["surface_mask"] = surface_mask ds["above_seafloor_mask"] = above_seafloor_mask + ds["below_seafloor_mask"] = below_seafloor_mask ds["valid_water_column"] = valid_water_column - ds["Sv_water_column"] = ( - ds["Sv"].where(valid_water_column) - ) + ds["Sv_water_column"] = ds["Sv"].where(valid_water_column) + sv_for_cps = ds[sv_var].where(valid_water_column) - # Broadcast the 2-D water-column mask - # (ping_time, range_sample) over channels. - # - # CPS calculations below therefore never - # see the upper 10 m or the seafloor. - sv_for_cps = ( - ds[sv_var].where( - valid_water_column + # ----------------------------------------- + # Mask and convolve above/below seafloor + # ----------------------------------------- + + # Dimension mappings for convolved2d + dim_map = {"ping_time": "x", "range_sample": "y"} + rev_map = {"x": "ping_time", "y": "range_sample"} + + # Rename variables once to match the expected x/y dims + sv_xy = ds["Sv"].rename(dim_map) + mask_above_xy = above_seafloor_mask.rename(dim_map) + mask_below_xy = below_seafloor_mask.rename(dim_map) + + # Mask and compute both regions + above_xy = sv_xy.where(mask_above_xy).compute() + below_xy = sv_xy.where(mask_below_xy).compute() + + # Convolve, combine, and revert dimension names + convolved_combined = xr.where( + mask_above_xy, + _apply_2d_convolution(above_xy), + _apply_2d_convolution(below_xy) + ).rename(rev_map) + + # Assign back to Sv, automatically restoring the original dimension order (e.g. channel) + ds["Sv"] = convolved_combined.transpose(*ds["Sv"].dims) + + # ----------------------------------------- + # Background noise + # ----------------------------------------- + + try: + ds = ( + ep.clean + .remove_background_noise( + ds, + ping_num=20, + range_sample_num=5, + SNR_threshold="5.0dB", + ) + ) + + except Exception as exc: + print( + f"{name}: background-noise " + f"removal failed: {exc}" + ) + + ds["Sv_corrected"] = ( + ds["Sv"] ) - ) + sv_var = ( + "Sv_corrected" + if "Sv_corrected" in ds + else "Sv" + ) # ----------------------------------------- # CPS classifier # ----------------------------------------- From 82aa932abccc2a834aae0bae4e0722318597eefa Mon Sep 17 00:00:00 2001 From: Elias Capriles Date: Mon, 24 Aug 2026 06:58:08 -0700 Subject: [PATCH 2/8] Fix typo --- src/echodataflow/flows/flows_CPS.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/echodataflow/flows/flows_CPS.py b/src/echodataflow/flows/flows_CPS.py index 034f8506..8abd88f5 100644 --- a/src/echodataflow/flows/flows_CPS.py +++ b/src/echodataflow/flows/flows_CPS.py @@ -705,7 +705,7 @@ async def cancel_run(): } bottom = ep.mask.detect_seafloor( - ds=ds_Sv, + ds=ds, method="blackwell", params=seafloor_params, ) From 5fd9c95d55caf23342d5a48014b2118d099074c7 Mon Sep 17 00:00:00 2001 From: Lloyd Izard <76954858+LOCEANlloydizard@users.noreply.github.com> Date: Tue, 25 Aug 2026 22:21:08 -0700 Subject: [PATCH 3/8] Add CPS raw2Sv realtime processing flow --- src/echodataflow/deployment/flow_registry.py | 7 +- src/echodataflow/flows/flows_CPS.py | 27 +-- src/echodataflow/flows/flows_acoustics.py | 185 +++++++++++++++++- src/echodataflow/flows/flows_simulation.py | 11 +- src/echodataflow/flows/flows_viz_cloud.py | 10 +- .../operations/operations_watchdog.py | 38 +++- src/echodataflow/utils/processing_ledger.py | 60 ++++-- tests/test_flow_raw2Sv.py | 8 +- tests/test_flow_simulate_transects.py | 12 +- tests/test_operations_watchdog.py | 38 ++++ tests/test_processing_ledger.py | 43 +++- 11 files changed, 363 insertions(+), 76 deletions(-) diff --git a/src/echodataflow/deployment/flow_registry.py b/src/echodataflow/deployment/flow_registry.py index 92c9822d..5425e920 100644 --- a/src/echodataflow/deployment/flow_registry.py +++ b/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", ), diff --git a/src/echodataflow/flows/flows_CPS.py b/src/echodataflow/flows/flows_CPS.py index 4c8b9846..469198ab 100644 --- a/src/echodataflow/flows/flows_CPS.py +++ b/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( diff --git a/src/echodataflow/flows/flows_acoustics.py b/src/echodataflow/flows/flows_acoustics.py index 5d8c0152..4169b498 100644 --- a/src/echodataflow/flows/flows_acoustics.py +++ b/src/echodataflow/flows/flows_acoustics.py @@ -31,11 +31,13 @@ RawToSvSettings, RawToSvWorkItem, ) -from echodataflow.operations.operations_postprocessing import ( - build_MVBS_ledger, - build_Sv_ledger, - 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 ( @@ -227,6 +229,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, diff --git a/src/echodataflow/flows/flows_simulation.py b/src/echodataflow/flows/flows_simulation.py index 1e3b8752..081d98b4 100644 --- a/src/echodataflow/flows/flows_simulation.py +++ b/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.") diff --git a/src/echodataflow/flows/flows_viz_cloud.py b/src/echodataflow/flows/flows_viz_cloud.py index 7e472b2f..afd62ac7 100644 --- a/src/echodataflow/flows/flows_viz_cloud.py +++ b/src/echodataflow/flows/flows_viz_cloud.py @@ -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: diff --git a/src/echodataflow/operations/operations_watchdog.py b/src/echodataflow/operations/operations_watchdog.py index e49c0cbc..ee0a5995 100644 --- a/src/echodataflow/operations/operations_watchdog.py +++ b/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,11 +35,20 @@ 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: - """Register a RAW file in the ledger, then emit its Prefect event.""" - register_raw_file(db_path, path) - emit_raw_update_event(path) + +def register_and_emit_raw_update( + path: Path, + db_path: str | Path, +) -> None: + """Register a RAW file and emit an event only when processing is needed.""" + if register_raw_file(db_path, path): + emit_raw_update_event(path) def watch_raw_directory( @@ -57,7 +73,10 @@ 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", ) @@ -65,7 +84,7 @@ def watch_raw_directory( 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,6 +99,9 @@ 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.""" @@ -87,4 +109,4 @@ def watch_transect_file(path: str | Path): return watch_file( target_file=path, callback=emit_transect_update_event, - ) + ) \ No newline at end of file diff --git a/src/echodataflow/utils/processing_ledger.py b/src/echodataflow/utils/processing_ledger.py index 01be1134..6c2aa0cb 100644 --- a/src/echodataflow/utils/processing_ledger.py +++ b/src/echodataflow/utils/processing_ledger.py @@ -15,6 +15,8 @@ select, update, ) +from sqlalchemy.dialects.postgresql import insert as postgresql_insert +from sqlalchemy.dialects.sqlite import insert as sqlite_insert from sqlalchemy.engine import Engine from sqlalchemy.sql import func @@ -115,8 +117,12 @@ def initialize_ledger(db_path: str | Path) -> None: def register_raw_file( db_path: str | Path, raw_path: str | Path, -) -> None: - """Register a RAW file, re-queueing it only when its contents changed.""" +) -> bool: + """Register a RAW file, re-queueing it only when its contents changed. + + Returns True when the ledger changed and downstream processing should + be notified, otherwise False. + """ raw_path = Path(raw_path) stat = raw_path.stat() @@ -132,16 +138,46 @@ def register_raw_file( ).first() if existing is None: - conn.execute( - insert(raw_sv).values( - raw_path=str(raw_path), - raw_filename=raw_path.name, - file_size=stat.st_size, - file_mtime_ns=stat.st_mtime_ns, - status="pending", + values = { + "raw_path": str(raw_path), + "raw_filename": raw_path.name, + "file_size": stat.st_size, + "file_mtime_ns": stat.st_mtime_ns, + "status": "pending", + } + + if engine.dialect.name == "postgresql": + stmt = ( + postgresql_insert(raw_sv) + .values(**values) + .on_conflict_do_nothing( + index_elements=[raw_sv.c.raw_path] + ) ) - ) - return + elif engine.dialect.name == "sqlite": + stmt = ( + sqlite_insert(raw_sv) + .values(**values) + .on_conflict_do_nothing( + index_elements=[raw_sv.c.raw_path] + ) + ) + else: + stmt = insert(raw_sv).values(**values) + + result = conn.execute(stmt) + + if result.rowcount == 1: + return True + + # Another callback registered the same RAW between our + # SELECT and INSERT. Re-read it instead of failing. + existing = conn.execute( + select( + raw_sv.c.file_size, + raw_sv.c.file_mtime_ns, + ).where(raw_sv.c.raw_path == str(raw_path)) + ).one() if ( existing.file_size != stat.st_size @@ -161,7 +197,9 @@ def register_raw_file( updated_at=func.current_timestamp(), ) ) + return True + return False def get_raw_files_to_process( db_path: str | Path, diff --git a/tests/test_flow_raw2Sv.py b/tests/test_flow_raw2Sv.py index 3298596d..2e3fb92a 100644 --- a/tests/test_flow_raw2Sv.py +++ b/tests/test_flow_raw2Sv.py @@ -1,10 +1,6 @@ import pytest from types import SimpleNamespace -pytestmark = pytest.mark.skip( - reason="Temporarily disabled while flows_acoustics.py is reverted to 3c1b9d3" -) - from echodataflow.flows import flows_acoustics @@ -65,7 +61,7 @@ def __call__(self, work_item, settings): FakeTask(), ) - flows_acoustics.flow_raw2Sv.fn( + flows_acoustics.flow_raw2Sv_CPS.fn( path_main=str(tmp_path), new_file_num_limit=1, ) @@ -148,7 +144,7 @@ async def set_flow_run_state(self, **kwargs): ) with pytest.raises(RuntimeError, match="1 errors during raw to Sv conversion"): - flows_acoustics.flow_raw2Sv.fn( + flows_acoustics.flow_raw2Sv_CPS.fn( path_main=str(tmp_path), new_file_num_limit=1, ) diff --git a/tests/test_flow_simulate_transects.py b/tests/test_flow_simulate_transects.py index 09f80dfc..09a0d5ab 100644 --- a/tests/test_flow_simulate_transects.py +++ b/tests/test_flow_simulate_transects.py @@ -34,23 +34,23 @@ def test_flow_simulate_transects_opens_closes_and_advances(monkeypatch, tmp_path "max_transects": 2, } - # First run: write complete transect 001. + # First run: open transect 001. flows_simulation.flow_simulate_transects.fn(**kwargs) df = pd.read_csv(transect_csv, dtype="string") assert df["transectPart"].tolist() == ["001"] assert df.loc[0, "transectStart"] == "2024-07-07T00:00:00+00:00" - assert df.loc[0, "transectEnd"] == "2024-07-07T00:10:00+00:00" + assert pd.isna(df.loc[0, "transectEnd"]) - # Second run: append complete transect 002. + # Second run: close transect 001. flows_simulation.flow_simulate_transects.fn(**kwargs) df = pd.read_csv(transect_csv, dtype="string") - assert df["transectPart"].tolist() == ["001", "002"] - assert df.loc[1, "transectStart"] == "2024-07-07T00:10:00+00:00" - assert df.loc[1, "transectEnd"] == "2024-07-07T00:20:00+00:00" + assert df["transectPart"].tolist() == ["001"] + assert df.loc[0, "transectStart"] == "2024-07-07T00:00:00+00:00" + assert df.loc[0, "transectEnd"] == "2024-07-07T00:10:00+00:00" def test_flow_simulate_transects_stops_after_maximum(monkeypatch, tmp_path, capsys): diff --git a/tests/test_operations_watchdog.py b/tests/test_operations_watchdog.py index 1c9b4f72..5f56a3ec 100644 --- a/tests/test_operations_watchdog.py +++ b/tests/test_operations_watchdog.py @@ -72,6 +72,7 @@ def test_register_and_emit_raw_update(monkeypatch, tmp_path): def fake_register_raw_file(db, path): called["registered"] = (db, path) + return True def fake_emit_raw_update_event(path): called["emitted"] = path @@ -97,6 +98,43 @@ def fake_emit_raw_update_event(path): assert called["emitted"] == raw_file +def test_register_and_emit_raw_update_skips_unchanged_duplicate( + monkeypatch, + tmp_path, +): + raw_file = tmp_path / "example.raw" + db_path = tmp_path / "processing.db" + + called = {} + + def fake_register_raw_file(db, path): + called["registered"] = (db, path) + return False + + def fake_emit_raw_update_event(path): + called["emitted"] = path + + monkeypatch.setattr( + operations_watchdog, + "register_raw_file", + fake_register_raw_file, + ) + + monkeypatch.setattr( + operations_watchdog, + "emit_raw_update_event", + fake_emit_raw_update_event, + ) + + operations_watchdog.register_and_emit_raw_update( + raw_file, + db_path, + ) + + assert called["registered"] == (db_path, raw_file) + assert "emitted" not in called + + def test_watch_raw_directory_reconciles_existing_raw_files(monkeypatch, tmp_path): raw_a = tmp_path / "a.raw" raw_b = tmp_path / "b.raw" diff --git a/tests/test_processing_ledger.py b/tests/test_processing_ledger.py index 28f310e6..ec2bcea7 100644 --- a/tests/test_processing_ledger.py +++ b/tests/test_processing_ledger.py @@ -11,6 +11,7 @@ register_raw_file, resolve_database, ) +from concurrent.futures import ThreadPoolExecutor def test_database_url_from_path(tmp_path): @@ -73,8 +74,9 @@ def test_register_raw_file(tmp_path): raw_path.touch() initialize_ledger(db_path) - register_raw_file(db_path, raw_path) - register_raw_file(db_path, raw_path) + + assert register_raw_file(db_path, raw_path) is True + assert register_raw_file(db_path, raw_path) is False with sqlite3.connect(db_path) as conn: rows = conn.execute( @@ -234,7 +236,8 @@ def test_register_raw_file_requeues_changed_file(tmp_path): raw_path.write_bytes(b"first") initialize_ledger(db_path) - register_raw_file(db_path, raw_path) + + assert register_raw_file(db_path, raw_path) is True mark_raw_completed( db_path, @@ -245,7 +248,8 @@ def test_register_raw_file_requeues_changed_file(tmp_path): ) raw_path.write_bytes(b"first plus more data") - register_raw_file(db_path, raw_path) + + assert register_raw_file(db_path, raw_path) is True with sqlite3.connect(db_path) as conn: row = conn.execute( @@ -268,4 +272,33 @@ def test_register_raw_file_large_mtime(tmp_path): register_raw_file(db_path, raw_path) # A nanosecond mtime is much larger than a 32-bit integer. - assert raw_path.stat().st_mtime_ns > 2**31 \ No newline at end of file + assert raw_path.stat().st_mtime_ns > 2**31 + +def test_register_raw_file_concurrently_is_idempotent(tmp_path): + db_path = tmp_path / "processing.db" + raw_path = tmp_path / "test.raw" + + raw_path.write_bytes(b"raw-data") + initialize_ledger(db_path) + + with ThreadPoolExecutor(max_workers=2) as executor: + futures = [ + executor.submit(register_raw_file, db_path, raw_path) + for _ in range(2) + ] + + results = [future.result() for future in futures] + + assert sorted(results) == [False, True] + + with sqlite3.connect(db_path) as conn: + rows = conn.execute( + """ + SELECT raw_path, raw_filename, status + FROM raw_sv + """ + ).fetchall() + + assert rows == [ + (str(raw_path), "test.raw", "pending"), + ] \ No newline at end of file From 209d8a5e374f9e96cc1ab9b75b6f1d78ed94ce68 Mon Sep 17 00:00:00 2001 From: Lloyd Izard <76954858+LOCEANlloydizard@users.noreply.github.com> Date: Wed, 26 Aug 2026 06:31:49 -0700 Subject: [PATCH 4/8] fix imports and lint issues --- src/echodataflow/flows/flows_acoustics.py | 9 ++++++++- src/echodataflow/flows/flows_viz_cloud.py | 2 +- 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/src/echodataflow/flows/flows_acoustics.py b/src/echodataflow/flows/flows_acoustics.py index 4169b498..7e214aa9 100644 --- a/src/echodataflow/flows/flows_acoustics.py +++ b/src/echodataflow/flows/flows_acoustics.py @@ -31,6 +31,13 @@ RawToSvSettings, RawToSvWorkItem, ) +from echodataflow.operations.operations_postprocessing import ( + build_MVBS_ledger, + build_Sv_ledger, + plan_mvbs_slices, + read_or_create_ledger, +) + from echodataflow.utils.processing_ledger import ( get_raw_files_to_process, initialize_ledger, @@ -144,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, diff --git a/src/echodataflow/flows/flows_viz_cloud.py b/src/echodataflow/flows/flows_viz_cloud.py index afd62ac7..15599c20 100644 --- a/src/echodataflow/flows/flows_viz_cloud.py +++ b/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() From 2b396d8a1cc9e22a909883bfe244a8e939017ca7 Mon Sep 17 00:00:00 2001 From: Lloyd Izard <76954858+LOCEANlloydizard@users.noreply.github.com> Date: Wed, 26 Aug 2026 08:04:48 -0700 Subject: [PATCH 5/8] fix raw watcher event emission --- src/echodataflow/operations/operations_watchdog.py | 6 +++--- tests/test_operations_watchdog.py | 5 ++--- 2 files changed, 5 insertions(+), 6 deletions(-) diff --git a/src/echodataflow/operations/operations_watchdog.py b/src/echodataflow/operations/operations_watchdog.py index ee0a5995..97cf846d 100644 --- a/src/echodataflow/operations/operations_watchdog.py +++ b/src/echodataflow/operations/operations_watchdog.py @@ -46,9 +46,9 @@ def register_and_emit_raw_update( path: Path, db_path: str | Path, ) -> None: - """Register a RAW file and emit an event only when processing is needed.""" - if register_raw_file(db_path, path): - emit_raw_update_event(path) + """Register a RAW file in the ledger, then emit its Prefect event.""" + register_raw_file(db_path, path) + emit_raw_update_event(path) def watch_raw_directory( diff --git a/tests/test_operations_watchdog.py b/tests/test_operations_watchdog.py index 5f56a3ec..35dc9f49 100644 --- a/tests/test_operations_watchdog.py +++ b/tests/test_operations_watchdog.py @@ -98,7 +98,7 @@ def fake_emit_raw_update_event(path): assert called["emitted"] == raw_file -def test_register_and_emit_raw_update_skips_unchanged_duplicate( +def test_register_and_emit_raw_update_emits_after_registration( monkeypatch, tmp_path, ): @@ -109,7 +109,6 @@ def test_register_and_emit_raw_update_skips_unchanged_duplicate( def fake_register_raw_file(db, path): called["registered"] = (db, path) - return False def fake_emit_raw_update_event(path): called["emitted"] = path @@ -132,7 +131,7 @@ def fake_emit_raw_update_event(path): ) assert called["registered"] == (db_path, raw_file) - assert "emitted" not in called + assert called["emitted"] == raw_file def test_watch_raw_directory_reconciles_existing_raw_files(monkeypatch, tmp_path): From fe02a5e566bf10c0db8f2cc4b3d5b191691c5165 Mon Sep 17 00:00:00 2001 From: Wu-Jung Lee Date: Wed, 26 Aug 2026 10:58:34 -0700 Subject: [PATCH 6/8] small patch --- .../operations/operations_watchdog.py | 17 ++++++----- tests/test_operations_watchdog.py | 29 +++++++++++++++++++ 2 files changed, 38 insertions(+), 8 deletions(-) diff --git a/src/echodataflow/operations/operations_watchdog.py b/src/echodataflow/operations/operations_watchdog.py index 97cf846d..3a1fb308 100644 --- a/src/echodataflow/operations/operations_watchdog.py +++ b/src/echodataflow/operations/operations_watchdog.py @@ -2,6 +2,7 @@ from prefect.events import emit_event from prefect.events.worker import EventsWorker +from watchdog.observers import Observer from echodataflow.utils.file_watcher import watch_directory, watch_file from echodataflow.utils.processing_ledger import ( @@ -46,16 +47,16 @@ 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) + """Emit a Prefect event only when registration changes the ledger.""" + if register_raw_file(db_path, path): # emit only if ledger is udpated + emit_raw_update_event(path) def watch_raw_directory( path: str | Path, db_path: str | Path, -): - """Watch a directory for new RAW files.""" +) -> Observer: + """Watch a directory for new RAW files and returning a running observer.""" raw_directory = Path(path).resolve() @@ -67,11 +68,11 @@ def watch_raw_directory( for raw_path in existing_raw_files: register_raw_file(db_path, raw_path) - # Wake raw2Sv once after reconciliation. + # Wake raw2Sv once after reconciliation if existing_raw_files: emit_raw_update_event(raw_directory) - return watch_directory( + return watch_directory( # this returns a running observer directory=raw_directory, callback=lambda raw_path: register_and_emit_raw_update( raw_path, @@ -109,4 +110,4 @@ def watch_transect_file(path: str | Path): return watch_file( target_file=path, callback=emit_transect_update_event, - ) \ No newline at end of file + ) diff --git a/tests/test_operations_watchdog.py b/tests/test_operations_watchdog.py index 35dc9f49..183a01b2 100644 --- a/tests/test_operations_watchdog.py +++ b/tests/test_operations_watchdog.py @@ -109,6 +109,7 @@ def test_register_and_emit_raw_update_emits_after_registration( def fake_register_raw_file(db, path): called["registered"] = (db, path) + return True def fake_emit_raw_update_event(path): called["emitted"] = path @@ -134,6 +135,34 @@ def fake_emit_raw_update_event(path): assert called["emitted"] == raw_file +def test_register_and_emit_raw_update_skips_unchanged_file( + monkeypatch, + tmp_path, +): + raw_file = tmp_path / "example.raw" + db_path = tmp_path / "processing.db" + + monkeypatch.setattr( + operations_watchdog, + "register_raw_file", + lambda db, path: False, + ) + + emitted = [] + monkeypatch.setattr( + operations_watchdog, + "emit_raw_update_event", + emitted.append, + ) + + operations_watchdog.register_and_emit_raw_update( + raw_file, + db_path, + ) + + assert emitted == [] + + def test_watch_raw_directory_reconciles_existing_raw_files(monkeypatch, tmp_path): raw_a = tmp_path / "a.raw" raw_b = tmp_path / "b.raw" From dd53edce2df432ba819ebb2425ef0fafee99a205 Mon Sep 17 00:00:00 2001 From: Lloyd Izard <76954858+LOCEANlloydizard@users.noreply.github.com> Date: Thu, 27 Aug 2026 07:30:56 -0700 Subject: [PATCH 7/8] Revert "Merge branch 'Add_cps_flow_edits' into update-raw2sv-for-cps" This reverts commit 1d7a4331bd4c16d33296161f51f984474de0e46b, reversing changes made to fe02a5e566bf10c0db8f2cc4b3d5b191691c5165. --- src/echodataflow/flows/flows_CPS.py | 333 +++++++++------------------- 1 file changed, 101 insertions(+), 232 deletions(-) diff --git a/src/echodataflow/flows/flows_CPS.py b/src/echodataflow/flows/flows_CPS.py index 983d41ac..469198ab 100644 --- a/src/echodataflow/flows/flows_CPS.py +++ b/src/echodataflow/flows/flows_CPS.py @@ -6,10 +6,7 @@ import numpy as np import pandas as pd import xarray as xr -from scipy.signal import convolve2d -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 @@ -42,29 +39,6 @@ def _dilate_7x7( coords=da.coords, ) -def _apply_2d_convolution(da): - """Applies a 2D convolution to an xarray DataArray using apply_ufunc.""" - def conv_wrapper(image, kern): - return convolve2d( - image, kern, mode="same", boundary="symm" - ) - kernel_data = np.ones((5, 5)) / 25.0 - kernel = xr.DataArray(kernel_data, dims=["ky", "kx"]) - - convolved = xr.apply_ufunc( - conv_wrapper, - da, - kernel, - input_core_dims=[ - ["y", "x"], - ["ky", "kx"], - ], - output_core_dims=[["y", "x"]], - vectorize=True, - ) - - return convolved - def _mask_above_seafloor( ds: xr.Dataset, @@ -158,99 +132,6 @@ def _mask_above_seafloor( fill_value=False, ) -def _mask_below_seafloor( - ds: xr.Dataset, - bottom_line_path: Path, - channel: str, -) -> xr.DataArray: - lines = er.read_lines_csv( - str(bottom_line_path) - ) - - depth_da = ds["depth"].sel( - channel=channel - ) - - deepest_ping_idx = int( - depth_da.max( - dim="range_sample", - skipna=True, - ) - .argmax(dim="ping_time") - .values - ) - - depth = ( - depth_da - .isel(ping_time=deepest_ping_idx) - .dropna( - dim="range_sample", - how="all", - ) - ) - - valid_length = depth.sizes[ - "range_sample" - ] - - ds_trimmed = ds.isel( - range_sample=slice( - 0, - valid_length, - ) - ) - - sv_target = ds_trimmed["Sv"].sel( - channel=[channel] - ) - - sv_for_regions = xr.DataArray( - sv_target.values, - dims=[ - "channel", - "ping_time", - "depth", - ], - coords={ - "channel": [channel], - "ping_time": sv_target["ping_time"], - "depth": depth.values, - }, - name="Sv", - ) - - bottom_mask, _ = lines.seafloor_mask( - sv_for_regions, - operation="above_below", - method="slinear", - limit_area=None, - limit_direction="both", - ) - - # Removed the ~ operator so True represents below the seafloor - below_mask = bottom_mask.astype(bool) - - if "channel" in below_mask.dims: - below_mask = below_mask.squeeze( - "channel", - drop=True, - ) - - below_mask = below_mask.rename( - {"depth": "range_sample"} - ) - - below_mask = below_mask.assign_coords( - range_sample=ds_trimmed[ - "range_sample" - ] - ) - - # Changed fill_value to True because truncated deep samples are below the seafloor - return below_mask.reindex( - range_sample=ds["range_sample"], - fill_value=True, - ) def _export_nasc_to_echoview_csv( ds_nasc: xr.Dataset, @@ -637,25 +518,71 @@ def flow_process_CPS( # ----------------------------------------- # Common geometry # ----------------------------------------- - aligned_ds_Sv = ep.commongrid.resample_to_geometry( - ds, - target_variable="Sv", - target_channel=target_channel, + + aligned = ( + ep.commongrid + .resample_to_geometry( + chunked, + target_variable="Sv", + target_channel=target_channel, + ) ) - aligned_ds_Sv["sound_absorption"] = ds["sound_absorption"] - aligned_ds_Sv = ep.consolidate.add_depth(aligned_ds_Sv) - ds[["Sv", "echo_range"]] = aligned_ds_Sv[["Sv", "echo_range"]] + if "sound_absorption" in ds: + aligned[ + "sound_absorption" + ] = ds[ + "sound_absorption" + ] - for angle_var in ["angle_athwartship", "angle_alongship"]: - if angle_var in ds: - ds[angle_var] = ep.commongrid.resample_to_geometry( + aligned = ( + ep.consolidate.add_depth( + aligned + ) + ) + + ds[ + ["Sv", "echo_range"] + ] = aligned[ + ["Sv", "echo_range"] + ] + + ds = ( + ep.consolidate.add_depth( + ds + ) + ) + + # ----------------------------------------- + # Background noise + # ----------------------------------------- + + try: + ds = ( + ep.clean + .remove_background_noise( ds, - target_variable=angle_var, - target_channel=target_channel, - )[angle_var] + ping_num=20, + range_sample_num=5, + SNR_threshold="5.0dB", + ) + ) + except Exception as exc: + print( + f"{name}: background-noise " + f"removal failed: {exc}" + ) + ds["Sv_corrected"] = ( + ds["Sv"] + ) + + sv_var = ( + "Sv_corrected" + if "Sv_corrected" in ds + else "Sv" + ) # ----------------------------------------- # Detect seafloor with Blackwell @@ -664,27 +591,35 @@ def flow_process_CPS( bottom_path = None try: - q = 0.95 - - angle_alonghsip_threshold = ds["angle_alongship"].isel(channel=1).rolling(ping_time=7, range_sample=7).mean().quantile([q]).values[0] - angle_athwartship_threshold = ds["angle_athwartship"].isel(channel=1).rolling(ping_time=7, range_sample=7).mean().quantile([q]).values[0] - - seafloor_threshold = [-50, angle_alonghsip_threshold, angle_athwartship_threshold] - seafloor_params = { - "channel": target_channel, - "var_name": "Sv", - "threshold": seafloor_threshold, - "offset": seafloor_offset, - "r0": seafloor_r0, - "r1": seafloor_r1, - "wtheta": 7, - "wphi": 7, - } - - bottom = ep.mask.detect_seafloor( - ds=ds, - method="blackwell", - params=seafloor_params, + bottom = ( + ep.mask.detect_seafloor( + ds=ds, + method="blackwell", + params={ + "channel": ( + target_channel + ), + "var_name": "Sv", + "threshold": ( + seafloor_threshold + ), + "offset": ( + seafloor_offset + ), + "r0": ( + seafloor_r0 + ), + "r1": ( + seafloor_r1 + ), + "wtheta": ( + seafloor_wtheta + ), + "wphi": ( + seafloor_wphi + ), + }, + ) ) bottom_df = pd.DataFrame( @@ -733,13 +668,6 @@ def flow_process_CPS( # This happens BEFORE CPS classification. # ----------------------------------------- - - sv_var = ( - "Sv_corrected" - if "Sv_corrected" in ds - else "Sv" - ) - target_depth = ( ds["depth"].sel( channel=target_channel @@ -759,12 +687,6 @@ def flow_process_CPS( dtype=bool, ) ) - below_seafloor_mask = ( - xr.ones_like( - surface_mask, - dtype=bool, - ) - ) if bottom_path is not None: try: @@ -775,13 +697,6 @@ def flow_process_CPS( target_channel, ) ) - below_seafloor_mask = ( - _mask_below_seafloor( - ds, - bottom_path, - target_channel, - ) - ) except Exception as exc: print( @@ -797,69 +712,23 @@ def flow_process_CPS( # Save intermediate masks/products for diagnostics ds["surface_mask"] = surface_mask ds["above_seafloor_mask"] = above_seafloor_mask - ds["below_seafloor_mask"] = below_seafloor_mask ds["valid_water_column"] = valid_water_column - ds["Sv_water_column"] = ds["Sv"].where(valid_water_column) - sv_for_cps = ds[sv_var].where(valid_water_column) - - # ----------------------------------------- - # Mask and convolve above/below seafloor - # ----------------------------------------- - - # Dimension mappings for convolved2d - dim_map = {"ping_time": "x", "range_sample": "y"} - rev_map = {"x": "ping_time", "y": "range_sample"} - - # Rename variables once to match the expected x/y dims - sv_xy = ds["Sv"].rename(dim_map) - mask_above_xy = above_seafloor_mask.rename(dim_map) - mask_below_xy = below_seafloor_mask.rename(dim_map) - - # Mask and compute both regions - above_xy = sv_xy.where(mask_above_xy).compute() - below_xy = sv_xy.where(mask_below_xy).compute() - - # Convolve, combine, and revert dimension names - convolved_combined = xr.where( - mask_above_xy, - _apply_2d_convolution(above_xy), - _apply_2d_convolution(below_xy) - ).rename(rev_map) - - # Assign back to Sv, automatically restoring the original dimension order (e.g. channel) - ds["Sv"] = convolved_combined.transpose(*ds["Sv"].dims) - - # ----------------------------------------- - # Background noise - # ----------------------------------------- - - try: - ds = ( - ep.clean - .remove_background_noise( - ds, - ping_num=20, - range_sample_num=5, - SNR_threshold="5.0dB", - ) - ) - - except Exception as exc: - print( - f"{name}: background-noise " - f"removal failed: {exc}" - ) + ds["Sv_water_column"] = ( + ds["Sv"].where(valid_water_column) + ) - ds["Sv_corrected"] = ( - ds["Sv"] + # Broadcast the 2-D water-column mask + # (ping_time, range_sample) over channels. + # + # CPS calculations below therefore never + # see the upper 10 m or the seafloor. + sv_for_cps = ( + ds[sv_var].where( + valid_water_column ) - - sv_var = ( - "Sv_corrected" - if "Sv_corrected" in ds - else "Sv" ) + # ----------------------------------------- # CPS classifier # ----------------------------------------- From 1a547d254a4f7934ade04cbd65d5f4784ad8fee9 Mon Sep 17 00:00:00 2001 From: Lloyd Izard <76954858+LOCEANlloydizard@users.noreply.github.com> Date: Thu, 27 Aug 2026 16:35:58 -0700 Subject: [PATCH 8/8] finalize CPS polling and deployment behavior --- pyproject.toml | 2 +- src/echodataflow/flows/flows_CPS.py | 35 ++++- src/echodataflow/flows/flows_viz_cloud.py | 12 +- .../operations/operations_watchdog.py | 47 ++----- .../services/viz_echogram_track_cps.py | 75 ++++++----- tests/test_operations_watchdog.py | 121 ++---------------- 6 files changed, 98 insertions(+), 194 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 86c587c5..7aba605e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -44,7 +44,7 @@ dependencies = [ "openpyxl", "pandas", "panel", - "prefect[shell,dask]", + "prefect[shell,dask]==3.8.4", "PyYAML", "s3fs", "shapely", diff --git a/src/echodataflow/flows/flows_CPS.py b/src/echodataflow/flows/flows_CPS.py index 469198ab..b80c313e 100644 --- a/src/echodataflow/flows/flows_CPS.py +++ b/src/echodataflow/flows/flows_CPS.py @@ -252,6 +252,7 @@ def flow_process_CPS( range_bin: str = "10m", dist_bin: str = "0.5nmi", nasc_process_id: int = 1928, + exclude_before: str | None = None, ): path_main = Path(path_main) @@ -298,7 +299,7 @@ def flow_process_CPS( ) # --------------------------------------------- - # Find completed transects still needing CPS + # Select eligible completed transects # --------------------------------------------- current = pd.read_csv( @@ -311,8 +312,38 @@ def flow_process_CPS( }, ) + eligible = current.copy() + + eligible["transectStart"] = pd.to_datetime( + eligible["transectStart"], + utc=True, + errors="coerce", + ) + + eligible["transectEnd"] = pd.to_datetime( + eligible["transectEnd"], + utc=True, + errors="coerce", + ) + + # Ignore historical transects that started + # before this deployment's processing window. + if exclude_before is not None: + cutoff = pd.Timestamp( + exclude_before + ) + + if cutoff.tzinfo is None: + cutoff = cutoff.tz_localize("UTC") + else: + cutoff = cutoff.tz_convert("UTC") + + eligible = eligible.loc[ + eligible["transectStart"] >= cutoff + ].copy() + # Ignore transects that have not finished yet. - completed = current.dropna( + completed = eligible.dropna( subset=[ "transectPart", "transectStart", diff --git a/src/echodataflow/flows/flows_viz_cloud.py b/src/echodataflow/flows/flows_viz_cloud.py index 15599c20..6158381e 100644 --- a/src/echodataflow/flows/flows_viz_cloud.py +++ b/src/echodataflow/flows/flows_viz_cloud.py @@ -257,17 +257,9 @@ def flow_update_cache_CPS( # ----------------------------------------------------- def _transect_number(path: Path) -> int: - return int( - path.name - .replace("transect_", "") - .replace("_CPS.zarr", "") - ) - + return int(path.name.replace("transect_", "").replace("_CPS.zarr", "")) - cps_files = sorted( - path_CPS.glob("transect_*_CPS.zarr"), - key=_transect_number, - ) + cps_files = sorted(path_CPS.glob("transect_*_CPS.zarr"), key=_transect_number) if not cps_files: print( diff --git a/src/echodataflow/operations/operations_watchdog.py b/src/echodataflow/operations/operations_watchdog.py index 3a1fb308..8649d5f9 100644 --- a/src/echodataflow/operations/operations_watchdog.py +++ b/src/echodataflow/operations/operations_watchdog.py @@ -11,9 +11,6 @@ ) -RAW_UPDATE_EVENT = "echodataflow.raw.updated" -RAW_RESOURCE_ID = "raw-monitor" - TRANSECT_UPDATE_EVENT = "echodataflow.transect.updated" TRANSECT_RESOURCE_ID = "transect-start-end-time" TRANSECT_RELATED_RESOURCE_ID = "transect-monitor" @@ -24,57 +21,31 @@ def _flush_events() -> None: EventsWorker.instance().wait_until_empty() -def emit_raw_update_event(path: Path) -> None: - """Emit a Prefect event when a RAW file arrives.""" - - event = emit_event( - event=RAW_UPDATE_EVENT, - resource={ - "prefect.resource.id": RAW_RESOURCE_ID, - "prefect.resource.name": RAW_RESOURCE_ID, - "path": str(path), - }, - ) - - 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( +def register_raw_update( path: Path, db_path: str | Path, ) -> None: - """Emit a Prefect event only when registration changes the ledger.""" - if register_raw_file(db_path, path): # emit only if ledger is udpated - emit_raw_update_event(path) + """Register a RAW file change in the processing ledger.""" + register_raw_file(db_path, path) def watch_raw_directory( path: str | Path, db_path: str | Path, ) -> Observer: - """Watch a directory for new RAW files and returning a running observer.""" + """Watch a directory for RAW files and keep the processing ledger updated.""" raw_directory = Path(path).resolve() initialize_ledger(db_path) - # Reconcile files that already exist when the watcher starts - existing_raw_files = list(raw_directory.glob("*.raw")) - - for raw_path in existing_raw_files: + # Reconcile files that already exist when the watcher starts. + for raw_path in raw_directory.glob("*.raw"): register_raw_file(db_path, raw_path) - # Wake raw2Sv once after reconciliation - if existing_raw_files: - emit_raw_update_event(raw_directory) - - return watch_directory( # this returns a running observer + return watch_directory( directory=raw_directory, - callback=lambda raw_path: register_and_emit_raw_update( + callback=lambda raw_path: register_raw_update( raw_path, db_path, ), @@ -110,4 +81,4 @@ def watch_transect_file(path: str | Path): return watch_file( target_file=path, callback=emit_transect_update_event, - ) + ) \ No newline at end of file diff --git a/src/echodataflow/services/viz_echogram_track_cps.py b/src/echodataflow/services/viz_echogram_track_cps.py index 8b4806d6..826d52f9 100644 --- a/src/echodataflow/services/viz_echogram_track_cps.py +++ b/src/echodataflow/services/viz_echogram_track_cps.py @@ -299,7 +299,7 @@ def plot_nasc( ds_nasc: xr.Dataset, title: str, ): - """Plot NASC as vertical bars along distance.""" + """Plot depth-integrated NASC along transect time.""" if "NASC" not in ds_nasc: return pn.pane.Markdown( @@ -310,11 +310,9 @@ def plot_nasc( # Select the channel closest to the configured target frequency. if "channel" in nasc.dims: - target_channel = ( - pick_channel_by_frequency( - ds_nasc, - TARGET_FREQUENCY, - ) + target_channel = pick_channel_by_frequency( + ds_nasc, + TARGET_FREQUENCY, ) nasc = nasc.sel( @@ -326,50 +324,62 @@ def plot_nasc( frequency_nominal=0 ) - # Remove singleton dimensions nasc = nasc.squeeze( drop=True ) print( - "NASC dims:", + "NASC dims before integration:", nasc.dims, ) print( - "NASC shape:", + "NASC shape before integration:", nasc.shape, ) - print( - "NASC coords:", - list( - nasc.coords - ), - ) - - # NASC should ultimately be one value per horizontal interval. - # If another dimension remains, integrate/sum over it for plotting. - while nasc.ndim > 1: - dim_to_reduce = ( - nasc.dims[0] - ) + # NASC is depth-resolved in the stored product: + # distance x depth. + # + # For the dashboard, integrate explicitly over depth so that + # one NASC value remains for each horizontal interval. + if "depth" in nasc.dims: nasc = nasc.sum( - dim=dim_to_reduce, + dim="depth", skipna=True, ) - dim = nasc.dims[0] + nasc = nasc.squeeze( + drop=True + ) + + print( + "NASC dims after integration:", + nasc.dims, + ) + print( + "NASC shape after integration:", + nasc.shape, + ) + + # Use the representative ping time associated with each + # horizontal NASC interval so the plot aligns with the echogram. + if "ping_time" in ds_nasc: + x = pd.to_datetime( + ds_nasc["ping_time"].values + ) + xlabel = "Time" + kdim = "ping_time" - if "distance" in nasc.coords: - x = nasc[ - "distance" - ].values + elif "distance" in nasc.coords: + x = nasc["distance"].values xlabel = "Distance (nmi)" + kdim = "distance" + else: - x = nasc[ - dim - ].values + dim = nasc.dims[0] + x = nasc[dim].values xlabel = dim + kdim = dim curve = hv.Curve( ( @@ -377,7 +387,7 @@ def plot_nasc( nasc.values, ), kdims=[ - xlabel, + kdim, ], vdims=[ "NASC", @@ -400,7 +410,6 @@ def plot_nasc( ylabel="NASC", ) - # --------------------------------------------------------------------- # Latest transect plotting # --------------------------------------------------------------------- diff --git a/tests/test_operations_watchdog.py b/tests/test_operations_watchdog.py index 183a01b2..e7883971 100644 --- a/tests/test_operations_watchdog.py +++ b/tests/test_operations_watchdog.py @@ -1,29 +1,6 @@ from echodataflow.operations import operations_watchdog -def test_emit_raw_update_event(monkeypatch, tmp_path): - raw_file = tmp_path / "example.raw" - raw_file.touch() - - emitted = {} - - def fake_emit_event(**kwargs): - emitted.update(kwargs) - - monkeypatch.setattr( - operations_watchdog, - "emit_event", - fake_emit_event, - ) - - operations_watchdog.emit_raw_update_event(raw_file) - - assert emitted["event"] == "echodataflow.raw.updated" - assert emitted["resource"]["prefect.resource.id"] == "raw-monitor" - assert emitted["resource"]["prefect.resource.name"] == "raw-monitor" - assert emitted["resource"]["path"] == str(raw_file) - - def test_watch_raw_directory(monkeypatch, tmp_path): called = {} db_path = tmp_path / "processing.db" @@ -59,12 +36,12 @@ def fake_watch_directory( ) assert result == "observer" - assert called["directory"] == tmp_path + assert called["directory"] == tmp_path.resolve() assert called["db_path"] == db_path assert called["pattern"] == "*.raw" -def test_register_and_emit_raw_update(monkeypatch, tmp_path): +def test_register_raw_update(monkeypatch, tmp_path): raw_file = tmp_path / "example.raw" db_path = tmp_path / "processing.db" @@ -72,10 +49,6 @@ def test_register_and_emit_raw_update(monkeypatch, tmp_path): def fake_register_raw_file(db, path): called["registered"] = (db, path) - return True - - def fake_emit_raw_update_event(path): - called["emitted"] = path monkeypatch.setattr( operations_watchdog, @@ -83,96 +56,27 @@ def fake_emit_raw_update_event(path): fake_register_raw_file, ) - monkeypatch.setattr( - operations_watchdog, - "emit_raw_update_event", - fake_emit_raw_update_event, - ) - - operations_watchdog.register_and_emit_raw_update( + operations_watchdog.register_raw_update( raw_file, db_path, ) assert called["registered"] == (db_path, raw_file) - assert called["emitted"] == raw_file -def test_register_and_emit_raw_update_emits_after_registration( +def test_watch_raw_directory_reconciles_existing_raw_files( monkeypatch, tmp_path, ): - raw_file = tmp_path / "example.raw" - db_path = tmp_path / "processing.db" - - called = {} - - def fake_register_raw_file(db, path): - called["registered"] = (db, path) - return True - - def fake_emit_raw_update_event(path): - called["emitted"] = path - - monkeypatch.setattr( - operations_watchdog, - "register_raw_file", - fake_register_raw_file, - ) - - monkeypatch.setattr( - operations_watchdog, - "emit_raw_update_event", - fake_emit_raw_update_event, - ) - - operations_watchdog.register_and_emit_raw_update( - raw_file, - db_path, - ) - - assert called["registered"] == (db_path, raw_file) - assert called["emitted"] == raw_file - - -def test_register_and_emit_raw_update_skips_unchanged_file( - monkeypatch, - tmp_path, -): - raw_file = tmp_path / "example.raw" - db_path = tmp_path / "processing.db" - - monkeypatch.setattr( - operations_watchdog, - "register_raw_file", - lambda db, path: False, - ) - - emitted = [] - monkeypatch.setattr( - operations_watchdog, - "emit_raw_update_event", - emitted.append, - ) - - operations_watchdog.register_and_emit_raw_update( - raw_file, - db_path, - ) - - assert emitted == [] - - -def test_watch_raw_directory_reconciles_existing_raw_files(monkeypatch, tmp_path): raw_a = tmp_path / "a.raw" raw_b = tmp_path / "b.raw" + raw_a.touch() raw_b.touch() db_path = tmp_path / "processing.db" registered = [] - emitted = [] monkeypatch.setattr( operations_watchdog, @@ -180,12 +84,6 @@ def test_watch_raw_directory_reconciles_existing_raw_files(monkeypatch, tmp_path lambda db, path: registered.append((db, path)), ) - monkeypatch.setattr( - operations_watchdog, - "emit_raw_update_event", - emitted.append, - ) - monkeypatch.setattr( operations_watchdog, "watch_directory", @@ -198,11 +96,11 @@ def test_watch_raw_directory_reconciles_existing_raw_files(monkeypatch, tmp_path ) assert result == "observer" + assert {path for _, path in registered} == { raw_a.resolve(), raw_b.resolve(), } - assert emitted == [tmp_path.resolve()] def test_emit_transect_update_event(monkeypatch, tmp_path): @@ -223,5 +121,8 @@ def fake_emit_event(**kwargs): operations_watchdog.emit_transect_update_event(target) assert emitted["event"] == "echodataflow.transect.updated" - assert emitted["resource"]["prefect.resource.id"] == "transect-start-end-time" - assert emitted["resource"]["path"] == str(target) + assert ( + emitted["resource"]["prefect.resource.id"] + == "transect-start-end-time" + ) + assert emitted["resource"]["path"] == str(target) \ No newline at end of file