Spaces:
Paused
Paused
File size: 2,551 Bytes
68b18cf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | """
Vision Sampler — frame capture, normalization, entropy, motion detection
Samples camera frames at adaptive rate, computes evidence metrics:
- Shannon entropy (information density)
- Perceptual hash (for frame dedup / novelty)
- Motion score (frame delta vs previous)
"""
import base64
import hashlib
import io
import time
import numpy as np
from PIL import Image
from .stream_state import FrameEvidence, sha256_bytes
def parse_data_url(data_url: str) -> bytes:
if "," in data_url and data_url.startswith("data:"):
data_url = data_url.split(",", 1)[1]
return base64.b64decode(data_url)
def normalize_jpeg(image_bytes: bytes, max_side: int = 960) -> tuple:
im = Image.open(io.BytesIO(image_bytes)).convert("RGB")
scale = min(1.0, max_side / max(im.width, im.height))
if scale < 1:
im = im.resize((int(im.width * scale), int(im.height * scale)))
out = io.BytesIO()
im.save(out, format="JPEG", quality=82, optimize=True)
return out.getvalue(), im
def image_entropy(im: Image.Image) -> float:
g = im.convert("L").resize((128, 128))
arr = np.asarray(g, dtype=np.uint8)
hist = np.bincount(arr.flatten(), minlength=256).astype(np.float64)
probs = hist / max(1, hist.sum())
probs = probs[probs > 0]
return float(-(probs * np.log2(probs)).sum())
def average_hash(im: Image.Image, size: int = 8) -> str:
g = im.convert("L").resize((size, size))
arr = np.asarray(g, dtype=np.float32)
mean = arr.mean()
bits = (arr > mean).astype(np.uint8).flatten()
value = 0
for bit in bits:
value = (value << 1) | int(bit)
return f"{value:016x}"
def phash_delta(a: str | None, b: str) -> float:
if not a:
return 1.0
x = int(a, 16)
y = int(b, 16)
return bin(x ^ y).count("1") / 64.0
def process_frame(data_url: str, previous_phash: str | None = None) -> FrameEvidence:
raw = parse_data_url(data_url)
jpeg, im = normalize_jpeg(raw)
ph = average_hash(im)
entropy = image_entropy(im)
motion = phash_delta(previous_phash, ph)
return FrameEvidence(
ts=time.time(),
sha256=sha256_bytes(jpeg),
phash=ph,
entropy=entropy,
motion_score=motion,
width=im.width,
height=im.height,
jpeg_b64=base64.b64encode(jpeg).decode("ascii"),
)
def should_increase_fps(motion_scores: list, threshold: float = 0.15) -> bool:
if len(motion_scores) < 3:
return False
recent = motion_scores[-3:]
return sum(recent) / len(recent) > threshold
|