diff --git a/src/echodataflow/flows/flows_viz_cloud.py b/src/echodataflow/flows/flows_viz_cloud.py index 96642c9..424e12c 100644 --- a/src/echodataflow/flows/flows_viz_cloud.py +++ b/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, - ) \ No newline at end of file + ) + + +@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, + ) diff --git a/src/echodataflow/services/viz_echogram_track_lasker.py b/src/echodataflow/services/viz_echogram_track_lasker.py index c42a1a2..79fe37b 100644 --- a/src/echodataflow/services/viz_echogram_track_lasker.py +++ b/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="*",