Improve WEL file handling and naming in Dash app

Adds lazy loading for MFUSG WEL files, improving performance for large datasets. Refactors period key access with a new get_wel_period_keys utility, updates dataset naming to use a hash of change parameters, and enhances resource publishing to use output filenames. Also adds a helper to flatten single nested directories after downloads and improves period summary display in the UI.

Author: wmobley

dash/Dockerfile.dash

@@ -26,4 +26,4 @@ COPY flopy_wel_map.py /app/dash/
 
 EXPOSE 8050
 
-CMD ["gunicorn", "--bind", "0.0.0.0:8050", "dash_app:server"]
+CMD ["gunicorn", "--bind", "0.0.0.0:8050", "--timeout", "600", "dash_app:server"]

dash/dash_app.py

@@ -8,6 +8,8 @@
 import uuid
 from pathlib import Path
 import re
+import hashlib
+import json
 from typing import Dict, Iterable, List, Sequence
 
 import dash
@@ -26,14 +28,15 @@
     search_ckan_datasets,
     search_ckan_datasets_wel_rch,
 )
-from flopy_interactive.config import GRID_STANDARD_VAR
+from flopy_interactive.config import CKAN_BASE_URL, GRID_STANDARD_VAR
 from flopy_interactive.data.download import download_ckan_resource
 from flopy_interactive.data.grid import load_grid_resource
 from flopy_interactive.data.rch import apply_rch_rate_update, build_rch_cells_for_periods, load_rch
 from flopy_interactive.data.wel import (
     apply_rate_update,
     build_cell_id_lookup,
     collect_wel_cells_for_period_data,
+    get_wel_period_keys,
     load_wel,
 )
 from flopy_interactive.viz.color_modes import apply_color_mode
@@ -42,6 +45,7 @@
 DATA_DIR = Path(os.environ.get("FLOPY_DATA_DIR", "ckan_data"))
 OUTPUT_WEL = Path(os.environ.get("FLOPY_OUTPUT_WEL", "barton_springs_updated.wel"))
 SUGGEST_TITLE_FILTER = "Barton Springs Edwards Aquifer"
+CKAN_URL = os.environ.get("FLOPY_CKAN_URL", CKAN_BASE_URL)
 
 
 def get_datasets() -> List[Dict]:
@@ -362,6 +366,22 @@ def _dataset_options_without_grid(datasets: List[Dict]) -> List[Dict[str, str]]:
     return filtered
 
 
+def _summarize_periods(periods: List[int], total: int | None) -> str:
+    """Return a compact stress-period summary string."""
+    if not periods:
+        return f"All periods ({total})" if total else "All periods"
+    unique = sorted({int(p) for p in periods})
+    if total and len(unique) >= total:
+        return f"All periods ({total})"
+    if total and total > 0 and len(unique) / total >= 0.7:
+        return f"Periods: {len(unique)}/{total}"
+    if len(unique) > 1 and unique[-1] - unique[0] + 1 == len(unique):
+        return f"Periods: {unique[0] + 1}-{unique[-1] + 1}"
+    if len(unique) <= 5:
+        return "Periods: " + ", ".join(str(p + 1) for p in unique)
+    return "Periods: " + ", ".join(str(p + 1) for p in unique[:3]) + f" (+{len(unique) - 3} more)"
+
+
 def _downsample_for_choropleth(gdf_valid, gdf_full, zoom: float | None) -> pd.DataFrame:
     """Downsample grid polygons for choropleth rendering based on zoom."""
     if zoom is None:
@@ -685,8 +705,7 @@ def update_dataset_controls(loaded_dataset: str | None):
         return [], []
     data = load_dataset(loaded_dataset)
     wel = data["wel"]
-    spd = wel.stress_period_data.data
-    period_keys = sorted(list(spd.keys())) if spd else [0]
+    period_keys = get_wel_period_keys(wel) or [0]
     period_options = [{"label": f"SP {idx + 1}", "value": idx} for idx in period_keys]
     nlay = data["nlay"]
     layer_options = [{"label": str(layer), "value": layer} for layer in range(1, nlay + 1)]
@@ -709,8 +728,7 @@ def update_color_period(loaded_dataset, color_by, current_value):
         return {"display": "none"}, [], None
     data = load_dataset(loaded_dataset)
     wel = data["wel"]
