diff --git a/pyproject.toml b/pyproject.toml index c850499..34a2cee 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -43,6 +43,7 @@ dependencies = [ "openpyxl", "pandas", "panel", + "paramiko", "prefect[shell,dask]==3.8.4", "PyYAML", "s3fs", diff --git a/src/echodataflow/deployment/flow_registry.py b/src/echodataflow/deployment/flow_registry.py index 5425e92..7ee321c 100644 --- a/src/echodataflow/deployment/flow_registry.py +++ b/src/echodataflow/deployment/flow_registry.py @@ -89,4 +89,8 @@ class FlowRegistration: ), description="Simulate realtime transect updates for testing.", ), + "fetch_spasso": FlowRegistration( + entrypoint="echodataflow/flows/flows_spasso.py:flow_fetch_spasso", + description="Incrementally fetch newly available SPASSO products over SFTP.", + ), } \ No newline at end of file diff --git a/src/echodataflow/flows/flows_spasso.py b/src/echodataflow/flows/flows_spasso.py new file mode 100644 index 0000000..2902f35 --- /dev/null +++ b/src/echodataflow/flows/flows_spasso.py @@ -0,0 +1,208 @@ +"""Prefect flows for retrieving SPASSO products.""" + +from __future__ import annotations + +import re +from pathlib import Path + +from prefect import flow, get_run_logger +from prefect.blocks.system import Secret +from prefect.futures import as_completed + +from echodataflow.deployment.task_runners import task_runner_from_environment +from echodataflow.operations.operations_spasso import ( + SpassoConnectionSettings, + SpassoDownloadSettings, +) +from echodataflow.tasks.tasks_spasso import ( + task_download_spasso_file, + task_list_spasso_files, +) + + +@flow(log_prints=True, task_runner=task_runner_from_environment()) +def flow_fetch_spasso( + host: str, + port: int = 2221, + remote_path: str = "", + path_main: str = "", + product_patterns: list[str] | None = None, + username_secret: str = "spasso-username", + password_secret: str = "spasso-password", + task_retries: int = 3, + task_retry_delay_seconds: int = 30, +) -> None: + """Download the latest available SPASSO file for each requested product.""" + + logger = get_run_logger() + + # Credentials + username = Secret.load(username_secret).get() + password = Secret.load(password_secret).get() + + # Local output directory + local_dir = Path(path_main) + local_dir.mkdir(parents=True, exist_ok=True) + + # SFTP connection settings + connection = SpassoConnectionSettings( + host=host, + port=port, + username=username, + password=password, + ) + + # List remote SPASSO files + remote_files = task_list_spasso_files.with_options( + retries=task_retries, + retry_delay_seconds=task_retry_delay_seconds, + )( + connection, + remote_path, + ) + + logger.info( + "Found %d remote SPASSO file(s).", + len(remote_files), + ) + + if not product_patterns: + raise ValueError( + "At least one SPASSO product must be provided in " + "'product_patterns'." + ) + + selected = [] + + for product in product_patterns: + + # Exact product-family matching + # This is important because "Copernicus_PHY" should match: + # + # 20260915_Copernicus_PHY.nc + # + # but NOT: + # + # 20260915_FTLE_Copernicus_PHY.nc + # 20260915_OW_Copernicus_PHY.nc + # 20260915_KE_Copernicus_PHY.nc + + pattern = re.compile( + rf"^\d{{8}}_{re.escape(product)}\.nc$" + ) + + matches = [ + item + for item in remote_files + if pattern.fullmatch(item.filename) + ] + + if not matches: + logger.warning( + "No remote SPASSO file found for product %r.", + product, + ) + continue + + # Filenames start with YYYYMMDD, so lexical ordering gives us + # the newest product date + latest = max( + matches, + key=lambda item: item.filename[:8], + ) + + logger.info( + "Latest %-25s -> %s", + product, + latest.filename, + ) + + selected.append(latest) + + if not selected: + logger.warning( + "None of the requested SPASSO products were found." + ) + return + + # Check which selected files already exist locally + # If filename AND size match, there is nothing to download + # If the file is absent or its size differs, download it + + existing = { + path.name: path.stat().st_size + for path in local_dir.iterdir() + if path.is_file() + } + + to_download = [ + item + for item in selected + if existing.get(item.filename) != item.size_bytes + ] + + logger.info( + "Selected %d latest product file(s); " + "%d require download.", + len(selected), + len(to_download), + ) + + if not to_download: + logger.info( + "All latest SPASSO products are already available locally." + ) + return + + # Download settings + settings = SpassoDownloadSettings( + local_directory=str(local_dir), + ) + + # Submit downloads concurrently through Prefect + futures = {} + + for item in to_download: + + future = task_download_spasso_file.with_options( + task_run_name=f"spasso_{item.filename}", + retries=task_retries, + retry_delay_seconds=task_retry_delay_seconds, + ).submit( + connection, + item, + settings, + ) + + futures[future] = item + + # Collect results + errors = [] + for future in as_completed(futures): + + item = futures[future] + + try: + result = future.result() + + logger.info( + "Downloaded %s -> %s", + result.remote_path, + result.local_path, + ) + + except Exception as exc: + errors.append(exc) + + logger.error( + "Failed to download %s: %s", + item.remote_path, + exc, + ) + + # Fail the flow if one or more downloads failed. + if errors: + raise RuntimeError( + f"{len(errors)} SPASSO download(s) failed " + f"out of {len(to_download)}." + ) diff --git a/src/echodataflow/operations/operations_spasso.py b/src/echodataflow/operations/operations_spasso.py new file mode 100644 index 0000000..12cc117 --- /dev/null +++ b/src/echodataflow/operations/operations_spasso.py @@ -0,0 +1,108 @@ +"""SPASSO data-access operations and data contracts.""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + +import paramiko + + +@dataclass(frozen=True) +class SpassoConnectionSettings: + """Connection settings shared by SPASSO SFTP operations.""" + + host: str + port: int = 2221 + username: str = "" + password: str = "" + + +@dataclass(frozen=True) +class SpassoDownloadSettings: + """Local settings for SPASSO downloads.""" + + local_directory: str + + +@dataclass(frozen=True) +class SpassoRemoteFile: + """Metadata for one remote SPASSO file.""" + + filename: str + remote_path: str + size_bytes: int + mtime: int + + +@dataclass(frozen=True) +class SpassoDownloadResult: + """Metadata describing one successfully downloaded SPASSO file.""" + + filename: str + remote_path: str + local_path: str + size_bytes: int + + +def _open_sftp( + connection: SpassoConnectionSettings, +) -> tuple[paramiko.Transport, paramiko.SFTPClient]: + """Open an authenticated SFTP connection.""" + transport = paramiko.Transport((connection.host, connection.port)) + transport.connect( + username=connection.username, + password=connection.password, + ) + sftp = paramiko.SFTPClient.from_transport(transport) + return transport, sftp + + +def list_spasso_files( + connection: SpassoConnectionSettings, + remote_directory: str, +) -> list[SpassoRemoteFile]: + """List regular files in one SPASSO remote directory.""" + transport, sftp = _open_sftp(connection) + try: + rows: list[SpassoRemoteFile] = [] + for entry in sftp.listdir_attr(remote_directory): + remote_path = f"{remote_directory.rstrip('/')}/{entry.filename}" + rows.append( + SpassoRemoteFile( + filename=entry.filename, + remote_path=remote_path, + size_bytes=int(entry.st_size), + mtime=int(entry.st_mtime), + ) + ) + return rows + finally: + sftp.close() + transport.close() + + +def download_spasso_file( + connection: SpassoConnectionSettings, + item: SpassoRemoteFile, + settings: SpassoDownloadSettings, +) -> SpassoDownloadResult: + """Download one SPASSO file to the configured local directory.""" + local_dir = Path(settings.local_directory) + local_dir.mkdir(parents=True, exist_ok=True) + + local_path = local_dir / item.filename + + transport, sftp = _open_sftp(connection) + try: + sftp.get(item.remote_path, str(local_path)) + finally: + sftp.close() + transport.close() + + return SpassoDownloadResult( + filename=item.filename, + remote_path=item.remote_path, + local_path=str(local_path), + size_bytes=local_path.stat().st_size, + ) diff --git a/src/echodataflow/services/spasso/__init__.py b/src/echodataflow/services/spasso/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/echodataflow/services/spasso/app.py b/src/echodataflow/services/spasso/app.py new file mode 100644 index 0000000..31832e6 --- /dev/null +++ b/src/echodataflow/services/spasso/app.py @@ -0,0 +1,279 @@ +"""Panel application for live SPASSO visualization.""" + +from __future__ import annotations + +from pathlib import Path + +import cartopy +import holoviews as hv +import pandas as pd +import panel as pn +from holoviews.streams import Pipe + +from echodataflow.services.spasso.data import load_navigation +from echodataflow.services.spasso.layers import ( + add_navigation, + geographic_bounds, + navigation_layer, +) +from echodataflow.services.spasso.plotting import ( + create_currents_plot, + create_product_plot, +) + + +def create_spasso_app(config: dict): + """Create the live SPASSO dashboard from configuration.""" + + # Configuration + dashboard_config = config["dashboard"] + products_config = config["products"] + currents_config = config.get("currents") + + spasso_dir = Path( + dashboard_config["spasso_path"] + ) + + navigation_dir = Path( + dashboard_config["navigation_path"] + ) + + cartopy_dir = Path( + dashboard_config["cartopy_path"] + ) + + cartopy_dir.mkdir( + parents=True, + exist_ok=True, + ) + + cartopy.config["data_dir"] = str( + cartopy_dir + ) + + map_radius_km = dashboard_config.get( + "map_radius_km", + 500, + ) + + navigation_history_hours = float( + dashboard_config.get( + "navigation_history_hours", + 6, + ) + ) + + refresh_seconds = dashboard_config.get( + "refresh_seconds", + 60, + ) + + plot_config = dashboard_config.get( + "plot", + {}, + ) + + plot_width = plot_config.get( + "width", + 520, + ) + + plot_height = plot_config.get( + "height", + 400, + ) + + # Initial navigation + # Load the complete available track. navigation_history_hours is + # used only to control which part is highlighted as "recent" + navigation = load_navigation( + navigation_dir=navigation_dir, + ) + + latest = navigation.iloc[-1] + + ship_lat = float( + latest["latitude"] + ) + + ship_lon = float( + latest["longitude"] + ) + + bounds = geographic_bounds( + latitude=ship_lat, + longitude=ship_lon, + radius_km=map_radius_km, + ) + + # Shared dynamic navigation + navigation_pipe = Pipe( + data=navigation + ) + + def make_navigation_layer(data): + """Render navigation using the configured recent-track window.""" + + return navigation_layer( + data=data, + history_hours=navigation_history_hours, + ) + + dynamic_navigation = hv.DynamicMap( + make_navigation_layer, + streams=[navigation_pipe], + ) + + + # SPASSO products + # These are deliberately created ONCE + # The periodic callback below only updates navigation + plots_by_tab: dict[str, list] = {} + + for product_name, product_config in products_config.items(): + + plot = create_product_plot( + product_name=product_name, + product_config=product_config, + spasso_dir=spasso_dir, + bounds=bounds, + width=plot_width, + height=plot_height, + ) + + plot = add_navigation( + plot, + dynamic_navigation, + ) + + tab_name = product_config.get( + "tab", + "Products", + ) + + plots_by_tab.setdefault( + tab_name, + [], + ).append(plot) + + # Surface currents + if currents_config is not None: + + currents_plot = create_currents_plot( + currents_config=currents_config, + spasso_dir=spasso_dir, + bounds=bounds, + width=plot_width, + height=plot_height, + ) + + currents_plot = add_navigation( + currents_plot, + dynamic_navigation, + ) + + tab_name = currents_config.get( + "tab", + "Currents / Diagnostics", + ) + + # Put currents first in its tab, matching the previous dashboard + plots_by_tab.setdefault( + tab_name, + [], + ).insert( + 0, + currents_plot, + ) + + # Status + status = pn.pane.Markdown( + ( + f"**Ship position:** " + f"{latest['latitude']:.5f}, " + f"{latest['longitude']:.5f} \n" + f"**Navigation time:** " + f"{latest['timestamp_utc']:%Y-%m-%d %H:%M:%S UTC}" + ) + ) + + # Tabs + tabs_content = [] + + for tab_name, plots in plots_by_tab.items(): + + grid = pn.GridBox( + *plots, + ncols=2, + sizing_mode="stretch_width", + ) + + tabs_content.append( + ( + tab_name, + grid, + ) + ) + + tabs = pn.Tabs( + *tabs_content, + dynamic=True, + sizing_mode="stretch_width", + ) + + # Navigation refresh + # IMPORTANT: Only navigation is reloaded here + # SPASSO NetCDF files and rasterized products are NOT recreated + # This preserves the current zoom/pan behavior and avoids repeatedly + # processing the heavy gridded products + def scheduled_update(): + """Update only the ship-navigation layer.""" + + try: + new_navigation = load_navigation( + navigation_dir=navigation_dir, + ) + + new_latest = new_navigation.iloc[-1] + + navigation_pipe.send( + new_navigation + ) + + status.object = ( + f"**Ship position:** " + f"{new_latest['latitude']:.5f}, " + f"{new_latest['longitude']:.5f} \n" + f"**Navigation time:** " + f"{new_latest['timestamp_utc']:%Y-%m-%d %H:%M:%S UTC} \n" + f"**Dashboard refresh:** " + f"{pd.Timestamp.now(tz='UTC'):%H:%M:%S UTC}" + ) + + except Exception as exc: + + status.object = ( + f"**Navigation update error:** `{exc}`" + ) + + print( + f"SPASSO navigation update error: {exc}" + ) + + pn.state.add_periodic_callback( + scheduled_update, + period=refresh_seconds * 1000, + ) + + # Dashboard + title = dashboard_config.get( + "title", + "SPASSO / Ship Navigation", + ) + + return pn.Column( + f"# {title}", + status, + tabs, + sizing_mode="stretch_width", + ) diff --git a/src/echodataflow/services/spasso/config.py b/src/echodataflow/services/spasso/config.py new file mode 100644 index 0000000..8f51e0c --- /dev/null +++ b/src/echodataflow/services/spasso/config.py @@ -0,0 +1,39 @@ +"""Configuration loading for SPASSO visualization services.""" + +from __future__ import annotations + +from pathlib import Path + +import yaml + + +def load_spasso_viz_config( + config_path: str | Path, +) -> dict: + """Load a SPASSO visualization configuration.""" + + config_path = Path(config_path) + + with config_path.open( + "r", + encoding="utf-8", + ) as file: + config = yaml.safe_load(file) + + if not isinstance(config, dict): + raise ValueError( + f"SPASSO visualization config must be a mapping: " + f"{config_path}" + ) + + if "dashboard" not in config: + raise ValueError( + f"Missing 'dashboard' section in {config_path}" + ) + + if "products" not in config: + raise ValueError( + f"Missing 'products' section in {config_path}" + ) + + return config \ No newline at end of file diff --git a/src/echodataflow/services/spasso/data.py b/src/echodataflow/services/spasso/data.py new file mode 100644 index 0000000..5de589c --- /dev/null +++ b/src/echodataflow/services/spasso/data.py @@ -0,0 +1,205 @@ +"""Data access and preparation for SPASSO visualizations.""" + +from __future__ import annotations + +from pathlib import Path + +import numpy as np +import pandas as pd +import xarray as xr + + +def load_navigation( + navigation_dir: Path, +) -> pd.DataFrame: + """Load all available ship-navigation Parquet files.""" + + files = sorted( + navigation_dir.glob("ship_navigation_*.parquet") + ) + + if not files: + raise FileNotFoundError( + f"No navigation Parquet files found in {navigation_dir}" + ) + + navigation = pd.concat( + [pd.read_parquet(path) for path in files], + ignore_index=True, + ) + + navigation["timestamp_utc"] = pd.to_datetime( + navigation["timestamp_utc"], + utc=True, + ) + + navigation = ( + navigation + .sort_values("timestamp_utc") + .drop_duplicates(subset="timestamp_utc") + .reset_index(drop=True) + ) + + return navigation + + +def find_latest_product_file( + spasso_dir: Path, + patterns: list[str], +) -> Path | None: + """Find the newest local file matching any product pattern.""" + + files: list[Path] = [] + + for pattern in patterns: + files.extend(spasso_dir.glob(pattern)) + + files = list(set(files)) + + if not files: + return None + + return max( + files, + key=lambda path: path.stat().st_mtime, + ) + + +def find_variable( + ds: xr.Dataset, + candidates: list[str], +) -> str | None: + """Find the first matching variable in a dataset.""" + + for variable in candidates: + if variable in ds.data_vars: + return variable + + return None + + +def get_coordinate_values( + ds: xr.Dataset, +) -> tuple[np.ndarray, np.ndarray] | None: + """Find latitude and longitude values in a SPASSO dataset.""" + + lat = next( + ( + ds[name].values + for name in ("lats", "lat", "latitude") + if name in ds.variables + ), + None, + ) + + lon = next( + ( + ds[name].values + for name in ("lons", "lon", "longitude") + if name in ds.variables + ), + None, + ) + + if lat is None or lon is None: + return None + + return np.asarray(lat), np.asarray(lon) + + +def spatial_subset( + data: xr.DataArray, + lon_min: float, + lon_max: float, + lat_min: float, + lat_max: float, +) -> xr.DataArray: + """Subset a DataArray regardless of coordinate direction.""" + + lon = data["lon"] + lat = data["lat"] + + lon_slice = ( + slice(lon_min, lon_max) + if lon.values[0] < lon.values[-1] + else slice(lon_max, lon_min) + ) + + lat_slice = ( + slice(lat_min, lat_max) + if lat.values[0] < lat.values[-1] + else slice(lat_max, lat_min) + ) + + return data.sel( + lon=lon_slice, + lat=lat_slice, + ) + + +def prepare_data_array( + ds: xr.Dataset, + variable: str, +) -> xr.DataArray: + """Prepare a SPASSO variable for geographic plotting.""" + + data = ds[variable] + + if "time" in data.dims: + data = data.isel(time=0) + + if "depth" in data.dims and data.sizes["depth"] == 1: + data = data.isel(depth=0) + + coordinates = get_coordinate_values(ds) + + if coordinates is None: + raise ValueError( + "Could not identify latitude/longitude coordinates." + ) + + lat_values, lon_values = coordinates + + lat_dim = next( + ( + dim + for dim in data.dims + if dim.lower() in {"lat", "latitude", "y"} + ), + None, + ) + + lon_dim = next( + ( + dim + for dim in data.dims + if dim.lower() in {"lon", "longitude", "x"} + ), + None, + ) + + if lat_dim is None or lon_dim is None: + raise ValueError( + f"Could not identify geographic dimensions for {variable}. " + f"Dimensions are {data.dims}." + ) + + data = data.assign_coords( + { + lat_dim: lat_values, + lon_dim: lon_values, + } + ) + + rename = {} + + if lat_dim != "lat": + rename[lat_dim] = "lat" + + if lon_dim != "lon": + rename[lon_dim] = "lon" + + if rename: + data = data.rename(rename) + + return data \ No newline at end of file diff --git a/src/echodataflow/services/spasso/layers.py b/src/echodataflow/services/spasso/layers.py new file mode 100644 index 0000000..e9066f9 --- /dev/null +++ b/src/echodataflow/services/spasso/layers.py @@ -0,0 +1,189 @@ +"""Geographic and navigation layers for SPASSO visualizations.""" + +from __future__ import annotations + +import math + +import cartopy.crs as ccrs +import cartopy.feature as cfeature +import geoviews as gv +import pandas as pd +import panel as pn + + +CRS = ccrs.PlateCarree() +def geographic_bounds( + latitude: float, + longitude: float, + radius_km: float, +) -> tuple[float, float, float, float]: + """Return approximate geographic bounds around a position.""" + + delta_lat = radius_km / 111.0 + + delta_lon = radius_km / ( + 111.0 * math.cos(math.radians(latitude)) + ) + + return ( + longitude - delta_lon, + longitude + delta_lon, + latitude - delta_lat, + latitude + delta_lat, + ) + + +def create_geographic_layers(): + """Create reusable land/coastline/state layers.""" + + land = gv.feature.land.opts( + fill_color="lightgray", + line_color="black", + ) + + coastline = gv.feature.coastline.opts( + line_color="black", + line_width=1, + ) + + states_feature = cfeature.NaturalEarthFeature( + category="cultural", + name="admin_1_states_provinces_lines", + scale="50m", + facecolor="none", + ) + + states = gv.Feature(states_feature).opts( + line_color="gray", + line_width=0.6, + ) + + return land * coastline * states + + +def navigation_layer( + data, + history_hours: float = 6, +): + """ + Create complete and recent ship-track layers. + + The complete available track is shown as a thin dashed line. + The most recent ``history_hours`` are shown as a thicker solid line. + """ + + if data is None or data.empty: + return gv.Overlay([]) + + navigation = ( + data + .sort_values("timestamp_utc") + .reset_index(drop=True) + ) + + latest = navigation.iloc[-1] + + ship_lon = float( + latest["longitude"] + ) + + ship_lat = float( + latest["latitude"] + ) + + # Complete ship track + all_coordinates = navigation[ + ["longitude", "latitude"] + ].to_numpy() + + full_track = gv.Path( + [all_coordinates], + crs=CRS, + ).opts( + line_width=1, + line_dash="dashed", + color="deepskyblue", + alpha=0.6, + ) + + # Recent ship track + latest_time = navigation[ + "timestamp_utc" + ].max() + + cutoff = latest_time - pd.Timedelta( + hours=history_hours + ) + + recent = navigation.loc[ + navigation["timestamp_utc"] >= cutoff + ].copy() + + # Include the final historical point so that the recent track + # connects cleanly to the complete track + older = navigation.loc[ + navigation["timestamp_utc"] < cutoff + ] + + if not older.empty: + recent = pd.concat( + [ + older.tail(1), + recent, + ], + ignore_index=True, + ) + + recent_coordinates = recent[ + ["longitude", "latitude"] + ].to_numpy() + + recent_track = gv.Path( + [recent_coordinates], + crs=CRS, + ).opts( + line_width=3, + color="deepskyblue", + ) + + # Current ship position + ship = gv.Points( + [ + ( + ship_lon, + ship_lat, + ) + ], + kdims=[ + "longitude", + "latitude", + ], + crs=CRS, + ).opts( + size=12, + marker="triangle", + color="deepskyblue", + tools=["hover"], + ) + + return ( + full_track + * recent_track + * ship + ) + + +def add_navigation( + plot, + dynamic_navigation, +): + """Overlay live navigation on a geographic plot.""" + + if isinstance(plot, pn.viewable.Viewable): + return plot + + return ( + plot * dynamic_navigation + ).opts( + framewise=False + ) diff --git a/src/echodataflow/services/spasso/plotting.py b/src/echodataflow/services/spasso/plotting.py new file mode 100644 index 0000000..b7924c5 --- /dev/null +++ b/src/echodataflow/services/spasso/plotting.py @@ -0,0 +1,518 @@ +"""Plotting functions for SPASSO visualization services.""" + +from __future__ import annotations + +from pathlib import Path + +import geoviews as gv +import numpy as np +import panel as pn +import xarray as xr + +from holoviews.operation.datashader import rasterize + +from echodataflow.services.spasso.data import ( + find_latest_product_file, + find_variable, + prepare_data_array, + spatial_subset, +) +from echodataflow.services.spasso.layers import ( + CRS, + create_geographic_layers, +) + + +def create_raster_plot( + data, + *, + variable, + title, + cmap, + clim, + width, + height, +): + """Render a geographic DataArray using Datashader.""" + + quadmesh = gv.QuadMesh( + data, + kdims=["lon", "lat"], + vdims=[variable], + crs=CRS, + ) + + return rasterize( + quadmesh, + width=width, + height=height, + ).opts( + cmap=cmap, + clim=clim, + colorbar=True, + width=width, + height=height, + xlabel="Longitude", + ylabel="Latitude", + tools=[ + "hover", + "pan", + "wheel_zoom", + "box_zoom", + "reset", + ], + active_tools=["wheel_zoom"], + framewise=False, + title=title, + ) + + +def _apply_transform( + data: xr.DataArray, + transform: str | None, +) -> xr.DataArray: + """Apply a configured transformation to a product.""" + + if transform is None: + return data + + if transform == "kelvin_to_celsius": + return data - 273.15 + + raise ValueError( + f"Unknown SPASSO visualization transform: {transform}" + ) + + +def _compute_clim( + data: xr.DataArray, + config: dict, +) -> tuple[float, float]: + """Determine plotting color limits.""" + + configured_clim = config.get("clim") + + if configured_clim is not None: + return ( + float(configured_clim[0]), + float(configured_clim[1]), + ) + + values = np.asarray(data.values) + finite = values[np.isfinite(values)] + + if finite.size == 0: + return (0.0, 1.0) + + vmin = float(finite.min()) + vmax = float(finite.max()) + + if config.get("zero_min", False): + vmin = 0.0 + + if vmin == vmax: + vmax = vmin + 1.0 + + return vmin, vmax + + +def _product_time( + data: xr.DataArray, +) -> str | None: + """Extract a readable product time when available.""" + + if "time" not in data.coords: + return None + + values = np.asarray( + data["time"].values + ).reshape(-1) + + if values.size == 0: + return None + + try: + value = np.datetime_as_string( + values[0], + unit="m", + ) + except (TypeError, ValueError): + return str(values[0]) + + return value.replace( + "T", + " ", + ) + + +def create_product_plot( + *, + product_name: str, + product_config: dict, + spasso_dir: Path, + bounds, + width: int, + height: int, +): + """Create one configured SPASSO product plot.""" + + patterns = product_config.get( + "patterns", + [], + ) + + product_file = find_latest_product_file( + spasso_dir, + patterns, + ) + + title = product_config.get( + "title", + product_name, + ) + + if product_file is None: + return pn.pane.Markdown( + ( + f"### {title}\n\n" + "No matching SPASSO product found." + ), + width=width, + height=height, + ) + + try: + with xr.open_dataset( + product_file + ) as ds: + + variable = find_variable( + ds, + product_config["variables"], + ) + + data = prepare_data_array( + ds, + variable, + ) + + data = _apply_transform( + data, + product_config.get( + "transform" + ), + ) + + lon_min, lon_max, lat_min, lat_max = bounds + + data = spatial_subset( + data, + lon_min, + lon_max, + lat_min, + lat_max, + ) + + # Load before closing the NetCDF file. + data = data.load() + + clim = _compute_clim( + data, + product_config, + ) + + product_time = _product_time( + data + ) + + plot_title = title + + if product_time is not None: + plot_title += ( + f" — {product_time}" + ) + + raster = create_raster_plot( + data, + variable=variable, + title=plot_title, + cmap=product_config.get( + "cmap", + "Viridis", + ), + clim=clim, + width=width, + height=height, + ) + + geographic_layers = ( + create_geographic_layers() + ) + + return ( + raster + * geographic_layers + ) + + except Exception as exc: + return pn.pane.Markdown( + ( + f"### {title}\n\n" + f"Could not load `{product_file.name}`.\n\n" + f"`{exc}`" + ), + width=width, + height=height, + ) + + +def create_currents_plot( + *, + currents_config: dict, + spasso_dir: Path, + bounds, + width: int, + height: int, +): + """Create surface-current speed and vector visualization.""" + + title = currents_config.get( + "title", + "Surface Geostrophic Currents", + ) + + pattern = currents_config.get( + "pattern", + "????????_Copernicus_PHY.nc", + ) + + product_file = find_latest_product_file( + spasso_dir, + [pattern], + ) + + if product_file is None: + return pn.pane.Markdown( + ( + f"### {title}\n\n" + "No matching physical-oceanography " + "product found." + ), + width=width, + height=height, + ) + + try: + with xr.open_dataset( + product_file + ) as ds: + + u_name = find_variable( + ds, + [ + currents_config.get( + "u_variable", + "ugos", + ) + ], + ) + + v_name = find_variable( + ds, + [ + currents_config.get( + "v_variable", + "vgos", + ) + ], + ) + + u = prepare_data_array( + ds, + u_name, + ) + + v = prepare_data_array( + ds, + v_name, + ) + + lon_min, lon_max, lat_min, lat_max = bounds + + u = spatial_subset( + u, + lon_min, + lon_max, + lat_min, + lat_max, + ) + + v = spatial_subset( + v, + lon_min, + lon_max, + lat_min, + lat_max, + ) + + u, v = xr.align( + u, + v, + join="inner", + ) + + speed = np.sqrt( + u**2 + v**2 + ) + + speed.name = ( + "current_speed" + ) + + u = u.load() + v = v.load() + speed = speed.load() + + clim = _compute_clim( + speed, + currents_config, + ) + + product_time = _product_time( + speed + ) + + plot_title = title + + if product_time is not None: + plot_title += ( + f" — {product_time}" + ) + + raster = create_raster_plot( + speed, + variable="current_speed", + title=plot_title, + cmap=currents_config.get( + "cmap", + "Viridis", + ), + clim=clim, + width=width, + height=height, + ) + + result = raster + + vector_config = ( + currents_config.get( + "vectors", + {}, + ) + ) + + if vector_config.get( + "enabled", + True, + ): + + step = int( + vector_config.get( + "step", + 2, + ) + ) + + scale = float( + vector_config.get( + "scale", + 0.5, + ) + ) + + u_vector = u.isel( + lat=slice(None, None, step), + lon=slice(None, None, step), + ) + + v_vector = v.isel( + lat=slice(None, None, step), + lon=slice(None, None, step), + ) + + lon, lat = np.meshgrid( + u_vector["lon"].values, + u_vector["lat"].values, + ) + + u_values = np.asarray( + u_vector.values + ) + + v_values = np.asarray( + v_vector.values + ) + + angle = np.arctan2( + v_values, + u_values, + ) + + magnitude = np.sqrt( + u_values**2 + + v_values**2 + ) + + valid = ( + np.isfinite(lon) + & np.isfinite(lat) + & np.isfinite(angle) + & np.isfinite(magnitude) + ) + + vectors = gv.VectorField( + ( + lon[valid], + lat[valid], + angle[valid], + magnitude[valid], + ), + kdims=[ + "Longitude", + "Latitude", + ], + vdims=[ + "Angle", + "Magnitude", + ], + crs=CRS, + ).opts( + magnitude="Magnitude", + scale=scale, + pivot="mid", + ) + + result = ( + result + * vectors + ) + + geographic_layers = ( + create_geographic_layers() + ) + + return ( + result + * geographic_layers + ) + + except Exception as exc: + return pn.pane.Markdown( + ( + f"### {title}\n\n" + f"Could not load `{product_file.name}`.\n\n" + f"`{exc}`" + ), + width=width, + height=height, + ) \ No newline at end of file diff --git a/src/echodataflow/services/spasso/run.py b/src/echodataflow/services/spasso/run.py new file mode 100644 index 0000000..4fd1c7b --- /dev/null +++ b/src/echodataflow/services/spasso/run.py @@ -0,0 +1,58 @@ +"""Run the SPASSO visualization service.""" + +from __future__ import annotations + +import argparse +from pathlib import Path + +import geoviews as gv +import holoviews as hv +import panel as pn + +from echodataflow.services.spasso.app import create_spasso_app +from echodataflow.services.spasso.config import load_spasso_viz_config + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Run the SPASSO visualization dashboard." + ) + + parser.add_argument( + "--config", + required=True, + type=Path, + help="Path to the SPASSO visualization YAML configuration.", + ) + + return parser.parse_args() + + +def main() -> None: + args = parse_args() + + config = load_spasso_viz_config(args.config) + + hv.extension("bokeh") + gv.extension("bokeh") + pn.extension( + loading_indicator=True, + loading_spinner="dots", + ) + + app = lambda: create_spasso_app(config) + + pn.serve( + {"spasso": app}, + port=1804, + websocket_origin="*", + admin=True, + show=False, + autoreload=False, + keep_alive=40000, + check_unused_sessions_milliseconds=30000, + ) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/src/echodataflow/tasks/tasks_spasso.py b/src/echodataflow/tasks/tasks_spasso.py new file mode 100644 index 0000000..0a20524 --- /dev/null +++ b/src/echodataflow/tasks/tasks_spasso.py @@ -0,0 +1,38 @@ +"""Reusable Prefect tasks for SPASSO data access.""" + +from prefect import task + +from echodataflow.operations.operations_spasso import ( + SpassoConnectionSettings, + SpassoDownloadResult, + SpassoDownloadSettings, + SpassoRemoteFile, + download_spasso_file, + list_spasso_files, +) + + +@task(log_prints=True) +def task_list_spasso_files( + connection: SpassoConnectionSettings, + remote_directory: str, +) -> list[SpassoRemoteFile]: + """List files available in a SPASSO remote directory.""" + return list_spasso_files( + connection=connection, + remote_directory=remote_directory, + ) + + +@task(log_prints=True) +def task_download_spasso_file( + connection: SpassoConnectionSettings, + item: SpassoRemoteFile, + settings: SpassoDownloadSettings, +) -> SpassoDownloadResult: + """Download one SPASSO file as a Prefect task.""" + return download_spasso_file( + connection=connection, + item=item, + settings=settings, + ) diff --git a/tests/test_flows_spasso.py b/tests/test_flows_spasso.py new file mode 100644 index 0000000..6e81c78 --- /dev/null +++ b/tests/test_flows_spasso.py @@ -0,0 +1,516 @@ +from unittest.mock import MagicMock + +import pytest + +import echodataflow.flows.flows_spasso as flows_spasso +from echodataflow.operations.operations_spasso import ( + SpassoDownloadResult, + SpassoRemoteFile, +) + + +def remote_file( + filename: str, + size: int = 100, +) -> SpassoRemoteFile: + """Create a remote SPASSO file for testing.""" + + return SpassoRemoteFile( + filename=filename, + remote_path=f"remote/{filename}", + size_bytes=size, + mtime=1000, + ) + + +@pytest.fixture +def mock_prefect(monkeypatch): + """Mock Prefect runtime dependencies.""" + + secret = MagicMock() + secret.get.return_value = "test-value" + + monkeypatch.setattr( + flows_spasso.Secret, + "load", + MagicMock(return_value=secret), + ) + + logger = MagicMock() + + monkeypatch.setattr( + flows_spasso, + "get_run_logger", + MagicMock(return_value=logger), + ) + + return logger + + +def configure_remote_listing( + monkeypatch, + files, +): + """Mock the Prefect task used to list remote files.""" + + list_task = MagicMock() + configured_task = MagicMock() + + list_task.with_options.return_value = configured_task + configured_task.return_value = files + + monkeypatch.setattr( + flows_spasso, + "task_list_spasso_files", + list_task, + ) + + return list_task + + +def test_requires_product_patterns( + monkeypatch, + mock_prefect, + tmp_path, +): + """At least one requested product family is required.""" + + configure_remote_listing( + monkeypatch, + [], + ) + + with pytest.raises( + ValueError, + match="At least one SPASSO product", + ): + flows_spasso.flow_fetch_spasso.fn( + host="example.org", + path_main=str(tmp_path), + product_patterns=[], + ) + + +def test_selects_latest_file_for_each_product( + monkeypatch, + mock_prefect, + tmp_path, +): + """Latest products are selected independently by date.""" + + files = [ + remote_file( + "20260915_Copernicus_PHY.nc" + ), + remote_file( + "20260917_Copernicus_PHY.nc" + ), + remote_file( + "20260914_Copernicus_SST_L4.nc" + ), + remote_file( + "20260916_Copernicus_SST_L4.nc" + ), + ] + + configure_remote_listing( + monkeypatch, + files, + ) + + submitted = [] + + download_task = MagicMock() + configured_download = MagicMock() + + download_task.with_options.return_value = ( + configured_download + ) + + def fake_submit(connection, item, settings): + submitted.append(item) + + future = MagicMock() + future.result.return_value = SpassoDownloadResult( + filename=item.filename, + remote_path=item.remote_path, + local_path=str( + tmp_path / item.filename + ), + size_bytes=item.size_bytes, + ) + + return future + + configured_download.submit.side_effect = fake_submit + + monkeypatch.setattr( + flows_spasso, + "task_download_spasso_file", + download_task, + ) + + monkeypatch.setattr( + flows_spasso, + "as_completed", + lambda futures: list(futures), + ) + + flows_spasso.flow_fetch_spasso.fn( + host="example.org", + path_main=str(tmp_path), + product_patterns=[ + "Copernicus_PHY", + "Copernicus_SST_L4", + ], + task_retries=0, + ) + + assert { + item.filename + for item in submitted + } == { + "20260917_Copernicus_PHY.nc", + "20260916_Copernicus_SST_L4.nc", + } + + +def test_product_matching_is_exact( + monkeypatch, + mock_prefect, + tmp_path, +): + """ + Copernicus_PHY must not match FTLE, KE, OW, + or LLADV products. + """ + + files = [ + remote_file( + "20260915_Copernicus_PHY.nc" + ), + remote_file( + "20260917_FTLE_Copernicus_PHY.nc" + ), + remote_file( + "20260917_KE_Copernicus_PHY.nc" + ), + remote_file( + "20260917_OW_Copernicus_PHY.nc" + ), + remote_file( + "20260917_LLADV_Copernicus_PHY.nc" + ), + ] + + configure_remote_listing( + monkeypatch, + files, + ) + + submitted = [] + + download_task = MagicMock() + configured_download = MagicMock() + + download_task.with_options.return_value = ( + configured_download + ) + + def fake_submit(connection, item, settings): + submitted.append(item) + + future = MagicMock() + future.result.return_value = SpassoDownloadResult( + filename=item.filename, + remote_path=item.remote_path, + local_path=str( + tmp_path / item.filename + ), + size_bytes=item.size_bytes, + ) + + return future + + configured_download.submit.side_effect = fake_submit + + monkeypatch.setattr( + flows_spasso, + "task_download_spasso_file", + download_task, + ) + + monkeypatch.setattr( + flows_spasso, + "as_completed", + lambda futures: list(futures), + ) + + flows_spasso.flow_fetch_spasso.fn( + host="example.org", + path_main=str(tmp_path), + product_patterns=["Copernicus_PHY"], + task_retries=0, + ) + + assert len(submitted) == 1 + assert ( + submitted[0].filename + == "20260915_Copernicus_PHY.nc" + ) + + +def test_existing_file_with_same_size_is_skipped( + monkeypatch, + mock_prefect, + tmp_path, +): + """Do not download a file already present at the same size.""" + + filename = "20260917_Copernicus_PHY.nc" + + local_file = tmp_path / filename + local_file.write_bytes(b"12345") + + files = [ + remote_file( + filename, + size=5, + ) + ] + + configure_remote_listing( + monkeypatch, + files, + ) + + download_task = MagicMock() + + monkeypatch.setattr( + flows_spasso, + "task_download_spasso_file", + download_task, + ) + + flows_spasso.flow_fetch_spasso.fn( + host="example.org", + path_main=str(tmp_path), + product_patterns=["Copernicus_PHY"], + task_retries=0, + ) + + download_task.with_options.assert_not_called() + + +def test_existing_file_with_different_size_is_downloaded( + monkeypatch, + mock_prefect, + tmp_path, +): + """Redownload an existing file when its size differs.""" + + filename = "20260917_Copernicus_PHY.nc" + + local_file = tmp_path / filename + local_file.write_bytes(b"old") + + files = [ + remote_file( + filename, + size=100, + ) + ] + + configure_remote_listing( + monkeypatch, + files, + ) + + download_task = MagicMock() + configured_download = MagicMock() + + download_task.with_options.return_value = ( + configured_download + ) + + future = MagicMock() + future.result.return_value = SpassoDownloadResult( + filename=filename, + remote_path=f"remote/{filename}", + local_path=str(local_file), + size_bytes=100, + ) + + configured_download.submit.return_value = future + + monkeypatch.setattr( + flows_spasso, + "task_download_spasso_file", + download_task, + ) + + monkeypatch.setattr( + flows_spasso, + "as_completed", + lambda futures: list(futures), + ) + + flows_spasso.flow_fetch_spasso.fn( + host="example.org", + path_main=str(tmp_path), + product_patterns=["Copernicus_PHY"], + task_retries=0, + ) + + configured_download.submit.assert_called_once() + + +def test_missing_product_does_not_fail_flow( + monkeypatch, + mock_prefect, + tmp_path, +): + """A product not yet available remotely is skipped.""" + + configure_remote_listing( + monkeypatch, + [ + remote_file( + "20260917_Copernicus_PHY.nc" + ) + ], + ) + + download_task = MagicMock() + configured_download = MagicMock() + + download_task.with_options.return_value = ( + configured_download + ) + + future = MagicMock() + future.result.return_value = SpassoDownloadResult( + filename="20260917_Copernicus_PHY.nc", + remote_path=( + "remote/20260917_Copernicus_PHY.nc" + ), + local_path=str( + tmp_path / "20260917_Copernicus_PHY.nc" + ), + size_bytes=100, + ) + + configured_download.submit.return_value = future + + monkeypatch.setattr( + flows_spasso, + "task_download_spasso_file", + download_task, + ) + + monkeypatch.setattr( + flows_spasso, + "as_completed", + lambda futures: list(futures), + ) + + flows_spasso.flow_fetch_spasso.fn( + host="example.org", + path_main=str(tmp_path), + product_patterns=[ + "Copernicus_PHY", + "Copernicus_SSS_L4", + ], + task_retries=0, + ) + + configured_download.submit.assert_called_once() + + +def test_no_requested_products_found_returns_cleanly( + monkeypatch, + mock_prefect, + tmp_path, +): + """No matching remote products is a valid no-op.""" + + configure_remote_listing( + monkeypatch, + [ + remote_file( + "20260917_SOMETHING_ELSE.nc" + ) + ], + ) + + download_task = MagicMock() + + monkeypatch.setattr( + flows_spasso, + "task_download_spasso_file", + download_task, + ) + + result = flows_spasso.flow_fetch_spasso.fn( + host="example.org", + path_main=str(tmp_path), + product_patterns=["Copernicus_PHY"], + task_retries=0, + ) + + assert result is None + download_task.with_options.assert_not_called() + + +def test_download_failure_fails_flow( + monkeypatch, + mock_prefect, + tmp_path, +): + """A failed download must make the flow fail.""" + + filename = "20260917_Copernicus_PHY.nc" + + configure_remote_listing( + monkeypatch, + [ + remote_file(filename) + ], + ) + + download_task = MagicMock() + configured_download = MagicMock() + + download_task.with_options.return_value = ( + configured_download + ) + + future = MagicMock() + future.result.side_effect = RuntimeError( + "download failed" + ) + + configured_download.submit.return_value = future + + monkeypatch.setattr( + flows_spasso, + "task_download_spasso_file", + download_task, + ) + + monkeypatch.setattr( + flows_spasso, + "as_completed", + lambda futures: list(futures), + ) + + with pytest.raises( + RuntimeError, + match="1 SPASSO download", + ): + flows_spasso.flow_fetch_spasso.fn( + host="example.org", + path_main=str(tmp_path), + product_patterns=["Copernicus_PHY"], + task_retries=0, + ) diff --git a/tests/test_operations_spasso.py b/tests/test_operations_spasso.py new file mode 100644 index 0000000..58e26ec --- /dev/null +++ b/tests/test_operations_spasso.py @@ -0,0 +1,212 @@ +from types import SimpleNamespace +from unittest.mock import MagicMock + +from echodataflow.operations.operations_spasso import ( + SpassoConnectionSettings, + SpassoDownloadSettings, + SpassoRemoteFile, + download_spasso_file, + list_spasso_files, +) + + +def test_list_spasso_files(monkeypatch): + """Remote SFTP entries are converted to SpassoRemoteFile objects.""" + + transport = MagicMock() + sftp = MagicMock() + + sftp.listdir_attr.return_value = [ + SimpleNamespace( + filename="20260917_Copernicus_PHY.nc", + st_size=100, + st_mtime=1000, + ), + SimpleNamespace( + filename="20260916_Copernicus_SST_L4.nc", + st_size=200, + st_mtime=2000, + ), + ] + + monkeypatch.setattr( + "echodataflow.operations.operations_spasso._open_sftp", + lambda connection: (transport, sftp), + ) + + connection = SpassoConnectionSettings( + host="example.org", + username="user", + password="password", + ) + + files = list_spasso_files( + connection, + "data/Cruises/test/Processed", + ) + + assert len(files) == 2 + + assert files[0] == SpassoRemoteFile( + filename="20260917_Copernicus_PHY.nc", + remote_path=( + "data/Cruises/test/Processed/" + "20260917_Copernicus_PHY.nc" + ), + size_bytes=100, + mtime=1000, + ) + + assert files[1].filename == "20260916_Copernicus_SST_L4.nc" + assert files[1].size_bytes == 200 + assert files[1].mtime == 2000 + + sftp.listdir_attr.assert_called_once_with( + "data/Cruises/test/Processed" + ) + + sftp.close.assert_called_once() + transport.close.assert_called_once() + + +def test_list_spasso_files_closes_connection_on_error(monkeypatch): + """SFTP resources are closed even when listing fails.""" + + transport = MagicMock() + sftp = MagicMock() + + sftp.listdir_attr.side_effect = RuntimeError("SFTP listing failed") + + monkeypatch.setattr( + "echodataflow.operations.operations_spasso._open_sftp", + lambda connection: (transport, sftp), + ) + + connection = SpassoConnectionSettings( + host="example.org", + username="user", + password="password", + ) + + try: + list_spasso_files(connection, "remote") + except RuntimeError as exc: + assert str(exc) == "SFTP listing failed" + else: + raise AssertionError("Expected RuntimeError") + + sftp.close.assert_called_once() + transport.close.assert_called_once() + + +def test_download_spasso_file(monkeypatch, tmp_path): + """A remote SPASSO file is downloaded to the configured directory.""" + + transport = MagicMock() + sftp = MagicMock() + + monkeypatch.setattr( + "echodataflow.operations.operations_spasso._open_sftp", + lambda connection: (transport, sftp), + ) + + def fake_get(remote_path, local_path): + assert remote_path == ( + "remote/20260917_Copernicus_PHY.nc" + ) + + with open(local_path, "wb") as file: + file.write(b"spasso-data") + + sftp.get.side_effect = fake_get + + connection = SpassoConnectionSettings( + host="example.org", + username="user", + password="password", + ) + + item = SpassoRemoteFile( + filename="20260917_Copernicus_PHY.nc", + remote_path="remote/20260917_Copernicus_PHY.nc", + size_bytes=11, + mtime=1000, + ) + + settings = SpassoDownloadSettings( + local_directory=str(tmp_path), + ) + + result = download_spasso_file( + connection, + item, + settings, + ) + + expected_path = ( + tmp_path / "20260917_Copernicus_PHY.nc" + ) + + assert expected_path.exists() + assert expected_path.read_bytes() == b"spasso-data" + + assert result.filename == item.filename + assert result.remote_path == item.remote_path + assert result.local_path == str(expected_path) + assert result.size_bytes == 11 + + sftp.get.assert_called_once_with( + item.remote_path, + str(expected_path), + ) + + sftp.close.assert_called_once() + transport.close.assert_called_once() + + +def test_download_spasso_file_closes_connection_on_error( + monkeypatch, + tmp_path, +): + """SFTP resources are closed even when a download fails.""" + + transport = MagicMock() + sftp = MagicMock() + + sftp.get.side_effect = RuntimeError("download failed") + + monkeypatch.setattr( + "echodataflow.operations.operations_spasso._open_sftp", + lambda connection: (transport, sftp), + ) + + connection = SpassoConnectionSettings( + host="example.org", + username="user", + password="password", + ) + + item = SpassoRemoteFile( + filename="20260917_Copernicus_PHY.nc", + remote_path="remote/20260917_Copernicus_PHY.nc", + size_bytes=100, + mtime=1000, + ) + + settings = SpassoDownloadSettings( + local_directory=str(tmp_path), + ) + + try: + download_spasso_file( + connection, + item, + settings, + ) + except RuntimeError as exc: + assert str(exc) == "download failed" + else: + raise AssertionError("Expected RuntimeError") + + sftp.close.assert_called_once() + transport.close.assert_called_once() \ No newline at end of file