Spaces:
Sleeping
Sleeping
| """Delay-embedded image rendering for time series. | |
| Each 1-D series is turned into a (H, W) matrix by sliding a fixed-length window across the | |
| series with a step proportional to (length - window) / (W - 1). The window | |
| values are interpolated to fill H rows. The 2-D map is normalised to [0, 1] | |
| using min/max and broadcast across 3 channels. | |
| Output is a **(3, H, W) float32 tensor in [0, 1]** — no uint8 quantisation, | |
| no PIL round-trip. The collators feed it straight to the DINO normaliser | |
| ((x - mean) / std) without any /255 step. `delay_tensor_to_pil` is provided | |
| for visualisation paths that still want a PIL image. | |
| Normalisation scope: | |
| - Default: per-series (uses the current 1-D array's own min/max). | |
| - With `vmin/vmax` overrides (`channel_stats`, a {field -> {vmin, vmax}} | |
| JSON given explicitly via `channel_stats_path`): the per-channel min/max | |
| is fixed across the dataset. Off in every shipped config and evaluation. | |
| For TimeSeriesExam examples that carry multiple series (ts1+ts2), we emit one | |
| delay-embedded tensor per series (paired examples -> 2 tensors). | |
| """ | |
| from __future__ import annotations | |
| import math | |
| import os | |
| from typing import Dict, List, Optional, Tuple | |
| import numpy as np | |
| import torch | |
| from PIL import Image | |
| from src import paths | |
| DEFAULT_HW = 256 | |
| DEFAULT_EMBED_RATIO = 0.6 | |
| DEFAULT_EMBED_LMIN = 48 | |
| DEFAULT_EMBED_LMAX = 192 | |
| def _to_1d(ts) -> np.ndarray: | |
| arr = np.asarray(ts, dtype=np.float32).squeeze() | |
| if arr.ndim == 0: | |
| arr = arr[None] | |
| if arr.ndim != 1: | |
| raise ValueError(f"expected 1-D series, got shape {arr.shape}") | |
| return arr | |
| def _delay_embed_2d( | |
| x: np.ndarray, | |
| height: int, | |
| width: int, | |
| embed_ratio: float, | |
| embed_lmin: int, | |
| embed_lmax: int, | |
| ) -> np.ndarray: | |
| if x.size == 0: | |
| x = np.zeros(height, dtype=np.float32) | |
| length = int(x.size) | |
| raw_l = int(math.floor(embed_ratio * length)) | |
| l = min(embed_lmax, max(embed_lmin, raw_l)) | |
| l = max(1, min(l, length)) | |
| max_start = max(0, length - l) | |
| delay = (max_start / float(width - 1)) if width > 1 else 0.0 | |
| out = np.empty((height, width), dtype=np.float32) | |
| src = np.linspace(0.0, 1.0, num=l, dtype=np.float32) | |
| dst = np.linspace(0.0, 1.0, num=height, dtype=np.float32) | |
| for col in range(width): | |
| start = int(round(col * delay)) | |
| if start > max_start: | |
| start = max_start | |
| window = x[start : start + l] | |
| if l == height: | |
| out[:, col] = window | |
| elif l == 1: | |
| out[:, col] = window[0] | |
| else: | |
| out[:, col] = np.interp(dst, src, window).astype(np.float32) | |
| return out | |
| def render_single_delay( | |
| ts, | |
| base_height: int = DEFAULT_HW, | |
| base_width: int = DEFAULT_HW, | |
| embed_ratio: float = DEFAULT_EMBED_RATIO, | |
| embed_lmin: int = DEFAULT_EMBED_LMIN, | |
| embed_lmax: int = DEFAULT_EMBED_LMAX, | |
| vmin: Optional[float] = None, | |
| vmax: Optional[float] = None, | |
| ) -> torch.Tensor: | |
| """Render one 1-D series to a delay-embedded RGB tensor in float32 [0, 1]. | |
| Returns a (3, H, W) float32 tensor. No uint8 quantisation, no PIL — the | |
| collators apply DINO's (x - mean) / std directly. | |
| If `vmin` and `vmax` are provided, use them for the [0,1] rescale (matches | |
| the original repo, where per-channel min/max are computed once across the | |
| whole dataset). Otherwise fall back to per-series min/max. | |
| """ | |
| x = _to_1d(ts) | |
| if vmin is None or vmax is None: | |
| vmin = float(np.nanmin(x)) if np.any(np.isfinite(x)) else 0.0 | |
| vmax = float(np.nanmax(x)) if np.any(np.isfinite(x)) else 1.0 | |
| if vmax > vmin: | |
| xn = (x - vmin) / (vmax - vmin) | |
| else: | |
| xn = np.zeros_like(x) | |
| xn = np.clip(xn, 0.0, 1.0) | |
| x2d = _delay_embed_2d(xn, base_height, base_width, embed_ratio, embed_lmin, embed_lmax) | |
| # (H, W) -> (3, H, W) float32 in [0, 1] | |
| t = torch.from_numpy(x2d).to(torch.float32) | |
| return t.unsqueeze(0).expand(3, -1, -1).contiguous() | |
| def render_for_example_delay( | |
| example: dict, | |
| channel_stats: Optional[Dict[str, Dict[str, float]]] = None, | |
| **kwargs, | |
| ) -> List[torch.Tensor]: | |
| """Return one (3, H, W) float32 tensor per time-series in the example. | |
| Inspects standard TimeSeriesExam fields: | |
| - `ts` -> [one tensor] | |
| - `ts1`,`ts2` -> [t1, t2] | |
| - fallback: any field whose value looks like a numeric 1-D series. | |
| `channel_stats`, when given, maps field name -> {"vmin": float, "vmax": float} | |
| and overrides the per-series min/max so all examples share a per-field scale. | |
| """ | |
| ts1 = example.get("ts1") | |
| ts2 = example.get("ts2") | |
| ts = example.get("ts") | |
| named_series: List[Tuple[str, object]] = [] | |
| if ts1 is not None and len(ts1): | |
| named_series.append(("ts1", ts1)) | |
| if ts2 is not None and len(ts2): | |
| named_series.append(("ts2", ts2)) | |
| if not named_series and ts is not None and len(ts): | |
| named_series.append(("ts", ts)) | |
| if not named_series: | |
| for k, v in example.items(): | |
| if isinstance(v, (list, tuple)) and len(v) > 8 and all( | |
| isinstance(x, (int, float)) for x in v[:5] | |
| ): | |
| named_series.append((k, v)) | |
| images: List[torch.Tensor] = [] | |
| for name, s in named_series: | |
| if channel_stats and name in channel_stats: | |
| stats = channel_stats[name] | |
| images.append(render_single_delay( | |
| s, vmin=stats["vmin"], vmax=stats["vmax"], **kwargs, | |
| )) | |
| else: | |
| images.append(render_single_delay(s, **kwargs)) | |
| return images | |
| def delay_tensor_to_pil(t: torch.Tensor) -> Image.Image: | |
| """Visualisation helper: (3, H, W) float32 [0, 1] tensor -> PIL.RGB. | |
| Used by preview / sample-HTML scripts that need a PIL image for `.save()`. | |
| Loses the same precision the old pipeline did; the training path never | |
| touches this — it stays in float.""" | |
| if t.ndim == 3 and t.shape[0] == 3: | |
| arr = t.permute(1, 2, 0).cpu().float().clamp(0.0, 1.0).numpy() | |
| elif t.ndim == 2: | |
| arr = t.cpu().float().clamp(0.0, 1.0).numpy() | |
| arr = np.stack([arr] * 3, axis=-1) | |
| else: | |
| raise ValueError(f"unexpected shape: {tuple(t.shape)}") | |
| return Image.fromarray((arr * 255.0).astype(np.uint8)) | |
| if __name__ == "__main__": | |
| import os | |
| out = paths.resolve("results/preview_delay") | |
| os.makedirs(out, exist_ok=True) | |
| rng = np.random.default_rng(0) | |
| for i in range(3): | |
| ts = np.cumsum(rng.standard_normal(256)) + np.sin(np.linspace(0, 6 * np.pi, 256)) | |
| t = render_single_delay(ts) | |
| delay_tensor_to_pil(t).save(f"{out}/delay_{i}.jpg") | |
| print(f"Wrote delay-embedding previews to {out}") | |