-    spd = getattr(wel, "stress_period_data", None)
-    spd_keys = sorted(list(spd.data.keys())) if spd is not None and hasattr(spd, "data") else [0]
+    spd_keys = get_wel_period_keys(wel) or [0]
     options = [{"label": f"SP {idx + 1}", "value": idx} for idx in spd_keys]
     if current_value in spd_keys:
         value = current_value
@@ -788,8 +806,7 @@ def update_periods_layers(
         if not period_options or not layer_options:
             data = load_dataset(loaded_dataset)
             wel = data["wel"]
-            spd = wel.stress_period_data.data
-            period_keys = sorted(list(spd.keys())) if spd else [0]
+            period_keys = get_wel_period_keys(wel) or [0]
             period_options = [
                 {"label": f"SP {idx + 1}", "value": idx} for idx in period_keys
             ]
@@ -925,6 +942,7 @@ def update_dataset_suggestions(username, jwt_token):
     Input("load-counter", "data"),
     State("name-seed", "data"),
     State("last-loaded-dataset", "data"),
+    State("periods", "options"),
     State("dataset-name", "value"),
     State("output-wel", "value"),
     State("source-url", "value"),
@@ -946,6 +964,7 @@ def suggest_names(
     _load_counter,
     name_seed,
     last_loaded_dataset,
+    period_options,
     current_dataset_name,
     current_output_name,
     current_source_url,
@@ -992,33 +1011,34 @@ def suggest_names(
             suffix = "0% change"
     else:
         suffix = f"set-{rate_value:.0f}"
-    base_name = _slugify(f"{loaded_dataset}-{name_seed}")
+    period_total = len(period_options or [])
+    period_summary = _summarize_periods(list(periods or []), period_total or None)
+    change_spec = {
+        "flux_source": flux_source,
+        "rate_mode": rate_mode,
+        "new_rate": rate_value,
+        "periods": sorted(list(periods or [])),
+        "layers": sorted(list(layers or [])),
+        "add_missing": bool(add_missing),
+        "selection_count": len(selected_ids or []),
+        "color_by": color_by,
+        "category": category_value,
+    }
+    change_hash = hashlib.sha1(json.dumps(change_spec, sort_keys=True).encode("utf-8")).hexdigest()[:8]
+    base_name = _slugify(f"{loaded_dataset}-{change_hash}")
     dataset_name = current_dataset_name or base_name
     output_ext = ".rch" if flux_source == "rch" else ".wel"
-    output_name = f"{loaded_dataset}_{suffix}{output_ext}"
+    output_name = f"{loaded_dataset}_{suffix}_{change_hash}{output_ext}"
     if not output_name.lower().endswith(output_ext):
         output_name = f"{Path(output_name).stem}{output_ext}"
     source_url = current_source_url
-    if not source_url and jwt_token and loaded_dataset:
-        try:
-            details = ckanp.package_show(jwt_token, loaded_dataset)
-            source_url = details.get("url")
-            if not source_url:
-                resources = details.get("resources", [])
-                target_var = ckanp.RCH_STANDARD_VAR if flux_source == "rch" else ckanp.WEL_STANDARD_VAR
-                target_res = next(
-                    (res for res in resources if resource_has_standard_var(res, target_var)),
-                    None,
-                )
-                if target_res:
-                    source_url = target_res.get("url")
-        except Exception:
-            source_url = current_source_url
+    if not source_url and loaded_dataset:
+        source_url = f"{CKAN_URL}/dataset/{loaded_dataset}"
     selection_count = len(selected_ids or [])
     selection_desc = f"Selected cells: {selection_count}"
     if color_by in ("GCD_Name", "PGMA_Name") and category_value:
         selection_desc = f"Category {color_by} = {category_value}"
-    period_desc = "All periods" if not periods else f"Periods: {', '.join(str(p) for p in periods)}"
+    period_desc = period_summary
     if flux_source == "rch":
         layer_desc = "Layers: n/a"
         add_desc = "Add missing wells: n/a"
@@ -1234,6 +1254,16 @@ def apply_rate(
     data = load_dataset(loaded_dataset)
     wel = data["wel"]
     gdf = data["gdf"]
+    print(
+        "[apply] "
+        f"dataset={loaded_dataset} flux_source={flux_source} "
+        f"rate_mode={rate_mode} new_rate={new_rate} "
+        f"periods={periods} layers={layers} add_missing={add_missing} "
+        f"selected_count={len(selected_ids)} "
+        f"dataset_name={dataset_name} output={output_wel} "
+        f"source_url={source_url} change_summary={change_summary} "
+        f"jwt={'yes' if jwt_token else 'no'}"
+    )
     output_path = Path(output_wel or OUTPUT_WEL)
     target_ext = ".rch" if flux_source == "rch" else ".wel"
     if output_path.suffix.lower() != target_ext:

flopy_interactive/app/notebook_ui.py

@@ -22,6 +22,7 @@
     apply_rate_update,
     build_cell_id_lookup,
     collect_wel_cells_for_period_data,
+    get_wel_period_keys,
     load_wel,
 )
 from flopy_interactive.viz.color_modes import apply_color_mode, update_flux_customdata
@@ -121,9 +122,7 @@ def _collect_wel_cells_for_period(
         value="flux",
         description="Color by",
     )
-    spd_keys = sorted(list(wel.stress_period_data.data.keys()))
-    if not spd_keys:
-        spd_keys = [0]
+    spd_keys = get_wel_period_keys(wel) or [0]
     period_options = [(f"SP {idx + 1}", idx) for idx in spd_keys]
     period_select = widgets.SelectMultiple(
         options=period_options,

flopy_interactive/ckankit/publish.py

@@ -325,6 +325,7 @@ def build_resource_payload(
     change_summary: str | None = None,
     default_name: str = "WEL",
     default_format: str = "WEL",
+    resource_name: str | None = None,
 ) -> Dict:
     """Create a resource payload derived from a source resource.
 
@@ -358,7 +359,7 @@ def build_resource_payload(
     if change_summary:
         description = f"{description}\nMetadata Description of Changes Made: {change_summary}".strip()
     return {
-        "name": f"{source_resource.get('name', default_name)} (updated)",
+        "name": resource_name or f"{source_resource.get('name', default_name)} (updated)",
         "description": description,
         "format": source_resource.get("format", default_format),
         "mint_standard_variables": mint_svo,
@@ -456,6 +457,7 @@ def publish_updated_wel(
         mint_svo,
         source_url=source_url,
         change_summary=change_summary,
+        resource_name=f"{output_path.stem}",
     )
     created_resource = create_resource_upload(
         jwt_token,
@@ -558,6 +560,7 @@ def publish_updated_rch(
         change_summary=change_summary,
         default_name="RCH",
         default_format="RCH",
+        resource_name=f"{output_path.stem}",
     )
     created_resource = create_resource_upload(
         jwt_token,

flopy_interactive/data/download.py

@@ -34,6 +34,21 @@ def download_ckan_resource(resource: Dict, dest_dir: Path) -> Path:
     return dest_path
 
 
+def _flatten_single_dir(root: Path) -> None:
+    """Flatten a single nested directory in place."""
+    if not root.is_dir():
+        return
+    entries = list(root.iterdir())
+    subdirs = [entry for entry in entries if entry.is_dir()]
+    files = [entry for entry in entries if entry.is_file()]
+    if files or len(subdirs) != 1:
+        return
+    nested = subdirs[0]
+    for entry in nested.iterdir():
+        shutil.move(str(entry), root / entry.name)
+    nested.rmdir()
+
+
 def extract_zip(zip_path: Path) -> Path:
     """Extract a zip to a folder alongside the archive.
 

flopy_interactive/data/wel.py

@@ -40,6 +40,48 @@ def write_file(self, output_path: Path) -> None:
         output_path.write_text("\n".join(lines) + "\n")
 
 
+class LazyMfusgWel(MfusgWel):
+    """Lazy MFUSG WEL reader that loads selected stress periods on demand."""
+
+    is_lazy = True
+
+    def __init__(self, path: Path, header_line: str, index: list, dtype: np.dtype) -> None:
+        super().__init__(header_line, {}, dtype)
+        self.path = path
+        self.index = index
+        self.period_count = len(index)
+
+    def get_period(self, period: int) -> np.recarray:
+        if period < 0 or period >= self.period_count:
+            return np.recarray(0, dtype=self.stress_period_data.dtype)
+        entry = self.index[period]
+        itmp = entry["itmp"]
+        if itmp < 0:
+            return self.get_period(period - 1) if period > 0 else np.recarray(0, dtype=self.stress_period_data.dtype)
+        if itmp == 0:
+            return np.recarray(0, dtype=self.stress_period_data.dtype)
+        records = []
+        with self.path.open() as handle:
+            handle.seek(entry["offset"])
+            for _ in range(itmp):
+                line = _strip_comment(handle.readline())
+                if not line:
+                    continue
+                parts = line.split()
+                if len(parts) < 2:
+                    continue
+                node = int(float(parts[0]))
+                flux = float(parts[1])
+                records.append((node, flux))
+        return np.rec.array(records, dtype=self.stress_period_data.dtype) if records else np.recarray(0, dtype=self.stress_period_data.dtype)
+
+    def load_all(self) -> Dict[int, np.recarray]:
+        spd: Dict[int, np.recarray] = {}
+        for per in range(self.period_count):
+            spd[per] = self.get_period(per)
+        return spd
+
+
 def _strip_comment(line: str) -> str:
     for token in ("#", ";"):
         if token in line:
@@ -102,6 +144,31 @@ def _load_mfusg_wel(wel_path: Path) -> MfusgWel:
     return MfusgWel(header, spd, dtype)
 
 
+def _index_mfusg_wel(wel_path: Path) -> LazyMfusgWel:
+    """Index MFUSG WEL file for lazy period access."""
+    dtype = np.dtype([("node", "i4"), ("flux", "f8")])
+    index = []
+    with wel_path.open() as handle:
+        header = _strip_comment(handle.readline()).rstrip("\n")
+        per = 0
+        while True:
+            line = handle.readline()
+            if not line:
+                break
+            line = _strip_comment(line)
+            if not line:
+                continue
+            tokens = line.split()
+            itmp = int(tokens[0])
+            offset = handle.tell()
+            if itmp > 0:
+                for _ in range(itmp):
+                    handle.readline()
+            index.append({"itmp": itmp, "offset": offset})
+            per += 1
+    return LazyMfusgWel(wel_path, header, index, dtype)
+
+
 def scan_wel_metadata(path: Path) -> Tuple[int, int, int, int]:
     """Scan a MODFLOW WEL file for grid dimensions.
 
@@ -150,7 +217,7 @@ def load_wel(wel_path: Path) -> flopy.modflow.ModflowWel:
         FloPy WEL package.
     """
     if _detect_mfusg_wel(wel_path):
-        return _load_mfusg_wel(wel_path)
+        return _index_mfusg_wel(wel_path)
     nper, nlay, nrow, ncol = scan_wel_metadata(wel_path)
     model = flopy.modflow.Modflow(modelname="wel_read", model_ws=str(wel_path.parent))
     flopy.modflow.ModflowDis(
@@ -194,10 +261,19 @@ def collect_wel_cells_for_period_data(
         Mapping of CELL_ID to flux value.
     """
     cells: Dict[int, float] = {}
-    spd = wel.stress_period_data.data
-    if period not in spd:
-        return cells
-    recs = spd[period]
+    if hasattr(wel, "is_mfusg") and getattr(wel, "is_mfusg"):
+        if hasattr(wel, "get_period"):
+            recs = wel.get_period(period)
+        else:
+            spd = wel.stress_period_data.data
+            if period not in spd:
+                return cells
+            recs = spd[period]
+    else:
+        spd = wel.stress_period_data.data
+        if period not in spd:
+            return cells
+        recs = spd[period]
     for rec in recs:
         if hasattr(wel, "is_mfusg") and getattr(wel, "is_mfusg"):
             node = int(rec["node"])
@@ -247,6 +323,9 @@ def apply_rate_update(
     if hasattr(wel, "is_mfusg") and getattr(wel, "is_mfusg"):
         node_lookup = dict(zip(gdf["CELL_ID"], gdf["NODE_NUM"]))
         selected_cells = {int(node_lookup[cid]) for cid in selected_ids if cid in node_lookup}
+        if hasattr(wel, "load_all"):
+            spd = wel.load_all()
+            wel.stress_period_data.data = spd
     else:
         cell_lookup = dict(zip(gdf["CELL_ID"], zip(gdf["ROW"], gdf["COL"])))
         selected_cells = {cell_lookup[cid] for cid in selected_ids if cid in cell_lookup}
@@ -310,3 +389,13 @@ def apply_rate_update(
                 updated_spd[per] = np.rec.array(recs, dtype=base_dtype)
     wel.write_file(str(output_path))
     return len(selected_cells)
+
+
+def get_wel_period_keys(wel) -> list[int]:
+    """Return stress period indices for WEL, including lazy MFUSG."""
+    if hasattr(wel, "period_count"):
+        return list(range(int(wel.period_count)))
+    spd = getattr(wel, "stress_period_data", None)
+    if spd is not None and hasattr(spd, "data"):
+        return sorted(list(spd.data.keys()))
+    return []