+464

observe/gnome/screencast_diff.py

reviewed

··· 1 1 + #!/usr/bin/env python3 2 2 + """ 3 3 + screencast_diff.py — find the most visually different frames in a screencast 4 4 + 5 5 + Efficiently scans video and computes the minimum distance from each frame to 6 6 + all earlier frames using intensity, gradient, and histogram-based scoring. 7 7 + Frames with high min-distance are novel (never seen before), while repeated 8 8 + frames (e.g., A→B→A toggles) score low. Identifies the top 10 most divergent 9 9 + frames, then re-extracts only those frames as WebP images. 10 10 + 11 11 + Usage: 12 12 + gnome-screencast-diff screencast.webm 13 13 + gnome-screencast-diff screencast.webm --interval 0.5 # sample every 0.5s 14 14 + """ 15 15 + 16 16 + import argparse 17 17 + import io 18 18 + import sys 19 19 + import time 20 20 + from http.server import BaseHTTPRequestHandler, HTTPServer 21 21 + from pathlib import Path 22 22 + 23 23 + import av 24 24 + import numpy as np 25 25 + from PIL import Image 26 26 + from scipy.spatial.distance import jensenshannon 27 27 + 28 28 + 29 29 + # Frame comparison helper functions 30 30 + 31 31 + 32 32 + def to_luma(img: Image.Image) -> np.ndarray: 33 33 + """Convert PIL image to luma (grayscale) as float32 normalized to [0, 1].""" 34 34 + if img.mode != "L": 35 35 + img = img.convert("L") 36 36 + return np.array(img, dtype=np.float32) / 255.0 37 37 + 38 38 + 39 39 + def avg_pool(Y: np.ndarray, pool_h: int, pool_w: int) -> np.ndarray: 40 40 + """Average pooling via reshape and mean.""" 41 41 + h, w = Y.shape 42 42 + new_h = h // pool_h 43 43 + new_w = w // pool_w 44 44 + 45 45 + # Trim to make dimensions divisible 46 46 + Y_trimmed = Y[: new_h * pool_h, : new_w * pool_w] 47 47 + 48 48 + # Reshape and compute mean 49 49 + reshaped = Y_trimmed.reshape(new_h, pool_h, new_w, pool_w) 50 50 + return reshaped.mean(axis=(1, 3)) 51 51 + 52 52 + 53 53 + def gradient_mag(Y: np.ndarray) -> np.ndarray: 54 54 + """Compute gradient magnitude using np.diff in x and y directions.""" 55 55 + dx = np.diff(Y, axis=1, prepend=Y[:, :1]) 56 56 + dy = np.diff(Y, axis=0, prepend=Y[:1, :]) 57 57 + return np.sqrt(dx**2 + dy**2) 58 58 + 59 59 + 60 60 + def hist32(Y: np.ndarray) -> np.ndarray: 61 61 + """Compute 32-bin normalized histogram of image.""" 62 62 + hist, _ = np.histogram(Y, bins=32, range=(0, 1)) 63 63 + hist = hist.astype(np.float32) 64 64 + # Normalize to probability distribution 65 65 + hist_sum = hist.sum() 66 66 + if hist_sum > 0: 67 67 + hist = hist / hist_sum 68 68 + return hist 69 69 + 70 70 + 71 71 + def js_div(p: np.ndarray, q: np.ndarray) -> float: 72 72 + """Jensen-Shannon divergence between two probability distributions.""" 73 73 + # Add small epsilon to avoid log(0) 74 74 + eps = 1e-10 75 75 + p = np.clip(p, eps, 1.0) 76 76 + q = np.clip(q, eps, 1.0) 77 77 + return float(jensenshannon(p, q) ** 2) 78 78 + 79 79 + 80 80 + def compute_frame_score( 81 81 + Y_prev: np.ndarray, 82 82 + Y_curr: np.ndarray, 83 83 + G_prev: np.ndarray, 84 84 + G_curr: np.ndarray, 85 85 + hist_prev: np.ndarray, 86 86 + hist_curr: np.ndarray, 87 87 + ) -> float: 88 88 + """ 89 89 + Compute comprehensive frame difference score. 90 90 + 91 91 + S = 0.5*S_int + 0.3*S_grad + 0.2*JSD 92 92 + where: 93 93 + S_int = mean((Y_curr - Y_prev)**2) / (var(Y_prev) + eps) 94 94 + S_grad = mean((G_curr - G_prev)**2) / (mean(G_prev**2) + eps) 95 95 + JSD = jensen_shannon_divergence(hist_prev, hist_curr) 96 96 + """ 97 97 + eps = 1e-10 98 98 + 99 99 + # Intensity difference score 100 100 + var_prev = np.var(Y_prev) 101 101 + S_int = np.mean((Y_curr - Y_prev) ** 2) / (var_prev + eps) 102 102 + 103 103 + # Gradient difference score 104 104 + mean_G_prev_sq = np.mean(G_prev**2) 105 105 + S_grad = np.mean((G_curr - G_prev) ** 2) / (mean_G_prev_sq + eps) 106 106 + 107 107 + # Histogram JSD 108 108 + JSD = js_div(hist_prev, hist_curr) 109 109 + 110 110 + # Weighted combination 111 111 + S = 0.5 * S_int + 0.3 * S_grad + 0.2 * JSD 112 112 + 113 113 + return float(S) 114 114 + 115 115 + 116 116 + class ScreencastDiffer: 117 117 + def __init__(self, video_path: str, sample_interval: float = 1.0): 118 118 + self.video_path = Path(video_path) 119 119 + if not self.video_path.exists(): 120 120 + raise FileNotFoundError(f"Video file not found: {video_path}") 121 121 + 122 122 + self.sample_interval = sample_interval 123 123 + self.frame_scores = [] # List of (timestamp, score) - computed during scan 124 124 + self.divergence_scores = [] # List of (timestamp, score) - sorted by score 125 125 + self.top_frames = {} # Dict of {idx: (timestamp, score, webp_bytes)} for top 10 126 126 + 127 127 + # Performance timing 128 128 + self.timings = { 129 129 + "video_scan": 0.0, 130 130 + "frame_to_image": 0.0, 131 131 + "luma_compute": 0.0, 132 132 + "pooling": 0.0, 133 133 + "gradient": 0.0, 134 134 + "histogram": 0.0, 135 135 + "score_compute": 0.0, 136 136 + "top_frames_extract": 0.0, 137 137 + "webp_encode": 0.0, 138 138 + } 139 139 + 140 140 + print( 141 141 + f"Scanning {video_path} with min-distance-to-history comparison...", 142 142 + file=sys.stderr, 143 143 + ) 144 144 + self._process_video() 145 145 + print("Sorting divergence scores...", file=sys.stderr) 146 146 + self._compute_divergence() 147 147 + print("Extracting top 10 frames as WebP...", file=sys.stderr) 148 148 + self._extract_top_frames() 149 149 + print( 150 150 + f"Found {len(self.divergence_scores)} scored frames, ready to serve top 10", 151 151 + file=sys.stderr, 152 152 + ) 153 153 + self._print_timings() 154 154 + 155 155 + def _process_video(self): 156 156 + """Scan video at intervals and compute min distance to any earlier frame.""" 157 157 + try: 158 158 + t_scan_start = time.perf_counter() 159 159 + with av.open(str(self.video_path)) as container: 160 160 + stream = container.streams.video[0] 161 161 + duration = ( 162 162 + float(stream.duration * stream.time_base) 163 163 + if stream.duration 164 164 + else None 165 165 + ) 166 166 + 167 167 + if duration: 168 168 + print(f"Video duration: {duration:.2f}s", file=sys.stderr) 169 169 + 170 170 + last_sampled = -self.sample_interval 171 171 + frame_count = 0 172 172 + 173 173 + # Store all previous frame features for comparison 174 174 + previous_frames = [] # List of (Yd, G, hist) tuples 175 175 + 176 176 + for frame in container.decode(video=0): 177 177 + if frame.pts is None: 178 178 + continue 179 179 + 180 180 + timestamp = frame.time if frame.time is not None else 0.0 181 181 + 182 182 + # Sample at intervals 183 183 + if timestamp - last_sampled >= self.sample_interval: 184 184 + # Convert frame to image 185 185 + t_img_start = time.perf_counter() 186 186 + img = frame.to_image() 187 187 + self.timings["frame_to_image"] += ( 188 188 + time.perf_counter() - t_img_start 189 189 + ) 190 190 + 191 191 + # Convert to luma 192 192 + t_luma_start = time.perf_counter() 193 193 + Y = to_luma(img) 194 194 + self.timings["luma_compute"] += ( 195 195 + time.perf_counter() - t_luma_start 196 196 + ) 197 197 + 198 198 + # Average pool to reduce resolution 199 199 + t_pool_start = time.perf_counter() 200 200 + Yd = avg_pool(Y, 128, 128) 201 201 + self.timings["pooling"] += time.perf_counter() - t_pool_start 202 202 + 203 203 + # Compute gradient magnitude 204 204 + t_grad_start = time.perf_counter() 205 205 + G = gradient_mag(Yd) 206 206 + self.timings["gradient"] += time.perf_counter() - t_grad_start 207 207 + 208 208 + # Compute histogram 209 209 + t_hist_start = time.perf_counter() 210 210 + hist = hist32(Yd) 211 211 + self.timings["histogram"] += time.perf_counter() - t_hist_start 212 212 + 213 213 + # Compute min score against all previous frames 214 214 + if previous_frames: 215 215 + t_score_start = time.perf_counter() 216 216 + min_score = float("inf") 217 217 + for Y_prev, G_prev, hist_prev in previous_frames: 218 218 + score = compute_frame_score( 219 219 + Y_prev, Yd, G_prev, G, hist_prev, hist 220 220 + ) 221 221 + min_score = min(min_score, score) 222 222 + self.timings["score_compute"] += ( 223 223 + time.perf_counter() - t_score_start 224 224 + ) 225 225 + self.frame_scores.append((timestamp, min_score)) 226 226 + 227 227 + # Store current frame features for future comparisons 228 228 + previous_frames.append((Yd, G, hist)) 229 229 + 230 230 + last_sampled = timestamp 231 231 + frame_count += 1 232 232 + 233 233 + if frame_count % 10 == 0: 234 234 + print( 235 235 + f" Scanned {frame_count} frames at {timestamp:.1f}s", 236 236 + file=sys.stderr, 237 237 + ) 238 238 + 239 239 + self.timings["video_scan"] = time.perf_counter() - t_scan_start 240 240 + 241 241 + except Exception as e: 242 242 + print(f"ERROR: Failed to process video: {e}", file=sys.stderr) 243 243 + import traceback 244 244 + 245 245 + traceback.print_exc(file=sys.stderr) 246 246 + raise 247 247 + 248 248 + def _compute_divergence(self): 249 249 + """Sort frame scores by divergence (highest first).""" 250 250 + # Scores are already computed during scan, just need to sort 251 251 + self.divergence_scores = sorted( 252 252 + self.frame_scores, key=lambda x: x[1], reverse=True 253 253 + ) 254 254 + 255 255 + def _extract_top_frames(self, n: int = 10): 256 256 + """Extract and encode top N most divergent frames as WebP.""" 257 257 + if not self.divergence_scores: 258 258 + return 259 259 + 260 260 + t_extract_start = time.perf_counter() 261 261 + 262 262 + # Create mapping from timestamp to rank (1-indexed) 263 263 + timestamp_to_idx = { 264 264 + ts: idx for idx, (ts, _) in enumerate(self.divergence_scores[:n], 1) 265 265 + } 266 266 + 267 267 + try: 268 268 + with av.open(str(self.video_path)) as container: 269 269 + last_sampled = -self.sample_interval 270 270 + 271 271 + for frame in container.decode(video=0): 272 272 + if frame.pts is None: 273 273 + continue 274 274 + 275 275 + timestamp = frame.time if frame.time is not None else 0.0 276 276 + 277 277 + # Sample at intervals and check if this is a top frame 278 278 + if timestamp - last_sampled >= self.sample_interval: 279 279 + if timestamp in timestamp_to_idx: 280 280 + # Convert frame to image 281 281 + img = frame.to_image() 282 282 + 283 283 + # Encode as WebP with quality setting 284 284 + t_webp_start = time.perf_counter() 285 285 + buf = io.BytesIO() 286 286 + img.save(buf, format="WEBP", quality=85) 287 287 + webp_bytes = buf.getvalue() 288 288 + self.timings["webp_encode"] += ( 289 289 + time.perf_counter() - t_webp_start 290 290 + ) 291 291 + 292 292 + # Find the divergence score for this timestamp 293 293 + score = next( 294 294 + s for ts, s in self.divergence_scores if ts == timestamp 295 295 + ) 296 296 + 297 297 + # Store with index based on divergence rank 298 298 + idx = timestamp_to_idx[timestamp] 299 299 + self.top_frames[idx] = (timestamp, score, webp_bytes) 300 300 + 301 301 + # Exit early if we've found all top frames 302 302 + if len(self.top_frames) >= n: 303 303 + break 304 304 + 305 305 + last_sampled = timestamp 306 306 + 307 307 + except Exception as e: 308 308 + print(f"ERROR: Failed to extract top frames: {e}", file=sys.stderr) 309 309 + import traceback 310 310 + 311 311 + traceback.print_exc(file=sys.stderr) 312 312 + raise 313 313 + 314 314 + self.timings["top_frames_extract"] = time.perf_counter() - t_extract_start 315 315 + 316 316 + def get_top_divergent(self, n: int = 10): 317 317 + """Get the top N most divergent frames.""" 318 318 + return self.divergence_scores[:n] 319 319 + 320 320 + def _print_timings(self): 321 321 + """Print performance timing breakdown.""" 322 322 + total = sum(self.timings.values()) 323 323 + 324 324 + print(f"\n{'='*60}", file=sys.stderr) 325 325 + print(f"Performance Breakdown (total: {total:.2f}s)", file=sys.stderr) 326 326 + print(f"{'='*60}", file=sys.stderr) 327 327 + 328 328 + # Sort by time descending 329 329 + sorted_timings = sorted(self.timings.items(), key=lambda x: x[1], reverse=True) 330 330 + 331 331 + for name, duration in sorted_timings: 332 332 + pct = (duration / total * 100) if total > 0 else 0 333 333 + bar_width = int(pct / 2) # Scale to 50 chars max 334 334 + bar = "█" * bar_width 335 335 + print(f"{name:20s} {duration:7.2f}s {pct:5.1f}% {bar}", file=sys.stderr) 336 336 + 337 337 + print(f"{'='*60}", file=sys.stderr) 338 338 + 339 339 + # Calculate derived metrics 340 340 + if len(self.frame_scores) > 0: 341 341 + per_frame = total / ( 342 342 + len(self.frame_scores) + 1 343 343 + ) # +1 for first frame with no score 344 344 + print(f"Frames scanned: {len(self.frame_scores) + 1}", file=sys.stderr) 345 345 + print(f"Frames scored: {len(self.frame_scores)}", file=sys.stderr) 346 346 + print(f"Time per frame: {per_frame:.3f}s", file=sys.stderr) 347 347 + 348 348 + if len(self.frame_scores) > 0: 349 349 + per_score = self.timings["score_compute"] / len(self.frame_scores) 350 350 + print(f"Time per score: {per_score*1000:.3f}ms", file=sys.stderr) 351 351 + 352 352 + print(f"{'='*60}\n", file=sys.stderr) 353 353 + 354 354 + 355 355 + def make_handler(differ: ScreencastDiffer): 356 356 + """Create request handler with access to differ instance.""" 357 357 + 358 358 + class RequestHandler(BaseHTTPRequestHandler): 359 359 + def log_message(self, format, *args): 360 360 + """Suppress default request logging.""" 361 361 + sys.stderr.write(f"{self.address_string()} - {format % args}\n") 362 362 + 363 363 + def do_GET(self): 364 364 + if self.path == "/": 365 365 + # Serve HTML page with top 10 frames 366 366 + self.send_response(200) 367 367 + self.send_header("Content-Type", "text/html; charset=utf-8") 368 368 + self.end_headers() 369 369 + 370 370 + html = ["<!DOCTYPE html>"] 371 371 + html.append("<html><head>") 372 372 + html.append("<title>Screencast Divergence</title>") 373 373 + html.append("<style>") 374 374 + html.append( 375 375 + "body { font-family: monospace; margin: 20px; background: #1e1e1e; color: #ccc; }" 376 376 + ) 377 377 + html.append("h1 { color: #fff; }") 378 378 + html.append( 379 379 + ".frame { margin: 20px 0; padding: 10px; border: 1px solid #444; background: #2e2e2e; }" 380 380 + ) 381 381 + html.append( 382 382 + ".frame img { max-width: 800px; display: block; margin: 10px 0; }" 383 383 + ) 384 384 + html.append(".info { color: #8cf; }") 385 385 + html.append("</style>") 386 386 + html.append("</head><body>") 387 387 + html.append("<h1>Top 10 Most Divergent Frames</h1>") 388 388 + html.append(f"<p>Video: {differ.video_path.name}</p>") 389 389 + html.append( 390 390 + "<p>Scoring method: Min distance to any earlier frame (intensity + gradient + histogram JSD)</p>" 391 391 + ) 392 392 + html.append( 393 393 + f"<p>Total frames analyzed: {len(differ.frame_scores)} (sampled every {differ.sample_interval}s)</p>" 394 394 + ) 395 395 + 396 396 + for idx, (timestamp, score) in enumerate( 397 397 + differ.get_top_divergent(10), 1 398 398 + ): 399 399 + html.append('<div class="frame">') 400 400 + html.append( 401 401 + f'<div class="info">#{idx} - Timestamp: {timestamp:.2f}s - Divergence Score: {score}</div>' 402 402 + ) 403 403 + html.append( 404 404 + f'<img src="/frame/{idx}" alt="Frame at {timestamp:.2f}s">' 405 405 + ) 406 406 + html.append("</div>") 407 407 + 408 408 + html.append("</body></html>") 409 409 + 410 410 + output = "\n".join(html).encode("utf-8") 411 411 + self.wfile.write(output) 412 412 + 413 413 + elif self.path.startswith("/frame/"): 414 414 + # Serve individual frame image 415 415 + try: 416 416 + idx = int(self.path.split("/")[-1]) 417 417 + if idx in differ.top_frames: 418 418 + _, _, webp_bytes = differ.top_frames[idx] 419 419 + self.send_response(200) 420 420 + self.send_header("Content-Type", "image/webp") 421 421 + self.send_header("Content-Length", str(len(webp_bytes))) 422 422 + self.end_headers() 423 423 + self.wfile.write(webp_bytes) 424 424 + else: 425 425 + self.send_error(404, "Frame not found") 426 426 + except (ValueError, IndexError): 427 427 + self.send_error(400, "Invalid frame index") 428 428 + else: 429 429 + self.send_error(404, "Not found") 430 430 + 431 431 + return RequestHandler 432 432 + 433 433 + 434 434 + def main(): 435 435 + parser = argparse.ArgumentParser( 436 436 + description="Find and display the most visually different frames in a screencast" 437 437 + ) 438 438 + parser.add_argument("video", help="Path to screencast webm file") 439 439 + parser.add_argument( 440 440 + "--interval", 441 441 + type=float, 442 442 + default=1.0, 443 443 + help="Sample interval in seconds (default: 1.0)", 444 444 + ) 445 445 + parser.add_argument( 446 446 + "--port", type=int, default=9999, help="Server port (default: 9999)" 447 447 + ) 448 448 + args = parser.parse_args() 449 449 + 450 450 + differ = ScreencastDiffer(args.video, sample_interval=args.interval) 451 451 + 452 452 + print(f"\nServer running at http://0.0.0.0:{args.port}/") 453 453 + print("Press Ctrl+C to stop") 454 454 + 455 455 + server = HTTPServer(("0.0.0.0", args.port), make_handler(differ)) 456 456 + try: 457 457 + server.serve_forever() 458 458 + except KeyboardInterrupt: 459 459 + print("\nShutting down...") 460 460 + server.shutdown() 461 461 + 462 462 + 463 463 + if __name__ == "__main__": 464 464 + main()

+2

pyproject.toml

reviewed

··· 44 44 "numpy", 45 45 "dbus-next", 46 46 "av", 47 47 + "imagehash", 47 48 48 49 "sqlite-utils", 49 50 "openai>=1.2.0", ··· 107 108 think-messages = "think.messages:main" 108 109 gnome-screencast = "observe.gnome.screencast:main" 109 110 gnome-screencast-viewer = "observe.gnome.screencast_viewer:main" 111 111 + gnome-screencast-diff = "observe.gnome.screencast_diff:main" 110 112 111 113 [project.urls] 112 114 Homepage = "https://github.com/yourusername/sunstone"