Update ledger

src/echodataflow/flows/flows_transect.py

@@ -9,29 +9,31 @@
 )
 
 
+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",
             indicator=True,
         )
         .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)
 

src/echodataflow/utils/raw_monitor.py → 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."""
@@ -55,4 +59,32 @@ def watch_raw_directory(
         directory=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,
+    )

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
+    return observer