forked from manbo/internal-docs
273 lines
8.1 KiB
Python
273 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Option A: "Synthetic ICS Data" mini-panel (high-level features, not packets)
|
||
|
||
What it draws (one SVG, transparent background):
|
||
- Top: 2–3 continuous feature curves (smooth, time-aligned)
|
||
- Bottom: discrete/categorical feature strip (colored blocks)
|
||
- One vertical dashed alignment line crossing both
|
||
- Optional shaded regime window
|
||
- Optional "real vs synthetic" ghost overlay (faint gray behind one curve)
|
||
|
||
Usage:
|
||
uv run python draw_synthetic_ics_optionA.py --out ./assets/synth_ics_optionA.svg
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
|
||
import numpy as np
|
||
import matplotlib.pyplot as plt
|
||
from matplotlib.patches import Rectangle
|
||
|
||
|
||
@dataclass
|
||
class Params:
|
||
seed: int = 7
|
||
seconds: float = 10.0
|
||
fs: int = 300
|
||
|
||
n_curves: int = 3 # continuous channels shown
|
||
n_bins: int = 40 # discrete blocks across x
|
||
disc_vocab: int = 8 # number of discrete categories
|
||
|
||
# Layout / style
|
||
width_in: float = 6.0
|
||
height_in: float = 2.2
|
||
curve_lw: float = 2.3
|
||
ghost_lw: float = 2.0 # "real" overlay line width
|
||
strip_height: float = 0.65 # bar height in [0,1] strip axis
|
||
strip_gap_frac: float = 0.10 # gap between blocks (fraction of block width)
|
||
|
||
# Visual cues
|
||
show_alignment_line: bool = True
|
||
align_x_frac: float = 0.58 # where to place dashed line, fraction of timeline
|
||
show_regime_window: bool = True
|
||
regime_start_frac: float = 0.30
|
||
regime_end_frac: float = 0.45
|
||
show_real_ghost: bool = True # faint gray "real" behind first synthetic curve
|
||
|
||
|
||
def _smooth(x: np.ndarray, win: int) -> np.ndarray:
|
||
win = max(3, int(win) | 1) # odd
|
||
k = np.ones(win, dtype=float)
|
||
k /= k.sum()
|
||
return np.convolve(x, k, mode="same")
|
||
|
||
|
||
def make_continuous_curves(p: Params) -> tuple[np.ndarray, np.ndarray, np.ndarray | None]:
|
||
"""
|
||
Returns:
|
||
t: (T,)
|
||
Y_syn: (n_curves, T) synthetic curves
|
||
y_real: (T,) or None optional "real" ghost curve (for one channel)
|
||
"""
|
||
rng = np.random.default_rng(p.seed)
|
||
T = int(p.seconds * p.fs)
|
||
t = np.linspace(0, p.seconds, T, endpoint=False)
|
||
|
||
Y = []
|
||
for i in range(p.n_curves):
|
||
# multi-scale smooth temporal patterns
|
||
f_slow = 0.09 + 0.03 * (i % 3)
|
||
f_mid = 0.65 + 0.18 * (i % 4)
|
||
ph = rng.uniform(0, 2 * np.pi)
|
||
|
||
y = (
|
||
0.95 * np.sin(2 * np.pi * f_slow * t + ph)
|
||
+ 0.30 * np.sin(2 * np.pi * f_mid * t + 0.7 * ph)
|
||
)
|
||
|
||
# regime-like bumps
|
||
bumps = np.zeros_like(t)
|
||
for _ in range(2):
|
||
mu = rng.uniform(0.8, p.seconds - 0.8)
|
||
sig = rng.uniform(0.35, 0.85)
|
||
bumps += np.exp(-0.5 * ((t - mu) / (sig + 1e-9)) ** 2)
|
||
y += 0.55 * bumps
|
||
|
||
# mild smooth noise
|
||
noise = _smooth(rng.normal(0, 1, size=T), win=int(p.fs * 0.04))
|
||
y += 0.10 * noise
|
||
|
||
# normalize for clean presentation
|
||
y = (y - y.mean()) / (y.std() + 1e-9)
|
||
y *= 0.42
|
||
Y.append(y)
|
||
|
||
Y_syn = np.vstack(Y)
|
||
|
||
# Optional "real" ghost: similar to first curve, but slightly different
|
||
y_real = None
|
||
if p.show_real_ghost:
|
||
base = Y_syn[0].copy()
|
||
drift = _smooth(rng.normal(0, 1, size=T), win=int(p.fs * 0.18))
|
||
drift = drift / (np.std(drift) + 1e-9)
|
||
y_real = base * 0.95 + 0.07 * drift
|
||
|
||
return t, Y_syn, y_real
|
||
|
||
|
||
def make_discrete_strip(p: Params) -> np.ndarray:
|
||
"""
|
||
Piecewise-constant categorical IDs across n_bins.
|
||
Returns:
|
||
ids: (n_bins,) in [0, disc_vocab-1]
|
||
"""
|
||
rng = np.random.default_rng(p.seed + 123)
|
||
n = p.n_bins
|
||
ids = np.zeros(n, dtype=int)
|
||
|
||
cur = rng.integers(0, p.disc_vocab)
|
||
for i in range(n):
|
||
# occasional change
|
||
if i == 0 or rng.random() < 0.28:
|
||
cur = rng.integers(0, p.disc_vocab)
|
||
ids[i] = cur
|
||
|
||
return ids
|
||
|
||
|
||
def _axes_clean(ax: plt.Axes) -> None:
|
||
"""Keep axes lines optional but remove all text/numbers (diagram-friendly)."""
|
||
ax.set_xlabel("")
|
||
ax.set_ylabel("")
|
||
ax.set_title("")
|
||
ax.set_xticks([])
|
||
ax.set_yticks([])
|
||
ax.tick_params(
|
||
axis="both",
|
||
which="both",
|
||
bottom=False,
|
||
left=False,
|
||
top=False,
|
||
right=False,
|
||
labelbottom=False,
|
||
labelleft=False,
|
||
)
|
||
|
||
|
||
def draw_optionA(out_path: Path, p: Params) -> None:
|
||
# Figure
|
||
fig = plt.figure(figsize=(p.width_in, p.height_in), dpi=200)
|
||
fig.patch.set_alpha(0.0)
|
||
|
||
# Two stacked axes (shared x)
|
||
ax_top = fig.add_axes([0.08, 0.32, 0.90, 0.62])
|
||
ax_bot = fig.add_axes([0.08, 0.12, 0.90, 0.16], sharex=ax_top)
|
||
ax_top.patch.set_alpha(0.0)
|
||
ax_bot.patch.set_alpha(0.0)
|
||
|
||
# Generate data
|
||
t, Y_syn, y_real = make_continuous_curves(p)
|
||
ids = make_discrete_strip(p)
|
||
|
||
x0, x1 = float(t[0]), float(t[-1])
|
||
span = x1 - x0
|
||
|
||
# Optional shaded regime window
|
||
if p.show_regime_window:
|
||
rs = x0 + p.regime_start_frac * span
|
||
re = x0 + p.regime_end_frac * span
|
||
ax_top.axvspan(rs, re, alpha=0.12) # default color, semi-transparent
|
||
ax_bot.axvspan(rs, re, alpha=0.12)
|
||
|
||
# Optional vertical dashed alignment line
|
||
if p.show_alignment_line:
|
||
vx = x0 + p.align_x_frac * span
|
||
ax_top.axvline(vx, linestyle="--", linewidth=1.2, alpha=0.7)
|
||
ax_bot.axvline(vx, linestyle="--", linewidth=1.2, alpha=0.7)
|
||
|
||
# Continuous curves (use fixed colors for consistency)
|
||
curve_colors = ["#1f77b4", "#ff7f0e", "#2ca02c", "#9467bd"] # blue, orange, green, purple
|
||
|
||
# Ghost "real" behind the first curve (faint gray)
|
||
if y_real is not None:
|
||
ax_top.plot(t, y_real, linewidth=p.ghost_lw, color="0.65", alpha=0.55, zorder=1)
|
||
|
||
for i in range(Y_syn.shape[0]):
|
||
ax_top.plot(
|
||
t, Y_syn[i],
|
||
linewidth=p.curve_lw,
|
||
color=curve_colors[i % len(curve_colors)],
|
||
zorder=2
|
||
)
|
||
|
||
# Set top y-limits with padding
|
||
ymin, ymax = float(Y_syn.min()), float(Y_syn.max())
|
||
ypad = 0.10 * (ymax - ymin + 1e-9)
|
||
ax_top.set_xlim(x0, x1)
|
||
ax_top.set_ylim(ymin - ypad, ymax + ypad)
|
||
|
||
# Discrete strip as colored blocks
|
||
palette = [
|
||
"#e41a1c", "#377eb8", "#4daf4a", "#984ea3",
|
||
"#ff7f00", "#ffff33", "#a65628", "#f781bf",
|
||
]
|
||
|
||
n = len(ids)
|
||
bin_w = span / n
|
||
gap = p.strip_gap_frac * bin_w
|
||
ax_bot.set_ylim(0, 1)
|
||
|
||
y = (1 - p.strip_height) / 2
|
||
for i, cat in enumerate(ids):
|
||
left = x0 + i * bin_w + gap / 2
|
||
width = bin_w - gap
|
||
ax_bot.add_patch(
|
||
Rectangle(
|
||
(left, y), width, p.strip_height,
|
||
facecolor=palette[int(cat) % len(palette)],
|
||
edgecolor="none",
|
||
)
|
||
)
|
||
|
||
# Clean axes: no ticks/labels; keep spines (axes lines) visible
|
||
_axes_clean(ax_top)
|
||
_axes_clean(ax_bot)
|
||
for ax in (ax_top, ax_bot):
|
||
for side in ("left", "bottom", "top", "right"):
|
||
ax.spines[side].set_visible(True)
|
||
|
||
out_path.parent.mkdir(parents=True, exist_ok=True)
|
||
fig.savefig(out_path, format="svg", transparent=True, bbox_inches="tight", pad_inches=0.0)
|
||
plt.close(fig)
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--out", type=Path, default=Path("synth_ics_optionA.svg"))
|
||
ap.add_argument("--seed", type=int, default=7)
|
||
ap.add_argument("--seconds", type=float, default=10.0)
|
||
ap.add_argument("--fs", type=int, default=300)
|
||
ap.add_argument("--curves", type=int, default=3)
|
||
ap.add_argument("--bins", type=int, default=40)
|
||
ap.add_argument("--vocab", type=int, default=8)
|
||
|
||
ap.add_argument("--no-align", action="store_true")
|
||
ap.add_argument("--no-regime", action="store_true")
|
||
ap.add_argument("--no-ghost", action="store_true")
|
||
args = ap.parse_args()
|
||
|
||
p = Params(
|
||
seed=args.seed,
|
||
seconds=args.seconds,
|
||
fs=args.fs,
|
||
n_curves=args.curves,
|
||
n_bins=args.bins,
|
||
disc_vocab=args.vocab,
|
||
show_alignment_line=not args.no_align,
|
||
show_regime_window=not args.no_regime,
|
||
show_real_ghost=not args.no_ghost,
|
||
)
|
||
|
||
draw_optionA(args.out, p)
|
||
print(f"Wrote: {args.out}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|