Add contour visualization lasker

src/echodataflow/flows/flows_viz_cloud.py

@@ -1,11 +1,14 @@
 from pathlib import Path
 import datetime
 import configparser
+import tempfile
 
 import pandas as pd
 import xarray as xr
 import s3fs
 
+import echoregions as er
+
 from prefect import flow, get_run_logger
 
 from echodataflow.utils.utils import round_up_mins, get_slice_start_end_times
@@ -123,4 +126,125 @@ def flow_update_cache_MVBS(
             Path(path_cache) / file_MVBS_zarr,  # cache is local
             mode="w",
             consolidated=True,
-        )
+        )
+
+
+@flow()
+def flow_update_cache_contours(
+    time_offset_seconds: float = 0.0,
+    slice_mins: int = 180,
+    path_cache: str = "PATH_TO_DATA_CACHE",
+    path_EVR: str = "PATH_TO_EVR_DATA_STORE",
+    cred_file: str = "PATH_TO_CREDENTIALS_FILE",
+    file_contours_csv: str = "latest_contours.csv",
+):
+    logger = get_run_logger()
+
+    # Set end time
+    end_time = (
+        datetime.datetime.now(datetime.timezone.utc)
+        - datetime.timedelta(seconds=time_offset_seconds)
+    )
+
+    logger.info(
+        "flow started with parameters:\n"
+        f"- end_time: {end_time}\n"
+        f"- slice_mins: {slice_mins}\n"
+    )
+
+    # Compute slice time range
+    start_time, end_time = get_slice_start_end_times(
+        end_time=end_time,
+        slice_mins=slice_mins,
+        num_slices=1,
+    )
+
+    # Get cloud bucket
+    config = configparser.ConfigParser()
+    config.read(cred_file)
+    fs = s3fs.S3FileSystem(
+        key=config["osn_sdsc_hake"]["access_key_id"],
+        secret=config["osn_sdsc_hake"]["secret_access_key"],
+        client_kwargs={"endpoint_url": config["osn_sdsc_hake"]["endpoint"]},
+    )
+
+    # Find EVR files
+    evr_files = fs.glob(f"{path_EVR}/*.evr")
+    selected_evr = []
+    for evr_file in evr_files:
+        filename = Path(evr_file).stem
+        # Example filename:
+        # prediction_20260710T182000
+        timestamp = datetime.datetime.strptime(
+            filename.split("_")[-1],
+            "%Y%m%dT%H%M%S",
+        ).replace(tzinfo=datetime.timezone.utc)
+        # Subset for time range
+        if start_time[0] <= timestamp <= end_time[0]:
+            selected_evr.append(
+                (timestamp, evr_file)
+            )
+    selected_evr.sort(key=lambda x: x[0])
+
+    logger.info(
+        f"Found {len(selected_evr)} EVR files in time range:\n"
+        + "".join(
+            [
+                f"- {Path(f).name} ({t})\n"
+                for t, f in selected_evr
+            ]
+        )
+    )
+
+    if len(selected_evr) == 0:
+        logger.info(
+            "Contours cache not updated: no EVR files in specified time range"
+        )
+    else:
+        # Read EVRs and collect dataframes
+        contours_dfs = []
+
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            tmp_path = Path(tmp_dir)
+
+            for _, evr_file in selected_evr:
+                logger.info(f"Downloading EVR: {evr_file}")
+
+                local_evr = tmp_path / Path(evr_file).name
+
+                fs.get(
+                    evr_file,
+                    str(local_evr),
+                )
+
+                logger.info(f"Reading local EVR: {local_evr}")
+
+                try:
+                    regions = er.read_evr(str(local_evr))
+
+                    contours_dfs.append(
+                        regions.data
+                    )
+
+                finally:
+                    if local_evr.exists():
+                        local_evr.unlink()
+                        logger.info(f"Removed temporary EVR: {local_evr}")
+
+        # Merge all regions
+        df_contours = pd.concat(
+            contours_dfs,
+            ignore_index=True,
+        )
+
+        logger.info(
+            f"Merged and saving {len(df_contours)} contour regions"
+        )
+
+        # Save CSV cache
+        output_file = Path(path_cache) / file_contours_csv
+
+        df_contours.to_csv(
+            output_file,
+            index=False,
+        )

