#!/usr/bin/env python3 """ MIDI Validator - Parallel Processing Version Detects MIDI files with patterns that cause VST crashes and malloc errors Outputs only risky files to JSON Usage: python midi_validator.py /path/to/midi/folder python midi_validator.py /path/to/midi/folder --delete python midi_validator.py /path/to/midi/folder --jobs 8 """ import mido import os import sys import json import glob from pathlib import Path from tqdm import tqdm from collections import defaultdict from joblib import Parallel, delayed import multiprocessing from enum import Enum class MidiRiskType(Enum): """Enumeration of MIDI file risk patterns that can cause VST crashes""" # Timing and Control Change Issues CC123_TEMPO_CRASH = "CC123_TEMPO_CRASH" RAPID_CC_BURST = "RAPID_CC_BURST" LARGE_DELTA_TIMES = "LARGE_DELTA_TIMES" LARGE_DELTA_CC123 = "LARGE_DELTA_CC123" MALLOC_CRASH = "MALLOC_CRASH" EXCESSIVE_CC123 = "EXCESSIVE_CC123" # Sustain Pedal Issues SUSTAIN_CRASH = "SUSTAIN_CRASH" SUSTAIN_RAPID = "SUSTAIN_RAPID" EXCESSIVE_SUSTAIN = "EXCESSIVE_SUSTAIN" # Pitch Bend Issues PITCH_BEND_BURST = "PITCH_BEND_BURST" PITCH_BEND_EXTREME = "PITCH_BEND_EXTREME" PITCH_BEND_DENSITY = "PITCH_BEND_DENSITY" # Note Issues NOTE_BURST = "NOTE_BURST" NOTE_OVERFLOW = "NOTE_OVERFLOW" # Program Change Issues PROGRAM_CHANGE_BURST = "PROGRAM_CHANGE_BURST" # Tempo Issues EXCESSIVE_TEMPO_CHANGES = "EXCESSIVE_TEMPO_CHANGES" # Message Burst Issues MESSAGE_BURST = "MESSAGE_BURST" HIGH_DENSITY = "HIGH_DENSITY" ZERO_TIME_BURST = "ZERO_TIME_BURST" COMPLEX_BURST = "COMPLEX_BURST" CC_BURST = "CC_BURST" CC_COMPLEXITY = "CC_COMPLEXITY" # System Exclusive Issues SYSEX_BURST = "SYSEX_BURST" SYSEX_OVERFLOW = "SYSEX_OVERFLOW" # File Size Issues EXCESSIVE_MESSAGES = "EXCESSIVE_MESSAGES" EXCESSIVE_LENGTH = "EXCESSIVE_LENGTH" HIGH_DENSITY_OVERALL = "HIGH_DENSITY_OVERALL" # Parse Issues PARSE_ERROR = "PARSE_ERROR" def find_midi_files(path): """Find all MIDI files recursively""" if os.path.isfile(path): return [path] if path.lower().endswith((".mid", ".midi")) else [] midi_files = [] for pattern in ["*.mid", "*.midi", "*.MID", "*.MIDI"]: midi_files.extend(Path(path).rglob(pattern)) return [str(f) for f in sorted(midi_files)] def is_risky_file(midi_path): """Check if MIDI file has crash patterns - returns (is_risky, risk_type, details, filepath)""" try: midi_file = mido.MidiFile(midi_path) # Track message timing patterns timestamp_messages = {} # timestamp -> message_count cc123_total = 0 cc123_zero_time = 0 cc123_large_delta = 0 cc64_total = 0 cc64_zero_time = 0 cc64_rapid_changes = 0 pitch_bend_total = 0 pitch_bend_zero_time = 0 pitch_bend_extreme = 0 program_change_bursts = 0 large_delta_times = 0 tempo_changes = 0 cc123_near_tempo = 0 rapid_cc_bursts = 0 for track in midi_file.tracks: current_time = 0 prev_cc64_value = None cc64_changes_in_sequence = 0 program_changes_at_timestamp = {} recent_messages = [] consecutive_rapid_ccs = 0 for msg in track: current_time += msg.time # Track recent messages for complex pattern detection recent_messages.append(msg) if len(recent_messages) > 10: recent_messages.pop(0) # Check for extremely large delta times if msg.time > 10000: large_delta_times += 1 # Count messages at each timestamp if current_time not in timestamp_messages: timestamp_messages[current_time] = 0 timestamp_messages[current_time] += 1 # Check for tempo changes if msg.type == "set_tempo": tempo_changes += 1 # Check if CC 123 occurred recently for recent_msg in recent_messages[-5:]: if ( recent_msg.type == "control_change" and getattr(recent_msg, "control", None) == 123 ): cc123_near_tempo += 1 break # Control Change tracking elif msg.type == "control_change": control = getattr(msg, "control", None) value = getattr(msg, "value", None) # Track rapid CC messages if msg.time <= 10: consecutive_rapid_ccs += 1 if consecutive_rapid_ccs > 5: rapid_cc_bursts += 1 consecutive_rapid_ccs = 0 else: consecutive_rapid_ccs = 0 # CC 123 (All Notes Off) patterns if control == 123: cc123_total += 1 if msg.time == 0: cc123_zero_time += 1 elif msg.time > 5000: cc123_large_delta += 1 # CC 64 (Sustain Pedal) patterns elif control == 64: cc64_total += 1 if msg.time == 0: cc64_zero_time += 1 # Track rapid sustain pedal changes if prev_cc64_value is not None: if (prev_cc64_value >= 64 and value < 64) or ( prev_cc64_value < 64 and value >= 64 ): cc64_changes_in_sequence += 1 if msg.time <= 10: cc64_rapid_changes += 1 else: cc64_changes_in_sequence = 0 prev_cc64_value = value # Pitch bend patterns elif msg.type == "pitchwheel": pitch_bend_total += 1 pitch_value = getattr(msg, "pitch", 0) if msg.time == 0: pitch_bend_zero_time += 1 if abs(pitch_value) > 7000: pitch_bend_extreme += 1 # Program change patterns elif msg.type == "program_change": if current_time not in program_changes_at_timestamp: program_changes_at_timestamp[current_time] = 0 program_changes_at_timestamp[current_time] += 1 if program_changes_at_timestamp[current_time] > 5: program_change_bursts += 1 # Check for risk patterns and return first match # Critical timing corruption patterns if cc123_near_tempo > 0: return ( True, MidiRiskType.CC123_TEMPO_CRASH, f"CC 123 messages near tempo changes (timing corruption): {cc123_near_tempo}", midi_path, ) if rapid_cc_bursts > 0: return ( True, MidiRiskType.RAPID_CC_BURST, f"sequences of rapid control changes: {rapid_cc_bursts}", midi_path, ) if large_delta_times > 5: return ( True, MidiRiskType.LARGE_DELTA_TIMES, f"messages with >10k tick deltas: {large_delta_times}", midi_path, ) if cc123_large_delta > 0: return ( True, MidiRiskType.LARGE_DELTA_CC123, f"CC 123 after large time gap (timing buffer corruption): {cc123_large_delta}", midi_path, ) # Memory allocation patterns if cc123_zero_time > 30: return ( True, MidiRiskType.MALLOC_CRASH, f"zero-time CC123 messages: {cc123_zero_time}", midi_path, ) if cc123_total > 100: return True, MidiRiskType.EXCESSIVE_CC123, f"total CC123 messages: {cc123_total}", midi_path # Sustain pedal issues if cc64_zero_time > 5: return ( True, MidiRiskType.SUSTAIN_CRASH, f"zero-time CC64 (sustain) messages: {cc64_zero_time}", midi_path, ) if cc64_rapid_changes > 20: return ( True, MidiRiskType.SUSTAIN_RAPID, f"rapid sustain pedal changes: {cc64_rapid_changes}", midi_path, ) if cc64_total > 200: return ( True, MidiRiskType.EXCESSIVE_SUSTAIN, f"total sustain messages: {cc64_total}", midi_path, ) # Pitch bend issues if pitch_bend_zero_time > 100: return ( True, MidiRiskType.PITCH_BEND_BURST, f"zero-time pitch bend messages: {pitch_bend_zero_time}", midi_path, ) if pitch_bend_extreme > 50: return ( True, MidiRiskType.PITCH_BEND_EXTREME, f"extreme pitch bend values: {pitch_bend_extreme}", midi_path, ) if pitch_bend_total > 5000 and midi_file.length > 0: pitch_density = pitch_bend_total / midi_file.length if pitch_density > 100: return ( True, MidiRiskType.PITCH_BEND_DENSITY, f"pitch bends per second: {pitch_density:.1f}", midi_path, ) # Program change issues if program_change_bursts > 0: return ( True, MidiRiskType.PROGRAM_CHANGE_BURST, f"timestamps with >5 program changes: {program_change_bursts}", midi_path, ) # Tempo issues if tempo_changes > 10: return ( True, MidiRiskType.EXCESSIVE_TEMPO_CHANGES, f"tempo changes: {tempo_changes}", midi_path, ) # Message density issues for timestamp, count in timestamp_messages.items(): if count > 500: return ( True, MidiRiskType.MESSAGE_BURST, f"messages at timestamp {timestamp}: {count}", midi_path, ) high_density_timestamps = sum(1 for count in timestamp_messages.values() if count > 100) if high_density_timestamps > 50: return ( True, MidiRiskType.HIGH_DENSITY, f"timestamps with >100 messages each: {high_density_timestamps}", midi_path, ) # Check for burst patterns in tracks for track in midi_file.tracks: consecutive_zero_time = 0 zero_time_message_types = set() for msg in track: if msg.time == 0: consecutive_zero_time += 1 zero_time_message_types.add(msg.type) if consecutive_zero_time > 1000: return ( True, MidiRiskType.ZERO_TIME_BURST, f"consecutive zero-time messages: {consecutive_zero_time}", midi_path, ) if len(zero_time_message_types) > 10 and consecutive_zero_time > 100: return ( True, MidiRiskType.COMPLEX_BURST, f"zero-time messages of {len(zero_time_message_types)} different types: {consecutive_zero_time}", midi_path, ) else: consecutive_zero_time = 0 zero_time_message_types.clear() # Check for control change bursts for track in midi_file.tracks: cc_burst_count = 0 cc_types_in_burst = set() for msg in track: if msg.type == "control_change" and msg.time == 0: cc_burst_count += 1 cc_types_in_burst.add(getattr(msg, "control", -1)) if cc_burst_count > 200: return ( True, MidiRiskType.CC_BURST, f"zero-time control changes: {cc_burst_count}", midi_path, ) if len(cc_types_in_burst) > 20 and cc_burst_count > 50: return ( True, MidiRiskType.CC_COMPLEXITY, f"zero-time CCs of {len(cc_types_in_burst)} different types: {cc_burst_count}", midi_path, ) else: cc_burst_count = 0 cc_types_in_burst.clear() # Check for note bursts for track in midi_file.tracks: note_burst_count = 0 simultaneous_notes = set() for msg in track: if msg.type in ["note_on", "note_off"] and msg.time == 0: note_burst_count += 1 if msg.type == "note_on" and getattr(msg, "velocity", 0) > 0: note_key = (getattr(msg, "channel", 0), getattr(msg, "note", 0)) simultaneous_notes.add(note_key) if note_burst_count > 300: return ( True, MidiRiskType.NOTE_BURST, f"zero-time note messages: {note_burst_count}", midi_path, ) if len(simultaneous_notes) > 128: return ( True, MidiRiskType.NOTE_OVERFLOW, f"simultaneous notes: {len(simultaneous_notes)}", midi_path, ) else: note_burst_count = 0 simultaneous_notes.clear() # Check for system exclusive bursts for track in midi_file.tracks: sysex_burst_count = 0 sysex_total_bytes = 0 for msg in track: if msg.type == "sysex" and msg.time == 0: sysex_burst_count += 1 if hasattr(msg, "data"): sysex_total_bytes += len(msg.data) if sysex_burst_count > 10: return ( True, MidiRiskType.SYSEX_BURST, f"zero-time sysex messages: {sysex_burst_count}", midi_path, ) if sysex_total_bytes > 32768: return ( True, MidiRiskType.SYSEX_OVERFLOW, f"bytes of sysex data: {sysex_total_bytes}", midi_path, ) else: sysex_burst_count = 0 sysex_total_bytes = 0 # Check overall file characteristics total_messages = sum(len(track) for track in midi_file.tracks) if total_messages > 500000: return True, MidiRiskType.EXCESSIVE_MESSAGES, f"total messages: {total_messages}", midi_path if midi_file.length > 3600: return True, MidiRiskType.EXCESSIVE_LENGTH, f"seconds: {midi_file.length:.1f}", midi_path if total_messages > 0 and midi_file.length > 0: messages_per_second = total_messages / midi_file.length if messages_per_second > 10000: return ( True, MidiRiskType.HIGH_DENSITY_OVERALL, f"messages per second: {messages_per_second:.1f}", midi_path, ) return False, None, None, midi_path except Exception as e: return True, MidiRiskType.PARSE_ERROR, str(e), midi_path def process_single_file(midi_file): """Process a single MIDI file - wrapper for joblib""" is_risky, risk_type, details, filepath = is_risky_file(midi_file) if is_risky: return { "file": filepath, "risk_type": risk_type.value, "details": details, "reason": f"{risk_type.value}: {details}", # Keep for backwards compatibility } return None def validate_midi_folder(folder_path, delete_risky=False, n_jobs=-1): """Validate all MIDI files in folder using parallel processing""" print(f"Scanning for MIDI files in: {folder_path}") midi_files = find_midi_files(folder_path) if not midi_files: print("No MIDI files found!") return [] print(f"Found {len(midi_files)} MIDI files") # Determine number of jobs if n_jobs == -1: n_jobs = multiprocessing.cpu_count() print(f"Using {n_jobs} parallel processes") # Process files in parallel with progress bar print("Processing files in parallel...") results = Parallel(n_jobs=n_jobs, backend="multiprocessing")( delayed(process_single_file)(midi_file) for midi_file in tqdm(midi_files, desc="Validating MIDI files") ) # Filter out None results (safe files) risky_files = [result for result in results if result is not None] # Handle deletion if requested deleted_files = [] if delete_risky and risky_files: print(f"\nDeleting {len(risky_files)} risky files...") for file_info in tqdm(risky_files, desc="Deleting files"): try: os.remove(file_info["file"]) deleted_files.append(file_info["file"]) except Exception as e: print(f"Warning: Could not delete {file_info['file']}: {e}") print(f"Successfully deleted {len(deleted_files)} files") return risky_files def main(): import argparse parser = argparse.ArgumentParser( description="Validate MIDI files for VST crash patterns (Parallel Processing)" ) parser.add_argument("folder", help="Path to MIDI folder") parser.add_argument("--delete", action="store_true", help="Delete risky files automatically") parser.add_argument( "--jobs", "-j", type=int, default=-1, help="Number of parallel jobs (default: use all CPU cores)" ) args = parser.parse_args() if not os.path.exists(args.folder): print(f"Error: Path does not exist: {args.folder}") sys.exit(1) # Show warning for delete mode if args.delete: print("⚠️ DELETE MODE: Risky files will be permanently deleted!") response = input("Continue? (y/N): ") if response.lower() != "y": print("Cancelled.") sys.exit(0) # Validate files with parallel processing print(f"Starting parallel validation with {args.jobs} jobs...") risky_files = validate_midi_folder(args.folder, delete_risky=args.delete, n_jobs=args.jobs) # Output results output_file = "risky_midi_files.json" with open(output_file, "w") as f: json.dump( { "scan_path": args.folder, "total_risky_files": len(risky_files), "files_deleted": args.delete, "parallel_jobs_used": args.jobs if args.jobs != -1 else multiprocessing.cpu_count(), "risky_files": risky_files, }, f, indent=2, ) print(f"\nResults:") print(f" Risky files found: {len(risky_files)}") if args.delete: print(f" Files deleted: {len(risky_files)}") print(f" Results saved to: {output_file}") if risky_files and not args.delete: print(f"\nTop 10 risky files:") for i, file_info in enumerate(risky_files[:10]): print(f" {i+1}. {os.path.basename(file_info['file'])}: {file_info['risk_type']}") elif args.delete and risky_files: print(f"\nDeleted files by risk type:") risk_counts = {} for file_info in risky_files: risk_type = file_info["risk_type"] risk_counts[risk_type] = risk_counts.get(risk_type, 0) + 1 for risk_type, count in sorted(risk_counts.items()): print(f" {risk_type}: {count} files") if __name__ == "__main__": main()