Containerise echodata and fix the pipeline for non-Azure backends

Everything here was required to run the pipeline outside Azure, on
Kubernetes, for the Denoise Lab deployment.

Packaging
- deploy/Dockerfile.echodata builds the oceanstream-echodata base image
  (echopype from the oceanstream-integration branch, so the builder needs
  git for setuptools_scm). boto3 is installed in the same pip run as the
  echodata extra: resolving it separately lets pip move botocore out from
  under s3fs, which then fails inside a Dask worker at flow-run time.
- .dockerignore keeps data/, raw_data/ and outputs out of the build
  context; ACR Tasks uploads the whole thing before it starts.

Dask worker plugins
- S3StoragePlugin and LocalStoragePlugin now subclass WorkerPlugin.
  distributed 2025.x rejects a plain class with a setup method:
  "TypeError: Registering duck-typed plugins is not allowed". This broke
  every non-Azure backend, local included.

Memory sizing in a container
- effective_parallel_workers sized itself from os.sysconf(SC_PHYS_PAGES),
  which reports the HOST's RAM inside a container. A pod with a 24 GiB
  limit on a 110 GiB node picked 7 stage workers and was OOM-killed.
  Now takes the minimum of that and the cgroup v1/v2 memory limit.
- keep_raw was swallowed into the previous line by a literal \n, so it
  was a comment rather than a dataclass field.

Rendering
- combine_38khz_day resampled with np.interp, which bridges interior NaN
  runs. On denoised input those runs ARE the removed noise, so the
  combined panel painted it back in and looked undenoised. The validity
  mask is now interpolated alongside and the output blanked where the
  nearest source samples were mostly NaN.
- Combined panels use the same canvas as the per-category ones
  (34x12 in at 180 dpi) so the two can be compared side by side.
- --force no longer reuses existing echogram PNGs. It means the denoise
  parameters changed, so the previous run's images were being shown for
  the new parameters, which is silently wrong.

Author: arusu

.dockerignore

@@ -0,0 +1,73 @@
+# =============================================================================
+# Build context filter for `az acr build` / `docker build`
+# =============================================================================
+# ACR Tasks uploads the whole context before it starts, so anything listed here
+# is bandwidth saved on every build. `raw_data/` and `data/` in particular sit
+# at the repo root on a developer machine and can run to hundreds of GB.
+# =============================================================================
+
+# Local data and pipeline outputs — never belongs in an image
+data/
+raw_data/
+out/
+**/out/
+output/
+**/geoparquet/
+.oceanstream_staging/
+.oceanstream_work/
+scripts/batch_processing/raw_cache/
+scripts/batch_processing/report_assets/
+*.out
+
+# Test fixtures. The image runs the pipeline, it does not run the test suite.
+oceanstream/tests/
+
+# Host artifacts that must never overlay what the image builds itself
+**/.venv
+venv/
+venv311/
+env/
+**/__pycache__
+*.py[cod]
+*.egg-info/
+build/
+dist/
+.eggs/
+
+# Caches and coverage
+.pytest_cache/
+.cache/
+.coverage
+.coverage.*
+htmlcov/
+coverage_html/
+coverage.json
+.mypy_cache/
+.ruff_cache/
+
+# Docs live in their own repo; notebooks and examples are not runtime code
+docs/
+docs-internal/
+notebooks/
+examples/
+.docs-website/
+.mkdocs-shadcn-fork/
+_oceanstream-website-static/
+_echodata-legacy-code/
+
+# Secrets and VCS
+.env
+.env.*
+.git/
+.github/
+
+# OS / editor cruft
+.DS_Store
+.AppleDouble
+.LSOverride
+Thumbs.db
+.vscode/
+.idea/
+.history/
+*.bak*
+*.log

deploy/Dockerfile.echodata

