Simplify CLI: normalize schema, deduplicate, remove dead code

- Normalize medium state to unified {field}_scalar/{field}_path schema
  (phantom generate now uses same format as phantom load)
- Remove bogus PPW check from plan (Nyquist-based PPW was a false positive
  for all valid CFL values)
- Remove dead code: unused min_wavelength, unused sys import, unused Optional
- Add Session.assert_ready() and Session.update_many()
- Rename _completeness() to completeness() (public API)
- Extract _invoke helper to tests/test_cli/conftest.py
- Shrink e2e test grid from 128x128 to 48x48
- Extract _parse_int_tuple and _resolve_scalar_or_path in phantom.py

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: waltsims

kwave/cli/commands/phantom.py

@@ -7,6 +7,19 @@
 from kwave.cli.schema import CLIError, CLIResponse, ValidationError, json_command
 
 
+def _parse_int_tuple(s: str) -> tuple[int, ...]:
+    return tuple(int(x) for x in s.split(","))
+
+
+def _resolve_scalar_or_path(value: str, name: str, sess) -> dict:
+    """Parse a CLI value as scalar float or .npy path. Returns {name_scalar, name_path} dict."""
+    if value.endswith(".npy"):
+        arr = np.load(value)
+        path = sess.save_array(name, arr)
+        return {f"{name}_scalar": None, f"{name}_path": path}
+    return {f"{name}_scalar": float(value), f"{name}_path": None}
+
+
 @click.group("phantom")
 def phantom():
     """Define the simulation phantom (medium + initial pressure)."""
@@ -25,60 +38,23 @@ def load(sess, grid_size, spacing, sound_speed, density, cfl):
     """Load medium properties from scalar values or .npy files."""
     sess.load()
 
-    grid_n = tuple(int(x) for x in grid_size.split(","))
+    grid_n = _parse_int_tuple(grid_size)
     ndim = len(grid_n)
     grid_spacing = (spacing,) * ndim
 
-    medium_state = {}
-
-    # Sound speed: scalar or .npy path
-    if sound_speed.endswith(".npy"):
-        arr = np.load(sound_speed)
-        path = sess.save_array("sound_speed", arr)
-        medium_state["sound_speed_path"] = path
-        medium_state["sound_speed_scalar"] = None
-        # Store path so makeTime gets the full array (not just max)
-        sound_speed_for_time_path = path
-        sound_speed_for_time_scalar = None
-    else:
-        medium_state["sound_speed_scalar"] = float(sound_speed)
-        medium_state["sound_speed_path"] = None
-        sound_speed_for_time_path = None
-        sound_speed_for_time_scalar = float(sound_speed)
-
-    # Density: scalar, .npy path, or None
+    medium_state = _resolve_scalar_or_path(sound_speed, "sound_speed", sess)
     if density is not None:
-        if density.endswith(".npy"):
-            arr = np.load(density)
-            path = sess.save_array("density", arr)
-            medium_state["density_path"] = path
-            medium_state["density_scalar"] = None
-        else:
-            medium_state["density_scalar"] = float(density)
-            medium_state["density_path"] = None
-
-    grid_state = {
-        "N": list(grid_n),
-        "spacing": list(grid_spacing),
-        "sound_speed_for_time_path": sound_speed_for_time_path,
-        "sound_speed_for_time_scalar": sound_speed_for_time_scalar,
-    }
+        medium_state.update(_resolve_scalar_or_path(density, "density", sess))
+
+    grid_state = {"N": list(grid_n), "spacing": list(grid_spacing)}
     if cfl is not None:
         grid_state["cfl"] = cfl
 
-    sess.update("grid", grid_state)
-    sess.update("medium", medium_state)
+    sess.update_many({"grid": grid_state, "medium": medium_state})
 
     return CLIResponse(
-        result={
-            "grid_size": list(grid_n),
-            "spacing": list(grid_spacing),
-            "medium": medium_state,
-        },
-        derived={
-            "ndim": ndim,
-            "grid_points": int(np.prod(grid_n)),
-        },
+        result={"grid_size": list(grid_n), "spacing": list(grid_spacing), "medium": medium_state},
+        derived={"ndim": ndim, "grid_points": int(np.prod(grid_n))},
     )
 
 
