""" Portable CPU Inference — Read vs Spontaneous Speech Classifier ================================================================= This is a SELF-CONTAINED script for running inference on a CPU-only "potato laptop". It uses ONNX Runtime (no PyTorch/transformers needed). Requirements (install on the laptop): pip install onnxruntime librosa soundfile numpy scipy tqdm Optional (strongly recommended — much better silence detection): pip install torch torchaudio # for Silero VAD # Silero model (~2MB) is downloaded automatically on first run Files to copy to the laptop: 1. This script (predict_cpu.py) 2. One or more ONNX model files, for example: - checkpoints/speech_classifier_quant.onnx (5sec wav2vec2) - checkpoints/speech_classifier_wav2vec2_7_5sec_quant.onnx - checkpoints/speech_classifier_wav2vec2_10sec_quant.onnx - checkpoints/speech_classifier_wav2vec2_12_5sec_quant.onnx - checkpoints/speech_classifier_wav2vec2_15sec_quant.onnx - checkpoints/speech_classifier_wavlm_5sec_quant.onnx Usage: python predict_cpu.py --audio interview.wav python predict_cpu.py --audio company_recordings/ python predict_cpu.py --audio company_recordings/ --output results.json python predict_cpu.py --audio interview.wav --verbose python predict_cpu.py --audio interview.wav --model 10sec python predict_cpu.py --audio interview.wav --model wavlm """ import os import sys import json import argparse import time from pathlib import Path import numpy as np import librosa import onnxruntime as ort from scipy.ndimage import median_filter from tqdm import tqdm # ============================================================ # Configuration # ============================================================ DEFAULT_CONFIG = { "sample_rate": 16000, "window_sec": 5.0, "min_speech_ratio": 0.20, "vad_energy_threshold": 0.01, "max_duration_sec": 120, "temporal_smooth_window": 3, "read_threshold": 0.45, # tuned on company data "min_conf": 0.0, # disabled — was hurting recall "min_segment_sec": 3.0, "vad_merge_gap_sec": 1.0, } MODEL_ALIAS_TO_FILE = { # 5sec wav2vec2 (production) "5sec": "speech_classifier_quant.onnx", "wav2vec2": "speech_classifier_quant.onnx", "wav2vec2_5sec": "speech_classifier_quant.onnx", # 7.5sec "7sec": "speech_classifier_wav2vec2_7_5sec_quant.onnx", "7.5": "speech_classifier_wav2vec2_7_5sec_quant.onnx", "7.5sec": "speech_classifier_wav2vec2_7_5sec_quant.onnx", "7_5sec": "speech_classifier_wav2vec2_7_5sec_quant.onnx", "wav2vec2_7sec": "speech_classifier_wav2vec2_7_5sec_quant.onnx", "wav2vec2_7_5sec": "speech_classifier_wav2vec2_7_5sec_quant.onnx", # 10sec "10": "speech_classifier_wav2vec2_10sec_quant.onnx", "10sec": "speech_classifier_wav2vec2_10sec_quant.onnx", "wav2vec2_10sec": "speech_classifier_wav2vec2_10sec_quant.onnx", # 12.5sec "12.5": "speech_classifier_wav2vec2_12_5sec_quant.onnx", "12.5sec": "speech_classifier_wav2vec2_12_5sec_quant.onnx", "12_5sec": "speech_classifier_wav2vec2_12_5sec_quant.onnx", "wav2vec2_12_5sec": "speech_classifier_wav2vec2_12_5sec_quant.onnx", # 15sec "15": "speech_classifier_wav2vec2_15sec_quant.onnx", "15sec": "speech_classifier_wav2vec2_15sec_quant.onnx", "wav2vec2_15sec": "speech_classifier_wav2vec2_15sec_quant.onnx", # WavLM "wavlm": "speech_classifier_wavlm_5sec_quant.onnx", "wavlm5sec": "speech_classifier_wavlm_5sec_quant.onnx", "wavlm_5sec": "speech_classifier_wavlm_5sec_quant.onnx", } # Exact filename -> window size mapping (checked before pattern matching) WINDOW_SEC_MAP = { "speech_classifier_quant.onnx": 5.0, "speech_classifier_wav2vec2_7_5sec_quant.onnx": 7.5, "speech_classifier_wav2vec2_10sec_quant.onnx": 10.0, "speech_classifier_wav2vec2_12_5sec_quant.onnx": 12.5, "speech_classifier_wav2vec2_15sec_quant.onnx": 15.0, "speech_classifier_wavlm_5sec_quant.onnx": 5.0, } # ============================================================ # Audio Loading # ============================================================ def load_audio(path: str, sr: int = 16000, max_duration: float = 120.0) -> np.ndarray: """Load audio file, convert to mono, resample. Returns waveform array.""" audio, _ = librosa.load(path, sr=sr, mono=True, duration=max_duration) peak = np.max(np.abs(audio)) if peak > 1e-6: audio = audio / peak * 0.95 return audio # ============================================================ # Voice Activity Detection # ============================================================ def load_silero_vad(): """Load Silero VAD model. Returns (model, get_speech_timestamps_fn).""" import torch model, utils = torch.hub.load( repo_or_dir="snakers4/silero-vad", model="silero_vad", force_reload=False, verbose=False, ) get_speech_timestamps = utils[0] return model, get_speech_timestamps def get_speech_segments_silero( audio: np.ndarray, sr: int, vad_model, get_ts_fn, min_silence_ms: int = 300, min_speech_ms: int = 250, ) -> list[dict]: """Run Silero VAD. Returns list of {start, end} dicts in seconds.""" import torch audio_t = torch.from_numpy(audio).float() timestamps = get_ts_fn( audio_t, vad_model, sampling_rate=sr, min_silence_duration_ms=min_silence_ms, min_speech_duration_ms=min_speech_ms, return_seconds=True, ) return [{"start": float(t["start"]), "end": float(t["end"])} for t in timestamps] def get_speech_segments_rms( audio: np.ndarray, sr: int, energy_threshold: float = 0.01, ) -> list[dict]: """ RMS-based VAD using a global threshold from the full recording. Returns list of {start, end} dicts in seconds. """ rms = librosa.feature.rms(y=audio, frame_length=512, hop_length=256)[0] hop = 256 p30 = np.percentile(rms, 30) p90 = np.percentile(rms, 90) dynamic_range = p90 - p30 if dynamic_range < 0.001: thresh = max(energy_threshold, 0.002) else: thresh = max(p30 + 0.2 * dynamic_range, 0.002) speech_mask = rms > thresh segments = [] in_speech = False seg_start = 0 for i, is_speech in enumerate(speech_mask): if is_speech and not in_speech: seg_start = i in_speech = True elif not is_speech and in_speech: segments.append({ "start": round(seg_start * hop / sr, 3), "end": round(i * hop / sr, 3), }) in_speech = False if in_speech: segments.append({ "start": round(seg_start * hop / sr, 3), "end": round(len(audio) / sr, 3), }) return segments # ============================================================ # Windowing # ============================================================ def adaptive_hop(window_sec: float, floor_sec: float = 2.5, ratio: float = 0.4) -> float: """Compute hop size scaling with window, floored at floor_sec. window_sec -> hop_sec 5.0 -> 2.5 7.5 -> 3.0 10.0 -> 4.0 12.5 -> 5.0 15.0 -> 6.0 """ return max(floor_sec, window_sec * ratio) def make_vad_gated_windows( audio: np.ndarray, sr: int, speech_segments: list[dict], window_samples: int, hop_sec: float, merge_gap_sec: float = 1.0, ) -> list[tuple]: """ Return (chunk, start_sec, end_sec) tuples windowed ONLY over VAD-confirmed speech regions. Silence is never passed to the model. """ merged = [] for seg in speech_segments: if merged and (seg["start"] - merged[-1]["end"]) < merge_gap_sec: merged[-1]["end"] = seg["end"] else: merged.append(dict(seg)) win_samp = int(window_samples) hop_samp = int(hop_sec * sr) windows = [] for seg in merged: seg_start_samp = int(seg["start"] * sr) seg_end_samp = int(seg["end"] * sr) seg_audio = audio[seg_start_samp:seg_end_samp] if len(seg_audio) < win_samp // 2: chunk = np.pad(seg_audio, (0, win_samp - len(seg_audio))) windows.append((chunk, seg["start"], seg["end"])) continue pos = 0 while pos < len(seg_audio): chunk = seg_audio[pos : pos + win_samp] if len(chunk) < win_samp: chunk = np.pad(chunk, (0, win_samp - len(chunk))) start_sec = seg["start"] + pos / sr end_sec = seg["start"] + (pos + win_samp) / sr windows.append((chunk, start_sec, end_sec)) pos += hop_samp if pos + win_samp // 4 >= len(seg_audio): break return windows # ============================================================ # ONNX Inference # ============================================================ class ONNXClassifier: """Lightweight ONNX Runtime wrapper for the speech classifier.""" def __init__(self, model_path: str, window_samples: int): opts = ort.SessionOptions() opts.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL opts.intra_op_num_threads = os.cpu_count() or 4 opts.inter_op_num_threads = 2 self.session = ort.InferenceSession( model_path, sess_options=opts, providers=["CPUExecutionProvider"], ) self.input_name = self.session.get_inputs()[0].name self.output_name = self.session.get_outputs()[0].name # Prefer the static ONNX input length if available. # This avoids mismatches like configured 120000 vs model-required 80000. model_shape = self.session.get_inputs()[0].shape model_samples = None if isinstance(model_shape, (list, tuple)) and len(model_shape) >= 2: dim = model_shape[-1] if isinstance(dim, int) and dim > 0: model_samples = dim self.window_samples = int(model_samples or window_samples) # Warmup with correct window size dummy = np.zeros((1, self.window_samples), dtype=np.float32) self.session.run([self.output_name], {self.input_name: dummy}) def predict_batch(self, waveforms: np.ndarray) -> np.ndarray: """ Args: waveforms: (B, window_samples) float32 Returns: probs: (B, 2) — [p_spontaneous, p_read] """ logits = self.session.run( [self.output_name], {self.input_name: waveforms.astype(np.float32)}, )[0] exp_l = np.exp(logits - np.max(logits, axis=-1, keepdims=True)) return exp_l / exp_l.sum(axis=-1, keepdims=True) # ============================================================ # Segment Construction # ============================================================ def _merge_segments(window_preds: list[dict]) -> list[dict]: """Merge consecutive windows with the same label into segments.""" if not window_preds: return [] segments = [] cur = window_preds[0] seg_start = cur["start_sec"] seg_confs = [cur["confidence"]] for wp in window_preds[1:]: if wp["label"] != cur["label"]: seg_end = wp["start_sec"] segments.append({ "start_sec": round(seg_start, 2), "end_sec": round(seg_end, 2), "duration_sec": round(seg_end - seg_start, 2), "label": cur["label"], "confidence": round(float(np.mean(seg_confs)), 3), }) seg_start = wp["start_sec"] seg_confs = [wp["confidence"]] cur = wp else: seg_confs.append(wp["confidence"]) cur = wp segments.append({ "start_sec": round(seg_start, 2), "end_sec": round(cur["end_sec"], 2), "duration_sec": round(cur["end_sec"] - seg_start, 2), "label": cur["label"], "confidence": round(float(np.mean(seg_confs)), 3), }) return segments def enforce_min_segment_length(segments: list[dict], min_sec: float = 3.0) -> list[dict]: """Merge segments shorter than min_sec into their neighbor.""" changed = True while changed: changed = False out = [] i = 0 while i < len(segments): seg = segments[i] if seg["duration_sec"] < min_sec and seg["label"] not in ("silence", "uncertain"): if out: out[-1]["end_sec"] = seg["end_sec"] out[-1]["duration_sec"] = round( out[-1]["end_sec"] - out[-1]["start_sec"], 2) changed = True elif i + 1 < len(segments): segments[i + 1]["start_sec"] = seg["start_sec"] segments[i + 1]["duration_sec"] = round( segments[i + 1]["end_sec"] - segments[i + 1]["start_sec"], 2) changed = True else: out.append(seg) else: out.append(seg) i += 1 segments = out return segments def _empty_result(path: str, duration: float) -> dict: return { "filepath": path, "filename": Path(path).name, "duration_sec": round(duration, 2), "overall_label": "silence", "overall_confidence": 1.0, "read_ratio": 0.0, "cheating_suspected": False, "segments": [], "window_predictions": [], "processing_time_sec": 0.0, } # ============================================================ # Main Prediction Pipeline # ============================================================ def predict_file( audio_path: str, classifier: ONNXClassifier, cfg: dict, vad_model=None, get_ts_fn=None, batch_size: int = 4, ) -> dict: sr = cfg["sample_rate"] window_samples = int(cfg.get("window_samples", cfg["window_sec"] * sr)) window_sec = window_samples / sr min_conf = cfg["min_conf"] smooth_window = cfg["temporal_smooth_window"] read_threshold = cfg["read_threshold"] t0 = time.perf_counter() audio = load_audio(audio_path, sr=sr, max_duration=cfg["max_duration_sec"]) total_duration = len(audio) / sr # --- VAD --- if vad_model is not None: try: speech_segments = get_speech_segments_silero(audio, sr, vad_model, get_ts_fn) except Exception as e: print(f" [VAD] Silero failed ({e}), using RMS fallback") speech_segments = None else: speech_segments = None if not speech_segments: speech_segments = get_speech_segments_rms(audio, sr, cfg["vad_energy_threshold"]) if not speech_segments: result = _empty_result(audio_path, total_duration) result["processing_time_sec"] = round(time.perf_counter() - t0, 2) return result # --- Windowing --- hop_sec = adaptive_hop(window_sec) all_windows = make_vad_gated_windows( audio, sr, speech_segments, window_samples, hop_sec, merge_gap_sec=cfg["vad_merge_gap_sec"], ) if not all_windows: result = _empty_result(audio_path, total_duration) result["processing_time_sec"] = round(time.perf_counter() - t0, 2) return result chunks = np.stack([w[0] for w in all_windows]) starts = [w[1] for w in all_windows] ends = [w[2] for w in all_windows] # --- Inference --- all_probs = [] for i in range(0, len(chunks), batch_size): all_probs.append(classifier.predict_batch(chunks[i : i + batch_size])) probs = np.concatenate(all_probs, axis=0) # --- Per-window predictions --- window_preds = [] for i, (start, end) in enumerate(zip(starts, ends)): pred_cls = int(np.argmax(probs[i])) conf = float(probs[i][pred_cls]) if min_conf > 0 and conf < min_conf: label = "uncertain" else: label = "spontaneous" if pred_cls == 0 else "read" window_preds.append({ "start_sec": round(start, 2), "end_sec": round(end, 2), "label": label, "confidence": round(conf, 3), }) # --- Temporal smoothing --- voting_idx = [ i for i, wp in enumerate(window_preds) if wp["label"] in ("spontaneous", "read") ] if len(voting_idx) >= smooth_window: labels_num = np.array([ 0 if window_preds[i]["label"] == "spontaneous" else 1 for i in voting_idx ]) smoothed = median_filter(labels_num, size=smooth_window).astype(int) for j, i in enumerate(voting_idx): window_preds[i]["label"] = "spontaneous" if smoothed[j] == 0 else "read" # --- Segments --- segments = _merge_segments(window_preds) segments = enforce_min_segment_length(segments, min_sec=cfg["min_segment_sec"]) # --- Overall label --- speaking = [wp for wp in window_preds if wp["label"] in ("spontaneous", "read")] if not speaking: result = _empty_result(audio_path, total_duration) result["processing_time_sec"] = round(time.perf_counter() - t0, 2) return result read_count = sum(1 for wp in speaking if wp["label"] == "read") read_ratio = read_count / len(speaking) overall_label = "read" if read_ratio >= read_threshold else "spontaneous" same_label = [wp for wp in speaking if wp["label"] == overall_label] overall_conf = float(np.mean([wp["confidence"] for wp in same_label])) return { "filepath": audio_path, "filename": Path(audio_path).name, "duration_sec": round(total_duration, 2), "overall_label": overall_label, "overall_confidence": round(overall_conf, 3), "read_ratio": round(read_ratio, 3), "cheating_suspected": overall_label == "read", "segments": segments, "window_predictions": window_preds, "processing_time_sec": round(time.perf_counter() - t0, 2), } # ============================================================ # Reporting # ============================================================ def format_report(result: dict, verbose: bool = False) -> str: lines = [] lines.append(f"{'='*65}") lines.append(f" File: {result['filename']}") lines.append(f" Duration: {result['duration_sec']}s | " f"Processed in: {result['processing_time_sec']}s") lines.append(f"{'='*65}") verdict = ("!! READING DETECTED !!" if result["overall_label"] == "read" else "OK — Spontaneous") lines.append(f" VERDICT: {verdict}") lines.append(f" Confidence: {result['overall_confidence']:.1%}") lines.append(f" Read ratio: {result['read_ratio']:.1%} of speaking time") lines.append("") lines.append(" --- TIMELINE ---") for seg in result["segments"]: marker = "██" if seg["label"] == "read" else ("░░" if seg["label"] == "spontaneous" else "··") lines.append( f" {marker} [{seg['start_sec']:6.1f}s - {seg['end_sec']:6.1f}s] " f"{seg['label']:12s} conf={seg['confidence']:.0%} " f"({seg['duration_sec']:.1f}s)" ) if verbose: lines.append("") lines.append(f" --- WINDOWS ({len(result['window_predictions'])}) ---") for wp in result["window_predictions"]: lines.append( f" [{wp['start_sec']:6.1f}s-{wp['end_sec']:6.1f}s] " f"{wp['label']:12s} conf={wp['confidence']:.2f}" ) lines.append("") return "\n".join(lines) def format_summary_table(results: list[dict]) -> str: lines = [] lines.append(f"\n{'='*85}") lines.append(f" BATCH SUMMARY — {len(results)} files") lines.append(f"{'='*85}") lines.append(f" {'Filename':<40s} {'Verdict':<14s} {'Conf':>6s} {'Read%':>6s} {'Time':>6s}") lines.append(f" {'-'*40} {'-'*14} {'-'*6} {'-'*6} {'-'*6}") for r in results: flag = "** READ **" if r["overall_label"] == "read" else "spontaneous" lines.append( f" {r['filename']:<40s} {flag:<14s} " f"{r['overall_confidence']:5.0%} {r['read_ratio']:5.0%} " f"{r['processing_time_sec']:5.1f}s" ) read_n = sum(1 for r in results if r["overall_label"] == "read") spont_n = sum(1 for r in results if r["overall_label"] == "spontaneous") avg_conf = np.mean([r["overall_confidence"] for r in results]) total_t = sum(r["processing_time_sec"] for r in results) lines.append(f" {'-'*78}") lines.append(f" Read (cheating suspected): {read_n}") lines.append(f" Spontaneous (OK): {spont_n}") lines.append(f" Avg confidence: {avg_conf:.1%}") lines.append(f" Total processing time: {total_t:.1f}s") lines.append(f"{'='*85}") return "\n".join(lines) # ============================================================ # CLI helpers # ============================================================ def find_audio_files(path: Path) -> list[str]: AUDIO_EXTS = {".wav", ".mp3", ".m4a", ".flac", ".ogg", ".wma", ".aac", ".webm"} if path.is_file(): return [str(path)] elif path.is_dir(): files = [] for ext in AUDIO_EXTS: files.extend(path.rglob(f"*{ext}")) files.extend(path.rglob(f"*{ext.upper()}")) return sorted(set(str(f) for f in files)) return [] def infer_window_sec_from_model_name(model_path: str, fallback: float = 5.0) -> float: """Infer window size from model filename. Exact map checked first, then patterns.""" name = Path(model_path).name.lower() # Check exact filename map first — most reliable for fname, window in WINDOW_SEC_MAP.items(): if fname.lower() == name: return window # Pattern matching — ORDER MATTERS (specific before generic) if "12_5sec" in name or "12.5sec" in name: return 12.5 if "15sec" in name: return 15.0 if "10sec" in name: return 10.0 if "7_5sec" in name: # must check before plain "7sec" return 7.5 if "7sec" in name: return 7.5 if "wavlm" in name: return 5.0 if "5sec" in name: return 5.0 return fallback def resolve_model_path(model_arg: str) -> str: """Resolve alias or path string to an existing ONNX file path.""" # Direct path candidate = Path(model_arg) if candidate.exists(): return str(candidate) # Alias lookup alias_key = model_arg.lower().strip() if alias_key in MODEL_ALIAS_TO_FILE: for base in (Path("checkpoints"), Path(__file__).parent / "checkpoints"): p = base / MODEL_ALIAS_TO_FILE[alias_key] if p.exists(): return str(p) # Relative to script directory script_relative = Path(__file__).parent / model_arg if script_relative.exists(): return str(script_relative) return model_arg # ============================================================ # Entry point # ============================================================ def main(): parser = argparse.ArgumentParser( description="CPU Inference — Read vs Spontaneous Speech Classifier", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Model aliases: 5sec / wav2vec2 -> speech_classifier_quant.onnx (production) 7sec / 7.5sec / 7_5sec -> speech_classifier_wav2vec2_7_5sec_quant.onnx 10sec -> speech_classifier_wav2vec2_10sec_quant.onnx 12.5sec / 12_5sec -> speech_classifier_wav2vec2_12_5sec_quant.onnx 15sec -> speech_classifier_wav2vec2_15sec_quant.onnx wavlm / wavlm_5sec -> speech_classifier_wavlm_5sec_quant.onnx Examples: python predict_cpu.py --audio interview.wav python predict_cpu.py --audio interview.wav --model 10sec python predict_cpu.py --audio interview.wav --model 7_5sec python predict_cpu.py --audio interview.wav --model wavlm python predict_cpu.py --audio recordings/ --output results.json --no-silero """, ) parser.add_argument("--audio", type=str, required=True, help="Path to audio file or folder") parser.add_argument("--model", type=str, default="checkpoints/speech_classifier_quant.onnx", help="Path or alias (default: 5sec wav2vec2)") parser.add_argument("--window-sec", type=float, default=None, help="Override window size in seconds (inferred from filename by default)") parser.add_argument("--output", type=str, default="outputs/cpu_predictions.json") parser.add_argument("--batch-size", type=int, default=4) parser.add_argument("--read-threshold", type=float, default=None, help="Override read_threshold (default: 0.45)") parser.add_argument("--verbose", action="store_true", help="Print per-window details") parser.add_argument("--no-silero", action="store_true", help="Use RMS VAD instead of Silero") args = parser.parse_args() # --- Find audio files --- audio_files = find_audio_files(Path(args.audio)) if not audio_files: print(f"ERROR: No audio files found at '{args.audio}'") sys.exit(1) print(f"Found {len(audio_files)} audio file(s)") # --- Resolve model --- model_path = resolve_model_path(args.model) if not Path(model_path).exists(): print(f"ERROR: Model not found: '{args.model}'") print("Run with --help to see available aliases") sys.exit(1) # --- Build config --- cfg = DEFAULT_CONFIG.copy() if args.read_threshold is not None: cfg["read_threshold"] = args.read_threshold # Window size: explicit flag > exact filename map > pattern matching > default if args.window_sec is not None: cfg["window_sec"] = args.window_sec else: cfg["window_sec"] = infer_window_sec_from_model_name( model_path, fallback=DEFAULT_CONFIG["window_sec"] ) window_samples = int(cfg["window_sec"] * cfg["sample_rate"]) model_mb = Path(model_path).stat().st_size / 1e6 classifier = ONNXClassifier(model_path, window_samples=window_samples) # Use the model-required input length as the source of truth. # This preserves compatibility for models whose true input differs from filename alias. cfg["window_samples"] = int(classifier.window_samples) effective_window_sec = cfg["window_samples"] / cfg["sample_rate"] print(f"Model: {Path(model_path).name} ({model_mb:.0f} MB)") print(f"Window: {effective_window_sec:.2f}s | " f"Hop: {adaptive_hop(effective_window_sec):.2f}s | " f"Samples: {cfg['window_samples']}") if cfg["window_samples"] != window_samples: print(f"Note: requested {window_samples} samples from --window-sec/name, " f"but ONNX expects {cfg['window_samples']}; using ONNX shape.") print(f"Threshold: {cfg['read_threshold']}") print("Model loaded") # --- Load Silero VAD --- vad_model, get_ts_fn = None, None if not args.no_silero: try: vad_model, get_ts_fn = load_silero_vad() print("Silero VAD loaded") except Exception as e: print(f"Silero VAD unavailable ({e}) — using RMS VAD") else: print("Using RMS VAD (--no-silero)") print() # --- Run inference --- results = [] for fpath in tqdm(audio_files, desc="Processing", disable=len(audio_files) == 1): try: result = predict_file( fpath, classifier, cfg, vad_model=vad_model, get_ts_fn=get_ts_fn, batch_size=args.batch_size, ) results.append(result) print(format_report(result, verbose=args.verbose)) except Exception as e: print(f"ERROR processing {fpath}: {e}") if len(results) > 1: print(format_summary_table(results)) # --- Save JSON --- out_path = Path(args.output) out_path.parent.mkdir(parents=True, exist_ok=True) with open(out_path, "w") as f: json.dump(results, f, indent=2, default=str) print(f"\nResults saved to: {out_path}") if __name__ == "__main__": main()