Add data dir and docs

.gitignore

@@ -38,7 +38,6 @@ MANIFEST
 *.h5
 *.hdf5
 cache/
-data/
 !examples/data/
 !examples/data/*.p
 !examples/data/README.md

LWMTemporal/data/__init__.py

@@ -0,0 +1,32 @@
+"""Data utilities for the LWM foundation package."""
+
+from .angle_delay import AngleDelayConfig, AngleDelayProcessor
+from .datasets import AngleDelaySequenceDataset, AngleDelayDatasetConfig, load_adseq_dataset
+from .scenario_generation import (
+    AntennaArrayConfig,
+    DynamicScenarioGenerator,
+    GridConfig,
+    ScenarioGenerationConfig,
+    ScenarioGenerationResult,
+    ScenarioSamplingConfig,
+    TrafficConfig,
+    generate_dynamic_scenario,
+    generate_dynamic_scenario_dataset,
+)
+
+__all__ = [
+    "AngleDelayConfig",
+    "AngleDelayProcessor",
+    "AngleDelaySequenceDataset",
+    "AngleDelayDatasetConfig",
+    "load_adseq_dataset",
+    "AntennaArrayConfig",
+    "TrafficConfig",
+    "GridConfig",
+    "ScenarioSamplingConfig",
+    "ScenarioGenerationConfig",
+    "ScenarioGenerationResult",
+    "DynamicScenarioGenerator",
+    "generate_dynamic_scenario_dataset",
+    "generate_dynamic_scenario",
+]

LWMTemporal/data/angle_delay.py

@@ -0,0 +1,272 @@
+from __future__ import annotations
+
+import dataclasses
+import math
+from pathlib import Path
+from typing import Iterable, List, Optional, Sequence
+
+import imageio.v2 as imageio
+import matplotlib.animation as animation
+import matplotlib.pyplot as plt
+import numpy as np
+import torch
+
+
+@dataclasses.dataclass
+class AngleDelayConfig:
+    """Configuration options for angle-delay processing."""
+
+    angle_range: tuple[float, float] = (-math.pi / 2, math.pi / 2)
+    delay_range: tuple[float, float] = (0.0, 100.0)
+    keep_percentage: float = 0.25
+    fps: int = 4
+    dpi: int = 120
+    num_bins: int = 6
+    output_dir: Path = Path("figs")
+
+    def validate(self) -> None:
+        if not 0.0 < self.keep_percentage <= 1.0:
+            raise ValueError("keep_percentage must be in (0, 1]")
+        if self.fps <= 0:
+            raise ValueError("fps must be positive")
+        if self.dpi <= 0:
+            raise ValueError("dpi must be positive")
+        if self.num_bins <= 0:
+            raise ValueError("num_bins must be positive")
+
+
+class AngleDelayProcessor:
+    """Project complex channels into the angle-delay domain and visualise them."""
+
+    def __init__(self, config: AngleDelayConfig | None = None) -> None:
+        self.config = config or AngleDelayConfig()
+        self.config.validate()
+
+    # ------------------------------------------------------------------
+    # Core transforms
+    # ------------------------------------------------------------------
+    @staticmethod
+    def _ensure_complex(tensor: torch.Tensor) -> torch.Tensor:
+        if not torch.is_complex(tensor):
+            raise TypeError("expected complex tensor")
+        return tensor
+
+    def forward(self, channel: torch.Tensor) -> torch.Tensor:
+        channel = self._ensure_complex(channel)
+        angle_domain = torch.fft.fft(channel, dim=1, norm="ortho")
+        delay_domain = torch.fft.ifft(angle_domain, dim=2, norm="ortho")
+        return delay_domain
+
+    def inverse(self, angle_delay: torch.Tensor) -> torch.Tensor:
+        angle_delay = self._ensure_complex(angle_delay)
+        subcarrier = torch.fft.fft(angle_delay, dim=2, norm="ortho")
+        antenna = torch.fft.ifft(subcarrier, dim=1, norm="ortho")
+        return antenna
+
+    # ------------------------------------------------------------------
+    # Truncation helpers and metrics
+    # ------------------------------------------------------------------
+    def truncate_delay_bins(self, tensor: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor]:
+        tensor = self._ensure_complex(tensor)
+        if tensor.ndim != 3:
+            raise ValueError("angle-delay tensor must have shape (T, N, M)")
+        keep = max(1, int(round(tensor.size(-1) * self.config.keep_percentage)))
+        truncated = tensor[..., :keep]
+        padded = torch.zeros_like(tensor)
+        padded[..., :keep] = truncated
+        return truncated, padded
+
+    @staticmethod
+    def nmse(reference: torch.Tensor, reconstruction: torch.Tensor) -> float:
+        reference = AngleDelayProcessor._ensure_complex(reference)
+        reconstruction = AngleDelayProcessor._ensure_complex(reconstruction)
+        mse = torch.mean(torch.abs(reference - reconstruction) ** 2)
+        power = torch.mean(torch.abs(reference) ** 2).clamp_min(1e-12)
+        return float(10.0 * torch.log10(mse / power))
+
+    def reconstruction_nmse(self, channel: torch.Tensor) -> tuple[float, float]:
+        ad_full = self.forward(channel)
+        recon_full = self.inverse(ad_full)
+        nmse_full = self.nmse(channel, recon_full)
+        truncated, padded = self.truncate_delay_bins(ad_full)
+        recon_trunc = self.inverse(padded)
+        nmse_trunc = self.nmse(channel, recon_trunc)
+        return nmse_full, nmse_trunc
+
+    # ------------------------------------------------------------------
+    # Visualisation helpers
+    # ------------------------------------------------------------------
+    def save_angle_delay_gif(self, tensor: torch.Tensor, output_path: Path, fps: Optional[int] = None) -> None:
+        tensor = self._ensure_complex(tensor)
+        output_path = Path(output_path)
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+
+        magnitude = tensor.abs().cpu()
+        vmin, vmax = float(magnitude.min()), float(magnitude.max())
+        frames: List[np.ndarray] = []
+        for frame_idx in range(magnitude.size(0)):
+            fig, ax = plt.subplots(figsize=(8, 6))
+            im = ax.imshow(
+                magnitude[frame_idx].numpy(),
+                cmap="gray_r",
+                origin="lower",
+                aspect="auto",
+                extent=[*self.config.delay_range, *self.config.angle_range],
+                vmin=vmin,
+                vmax=vmax,
+            )
+            ax.set_xlabel("Delay (samples)")
+            ax.set_ylabel("Angle (radians)")
+            ax.set_title(f"Angle-Delay Map – Frame {frame_idx}")
+            fig.colorbar(im, ax=ax, label="Power (linear)")
+            fig.canvas.draw()
+            frames.append(np.asarray(fig.canvas.buffer_rgba()))
+            plt.close(fig)
+        imageio.mimsave(output_path, frames, fps=fps or self.config.fps)
+
+    def _save_animation(self, fig: plt.Figure, animate_fn, output_path: Path, fps: Optional[int] = None, dpi: Optional[int] = None) -> None:
+        anim = animation.FuncAnimation(fig, animate_fn)
+        output_path = Path(output_path)
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        anim.save(output_path, writer="pillow", fps=fps or self.config.fps, dpi=dpi or self.config.dpi)
+        plt.close(fig)
+
+    def save_channel_animation(self, channel: torch.Tensor, output_path: Path) -> None:
+        channel = self._ensure_complex(channel)
+        magnitude = channel.abs().cpu()
+        phase = torch.angle(channel).cpu()
+        vmin, vmax = float(magnitude.min()), float(magnitude.max())
+        temporal_mag = magnitude.mean(dim=(1, 2))
+        temporal_phase = np.unwrap(phase.mean(dim=(1, 2)).numpy())
+
+        fig, (ax_mag, ax_mag_line, ax_phase_line) = plt.subplots(1, 3, figsize=(20, 6))
+        mag_img = ax_mag.imshow(magnitude[0].numpy(), cmap="gray_r", origin="upper", aspect="auto", vmin=vmin, vmax=vmax)
+        ax_mag.set_xlabel("Subcarrier")
+        ax_mag.set_ylabel("Antenna")
+        fig.colorbar(mag_img, ax=ax_mag, label="Magnitude")
+
+        mag_line, = ax_mag_line.plot([], [], "b-o", linewidth=2, markersize=4)
+        ax_mag_line.set_xlim(0, magnitude.size(0) - 1)
+        ax_mag_line.set_ylim(float(temporal_mag.min()) * 0.9, float(temporal_mag.max()) * 1.1)
+        ax_mag_line.grid(True, alpha=0.3)
+        ax_mag_line.set_xlabel("Frame")
+        ax_mag_line.set_ylabel("Average Magnitude")
+
+        phase_line, = ax_phase_line.plot([], [], "r-s", linewidth=2, markersize=4)
+        ax_phase_line.set_xlim(0, magnitude.size(0) - 1)
+        ax_phase_line.set_ylim(float(temporal_phase.min()) - 0.1, float(temporal_phase.max()) + 0.1)
+        ax_phase_line.grid(True, alpha=0.3)
+        ax_phase_line.set_xlabel("Frame")
+        ax_phase_line.set_ylabel("Average Phase (rad)")
+
+        def animate(idx: int):
+            mag_img.set_array(magnitude[idx].numpy())
+            ax_mag.set_title(f"Magnitude – Frame {idx}")
+            xs = np.arange(idx + 1)
+            mag_line.set_data(xs, temporal_mag[: idx + 1].numpy())
+            phase_line.set_data(xs, temporal_phase[: idx + 1])
+            return mag_img, mag_line, phase_line
+
+        self._save_animation(fig, animate, output_path)
+
+    def save_angle_delay_animation(self, tensor: torch.Tensor, output_path: Path, keep_percentage: Optional[float] = None) -> None:
+        tensor = self._ensure_complex(tensor)
+        magnitude = tensor.abs().cpu()
+        phase = torch.angle(tensor).cpu()
+        keep_suffix = "" if keep_percentage is None else f" (keep={keep_percentage * 100:.0f}%)"
+
+        fig, axes = plt.subplots(2, 2, figsize=(18, 10))
+        mag_ax, phase_ax, mag_line_ax, phase_line_ax = axes.flat
+        mag_img = mag_ax.imshow(magnitude[0].numpy(), cmap="gray_r", origin="upper", aspect="auto")
+        mag_ax.set_xlabel("Delay Bin")
+        mag_ax.set_ylabel("Angle Bin")
+        fig.colorbar(mag_img, ax=mag_ax, label="Magnitude")
+
+        phase_img = phase_ax.imshow(phase[0].numpy(), cmap="twilight", origin="upper", aspect="auto", vmin=-math.pi, vmax=math.pi)
+        phase_ax.set_xlabel("Delay Bin")
+        phase_ax.set_ylabel("Angle Bin")
+        fig.colorbar(phase_img, ax=phase_ax, label="Phase (rad)")
+
+        temporal_mag = magnitude.mean(dim=(1, 2))
+        temporal_phase = np.unwrap(phase.mean(dim=(1, 2)).numpy())
+        mag_line, = mag_line_ax.plot([], [], "r-o", linewidth=2)
+        phase_line, = phase_line_ax.plot([], [], "b-s", linewidth=2)
+
+        for axis, label in ((mag_line_ax, "Average Magnitude"), (phase_line_ax, "Average Phase (rad)")):
+            axis.set_xlabel("Frame")
+            axis.set_ylabel(label)
+            axis.set_xlim(0, tensor.size(0) - 1)
+            axis.grid(True, alpha=0.3)
+
+        def animate(idx: int):
+            mag_img.set_array(magnitude[idx].numpy())
+            phase_img.set_array(phase[idx].numpy())
+            mag_ax.set_title(f"AD Magnitude – Frame {idx}{keep_suffix}")
+            phase_ax.set_title(f"AD Phase – Frame {idx}{keep_suffix}")
+            xs = np.arange(idx + 1)
+            mag_line.set_data(xs, temporal_mag[: idx + 1].numpy())
+            phase_line.set_data(xs, temporal_phase[: idx + 1])
+            return mag_img, phase_img, mag_line, phase_line
+
+        self._save_animation(fig, animate, output_path)
+
+    def save_dominant_bin_animation(self, tensor: torch.Tensor, output_path: Path, threshold_ratio: float = 0.05) -> None:
+        tensor = self._ensure_complex(tensor)
+        magnitude = tensor.abs().cpu()
+        threshold = float(magnitude.max()) * threshold_ratio
+        dominant_counts = (magnitude > threshold).sum(dim=(1, 2)).numpy()
+
+        fig, (heat_ax, line_ax) = plt.subplots(1, 2, figsize=(16, 6))
+        heat_img = heat_ax.imshow(magnitude[0].numpy(), cmap="gray_r", origin="upper", aspect="auto")
+        heat_ax.set_xlabel("Delay Bin")
+        heat_ax.set_ylabel("Angle Bin")
+        fig.colorbar(heat_img, ax=heat_ax, label="Magnitude")
+
+        count_line, = line_ax.plot([], [], "r-s", linewidth=2)
+        line_ax.set_xlabel("Frame")
+        line_ax.set_ylabel("Dominant Bin Count")
+        line_ax.set_xlim(0, tensor.size(0) - 1)
+        line_ax.set_ylim(0, dominant_counts.max() * 1.1)
+        line_ax.grid(True, alpha=0.3)
+
+        def animate(idx: int):
+            heat_img.set_array(magnitude[idx].numpy())
+            heat_ax.set_title(f"Magnitude – Frame {idx}")
+            xs = np.arange(idx + 1)
+            count_line.set_data(xs, dominant_counts[: idx + 1])
+            return heat_img, count_line
+
+        self._save_animation(fig, animate, output_path)
+
+    def save_bin_evolution_plot(self, tensor: torch.Tensor, output_path: Path) -> None:
+        tensor = self._ensure_complex(tensor)
+        magnitude = tensor.abs()
+        avg_mag = magnitude.mean(dim=0)
+        flat_mag = avg_mag.flatten()
+        k = min(self.config.num_bins, flat_mag.numel())
+        _, indices = torch.topk(flat_mag, k)
+        angle_indices = (indices // tensor.size(-1)).tolist()
+        delay_indices = (indices % tensor.size(-1)).tolist()
+
+        time_axis = np.arange(tensor.size(0))
+        fig, axes = plt.subplots(k, 2, figsize=(12, 3 * k), squeeze=False)
+        for row in range(k):
+            series = tensor[:, angle_indices[row], delay_indices[row]]
+            mag_series = torch.abs(series).cpu().numpy()
+            phase_series = np.unwrap(torch.angle(series).cpu().numpy())
+            ax_mag, ax_phase = axes[row]
+            ax_mag.plot(time_axis, mag_series, "b-", linewidth=2)
+            ax_mag.set_title(f"Bin (angle={angle_indices[row]}, delay={delay_indices[row]}) – Magnitude")
+            ax_mag.set_xlabel("Frame")
+            ax_mag.set_ylabel("Magnitude")
+            ax_mag.grid(True, alpha=0.3)
+            ax_phase.plot(time_axis, phase_series, "r-", linewidth=2)
+            ax_phase.set_title(f"Bin (angle={angle_indices[row]}, delay={delay_indices[row]}) – Phase")
+            ax_phase.set_xlabel("Frame")
+            ax_phase.set_ylabel("Phase (rad)")
+            ax_phase.grid(True, alpha=0.3)
+        fig.tight_layout()
+        output_path = Path(output_path)
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        fig.savefig(output_path, dpi=self.config.dpi, bbox_inches="tight")
+        plt.close(fig)

LWMTemporal/data/datasets.py

@@ -0,0 +1,241 @@
+from __future__ import annotations
+
+import dataclasses
+import math
+import pickle
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple
+
+import numpy as np
+import torch
+from torch.utils.data import Dataset
+
+from .angle_delay import AngleDelayConfig, AngleDelayProcessor
+from ..models.lwm import ComplexPatchTokenizer
+
+
+@dataclasses.dataclass
+class AngleDelayDatasetConfig:
+    raw_path: Path
+    keep_percentage: float = 0.25
+    normalize: str = "global_rms"
+    cache_dir: Optional[Path] = Path("cache")
+    use_cache: bool = True
+    overwrite_cache: bool = False
+    snr_db: Optional[float] = None
+    noise_seed: Optional[int] = None
+    max_time_steps: Optional[int] = None
+    patch_size: Tuple[int, int] = (1, 1)
+    phase_mode: str = "real_imag"
+
+
+class AngleDelaySequenceDataset(Dataset):
+    """Angle-delay dataset with optional caching and metadata retention."""
+
+    def __init__(self, config: AngleDelayDatasetConfig, logger: Optional[Any] = None) -> None:
+        super().__init__()
+        self.config = config
+        self.logger = logger
+        self.tokenizer = ComplexPatchTokenizer(config.phase_mode)
+        self.samples: List[torch.Tensor]
+        self.avg_speed_mps: List[Optional[float]]
+        cache_hit = False
+        cache_path = self._cache_path() if config.use_cache and config.cache_dir is not None else None
+        if cache_path and cache_path.exists() and not config.overwrite_cache:
+            try:
+                payload = torch.load(cache_path, map_location="cpu")
+                if isinstance(payload, dict) and "samples" in payload:
+                    self.samples = payload["samples"]
+                    self.avg_speed_mps = payload.get("avg_speed_mps", [None] * len(self.samples))
+                else:
+                    self.samples = payload
+                    self.avg_speed_mps = [None] * len(self.samples)
+                cache_hit = True
+            except Exception:
+                cache_path.unlink(missing_ok=True)
+                cache_hit = False
+        if not cache_hit:
+            self.samples, self.avg_speed_mps = self._build_samples()
+            if cache_path is not None:
+                cache_path.parent.mkdir(parents=True, exist_ok=True)
+                torch.save({"samples": self.samples, "avg_speed_mps": self.avg_speed_mps}, cache_path)
+        if self.config.snr_db is not None:
+            self._apply_noise()
+
+    def _cache_path(self) -> Path:
+        cfg = self.config
+        name = cfg.raw_path.stem
+        # Include patch_size and phase_mode in cache name to ensure cache invalidation
+        # when these parameters change. Also add 'v2' to invalidate old caches with wrong normalization.
+        ph, pw = cfg.patch_size
+        cache_name = f"adseq_{name}_keep{int(cfg.keep_percentage * 100)}_{cfg.normalize}_p{ph}x{pw}_{cfg.phase_mode}_v2.pt"
+        return cfg.cache_dir / cache_name  # type: ignore[operator]
+
+    def _load_raw(self) -> Any:
+        with self.config.raw_path.open("rb") as handle:
+            return pickle.load(handle)
+
+    def _normalize_sample(self, tensor: torch.Tensor) -> torch.Tensor:
+        """Normalize a single sample by its own RMS."""
+        rms = torch.sqrt((tensor.real.float() ** 2 + tensor.imag.float() ** 2).mean()).clamp_min(1e-8)
+        return tensor / rms.to(tensor.dtype)
+
+    def _estimate_speed(self, pos_meta: Any, dt_meta: Any, index: int) -> Optional[float]:
+        if pos_meta is None or dt_meta is None:
+            return None
+
+        def _extract_dt(meta: Any) -> Optional[float]:
+            if isinstance(meta, (list, tuple)) and index < len(meta):
+                return float(meta[index])
+            if isinstance(meta, np.ndarray) and meta.ndim == 1 and index < meta.shape[0]:
+                return float(meta[index])
+            if isinstance(meta, (int, float)):
+                return float(meta)
+            return None
+
+        dt = _extract_dt(dt_meta)
+        if dt is None or dt <= 0:
+            return None
+
+        def _extract_positions(meta: Any) -> Optional[List[np.ndarray]]:
+            if isinstance(meta, (list, tuple)):
+                if index >= len(meta):
+                    return None
+                candidate = meta[index]
+            elif isinstance(meta, np.ndarray):
+                if meta.ndim < 2 or index >= meta.shape[0]:
+                    return None
+                candidate = meta[index]
+            else:
+                return None
+            if isinstance(candidate, list):
+                trajs = [np.asarray(traj) for traj in candidate if isinstance(traj, (list, tuple, np.ndarray))]
+            else:
+                trajs = [np.asarray(candidate)]
+            usable = [traj for traj in trajs if traj.ndim >= 2 and traj.shape[0] >= 2]
+            return usable if usable else None
+
+        trajectories = _extract_positions(pos_meta)
+        if not trajectories:
+            return None
+
+        speeds: List[float] = []
+        for traj in trajectories:
+            diffs = np.linalg.norm(np.diff(traj[:, :2], axis=0), axis=1)
+            if diffs.size == 0:
+                continue
+            velocity = diffs / dt
+            finite = velocity[np.isfinite(velocity)]
+            if finite.size > 0:
+                speeds.append(float(finite.mean()))
+        if not speeds:
+            return None
+        return float(np.mean(speeds))
+
+    def _build_samples(self) -> tuple[List[torch.Tensor], List[Optional[float]]]:
+        payload = self._load_raw()
+        pos_meta = payload.get("pos") if isinstance(payload, dict) else None
+        dt_meta = payload.get("dt") if isinstance(payload, dict) else None
+        channel = payload["channel"] if isinstance(payload, dict) and "channel" in payload else payload
+        channel_tensor = torch.as_tensor(channel, dtype=torch.complex64)
+        if channel_tensor.ndim == 3:
+            channel_tensor = channel_tensor.unsqueeze(0)
+        if self.config.max_time_steps is not None and channel_tensor.size(1) > self.config.max_time_steps:
+            channel_tensor = channel_tensor[:, : self.config.max_time_steps]
+        processor = AngleDelayProcessor(AngleDelayConfig(keep_percentage=self.config.keep_percentage))
+        samples: List[torch.Tensor] = []
+        avg_speeds: List[Optional[float]] = []
+        for idx, seq in enumerate(channel_tensor):
+            ad = processor.forward(seq)
+            truncated, _ = processor.truncate_delay_bins(ad)
+            samples.append(truncated)
+            avg_speeds.append(self._estimate_speed(pos_meta, dt_meta, idx))
+        
+        # Apply normalization after collecting all samples
+        if self.config.normalize == "per_sample_rms":
+            samples = [self._normalize_sample(s) for s in samples]
+        elif self.config.normalize == "global_rms":
+            # Compute global RMS across all samples
+            total_sum_sq = 0.0
+            total_count = 0
+            for s in samples:
+                s_real = s.real.float()
+                s_imag = s.imag.float()
+                total_sum_sq += (s_real ** 2 + s_imag ** 2).sum().item()
+                total_count += s_real.numel()
+            if total_count > 0:
+                global_rms = math.sqrt(total_sum_sq / total_count)
+                global_rms = max(global_rms, 1e-8)
+                samples = [s / torch.tensor(global_rms, dtype=torch.float32).to(s.dtype) for s in samples]
+        
+        return samples, avg_speeds
+
+    def _apply_noise(self) -> None:
+        if self.config.noise_seed is not None:
+            torch.manual_seed(int(self.config.noise_seed))
+        noisy: List[torch.Tensor] = []
+        snr_lin = 10.0 ** (float(self.config.snr_db) / 10.0)
+        for sample in self.samples:
+            real = sample.real.float()
+            imag = sample.imag.float()
+            power = (real.square() + imag.square()).mean().item()
+            if power <= 0:
+                noisy.append(sample)
+                continue
+            noise_var = power / snr_lin
+            std = math.sqrt(noise_var / 2.0)
+            noise_real = torch.randn_like(real) * std
+            noise_imag = torch.randn_like(imag) * std
+            noise = torch.complex(noise_real.to(sample.dtype), noise_imag.to(sample.dtype))
+            noisy.append((sample + noise).to(sample.dtype))
+        self.samples = noisy
+
+    def __len__(self) -> int:
+        return len(self.samples)
+
+    def __getitem__(self, index: int) -> Dict[str, Any]:
+        sample = self.samples[index]
+        tokens, base_mask = self.tokenizer(sample.unsqueeze(0), self.config.patch_size)
+        tokens = tokens.squeeze(0)
+        base_mask = base_mask.squeeze(0)
+        T, N, M = sample.shape
+        ph, pw = self.config.patch_size
+        H = N // ph
+        W = M // pw
+        shape = torch.tensor([T, H, W], dtype=torch.long)
+        avg_speed = self.avg_speed_mps[index] if index < len(self.avg_speed_mps) else None
+        payload: Dict[str, Any] = {
+            "sequence": sample,
+            "tokens": tokens,
+            "base_mask": base_mask,
+            "shape": shape,
+        }
+        if avg_speed is not None:
+            payload["avg_speed"] = torch.tensor(avg_speed, dtype=torch.float32)
+        return payload
+
+
+def load_adseq_dataset(
+    data_path: str | Path,
+    keep_percentage: float = 0.25,
+    normalize: str = "global_rms",
+    cache_dir: Optional[str | Path] = "cache",
+    use_cache: bool = True,
+    overwrite_cache: bool = False,
+    logger: Optional[Any] = None,
+    snr_db: Optional[float] = None,
+    noise_seed: Optional[int] = None,
+    max_time_steps: Optional[int] = None,
+) -> AngleDelaySequenceDataset:
+    cfg = AngleDelayDatasetConfig(
+        raw_path=Path(data_path),
+        keep_percentage=keep_percentage,
+        normalize=normalize,
+        cache_dir=None if cache_dir is None else Path(cache_dir),
+        use_cache=use_cache,
+        overwrite_cache=overwrite_cache,
+        snr_db=snr_db,
+        noise_seed=noise_seed,
+        max_time_steps=max_time_steps,
+    )
+    return AngleDelaySequenceDataset(cfg, logger=logger)

LWMTemporal/data/deepmimo_adapter.py

@@ -0,0 +1,236 @@
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Dict, Iterable, List, Optional, Sequence, Union
+
+import numpy as np  # type: ignore[import]
+
+_LOGGER = logging.getLogger(__name__)
+
+try:  # pragma: no cover - optional dependency
+    import deepmimo  # type: ignore[import]
+    from deepmimo import config as deepmimo_config  # type: ignore[import]
+
+    _HAS_DEEPMIMO_V4 = True
+except Exception:  # pragma: no cover - DeepMIMO v4 not installed
+    deepmimo = None  # type: ignore[assignment]
+    deepmimo_config = None  # type: ignore[assignment]
+    _HAS_DEEPMIMO_V4 = False
+
+try:  # pragma: no cover - legacy fallback
+    from input_preprocess import DeepMIMO_data_gen as _legacy_data_gen  # type: ignore[import]
+except Exception:  # pragma: no cover - legacy loader unavailable
+    _legacy_data_gen = None
+
+ArrayLike = Union[np.ndarray, "np.typing.NDArray[np.floating[Any]]"]
+
+
+@dataclass
+class _PathTable:
+    power: np.ndarray
+    phase: np.ndarray
+    delay: np.ndarray
+    aoa_az: np.ndarray
+    aoa_el: np.ndarray
+    aod_az: np.ndarray
+    aod_el: np.ndarray
+    interactions: np.ndarray
+    num_paths: np.ndarray
+    los_user: np.ndarray
+    locations: np.ndarray
+
+
+class _LazyPathAccessor:
+    """Lazy view over per-user path dictionaries compatible with v3 interface."""
+
+    def __init__(self, data: _PathTable) -> None:
+        self._data = data
+
+    def __len__(self) -> int:
+        return int(self._data.num_paths.shape[0])
+
+    def __getitem__(self, index: Union[int, slice, Sequence[int]]) -> Union[Dict[str, np.ndarray], List[Dict[str, np.ndarray]]]:
+        if isinstance(index, slice):
+            return [self[i] for i in range(*index.indices(len(self)))]
+        if isinstance(index, Sequence) and not isinstance(index, (str, bytes)):
+            return [self[int(i)] for i in index]
+        idx = int(index)
+        count = int(self._data.num_paths[idx])
+        if count <= 0:
+            empty = np.empty((0,), dtype=np.float32)
+            return {
+                "num_paths": 0,
+                "DoD_theta": empty,
+                "DoD_phi": empty,
+                "DoA_theta": empty,
+                "DoA_phi": empty,
+                "phase": empty,
+                "ToA": empty,
+                "power": empty,
+                "LoS": np.empty((0,), dtype=np.int32),
+            }
+        sl = slice(0, count)
+        interactions = np.asarray(self._data.interactions[idx, sl])
+        los_per_path = np.where(np.isnan(interactions), 0, (interactions == 0).astype(np.int32))
+        return {
+            "num_paths": count,
+            "DoD_theta": np.asarray(self._data.aod_el[idx, sl]),
+            "DoD_phi": np.asarray(self._data.aod_az[idx, sl]),
+            "DoA_theta": np.asarray(self._data.aoa_el[idx, sl]),
+            "DoA_phi": np.asarray(self._data.aoa_az[idx, sl]),
+            "phase": np.asarray(self._data.phase[idx, sl]),
+            "ToA": np.asarray(self._data.delay[idx, sl]),
+            "power": np.asarray(self._data.power[idx, sl]),
+            "LoS": los_per_path.astype(np.int32),
+        }
+
+
+def _cast(array: ArrayLike, dtype: np.dtype[Any]) -> np.ndarray:
+    arr = np.asarray(array)
+    if arr.dtype == dtype:
+        return arr
+    return arr.astype(dtype, copy=True)
+
+
+def _load_v4_dataset(
+    scenario: str,
+    *,
+    scenarios_dir: Optional[Path],
+    load_params: Optional[Dict[str, Any]],
+    max_paths: Optional[int],
+    array_dtype: np.dtype[Any],
+    logger: Optional[logging.Logger],
+) -> Dict[str, Any]:
+    if not _HAS_DEEPMIMO_V4:
+        raise RuntimeError("DeepMIMO v4 package is not available in the current environment")
+
+    if scenarios_dir is not None:
+        deepmimo_config.set("scenarios_folder", str(scenarios_dir))  # type: ignore[attr-defined]
+
+    params = dict(load_params or {})
+    if max_paths is not None:
+        params.setdefault("max_paths", int(max_paths))
+
+    dataset = deepmimo.load(scenario, **params)  # type: ignore[call-arg]
+    logger = logger or _LOGGER
+    logger.info(
+        "Loaded DeepMIMO v4 scenario '%s' with %s users and %s max paths",  # pragma: no cover - logging
+        scenario,
+        getattr(dataset, "n_ue", "unknown"),
+        params.get("max_paths", "default"),
+    )
+
+    num_paths_raw = np.asarray(dataset.num_paths)
+    tx_axis: Optional[int] = None
+    if num_paths_raw.ndim > 1 and num_paths_raw.shape[0] > 1:
+        axes = tuple(range(1, num_paths_raw.ndim))
+        scores = num_paths_raw.sum(axis=axes)
+        tx_axis = int(np.argmax(scores))
+
+    def _select_tx(arr: Any, dtype: Optional[np.dtype[Any]] = None) -> np.ndarray:
+        out = np.asarray(arr)
+        if out.dtype == object:
+            out = np.stack([np.asarray(v) for v in out], axis=0)
+        if tx_axis is not None and out.ndim >= 1 and out.shape[0] == num_paths_raw.shape[0]:
+            out = out[tx_axis]
+        if dtype is not None:
+            out = out.astype(dtype, copy=False)
+        return out
+
+    num_paths = _select_tx(num_paths_raw, dtype=np.int32).reshape(-1)
+    power = _select_tx(dataset.power, dtype=array_dtype)
+    phase = _select_tx(dataset.phase, dtype=array_dtype)
+    delay = _select_tx(dataset.delay, dtype=array_dtype)
+    aoa_az = _select_tx(dataset.aoa_az, dtype=array_dtype)
+    aoa_el = _select_tx(dataset.aoa_el, dtype=array_dtype)
+    aod_az = _select_tx(dataset.aod_az, dtype=array_dtype)
+    aod_el = _select_tx(dataset.aod_el, dtype=array_dtype)
+    interactions = _select_tx(dataset.inter, dtype=array_dtype)
+    los_raw = getattr(dataset, "los", None)
+    if los_raw is None:
+        los_selected = np.zeros_like(num_paths, dtype=np.int8)
+    else:
+        los_selected = _select_tx(los_raw, dtype=np.int8)
+    locations = _select_tx(dataset.rx_pos, dtype=np.float32)
+    if locations.ndim == 1:
+        if locations.size % 3 == 0:
+            locations = locations.reshape(-1, 3)
+        else:
+            locations = locations.reshape(-1, 1)
+    if locations.ndim > 2:
+        locations = locations.reshape(locations.shape[0], -1)
+
+    path_table = _PathTable(
+        power=power,
+        phase=phase,
+        delay=delay,
+        aoa_az=aoa_az,
+        aoa_el=aoa_el,
+        aod_az=aod_az,
+        aod_el=aod_el,
+        interactions=interactions,
+        num_paths=num_paths,
+        los_user=los_selected.reshape(-1),
+        locations=locations,
+    )
+
+    # Help GC release original dataset arrays early
+    del dataset
+
+    user_payload = {
+        "paths": _LazyPathAccessor(path_table),
+        "LoS": path_table.los_user,
+        "location": path_table.locations,
+    }
+
+    return {
+        "user": user_payload,
+        "_path_data": path_table,
+        "_source": "deepmimo_v4",
+    }
+
+
+def load_deepmimo_user_data(
+    scenario: str,
+    *,
+    scenarios_dir: Optional[Path] = None,
+    load_params: Optional[Dict[str, Any]] = None,
+    max_paths: Optional[int] = None,
+    array_dtype: np.dtype[Any] = np.float32,
+    logger: Optional[logging.Logger] = None,
+) -> Dict[str, Any]:
+    """Load DeepMIMO scenario data in a form compatible with legacy utilities.
+
+    The returned dictionary mimics the structure produced by DeepMIMO v3's
+    ``DeepMIMO_data_gen`` so downstream utilities (e.g., dynamic scenario
+    generation) can operate without modification. When DeepMIMO v4 is not
+    available, the function falls back to the legacy generator if present.
+    """
+
+    if _HAS_DEEPMIMO_V4:
+        return _load_v4_dataset(
+            scenario,
+            scenarios_dir=scenarios_dir,
+            load_params=load_params,
+            max_paths=max_paths,
+            array_dtype=array_dtype,
+            logger=logger,
+        )
+
+    if _legacy_data_gen is not None:
+        raise RuntimeError(
+            "DeepMIMO v4 is not installed. The repository still includes the legacy "
+            "DeepMIMO_data_gen interface, but integration parameters must be provided "
+            "explicitly. Please migrate to the official DeepMIMO package or invoke "
+            "DeepMIMO_data_gen directly from your own tooling."
+        )
+
+    raise RuntimeError(
+        "Neither DeepMIMO v4 nor the legacy DeepMIMO_data_gen function is available. "
+        "Please install the DeepMIMO package or provide the legacy generator."
+    )
+
+
+__all__ = ["load_deepmimo_user_data"]

LWMTemporal/data/scenario_generation.py

@@ -0,0 +1,1161 @@
+from __future__ import annotations
+
+import dataclasses
+import logging
+import warnings
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Sequence, Tuple
+
+import matplotlib.pyplot as plt  # type: ignore[import]
+from matplotlib.lines import Line2D  # type: ignore[attr-defined]
+import networkx as nx  # type: ignore[import]
+import numpy as np  # type: ignore[import]
+from scipy.spatial import KDTree  # type: ignore[import]
+
+from .deepmimo_adapter import load_deepmimo_user_data
+
+try:  # pragma: no cover - DeepMIMO may not be installed in all environments
+    import DeepMIMOv3.consts as c  # type: ignore
+except Exception:  # pragma: no cover - maintain compatibility when DeepMIMO is missing
+    class _C:  # noqa: D401 - minimal stub to satisfy typing
+        """Fallback constants when DeepMIMOv3 is not available."""
+
+    c = _C()
+    c.PARAMSET_OFDM = "OFDM"
+    c.PARAMSET_OFDM_BW = "bandwidth"
+    c.PARAMSET_OFDM_BW_MULT = 1e9
+    c.PARAMSET_OFDM_SC_SAMP = "selected_subcarriers"
+    c.PARAMSET_OFDM_SC_NUM = "subcarriers"
+    c.PARAMSET_OFDM_LPF = "LPF"
+    c.PARAMSET_FDTD = "FDTD"
+    c.PARAMSET_ANT_SHAPE = "shape"
+    c.PARAMSET_ANT_SPACING = "spacing"
+    c.PARAMSET_ANT_ROTATION = "rotation"
+    c.PARAMSET_ANT_FOV = "FoV"
+    c.PARAMSET_ANT_RAD_PAT = "radiation_pattern"
+    c.OUT_PATH_NUM = "num_paths"
+    c.OUT_PATH_DOD_THETA = "DoD_theta"
+    c.OUT_PATH_DOD_PHI = "DoD_phi"
+    c.OUT_PATH_DOA_THETA = "DoA_theta"
+    c.OUT_PATH_DOA_PHI = "DoA_phi"
+    c.OUT_PATH_PHASE = "phase"
+    c.OUT_PATH_TOA = "ToA"
+    c.OUT_PATH_RX_POW = "power"
+    c.OUT_PATH_DOP_VEL = "Doppler_vel"
+    c.OUT_PATH_DOP_ACC = "Doppler_acc"
+    c.PARAMSET_DOPPLER_EN = "Doppler"
+    c.PARAMSET_SCENARIO_PARAMS = "scenario_params"
+    c.PARAMSET_SCENARIO_PARAMS_DOPPLER_EN = "Doppler_enabled"
+    c.PARAMSET_SCENARIO_PARAMS_CF = "carrier_freq"
+    c.LIGHTSPEED = 3e8
+
+
+_LOGGER = logging.getLogger(__name__)
+
+
+@dataclass
+class AntennaArrayConfig:
+    """Configuration describing the transmit and receive array geometry."""
+
+    tx_horizontal: int = 32
+    tx_vertical: int = 1
+    rx_horizontal: int = 1
+    rx_vertical: int = 1
+    subcarriers: int = 32
+    spacing: float = 0.5
+    tx_rotation: Tuple[float, float, float] = (0.0, 0.0, -135.0)
+    rx_rotation: Tuple[float, float, float] = (0.0, 0.0, 0.0)
+    field_of_view: Tuple[float, float] = (360.0, 180.0)
+
+    def total_tx_elements(self) -> int:
+        return int(self.tx_horizontal * self.tx_vertical)
+
+    def tx_shape(self) -> np.ndarray:
+        return np.asarray([self.tx_horizontal, self.tx_vertical, 1])
+
+    def rx_shape(self) -> np.ndarray:
+        return np.asarray([self.rx_horizontal, self.rx_vertical, 1])
+
+
+@dataclass
+class TrafficConfig:
+    """Parameters controlling vehicle and pedestrian traffic synthesis."""
+
+    num_vehicles: int = 50
+    num_pedestrians: int = 10
+    vehicle_speed_range: Tuple[float, float] = (5 / 3.6, 60 / 3.6)
+    pedestrian_speed_range: Tuple[float, float] = (0.5, 2.0)
+    turn_probability: float = 0.1
+    max_attempts: int = 300
+    pedestrian_angle_std: float = 0.1
+
+
+@dataclass
+class GridConfig:
+    """Describe the road grid geometry to be inferred from DeepMIMO positions."""
+
+    road_width: float = 6.0
+    road_center_spacing: float = 25.0
+    step_size: float = 1.0
+    auto_step_size: bool = True
+
+
+@dataclass
+class ScenarioSamplingConfig:
+    """Temporal sampling configuration for the dynamic scenario."""
+
+    time_steps: int = 20
+    continuous_length: Optional[int] = None
+    sample_dt: float = 1e-3
+    continuous_mode: bool = True
+
+
+@dataclass
+class ScenarioGenerationConfig:
+    """All inputs required to synthesize a dynamic DeepMIMO scenario."""
+
+    scenario: str
+    antenna: AntennaArrayConfig = field(default_factory=AntennaArrayConfig)
+    sampling: ScenarioSamplingConfig = field(default_factory=ScenarioSamplingConfig)
+    grid: GridConfig = field(default_factory=GridConfig)
+    traffic: TrafficConfig = field(default_factory=TrafficConfig)
+    carrier_frequency_hz: float = 3.5e9
+    output_dir: Path = Path("examples/data")
+    full_output_dir: Path = Path("examples/full_data")
+    figures_dir: Optional[Path] = Path("figs")
+    export_environment_plot: bool = True
+    rng_seed: Optional[int] = None
+    scenarios_dir: Optional[Path] = None
+    deepmimo_max_paths: Optional[int] = 6
+    deepmimo_load_params: Optional[Dict[str, Any]] = None
+    deepmimo_array_dtype: np.dtype[Any] = np.float32
+
+
+@dataclasses.dataclass
+class ScenarioGenerationResult:
+    """Container for the generated dynamic scenario payload."""
+
+    payload: Dict[str, Any]
+    output_path: Path
+    full_output_path: Path
+    generated: bool
+
+
+# ---------------------------------------------------------------------------
+# Geometry helpers
+# ---------------------------------------------------------------------------
+
+def infer_grid_step(positions: np.ndarray, max_samples: int = 20000) -> float:
+    """Infer the base grid spacing from the DeepMIMO user positions."""
+
+    unique_xy = np.unique(positions[:, :2], axis=0)
+    if unique_xy.shape[0] < 2:
+        return 1.0
+    if unique_xy.shape[0] > max_samples:
+        rng = np.random.default_rng()
+        unique_xy = unique_xy[rng.choice(unique_xy.shape[0], size=max_samples, replace=False)]
+    tree = KDTree(unique_xy)
+    distances, _ = tree.query(unique_xy, k=2)
+    nearest = distances[:, 1]
+    nearest = nearest[nearest > 0]
+    if nearest.size == 0:
+        return 1.0
+    min_distance = float(np.min(nearest))
+    tolerance = min_distance * 0.1
+    count_at_min = np.sum(np.abs(nearest - min_distance) < tolerance)
+    if count_at_min >= 0.1 * nearest.size:
+        return min_distance
+    lo, hi = np.percentile(nearest, [5, 30])
+    mask = (nearest >= lo) & (nearest <= hi)
+    candidate = nearest[mask]
+    return float(np.median(candidate if candidate.size else nearest))
+
+
+def filter_road_positions(valid_positions: np.ndarray, road_width: float, spacing: float) -> Tuple[np.ndarray, Dict[Tuple[float, float, float], Tuple[Tuple[int, int], str]]]:
+    road_positions: List[np.ndarray] = []
+    lane_info: Dict[Tuple[float, float, float], Tuple[Tuple[int, int], str]] = {}
+    half = road_width / 2.0
+    for pos in valid_positions:
+        x, y, _ = pos
+        cx = round(x / spacing) * spacing
+        cy = round(y / spacing) * spacing
+        dx, dy = x - cx, y - cy
+        on_vertical = abs(dx) < half
+        on_horizontal = abs(dy) < half
+        key = tuple(pos)
+        if on_vertical and not on_horizontal:
+            direction = (0, 1) if dx >= 0 else (0, -1)
+            lane_info[key] = (direction, "vertical")
+            road_positions.append(pos)
+        elif on_horizontal and not on_vertical:
+            direction = (1, 0) if dy < 0 else (-1, 0)
+            lane_info[key] = (direction, "horizontal")
+            road_positions.append(pos)
+        elif on_vertical and on_horizontal:
+            direction = (0, 1) if dx >= 0 else (0, -1)
+            lane_info[key] = (direction, "intersection")
+            road_positions.append(pos)
+    return np.asarray(road_positions), lane_info
+
+
+def create_grid_road_network(road_positions: np.ndarray, lane_info: Dict[Tuple[float, float, float], Tuple[Tuple[int, int], str]], step_size: float) -> nx.DiGraph:
+    graph: nx.DiGraph = nx.DiGraph()
+    pos_dict = {tuple(pos): idx for idx, pos in enumerate(road_positions)}
+    for pos, idx in pos_dict.items():
+        if pos not in lane_info:
+            continue
+        direction, lane_type = lane_info[pos]
+        graph.add_node(idx, pos=np.array(pos), direction=direction, lane_type=lane_type)
+    tree = KDTree(road_positions)
+    for idx, pos in enumerate(road_positions):
+        if idx not in graph.nodes:
+            continue
+        neighbors = tree.query_ball_point(pos, r=step_size + 0.1)
+        for nb in neighbors:
+            if nb == idx or nb not in graph.nodes:
+                continue
+            target = road_positions[nb]
+            distance = np.linalg.norm(pos - target)
+            if not np.isclose(distance, step_size, atol=0.1):
+                continue
+            move_dir = (int(np.sign(target[0] - pos[0])), int(np.sign(target[1] - pos[1])))
+            lane_type = graph.nodes[idx].get("lane_type", "vertical")
+            if lane_type == "intersection":
+                if move_dir in [(0, 1), (0, -1), (1, 0), (-1, 0)]:
+                    graph.add_edge(idx, nb, weight=distance)
+            elif move_dir == graph.nodes[idx]["direction"]:
+                graph.add_edge(idx, nb, weight=distance)
+    return graph
+
+
+def _fallback_anywalk(road_positions: np.ndarray, step_size: float, length: int, start_idx: Optional[int] = None) -> np.ndarray:
+    if start_idx is None:
+        start_idx = np.random.randint(0, len(road_positions))
+    pos = road_positions[start_idx]
+    trajectory = [pos]
+    tree = KDTree(road_positions)
+    for _ in range(length - 1):
+        idxs = tree.query_ball_point(pos, r=step_size * 1.1)
+        candidates = []
+        for idx in idxs:
+            if np.allclose(road_positions[idx], pos):
+                continue
+            d = np.linalg.norm(road_positions[idx] - pos)
+            if np.isclose(d, step_size, atol=0.15 * max(1.0, step_size)):
+                candidates.append(idx)
+        if not candidates:
+            idxs = tree.query_ball_point(pos, r=max(1.5 * step_size, step_size + 0.5))
+            if not idxs:
+                break
+            candidate_idx = min(idxs, key=lambda j: np.linalg.norm(road_positions[j] - pos))
+        else:
+            candidate_idx = np.random.choice(candidates)
+        pos = road_positions[candidate_idx]
+        trajectory.append(pos)
+    return np.asarray(trajectory)
+
+
+def generate_smooth_grid_trajectory(graph: nx.DiGraph, road_positions: np.ndarray, turn_probability: float, sequence_length: int = 12, start_node: Optional[int] = None) -> np.ndarray:
+    if start_node is None:
+        nodes = list(graph.nodes)
+        if not nodes:
+            return np.empty((0, 3))
+        start_node = np.random.choice(nodes)
+    trajectory = [road_positions[start_node]]
+    current = start_node
+    previous: Optional[int] = None
+    for _ in range(sequence_length - 1):
+        if current not in graph:
+            break
+        neighbors = list(graph.neighbors(current))
+        if previous in neighbors:
+            neighbors.remove(previous)
+        if not neighbors:
+            remaining = sequence_length - len(trajectory)
+            if remaining > 0:
+                trajectory.extend([road_positions[current]] * remaining)
+            break
+        node_data = graph.nodes[current]
+        lane_type = node_data.get("lane_type", "vertical")
+        if lane_type == "intersection" and previous is not None:
+            prev_pos = np.asarray(graph.nodes[previous]["pos"])
+            curr_pos = np.asarray(node_data["pos"])
+            incoming = (
+                int(np.sign(curr_pos[0] - prev_pos[0])),
+                int(np.sign(curr_pos[1] - prev_pos[1])),
+            )
+            default_direction = incoming
+        else:
+            default_direction = node_data.get("direction", None)
+        position = np.asarray(node_data["pos"])
+        forward_neighbors: List[Tuple[int, float]] = []
+        turn_neighbors: List[Tuple[int, float]] = []
+        for nb in neighbors:
+            neighbor_pos = np.asarray(graph.nodes[nb]["pos"])
+            move_dir = (
+                int(np.sign(neighbor_pos[0] - position[0])),
+                int(np.sign(neighbor_pos[1] - position[1])),
+            )
+            distance = float(np.linalg.norm(neighbor_pos - position))
+            if move_dir == default_direction:
+                forward_neighbors.append((nb, distance))
+            else:
+                turn_neighbors.append((nb, distance))
+        if lane_type == "intersection":
+            r = np.random.rand()
+            if forward_neighbors and r > turn_probability:
+                nxt = min(forward_neighbors, key=lambda item: item[1])[0]
+            elif turn_neighbors and r < turn_probability:
+                nxt = min(turn_neighbors, key=lambda item: item[1])[0]
+            elif forward_neighbors:
+                nxt = min(forward_neighbors, key=lambda item: item[1])[0]
+            elif turn_neighbors:
+                nxt = min(turn_neighbors, key=lambda item: item[1])[0]
+            else:
+                remaining = sequence_length - len(trajectory)
+                if remaining > 0:
+                    trajectory.extend([road_positions[current]] * remaining)
+                break
+        else:
+            if forward_neighbors:
+                nxt = min(forward_neighbors, key=lambda item: item[1])[0]
+            else:
+                remaining = sequence_length - len(trajectory)
+                if remaining > 0:
+                    trajectory.extend([road_positions[current]] * remaining)
+                break
+        trajectory.append(road_positions[nxt])
+        previous, current = current, nxt
+    return np.asarray(trajectory)
+
+
+def generate_n_smooth_grid_trajectories(graph: nx.DiGraph, road_positions: np.ndarray, n: int, sequence_length: int = 12, turn_probability: float = 0.15, max_attempts: int = 2000, step_size: float = 1.0) -> List[np.ndarray]:
+    trajectories: List[np.ndarray] = []
+    attempts = 0
+    hard_cap = n * max_attempts
+    tree = KDTree(road_positions)
+    min_x, min_y = np.min(road_positions[:, 0]), np.min(road_positions[:, 1])
+    max_x, max_y = np.max(road_positions[:, 0]), np.max(road_positions[:, 1])
+    while len(trajectories) < n and attempts < hard_cap:
+        rand = [np.random.uniform(min_x, max_x), np.random.uniform(min_y, max_y), 0]
+        _, start_idx = tree.query(rand)
+        traj = generate_smooth_grid_trajectory(graph, road_positions, turn_probability, sequence_length, start_node=start_idx)
+        if traj.shape[0] < sequence_length:
+            traj = _fallback_anywalk(road_positions, step_size, sequence_length, start_idx=start_idx)
+        if traj.shape[0] >= sequence_length:
+            trajectories.append(traj[:sequence_length])
+        attempts += 1
+    if len(trajectories) < n:
+        _LOGGER.warning("Only generated %d/%d vehicle trajectories", len(trajectories), n)
+    return trajectories
+
+
+def generate_pedestrian_trajectory(valid_positions: np.ndarray, sequence_length: int = 10, step_size: float = 2.5, angle_std: float = 0.1, start: Optional[np.ndarray] = None) -> np.ndarray:
+    tree = KDTree(valid_positions)
+    if start is None:
+        start = valid_positions[np.random.choice(len(valid_positions))]
+    trajectory = [start]
+    angle = np.random.uniform(0, 2 * np.pi)
+    current = start
+    for _ in range(sequence_length - 1):
+        angle += np.random.normal(0, angle_std)
+        candidate = current + np.array([step_size * np.cos(angle), step_size * np.sin(angle), 0])
+        _, idx = tree.query(candidate)
+        nxt = valid_positions[idx]
+        trajectory.append(nxt)
+        current = nxt
+    return np.asarray(trajectory)
+
+
+def generate_n_pedestrian_trajectories(valid_positions: np.ndarray, n: int, sequence_length: int = 10, step_size: float = 2.5, angle_std: float = 0.1) -> List[np.ndarray]:
+    return [generate_pedestrian_trajectory(valid_positions, sequence_length, step_size, angle_std) for _ in range(n)]
+
+
+def get_trajectory_indices(trajectories: Sequence[np.ndarray], pos_total: np.ndarray) -> List[List[int]]:
+    def _pos_key(pos: Any) -> Tuple[float, ...]:
+        arr = np.asarray(pos)
+        if arr.ndim == 0:
+            return (float(arr),)
+        flat = arr.reshape(-1)
+        return tuple(float(x) for x in flat[:3])
+
+    pos_to_idx = {_pos_key(pos): i for i, pos in enumerate(pos_total)}
+    indices: List[List[int]] = []
+    for traj in trajectories:
+        indices.append([pos_to_idx.get(_pos_key(point), -1) for point in traj])
+    return indices
+
+
+def sample_continuous_along_polyline(
+    traj_pos: np.ndarray,
+    idxs: Sequence[int],
+    speed: float,
+    dt: float,
+    n_samples: int,
+    speed_profile: Optional[np.ndarray] = None,
+) -> Tuple[np.ndarray, List[Tuple[int, int]], np.ndarray, np.ndarray]:
+    traj_pos = np.asarray(traj_pos, float)
+    if traj_pos.shape[0] < 2:
+        p = np.repeat(traj_pos[:1], n_samples, axis=0)
+        pairs = [(idxs[0], idxs[0])] * n_samples
+        alphas = np.zeros(n_samples, float)
+        vdirs = np.zeros((n_samples, 2), float)
+        return p.astype(np.float32), pairs, alphas.astype(np.float32), vdirs.astype(np.float32)
+    segment_vectors = traj_pos[1:] - traj_pos[:-1]
+    segment_lengths = np.linalg.norm(segment_vectors[:, :2], axis=1)
+    cumulative = np.zeros(len(segment_lengths) + 1, float)
+    cumulative[1:] = np.cumsum(segment_lengths)
+    if cumulative[-1] <= 1e-12:
+        p = np.repeat(traj_pos[:1], n_samples, axis=0)
+        pairs = [(idxs[0], idxs[0])] * n_samples
+        alphas = np.zeros(n_samples, float)
+        vdirs = np.zeros((n_samples, 2), float)
+        return p.astype(np.float32), pairs, alphas.astype(np.float32), vdirs.astype(np.float32)
+    if speed_profile is not None:
+        speeds = np.asarray(speed_profile, float)
+        if speeds.shape[0] != n_samples:
+            raise ValueError("speed_profile must match number of samples")
+        ds = np.zeros(n_samples, float)
+        for k in range(1, n_samples):
+            ds[k] = ds[k - 1] + speeds[k - 1] * dt
+        ds = np.clip(ds, 0.0, max(cumulative[-1] - 1e-12, 0.0))
+    else:
+        ds = speed * dt * np.arange(n_samples, dtype=float)
+        ds = np.clip(ds, 0.0, max(cumulative[-1] - 1e-12, 0.0))
+    pos_c = np.zeros((n_samples, 3), float)
+    alphas = np.zeros(n_samples, float)
+    vdirs = np.zeros((n_samples, 2), float)
+    pairs: List[Tuple[int, int]] = []
+    for idx, distance in enumerate(ds):
+        seg_idx = int(np.searchsorted(cumulative, distance, side="right") - 1)
+        seg_idx = min(max(seg_idx, 0), len(segment_lengths) - 1)
+        seg_start = cumulative[seg_idx]
+        seg_length = segment_lengths[seg_idx]
+        alpha = (distance - seg_start) / (seg_length if seg_length > 1e-12 else 1.0)
+        alpha = min(max(alpha, 0.0), 1.0 - 1e-9)
+        p0 = traj_pos[seg_idx]
+        p1 = traj_pos[seg_idx + 1]
+        position = p0 + alpha * (p1 - p0)
+        direction = segment_vectors[seg_idx, :2] / (seg_length if seg_length > 1e-12 else 1.0)
+        pos_c[idx] = position
+        alphas[idx] = alpha
+        vdirs[idx] = direction
+        pairs.append((idxs[seg_idx], idxs[seg_idx + 1]))
+    return pos_c.astype(np.float32), pairs, alphas.astype(np.float32), vdirs.astype(np.float32)
+
+
+# ---------------------------------------------------------------------------
+# Channel construction helpers
+# ---------------------------------------------------------------------------
+
+def array_response_phase(theta: np.ndarray, phi: np.ndarray, kd: float) -> np.ndarray:
+    gamma_x = 1j * kd * np.sin(theta) * np.cos(phi)
+    gamma_y = 1j * kd * np.sin(theta) * np.sin(phi)
+    gamma_z = 1j * kd * np.cos(theta)
+    return np.vstack([gamma_x, gamma_y, gamma_z]).T
+
+
+def array_response(ant_indices: np.ndarray, theta: np.ndarray, phi: np.ndarray, kd: float) -> np.ndarray:
+    gamma = array_response_phase(theta, phi, kd)
+    return np.exp(ant_indices @ gamma.T)
+
+
+def ant_indices(panel_size: np.ndarray) -> np.ndarray:
+    gx = np.tile(np.arange(1), int(panel_size[0] * panel_size[1]))
+    gy = np.tile(np.repeat(np.arange(int(panel_size[0])), 1), int(panel_size[1]))
+    gz = np.repeat(np.arange(int(panel_size[1])), int(panel_size[0]))
+    return np.vstack([gx, gy, gz]).T
+
+
+def apply_fov(fov: Sequence[float], theta: np.ndarray, phi: np.ndarray) -> np.ndarray:
+    theta = np.mod(theta, 2 * np.pi)
+    phi = np.mod(phi, 2 * np.pi)
+    fov = np.deg2rad(fov)
+    include_phi = np.logical_or(phi <= 0 + fov[0] / 2, phi >= 2 * np.pi - fov[0] / 2)
+    include_theta = np.logical_and(theta <= np.pi / 2 + fov[1] / 2, theta >= np.pi / 2 - fov[1] / 2)
+    return np.logical_and(include_phi, include_theta)
+
+
+def rotate_angles(rotation: Optional[np.ndarray], theta: np.ndarray, phi: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
+    theta = np.deg2rad(theta)
+    phi = np.deg2rad(phi)
+    if rotation is not None:
+        R = np.deg2rad(rotation)
+        sa = np.sin(phi - R[2])
+        sb = np.sin(R[1])
+        sg = np.sin(R[0])
+        ca = np.cos(phi - R[2])
+        cb = np.cos(R[1])
+        cg = np.cos(R[0])
+        st, ct = np.sin(theta), np.cos(theta)
+        theta = np.arccos(cb * cg * ct + st * (sb * cg * ca - sg * sa))
+        phi = np.angle(cb * st * ca - sb * ct + 1j * (cb * sg * ct + st * (sb * sg * ca + cg * sa)))
+    return theta, phi
+
+
+class OFDMPathGenerator:
+    def __init__(self, params: Dict[str, Any], subcarriers: np.ndarray) -> None:
+        self.params = params
+        self.ofdm = params[c.PARAMSET_OFDM]
+        self.subcarriers = subcarriers
+        self.total_subc = self.ofdm[c.PARAMSET_OFDM_SC_NUM]
+        self.delay_bins = np.arange(self.ofdm["subcarriers"])
+        self.delay_to_ofdm = np.exp(-1j * 2 * np.pi / self.total_subc * np.outer(self.delay_bins, self.subcarriers))
+
+    def _doppler_phase(self, raydata: Dict[str, Any]) -> Optional[np.ndarray]:
+        if not (
+            self.params[c.PARAMSET_DOPPLER_EN]
+            and self.params[c.PARAMSET_SCENARIO_PARAMS][c.PARAMSET_SCENARIO_PARAMS_DOPPLER_EN]
+        ):
+            return None
+        fc = self.params[c.PARAMSET_SCENARIO_PARAMS][c.PARAMSET_SCENARIO_PARAMS_CF]
+        velocities = np.asarray(raydata.get(c.OUT_PATH_DOP_VEL, 0.0)).reshape(-1, 1)
+        elapsed = np.asarray(raydata.get("elapsed_time", 0.0)).reshape(-1, 1)
+        return np.exp(-1j * 2 * np.pi * (fc / c.LIGHTSPEED) * (velocities * elapsed))
+
+    def generate(self, raydata: Dict[str, Any], ts: float) -> np.ndarray:
+        if self.ofdm[c.PARAMSET_OFDM_LPF] == 0:
+            return self.no_lpf(raydata, ts)
+        return self.with_lpf(raydata, ts)
+
+    def no_lpf(self, raydata: Dict[str, Any], ts: float) -> np.ndarray:
+        power = raydata[c.OUT_PATH_RX_POW].reshape(-1, 1)
+        delay_n = (raydata[c.OUT_PATH_TOA] / ts).reshape(-1, 1)
+        phase = raydata[c.OUT_PATH_PHASE].reshape(-1, 1)
+        over = delay_n >= self.ofdm["subcarriers"]
+        power[over] = 0
+        delay_n[over] = self.ofdm["subcarriers"]
+        path_const = np.sqrt(power / self.total_subc) * np.exp(
+            1j * (np.deg2rad(phase) - (2 * np.pi / self.total_subc) * np.outer(delay_n, self.subcarriers))
+        )
+        doppler = self._doppler_phase(raydata)
+        if doppler is not None:
+            path_const *= doppler
+        return path_const
+
+    def with_lpf(self, raydata: Dict[str, Any], ts: float) -> np.ndarray:
+        power = raydata[c.OUT_PATH_RX_POW].reshape(-1, 1)
+        delay_n = (raydata[c.OUT_PATH_TOA] / ts).reshape(-1, 1)
+        phase = raydata[c.OUT_PATH_PHASE].reshape(-1, 1)
+        over = delay_n >= self.ofdm["subcarriers"]
+        power[over] = 0
+        delay_n[over] = self.ofdm["subcarriers"]
+        pulse = np.sinc(self.delay_bins - delay_n) * np.sqrt(power / self.total_subc) * np.exp(1j * np.deg2rad(phase))
+        doppler = self._doppler_phase(raydata)
+        if doppler is not None:
+            pulse *= doppler
+        return pulse
+
+
+def generate_mimo_channel(raydata: Sequence[Dict[str, Any]], params: Dict[str, Any], tx_params: Dict[str, Any], rx_params: Dict[str, Any]) -> Tuple[np.ndarray, np.ndarray]:
+    bw = params[c.PARAMSET_OFDM][c.PARAMSET_OFDM_BW] * c.PARAMSET_OFDM_BW_MULT
+    kd_tx = 2 * np.pi * tx_params[c.PARAMSET_ANT_SPACING]
+    kd_rx = 2 * np.pi * rx_params[c.PARAMSET_ANT_SPACING]
+    ts = 1 / bw
+    subcarriers = params[c.PARAMSET_OFDM][c.PARAMSET_OFDM_SC_SAMP]
+    generator = OFDMPathGenerator(params, subcarriers)
+    m_tx = int(np.prod(tx_params[c.PARAMSET_ANT_SHAPE]))
+    tx_indices = ant_indices(tx_params[c.PARAMSET_ANT_SHAPE])
+    m_rx = int(np.prod(rx_params[c.PARAMSET_ANT_SHAPE]))
+    rx_indices = ant_indices(rx_params[c.PARAMSET_ANT_SHAPE])
+    channel = np.zeros((len(raydata), m_rx, m_tx, len(subcarriers)), dtype=np.csingle)
+    los = np.zeros((len(raydata)), dtype=np.int8) - 2
+    for idx, item in enumerate(raydata):
+        if item[c.OUT_PATH_NUM] == 0:
+            los[idx] = -1
+            continue
+        dod_theta, dod_phi = rotate_angles(tx_params[c.PARAMSET_ANT_ROTATION], item[c.OUT_PATH_DOD_THETA], item[c.OUT_PATH_DOD_PHI])
+        doa_theta, doa_phi = rotate_angles(rx_params[c.PARAMSET_ANT_ROTATION], item[c.OUT_PATH_DOA_THETA], item[c.OUT_PATH_DOA_PHI])
+        include_tx = apply_fov(tx_params[c.PARAMSET_ANT_FOV], dod_theta, dod_phi)
+        include_rx = apply_fov(rx_params[c.PARAMSET_ANT_FOV], doa_theta, doa_phi)
+        include = np.logical_and(include_tx, include_rx)
+        dod_theta, dod_phi, doa_theta, doa_phi = dod_theta[include], dod_phi[include], doa_theta[include], doa_phi[include]
+        for key in list(item.keys()):
+            if key == c.OUT_PATH_NUM:
+                item[key] = include.sum()
+            elif isinstance(item[key], np.ndarray) and item[key].shape[0] == include.shape[0]:
+                item[key] = item[key][include]
+        if item[c.OUT_PATH_NUM] == 0:
+            los[idx] = -1
+            continue
+        los[idx] = int(np.sum(item.get("LoS", np.zeros(int(item[c.OUT_PATH_NUM])))))
+        tx_response = array_response(tx_indices, dod_theta, dod_phi, kd_tx)
+        rx_response = array_response(rx_indices, doa_theta, doa_phi, kd_rx)
+        factors = generator.generate(item, ts)
+        if params[c.PARAMSET_OFDM][c.PARAMSET_OFDM_LPF] == 0:
+            channel[idx] = np.sum(
+                rx_response[:, None, None, :] * tx_response[None, :, None, :] * factors.T[None, None, :, :],
+                axis=3,
+            )
+        else:
+            channel[idx] = (
+                np.sum(
+                    rx_response[:, None, None, :] * tx_response[None, :, None, :] * factors.T[None, None, :, :],
+                    axis=3,
+                )
+            ) @ generator.delay_to_ofdm
+    return channel, los
+
+
+def generate_channel_from_interpolated_ray(ray_interp: Dict[str, Any], antenna_cfg: AntennaArrayConfig, carrier_frequency_hz: float) -> Tuple[np.ndarray, Optional[np.ndarray], np.ndarray]:
+    raydata = [
+        {
+            c.OUT_PATH_NUM: int(ray_interp["num_paths"]),
+            c.OUT_PATH_DOD_THETA: np.asarray(ray_interp["DoD_theta"]),
+            c.OUT_PATH_DOD_PHI: np.asarray(ray_interp["DoD_phi"]),
+            c.OUT_PATH_DOA_THETA: np.asarray(ray_interp["DoA_theta"]),
+            c.OUT_PATH_DOA_PHI: np.asarray(ray_interp["DoA_phi"]),
+            c.OUT_PATH_PHASE: np.asarray(ray_interp["phase"]),
+            c.OUT_PATH_TOA: np.asarray(ray_interp["ToA"]),
+            c.OUT_PATH_RX_POW: np.asarray(ray_interp["power"]),
+            "LoS": np.asarray(ray_interp.get("LoS", np.zeros(int(ray_interp["num_paths"])))),
+            c.OUT_PATH_DOP_VEL: np.asarray(ray_interp.get("Doppler_vel", np.zeros(int(ray_interp["num_paths"])))),
+            "elapsed_time": np.asarray(ray_interp.get("elapsed_time", np.zeros(int(ray_interp["num_paths"])))),
+        }
+    ]
+    params = {
+        c.PARAMSET_OFDM: {
+            c.PARAMSET_OFDM_BW: 1.92e-3,
+            c.PARAMSET_OFDM_SC_SAMP: np.arange(antenna_cfg.subcarriers),
+            c.PARAMSET_OFDM_SC_NUM: antenna_cfg.subcarriers,
+            c.PARAMSET_OFDM_LPF: 0,
+            "subcarriers": antenna_cfg.subcarriers,
+        },
+        c.PARAMSET_FDTD: True,
+        c.PARAMSET_DOPPLER_EN: True,
+        c.PARAMSET_SCENARIO_PARAMS: {
+            c.PARAMSET_SCENARIO_PARAMS_DOPPLER_EN: True,
+            c.PARAMSET_SCENARIO_PARAMS_CF: float(carrier_frequency_hz),
+        },
+    }
+    tx = {
+        c.PARAMSET_ANT_SHAPE: antenna_cfg.tx_shape(),
+        c.PARAMSET_ANT_SPACING: antenna_cfg.spacing,
+        c.PARAMSET_ANT_ROTATION: np.asarray(antenna_cfg.tx_rotation),
+        c.PARAMSET_ANT_FOV: list(antenna_cfg.field_of_view),
+        c.PARAMSET_ANT_RAD_PAT: "isotropic",
+    }
+    rx = {
+        c.PARAMSET_ANT_SHAPE: antenna_cfg.rx_shape(),
+        c.PARAMSET_ANT_SPACING: antenna_cfg.spacing,
+        c.PARAMSET_ANT_ROTATION: np.asarray(antenna_cfg.rx_rotation),
+        c.PARAMSET_ANT_FOV: list(antenna_cfg.field_of_view),
+        c.PARAMSET_ANT_RAD_PAT: "isotropic",
+    }
+    channel, los = generate_mimo_channel(raydata, params, tx, rx)
+    return channel, None, los
+
+
+def unwrap_angle_deg(a0: np.ndarray, a1: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
+    delta = ((a1 - a0 + 180.0) % 360.0) - 180.0
+    return a0, a0 + delta
+
+
+def interpolate_ray_params(deepmimo_data: Dict[str, Any], idx0: int, idx1: int, alpha: float) -> Dict[str, Any]:
+    p0 = deepmimo_data["user"]["paths"][idx0]
+    p1 = deepmimo_data["user"]["paths"][idx1]
+    l0, l1 = int(p0["num_paths"]), int(p1["num_paths"])
+    if l0 == 0 or l1 == 0:
+        return {
+            "num_paths": 0,
+            "DoD_theta": np.array([]),
+            "DoD_phi": np.array([]),
+            "DoA_theta": np.array([]),
+            "DoA_phi": np.array([]),
+            "phase": np.array([]),
+            "ToA": np.array([]),
+            "power": np.array([]),
+            "LoS": np.array([], int),
+        }
+
+    def _select_paths(path: Dict[str, Any], count: int) -> np.ndarray:
+        scores = np.asarray(path["power"]).flatten()
+        order = np.argsort(-scores)
+        return order[:count]
+
+    count = min(l0, l1)
+    idxs0 = _select_paths(p0, count)
+    idxs1 = _select_paths(p1, count)
+
+    def _g(source: Dict[str, Any], key: str, selection: np.ndarray) -> np.ndarray:
+        return np.asarray(source[key]).flatten()[selection].astype(float)
+
+    power0, power1 = _g(p0, "power", idxs0), _g(p1, "power", idxs1)
+    power = (1 - alpha) * power0 + alpha * power1
+    toa0, toa1 = _g(p0, "ToA", idxs0), _g(p1, "ToA", idxs1)
+    toa = (1 - alpha) * toa0 + alpha * toa1
+
+    def _ainterp(key: str) -> np.ndarray:
+        a0, a1 = _g(p0, key, idxs0), _g(p1, key, idxs1)
+        out = np.zeros_like(a0)
+        for n in range(count):
+            u0, u1 = unwrap_angle_deg(a0[n], a1[n])
+            out[n] = (1 - alpha) * u0 + alpha * u1
+        return out
+
+    dod_theta, dod_phi = _ainterp("DoD_theta"), _ainterp("DoD_phi")
+    doa_theta, doa_phi = _ainterp("DoA_theta"), _ainterp("DoA_phi")
+    phase0 = np.deg2rad(_g(p0, "phase", idxs0))
+    phase1 = np.deg2rad(_g(p1, "phase", idxs1))
+    dphi = np.angle(np.exp(1j * (phase1 - phase0)))
+    phase = np.rad2deg(phase0 + alpha * dphi)
+    los = (
+        _g(p0, "LoS", idxs0).astype(int) + _g(p1, "LoS", idxs1).astype(int) > 0
+    ).astype(int)
+
+    return {
+        "num_paths": count,
+        "DoD_theta": dod_theta,
+        "DoD_phi": dod_phi,
+        "DoA_theta": doa_theta,
+        "DoA_phi": doa_phi,
+        "phase": phase,
+        "ToA": toa,
+        "power": power,
+        "LoS": los,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Scenario generator
+# ---------------------------------------------------------------------------
+
+
+class DynamicScenarioGenerator:
+    """High-level orchestrator for dynamic DeepMIMO scenario synthesis."""
+
+    def __init__(self, config: ScenarioGenerationConfig, logger: Optional[logging.Logger] = None) -> None:
+        self.config = config
+        self.logger = logger or _LOGGER
+        if config.rng_seed is not None:
+            np.random.seed(config.rng_seed)
+
+    # ------------------------------------------------------------------
+    # public API
+    # ------------------------------------------------------------------
+    def generate(self, overwrite: bool = False) -> ScenarioGenerationResult:
+        cfg = self.config
+        cfg.output_dir.mkdir(parents=True, exist_ok=True)
+        cfg.full_output_dir.mkdir(parents=True, exist_ok=True)
+        filename = self._build_filename()
+        data_path = cfg.output_dir / filename
+        full_path = cfg.full_output_dir / filename.replace(".p", "_full.p")
+
+        if data_path.exists() and full_path.exists() and not overwrite:
+            payload = self._load_pickle(data_path)
+            self.logger.info("Loaded cached scenario from %s", data_path)
+            return ScenarioGenerationResult(payload=payload, output_path=data_path, full_output_path=full_path, generated=False)
+
+        deepmimo_data = self._load_or_build_deepmimo()
+        path_exist = np.asarray(deepmimo_data["user"]["LoS"])
+        pos_total = np.asarray(deepmimo_data["user"]["location"])
+
+        grid_step = self._infer_grid_step(pos_total)
+        road_positions, lane_info = filter_road_positions(
+            pos_total[(path_exist == 0) | (path_exist == 1)], cfg.grid.road_width, cfg.grid.road_center_spacing
+        )
+        road_graph = create_grid_road_network(road_positions, lane_info, grid_step)
+        self.logger.info("Road graph has %d nodes", len(road_graph.nodes))
+
+        vehicle_trajs = generate_n_smooth_grid_trajectories(
+            road_graph,
+            road_positions,
+            cfg.traffic.num_vehicles,
+            sequence_length=cfg.sampling.time_steps,
+            turn_probability=cfg.traffic.turn_probability,
+            max_attempts=cfg.traffic.max_attempts,
+            step_size=grid_step,
+        )
+        pedestrian_trajs = generate_n_pedestrian_trajectories(
+            pos_total[(path_exist == 0) | (path_exist == 1)],
+            cfg.traffic.num_pedestrians,
+            sequence_length=cfg.sampling.time_steps,
+            step_size=grid_step,
+            angle_std=cfg.traffic.pedestrian_angle_std,
+        )
+
+        veh_idx = get_trajectory_indices(vehicle_trajs, pos_total)
+        ped_idx = get_trajectory_indices(pedestrian_trajs, pos_total)
+
+        car_tracks = self._build_tracks(vehicle_trajs, veh_idx, cfg.traffic.vehicle_speed_range, cfg.sampling)
+        ped_tracks = self._build_tracks(pedestrian_trajs, ped_idx, cfg.traffic.pedestrian_speed_range, cfg.sampling)
+
+        if cfg.export_environment_plot and cfg.figures_dir is not None:
+            self._save_environment_plot(
+                cfg.figures_dir,
+                pos_total,
+                path_exist,
+                road_positions=road_positions,
+                vehicle_trajs=[track["pos"] for track in car_tracks],
+                ped_trajs=[track["pos"] for track in ped_tracks],
+            )
+
+        channel_payload = self._channels_for_tracks(
+            deepmimo_data,
+            car_tracks + ped_tracks,
+            cfg.antenna,
+            cfg.carrier_frequency_hz,
+            cfg.sampling,
+        )
+
+        discrete_channels = np.zeros(
+            (
+                cfg.traffic.num_vehicles + cfg.traffic.num_pedestrians,
+                cfg.sampling.time_steps,
+                cfg.antenna.total_tx_elements(),
+                cfg.antenna.subcarriers,
+            ),
+            np.complex128,
+        )
+
+        payload: Dict[str, Any] = {
+            "scenario": cfg.scenario,
+            "index_discrete": veh_idx + ped_idx,
+            "grid_step": grid_step,
+            "sample_dt": cfg.sampling.sample_dt,
+            "car_speed_range": cfg.traffic.vehicle_speed_range,
+            "ped_speed_range": cfg.traffic.pedestrian_speed_range,
+            "los": path_exist,
+            "channel_cont": channel_payload["channel"],
+            "pos_cont": channel_payload["pos"],
+            "vel_cont": channel_payload["vel"],
+            "acc_cont": channel_payload["acc"],
+            "doppler_vel_cont": channel_payload["doppler"],
+            "angle_cont": channel_payload["angle"],
+            "delay_cont": channel_payload["delay"],
+            "pos_step_xy": channel_payload["step_xy"],
+            "channel_discrete": discrete_channels,
+            "pos_discrete": vehicle_trajs + pedestrian_trajs,
+        }
+
+        if cfg.sampling.continuous_mode:
+            payload.update(
+                {
+                    "channel": channel_payload["channel"],
+                    "pos": channel_payload["pos"],
+                    "vel": channel_payload["vel"],
+                    "acc": channel_payload["acc"],
+                    "doppler_vel": channel_payload["doppler"],
+                    "angle": channel_payload["angle"],
+                    "delay": channel_payload["delay"],
+                }
+            )
+        else:
+            payload["channel"] = discrete_channels
+
+        self._dump_pickle(data_path, payload["channel"])
+        self._dump_pickle(full_path, payload)
+        self.logger.info("Generated scenario saved to %s", data_path)
+        return ScenarioGenerationResult(payload=payload, output_path=data_path, full_output_path=full_path, generated=True)
+
+    # ------------------------------------------------------------------
+    # internal helpers
+    # ------------------------------------------------------------------
+    def _load_or_build_deepmimo(self) -> Dict[str, Any]:
+        cfg = self.config
+        return load_deepmimo_user_data(
+            cfg.scenario,
+            scenarios_dir=cfg.scenarios_dir,
+            load_params=cfg.deepmimo_load_params,
+            max_paths=cfg.deepmimo_max_paths,
+            array_dtype=cfg.deepmimo_array_dtype,
+            logger=self.logger,
+        )
+
+    def _save_environment_plot(
+        self,
+        figures_dir: Path,
+        positions: np.ndarray,
+        los: np.ndarray,
+        road_positions: Optional[np.ndarray] = None,
+        vehicle_trajs: Optional[Sequence[np.ndarray]] = None,
+        ped_trajs: Optional[Sequence[np.ndarray]] = None,
+    ) -> None:
+        figures_dir.mkdir(parents=True, exist_ok=True)
+        x, y = positions[:, 0], positions[:, 1]
+        los_array = np.asarray(los)
+        if los_array.ndim > 1:
+            N = x.shape[0]
+            shape = list(los_array.shape)
+            if N in shape:
+                axis_user = shape.index(N)
+                los_matrix = np.moveaxis(los_array, axis_user, -1).reshape(-1, N)
+                los_per_user = np.full(N, -1, np.int8)
+                any_los = (los_matrix == 1).any(axis=0)
+                los_per_user[any_los] = 1
+                any_nlos = (los_matrix == 0).any(axis=0) & ~any_los
+                los_per_user[any_nlos] = 0
+            else:
+                los_per_user = np.full(x.shape[0], -1, np.int8)
+        else:
+            los_per_user = los_array.astype(np.int8, copy=False)
+
+        colors = {
+            1: "#2ca8c2",   # teal for LoS
+            0: "#f39c12",   # amber for NLoS
+            -1: "#5d6d7e",  # muted slate for inactive users
+        }
+        plt.figure(figsize=(8, 8), dpi=500)
+        legend_handles = []
+        legend_labels = []
+        for value, color in colors.items():
+            mask = los_per_user == value
+            if np.any(mask):
+                handle = plt.scatter(x[mask], y[mask], s=3, c=color, alpha=0.6)
+                legend_handles.append(handle)
+                legend_labels.append(f"LoS={value}")
+
+        if road_positions is not None and len(road_positions):
+            rp = np.asarray(road_positions)
+            handle = plt.scatter(rp[:, 0], rp[:, 1], s=1, c="#a0a0a0", alpha=0.25)
+            legend_handles.append(handle)
+            legend_labels.append("Road nodes")
+
+        if vehicle_trajs:
+            cmap_vehicle = plt.cm.get_cmap("tab20", max(len(vehicle_trajs), 1))
+            for idx, traj in enumerate(vehicle_trajs):
+                arr = np.asarray(traj, dtype=float)
+                arr = arr.reshape(arr.shape[0], -1)
+                if arr.size == 0:
+                    continue
+                color = cmap_vehicle(idx % cmap_vehicle.N)
+                plt.plot(arr[:, 0], arr[:, 1], color=color, alpha=0.85, linewidth=1.2)
+                plt.scatter(
+                    arr[0:1, 0],
+                    arr[0:1, 1],
+                    s=20,
+                    marker="o",
+                    facecolor="white",
+                    edgecolor=color,
+                    linewidths=1.2,
+                    zorder=3,
+                )
+            legend_handles.append(Line2D([0], [0], color=cmap_vehicle(0), linewidth=1.2))
+            legend_labels.append("Vehicle trajectories")
+            legend_handles.append(Line2D([0], [0], marker="o", color="white", markeredgecolor=cmap_vehicle(0), markerfacecolor="white", linestyle="None", markersize=5, markeredgewidth=1.2))
+            legend_labels.append("Vehicle start")
+
+        if ped_trajs:
+            cmap_ped = plt.cm.get_cmap("Set2", max(len(ped_trajs), 1))
+            for idx, traj in enumerate(ped_trajs):
+                arr = np.asarray(traj, dtype=float)
+                arr = arr.reshape(arr.shape[0], -1)
+                if arr.size == 0:
+                    continue
+                color = cmap_ped(idx % cmap_ped.N)
+                plt.plot(arr[:, 0], arr[:, 1], color=color, alpha=0.85, linewidth=1.2, linestyle="--")
+                plt.scatter(
+                    arr[0:1, 0],
+                    arr[0:1, 1],
+                    s=20,
+                    marker="o",
+                    facecolor="white",
+                    edgecolor=color,
+                    linewidths=1.2,
+                    zorder=3,
+                )
+            legend_handles.append(Line2D([0], [0], color=cmap_ped(0), linewidth=1.2, linestyle="--"))
+            legend_labels.append("Pedestrian trajectories")
+            legend_handles.append(Line2D([0], [0], marker="o", color="white", markeredgecolor=cmap_ped(0), markerfacecolor="white", linestyle="None", markersize=5, markeredgewidth=1.2))
+            legend_labels.append("Pedestrian start")
+
+        if legend_handles:
+            plt.legend(legend_handles, legend_labels, loc="best", fontsize=8)
+        # plt.grid(True)
+        plt.gca().set_aspect("equal", adjustable="box")
+        plt.savefig(figures_dir / "environment.png", bbox_inches="tight", pad_inches=0.1)
+        plt.close()
+
+    def _infer_grid_step(self, positions: np.ndarray) -> float:
+        cfg = self.config
+        if cfg.grid.auto_step_size:
+            inferred = infer_grid_step(positions)
+            self.logger.info("Inferred road step size %.6f m", inferred)
+            return inferred
+        return float(cfg.grid.step_size)
+
+    def _build_tracks(
+        self,
+        trajectories: Sequence[np.ndarray],
+        indices: Sequence[Sequence[int]],
+        speed_range: Tuple[float, float],
+        sampling_cfg: ScenarioSamplingConfig,
+    ) -> List[Dict[str, Any]]:
+        tracks: List[Dict[str, Any]] = []
+        horizon = sampling_cfg.continuous_length or sampling_cfg.time_steps
+        for traj, idxs in zip(trajectories, indices):
+            speed = float(np.random.uniform(*speed_range))
+            idxs_list = list(idxs)
+            speed_profile = np.full(horizon, speed, float)
+            last_change = -1
+            for k in range(len(idxs_list) - 1):
+                if idxs_list[k] != idxs_list[k + 1]:
+                    last_change = k
+            if last_change == -1:
+                speed_profile[:] = 0.0
+            else:
+                last_move_sample = last_change + 1
+                if last_move_sample < horizon - 1:
+                    ramp = min(5, last_move_sample + 1)
+                    start_idx = max(0, last_move_sample - ramp + 1)
+                    speed_profile[:start_idx] = speed
+                    for k in range(start_idx, last_move_sample + 1):
+                        factor = float(last_move_sample - k + 1) / float(ramp)
+                        speed_profile[k] = speed * factor
+                    speed_profile[last_move_sample + 1 :] = 0.0
+            pos_c, pairs, alpha, vdir = sample_continuous_along_polyline(
+                traj,
+                idxs,
+                speed,
+                sampling_cfg.sample_dt,
+                horizon,
+                speed_profile=speed_profile,
+            )
+            tracks.append(
+                {
+                    "speed": speed,
+                    "speed_profile": speed_profile,
+                    "pos": pos_c,
+                    "pairs": pairs,
+                    "alpha": alpha,
+                    "vdir": vdir,
+                }
+            )
+        return tracks
+
+    def _channels_for_tracks(
+        self,
+        deepmimo_data: Dict[str, Any],
+        tracks: Sequence[Dict[str, Any]],
+        antenna_cfg: AntennaArrayConfig,
+        carrier_frequency_hz: float,
+        sampling_cfg: ScenarioSamplingConfig,
+    ) -> Dict[str, Any]:
+        channels: List[np.ndarray] = []
+        positions: List[np.ndarray] = []
+        velocities: List[np.ndarray] = []
+        accelerations: List[np.ndarray] = []
+        dopplers: List[List[np.ndarray]] = []
+        angles: List[List[np.ndarray]] = []
+        delays: List[List[np.ndarray]] = []
+        steps_xy: List[np.ndarray] = []
+        for track in tracks:
+            base_speed = track["speed"]
+            speed_profile = track.get("speed_profile")
+            pos = track["pos"]
+            pairs = track["pairs"]
+            alpha = track["alpha"]
+            vdir = track["vdir"]
+            horizon = len(alpha)
+            channel_sequence: List[np.ndarray] = []
+            doppler_list: List[np.ndarray] = []
+            angle_list: List[np.ndarray] = []
+            delay_list: List[np.ndarray] = []
+            step_xy = np.zeros(horizon, float)
+            step_xy[1:] = np.linalg.norm(pos[1:, :2] - pos[:-1, :2], axis=1)
+            vel_series = np.zeros(horizon, float)
+            if horizon > 1:
+                vel_series[1:] = step_xy[1:] / sampling_cfg.sample_dt
+            if vel_series[0] == 0.0:
+                if speed_profile is not None:
+                    vel_series[0] = float(speed_profile[0])
+                else:
+                    vel_series[0] = base_speed
+            acc_series = np.zeros_like(vel_series)
+            if horizon > 1:
+                acc_series[1:] = (vel_series[1:] - vel_series[:-1]) / sampling_cfg.sample_dt
+                acc_series[0] = acc_series[1] if horizon > 1 else 0.0
+            for idx, (pair, a, vd) in enumerate(zip(pairs, alpha, vdir)):
+                i0, i1 = pair
+                ray_interp = interpolate_ray_params(deepmimo_data, i0, i1, float(a))
+                if ray_interp["num_paths"] == 0:
+                    channel_sequence.append(np.zeros((antenna_cfg.total_tx_elements(), antenna_cfg.subcarriers), np.complex128))
+                    doppler_list.append(np.zeros((0,), np.float32))
+                    angle_list.append(np.zeros((0,), np.float32))
+                    delay_list.append(np.zeros((0,), np.float32))
+                    continue
+                doa_phi = np.deg2rad(ray_interp["DoA_phi"])
+                aoa_unit = np.stack([np.cos(doa_phi), np.sin(doa_phi)], axis=1)
+                speed_inst = vel_series[idx] if idx < vel_series.shape[0] else base_speed
+                v_proj = -np.sum(aoa_unit * vd[None, :], axis=1) * speed_inst
+                ray_interp["Doppler_vel"] = v_proj.astype(np.float32)
+                ray_interp["elapsed_time"] = np.ones_like(ray_interp["power"], dtype=np.float32) * (idx * sampling_cfg.sample_dt)
+                pred, _, _ = generate_channel_from_interpolated_ray(ray_interp, antenna_cfg, carrier_frequency_hz)
+                channel_sequence.append(np.asarray(pred[0]).squeeze(0))
+                doppler_list.append(v_proj.astype(np.float32))
+                angle_list.append(ray_interp["DoA_phi"].astype(np.float32))
+                delay_list.append(ray_interp["ToA"].astype(np.float32))
+            channels.append(np.stack(channel_sequence, axis=0))
+            positions.append(pos.astype(np.float32))
+            velocities.append(vel_series.astype(np.float32))
+            accelerations.append(acc_series.astype(np.float32))
+            dopplers.append(doppler_list)
+            angles.append(angle_list)
+            delays.append(delay_list)
+            steps_xy.append(step_xy.astype(np.float32))
+        return {
+            "channel": np.stack(channels, axis=0),
+            "pos": np.stack(positions, axis=0),
+            "vel": np.stack(velocities, axis=0),
+            "acc": np.stack(accelerations, axis=0),
+            "doppler": dopplers,
+            "angle": angles,
+            "delay": delays,
+            "step_xy": np.stack(steps_xy, axis=0),
+        }
+
+    def _build_filename(self) -> str:
+        cfg = self.config
+        return f"{cfg.scenario}_{cfg.sampling.time_steps}_{cfg.antenna.total_tx_elements()}_{cfg.antenna.subcarriers}.p"
+
+    @staticmethod
+    def _load_pickle(path: Path) -> Any:
+        with path.open("rb") as handle:
+            import pickle
+
+            return pickle.load(handle)
+
+    @staticmethod
+    def _dump_pickle(path: Path, obj: Any) -> None:
+        with path.open("wb") as handle:
+            import pickle
+
+            pickle.dump(obj, handle)
+
+
+# Backwards compatibility helper -------------------------------------------------
+
+def generate_dynamic_scenario_dataset(
+    config: ScenarioGenerationConfig,
+    overwrite: bool = False,
+    logger: Optional[logging.Logger] = None,
+) -> ScenarioGenerationResult:
+    generator = DynamicScenarioGenerator(config, logger=logger)
+    return generator.generate(overwrite=overwrite)
+
+
+def generate_dynamic_scenario(
+    config: ScenarioGenerationConfig,
+    overwrite: bool = False,
+    logger: Optional[logging.Logger] = None,
+) -> ScenarioGenerationResult:
+    warnings.warn(
+        "`generate_dynamic_scenario` is deprecated. Use `generate_dynamic_scenario_dataset` instead.",
+        DeprecationWarning,
+        stacklevel=2,
+    )
+    return generate_dynamic_scenario_dataset(config, overwrite=overwrite, logger=logger)
+
+
+__all__ = [
+    "AntennaArrayConfig",
+    "TrafficConfig",
+    "GridConfig",
+    "ScenarioSamplingConfig",
+    "ScenarioGenerationConfig",
+    "ScenarioGenerationResult",
+    "DynamicScenarioGenerator",
+    "generate_dynamic_scenario_dataset",
+    "generate_dynamic_scenario",
+]

docs/dynamic_scenario_pipeline.md

@@ -214,14 +214,14 @@ from LWMTemporal.data.scenario_generation import (
 config = ScenarioGenerationConfig(
     scenario="city_0_newyork_3p5",
     antenna=AntennaArrayConfig(tx_horizontal=32, tx_vertical=1, subcarriers=32),
-    sampling=ScenarioSamplingConfig(time_steps=40, sample_dt=0.1),
-    traffic=TrafficConfig(num_vehicles=120, num_pedestrians=20, turn_probability=0.08),
-    grid=GridConfig(road_width=6.0, road_center_spacing=25.0),
+    sampling=ScenarioSamplingConfig(time_steps=20, sample_dt=0.1),
+    traffic=TrafficConfig(num_vehicles=120, num_pedestrians=20, turn_probability=0.1),
+    grid=GridConfig(road_width=2.0, road_center_spacing=8.0),
     output_dir=Path("examples/data"),
     full_output_dir=Path("examples/full_data"),
     figures_dir=Path("examples/figs/newyork"),
     scenarios_dir=Path("deepmimo_scenarios"),
-    deepmimo_max_paths=6,
+    deepmimo_max_paths=25,
 )
 
 generator = DynamicScenarioGenerator(config)