tools/detect_ref.py

"""
Reference implementation of the flash detector in Python/OpenCV, used to tune
the algorithm against the real video before porting the tuned logic to
detector.js. Runs the WHOLE video and prints candidate flashes.

Algorithm (the tuned design):
  - Downscale each frame to analysis width for speed + noise averaging.
  - Rolling background = mean of previous N frames.
  - diff = luma - background.
  - A candidate pixel must be a strong, LOCAL peak: its diff exceeds a high
    threshold AND it stands out from its local neighbourhood (top-hat), which
    rejects broad brightness shifts from jitter/glow.
  - Connected components -> blobs, size-filtered.
  - A blob is a flash only if it is a genuine transient: strong on its frame,
    weak (gone) on the next frame, and reasonably compact/round.
"""
import cv2, numpy as np, sys

PATH = r"E:\test\Kiro_flash_detection\Video\0921 (1)_flash.mp4"
# --- constants mirrored EXACTLY from detector.js (sensitivity 3) ---
ANALYSIS_W = 480
BG_HISTORY = 15
SENS_THRESH = 38          # SENSITIVITY_THRESHOLDS[3]
CONFIDENCE_PEAKDIFF = 135 # CONFIDENCE_PEAKDIFF[3]
MIN_SIZE = 2
MAX_SIZE = 1500
MIN_FILL = 0.35           # MIN_FILL
MAX_AREA_A = 400          # MAX_AREA_ANALYSIS
MAX_DUR_FRAMES = 24       # maxDurationMs 800 / (1000/30)

def main():
    cap = cv2.VideoCapture(PATH)
    fps = cap.get(cv2.CAP_PROP_FPS) or 30.0
    W = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    H = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    scale = W / ANALYSIS_W if W > ANALYSIS_W else 1.0
    aw = max(1, round(W/scale)); ah = max(1, round(H/scale))

    hist = []          # rolling luma frames (float32)
    bgsum = np.zeros((ah, aw), np.float32)
    events = []        # active events
    finished = []
    eid = 0
    fidx = -1

    while True:
        ok, frame = cap.read()
        if not ok: break
        fidx += 1
        small = cv2.resize(frame, (aw, ah), interpolation=cv2.INTER_AREA)
        luma = cv2.cvtColor(small, cv2.COLOR_BGR2GRAY).astype(np.float32)

        if len(hist) < 4:
            hist.append(luma); bgsum += luma
            continue

        bg = bgsum / len(hist)
        diff = luma - bg

        # Local top-hat: subtract a blurred version of the diff so only
        # spatially compact peaks survive (kills broad jitter glow).
        blur = cv2.GaussianBlur(diff, (0,0), sigmaX=9)
        tophat = diff - blur

        # noise floor from robust spread of tophat
        med = np.median(np.abs(tophat))
        thr = max(SENS_THRESH*0.5, med*6.0)

        hot = ((diff >= SENS_THRESH) & (tophat >= thr)).astype(np.uint8)

        # connected components
        blobs = []
        ncc, labels, stats, cents = cv2.connectedComponentsWithStats(hot, 8)
        for lab in range(1, ncc):
            area = stats[lab, cv2.CC_STAT_AREA]
            if area < MIN_SIZE or area > MAX_SIZE: continue
            x = stats[lab, cv2.CC_STAT_LEFT]; y = stats[lab, cv2.CC_STAT_TOP]
            bw = stats[lab, cv2.CC_STAT_WIDTH]; bh = stats[lab, cv2.CC_STAT_HEIGHT]
            cx, cy = cents[lab]
            mask = labels == lab
            peakdiff = float(diff[mask].max())
            peaklum = float(luma[mask].max())
            # compactness: area vs bbox area
            fill = area / max(1, bw*bh)
            blobs.append(dict(cx=cx, cy=cy, area=int(area), bw=int(bw), bh=int(bh),
                              peakdiff=peakdiff, peaklum=peaklum, fill=fill,
                              bbox=(x,y,x+bw-1,y+bh-1)))

        # update bg AFTER detection
        hist.append(luma); bgsum += luma
        if len(hist) > BG_HISTORY:
            old = hist.pop(0); bgsum -= old

        # link events (nearest within radius)
        link_r = max(6, aw*0.025)
        used = [False]*len(blobs)
        for ev in events:
            best=-1; bd=1e9
            for i,b in enumerate(blobs):
                if used[i]: continue
                d = ((b['cx']-ev['lcx'])**2+(b['cy']-ev['lcy'])**2)**0.5
                if d<bd and d<=link_r: bd=d; best=i
            if best>=0:
                b=blobs[best]; used[best]=True
                ev['frames']+=1; ev['end']=fidx; ev['lcx']=b['cx']; ev['lcy']=b['cy']
                ev['series'].append(b['peakdiff'])
                if b['peakdiff']>ev['peakdiff']:
                    ev.update(peakdiff=b['peakdiff'], peaklum=b['peaklum'],
                              pcx=b['cx'], pcy=b['cy'], parea=b['area'],
                              pbbox=b['bbox'], pframe=fidx, fill=b['fill'])
                ev['miss']=0
            else:
                ev['miss']=ev.get('miss',0)+1
        still=[]
        for ev in events:
            if ev['miss']>=1: finished.append(ev)
            else: still.append(ev)
        events=still
        for i,b in enumerate(blobs):
            if used[i]: continue
            eid+=1
            events.append(dict(id=eid, start=fidx, end=fidx, pframe=fidx, frames=1,
                lcx=b['cx'], lcy=b['cy'], pcx=b['cx'], pcy=b['cy'],
                peakdiff=b['peakdiff'], peaklum=b['peaklum'], parea=b['area'],
                pbbox=b['bbox'], fill=b['fill'], series=[b['peakdiff']], miss=0))

    finished += events

    # finalize: duration + transient + amplitude + shape filters
    MIN_FILL = 0.35      # roundness/compactness
    MIN_PEAKDIFF = 60    # amplitude floor
    MAX_AREA_A = 400     # analysis-space area cap for a point-like flash
    results=[]
    for ev in finished:
        dur_frames = ev['frames']
        if dur_frames > MAX_DUR_FRAMES: continue
        if ev['peakdiff'] < MIN_PEAKDIFF: continue
        if ev['fill'] < MIN_FILL: continue
        if ev['parea'] > MAX_AREA_A: continue
        s = ev['series']
        # transient: peak then decay
        if len(s) > 2:
            pv=max(s); last=s[-1]
            if not (last <= pv*0.6): continue
        results.append(ev)

    # sort by peak amplitude desc
    results.sort(key=lambda e:-e['peakdiff'])
    print(f"fps={fps} analysis={aw}x{ah} scale={scale:.2f}")

    # threshold sweep to find a robust separation
    for pd in [100, 120, 140, 150, 160, 180, 200]:
        cnt = sum(1 for e in results if e['peakdiff'] >= pd)
        print(f"  peakdiff>={pd}: {cnt} candidates")
    print(f"TOTAL candidate flashes: {len(results)}")
    for e in results[:25]:
        print(f"  #{e['id']} frame {e['start']}..{e['end']} peakF={e['pframe']} "
              f"t={e['pframe']/fps:.3f}s frames={e['frames']} "
              f"peakdiff={e['peakdiff']:.0f} peaklum={e['peaklum']:.0f} "
              f"area={e['parea']} fill={e['fill']:.2f} "
              f"pos=({e['pcx']*scale:.0f},{e['pcy']*scale:.0f})")
    cap.release()

if __name__=='__main__':
    main()