Add transect update monitoring flow

pyproject.toml

@@ -48,6 +48,7 @@ dependencies = [
     "PyYAML",
     "s3fs",
     "shapely",
+   "watchdog",
     "xarray",
 ]
 dynamic = ["version"]

src/echodataflow/deployment/core.py

@@ -27,6 +27,10 @@
     "processes",
     "threads_per_worker",
 }
-ALLOWED_TRIGGER_KEYS = {"expect", "resource_name"}
+ALLOWED_TRIGGER_KEYS = {
+    "expect",
+    "resource_name",
+    "resource_scope",
+}
 ALLOWED_SOURCE_KEYS = {"mode", "git"}
 ALLOWED_GIT_SOURCE_KEYS = {"url", "branch"}

src/echodataflow/deployment/deployment_engine.py

@@ -227,18 +227,31 @@ def build_cron(interval: int | None, cron_offset: int = 0) -> str | None:
     return f"*/{interval} * * * *"
 
 
-def build_triggers(trigger_items: list[dict[str, Any]]) -> list[Any]:
-    return [
-        DeploymentEventTrigger(
-            expect={item["expect"]},
-            match_related={
+def build_triggers(
+    trigger_items: list[dict[str, Any]],
+) -> list[Any]:
+    triggers = []
+
+    for item in trigger_items:
+        kwargs = {
+            "expect": {item["expect"]},
+        }
+
+        if item["resource_scope"] == "primary":
+            kwargs["match"] = {
+                "prefect.resource.name": item["resource_name"],
+            }
+        else:
+            kwargs["match_related"] = {
                 "prefect.resource.name": item["resource_name"],
                 "prefect.resource.role": "deployment",
-            },
+            }
+
+        triggers.append(
+            DeploymentEventTrigger(**kwargs)
         )
-        for item in trigger_items
-    ]
 
+    return triggers
 
 def _reject_unknown_keys(
     value: dict[str, Any],
@@ -397,34 +410,55 @@ def validate_triggers(
     When configured, triggers must be a non-empty list of mappings with
     non-empty string values for `expect` and `resource_name`.
     """
+
     triggers = validate_optional_non_empty_list(
         triggers,
         field_name=f"deploy_cfg.flows.{flow_key}.triggers",
         item_label="trigger",
     )
+
     if triggers is None:
         return None
 
     validated_triggers: list[dict[str, Any]] = []
+
     for trigger_item in triggers:
         if not isinstance(trigger_item, dict):
             raise ValueError(f"deploy_cfg.flows.{flow_key}.triggers entries must be mappings")
 
         expect = trigger_item.get("expect")
         resource_name = trigger_item.get("resource_name")
+        resource_scope = trigger_item.get(
+            "resource_scope",
+            "related",
+        )
+
         if not isinstance(expect, str) or not expect.strip():
             raise ValueError(
-                f"deploy_cfg.flows.{flow_key}.triggers entries must define a non-empty 'expect'"
+                f"deploy_cfg.flows.{flow_key}.triggers entries "
+                "must define a non-empty 'expect'"
             )
+
         if not isinstance(resource_name, str) or not resource_name.strip():
             raise ValueError(
-                f"deploy_cfg.flows.{flow_key}.triggers entries must define a non-empty 'resource_name'"
+                f"deploy_cfg.flows.{flow_key}.triggers entries "
+                "must define a non-empty 'resource_name'"
+            )
+
+        if resource_scope not in {
+            "primary",
+            "related",
+        }:
+            raise ValueError(
+                f"deploy_cfg.flows.{flow_key}.triggers "
+                "resource_scope must be 'primary' or 'related'"
             )
 
         validated_triggers.append(
             {
                 "expect": expect.strip(),
                 "resource_name": resource_name.strip(),
+                "resource_scope": resource_scope,
             }
         )
 

src/echodataflow/deployment/flow_registry.py

@@ -66,4 +66,8 @@ class FlowRegistration:
     "update_cache_MVBS": FlowRegistration(
         entrypoint="echodataflow/flows/flows_viz_cloud.py:flow_update_cache_MVBS",
     ),
+    "transect_update": FlowRegistration(
+        entrypoint="echodataflow/flows/flows_transect.py:flow_transect_update",
+        description="Process updates to transect start/end information.",
+    ),
 }

src/echodataflow/flows/flows_transect.py

@@ -0,0 +1,141 @@
+from pathlib import Path
+
+import pandas as pd
+from prefect import flow
+
+
+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",
+    ]
+
+    return (
+        current.merge(
+            previous[key_columns],
+            on=key_columns,
+            how="left",
+            indicator=True,
+        )
+        .query("_merge == 'left_only'")
+        .drop(columns="_merge")
+        .drop_duplicates(subset=key_columns)
+    )
+
+
+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",
+):
+    """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
+
+    # 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",
+        },
+    )
+
+    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",
+        },
+    )
+
+    changed = get_changed_transects(current, previous)
+
+    # Only process completed transects
+    changed = changed.dropna(subset=["transectEnd"])
+
+    if changed.empty:
+        print("No new or updated transect segments.")
+        current.to_csv(path_snapshot, index=False)
+        return
+
+    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 = sorted(overlapping_sv["Sv_filename"].tolist())
+
+        print(
+            f"\nTransect {transect['transectPart']}: "
+            f"{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)

src/echodataflow/utils/file_watcher.py

@@ -0,0 +1,107 @@
+from collections.abc import Callable
+from pathlib import Path
+
+from watchdog.events import FileSystemEvent, FileSystemEventHandler
+from watchdog.observers import Observer
+
+
+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)
+
+    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:
+        if not event.is_directory:
+            self._handle_path(event.src_path)
+
+    def on_moved(self, event: FileSystemEvent) -> None:
+        if not event.is_directory:
+            self._handle_path(event.dest_path)
+
+
+class FileCreatedHandler(FileSystemEventHandler):
+    """Run a callback when a matching file is created or modified."""
+
+    def __init__(
+        self,
+        callback: Callable[[Path], None],
+        pattern: str,
+    ):
+        self.callback = callback
+        self.pattern = pattern
+
+    def _handle(self, event: FileSystemEvent) -> None:
+        if event.is_directory:
+            return
+
+        event_path = Path(event.src_path).resolve()
+
+        if event_path.match(self.pattern):
+            self.callback(event_path)
+
+    def on_created(self, event: FileSystemEvent) -> None:
+        self._handle(event)
+
+    def on_modified(self, event: FileSystemEvent) -> None:
+        self._handle(event)
+
+
+def watch_file(
+    target_file: str | Path,
+    callback: Callable[[Path], None],
+) -> Observer:
+    """Start watching a file for modifications."""
+
+    target_file = Path(target_file).resolve()
+
+    observer = Observer()
+    observer.schedule(
+        FileUpdateHandler(
+            target_file=target_file,
+            callback=callback,
+        ),
+        str(target_file.parent),
+        recursive=False,
+    )
+    observer.start()
+
+    return observer
+
+
+def watch_directory(
+    directory: str | Path,
+    callback: Callable[[Path], None],
+    pattern: str,
+) -> Observer:
+    """Start watching a directory for matching file changes."""
+
+    directory = Path(directory).resolve()
+
+    observer = Observer()
+    observer.schedule(
+        FileCreatedHandler(
+            callback=callback,
+            pattern=pattern,
+        ),
+        str(directory),
+        recursive=False,
+    )
+    observer.start()
+
+    return observer