@@ -96,7 +72,7 @@ def generate(sess, phantom_type, grid_size, spacing, sound_speed, density, disc_
     """Generate an analytical phantom."""
     sess.load()
 
-    grid_n = tuple(int(x) for x in grid_size.split(","))
+    grid_n = _parse_int_tuple(grid_size)
     ndim = len(grid_n)
     grid_spacing = (spacing,) * ndim
 
@@ -117,12 +93,11 @@ def generate(sess, phantom_type, grid_size, spacing, sound_speed, density, disc_
         if disc_center is None:
             center = Vector([n // 2 for n in grid_n])
         else:
-            center = Vector([int(x) for x in disc_center.split(",")])
+            center = Vector(_parse_int_tuple(disc_center))
 
         p0 = make_disc(Vector(list(grid_n)), center, disc_radius).astype(float)
 
     elif phantom_type == "spherical":
-        # Simple spherical inclusion centered in grid
         center = np.array([n // 2 for n in grid_n])
         coords = np.mgrid[tuple(slice(0, n) for n in grid_n)]
         dist = np.sqrt(sum((c - cn) ** 2 for c, cn in zip(coords, center)))
@@ -133,32 +108,14 @@ def generate(sess, phantom_type, grid_size, spacing, sound_speed, density, disc_
         layer_pos = grid_n[0] // 4
         p0[layer_pos, ...] = 1.0
 
-    # Save arrays
     p0_path = sess.save_array("p0", p0)
 
-    # Update session
-    sess.update(
-        "grid",
+    sess.update_many(
         {
-            "N": list(grid_n),
-            "spacing": list(grid_spacing),
-            "sound_speed_for_time_scalar": sound_speed,
-            "sound_speed_for_time_path": None,
-        },
-    )
-    sess.update(
-        "medium",
-        {
-            "sound_speed": sound_speed,
-            "density": density,
-        },
-    )
-    sess.update(
-        "source",
-        {
-            "type": "initial-pressure",
-            "p0_path": p0_path,
-        },
+            "grid": {"N": list(grid_n), "spacing": list(grid_spacing)},
+            "medium": {"sound_speed_scalar": sound_speed, "sound_speed_path": None, "density_scalar": density, "density_path": None},
+            "source": {"type": "initial-pressure", "p0_path": p0_path},
+        }
     )
 
     return CLIResponse(
@@ -171,8 +128,5 @@ def generate(sess, phantom_type, grid_size, spacing, sound_speed, density, disc_
             "sound_speed": sound_speed,
             "density": density,
         },
-        derived={
-            "ndim": ndim,
-            "grid_points": int(np.prod(grid_n)),
-        },
+        derived={"ndim": ndim, "grid_points": int(np.prod(grid_n))},
     )

kwave/cli/commands/plan.py

@@ -4,7 +4,7 @@
 import numpy as np
 
 from kwave.cli.main import pass_session
-from kwave.cli.schema import CLIResponse, SessionError, json_command
+from kwave.cli.schema import CLIResponse, json_command
 
 
 @click.command("plan")
@@ -13,59 +13,29 @@
 def plan(sess):
     """Derive full simulation config and validate before running."""
     sess.load()
-
-    # Check completeness
-    comp = sess._completeness()
-    missing = [k for k, v in comp.items() if not v]
-    if missing:
-        raise SessionError(f"Cannot plan: missing {', '.join(missing)}. Complete setup first.")
+    sess.assert_ready("plan")
 
     kgrid = sess.make_grid()
     medium = sess.make_medium()
 
-    g = sess.state["grid"]
-    grid_n = tuple(g["N"])
-    spacing = tuple(g["spacing"])
+    grid_n = tuple(int(n) for n in kgrid.N)
+    spacing = tuple(float(d) for d in kgrid.spacing)
     ndim = len(grid_n)
     grid_points = int(np.prod(grid_n))
-
-    # Time stepping
     dt = float(kgrid.dt)
     Nt = int(kgrid.Nt)
 
-    # PPW check
-    c_min = float(np.min(medium.sound_speed)) if hasattr(medium.sound_speed, "__len__") else float(medium.sound_speed)
-    max_spacing = max(spacing)
-    # For IVP problems, use grid-based wavelength estimate
-    min_wavelength = c_min * dt * Nt / 2  # rough estimate
-    ppw = c_min / (max_spacing * (1 / (2 * dt)))  # Nyquist-based PPW
-
-    # CFL
     c_max = float(np.max(medium.sound_speed)) if hasattr(medium.sound_speed, "__len__") else float(medium.sound_speed)
+    c_min = float(np.min(medium.sound_speed)) if hasattr(medium.sound_speed, "__len__") else float(medium.sound_speed)
     cfl = c_max * dt / min(spacing)
 
-    # Memory estimate: ~(3 + 2*ndim) fields of float64
     n_fields = 3 + 2 * ndim
-    memory_bytes = grid_points * n_fields * 8
-    memory_mb = memory_bytes / (1024 * 1024)
+    memory_mb = grid_points * n_fields * 8 / (1024 * 1024)
+    estimated_runtime_s = grid_points * Nt * 50e-9  # ~50ns per grid point per step on CPU
 
-    # Runtime estimate
-    cost_per_point_step_ns = 50  # ~50ns per grid point per step on CPU
-    estimated_runtime_s = grid_points * Nt * cost_per_point_step_ns / 1e9
-
-    # PML
     pml_size = 20
 
-    # Warnings
     warnings = []
-    if ppw < 4:
-        warnings.append(
-            {
-                "code": "LOW_PPW",
-                "detail": f"PPW={ppw:.1f} is below recommended minimum of 4",
-                "suggestion": "Increase grid resolution or reduce maximum frequency",
-            }
-        )
     if cfl > 0.5:
         warnings.append(
             {
@@ -94,7 +64,6 @@ def plan(sess):
     }
 
     derived = {
-        "ppw": round(ppw, 2),
         "cfl": round(cfl, 4),
         "grid_points": grid_points,
         "estimated_memory_mb": round(memory_mb, 1),

kwave/cli/commands/run.py

@@ -1,14 +1,13 @@
 """Run command: execute simulation with structured JSON progress."""
 
 import json
-import sys
 import time
 
 import click
 import numpy as np
 
 from kwave.cli.main import pass_session
-from kwave.cli.schema import CLIResponse, SessionError, json_command
+from kwave.cli.schema import CLIResponse, json_command
 
 
 def _emit_event(event: dict):
@@ -24,12 +23,7 @@ def _emit_event(event: dict):
 def run(sess, backend, device):
     """Execute the simulation."""
     sess.load()
-
-    # Check completeness
-    comp = sess._completeness()
-    missing = [k for k, v in comp.items() if not v]
-    if missing:
-        raise SessionError(f"Cannot run: missing {', '.join(missing)}. Complete setup first.")
+    sess.assert_ready("run")
 
     kgrid = sess.make_grid()
     medium = sess.make_medium()

kwave/cli/main.py

@@ -1,7 +1,6 @@
 """Agent-first CLI for k-Wave simulations. All commands return structured JSON."""
 
 from pathlib import Path
-from typing import Optional
 
 import click