src/echodataflow/services/viz_echogram_track_lasker.py

@@ -1,20 +1,24 @@
 from pathlib import Path
+import ast
+
 import panel as pn
 import xarray as xr
 from holoviews import opts
+import holoviews as hv
 import echoshader
+import pandas as pd
+import numpy as np
 
 # Configure Panel to prevent automatic refreshes
 pn.config.autoreload = False
 
-path_MVBS = Path("/media/volume/shimada_202506_volume/viz_data_cache_2026/iwcsp_2026")
-
+path_latest = Path("/media/volume/shimada_202506_volume/viz_data_cache_2026/iwcsp_2026")
 
 def update_cache_multi_freq():
     """
     Load latest MVBS data and create multi-frequency echograms.
     """
-    ds_MVBS = xr.open_zarr(path_MVBS / "latest_MVBS.zarr")
+    ds_MVBS = xr.open_zarr(path_latest / "latest_MVBS.zarr")
     egram = ds_MVBS.eshader.echogram(
         channel=[
             "WBT 987760-15 ES18_ES",
@@ -65,7 +69,7 @@ def update_cache_tricolor():
     """
     Load latest MVBS data and create tricolor echogram.
     """
-    ds_MVBS = xr.open_zarr(path_MVBS / "latest_MVBS.zarr")
+    ds_MVBS = xr.open_zarr(path_latest / "latest_MVBS.zarr")
 
     tricolor = ds_MVBS.eshader.echogram(
         channel=[
@@ -110,11 +114,94 @@ def scheduled_update():
     return plot_pane
 
 
+def create_contours_overlay():
+    """
+    Convert hake contours dataframe into HoloViews paths.
+    """
+
+    contours_df = pd.read_csv(
+        path_latest / "latest_contours.csv"
+    )
+
+    # Convert string representations back to arrays
+    contours_df["depth"] = contours_df["depth"].apply(
+        lambda x: np.fromstring(
+            x.strip("[]"),
+            sep=" "
+        )
+    )
+    contours_df["time"] = contours_df["time"].apply(
+        lambda x: np.array(
+            x.strip("[]").replace("'", "").split()
+        )
+    )
+
+    # Set to datetime
+    contours_df["time"] = contours_df["time"].apply(
+        lambda x: pd.to_datetime(x).to_numpy(dtype="datetime64[ns]")
+    )
+
+    # Create holoviews path object
+    hv_paths = []
+    for _, row in contours_df.iterrows():
+        time = list(row["time"])
+        depth = list(row["depth"])
+
+        # Close contour by appending first point to the end
+        time.append(time[0])
+        depth.append(depth[0])
+
+        hv_paths.append(
+            {
+                "ping_time": time,
+                "depth": depth,
+            }
+        )
+    contours_hv = hv.Path(
+        hv_paths,
+        kdims=["ping_time", "depth"],
+    ).opts(
+        color="magenta",
+        line_width=3,
+    )
+
+    return contours_hv
+
+
+def tricolor_with_contour_app():
+    """
+    Plot tricolor echogram with regular updates.
+    """
+    # Create initial plot
+    tricolor = update_cache_tricolor()
+    contours_hv = create_contours_overlay()
+    tricolor_with_contour = tricolor() * contours_hv
+    plot_pane = pn.pane.HoloViews(tricolor_with_contour)
+
+    # Simple update function that only runs every 10 minutes
+    def scheduled_update():
+        try:
+            new_tricolor = update_cache_tricolor()
+            plot_pane.object = new_tricolor
+            print("Plot updated at scheduled interval")
+        except Exception as e:
+            print(f"Error during scheduled update: {e}")
+
+    # Add ONLY the 10-minute callback - no other automatic updates
+    pn.state.add_periodic_callback(
+        scheduled_update,
+        period=10*60*1000  # Update every 10 mins
+    )
+
+    return plot_pane
+
+
 # Deploy the application with stable configuration
 test_server = pn.serve(
     {
         "multi_freq_echogram": multi_freq_app,
         "tricolor_echogram": tricolor_app,
+        "tricolor_contour_echogram": tricolor_with_contour_app,
     },
     port=1802,
     websocket_origin="*",