From 60a31591a7e375ff2b8f3f59aaa8ceb6b5982efe Mon Sep 17 00:00:00 2001 From: Lloyd Izard <76954858+LOCEANlloydizard@users.noreply.github.com> Date: Fri, 14 Aug 2026 21:12:55 -0700 Subject: [PATCH 1/7] Add SQLite processing ledger for realtime workflows --- src/echodataflow/flows/flows_acoustics.py | 264 ++++++++++---------- src/echodataflow/flows/flows_transect.py | 54 +--- src/echodataflow/utils/processing_ledger.py | 226 +++++++++++++++++ src/echodataflow/utils/raw_monitor.py | 19 +- tests/test_flow_raw2sv.py | 170 +++++++++++++ tests/test_flow_transect.py | 100 ++++---- tests/test_processing_ledger.py | 233 +++++++++++++++++ tests/test_raw_monitor.py | 52 +++- 8 files changed, 879 insertions(+), 239 deletions(-) create mode 100644 src/echodataflow/utils/processing_ledger.py create mode 100644 tests/test_flow_raw2sv.py create mode 100644 tests/test_processing_ledger.py diff --git a/src/echodataflow/flows/flows_acoustics.py b/src/echodataflow/flows/flows_acoustics.py index 6068c4f7..eb0f0cf1 100644 --- a/src/echodataflow/flows/flows_acoustics.py +++ b/src/echodataflow/flows/flows_acoustics.py @@ -18,7 +18,6 @@ from echodataflow.utils.manifests import ( MVBS_COLUMNS_POSTPROCESSING, MVBS_COLUMNS_REALTIME, - SV_COLUMNS_REALTIME, SV_COLUMNS_POSTPROCESSING, filter_time_range, read_manifest, @@ -28,7 +27,6 @@ CreateMVBSResult, CreateMVBSSettings, CreateMVBSWorkItem, - RawToSvResult, RawToSvSettings, RawToSvWorkItem, ) @@ -50,6 +48,15 @@ extract_datetime_from_filename, ) +from echodataflow.utils.processing_ledger import ( + get_completed_sv_files, + get_raw_files_to_process, + initialize_ledger, + mark_raw_completed, + mark_raw_failed, + mark_raw_processing, +) + # Turn on verbose logging for echopype # otherwise all logging will be muted ep.utils.log.verbose() @@ -66,13 +73,10 @@ def flow_raw2Sv( sonar_model: str = "EK80", datagram_type: str | None = None, nmea_sentence: str | None = None, - filename_pattern: str = "*.raw", path_main: str = "", - path_raw: str = "", - file_Sv_csv: str = "Sv_files.csv", + processing_db: str = "processing.db", new_file_num_limit: int = 50, ): - # Check if the deployment is already running already_running = asyncio.run(deployment_already_running()) if already_running: @@ -81,83 +85,58 @@ 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"), + state=Cancelled( + message="Another instance of this flow is already running" + ), ) asyncio.run(cancel_run()) - return # exit the flow early + return # Assemble paths path_Sv_zarr = Path(path_main) / "Sv" - file_Sv_csv = Path(path_main) / file_Sv_csv - path_raw = Path(path_raw) + db_path = Path(path_main) / processing_db + + initialize_ledger(db_path) # Set up folder to store converted Sv zarr - if not path_Sv_zarr.exists(): - path_Sv_zarr.mkdir(parents=True, exist_ok=True) - path_Sv_zarr = str(path_Sv_zarr) # convert backto string to pass into task + path_Sv_zarr.mkdir(parents=True, exist_ok=True) - # Load info dataframe containing raw to Sv correspondence - sv_manifest_exists = file_Sv_csv.exists() - df_Sv = read_manifest( - file_Sv_csv, - SV_COLUMNS_REALTIME, - ["first_ping_time", "last_ping_time"], + # Get RAW files requiring processing directly from the database + raw_files = get_raw_files_to_process( + db_path, + limit=new_file_num_limit, ) - if not sv_manifest_exists: - write_manifest(df_Sv, file_Sv_csv) - if not df_Sv.empty: - df_Sv.sort_values(by="first_ping_time", inplace=True, ignore_index=True) - - # Exclude raw files before exclude_before datetime - if exclude_before is None: - raw_files_in_folder = set([filename.name for filename in path_raw.glob(filename_pattern)]) - else: - raw_files_in_folder = set( - [ - filename.name - for filename in path_raw.glob(filename_pattern) - if extract_datetime_from_filename(filename.name) - >= datetime.datetime.fromisoformat(exclude_before) - ] - ) - if df_Sv.empty: - raw_files_in_df = set() - else: - raw_files_in_df = set(df_Sv["raw_filename"].tolist()) - last_raw_filename = df_Sv.iloc[-1]["raw_filename"] if not df_Sv.empty else None - if last_raw_filename: - df_Sv = df_Sv[:-1] # drop the most recent file processed - - # Find new files to process - new_files = raw_files_in_folder.difference(raw_files_in_df) - print(f"Found {len(new_files)} new files to process") - - # Reprocess last file in case it was incomplete - if last_raw_filename: - print(f"Reprocess {last_raw_filename}") - new_files.add(last_raw_filename) - - # Skip files in exclude_raw_file list - if len(exclude_raw_file) > 0: - print(f"Exclude {exclude_raw_file} from processing") - new_files.difference_update(set(exclude_raw_file)) - - # Sort new files - new_files = sorted(list(new_files)) - - # Limit number of new files to process - if new_file_num_limit != -1 and len(new_files) > new_file_num_limit: - print( - f"More than {new_file_num_limit} new files to process. " - 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])) + # Exclude files before requested datetime + if exclude_before is not None: + exclude_before_dt = datetime.datetime.fromisoformat(exclude_before) + raw_files = [ + raw_path + for raw_path in raw_files + if extract_datetime_from_filename(raw_path.name) >= exclude_before_dt + ] + + # Skip explicitly excluded RAW files + 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=path_Sv_zarr, + output_directory=str(path_Sv_zarr), encode_mode=encode_mode, waveform_mode=waveform_mode, depth_offset=depth_offset, @@ -165,74 +144,105 @@ async def cancel_run(): datagram_type=datagram_type, nmea_sentence=nmea_sentence, ) + errors = [] - results: list[RawToSvResult] = [] if parallel: - # Convert raw files to Sv in parallel print("Processing raw files in parallel") - future_all = [] - for nf in new_files: - new_processed_raw = task_raw2Sv.with_options(task_run_name=nf, name=nf, retries=3) - future = new_processed_raw.submit( - RawToSvWorkItem(raw_path=str(path_raw / nf)), + + 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, ) - future_all.append(future) - for ff in future_all: - task_result = ff.result() - results.append(task_result) + 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: - # Convert raw files to Sv sequentially print("Processing raw files sequentially") - for nf in new_files: + + for raw_path in raw_files: try: - print(f"Converting {nf}") - task_result = task_raw2Sv.with_options(task_run_name=nf, name=nf, retries=3)( - RawToSvWorkItem(raw_path=str(path_raw / nf)), + 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, ) - results.append(task_result) - except Exception as e: - errors.append(e) - print(f"Error converting {nf}: {e}") - - # Add new entries to df_Sv - if len(results) > 0: - df_new = pd.DataFrame( - [ - { - "raw_filename": result.filename_raw, - "Sv_filename": result.filename_Sv, - "first_ping_time": result.first_ping_time, - "last_ping_time": result.last_ping_time, - } - for result in results - ] - ) - # Concatenate with existing df_Sv and save - df_Sv = pd.concat([df_Sv, df_new], ignore_index=True) - df_Sv.sort_values(by=["first_ping_time"], inplace=True, ignore_index=True) - write_manifest(df_Sv, file_Sv_csv) - print(f"Added {len(new_files)} new entries to tracking CSV") + mark_raw_completed( + db_path, + raw_path, + result.filename_Sv, + result.first_ping_time, + result.last_ping_time, + ) - # Set flow to Failed state if any errors occurred - if len(errors) > 0: + except Exception as exc: + mark_raw_failed( + db_path, + raw_path, + str(exc), + ) + errors.append(exc) + print(f"Error converting {raw_path.name}: {exc}") + + # Set flow to Failed state if any conversions failed + if errors: error_msg = ( - f"{len(errors)} errors during raw to Sv conversion out of {len(new_files)} files" + 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) + flow_run_id=runtime.flow_run.id, + state=Failed(message=error_msg), ) asyncio.run(set_failed_state()) - raise Exception(error_msg) + raise RuntimeError(error_msg) @flow(log_prints=True) @@ -243,7 +253,7 @@ async def flow_create_MVBS( range_bin: str = "1m", ping_time_bin: str = "5s", path_main: str = "", - file_Sv_csv: str = "Sv_files.csv", + processing_db: str = "processing.db", file_MVBS_csv: str = "MVBS_files.csv", ): """ @@ -281,7 +291,7 @@ async def flow_create_MVBS( ) # Assemble paths - file_Sv_csv = Path(path_main) / file_Sv_csv + db_path = Path(path_main) / processing_db file_MVBS_csv = Path(path_main) / file_MVBS_csv path_Sv_zarr = Path(path_main) / "Sv" path_MVBS_zarr = Path(path_main) / "MVBS" @@ -299,19 +309,6 @@ async def flow_create_MVBS( path_MVBS_zarr = str(path_MVBS_zarr) # convert back to string to pass into task # Load Sv and MVBS info dataframes - if not file_Sv_csv.exists(): - raise ValueError("Sv info csv does not exist, check raw2Sv flow!") - df_Sv = read_manifest( - file_Sv_csv, - SV_COLUMNS_REALTIME, - ["first_ping_time", "last_ping_time"], - ) - if df_Sv.empty: - logger.info( - "Sv info csv is empty, raw2Sv flow may have just started! " - "No MVBS can be created, exiting flow." - ) - return mvbs_manifest_exists = file_MVBS_csv.exists() df_MVBS = read_manifest( @@ -336,11 +333,10 @@ async def flow_create_MVBS( logger.info(f"Slice {snum+1}: {start_time[snum]} to {end_time[snum]}") # Get Sv files in the specified time range - Sv_filenames = sorted( - df_Sv[ - (pd.to_datetime(df_Sv["last_ping_time"]) >= start_time[snum]) - & (pd.to_datetime(df_Sv["first_ping_time"]) <= end_time[snum]) - ]["Sv_filename"].tolist() + Sv_filenames = get_completed_sv_files( + db_path, + start_time=start_time[snum], + end_time=end_time[snum], ) logger.info( f"Found {len(Sv_filenames)} Sv files in the specified time range: \n" diff --git a/src/echodataflow/flows/flows_transect.py b/src/echodataflow/flows/flows_transect.py index 5df85f3d..456603ac 100644 --- a/src/echodataflow/flows/flows_transect.py +++ b/src/echodataflow/flows/flows_transect.py @@ -3,6 +3,8 @@ import pandas as pd from prefect import flow +from echodataflow.utils.processing_ledger import get_completed_sv_files + def get_changed_transects( current: pd.DataFrame, @@ -30,31 +32,18 @@ def get_changed_transects( ) -def find_overlapping_sv_files( - df_sv: pd.DataFrame, - start_time: pd.Timestamp, - end_time: pd.Timestamp, -) -> pd.DataFrame: - """Return Sv registry rows overlapping a transect time interval.""" - - return df_sv[ - (df_sv["last_ping_time"] >= start_time) - & (df_sv["first_ping_time"] <= end_time) - ].copy() - - @flow(log_prints=True) def flow_transect_update( path_transect_csv: str, path_snapshot_csv: str, path_main: str, - file_Sv_csv: str = "Sv_files.csv", + processing_db: str = "processing.db", ): """Identify updated transects and find overlapping Sv files.""" path_transect = Path(path_transect_csv) path_snapshot = Path(path_snapshot_csv) - path_sv_csv = Path(path_main) / file_Sv_csv + db_path = Path(path_main) / processing_db # Read the current transect information, preserving transect identifiers # as strings so values with leading zeros (e.g., "002") are not converted @@ -97,45 +86,22 @@ def flow_transect_update( print(f"Found {len(changed)} new or updated transect segment(s):") print(changed) - # Load Sv tracking information created by raw2Sv - if not path_sv_csv.exists(): - print(f"Sv tracking file does not exist yet: {path_sv_csv}") - return - - df_sv = pd.read_csv( - path_sv_csv, - index_col=0, - date_format="ISO8601", - parse_dates=["first_ping_time", "last_ping_time"], - ) - - if df_sv["first_ping_time"].dt.tz is None: - df_sv["first_ping_time"] = df_sv["first_ping_time"].dt.tz_localize("UTC") - - if df_sv["last_ping_time"].dt.tz is None: - df_sv["last_ping_time"] = df_sv["last_ping_time"].dt.tz_localize("UTC") - # Find Sv files overlapping each changed transect for _, transect in changed.iterrows(): start_time = pd.to_datetime(transect["transectStart"], utc=True) end_time = pd.to_datetime(transect["transectEnd"], utc=True) - overlapping_sv = find_overlapping_sv_files( - df_sv, - start_time, - end_time, + sv_filenames = get_completed_sv_files( + db_path, + start_time=start_time, + end_time=end_time, ) - sv_filenames = sorted(overlapping_sv["Sv_filename"].tolist()) - - print( - f"\nTransect {transect['transectPart']}: " - f"{start_time} to {end_time}" - ) + print(f"\nTransect {transect['transectPart']}: {start_time} to {end_time}") print(f"Found {len(sv_filenames)} overlapping Sv file(s):") for filename in sv_filenames: print(f"- {filename}") # Save snapshot only after processing the current CSV - current.to_csv(path_snapshot, index=False) \ No newline at end of file + current.to_csv(path_snapshot, index=False) diff --git a/src/echodataflow/utils/processing_ledger.py b/src/echodataflow/utils/processing_ledger.py new file mode 100644 index 00000000..dfaaf61e --- /dev/null +++ b/src/echodataflow/utils/processing_ledger.py @@ -0,0 +1,226 @@ +from __future__ import annotations + +import sqlite3 +from pathlib import Path + + +def initialize_ledger(db_path: str | Path) -> None: + """Create the processing ledger database and required tables.""" + db_path = Path(db_path) + db_path.parent.mkdir(parents=True, exist_ok=True) + + with sqlite3.connect(db_path, timeout=30) as conn: + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA busy_timeout=30000") + + conn.execute( + """ + CREATE TABLE IF NOT EXISTS raw_sv ( + raw_path TEXT PRIMARY KEY, + raw_filename TEXT NOT NULL, + file_size INTEGER, + file_mtime_ns INTEGER, + status TEXT NOT NULL DEFAULT 'pending', + sv_filename TEXT, + first_ping_time TEXT, + last_ping_time TEXT, + error TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP + ) + """ + ) + + conn.execute( + """ + CREATE INDEX IF NOT EXISTS idx_raw_sv_status + ON raw_sv(status) + """ + ) + + conn.execute( + """ + CREATE INDEX IF NOT EXISTS idx_raw_sv_first_ping_time + ON raw_sv(first_ping_time) + """ + ) + + +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.""" + raw_path = Path(raw_path) + stat = raw_path.stat() + + with sqlite3.connect(db_path, timeout=30) as conn: + conn.execute("PRAGMA busy_timeout=30000") + + existing = conn.execute( + """ + SELECT file_size, file_mtime_ns + FROM raw_sv + WHERE raw_path = ? + """, + (str(raw_path),), + ).fetchone() + + if existing is None: + conn.execute( + """ + INSERT INTO raw_sv ( + raw_path, + raw_filename, + file_size, + file_mtime_ns, + status + ) + VALUES (?, ?, ?, ?, 'pending') + """, + ( + str(raw_path), + raw_path.name, + stat.st_size, + stat.st_mtime_ns, + ), + ) + return + + if existing != (stat.st_size, stat.st_mtime_ns): + conn.execute( + """ + UPDATE raw_sv + SET file_size = ?, + file_mtime_ns = ?, + status = 'pending', + sv_filename = NULL, + first_ping_time = NULL, + last_ping_time = NULL, + error = '', + updated_at = CURRENT_TIMESTAMP + WHERE raw_path = ? + """, + ( + stat.st_size, + stat.st_mtime_ns, + str(raw_path), + ), + ) + +def get_raw_files_to_process( + db_path: str | Path, + limit: int = -1, +) -> list[Path]: + """Return RAW files that are pending or failed.""" + query = """ + SELECT raw_path + FROM raw_sv + WHERE status IN ('pending', 'failed') + ORDER BY created_at, raw_path + """ + + params: tuple = () + + if limit != -1: + query += " LIMIT ?" + params = (limit,) + + with sqlite3.connect(db_path) as conn: + rows = conn.execute(query, params).fetchall() + + return [Path(row[0]) for row in rows] + +def mark_raw_processing(db_path: str | Path, raw_path: str | Path) -> None: + """Mark a RAW file as currently being processed.""" + with sqlite3.connect(db_path) as conn: + conn.execute( + """ + UPDATE raw_sv + SET status = 'processing', + error = '', + updated_at = CURRENT_TIMESTAMP + WHERE raw_path = ? + """, + (str(Path(raw_path)),), + ) + + +def mark_raw_completed( + db_path: str | Path, + raw_path: str | Path, + sv_filename: str, + first_ping_time, + last_ping_time, +) -> None: + """Mark a RAW file as successfully converted to Sv.""" + with sqlite3.connect(db_path) as conn: + conn.execute( + """ + UPDATE raw_sv + SET status = 'completed', + sv_filename = ?, + first_ping_time = ?, + last_ping_time = ?, + error = '', + updated_at = CURRENT_TIMESTAMP + WHERE raw_path = ? + """, + ( + sv_filename, + str(first_ping_time), + str(last_ping_time), + str(Path(raw_path)), + ), + ) + + +def mark_raw_failed( + db_path: str | Path, + raw_path: str | Path, + error: str, +) -> None: + """Mark a RAW file as failed.""" + with sqlite3.connect(db_path) as conn: + conn.execute( + """ + UPDATE raw_sv + SET status = 'failed', + error = ?, + updated_at = CURRENT_TIMESTAMP + WHERE raw_path = ? + """, + (error, str(Path(raw_path))), + ) + +def get_completed_sv_files( + db_path: str | Path, + start_time=None, + end_time=None, +) -> list[str]: + """Return completed Sv files, optionally overlapping a time range.""" + query = """ + SELECT sv_filename + FROM raw_sv + WHERE status = 'completed' + AND sv_filename IS NOT NULL + """ + params = [] + + if start_time is not None: + query += """ + AND datetime(last_ping_time) >= datetime(?) + """ + params.append(str(start_time)) + + if end_time is not None: + query += """ + AND datetime(first_ping_time) <= datetime(?) + """ + params.append(str(end_time)) + + query += """ + ORDER BY datetime(first_ping_time) + """ + + with sqlite3.connect(db_path) as conn: + rows = conn.execute(query, params).fetchall() + + return [row[0] for row in rows] \ No newline at end of file diff --git a/src/echodataflow/utils/raw_monitor.py b/src/echodataflow/utils/raw_monitor.py index 82488d28..80ea6215 100644 --- a/src/echodataflow/utils/raw_monitor.py +++ b/src/echodataflow/utils/raw_monitor.py @@ -3,6 +3,10 @@ from prefect.events import emit_event from echodataflow.utils.file_watcher import watch_directory +from echodataflow.utils.processing_ledger import ( + initialize_ledger, + register_raw_file, +) RAW_UPDATE_EVENT = "echodataflow.raw.updated" @@ -11,7 +15,6 @@ def emit_raw_update_event(path: Path) -> None: """Emit a Prefect event when a RAW file arrives.""" - emit_event( event=RAW_UPDATE_EVENT, resource={ @@ -22,11 +25,21 @@ def emit_raw_update_event(path: Path) -> None: ) -def watch_raw_directory(path: str | Path): +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 watch_raw_directory( + path: str | Path, + db_path: str | Path, +): """Watch a directory for new RAW files.""" + initialize_ledger(db_path) return watch_directory( directory=path, - callback=emit_raw_update_event, + callback=lambda raw_path: register_and_emit_raw_update(raw_path, db_path), pattern="*.raw", ) \ No newline at end of file diff --git a/tests/test_flow_raw2sv.py b/tests/test_flow_raw2sv.py new file mode 100644 index 00000000..34806ad5 --- /dev/null +++ b/tests/test_flow_raw2sv.py @@ -0,0 +1,170 @@ +import pytest +from types import SimpleNamespace + +from echodataflow.flows import flows_acoustics + + +async def _false_async(): + return False + + +def test_flow_raw2sv_uses_processing_ledger(monkeypatch, tmp_path): + raw_file = tmp_path / "example.raw" + + calls = { + "processing": [], + "completed": [], + } + + monkeypatch.setattr( + flows_acoustics, + "deployment_already_running", + lambda: _false_async(), + ) + + monkeypatch.setattr( + flows_acoustics, + "initialize_ledger", + lambda db_path: None, + ) + + monkeypatch.setattr( + flows_acoustics, + "get_raw_files_to_process", + lambda db_path, limit: [raw_file], + ) + + monkeypatch.setattr( + flows_acoustics, + "mark_raw_processing", + lambda db_path, path: calls["processing"].append(path), + ) + + monkeypatch.setattr( + flows_acoustics, + "mark_raw_completed", + lambda db_path, path, sv_filename, first_ping_time, last_ping_time: + calls["completed"].append( + ( + path, + sv_filename, + first_ping_time, + last_ping_time, + ) + ), + ) + + class FakeTask: + def with_options(self, **kwargs): + return self + + def __call__(self, work_item, settings): + return SimpleNamespace( + filename_Sv="example_Sv.zarr", + first_ping_time="2026-08-14T12:00:00Z", + last_ping_time="2026-08-14T12:05:00Z", + ) + + monkeypatch.setattr( + flows_acoustics, + "task_raw2Sv", + FakeTask(), + ) + + flows_acoustics.flow_raw2Sv.fn( + path_main=str(tmp_path), + new_file_num_limit=1, + ) + + assert calls["processing"] == [raw_file] + assert calls["completed"] == [ + ( + raw_file, + "example_Sv.zarr", + "2026-08-14T12:00:00Z", + "2026-08-14T12:05:00Z", + ) + ] + + +def test_flow_raw2sv_marks_failed(monkeypatch, tmp_path): + raw_file = tmp_path / "example.raw" + + calls = { + "failed": [], + } + + monkeypatch.setattr( + flows_acoustics, + "deployment_already_running", + lambda: _false_async(), + ) + + monkeypatch.setattr( + flows_acoustics, + "initialize_ledger", + lambda db_path: None, + ) + + monkeypatch.setattr( + flows_acoustics, + "get_raw_files_to_process", + lambda db_path, limit: [raw_file], + ) + + monkeypatch.setattr( + flows_acoustics, + "mark_raw_processing", + lambda db_path, path: None, + ) + + monkeypatch.setattr( + flows_acoustics, + "mark_raw_failed", + lambda db_path, path, error: calls["failed"].append((path, error)), + ) + + class FakeTask: + def with_options(self, **kwargs): + return self + + def __call__(self, work_item, settings): + raise RuntimeError("conversion failed") + + monkeypatch.setattr( + flows_acoustics, + "task_raw2Sv", + FakeTask(), + ) + + class FakeClient: + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + pass + + async def set_flow_run_state(self, **kwargs): + pass + + monkeypatch.setattr( + flows_acoustics, + "get_client", + lambda: FakeClient(), + ) + + monkeypatch.setattr( + flows_acoustics.runtime.flow_run, + "id", + "00000000-0000-0000-0000-000000000001", + ) + + with pytest.raises(RuntimeError, match="1 errors during raw to Sv conversion"): + flows_acoustics.flow_raw2Sv.fn( + path_main=str(tmp_path), + new_file_num_limit=1, + ) + + assert calls["failed"] == [ + (raw_file, "conversion failed"), + ] \ No newline at end of file diff --git a/tests/test_flow_transect.py b/tests/test_flow_transect.py index fa65cfaf..d7a23921 100644 --- a/tests/test_flow_transect.py +++ b/tests/test_flow_transect.py @@ -1,10 +1,14 @@ import pandas as pd from echodataflow.flows.flows_transect import ( - find_overlapping_sv_files, flow_transect_update, get_changed_transects, ) +from echodataflow.utils.processing_ledger import ( + initialize_ledger, + mark_raw_completed, + register_raw_file, +) def test_flow_transect_update_initializes_snapshot(tmp_path): @@ -75,25 +79,44 @@ def test_flow_transect_update_finds_overlapping_sv(tmp_path, capsys): previous.to_csv(snapshot_csv, index=False) current.to_csv(transect_csv, index=False) - pd.DataFrame( - { - "Sv_filename": [ - "before_Sv.zarr", - "overlap_Sv.zarr", - "after_Sv.zarr", - ], - "first_ping_time": [ - "2024-07-07T00:00:00Z", - "2024-07-07T00:15:00Z", - "2024-07-07T00:40:00Z", - ], - "last_ping_time": [ - "2024-07-07T00:05:00Z", - "2024-07-07T00:25:00Z", - "2024-07-07T00:50:00Z", - ], - } - ).to_csv(path_main / "Sv_files.csv") + db_path = path_main / "processing.db" + initialize_ledger(db_path) + + raw_before = tmp_path / "before.raw" + raw_overlap = tmp_path / "overlap.raw" + raw_after = tmp_path / "after.raw" + + raw_before.touch() + raw_overlap.touch() + raw_after.touch() + + register_raw_file(db_path, raw_before) + register_raw_file(db_path, raw_overlap) + register_raw_file(db_path, raw_after) + + mark_raw_completed( + db_path, + raw_before, + "before_Sv.zarr", + "2024-07-07T00:00:00Z", + "2024-07-07T00:05:00Z", + ) + + mark_raw_completed( + db_path, + raw_overlap, + "overlap_Sv.zarr", + "2024-07-07T00:15:00Z", + "2024-07-07T00:25:00Z", + ) + + mark_raw_completed( + db_path, + raw_after, + "after_Sv.zarr", + "2024-07-07T00:40:00Z", + "2024-07-07T00:50:00Z", + ) flow_transect_update.fn( path_transect_csv=str(transect_csv), @@ -107,6 +130,8 @@ def test_flow_transect_update_finds_overlapping_sv(tmp_path, capsys): assert "Transect 002" in output assert "Found 1 overlapping Sv file(s)" in output assert "overlap_Sv.zarr" in output + assert "before_Sv.zarr" not in output + assert "after_Sv.zarr" not in output def test_get_changed_transects(): @@ -140,41 +165,6 @@ def test_get_changed_transects(): assert changed.iloc[0]["transectPart"] == "002" -def test_find_overlapping_sv_files(): - df_sv = pd.DataFrame( - { - "Sv_filename": [ - "before_Sv.zarr", - "overlap_Sv.zarr", - "after_Sv.zarr", - ], - "first_ping_time": pd.to_datetime( - [ - "2024-07-07T00:00:00Z", - "2024-07-07T00:15:00Z", - "2024-07-07T00:40:00Z", - ], - utc=True, - ), - "last_ping_time": pd.to_datetime( - [ - "2024-07-07T00:05:00Z", - "2024-07-07T00:25:00Z", - "2024-07-07T00:50:00Z", - ], - utc=True, - ), - } - ) - - overlapping = find_overlapping_sv_files( - df_sv, - pd.Timestamp("2024-07-07T00:20:00Z"), - pd.Timestamp("2024-07-07T00:30:00Z"), - ) - - assert overlapping["Sv_filename"].tolist() == ["overlap_Sv.zarr"] - def test_flow_transect_update_ignores_open_transect(tmp_path, capsys): transect_csv = tmp_path / "transects.csv" snapshot_csv = tmp_path / "snapshot.csv" diff --git a/tests/test_processing_ledger.py b/tests/test_processing_ledger.py new file mode 100644 index 00000000..f6629bba --- /dev/null +++ b/tests/test_processing_ledger.py @@ -0,0 +1,233 @@ +import sqlite3 + +from echodataflow.utils.processing_ledger import ( + get_completed_sv_files, + get_raw_files_to_process, + initialize_ledger, + mark_raw_completed, + mark_raw_failed, + mark_raw_processing, + register_raw_file, +) + + +def test_initialize_ledger(tmp_path): + db_path = tmp_path / "processing.db" + + initialize_ledger(db_path) + + assert db_path.exists() + + with sqlite3.connect(db_path) as conn: + columns = { + row[1] + for row in conn.execute("PRAGMA table_info(raw_sv)").fetchall() + } + + assert columns == { + "raw_path", + "raw_filename", + "file_size", + "file_mtime_ns", + "status", + "sv_filename", + "first_ping_time", + "last_ping_time", + "error", + "created_at", + "updated_at", + } + + +def test_register_raw_file(tmp_path): + db_path = tmp_path / "processing.db" + raw_path = tmp_path / "test.raw" + + raw_path.touch() + + initialize_ledger(db_path) + register_raw_file(db_path, raw_path) + register_raw_file(db_path, raw_path) + + 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"), + ] + + +def test_get_raw_files_to_process(tmp_path): + db_path = tmp_path / "processing.db" + + raw_a = tmp_path / "a.raw" + raw_b = tmp_path / "b.raw" + + raw_a.touch() + raw_b.touch() + + initialize_ledger(db_path) + register_raw_file(db_path, raw_a) + register_raw_file(db_path, raw_b) + + result = get_raw_files_to_process(db_path) + + assert result == [raw_a, raw_b] + + +def test_get_raw_files_to_process_limit(tmp_path): + db_path = tmp_path / "processing.db" + + raw_a = tmp_path / "a.raw" + raw_b = tmp_path / "b.raw" + + raw_a.touch() + raw_b.touch() + + initialize_ledger(db_path) + register_raw_file(db_path, raw_a) + register_raw_file(db_path, raw_b) + + result = get_raw_files_to_process(db_path, limit=1) + + assert result == [raw_a] + + +def test_raw_processing_state_transitions(tmp_path): + db_path = tmp_path / "processing.db" + raw_path = tmp_path / "test.raw" + + raw_path.touch() + + initialize_ledger(db_path) + register_raw_file(db_path, raw_path) + + mark_raw_processing(db_path, raw_path) + + with sqlite3.connect(db_path) as conn: + status = conn.execute( + "SELECT status FROM raw_sv WHERE raw_path = ?", + (str(raw_path),), + ).fetchone()[0] + + assert status == "processing" + + mark_raw_completed( + db_path, + raw_path, + "test_Sv.zarr", + "2026-08-14T12:00:00Z", + "2026-08-14T12:05:00Z", + ) + + with sqlite3.connect(db_path) as conn: + row = conn.execute( + """ + SELECT status, sv_filename, first_ping_time, last_ping_time, error + FROM raw_sv + WHERE raw_path = ? + """, + (str(raw_path),), + ).fetchone() + + assert row == ( + "completed", + "test_Sv.zarr", + "2026-08-14T12:00:00Z", + "2026-08-14T12:05:00Z", + "", + ) + + mark_raw_failed(db_path, raw_path, "boom") + + with sqlite3.connect(db_path) as conn: + row = conn.execute( + """ + SELECT status, error + FROM raw_sv + WHERE raw_path = ? + """, + (str(raw_path),), + ).fetchone() + + assert row == ("failed", "boom") + + +def test_get_completed_sv_files_by_time_range(tmp_path): + db_path = tmp_path / "processing.db" + + raw_a = tmp_path / "a.raw" + raw_b = tmp_path / "b.raw" + + raw_a.touch() + raw_b.touch() + + initialize_ledger(db_path) + + register_raw_file(db_path, raw_a) + register_raw_file(db_path, raw_b) + + mark_raw_completed( + db_path, + raw_a, + "a_Sv.zarr", + "2026-08-14T12:00:00", + "2026-08-14T12:05:00", + ) + + mark_raw_completed( + db_path, + raw_b, + "b_Sv.zarr", + "2026-08-14T12:05:00", + "2026-08-14T12:10:00", + ) + + result = get_completed_sv_files( + db_path, + start_time="2026-08-14T12:04:00", + end_time="2026-08-14T12:06:00", + ) + + assert result == [ + "a_Sv.zarr", + "b_Sv.zarr", + ] + + +def test_register_raw_file_requeues_changed_file(tmp_path): + db_path = tmp_path / "processing.db" + raw_path = tmp_path / "test.raw" + + raw_path.write_bytes(b"first") + + initialize_ledger(db_path) + register_raw_file(db_path, raw_path) + + mark_raw_completed( + db_path, + raw_path, + "test_Sv.zarr", + "2026-08-14T12:00:00Z", + "2026-08-14T12:05:00Z", + ) + + raw_path.write_bytes(b"first plus more data") + register_raw_file(db_path, raw_path) + + with sqlite3.connect(db_path) as conn: + row = conn.execute( + """ + SELECT status, sv_filename, first_ping_time, last_ping_time + FROM raw_sv + WHERE raw_path = ? + """, + (str(raw_path),), + ).fetchone() + + assert row == ("pending", None, None, None) \ No newline at end of file diff --git a/tests/test_raw_monitor.py b/tests/test_raw_monitor.py index b90fc98b..3e3ededd 100644 --- a/tests/test_raw_monitor.py +++ b/tests/test_raw_monitor.py @@ -28,6 +28,10 @@ def fake_emit_event(**kwargs): def test_watch_raw_directory(monkeypatch, tmp_path): called = {} + db_path = tmp_path / "processing.db" + + def fake_initialize_ledger(path): + called["db_path"] = path def fake_watch_directory( directory, @@ -39,15 +43,57 @@ def fake_watch_directory( called["pattern"] = pattern return "observer" + monkeypatch.setattr( + raw_monitor, + "initialize_ledger", + fake_initialize_ledger, + ) + monkeypatch.setattr( raw_monitor, "watch_directory", fake_watch_directory, ) - result = raw_monitor.watch_raw_directory(tmp_path) + result = raw_monitor.watch_raw_directory( + tmp_path, + db_path, + ) assert result == "observer" assert called["directory"] == tmp_path - assert called["callback"] is raw_monitor.emit_raw_update_event - assert called["pattern"] == "*.raw" \ No newline at end of file + assert called["db_path"] == db_path + assert called["pattern"] == "*.raw" + + +def test_register_and_emit_raw_update(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) + + def fake_emit_raw_update_event(path): + called["emitted"] = path + + monkeypatch.setattr( + raw_monitor, + "register_raw_file", + fake_register_raw_file, + ) + + monkeypatch.setattr( + raw_monitor, + "emit_raw_update_event", + fake_emit_raw_update_event, + ) + + raw_monitor.register_and_emit_raw_update( + raw_file, + db_path, + ) + + assert called["registered"] == (db_path, raw_file) + assert called["emitted"] == raw_file From 147d409f4a7a52419a45f3bdea7ac48097865cb9 Mon Sep 17 00:00:00 2001 From: Lloyd Izard <76954858+LOCEANlloydizard@users.noreply.github.com> Date: Sun, 16 Aug 2026 14:54:32 -0700 Subject: [PATCH 2/7] Improve realtime file watcher reconciliation --- src/echodataflow/utils/file_watcher.py | 25 ++++++++++++---- src/echodataflow/utils/raw_monitor.py | 15 +++++++++- tests/test_file_watcher.py | 21 ++++++++++++- tests/test_raw_monitor.py | 41 ++++++++++++++++++++++++++ 4 files changed, 95 insertions(+), 7 deletions(-) diff --git a/src/echodataflow/utils/file_watcher.py b/src/echodataflow/utils/file_watcher.py index cc7f3add..54d3444d 100644 --- a/src/echodataflow/utils/file_watcher.py +++ b/src/echodataflow/utils/file_watcher.py @@ -1,3 +1,4 @@ +import logging from collections.abc import Callable from pathlib import Path @@ -5,6 +6,7 @@ from watchdog.observers import Observer +logger = logging.getLogger(__name__) class FileUpdateHandler(FileSystemEventHandler): """Run a callback when the target file is updated.""" @@ -46,14 +48,23 @@ def __init__( self.callback = callback self.pattern = pattern - def _handle(self, event: FileSystemEvent) -> None: - if event.is_directory: - return + def _handle_path(self, path: str | Path) -> None: + event_path = Path(path).resolve() - event_path = Path(event.src_path).resolve() + if not event_path.match(self.pattern): + return - if event_path.match(self.pattern): + try: self.callback(event_path) + except Exception: + logger.exception( + "Error handling filesystem update for %s", + event_path, + ) + + def _handle(self, event: FileSystemEvent) -> None: + if not event.is_directory: + self._handle_path(event.src_path) def on_created(self, event: FileSystemEvent) -> None: self._handle(event) @@ -61,6 +72,10 @@ def on_created(self, event: FileSystemEvent) -> None: def on_modified(self, event: FileSystemEvent) -> None: self._handle(event) + def on_moved(self, event: FileSystemEvent) -> None: + if not event.is_directory: + self._handle_path(event.dest_path) + def watch_file( target_file: str | Path, diff --git a/src/echodataflow/utils/raw_monitor.py b/src/echodataflow/utils/raw_monitor.py index 80ea6215..4365e01a 100644 --- a/src/echodataflow/utils/raw_monitor.py +++ b/src/echodataflow/utils/raw_monitor.py @@ -36,10 +36,23 @@ def watch_raw_directory( db_path: str | Path, ): """Watch a directory for new RAW files.""" + + 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: + 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( - directory=path, + directory=raw_directory, callback=lambda raw_path: register_and_emit_raw_update(raw_path, db_path), pattern="*.raw", ) \ No newline at end of file diff --git a/tests/test_file_watcher.py b/tests/test_file_watcher.py index e25df720..e3aab567 100644 --- a/tests/test_file_watcher.py +++ b/tests/test_file_watcher.py @@ -111,4 +111,23 @@ def test_file_created_handler_ignores_nonmatching_file(tmp_path): handler.on_created(FakeEvent(other_file)) - assert detected == [] \ No newline at end of file + assert detected == [] + +def test_file_created_handler_calls_callback_for_moved_matching_file(tmp_path): + temporary = tmp_path / "temporary.tmp" + raw_file = tmp_path / "example.raw" + detected = [] + + handler = FileCreatedHandler( + callback=detected.append, + pattern="*.raw", + ) + + handler.on_moved( + FakeEvent( + src_path=temporary, + dest_path=raw_file, + ) + ) + + assert detected == [raw_file.resolve()] \ No newline at end of file diff --git a/tests/test_raw_monitor.py b/tests/test_raw_monitor.py index 3e3ededd..9baa1f76 100644 --- a/tests/test_raw_monitor.py +++ b/tests/test_raw_monitor.py @@ -97,3 +97,44 @@ def fake_emit_raw_update_event(path): assert called["registered"] == (db_path, raw_file) assert called["emitted"] == raw_file + +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( + raw_monitor, + "register_raw_file", + lambda db, path: registered.append((db, path)), + ) + + monkeypatch.setattr( + raw_monitor, + "emit_raw_update_event", + emitted.append, + ) + + monkeypatch.setattr( + raw_monitor, + "watch_directory", + lambda **kwargs: "observer", + ) + + result = raw_monitor.watch_raw_directory( + tmp_path, + db_path, + ) + + assert result == "observer" + assert {path for _, path in registered} == { + raw_a.resolve(), + raw_b.resolve(), + } + assert emitted == [tmp_path.resolve()] From aaca653cbf419ab401a379d7e9326ca922fb64d3 Mon Sep 17 00:00:00 2001 From: Lloyd Izard <76954858+LOCEANlloydizard@users.noreply.github.com> Date: Wed, 19 Aug 2026 19:11:25 -0700 Subject: [PATCH 3/7] dd SQLAlchemy and postgreSQL support --- pyproject.toml | 4 +- src/echodataflow/flows/flows_acoustics.py | 5 +- src/echodataflow/flows/flows_transect.py | 7 +- src/echodataflow/utils/processing_ledger.py | 355 ++++++++++++-------- tests/test_processing_ledger.py | 40 ++- 5 files changed, 257 insertions(+), 154 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 83158dcc..465db797 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -48,7 +48,9 @@ dependencies = [ "PyYAML", "s3fs", "shapely", - "watchdog", + "sqlalchemy>=2.0", + "psycopg[binary]>=3.2", + "watchdog", "xarray", ] dynamic = ["version"] diff --git a/src/echodataflow/flows/flows_acoustics.py b/src/echodataflow/flows/flows_acoustics.py index eb0f0cf1..2659db39 100644 --- a/src/echodataflow/flows/flows_acoustics.py +++ b/src/echodataflow/flows/flows_acoustics.py @@ -55,6 +55,7 @@ mark_raw_completed, mark_raw_failed, mark_raw_processing, + resolve_database, ) # Turn on verbose logging for echopype @@ -95,7 +96,7 @@ async def cancel_run(): # Assemble paths path_Sv_zarr = Path(path_main) / "Sv" - db_path = Path(path_main) / processing_db + db_path = resolve_database(path_main, processing_db) initialize_ledger(db_path) @@ -291,7 +292,7 @@ async def flow_create_MVBS( ) # Assemble paths - db_path = Path(path_main) / processing_db + db_path = resolve_database(path_main, processing_db) file_MVBS_csv = Path(path_main) / file_MVBS_csv path_Sv_zarr = Path(path_main) / "Sv" path_MVBS_zarr = Path(path_main) / "MVBS" diff --git a/src/echodataflow/flows/flows_transect.py b/src/echodataflow/flows/flows_transect.py index 456603ac..e2b7d7a1 100644 --- a/src/echodataflow/flows/flows_transect.py +++ b/src/echodataflow/flows/flows_transect.py @@ -3,7 +3,10 @@ import pandas as pd from prefect import flow -from echodataflow.utils.processing_ledger import get_completed_sv_files +from echodataflow.utils.processing_ledger import ( + get_completed_sv_files, + resolve_database, +) def get_changed_transects( @@ -43,7 +46,7 @@ def flow_transect_update( path_transect = Path(path_transect_csv) path_snapshot = Path(path_snapshot_csv) - db_path = Path(path_main) / processing_db + db_path = resolve_database(path_main, processing_db) # Read the current transect information, preserving transect identifiers # as strings so values with leading zeros (e.g., "002") are not converted diff --git a/src/echodataflow/utils/processing_ledger.py b/src/echodataflow/utils/processing_ledger.py index dfaaf61e..01be1134 100644 --- a/src/echodataflow/utils/processing_ledger.py +++ b/src/echodataflow/utils/processing_ledger.py @@ -1,145 +1,208 @@ from __future__ import annotations -import sqlite3 +from functools import lru_cache from pathlib import Path +from sqlalchemy import ( + BigInteger, + Column, + Index, + MetaData, + Table, + Text, + create_engine, + insert, + select, + update, +) +from sqlalchemy.engine import Engine +from sqlalchemy.sql import func + + +metadata = MetaData() + +raw_sv = Table( + "raw_sv", + metadata, + Column("raw_path", Text, primary_key=True), + Column("raw_filename", Text, nullable=False), + Column("file_size", BigInteger), + Column("file_mtime_ns", BigInteger), + Column("status", Text, nullable=False, server_default="pending"), + Column("sv_filename", Text), + Column("first_ping_time", Text), + Column("last_ping_time", Text), + Column("error", Text, nullable=False, server_default=""), + Column( + "created_at", + Text, + nullable=False, + server_default=func.current_timestamp(), + ), + Column( + "updated_at", + Text, + nullable=False, + server_default=func.current_timestamp(), + ), +) + +Index("idx_raw_sv_status", raw_sv.c.status) +Index("idx_raw_sv_first_ping_time", raw_sv.c.first_ping_time) + +def _timestamp_string(value) -> str: + """Return timestamps in a consistent ISO-8601 representation.""" + if hasattr(value, "isoformat"): + return value.isoformat() + return str(value) + +def _database_url(db_path: str | Path) -> str: + """Convert a local database path to a SQLAlchemy URL.""" + + value = str(db_path) + + # Already a SQLAlchemy database URL, e.g. PostgreSQL. + if "://" in value: + return value + + path = Path(value).resolve() + path.parent.mkdir(parents=True, exist_ok=True) + + return f"sqlite:///{path.as_posix()}" + + +@lru_cache +def _get_engine(database: str) -> Engine: + """Create and cache a SQLAlchemy engine.""" + + url = _database_url(database) + + kwargs = {} + + if url.startswith("sqlite"): + kwargs["connect_args"] = {"timeout": 30} + + return create_engine(url, **kwargs) + + +def _engine(db_path: str | Path) -> Engine: + return _get_engine(str(db_path)) + +def resolve_database( + path_main: str | Path, + processing_db: str, +) -> str | Path: + """Resolve a local database filename or preserve a database URL.""" + if "://" in processing_db: + return processing_db + + return Path(path_main) / processing_db def initialize_ledger(db_path: str | Path) -> None: """Create the processing ledger database and required tables.""" - db_path = Path(db_path) - db_path.parent.mkdir(parents=True, exist_ok=True) - with sqlite3.connect(db_path, timeout=30) as conn: - conn.execute("PRAGMA journal_mode=WAL") - conn.execute("PRAGMA busy_timeout=30000") + engine = _engine(db_path) - conn.execute( - """ - CREATE TABLE IF NOT EXISTS raw_sv ( - raw_path TEXT PRIMARY KEY, - raw_filename TEXT NOT NULL, - file_size INTEGER, - file_mtime_ns INTEGER, - status TEXT NOT NULL DEFAULT 'pending', - sv_filename TEXT, - first_ping_time TEXT, - last_ping_time TEXT, - error TEXT NOT NULL DEFAULT '', - created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, - updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP - ) - """ - ) + # Keep the existing SQLite concurrency settings. + if engine.dialect.name == "sqlite": + with engine.connect() as conn: + conn.exec_driver_sql("PRAGMA journal_mode=WAL") + conn.exec_driver_sql("PRAGMA busy_timeout=30000") - conn.execute( - """ - CREATE INDEX IF NOT EXISTS idx_raw_sv_status - ON raw_sv(status) - """ - ) + metadata.create_all(engine) - conn.execute( - """ - CREATE INDEX IF NOT EXISTS idx_raw_sv_first_ping_time - ON raw_sv(first_ping_time) - """ - ) - -def register_raw_file(db_path: str | Path, raw_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.""" + raw_path = Path(raw_path) stat = raw_path.stat() - with sqlite3.connect(db_path, timeout=30) as conn: - conn.execute("PRAGMA busy_timeout=30000") + engine = _engine(db_path) + with engine.begin() as conn: existing = conn.execute( - """ - SELECT file_size, file_mtime_ns - FROM raw_sv - WHERE raw_path = ? - """, - (str(raw_path),), - ).fetchone() + select( + raw_sv.c.file_size, + raw_sv.c.file_mtime_ns, + ).where(raw_sv.c.raw_path == str(raw_path)) + ).first() if existing is None: conn.execute( - """ - INSERT INTO raw_sv ( - raw_path, - raw_filename, - file_size, - file_mtime_ns, - status + 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 (?, ?, ?, ?, 'pending') - """, - ( - str(raw_path), - raw_path.name, - stat.st_size, - stat.st_mtime_ns, - ), ) return - if existing != (stat.st_size, stat.st_mtime_ns): + if ( + existing.file_size != stat.st_size + or existing.file_mtime_ns != stat.st_mtime_ns + ): conn.execute( - """ - UPDATE raw_sv - SET file_size = ?, - file_mtime_ns = ?, - status = 'pending', - sv_filename = NULL, - first_ping_time = NULL, - last_ping_time = NULL, - error = '', - updated_at = CURRENT_TIMESTAMP - WHERE raw_path = ? - """, - ( - stat.st_size, - stat.st_mtime_ns, - str(raw_path), - ), + update(raw_sv) + .where(raw_sv.c.raw_path == str(raw_path)) + .values( + file_size=stat.st_size, + file_mtime_ns=stat.st_mtime_ns, + status="pending", + sv_filename=None, + first_ping_time=None, + last_ping_time=None, + error="", + updated_at=func.current_timestamp(), + ) ) + def get_raw_files_to_process( db_path: str | Path, limit: int = -1, ) -> list[Path]: """Return RAW files that are pending or failed.""" - query = """ - SELECT raw_path - FROM raw_sv - WHERE status IN ('pending', 'failed') - ORDER BY created_at, raw_path - """ - params: tuple = () + stmt = ( + select(raw_sv.c.raw_path) + .where(raw_sv.c.status.in_(("pending", "failed"))) + .order_by(raw_sv.c.created_at, raw_sv.c.raw_path) + ) if limit != -1: - query += " LIMIT ?" - params = (limit,) + stmt = stmt.limit(limit) - with sqlite3.connect(db_path) as conn: - rows = conn.execute(query, params).fetchall() + engine = _engine(db_path) - return [Path(row[0]) for row in rows] + with engine.connect() as conn: + rows = conn.execute(stmt).all() -def mark_raw_processing(db_path: str | Path, raw_path: str | Path) -> None: + return [Path(row.raw_path) for row in rows] + + +def mark_raw_processing( + db_path: str | Path, + raw_path: str | Path, +) -> None: """Mark a RAW file as currently being processed.""" - with sqlite3.connect(db_path) as conn: + + engine = _engine(db_path) + + with engine.begin() as conn: conn.execute( - """ - UPDATE raw_sv - SET status = 'processing', - error = '', - updated_at = CURRENT_TIMESTAMP - WHERE raw_path = ? - """, - (str(Path(raw_path)),), + update(raw_sv) + .where(raw_sv.c.raw_path == str(Path(raw_path))) + .values( + status="processing", + error="", + updated_at=func.current_timestamp(), + ) ) @@ -151,24 +214,21 @@ def mark_raw_completed( last_ping_time, ) -> None: """Mark a RAW file as successfully converted to Sv.""" - with sqlite3.connect(db_path) as conn: + + engine = _engine(db_path) + + with engine.begin() as conn: conn.execute( - """ - UPDATE raw_sv - SET status = 'completed', - sv_filename = ?, - first_ping_time = ?, - last_ping_time = ?, - error = '', - updated_at = CURRENT_TIMESTAMP - WHERE raw_path = ? - """, - ( - sv_filename, - str(first_ping_time), - str(last_ping_time), - str(Path(raw_path)), - ), + update(raw_sv) + .where(raw_sv.c.raw_path == str(Path(raw_path))) + .values( + status="completed", + sv_filename=sv_filename, + first_ping_time=_timestamp_string(first_ping_time), + last_ping_time=_timestamp_string(last_ping_time), + error="", + updated_at=func.current_timestamp(), + ) ) @@ -178,49 +238,48 @@ def mark_raw_failed( error: str, ) -> None: """Mark a RAW file as failed.""" - with sqlite3.connect(db_path) as conn: + + engine = _engine(db_path) + + with engine.begin() as conn: conn.execute( - """ - UPDATE raw_sv - SET status = 'failed', - error = ?, - updated_at = CURRENT_TIMESTAMP - WHERE raw_path = ? - """, - (error, str(Path(raw_path))), + update(raw_sv) + .where(raw_sv.c.raw_path == str(Path(raw_path))) + .values( + status="failed", + error=error, + updated_at=func.current_timestamp(), + ) ) + def get_completed_sv_files( db_path: str | Path, start_time=None, end_time=None, ) -> list[str]: """Return completed Sv files, optionally overlapping a time range.""" - query = """ - SELECT sv_filename - FROM raw_sv - WHERE status = 'completed' - AND sv_filename IS NOT NULL - """ - params = [] + + stmt = select(raw_sv.c.sv_filename).where( + raw_sv.c.status == "completed", + raw_sv.c.sv_filename.is_not(None), + ) if start_time is not None: - query += """ - AND datetime(last_ping_time) >= datetime(?) - """ - params.append(str(start_time)) + stmt = stmt.where( + raw_sv.c.last_ping_time >= _timestamp_string(start_time) + ) if end_time is not None: - query += """ - AND datetime(first_ping_time) <= datetime(?) - """ - params.append(str(end_time)) + stmt = stmt.where( + raw_sv.c.first_ping_time <= _timestamp_string(end_time) + ) + + stmt = stmt.order_by(raw_sv.c.first_ping_time) - query += """ - ORDER BY datetime(first_ping_time) - """ + engine = _engine(db_path) - with sqlite3.connect(db_path) as conn: - rows = conn.execute(query, params).fetchall() + with engine.connect() as conn: + rows = conn.execute(stmt).all() - return [row[0] for row in rows] \ No newline at end of file + return [row.sv_filename for row in rows] \ No newline at end of file diff --git a/tests/test_processing_ledger.py b/tests/test_processing_ledger.py index f6629bba..28f310e6 100644 --- a/tests/test_processing_ledger.py +++ b/tests/test_processing_ledger.py @@ -1,6 +1,7 @@ import sqlite3 from echodataflow.utils.processing_ledger import ( + _database_url, get_completed_sv_files, get_raw_files_to_process, initialize_ledger, @@ -8,9 +9,35 @@ mark_raw_failed, mark_raw_processing, register_raw_file, + resolve_database, ) +def test_database_url_from_path(tmp_path): + db_path = tmp_path / "processing.db" + + url = _database_url(db_path) + + assert url.startswith("sqlite:///") + assert url.endswith("processing.db") + + +def test_database_url_preserves_database_url(): + url = "postgresql+psycopg://user:password@localhost/test" + + assert _database_url(url) == url + +def test_resolve_database_local_path(tmp_path): + result = resolve_database(tmp_path, "processing.db") + + assert result == tmp_path / "processing.db" + + +def test_resolve_database_preserves_database_url(): + url = "postgresql+psycopg://user:password@localhost/test" + + assert resolve_database("/some/path", url) == url + def test_initialize_ledger(tmp_path): db_path = tmp_path / "processing.db" @@ -230,4 +257,15 @@ def test_register_raw_file_requeues_changed_file(tmp_path): (str(raw_path),), ).fetchone() - assert row == ("pending", None, None, None) \ No newline at end of file + assert row == ("pending", None, None, None) + +def test_register_raw_file_large_mtime(tmp_path): + db_path = tmp_path / "processing.db" + raw_path = tmp_path / "test.raw" + raw_path.touch() + + initialize_ledger(db_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 From aae5a52a127da56f27d9e37e63ceedc8e2f0931e Mon Sep 17 00:00:00 2001 From: Wu-Jung Lee Date: Thu, 20 Aug 2026 08:29:40 -0700 Subject: [PATCH 4/7] move drop duplicate for transect csv earlier for safeguard, simplify transect dataframe handling --- src/echodataflow/flows/flows_transect.py | 38 ++++++++---------------- 1 file changed, 12 insertions(+), 26 deletions(-) diff --git a/src/echodataflow/flows/flows_transect.py b/src/echodataflow/flows/flows_transect.py index e2b7d7a1..d3c9ad19 100644 --- a/src/echodataflow/flows/flows_transect.py +++ b/src/echodataflow/flows/flows_transect.py @@ -9,21 +9,24 @@ ) +TRANSECT_COLUMNS = [ + "transectPart", + "transectNumber", + "transectStart", + "transectEnd", +] + + def get_changed_transects( current: pd.DataFrame, previous: pd.DataFrame, ) -> pd.DataFrame: """Return transect segments that are new or have been updated.""" - key_columns = [ - "transectPart", - "transectNumber", - "transectStart", - "transectEnd", - ] + key_columns = TRANSECT_COLUMNS return ( - current.merge( + current.drop_duplicates(subset=key_columns).merge( previous[key_columns], on=key_columns, how="left", @@ -31,7 +34,6 @@ def get_changed_transects( ) .query("_merge == 'left_only'") .drop(columns="_merge") - .drop_duplicates(subset=key_columns) ) @@ -51,30 +53,14 @@ def flow_transect_update( # Read the current transect information, preserving transect identifiers # as strings so values with leading zeros (e.g., "002") are not converted # to integers by pandas - current = pd.read_csv( - path_transect, - dtype={ - "transectPart": "string", - "transectNumber": "string", - "transectStart": "string", - "transectEnd": "string", - }, - ) + current = pd.read_csv(path_transect, dtype="string") if not path_snapshot.exists(): print("No previous transect snapshot found. Initializing snapshot.") current.to_csv(path_snapshot, index=False) return - previous = pd.read_csv( - path_snapshot, - dtype={ - "transectPart": "string", - "transectNumber": "string", - "transectStart": "string", - "transectEnd": "string", - }, - ) + previous = pd.read_csv(path_snapshot, dtype="string") changed = get_changed_transects(current, previous) From 32acd5ab93d880f4d2accbc31b85ce640d063efc Mon Sep 17 00:00:00 2001 From: Wu-Jung Lee Date: Thu, 20 Aug 2026 08:52:45 -0700 Subject: [PATCH 5/7] consolidate event handler to a single FileChangeHandler for both watching file and directory --- src/echodataflow/utils/file_watcher.py | 65 ++++++++++------------- tests/test_file_watcher.py | 72 ++++++++++++++++---------- 2 files changed, 71 insertions(+), 66 deletions(-) diff --git a/src/echodataflow/utils/file_watcher.py b/src/echodataflow/utils/file_watcher.py index 54d3444d..0bc2119e 100644 --- a/src/echodataflow/utils/file_watcher.py +++ b/src/echodataflow/utils/file_watcher.py @@ -7,51 +7,42 @@ logger = logging.getLogger(__name__) -class FileUpdateHandler(FileSystemEventHandler): - """Run a callback when the target file is updated.""" - def __init__( - self, - target_file: str | Path, - callback: Callable[[Path], None], - ): - self.target_file = Path(target_file).resolve() - self.callback = callback - - def _handle_path(self, path: str | Path) -> None: - event_path = Path(path).resolve() - if event_path == self.target_file: - self.callback(event_path) +class FileChangeHandler(FileSystemEventHandler): + """Run a callback when a changed file satisfies ``matches``. - def on_modified(self, event: FileSystemEvent) -> None: - if not event.is_directory: - self._handle_path(event.src_path) + Match one specific file, for example a transect CSV:: - def on_created(self, event: FileSystemEvent) -> None: - if not event.is_directory: - self._handle_path(event.src_path) + target = Path("transects.csv").resolve() + FileChangeHandler( + callback=process, + matches=lambda path: path == target, + ) - def on_moved(self, event: FileSystemEvent) -> None: - if not event.is_directory: - self._handle_path(event.dest_path) + Match files by name, for example all RAW files in the watched directory:: + FileChangeHandler( + callback=process, + matches=lambda path: path.match("*.raw"), + ) -class FileCreatedHandler(FileSystemEventHandler): - """Run a callback when a matching file is created or modified.""" + Callback exceptions are logged so they do not stop the filesystem + observer. + """ def __init__( self, callback: Callable[[Path], None], - pattern: str, + matches: Callable[[Path], bool], ): self.callback = callback - self.pattern = pattern + self.matches = matches def _handle_path(self, path: str | Path) -> None: event_path = Path(path).resolve() - if not event_path.match(self.pattern): + if not self.matches(event_path): return try: @@ -62,15 +53,13 @@ def _handle_path(self, path: str | Path) -> None: event_path, ) - def _handle(self, event: FileSystemEvent) -> None: + def on_modified(self, event: FileSystemEvent) -> None: if not event.is_directory: self._handle_path(event.src_path) def on_created(self, event: FileSystemEvent) -> None: - self._handle(event) - - def on_modified(self, event: FileSystemEvent) -> None: - self._handle(event) + if not event.is_directory: + self._handle_path(event.src_path) def on_moved(self, event: FileSystemEvent) -> None: if not event.is_directory: @@ -87,9 +76,9 @@ def watch_file( observer = Observer() observer.schedule( - FileUpdateHandler( - target_file=target_file, + FileChangeHandler( callback=callback, + matches=lambda path: path == target_file, ), str(target_file.parent), recursive=False, @@ -110,13 +99,13 @@ def watch_directory( observer = Observer() observer.schedule( - FileCreatedHandler( + FileChangeHandler( callback=callback, - pattern=pattern, + matches=lambda path: path.match(pattern), ), str(directory), recursive=False, ) observer.start() - return observer \ No newline at end of file + return observer diff --git a/tests/test_file_watcher.py b/tests/test_file_watcher.py index e3aab567..8b7a17e8 100644 --- a/tests/test_file_watcher.py +++ b/tests/test_file_watcher.py @@ -1,9 +1,7 @@ from pathlib import Path -from echodataflow.utils.file_watcher import ( - FileCreatedHandler, - FileUpdateHandler, -) +from echodataflow.utils.file_watcher import FileChangeHandler + class FakeEvent: is_directory = False @@ -17,15 +15,15 @@ def __init__( self.dest_path = str(dest_path) if dest_path else str(src_path) -def test_file_update_handler_calls_callback_for_target(tmp_path): +def test_file_change_handler_calls_callback_for_target(tmp_path): target = tmp_path / "transect_start_end_time.csv" target.touch() detected = [] - handler = FileUpdateHandler( - target_file=target, + handler = FileChangeHandler( callback=detected.append, + matches=lambda path: path == target.resolve(), ) handler.on_modified(FakeEvent(target)) @@ -33,7 +31,7 @@ def test_file_update_handler_calls_callback_for_target(tmp_path): assert detected == [target.resolve()] -def test_file_update_handler_ignores_other_files(tmp_path): +def test_file_change_handler_ignores_other_files(tmp_path): target = tmp_path / "transect_start_end_time.csv" other = tmp_path / "other.csv" @@ -42,23 +40,23 @@ def test_file_update_handler_ignores_other_files(tmp_path): detected = [] - handler = FileUpdateHandler( - target_file=target, + handler = FileChangeHandler( callback=detected.append, + matches=lambda path: path == target.resolve(), ) handler.on_modified(FakeEvent(other)) assert detected == [] - - -def test_file_update_handler_calls_callback_for_created_target(tmp_path): + + +def test_file_change_handler_calls_callback_for_created_target(tmp_path): target = tmp_path / "transect_start_end_time.csv" detected = [] - handler = FileUpdateHandler( - target_file=target, + handler = FileChangeHandler( callback=detected.append, + matches=lambda path: path == target.resolve(), ) handler.on_created(FakeEvent(target)) @@ -66,14 +64,14 @@ def test_file_update_handler_calls_callback_for_created_target(tmp_path): assert detected == [target.resolve()] -def test_file_update_handler_calls_callback_for_moved_target(tmp_path): +def test_file_change_handler_calls_callback_for_moved_target(tmp_path): target = tmp_path / "transect_start_end_time.csv" temporary = tmp_path / "temporary.csv" detected = [] - handler = FileUpdateHandler( - target_file=target, + handler = FileChangeHandler( callback=detected.append, + matches=lambda path: path == target.resolve(), ) handler.on_moved( @@ -86,13 +84,13 @@ def test_file_update_handler_calls_callback_for_moved_target(tmp_path): assert detected == [target.resolve()] -def test_file_created_handler_calls_callback_for_matching_file(tmp_path): +def test_file_change_handler_calls_callback_for_matching_file(tmp_path): raw_file = tmp_path / "example.raw" detected = [] - handler = FileCreatedHandler( + handler = FileChangeHandler( callback=detected.append, - pattern="*.raw", + matches=lambda path: path.match("*.raw"), ) handler.on_created(FakeEvent(raw_file)) @@ -100,27 +98,28 @@ def test_file_created_handler_calls_callback_for_matching_file(tmp_path): assert detected == [raw_file.resolve()] -def test_file_created_handler_ignores_nonmatching_file(tmp_path): +def test_file_change_handler_ignores_nonmatching_file(tmp_path): other_file = tmp_path / "example.txt" detected = [] - handler = FileCreatedHandler( + handler = FileChangeHandler( callback=detected.append, - pattern="*.raw", + matches=lambda path: path.match("*.raw"), ) handler.on_created(FakeEvent(other_file)) assert detected == [] -def test_file_created_handler_calls_callback_for_moved_matching_file(tmp_path): + +def test_file_change_handler_calls_callback_for_moved_matching_file(tmp_path): temporary = tmp_path / "temporary.tmp" raw_file = tmp_path / "example.raw" detected = [] - handler = FileCreatedHandler( + handler = FileChangeHandler( callback=detected.append, - pattern="*.raw", + matches=lambda path: path.match("*.raw"), ) handler.on_moved( @@ -130,4 +129,21 @@ def test_file_created_handler_calls_callback_for_moved_matching_file(tmp_path): ) ) - assert detected == [raw_file.resolve()] \ No newline at end of file + assert detected == [raw_file.resolve()] + + +def test_file_change_handler_logs_callback_exception(tmp_path, caplog): + target = tmp_path / "transect_start_end_time.csv" + + def failing_callback(_path): + raise RuntimeError("callback failed") + + handler = FileChangeHandler( + callback=failing_callback, + matches=lambda path: path == target.resolve(), + ) + + handler.on_modified(FakeEvent(target)) + + assert "Error handling filesystem update" in caplog.text + assert "callback failed" in caplog.text From 19780785dd8dc0bcfa27bb8410aa28ee01f1cad9 Mon Sep 17 00:00:00 2001 From: Wu-Jung Lee Date: Thu, 20 Aug 2026 08:56:09 -0700 Subject: [PATCH 6/7] move raw_monitor and transect_minotor to operations/ --- src/echodataflow/{utils => operations}/raw_monitor.py | 2 +- src/echodataflow/{utils => operations}/transect_monitor.py | 1 - tests/test_raw_monitor.py | 2 +- tests/test_transect_monitor.py | 4 ++-- 4 files changed, 4 insertions(+), 5 deletions(-) rename src/echodataflow/{utils => operations}/raw_monitor.py (99%) rename src/echodataflow/{utils => operations}/transect_monitor.py (99%) diff --git a/src/echodataflow/utils/raw_monitor.py b/src/echodataflow/operations/raw_monitor.py similarity index 99% rename from src/echodataflow/utils/raw_monitor.py rename to src/echodataflow/operations/raw_monitor.py index 4365e01a..bbf58cce 100644 --- a/src/echodataflow/utils/raw_monitor.py +++ b/src/echodataflow/operations/raw_monitor.py @@ -55,4 +55,4 @@ def watch_raw_directory( directory=raw_directory, callback=lambda raw_path: register_and_emit_raw_update(raw_path, db_path), pattern="*.raw", - ) \ No newline at end of file + ) diff --git a/src/echodataflow/utils/transect_monitor.py b/src/echodataflow/operations/transect_monitor.py similarity index 99% rename from src/echodataflow/utils/transect_monitor.py rename to src/echodataflow/operations/transect_monitor.py index fe0ffc12..d793b595 100644 --- a/src/echodataflow/utils/transect_monitor.py +++ b/src/echodataflow/operations/transect_monitor.py @@ -36,4 +36,3 @@ def watch_transect_file(path: str | Path): target_file=path, callback=emit_transect_update_event, ) - \ No newline at end of file diff --git a/tests/test_raw_monitor.py b/tests/test_raw_monitor.py index 9baa1f76..6a6266d4 100644 --- a/tests/test_raw_monitor.py +++ b/tests/test_raw_monitor.py @@ -1,6 +1,6 @@ from pathlib import Path -from echodataflow.utils import raw_monitor +from echodataflow.operations import raw_monitor def test_emit_raw_update_event(monkeypatch, tmp_path): diff --git a/tests/test_transect_monitor.py b/tests/test_transect_monitor.py index 6aacd0fc..439f5bab 100644 --- a/tests/test_transect_monitor.py +++ b/tests/test_transect_monitor.py @@ -1,5 +1,6 @@ from pathlib import Path -from echodataflow.utils import transect_monitor + +from echodataflow.operations import transect_monitor def test_emit_transect_update_event(monkeypatch, tmp_path): @@ -22,4 +23,3 @@ def fake_emit_event(**kwargs): assert emitted["event"] == "echodataflow.transect.updated" assert emitted["resource"]["prefect.resource.id"] == "transect-start-end-time" assert emitted["resource"]["path"] == str(target) - \ No newline at end of file From 49955de008be8399394a477cae8c9c043acb72a8 Mon Sep 17 00:00:00 2001 From: Wu-Jung Lee Date: Thu, 20 Aug 2026 09:06:50 -0700 Subject: [PATCH 7/7] move raw_monitor and transect_monitor contents to operations_watchdog --- ...{raw_monitor.py => operations_watchdog.py} | 34 ++++++++++++- .../operations/transect_monitor.py | 38 -------------- ...monitor.py => test_operations_watchdog.py} | 51 +++++++++++++------ tests/test_transect_monitor.py | 25 --------- 4 files changed, 69 insertions(+), 79 deletions(-) rename src/echodataflow/operations/{raw_monitor.py => operations_watchdog.py} (59%) delete mode 100644 src/echodataflow/operations/transect_monitor.py rename tests/{test_raw_monitor.py => test_operations_watchdog.py} (71%) delete mode 100644 tests/test_transect_monitor.py diff --git a/src/echodataflow/operations/raw_monitor.py b/src/echodataflow/operations/operations_watchdog.py similarity index 59% rename from src/echodataflow/operations/raw_monitor.py rename to src/echodataflow/operations/operations_watchdog.py index bbf58cce..e49c0cbc 100644 --- a/src/echodataflow/operations/raw_monitor.py +++ b/src/echodataflow/operations/operations_watchdog.py @@ -2,7 +2,7 @@ from prefect.events import emit_event -from echodataflow.utils.file_watcher import watch_directory +from echodataflow.utils.file_watcher import watch_directory, watch_file from echodataflow.utils.processing_ledger import ( initialize_ledger, register_raw_file, @@ -12,6 +12,10 @@ 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" + def emit_raw_update_event(path: Path) -> None: """Emit a Prefect event when a RAW file arrives.""" @@ -56,3 +60,31 @@ def watch_raw_directory( 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=TRANSECT_UPDATE_EVENT, + resource={ + "prefect.resource.id": TRANSECT_RESOURCE_ID, + "path": str(path), + }, + related=[ + { + "prefect.resource.id": TRANSECT_RELATED_RESOURCE_ID, + "prefect.resource.name": TRANSECT_RELATED_RESOURCE_ID, + "prefect.resource.role": "deployment", + } + ], + ) + + +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, + ) diff --git a/src/echodataflow/operations/transect_monitor.py b/src/echodataflow/operations/transect_monitor.py deleted file mode 100644 index d793b595..00000000 --- a/src/echodataflow/operations/transect_monitor.py +++ /dev/null @@ -1,38 +0,0 @@ -from pathlib import Path - -from prefect.events import emit_event - -from echodataflow.utils.file_watcher import watch_file - - -TRANSECT_UPDATE_EVENT = "echodataflow.transect.updated" -TRANSECT_RESOURCE_ID = "transect-start-end-time" -TRANSECT_RELATED_RESOURCE_ID = "transect-monitor" - - -def emit_transect_update_event(path: Path) -> None: - """Emit a Prefect event when the transect CSV is updated.""" - - emit_event( - event=TRANSECT_UPDATE_EVENT, - resource={ - "prefect.resource.id": TRANSECT_RESOURCE_ID, - "path": str(path), - }, - related=[ - { - "prefect.resource.id": TRANSECT_RELATED_RESOURCE_ID, - "prefect.resource.name": TRANSECT_RELATED_RESOURCE_ID, - "prefect.resource.role": "deployment", - } - ], - ) - - -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, - ) diff --git a/tests/test_raw_monitor.py b/tests/test_operations_watchdog.py similarity index 71% rename from tests/test_raw_monitor.py rename to tests/test_operations_watchdog.py index 6a6266d4..1c9b4f72 100644 --- a/tests/test_raw_monitor.py +++ b/tests/test_operations_watchdog.py @@ -1,6 +1,4 @@ -from pathlib import Path - -from echodataflow.operations import raw_monitor +from echodataflow.operations import operations_watchdog def test_emit_raw_update_event(monkeypatch, tmp_path): @@ -13,12 +11,12 @@ def fake_emit_event(**kwargs): emitted.update(kwargs) monkeypatch.setattr( - raw_monitor, + operations_watchdog, "emit_event", fake_emit_event, ) - raw_monitor.emit_raw_update_event(raw_file) + operations_watchdog.emit_raw_update_event(raw_file) assert emitted["event"] == "echodataflow.raw.updated" assert emitted["resource"]["prefect.resource.id"] == "raw-monitor" @@ -44,18 +42,18 @@ def fake_watch_directory( return "observer" monkeypatch.setattr( - raw_monitor, + operations_watchdog, "initialize_ledger", fake_initialize_ledger, ) monkeypatch.setattr( - raw_monitor, + operations_watchdog, "watch_directory", fake_watch_directory, ) - result = raw_monitor.watch_raw_directory( + result = operations_watchdog.watch_raw_directory( tmp_path, db_path, ) @@ -79,18 +77,18 @@ def fake_emit_raw_update_event(path): called["emitted"] = path monkeypatch.setattr( - raw_monitor, + operations_watchdog, "register_raw_file", fake_register_raw_file, ) monkeypatch.setattr( - raw_monitor, + operations_watchdog, "emit_raw_update_event", fake_emit_raw_update_event, ) - raw_monitor.register_and_emit_raw_update( + operations_watchdog.register_and_emit_raw_update( raw_file, db_path, ) @@ -98,6 +96,7 @@ def fake_emit_raw_update_event(path): assert called["registered"] == (db_path, raw_file) assert called["emitted"] == raw_file + def test_watch_raw_directory_reconciles_existing_raw_files(monkeypatch, tmp_path): raw_a = tmp_path / "a.raw" raw_b = tmp_path / "b.raw" @@ -110,24 +109,24 @@ def test_watch_raw_directory_reconciles_existing_raw_files(monkeypatch, tmp_path emitted = [] monkeypatch.setattr( - raw_monitor, + operations_watchdog, "register_raw_file", lambda db, path: registered.append((db, path)), ) monkeypatch.setattr( - raw_monitor, + operations_watchdog, "emit_raw_update_event", emitted.append, ) monkeypatch.setattr( - raw_monitor, + operations_watchdog, "watch_directory", lambda **kwargs: "observer", ) - result = raw_monitor.watch_raw_directory( + result = operations_watchdog.watch_raw_directory( tmp_path, db_path, ) @@ -138,3 +137,25 @@ def test_watch_raw_directory_reconciles_existing_raw_files(monkeypatch, tmp_path raw_b.resolve(), } assert emitted == [tmp_path.resolve()] + + +def test_emit_transect_update_event(monkeypatch, tmp_path): + target = tmp_path / "transect_start_end_time.csv" + target.touch() + + emitted = {} + + def fake_emit_event(**kwargs): + emitted.update(kwargs) + + monkeypatch.setattr( + operations_watchdog, + "emit_event", + fake_emit_event, + ) + + 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) diff --git a/tests/test_transect_monitor.py b/tests/test_transect_monitor.py deleted file mode 100644 index 439f5bab..00000000 --- a/tests/test_transect_monitor.py +++ /dev/null @@ -1,25 +0,0 @@ -from pathlib import Path - -from echodataflow.operations import transect_monitor - - -def test_emit_transect_update_event(monkeypatch, tmp_path): - target = tmp_path / "transect_start_end_time.csv" - target.touch() - - emitted = {} - - def fake_emit_event(**kwargs): - emitted.update(kwargs) - - monkeypatch.setattr( - transect_monitor, - "emit_event", - fake_emit_event, - ) - - transect_monitor.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)