""" Flow Matching Animation: OT-based straight-line transport from noise to data @fminxyz Series 2, Post 4 — 7 марта 2026 1080x1080 px, 2 fps, 25 сек (50 frames) """ import numpy as np import matplotlib.pyplot as plt import matplotlib.animation as animation from matplotlib.colors import LinearSegmentedColormap OUTPUT = "/root/Strategy/content/drafts/flow_matching_animation.mp4" FPS = 2 N_FRAMES = 50 rng = np.random.default_rng(123) # ── Data distribution: two moons ───────────────────────────────────────────── N_PART = 60 def make_moons(n, noise=0.12): half = n // 2 # Moon 1 (top) angles1 = rng.uniform(0, np.pi, half) x1 = np.cos(angles1) * 1.8 - 0.9 y1 = np.sin(angles1) * 1.5 + 0.3 # Moon 2 (bottom, flipped) angles2 = rng.uniform(0, np.pi, n - half) x2 = np.cos(angles2) * 1.8 + 0.9 y2 = -np.sin(angles2) * 1.5 - 0.3 x = np.concatenate([x1, x2]) + rng.normal(0, noise, n) y = np.concatenate([y1, y2]) + rng.normal(0, noise, n) labels = np.array([0] * half + [1] * (n - half)) return x, y, labels # Target distribution (data) x_data, y_data, labels = make_moons(N_PART) data_pts = np.stack([x_data, y_data], axis=1) # Source: pure Gaussian noise noise_pts = rng.normal(0, 1.2, (N_PART, 2)) # ── OT Flow Matching: straight lines from noise to data ────────────────────── # In OT-FM: x(t) = (1-t)*x0 + t*x1 (linear interpolation) # Velocity field: u_t(x) = x1 - x0 (constant per path) def interpolate(t): """Linear interpolation at time t in [0, 1]""" return (1 - t) * noise_pts + t * data_pts # Precompute all positions T_VALUES = np.linspace(0, 1, N_FRAMES) all_positions = np.array([interpolate(t) for t in T_VALUES]) # ── Velocity field on grid ─────────────────────────────────────────────────── # Approximate velocity field by binning / averaging over particles # v_t(x) ≈ average (x1 - x0) for nearby particles at time t x_grid = np.linspace(-3.5, 3.5, 14) y_grid = np.linspace(-3, 3, 12) Xg, Yg = np.meshgrid(x_grid, y_grid) def velocity_field_at_t(t, sigma_rbf=1.0): """Approximate velocity field at time t using RBF weighting""" positions_t = interpolate(t) # (N, 2) velocities = data_pts - noise_pts # (N, 2) constant Vx = np.zeros_like(Xg) Vy = np.zeros_like(Yg) for gi in range(Xg.shape[0]): for gj in range(Xg.shape[1]): gx, gy = Xg[gi, gj], Yg[gi, gj] dists = (positions_t[:, 0] - gx)**2 + (positions_t[:, 1] - gy)**2 weights = np.exp(-dists / (2 * sigma_rbf**2)) w_sum = weights.sum() + 1e-10 Vx[gi, gj] = (weights * velocities[:, 0]).sum() / w_sum Vy[gi, gj] = (weights * velocities[:, 1]).sum() / w_sum return Vx, Vy # Precompute velocity fields at a few key times KEY_TIMES = [0.0, 0.25, 0.5, 0.75, 1.0] vel_fields = {t: velocity_field_at_t(t, sigma_rbf=0.8) for t in KEY_TIMES} def get_vel_field(t_val): closest = min(KEY_TIMES, key=lambda k: abs(k - t_val)) return vel_fields[closest] # ── Colors ─────────────────────────────────────────────────────────────────── bg_color = '#0a0a14' COLOR_MOON1 = '#4fc3f7' # blue moon COLOR_MOON2 = '#f06292' # pink moon COLORS = np.where(labels == 0, COLOR_MOON1, COLOR_MOON2) # ── Figure ─────────────────────────────────────────────────────────────────── DPI = 108 fig, ax = plt.subplots(figsize=(10, 10), dpi=DPI, facecolor=bg_color) def draw_frame(i): ax.clear() ax.set_facecolor(bg_color) ax.set_xlim(-4, 4) ax.set_ylim(-3.5, 3.5) ax.set_aspect('equal') ax.axis('off') t = T_VALUES[i] pos = all_positions[i] # ── Phase logic ────────────────────────────────────────────────────────── # Phase 1: frames 0-10 → noise source # Phase 2: frames 11-35 → transport (straight lines visible) # Phase 3: frames 36-49 → data distribution reached fade_in = min(1.0, (i + 1) / 5) # Show straight-line paths (always, with varying alpha) if i >= 5: path_alpha = min(0.35, (i - 5) * 0.025) for p_idx in range(N_PART): ax.plot([noise_pts[p_idx, 0], data_pts[p_idx, 0]], [noise_pts[p_idx, 1], data_pts[p_idx, 1]], '-', color=COLORS[p_idx], alpha=path_alpha, linewidth=0.8, zorder=2) # Current positions for p_idx in range(N_PART): ax.plot(pos[p_idx, 0], pos[p_idx, 1], 'o', color=COLORS[p_idx], markersize=9, alpha=0.92, zorder=5, markeredgecolor='white', markeredgewidth=0.5) # Velocity field arrows if 8 <= i <= 42: vel_alpha = 0.0 if i < 15: vel_alpha = (i - 8) / 7 * 0.5 elif i < 35: vel_alpha = 0.5 else: vel_alpha = (42 - i) / 7 * 0.5 Vx, Vy = get_vel_field(t) V_mag = np.sqrt(Vx**2 + Vy**2 + 1e-12) Vx_n = Vx / V_mag Vy_n = Vy / V_mag ax.quiver(Xg, Yg, Vx_n, Vy_n, color='#aaaaff', alpha=vel_alpha * 0.7, scale=18, headwidth=3.5, headlength=4.5, width=0.003, zorder=3) # Phase labels if i < 12: ax.text(0, -3.1, 'x₀ ~ N(0, I) — чистый шум', ha='center', fontsize=14, color='#8899cc', fontfamily='monospace', alpha=fade_in) elif i < 38: progress_t = t ax.text(0, -3.1, f'x(t) = (1−t)·x₀ + t·x₁ t={progress_t:.2f}', ha='center', fontsize=13, color='#aaaaff', fontfamily='monospace') else: fade_end = min(1.0, (i - 38) / 6) ax.text(0, -3.1, 'x₁ ~ p_data — данные достигнуты ✓', ha='center', fontsize=14, color='#66ff88', fontfamily='monospace', alpha=fade_end) # Title ax.text(0, 3.2, 'Flow Matching: прямые траектории', ha='center', fontsize=17, fontweight='bold', color='white', fontfamily='monospace', alpha=fade_in) # Subtitle: OT note if i >= 15: ax.text(0, 2.8, 'OT = оптимальный транспорт → прямые линии', ha='center', fontsize=12, color='#777799', fontfamily='monospace') # Progress bar bar_w = 8 * t ax.barh(-3.35, bar_w, height=0.08, left=-4, color='#4466ff', alpha=0.6, zorder=4) ax.barh(-3.35, 8, height=0.08, left=-4, color='#223355', alpha=0.4, zorder=3) # Frame counter ax.text(3.8, -3.3, f't={i:02d}', ha='right', fontsize=11, color='#445566', fontfamily='monospace') anim = animation.FuncAnimation(fig, draw_frame, frames=N_FRAMES, interval=1000 // FPS) anim.save(OUTPUT, writer='ffmpeg', fps=FPS, dpi=DPI, extra_args=['-vcodec', 'libx264', '-pix_fmt', 'yuv420p', '-crf', '22', '-preset', 'fast']) plt.close() print(f"Saved: {OUTPUT}")