"""Depth aware 3D drawing canvas.""" from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime from pathlib import Path import cv2 import numpy as np from canvas import PALETTE from smoothing import OneEuroFilter MAX_JUMP = 0.12 MIN_STEP = 2.0 DEPTH_SPAN = 1.15 DEPTH_PX_PER_METER = 1.7 DEPTH_MIN_METERS = 0.12 DEPTH_BAND = 0.015 DEPTH_MAX_SPEED = 1.1 DEPTH_CUTOFF = 0.8 DEPTH_BETA = 4.0 DEPTH_STEP = 5.0 GRADIENT_SPAN = 0.5 GRADIENT_LO = 40 GRADIENT_HI = 218 TAIL = 7 NEAR_GAIN = 1.35 FAR_GAIN = 0.55 GRADIENT_NEAR_GAIN = 1.12 GRADIENT_FAR_GAIN = 0.80 _TURBO: np.ndarray = cv2.applyColorMap( np.arange(256, dtype=np.uint8).reshape(-1, 1), cv2.COLORMAP_TURBO ).reshape(-1, 3) @dataclass class View: """Camera orientation and zoom.""" yaw: float = 0.0 pitch: float = 0.0 roll: float = 0.0 scale: float = 1.0 @dataclass class Stroke3D: """One stroke in world space.""" color: tuple[int, int, int] thickness: int gradient: bool = False view: View = field(default_factory=View) points: list[tuple[float, float, float]] = field(default_factory=list) screen: list[tuple[float, float]] = field(default_factory=list) class DepthPen: """Noise tolerant metric depth.""" def __init__(self, band: float = DEPTH_BAND, max_speed: float = DEPTH_MAX_SPEED) -> None: self.band = band self.max_speed = max_speed self.filter = OneEuroFilter(min_cutoff=DEPTH_CUTOFF, beta=DEPTH_BETA) self.value: float | None = None self.raw: float = 0.0 def reset(self) -> None: """Drop tracking state.""" self.filter.reset() self.value = None def __call__(self, depth_meters: float, dt: float) -> float | None: """Filter one depth sample.""" d = float(depth_meters) if not np.isfinite(d) or d < DEPTH_MIN_METERS: return self.value step = max(float(dt), 1e-3) self.raw = d smooth = float(self.filter(np.array([d], dtype=np.float64), step)[0]) if self.value is None: self.value = smooth return self.value delta = smooth - self.value if abs(delta) <= self.band: return self.value move = np.sign(delta) * (abs(delta) - self.band) limit = self.max_speed * step self.value += float(np.clip(move, -limit, limit)) return self.value def rotation(yaw: float, pitch: float, roll: float) -> np.ndarray: """Build rotation matrix.""" cy, sy = np.cos(yaw), np.sin(yaw) cp, sp = np.cos(pitch), np.sin(pitch) cr, sr = np.cos(roll), np.sin(roll) ry = np.array([[cy, 0.0, sy], [0.0, 1.0, 0.0], [-sy, 0.0, cy]]) rx = np.array([[1.0, 0.0, 0.0], [0.0, cp, -sp], [0.0, sp, cp]]) rz = np.array([[cr, -sr, 0.0], [sr, cr, 0.0], [0.0, 0.0, 1.0]]) return rz @ ry @ rx class Canvas3D: """Vector canvas with per point depth.""" def __init__(self, width: int, height: int, output_dir: str | Path = "output") -> None: self.width = width self.height = height self.output_dir = Path(output_dir) self.strokes: list[Stroke3D] = [] self._active: Stroke3D | None = None self.color: tuple[int, int, int] = PALETTE[0] self.gradient = True self.thickness: int = 6 self.revision = 0 self.depth_span = float(height) * DEPTH_SPAN self.gradient_span = float(height) * GRADIENT_SPAN self.distance = float(height) * 2.2 self.focal = self.distance self.px_per_meter = float(height) * DEPTH_PX_PER_METER self.depth_ref: float | None = None self.last_depth: float = 0.0 self.last_z: float = 0.0 self.view = View() self.pen = DepthPen() self._last_screen: tuple[float, float] | None = None self._last_z = 0.0 def set_view(self, view: View) -> None: """Store the current camera.""" self.view = view def depth_to_world_z(self, depth_meters: float, dt: float) -> float: """Map metric depth to world Z.""" held = self.pen(depth_meters, dt) if held is None: return self.last_z if self.depth_ref is None: self.depth_ref = held self.last_depth = held self.last_z = float(np.clip((held - self.depth_ref) * self.px_per_meter, -self.depth_span, self.depth_span)) return self.last_z def depth_color(self, world_z: float) -> tuple[int, int, int]: """Gradient color for a depth.""" t = float(np.clip((world_z + self.gradient_span) / (2.0 * self.gradient_span), 0.0, 1.0)) b, g, r = _TURBO[int(round(GRADIENT_LO + (1.0 - t) * (GRADIENT_HI - GRADIENT_LO)))] return int(b), int(g), int(r) def pen_color(self) -> tuple[int, int, int]: """Color the pen draws with now.""" return self.depth_color(self.last_z) if self.gradient else self.color def use_gradient(self) -> None: """Color strokes by depth.""" self.gradient = True def _matrix(self, view: View | None = None) -> np.ndarray: """Scaled rotation matrix.""" v = view or self.view return rotation(v.yaw, v.pitch, v.roll) * max(v.scale, 1e-3) def project(self, points: np.ndarray, view: View | None = None) -> tuple[np.ndarray, np.ndarray]: """Project world points to screen.""" pts = np.asarray(points, dtype=np.float64).reshape(-1, 3) cam = pts @ self._matrix(view).T z = np.maximum(cam[:, 2] + self.distance, 1e-3) cx, cy = self.width * 0.5, self.height * 0.5 sx = cx + self.focal * cam[:, 0] / z sy = cy + self.focal * cam[:, 1] / z return np.stack([sx, sy], axis=1), z def unproject(self, screen: np.ndarray, world_z: float, view: View | None = None) -> tuple[float, float, float]: """Screen point at known depth to world.""" a = self._matrix(view) cx, cy = self.width * 0.5, self.height * 0.5 u = float(screen[0]) - cx v = float(screen[1]) - cy f = self.focal m = np.array( [ [f * a[0, 0] - u * a[2, 0], f * a[0, 1] - u * a[2, 1]], [f * a[1, 0] - v * a[2, 0], f * a[1, 1] - v * a[2, 1]], ], dtype=np.float64, ) rhs = np.array( [ u * (a[2, 2] * world_z + self.distance) - f * a[0, 2] * world_z, v * (a[2, 2] * world_z + self.distance) - f * a[1, 2] * world_z, ], dtype=np.float64, ) det = m[0, 0] * m[1, 1] - m[0, 1] * m[1, 0] if abs(det) < 1e-9: return 0.0, 0.0, world_z x, y = np.linalg.solve(m, rhs) return float(x), float(y), float(world_z) def begin(self, view: View | None = None) -> None: """Start a new stroke.""" if self._active is None: self._active = Stroke3D(color=self.color, thickness=self.thickness, gradient=self.gradient, view=view or self.view) self.strokes.append(self._active) self._last_screen = None def add_point(self, screen_point, depth_meters: float, dt: float, view: View | None = None) -> None: """Append a screen point with metric depth.""" if self._active is None: self.begin(view) assert self._active is not None sx = float(np.clip(float(screen_point[0]), 0.0, self.width - 1.0)) sy = float(np.clip(float(screen_point[1]), 0.0, self.height - 1.0)) z = self.depth_to_world_z(depth_meters, dt) if self._last_screen is not None: jump = float(np.hypot(sx - self._last_screen[0], sy - self._last_screen[1])) if jump < MIN_STEP and abs(z - self._last_z) < DEPTH_STEP: return if jump > MAX_JUMP * float(np.hypot(self.width, self.height)): self.end() self.begin(view) assert self._active is not None stroke = self._active stroke.points.append(self.unproject(np.array([sx, sy]), z, stroke.view)) stroke.screen.append((sx, sy)) self._smooth_tail() self._last_screen = (sx, sy) self._last_z = z self.revision += 1 def _smooth_tail(self) -> None: """Low pass the newest depths.""" stroke = self._active if stroke is None or len(stroke.points) < 3: return n = len(stroke.points) zs = [p[2] for p in stroke.points] for i in range(max(1, n - TAIL), n - 1): z = 0.25 * zs[i - 1] + 0.5 * zs[i] + 0.25 * zs[i + 1] if abs(z - zs[i]) < 1e-3: continue stroke.points[i] = self.unproject(np.array(stroke.screen[i]), z, stroke.view) def end(self) -> None: """Finish current stroke.""" if self._active is not None and not self._active.points: self.strokes.remove(self._active) self.revision += 1 self._active = None self._last_screen = None self.pen.reset() def undo(self) -> None: """Remove last stroke.""" if self.strokes: self.strokes.pop() self._active = None self._last_screen = None self.revision += 1 def clear(self) -> None: """Erase everything.""" self.strokes.clear() self._active = None self._last_screen = None self._last_z = 0.0 self.pen.reset() self.depth_ref = None self.revision += 1 def set_color(self, index: int) -> None: """Select palette color.""" self.color = PALETTE[index % len(PALETTE)] self.gradient = False def set_thickness(self, value: int) -> None: """Set stroke thickness.""" self.thickness = int(max(1, min(48, value))) def scale_content(self, factor: float) -> None: """Scale drawing about origin.""" if not self.strokes or abs(factor - 1.0) < 1e-3: return for s in self.strokes: s.points = [(x * factor, y * factor, z * factor) for x, y, z in s.points] s.thickness = int(max(1, min(48, round(s.thickness * factor)))) self._active = None self._last_screen = None self.revision += 1 @property def is_empty(self) -> bool: """No strokes drawn.""" return not any(s.points for s in self.strokes) def all_points(self) -> np.ndarray: """Every world point.""" pts = [p for s in self.strokes for p in s.points] if not pts: return np.zeros((0, 3), dtype=np.float64) return np.asarray(pts, dtype=np.float64) def content_span(self, view: View | None = None) -> float: """Projected diagonal length.""" pts = self.all_points() if pts.shape[0] == 0: return 0.0 proj, _ = self.project(pts, view) x0, y0 = proj.min(axis=0) x1, y1 = proj.max(axis=0) return float(np.hypot(x1 - x0, y1 - y0)) def _segments(self, view: View | None = None) -> list[tuple[float, np.ndarray, np.ndarray, tuple[int, int, int], int]]: """Depth sorted screen segments.""" out: list[tuple[float, np.ndarray, np.ndarray, tuple[int, int, int], int]] = [] for s in self.strokes: if not s.points: continue pts = np.asarray(s.points, dtype=np.float64) proj, z = self.project(pts, view) if len(s.points) == 1: color = self.depth_color(pts[0, 2]) if s.gradient else s.color out.append((float(z[0]), proj[0], proj[0], self._shade(color, float(z[0]), s.gradient), s.thickness)) continue for i in range(len(s.points) - 1): zc = float((z[i] + z[i + 1]) * 0.5) if s.gradient: color = self.depth_color(float((pts[i, 2] + pts[i + 1, 2]) * 0.5)) else: color = s.color out.append((zc, proj[i], proj[i + 1], self._shade(color, zc, s.gradient), s.thickness)) out.sort(key=lambda item: item[0], reverse=True) return out def _shade(self, color: tuple[int, int, int], z: float, gradient: bool = False) -> tuple[int, int, int]: """Dim color by distance.""" t = float(np.clip((z - self.distance) / max(self.depth_span, 1e-6) + 0.5, 0.0, 1.0)) near, far = (GRADIENT_NEAR_GAIN, GRADIENT_FAR_GAIN) if gradient else (NEAR_GAIN, FAR_GAIN) gain = near + (far - near) * t return tuple(int(np.clip(c * gain, 0, 255)) for c in color) def _thickness_at(self, base: int, z: float) -> int: """Perspective scaled thickness.""" k = self.focal / max(z, 1e-3) return int(max(1, min(64, round(base * k)))) def render_over(self, frame: np.ndarray, view: View | None = None) -> np.ndarray: """Draw strokes onto a frame.""" out = frame.copy() for z, a, b, color, thick in self._segments(view): pa = (int(round(a[0])), int(round(a[1]))) pb = (int(round(b[0])), int(round(b[1]))) width = self._thickness_at(thick, z) if pa == pb: cv2.circle(out, pa, max(1, width // 2), color, -1, cv2.LINE_AA) else: cv2.line(out, pa, pb, color, width, cv2.LINE_AA) return out def composite_over(self, frame: np.ndarray, opacity: float = 1.0) -> np.ndarray: """Blend drawing onto frame.""" drawn = self.render_over(frame, self.view) if opacity >= 1.0: return drawn return cv2.addWeighted(drawn, opacity, frame, 1.0 - opacity, 0.0) def layers(self, view: View | None = None) -> tuple[np.ndarray, np.ndarray]: """Color layer and alpha mask.""" layer = np.zeros((self.height, self.width, 3), dtype=np.uint8) mask = np.zeros((self.height, self.width), dtype=np.uint8) for z, a, b, color, thick in self._segments(view): pa = (int(round(a[0])), int(round(a[1]))) pb = (int(round(b[0])), int(round(b[1]))) width = self._thickness_at(thick, z) if pa == pb: cv2.circle(layer, pa, max(1, width // 2), color, -1, cv2.LINE_AA) cv2.circle(mask, pa, max(1, width // 2), 255, -1, cv2.LINE_AA) else: cv2.line(layer, pa, pb, color, width, cv2.LINE_AA) cv2.line(mask, pa, pb, 255, width, cv2.LINE_AA) return layer, mask def to_bgra(self, view: View | None = None) -> np.ndarray: """Drawing with transparent background.""" layer, mask = self.layers(view) return np.dstack([layer, mask]) def save(self, tag: str = "3d") -> list[Path]: """Save current view as PNGs.""" self.output_dir.mkdir(parents=True, exist_ok=True) stamp = datetime.now().strftime("%Y%m%d_%H%M%S") suffix = f"_{tag}" if tag else "" saved: list[Path] = [] layer, mask = self.layers(self.view) transparent = self.output_dir / f"drawing_{stamp}{suffix}.png" cv2.imwrite(str(transparent), np.dstack([layer, mask])) saved.append(transparent) white = np.full((self.height, self.width, 3), 255, dtype=np.uint8) alpha = (mask.astype(np.float32) / 255.0)[:, :, None] flat = (white * (1.0 - alpha) + layer * alpha).astype(np.uint8) on_white = self.output_dir / f"drawing_{stamp}{suffix}_white.png" cv2.imwrite(str(on_white), flat) saved.append(on_white) return saved