@@ -0,0 +1,111 @@
+# =============================================================================
+# oceanstream-echodata — base image for the batch echodata pipeline
+# =============================================================================
+# Ships `oceanstream[echodata]` plus every third-party module the scripts under
+# scripts/batch_processing/ import, and the scripts themselves. It runs nothing
+# by itself; downstream images (denoise-lab-worker) add their own entrypoint.
+#
+# Build from the repo root:
+#   az acr build --registry oceanstreamdevacr --platform linux/amd64 \
+#     --image oceanstream-echodata:<tag> -f deploy/Dockerfile.echodata .
+# =============================================================================
+
+ARG PYTHON_VERSION=3.12
+
+# -----------------------------------------------------------------------------
+# Stage 1: build the virtualenv
+# -----------------------------------------------------------------------------
+# Pinned to bookworm: trixie renamed the GDAL runtime lib (libgdal32 ->
+# libgdal36), so the runtime stage below could not satisfy it.
+FROM python:${PYTHON_VERSION}-slim-bookworm AS builder
+
+# git is not optional: pyproject.toml pulls echopype from the OceanStreamIO
+# fork's `oceanstream-integration` branch, and it builds with setuptools_scm,
+# which reads the cloned .git to derive a version.
+RUN apt-get update && apt-get install -y --no-install-recommends \
+        build-essential \
+        git \
+        libgdal-dev \
+        libgeos-dev \
+        libproj-dev \
+    && rm -rf /var/lib/apt/lists/*
+
+RUN python -m venv /opt/venv
+ENV PATH="/opt/venv/bin:$PATH"
+RUN pip install --no-cache-dir --upgrade pip wheel
+
+WORKDIR /src
+
+# Resolve the dependency tree against a package skeleton first. Editing the
+# source then only re-runs the --no-deps install below, instead of re-resolving
+# echopype-from-git, dask[complete], copernicusmarine and friends.
+COPY pyproject.toml ./
+COPY oceanstream/README.md ./oceanstream/README.md
+
+# The trailing names are modules the batch scripts import that the `echodata`
+# extra does not cover. They go in the SAME pip invocation on purpose: s3fs
+# pulls aiobotocore, which pins botocore to a narrow range, so boto3 has to be
+# resolved alongside it rather than bolted on afterwards.
+# Deliberately absent is the `echodata-viz` extra — holoviews/hvplot/panel/
+# bokeh/datashader are reached only by plot_interactive_echogram(), which
+# imports them lazily and which this pipeline never calls.
+RUN touch oceanstream/__init__.py \
+    && pip install --no-cache-dir ".[echodata]" \
+        boto3 \
+        cartopy \
+        cmocean \
+        matplotlib \
+        rasterio \
+        geopandas \
+        scipy \
+        pillow \
+        psycopg2-binary \
+        azure-storage-file-share \
+        python-dotenv
+
+COPY oceanstream/ ./oceanstream/
+RUN pip install --no-cache-dir --no-deps --force-reinstall .
+
+# -----------------------------------------------------------------------------
+# Stage 2: runtime
+# -----------------------------------------------------------------------------
+# Must match the builder's Debian release — the venv links against its libs.
+FROM python:${PYTHON_VERSION}-slim-bookworm AS runtime
+
+LABEL org.opencontainers.image.title="oceanstream-echodata"
+LABEL org.opencontainers.image.description="oceanstream[echodata] plus the batch_processing pipeline scripts"
+LABEL org.opencontainers.image.source="https://github.com/OceanStreamIO/oceanstream-cli"
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+        libgdal32 \
+        libgeos-c1v5 \
+        libproj25 \
+        ca-certificates \
+        curl \
+    && rm -rf /var/lib/apt/lists/* \
+    && apt-get clean
+
+# uid/gid 1000 is fixed on purpose: the worker and web containers share PVCs,
+# and the pod's fsGroup has to match one number across every image.
+RUN groupadd -g 1000 oceanstream \
+    && useradd -u 1000 -g 1000 -m -d /home/oceanstream -s /usr/sbin/nologin oceanstream
+
+COPY --from=builder /opt/venv /opt/venv
+COPY scripts/ /opt/oceanstream/scripts/
+
+# MPLBACKEND: without Agg, matplotlib picks a GUI backend and dies on import.
+# MPLCONFIGDIR/XDG_CACHE_HOME default under $HOME, which is not writable once
+# the pod runs as an arbitrary fsGroup; /tmp always is.
+ENV PATH="/opt/venv/bin:$PATH" \
+    PYTHONUNBUFFERED=1 \
+    PYTHONDONTWRITEBYTECODE=1 \
+    HOME=/home/oceanstream \
+    OCEANSTREAM_SCRIPTS=/opt/oceanstream/scripts/batch_processing \
+    MPLBACKEND=Agg \
+    MPLCONFIGDIR=/tmp/matplotlib \
+    XDG_CACHE_HOME=/tmp/cache
+
+WORKDIR /opt/oceanstream/scripts/batch_processing
+USER oceanstream
+
+CMD ["python"]

oceanstream/echodata/plot/combined.py

@@ -55,8 +55,13 @@
 # Sv panel look indistinguishable from a gridded MVBS one.
 PINGS_PER_PIXEL: float = 2.0
 
-# Cap on figure width (inches). At dpi=150 this is 15 000 px.
-MAX_WIDTH_IN: float = 100.0
+# Canvas geometry, matched to the per-category echograms in echogram.py
+# (34 x 12 in at 180 dpi = 6120 x 2160 px) so the two can be compared side by
+# side. Honouring PINGS_PER_PIXEL for a 29 000-ping day would want 80 in, and
+# the resulting 14 500 x 1050 px strip is unreadable in a browser.
+MAX_WIDTH_IN: float = 34.0
+HEIGHT_IN: float = 12.0
+RENDER_DPI: int = 180
 
 # Pulse-mode indicator bar colours (matches legacy scripts)
 _PULSE_COLORS = {
@@ -192,10 +197,20 @@ def combine_38khz_day(
             row = sv_raw[i, valid]
             m = ~np.isnan(row)
             if m.sum() > 1:
-                sv_interp[i] = np.interp(
+                interp = np.interp(
                     common_depth, depth_valid[m], row[m],
                     left=np.nan, right=np.nan,
                 )
+                # np.interp bridges interior NaN runs, and on a denoised input
+                # those runs ARE the removed noise — bridging them paints it
+                # back in and the panel looks undenoised. Blank any output
+                # sample whose nearest source samples were mostly NaN.
+                kept = np.interp(
+                    common_depth, depth_valid, m.astype(np.float32),
+                    left=0.0, right=0.0,
+                )
+                interp[kept < 0.5] = np.nan  # noqa: PLR2004 - nearest-sample majority
+                sv_interp[i] = interp
 
         mode_code = 0 if mode == "long_pulse" else 1
         ds_new = xr.Dataset(
@@ -321,9 +336,10 @@ def render_combined_echogram(
     max_plot_depth: float = MAX_PLOT_DEPTH,
     vmin: float = SV_VMIN,
     vmax: float = SV_VMAX,
-    dpi: int = 150,
+    dpi: int = RENDER_DPI,
     pings_per_pixel: float = PINGS_PER_PIXEL,
     max_width_in: float = MAX_WIDTH_IN,
+    height_in: float = HEIGHT_IN,
 ) -> Optional[Path]:
     """Render a combined 24h echogram for the 38 kHz channel.
 
@@ -355,11 +371,12 @@ def render_combined_echogram(
     max_plot_depth : float
         Trim depth axis at this value (metres).
     pings_per_pixel : float
-        Target pings per horizontal pixel. The canvas is widened until this
-        density is met, so a 29 000-ping Sv panel gets a much wider figure
-        than an 8 500-bin MVBS one.
+        Target pings per horizontal pixel. The canvas is widened towards this
+        density until ``max_width_in`` stops it.
     max_width_in : float
         Upper bound on figure width (inches).
+    height_in : float
+        Figure height (inches), including the pulse-mode strip.
 
     Returns
     -------
@@ -418,7 +435,7 @@ def render_combined_echogram(
             max(14.0, time_span * 1.2, n_pings / max(pings_per_pixel * dpi, 1.0)),
         )
     )
-    height = 7.0
+    height = height_in
 
     has_pulse = pulse_mode is not None
     cbar_frac = 0.3 / width

scripts/batch_processing/config.py

@@ -13,6 +13,29 @@
 from typing import Optional
 
 
+def _cgroup_memory_limit_gb() -> Optional[float]:
+    """The container's memory ceiling, or None when unconfined."""
+    for path in (
+        "/sys/fs/cgroup/memory.max",                    # cgroup v2
+        "/sys/fs/cgroup/memory/memory.limit_in_bytes",  # cgroup v1
+    ):
+        try:
+            raw = Path(path).read_text().strip()
+        except OSError:
+            continue
+        if raw == "max":
+            return None
+        try:
+            value = int(raw)
+        except ValueError:
+            continue
+        # cgroup v1 reports a sentinel near 2**63 when there is no limit.
+        if value >= 1 << 62:
+            return None
+        return value / (1024 ** 3)
+    return None
+
+
 @dataclass
 class DaskConfig:
     """Dask distributed cluster settings."""
@@ -323,7 +346,8 @@ class PipelineConfig:
     skip_campaign_echograms: bool = False  # skip only the campaign echogram loop (keep campaign zarr)
     build_campaign_sv_zarr: bool = False  # experimental
     category_parallel: bool = True  # parallelize short_pulse/long_pulse within each day
-    resume_stage: int = 0               # resume from this stage (0 = start from beginning)\n    keep_raw: bool = False              # keep downloaded raw files after conversion
+    resume_stage: int = 0               # resume from this stage (0 = start from beginning)
+    keep_raw: bool = False              # keep downloaded raw files after conversion
     # Stop after this stage (0 = run everything). Single-day comparison runs
     # set this to 9 so the campaign-aggregation stages don't run on one day.
     stop_after_stage: int = 0
@@ -395,16 +419,22 @@ def from_env(cls) -> PipelineConfig:
     def effective_parallel_workers(self, mem_per_worker_gb: float = 2.0) -> int:
         """Return the number of parallel stage workers.
 
-        If ``parallel_workers`` is 0 (the default), auto-detect from total
-        system RAM, reserving memory for Dask workers and OS overhead.
-        Each parallel denoise/echogram task needs ~2 GB.
+        If ``parallel_workers`` is 0 (the default), auto-detect from the RAM
+        this process may actually use, reserving memory for Dask workers and
+        OS overhead. Each parallel denoise/echogram task needs ~2 GB.
         """
         if self.parallel_workers > 0:
             return self.parallel_workers
         try:
             total_gb = os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES") / (1024 ** 3)
         except (ValueError, OSError):
             return 2  # safe default if detection fails
+        # sysconf reports the HOST's RAM even inside a container, so a pod with
+        # a 24 GiB limit on a 256 GiB node would size itself for 256 GiB and be
+        # OOM-killed or evicted mid-run.
+        cgroup_gb = _cgroup_memory_limit_gb()
+        if cgroup_gb is not None:
+            total_gb = min(total_gb, cgroup_gb)
         # Reserve ~50% for the Dask LocalCluster, OS, and headroom
         available_gb = total_gb * 0.5
         workers = max(1, int(available_gb / mem_per_worker_gb))

scripts/batch_processing/local_storage.py

@@ -216,11 +216,21 @@ def ls(self, path: str, detail: bool = False) -> list:
 
 # ── Dask worker plugin ───────────────────────────────────────────────────
 
-class LocalStoragePlugin:
+try:  # distributed is only needed for the Dask path, not for patch_storage()
+    from distributed.diagnostics.plugin import WorkerPlugin as _WorkerPlugin
+except ImportError:
+    _WorkerPlugin = object
+
+
+class LocalStoragePlugin(_WorkerPlugin):
     """Dask worker plugin that applies local storage patches on each worker.
 
     Register with ``client.register_worker_plugin(LocalStoragePlugin(root))``
     so that Dask workers use local filesystem instead of Azure.
+
+    Subclassing ``WorkerPlugin`` is mandatory, not decorative: distributed
+    2025.x onwards raises ``TypeError: Registering duck-typed plugins is not
+    allowed`` for a plain class with a ``setup`` method.
     """
 
     name = "local-storage"

scripts/batch_processing/process_campaign.py

@@ -2237,8 +2237,11 @@ def run_echogram_generation(
     with ThreadPoolExecutor(max_workers=max_workers) as pool:
         for day_key in sorted(all_keys):
             for category in sorted(all_keys[day_key]):
-                # Skip if echograms already exist in Azure
-                existing = _count_existing_echograms(output_container, day_key, category)
+                # --force means the denoise parameters changed, so the existing
+                # PNGs render the previous run and must not be reused.
+                existing = (
+                    0 if cfg.force else _count_existing_echograms(output_container, day_key, category)
+                )
                 if existing > 0:
                     logger.info("  Skipping %s/%s — %d echograms already exist", day_key, category, existing)
                     continue

scripts/batch_processing/s3_storage.py

@@ -363,13 +363,23 @@ def get_mapper(self, path: str, **kwargs) -> fsspec.FSMap:
 
 # ── Dask worker plugin ───────────────────────────────────────────────────
 
-class S3StoragePlugin:
+try:  # distributed is only needed for the Dask path, not for patch_storage()
+    from distributed.diagnostics.plugin import WorkerPlugin as _WorkerPlugin
+except ImportError:
+    _WorkerPlugin = object
+
+
+class S3StoragePlugin(_WorkerPlugin):
     """Dask worker plugin that applies the S3 storage patches on each worker.
 
     Register with ``client.register_plugin(S3StoragePlugin(bucket, prefix))``.
     Credentials are deliberately not carried by default: leaving *key* and
     *secret* unset makes each worker read them from its own environment
     (a K8s secret) instead of shipping them through the scheduler.
+
+    Subclassing ``WorkerPlugin`` is mandatory, not decorative: distributed
+    2025.x onwards raises ``TypeError: Registering duck-typed plugins is not
+    allowed`` for a plain class with a ``setup`` method.
     """
 
     name = "s3-storage"