From a2cbf9558cfa5d952b7b20e70c558026f1b2e13d Mon Sep 17 00:00:00 2001 From: Wu-Jung Lee Date: Fri, 10 Jul 2026 20:41:30 +0000 Subject: [PATCH 1/8] add initial update cache EVR code --- src/echodataflow/flows/flows_viz_cloud.py | 127 +++++++++++++++++- .../services/viz_echogram_track_lasker.py | 98 +++++++++++++- 2 files changed, 220 insertions(+), 5 deletions(-) diff --git a/src/echodataflow/flows/flows_viz_cloud.py b/src/echodataflow/flows/flows_viz_cloud.py index 96642c9..93e0c7d 100644 --- a/src/echodataflow/flows/flows_viz_cloud.py +++ b/src/echodataflow/flows/flows_viz_cloud.py @@ -6,6 +6,8 @@ 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 +125,127 @@ def flow_update_cache_MVBS( Path(path_cache) / file_MVBS_zarr, # cache is local mode="w", consolidated=True, - ) \ No newline at end of file + ) + + +@flow() +def flow_update_cache_EVR( + 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, + ) + + logger.info( + f"Using time range:\n" + f"- start: {start_time[0]}\n" + f"- end: {end_time[0]}" + ) + + # Connect to object storage + 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( + "EVR cache not updated: no EVR files in specified time range" + ) + else: + # Read EVRs and collect dataframes + contours_dfs = [] + + for _, evr_file in selected_evr: + logger.info(f"Reading EVR: {evr_file}") + + regions = er.read_evr(evr_file) + + contours_dfs.append( + regions.data + ) + + # Merge all regions + df_contours = pd.concat( + contours_dfs, + ignore_index=True, + ) + + logger.info( + f"Merged {len(df_contours)} contour regions" + ) + + # Save CSV cache + output_file = Path(path_cache) / file_contours_csv + + logger.info( + f"Saving contour cache: {output_file}" + ) + + df_contours.to_csv( + output_file, + index=False, + ) + + logger.info("Contour cache update complete") diff --git a/src/echodataflow/services/viz_echogram_track_lasker.py b/src/echodataflow/services/viz_echogram_track_lasker.py index c42a1a2..812e38a 100644 --- a/src/echodataflow/services/viz_echogram_track_lasker.py +++ b/src/echodataflow/services/viz_echogram_track_lasker.py @@ -1,20 +1,23 @@ 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 # 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 +68,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 +113,98 @@ def scheduled_update(): return plot_pane +def create_contour_overlay(): + """ + Convert contours dataframe into HoloViews paths. + """ + + contours_df = pd.read_csv( + path_latest / "latest_contours.csv" + ) + + # Convert string representations back to arrays + contours_df["time"] = contours_df["time"].apply(ast.literal_eval) + contours_df["depth"] = contours_df["depth"].apply(ast.literal_eval) + + hv_paths = [] + + for _, row in contours_df.iterrows(): + hv_paths.append( + { + "time": row["time"], + "depth": row["depth"], + } + ) + + contour_hv = hv.Path( + hv_paths, + kdims=["time", "depth"], + ).opts( + color="magenta", + line_width=5, + ) + + return contour_hv + + +def update_cache_tricolor_with_contour(): + """ + Load latest MVBS data and create tricolor echogram. + """ + ds_MVBS = xr.open_zarr(path_latest / "latest_MVBS.zarr") + + tricolor = ds_MVBS.eshader.echogram( + channel=[ + "WBT 987753-15 ES120-7C_ES", + "WBT 987763-15 ES38-7_ES", + "WBT 987760-15 ES18_ES", + ], + vmin=-70, + vmax=-36, + rgb_composite=True, + opts=opts.RGB( + width=1200, height=600, + tools=["pan", "box_zoom", "wheel_zoom", "reset"], + ) + ) + + contour_hv = create_contour_overlay() + + return tricolor * contour_hv + + +def tricolor_with_contour_app(): + """ + Plot tricolor echogram with regular updates. + """ + # Create initial plot + tricolor_with_contour = update_cache_tricolor_with_contour() + 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_echogram_with_hake_contour": tricolor_with_contour_app, }, port=1802, websocket_origin="*", From c8c2c5986c9b3920200c8d69d60030b45a4c038a Mon Sep 17 00:00:00 2001 From: Wu-Jung Lee Date: Fri, 10 Jul 2026 20:42:21 +0000 Subject: [PATCH 2/8] contour to contours --- src/echodataflow/services/viz_echogram_track_lasker.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/echodataflow/services/viz_echogram_track_lasker.py b/src/echodataflow/services/viz_echogram_track_lasker.py index 812e38a..4aeb3ac 100644 --- a/src/echodataflow/services/viz_echogram_track_lasker.py +++ b/src/echodataflow/services/viz_echogram_track_lasker.py @@ -136,7 +136,7 @@ def create_contour_overlay(): } ) - contour_hv = hv.Path( + contours_hv = hv.Path( hv_paths, kdims=["time", "depth"], ).opts( @@ -144,7 +144,7 @@ def create_contour_overlay(): line_width=5, ) - return contour_hv + return contours_hv def update_cache_tricolor_with_contour(): @@ -168,9 +168,9 @@ def update_cache_tricolor_with_contour(): ) ) - contour_hv = create_contour_overlay() + contours_hv = create_contour_overlay() - return tricolor * contour_hv + return tricolor * contours_hv def tricolor_with_contour_app(): From b26580d0a4e409360c135042c35e1def29823774 Mon Sep 17 00:00:00 2001 From: Wu-Jung Lee Date: Fri, 10 Jul 2026 20:57:15 +0000 Subject: [PATCH 3/8] update flow name --- src/echodataflow/flows/flows_viz_cloud.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/echodataflow/flows/flows_viz_cloud.py b/src/echodataflow/flows/flows_viz_cloud.py index 93e0c7d..91227c5 100644 --- a/src/echodataflow/flows/flows_viz_cloud.py +++ b/src/echodataflow/flows/flows_viz_cloud.py @@ -129,7 +129,7 @@ def flow_update_cache_MVBS( @flow() -def flow_update_cache_EVR( +def flow_update_cache_contours( time_offset_seconds: float = 0.0, slice_mins: int = 180, path_cache: str = "PATH_TO_DATA_CACHE", From a981b34968e3f2e47026991ed57dc7ed7390d811 Mon Sep 17 00:00:00 2001 From: Wu-Jung Lee Date: Fri, 10 Jul 2026 21:23:24 +0000 Subject: [PATCH 4/8] better match mvbs cache --- src/echodataflow/flows/flows_viz_cloud.py | 35 +++---------------- .../services/viz_echogram_track_lasker.py | 2 +- 2 files changed, 6 insertions(+), 31 deletions(-) diff --git a/src/echodataflow/flows/flows_viz_cloud.py b/src/echodataflow/flows/flows_viz_cloud.py index 91227c5..1321eea 100644 --- a/src/echodataflow/flows/flows_viz_cloud.py +++ b/src/echodataflow/flows/flows_viz_cloud.py @@ -158,45 +158,31 @@ def flow_update_cache_contours( num_slices=1, ) - logger.info( - f"Using time range:\n" - f"- start: {start_time[0]}\n" - f"- end: {end_time[0]}" - ) - - # Connect to object storage + # 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"], - }, + 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( @@ -211,17 +197,14 @@ def flow_update_cache_contours( if len(selected_evr) == 0: logger.info( - "EVR cache not updated: no EVR files in specified time range" + "Contours cache not updated: no EVR files in specified time range" ) else: # Read EVRs and collect dataframes contours_dfs = [] - for _, evr_file in selected_evr: logger.info(f"Reading EVR: {evr_file}") - regions = er.read_evr(evr_file) - contours_dfs.append( regions.data ) @@ -233,19 +216,11 @@ def flow_update_cache_contours( ) logger.info( - f"Merged {len(df_contours)} contour regions" + f"Merged and saving {len(df_contours)} contour regions" ) - # Save CSV cache output_file = Path(path_cache) / file_contours_csv - - logger.info( - f"Saving contour cache: {output_file}" - ) - df_contours.to_csv( output_file, index=False, - ) - - logger.info("Contour cache update complete") + ) \ No newline at end of file diff --git a/src/echodataflow/services/viz_echogram_track_lasker.py b/src/echodataflow/services/viz_echogram_track_lasker.py index 4aeb3ac..fb7beb9 100644 --- a/src/echodataflow/services/viz_echogram_track_lasker.py +++ b/src/echodataflow/services/viz_echogram_track_lasker.py @@ -204,7 +204,7 @@ def scheduled_update(): { "multi_freq_echogram": multi_freq_app, "tricolor_echogram": tricolor_app, - "tricolor_echogram_with_hake_contour": tricolor_with_contour_app, + "tricolor_contour_echogram": tricolor_with_contour_app, }, port=1802, websocket_origin="*", From 5bd666030b23b94eadaa8c3f0eeb9d466441eca0 Mon Sep 17 00:00:00 2001 From: Wu-Jung Lee Date: Fri, 10 Jul 2026 22:10:28 +0000 Subject: [PATCH 5/8] add temp file to save evrs in before reading, fix time and depth reading --- src/echodataflow/flows/flows_viz_cloud.py | 38 ++++++++++++--- .../services/viz_echogram_track_lasker.py | 47 +++++++++++++------ 2 files changed, 64 insertions(+), 21 deletions(-) diff --git a/src/echodataflow/flows/flows_viz_cloud.py b/src/echodataflow/flows/flows_viz_cloud.py index 1321eea..424e12c 100644 --- a/src/echodataflow/flows/flows_viz_cloud.py +++ b/src/echodataflow/flows/flows_viz_cloud.py @@ -1,6 +1,7 @@ from pathlib import Path import datetime import configparser +import tempfile import pandas as pd import xarray as xr @@ -202,12 +203,33 @@ def flow_update_cache_contours( else: # Read EVRs and collect dataframes contours_dfs = [] - for _, evr_file in selected_evr: - logger.info(f"Reading EVR: {evr_file}") - regions = er.read_evr(evr_file) - contours_dfs.append( - regions.data - ) + + 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( @@ -218,9 +240,11 @@ def flow_update_cache_contours( 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, - ) \ No newline at end of file + ) diff --git a/src/echodataflow/services/viz_echogram_track_lasker.py b/src/echodataflow/services/viz_echogram_track_lasker.py index fb7beb9..7d2d1b9 100644 --- a/src/echodataflow/services/viz_echogram_track_lasker.py +++ b/src/echodataflow/services/viz_echogram_track_lasker.py @@ -7,6 +7,7 @@ import holoviews as hv import echoshader import pandas as pd +import numpy as np # Configure Panel to prevent automatic refreshes pn.config.autoreload = False @@ -113,9 +114,9 @@ def scheduled_update(): return plot_pane -def create_contour_overlay(): +def create_contours_overlay(): """ - Convert contours dataframe into HoloViews paths. + Convert hake contours dataframe into HoloViews paths. """ contours_df = pd.read_csv( @@ -123,25 +124,45 @@ def create_contour_overlay(): ) # Convert string representations back to arrays - contours_df["time"] = contours_df["time"].apply(ast.literal_eval) - contours_df["depth"] = contours_df["depth"].apply(ast.literal_eval) + 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() + ) + ) - hv_paths = [] + # Set to datetime + contours_df["time"] = contours_df["time"].apply( + lambda x: pd.to_datetime(x) + ) + # 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( { - "time": row["time"], - "depth": row["depth"], + "ping_time": time, + "depth": depth, } ) - contours_hv = hv.Path( hv_paths, - kdims=["time", "depth"], + kdims=["ping_time", "depth"], ).opts( color="magenta", - line_width=5, + line_width=3, ) return contours_hv @@ -166,10 +187,8 @@ def update_cache_tricolor_with_contour(): width=1200, height=600, tools=["pan", "box_zoom", "wheel_zoom", "reset"], ) - ) - - contours_hv = create_contour_overlay() - + )() + contours_hv = create_contours_overlay() return tricolor * contours_hv From d744c627776b48a9d4ab4662ba52ba2ea7342b8d Mon Sep 17 00:00:00 2001 From: Wu-Jung Lee Date: Sat, 11 Jul 2026 02:01:31 +0000 Subject: [PATCH 6/8] add avoid latest mins --- src/echodataflow/flows/flows_viz_cloud.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/echodataflow/flows/flows_viz_cloud.py b/src/echodataflow/flows/flows_viz_cloud.py index 424e12c..61e92a4 100644 --- a/src/echodataflow/flows/flows_viz_cloud.py +++ b/src/echodataflow/flows/flows_viz_cloud.py @@ -133,6 +133,9 @@ def flow_update_cache_MVBS( def flow_update_cache_contours( time_offset_seconds: float = 0.0, slice_mins: int = 180, + # `avoid_latest_minutes` subtracts from end time + # allows enough MVBS slices to accumulate prior to prediction contour showing on vis + avoid_latest_mins: int = 20, path_cache: str = "PATH_TO_DATA_CACHE", path_EVR: str = "PATH_TO_EVR_DATA_STORE", cred_file: str = "PATH_TO_CREDENTIALS_FILE", @@ -144,6 +147,7 @@ def flow_update_cache_contours( end_time = ( datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(seconds=time_offset_seconds) + - datetime.timedelta(minutes=avoid_latest_mins) ) logger.info( From e0b3827e3a0ac931ecc988225a76c37ab74c9962 Mon Sep 17 00:00:00 2001 From: Wu-Jung Lee Date: Mon, 13 Jul 2026 16:40:24 +0000 Subject: [PATCH 7/8] remove the avoid latest mins and cast to datetime64[ns] --- src/echodataflow/flows/flows_viz_cloud.py | 4 ---- src/echodataflow/services/viz_echogram_track_lasker.py | 2 +- 2 files changed, 1 insertion(+), 5 deletions(-) diff --git a/src/echodataflow/flows/flows_viz_cloud.py b/src/echodataflow/flows/flows_viz_cloud.py index 61e92a4..424e12c 100644 --- a/src/echodataflow/flows/flows_viz_cloud.py +++ b/src/echodataflow/flows/flows_viz_cloud.py @@ -133,9 +133,6 @@ def flow_update_cache_MVBS( def flow_update_cache_contours( time_offset_seconds: float = 0.0, slice_mins: int = 180, - # `avoid_latest_minutes` subtracts from end time - # allows enough MVBS slices to accumulate prior to prediction contour showing on vis - avoid_latest_mins: int = 20, path_cache: str = "PATH_TO_DATA_CACHE", path_EVR: str = "PATH_TO_EVR_DATA_STORE", cred_file: str = "PATH_TO_CREDENTIALS_FILE", @@ -147,7 +144,6 @@ def flow_update_cache_contours( end_time = ( datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(seconds=time_offset_seconds) - - datetime.timedelta(minutes=avoid_latest_mins) ) logger.info( diff --git a/src/echodataflow/services/viz_echogram_track_lasker.py b/src/echodataflow/services/viz_echogram_track_lasker.py index 7d2d1b9..bffeba2 100644 --- a/src/echodataflow/services/viz_echogram_track_lasker.py +++ b/src/echodataflow/services/viz_echogram_track_lasker.py @@ -138,7 +138,7 @@ def create_contours_overlay(): # Set to datetime contours_df["time"] = contours_df["time"].apply( - lambda x: pd.to_datetime(x) + lambda x: pd.to_datetime(x).to_numpy(dtype="datetime64[ns]") ) # Create holoviews path object From 7e7db997ef7ab9f8c09dc9688889a063f2386b77 Mon Sep 17 00:00:00 2001 From: Wu-Jung Lee Date: Mon, 13 Jul 2026 17:25:15 +0000 Subject: [PATCH 8/8] remove redundant lines --- .../services/viz_echogram_track_lasker.py | 28 ++----------------- 1 file changed, 3 insertions(+), 25 deletions(-) diff --git a/src/echodataflow/services/viz_echogram_track_lasker.py b/src/echodataflow/services/viz_echogram_track_lasker.py index bffeba2..79fe37b 100644 --- a/src/echodataflow/services/viz_echogram_track_lasker.py +++ b/src/echodataflow/services/viz_echogram_track_lasker.py @@ -168,36 +168,14 @@ def create_contours_overlay(): return contours_hv -def update_cache_tricolor_with_contour(): - """ - Load latest MVBS data and create tricolor echogram. - """ - ds_MVBS = xr.open_zarr(path_latest / "latest_MVBS.zarr") - - tricolor = ds_MVBS.eshader.echogram( - channel=[ - "WBT 987753-15 ES120-7C_ES", - "WBT 987763-15 ES38-7_ES", - "WBT 987760-15 ES18_ES", - ], - vmin=-70, - vmax=-36, - rgb_composite=True, - opts=opts.RGB( - width=1200, height=600, - tools=["pan", "box_zoom", "wheel_zoom", "reset"], - ) - )() - contours_hv = create_contours_overlay() - return tricolor * contours_hv - - def tricolor_with_contour_app(): """ Plot tricolor echogram with regular updates. """ # Create initial plot - tricolor_with_contour = update_cache_tricolor_with_contour() + 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