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 6068c4f7..2659db39 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,16 @@ 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, + resolve_database, +) + # Turn on verbose logging for echopype # otherwise all logging will be muted ep.utils.log.verbose() @@ -66,13 +74,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 +86,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 = resolve_database(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 +145,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 +254,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 +292,7 @@ async def flow_create_MVBS( ) # Assemble paths - file_Sv_csv = Path(path_main) / file_Sv_csv + 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" @@ -299,19 +310,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 +334,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..e2b7d7a1 100644 --- a/src/echodataflow/flows/flows_transect.py +++ b/src/echodataflow/flows/flows_transect.py @@ -3,6 +3,11 @@ import pandas as pd from prefect import flow +from echodataflow.utils.processing_ledger import ( + get_completed_sv_files, + resolve_database, +) + def get_changed_transects( current: pd.DataFrame, @@ -30,31 +35,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 = 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 @@ -97,45 +89,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/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/processing_ledger.py b/src/echodataflow/utils/processing_ledger.py new file mode 100644 index 00000000..01be1134 --- /dev/null +++ b/src/echodataflow/utils/processing_ledger.py @@ -0,0 +1,285 @@ +from __future__ import annotations + +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.""" + + engine = _engine(db_path) + + # 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") + + metadata.create_all(engine) + + +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() + + engine = _engine(db_path) + + with engine.begin() as conn: + existing = conn.execute( + 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(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", + ) + ) + return + + if ( + existing.file_size != stat.st_size + or existing.file_mtime_ns != stat.st_mtime_ns + ): + conn.execute( + 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.""" + + 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: + stmt = stmt.limit(limit) + + engine = _engine(db_path) + + with engine.connect() as conn: + rows = conn.execute(stmt).all() + + 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.""" + + engine = _engine(db_path) + + with engine.begin() as conn: + conn.execute( + update(raw_sv) + .where(raw_sv.c.raw_path == str(Path(raw_path))) + .values( + status="processing", + error="", + updated_at=func.current_timestamp(), + ) + ) + + +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.""" + + engine = _engine(db_path) + + with engine.begin() as conn: + conn.execute( + 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(), + ) + ) + + +def mark_raw_failed( + db_path: str | Path, + raw_path: str | Path, + error: str, +) -> None: + """Mark a RAW file as failed.""" + + engine = _engine(db_path) + + with engine.begin() as conn: + conn.execute( + 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.""" + + 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: + stmt = stmt.where( + raw_sv.c.last_ping_time >= _timestamp_string(start_time) + ) + + if end_time is not None: + stmt = stmt.where( + raw_sv.c.first_ping_time <= _timestamp_string(end_time) + ) + + stmt = stmt.order_by(raw_sv.c.first_ping_time) + + engine = _engine(db_path) + + with engine.connect() as conn: + rows = conn.execute(stmt).all() + + return [row.sv_filename 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..4365e01a 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,34 @@ 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.""" + 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, - callback=emit_raw_update_event, + 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_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..28f310e6 --- /dev/null +++ b/tests/test_processing_ledger.py @@ -0,0 +1,271 @@ +import sqlite3 + +from echodataflow.utils.processing_ledger import ( + _database_url, + get_completed_sv_files, + get_raw_files_to_process, + initialize_ledger, + mark_raw_completed, + 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" + + 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) + +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 diff --git a/tests/test_raw_monitor.py b/tests/test_raw_monitor.py index b90fc98b..9baa1f76 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,98 @@ 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 + +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()]