|
1 | 1 | """Canonical frame-selection algorithms shared by the CLI and TUI.""" |
2 | 2 |
|
3 | 3 | from collections import OrderedDict |
| 4 | +from math import log1p |
4 | 5 | from statistics import median |
5 | 6 | from typing import Any, Dict, List, Optional, Sequence, Set, Tuple |
6 | 7 |
|
|
10 | 11 | Frame = Dict[str, Any] |
11 | 12 | PositionedFrame = Tuple[int, Frame] |
12 | 13 |
|
| 14 | +OUTLIER_DEFAULT_WINDOW_SIZE = 15 |
| 15 | +OUTLIER_DEFAULT_SENSITIVITY = 60 |
| 16 | +OUTLIER_MIN_WINDOW_SIZE = 5 |
| 17 | +OUTLIER_MIN_RELATIVE_DROP = 0.05 |
| 18 | +OUTLIER_MAX_RELATIVE_DROP = 0.30 |
| 19 | +OUTLIER_MIN_ROBUST_THRESHOLD = 1.5 |
| 20 | + |
13 | 21 |
|
14 | 22 | def _source_key(frame: Frame) -> Optional[str]: |
15 | 23 | """Return a stable source identifier, if the frame belongs to a video.""" |
@@ -191,35 +199,61 @@ def _outlier_positions( |
191 | 199 | if sensitivity <= 0 or not group: |
192 | 200 | return set() |
193 | 201 |
|
194 | | - scores = [float(frame.get("sharpnessScore", 0) or 0) for _, frame in group] |
| 202 | + raw_scores = [ |
| 203 | + max(0.0, float(frame.get("sharpnessScore", 0) or 0)) |
| 204 | + for _, frame in group |
| 205 | + ] |
| 206 | + scores = [log1p(score) for score in raw_scores] |
195 | 207 |
|
196 | | - actual_window_size = max(1, window_size) |
| 208 | + actual_window_size = max(OUTLIER_MIN_WINDOW_SIZE, window_size) |
197 | 209 | if actual_window_size % 2 == 0: |
198 | 210 | actual_window_size += 1 |
199 | 211 | half_window = actual_window_size // 2 |
200 | 212 | maximum_threshold = threshold_divisor if threshold_divisor > 0 else 4.0 |
201 | 213 | sensitivity_ratio = min(100, sensitivity) / 100 |
202 | | - robust_threshold = 0.5 + ((maximum_threshold - 0.5) * (1 - sensitivity_ratio)) |
| 214 | + robust_threshold = OUTLIER_MIN_ROBUST_THRESHOLD + ( |
| 215 | + (maximum_threshold - OUTLIER_MIN_ROBUST_THRESHOLD) |
| 216 | + * (1 - sensitivity_ratio) |
| 217 | + ) |
| 218 | + minimum_relative_drop = OUTLIER_MAX_RELATIVE_DROP - ( |
| 219 | + (OUTLIER_MAX_RELATIVE_DROP - OUTLIER_MIN_RELATIVE_DROP) |
| 220 | + * sensitivity_ratio |
| 221 | + ) |
203 | 222 | outliers: Set[int] = set() |
204 | 223 |
|
205 | 224 | for position, current_score in enumerate(scores): |
206 | 225 | window_start = max(0, position - half_window) |
207 | 226 | window_end = min(len(group), position + half_window + 1) |
208 | 227 | neighbor_scores = scores[window_start:position] + scores[position + 1 : window_end] |
209 | | - if not neighbor_scores or len(neighbor_scores) < max(0, min_neighbors): |
| 228 | + required_neighbors = max(2, min_neighbors) |
| 229 | + if len(neighbor_scores) < required_neighbors: |
210 | 230 | continue |
211 | 231 | neighbor_median = median(neighbor_scores) |
| 232 | + raw_neighbor_scores = ( |
| 233 | + raw_scores[window_start:position] |
| 234 | + + raw_scores[position + 1 : window_end] |
| 235 | + ) |
| 236 | + raw_neighbor_median = median(raw_neighbor_scores) |
212 | 237 | absolute_deviations = [ |
213 | 238 | abs(score - neighbor_median) for score in neighbor_scores |
214 | 239 | ] |
215 | 240 | median_absolute_deviation = median(absolute_deviations) |
216 | 241 | robust_scale = median_absolute_deviation * 1.4826 |
217 | 242 | if robust_scale == 0: |
218 | | - robust_scale = max(abs(neighbor_median) * 0.05, 1e-9) |
| 243 | + robust_scale = max(abs(neighbor_median) * 0.01, 1e-9) |
219 | 244 |
|
220 | 245 | deficit = neighbor_median - current_score |
221 | 246 | robust_deficit = deficit / robust_scale |
222 | | - if deficit > 0 and robust_deficit > robust_threshold: |
| 247 | + relative_drop = ( |
| 248 | + (raw_neighbor_median - raw_scores[position]) / raw_neighbor_median |
| 249 | + if raw_neighbor_median > 0 |
| 250 | + else 0.0 |
| 251 | + ) |
| 252 | + if ( |
| 253 | + deficit > 0 |
| 254 | + and relative_drop >= minimum_relative_drop |
| 255 | + and robust_deficit > robust_threshold |
| 256 | + ): |
223 | 257 | outliers.add(position) |
224 | 258 | return outliers |
225 | 259 |
|
@@ -284,7 +318,7 @@ def select_outlier_removal_frames( |
284 | 318 | progress_bar.update(len(result)) |
285 | 319 | selected_count = sum(frame["selected"] for frame in result) |
286 | 320 | print( |
287 | | - f"Outlier removal: Marked {len(result) - selected_count} outliers. " |
| 321 | + f"Outlier detection: Marked {len(result) - selected_count} outliers. " |
288 | 322 | f"Keeping {selected_count} frames." |
289 | 323 | ) |
290 | 324 | return result |
|
0 commit comments