"""Effect layers. Each layer takes the current BGR frame plus a shared Context (motion analysis of the raw camera frame) and returns a new BGR frame. Layers run top to bottom in the order returned by default_layers(), like a chain of TOPs. """ import cv2 import numpy as np rng = np.random.default_rng() class Param: def __init__(self, value, lo=0.0, hi=1.0, integer=False): self.value, self.lo, self.hi, self.integer = value, lo, hi, integer @property def norm(self): return (self.value - self.lo) / (self.hi - self.lo) @norm.setter def norm(self, n): v = self.lo + min(max(n, 0.0), 1.0) * (self.hi - self.lo) self.value = round(v) if self.integer else v class Layer: name = "layer" enabled = True def __init__(self, **params): self.params = params def __getitem__(self, key): return self.params[key].value def apply(self, frame, ctx): return frame class Context: """Per-frame motion analysis, computed at 1/4 resolution and shared by all layers.""" SCALE = 4 def __init__(self, size): self.small = (size[0] // self.SCALE, size[1] // self.SCALE) self.prev = None self.t = 0.0 self.dt = 1 / 30 def update(self, frame, dt): self.dt = dt self.t += dt gray = cv2.cvtColor(cv2.resize(frame, self.small, interpolation=cv2.INTER_AREA), cv2.COLOR_BGR2GRAY) if self.prev is None: self.prev = gray self.motion = cv2.absdiff(gray, self.prev) # where things changed self.energy = float(self.motion.mean()) / 255 # how much changed overall self._pair = (self.prev, gray) self._flow = None self.prev = gray @property def flow(self): """Dense optical flow (small-res pixels/frame). Lazy: only paid for if a layer asks.""" if self._flow is None: self._flow = cv2.calcOpticalFlowFarneback(*self._pair, None, 0.5, 2, 15, 2, 5, 1.1, 0) return self._flow def hue_bgr(hue): """Hue in 0..1 (wraps) -> saturated BGR colour with components in 0..1.""" hsv = np.uint8([[[int(hue % 1.0 * 180), 255, 255]]]) return cv2.cvtColor(hsv, cv2.COLOR_HSV2BGR)[0, 0] / 255.0 def tint(gray, color): return cv2.merge([cv2.convertScaleAbs(gray, alpha=float(c)) for c in color]) class Colorize(Layer): name = "Colorize" enabled = False PALETTES = [cv2.COLORMAP_INFERNO, cv2.COLORMAP_OCEAN, cv2.COLORMAP_TWILIGHT_SHIFTED, cv2.COLORMAP_TURBO] def __init__(self): super().__init__(mix=Param(0.8), palette=Param(0, 0, len(self.PALETTES) - 1, integer=True)) def apply(self, frame, ctx): gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) mapped = cv2.applyColorMap(gray, self.PALETTES[self["palette"]]) return cv2.addWeighted(frame, 1 - self["mix"], mapped, self["mix"], 0) class Edges(Layer): name = "Neon Edges" def __init__(self): super().__init__(detail=Param(0.6), glow=Param(0.35), source=Param(0.35), hue=Param(0.55), cycle=Param(0.3)) def apply(self, frame, ctx): gray = cv2.GaussianBlur(cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY), (5, 5), 0) lo = 20 + 180 * (1 - self["detail"]) edges = cv2.Canny(gray, lo, lo * 2) sigma = self["glow"] * 12 if sigma > 0.5: edges = cv2.addWeighted(edges, 1, cv2.GaussianBlur(edges, (0, 0), sigma), 2.5, 0) color = hue_bgr(self["hue"] + ctx.t * self["cycle"] * 0.2) return cv2.add(cv2.convertScaleAbs(frame, alpha=self["source"]), tint(edges, color)) class Particles(Layer): """Particles spawn where there is motion and get carried along by the optical flow.""" name = "Flow Particles" MAX = 6000 def __init__(self): super().__init__(count=Param(3000, 200, self.MAX, integer=True), speed=Param(1.5, 0, 4), trail=Param(0.88, 0, 0.98), hue=Param(0.08)) self.buf = None def apply(self, frame, ctx): h, w = frame.shape[:2] if self.buf is None: self.buf = np.zeros((h, w), np.uint8) self.pos = (rng.random((self.MAX, 2)) * (w, h)).astype(np.float32) self.life = rng.integers(20, 90, self.MAX) n = self["count"] pos, life = self.pos[:n], self.life[:n] flow = ctx.flow fh, fw = flow.shape[:2] scale = w / fw ix = np.clip((pos[:, 0] / scale).astype(int), 0, fw - 1) iy = np.clip((pos[:, 1] / scale).astype(int), 0, fh - 1) new = pos + flow[iy, ix] * scale * self["speed"] + rng.normal(0, 0.5, (n, 2)).astype(np.float32) life -= 1 dead = (life <= 0) | (new[:, 0] < 0) | (new[:, 0] >= w) | (new[:, 1] < 0) | (new[:, 1] >= h) self.buf = cv2.convertScaleAbs(self.buf, alpha=self["trail"]) segments = np.stack([pos, new], axis=1).astype(np.int32)[~dead] cv2.polylines(self.buf, list(segments), False, 255, 1, cv2.LINE_AA) k = int(dead.sum()) if k: hot = np.flatnonzero(ctx.motion > 24) if hot.size > 50: ys, xs = np.divmod(rng.choice(hot, k), fw) new[dead] = (np.stack([xs, ys], axis=1) + rng.random((k, 2))) * scale else: new[dead] = rng.random((k, 2)) * (w, h) life[dead] = rng.integers(20, 90, k) self.pos[:n] = new glow = cv2.add(self.buf, cv2.GaussianBlur(self.buf, (0, 0), 3)) return cv2.add(frame, tint(glow, hue_bgr(self["hue"]))) class Feedback(Layer): """Classic feedback loop: the previous output is transformed, faded and blended back in.""" name = "Feedback" def __init__(self): super().__init__(decay=Param(0.92, 0, 0.995), zoom=Param(0.012, -0.05, 0.05), rotate=Param(0.3, -3, 3), **{"flow push": Param(0.0, 0, 4)}) self.prev = None def apply(self, frame, ctx): h, w = frame.shape[:2] if self.prev is None: self.prev = frame self.grid = np.stack(np.meshgrid(np.arange(w, dtype=np.float32), np.arange(h, dtype=np.float32)), axis=-1) m = cv2.getRotationMatrix2D((w / 2, h / 2), self["rotate"], 1 + self["zoom"]) fb = cv2.warpAffine(self.prev, m, (w, h), borderMode=cv2.BORDER_REFLECT) if self["flow push"] > 0: # smear the feedback along the direction things are moving flow = cv2.resize(ctx.flow, (w, h)) * (self["flow push"] * Context.SCALE) fb = cv2.remap(fb, self.grid - flow, None, cv2.INTER_LINEAR, borderMode=cv2.BORDER_REFLECT) out = cv2.max(frame, cv2.convertScaleAbs(fb, alpha=self["decay"])) self.prev = out return out class RGBShift(Layer): name = "RGB Shift" def __init__(self): super().__init__(amount=Param(4, 0, 40), **{"motion react": Param(0.5)}) def apply(self, frame, ctx): d = int(self["amount"] + self["motion react"] * ctx.energy * 400) if d == 0: return frame b, g, r = cv2.split(frame) return cv2.merge([np.roll(b, d, axis=1), g, np.roll(r, -d, axis=1)]) class Kaleido(Layer): name = "Kaleidoscope" enabled = False def __init__(self): super().__init__(segments=Param(6, 2, 16, integer=True), spin=Param(0.2, -2, 2)) self.radius = None self.angle = 0.0 def apply(self, frame, ctx): h, w = frame.shape[:2] if self.radius is None: ys, xs = np.indices((h, w), dtype=np.float32) self.radius, self.theta = cv2.cartToPolar(xs - w / 2, ys - h / 2) self.angle += self["spin"] * ctx.dt wedge = 2 * np.pi / self["segments"] # fold every angle into one mirrored wedge, then sample the source from that wedge a = np.abs(np.mod(self.theta + np.float32(self.angle), np.float32(wedge)) - np.float32(wedge / 2)) x, y = cv2.polarToCart(self.radius, a - np.float32(np.pi / 2)) return cv2.remap(frame, x + w / 2, y + h / 2, cv2.INTER_LINEAR, borderMode=cv2.BORDER_REFLECT) def default_layers(): return [Colorize(), Edges(), Particles(), Feedback(), RGBShift(), Kaleido()]