File size: 6,681 Bytes
1d377f1 | 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 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 | """
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()
|