diff --git a/src/ria_toolkit_oss_cli/ria_toolkit_oss/annotate.py b/src/ria_toolkit_oss_cli/ria_toolkit_oss/annotate.py new file mode 100644 index 0000000..74e1370 --- /dev/null +++ b/src/ria_toolkit_oss_cli/ria_toolkit_oss/annotate.py @@ -0,0 +1,813 @@ +"""Annotate command - Automatic detection and manual annotation management.""" + +import json +from pathlib import Path + +import click + +from ria_toolkit_oss.annotations import ( + annotate_with_cusum, + detect_signals_energy, + split_recording_annotations, + threshold_qualifier, +) +from ria_toolkit_oss.data import Annotation +from ria_toolkit_oss.data.recording import Recording +from ria_toolkit_oss.io import load_recording, to_blue, to_npy, to_sigmf, to_wav +from ria_toolkit_oss_cli.common import format_frequency, format_sample_count + + +def normalize_sigmf_path(filepath): + """Normalize SigMF path to base name without extension.""" + path = Path(filepath) + + # Handle .sigmf-data, .sigmf-meta, or .sigmf + if ".sigmf" in path.suffix: + # Remove the suffix to get base name + return path.with_suffix("") + else: + return path + + +def detect_input_format(filepath): + """Detect file format from extension.""" + path = Path(filepath) + ext = path.suffix.lower() + + if ext in [".sigmf-data", ".sigmf-meta"]: + return "sigmf" + elif path.name.endswith(".sigmf"): + return "sigmf" + elif ext == ".npy": + return "npy" + elif ext == ".wav": + return "wav" + elif ext == ".blue": + return "blue" + else: + raise click.ClickException(f"Unknown format for '{filepath}'. Supported: .sigmf, .npy, .wav, .blue") + + +def determine_output_path(input_path, output_path, fmt, quiet, overwrite): + input_path = Path(input_path) + + if output_path: + target = Path(output_path) + final_path = target + else: + annotated_name = f"{input_path.stem}_annotated" + target = input_path.with_name(f"{annotated_name}{input_path.suffix}") + + if fmt == "sigmf": + final_path = normalize_sigmf_path(target) + if not quiet: + click.echo(f"Saving SigMF metadata to: {final_path}") + else: + final_path = target + if not quiet: + click.echo(f"Saving to: {final_path}") + + if final_path.exists() and not overwrite and final_path != input_path: + click.echo(f"Error: {final_path} already exists. Use --overwrite to replace it.", err=True) + return None + + return final_path + + +def save_recording_auto(recording, output_path, input_path, quiet=False, overwrite=False): + """Save recording, auto-detecting format from extension. + + For SigMF: Only overwrites metadata file, data file is unchanged + For other formats: Creates _annotated copy by default, unless overwrite=True + """ + input_path = Path(input_path) + fmt = detect_input_format(input_path) + + # Determine output path + output_path = determine_output_path( + input_path=input_path, output_path=output_path, fmt=fmt, quiet=quiet, overwrite=overwrite + ) + + if fmt == "sigmf": + # Normalize path for SigMF + base_path = output_path + stem = base_path.name + parent = base_path.parent + + # For SigMF: only save metadata, copy data if needed + meta_path = parent / f"{stem}.sigmf-meta" + data_path = parent / f"{stem}.sigmf-data" + + # If output is different from input, copy data file + input_base = normalize_sigmf_path(input_path) + if input_base != base_path: + import shutil + + # Construct input data path correctly + # input_base is like /path/to/recording or /path/to/recording.sigmf + # We need /path/to/recording.sigmf-data + if str(input_base).endswith(".sigmf"): + input_data = Path(str(input_base).replace(".sigmf", ".sigmf-data")) + else: + input_data = input_base.parent / f"{input_base.name}.sigmf-data" + if not quiet: + click.echo(f" Copying: {data_path}") + shutil.copy2(input_data, data_path) + + # Always save metadata (this is the whole point) + to_sigmf(recording, filename=stem, path=parent, overwrite=True) + + if not quiet: + click.echo(f" Updated: {meta_path}") + if input_base != base_path: + click.echo(f" Created: {data_path}") + + elif fmt == "npy": + to_npy(recording, filename=output_path.stem, path=output_path.parent, overwrite=True) + if not quiet: + click.echo(f" Created: {output_path}") + elif fmt == "wav": + to_wav(recording, filename=output_path.stem, path=output_path.parent, overwrite=True) + if not quiet: + click.echo(f" Created: {output_path}") + elif fmt == "blue": + to_blue(recording, filename=output_path.stem, path=output_path.parent, overwrite=True) + if not quiet: + click.echo(f" Created: {output_path}") + + +def determine_frequency_bounds(recording: Recording, freq_lower, freq_upper): + # Handle frequency bounds + if (freq_lower is None) != (freq_upper is None): + raise click.ClickException("Must specify both --freq-lower and --freq-upper, or neither") + + if freq_lower is None: + # Default to full bandwidth + sample_rate = recording.metadata.get("sample_rate", 1) + center_freq = recording.metadata.get("center_frequency", 0) + freq_lower = center_freq - (sample_rate / 2) + freq_upper = center_freq + (sample_rate / 2) + freq_default = True + else: + freq_default = False + if freq_lower >= freq_upper: + raise click.ClickException( + f"Invalid frequency range: lower ({format_frequency(freq_lower)}) " + f"must be < upper ({format_frequency(freq_upper)})" + ) + + return freq_lower, freq_upper, freq_default + + +def get_indices_list(indices, recording: Recording): + if indices: + try: + indices_list = [int(idx.strip()) for idx in indices.split(",")] + # Validate indices + for idx in indices_list: + if idx < 0 or idx >= len(recording.annotations): + raise click.ClickException( + f"Invalid index {idx}. Recording has {len(recording.annotations)} annotation(s)" + ) + except ValueError as e: + raise click.ClickException(f"Invalid indices format. Expected comma-separated integers: {e}") + + return indices_list + else: + return None + + +# ============================================================================ +# Main command group +# ============================================================================ + + +@click.group() +def annotate(): + """Manage and auto-detect annotations on RF recordings. + + \b + MANUAL MANAGEMENT: + list - List all current annotations + add - Manually add a specific annotation + remove - Delete an annotation by its index + clear - Remove all annotations from the recording + + \b + DETECTION & SEPARATION: + energy - Auto-detect using energy-based thresholding + cusum - Auto-detect segments using signal state changes + threshold - Auto-detect samples above magnitude percentage + separate - Auto-detect parallel frequency-offset signals, split into sub-bands + + \b + File Path Handling: + - SigMF files: Pass .sigmf-data, .sigmf-meta, or base name + - Other formats: .npy, .wav, .blue files + + \b + Output Behavior: + - SigMF: Updates .sigmf-meta only (data unchanged), in-place + - Other: Creates _annotated copy unless --overwrite specified + """ + pass + + +# ============================================================================ +# List subcommand +# ============================================================================ + + +@annotate.command() +@click.argument("input", type=click.Path(exists=True)) +@click.option("--verbose", is_flag=True, help="Show detailed annotation info") +def list(input, verbose): + """List all annotations in a recording. + + \b + Examples: + utils annotate list recording.sigmf-data + utils annotate list signal.npy --verbose + """ + try: + recording = load_recording(input) + except Exception as e: + raise click.ClickException(f"Failed to load recording: {e}") + + if len(recording.annotations) == 0: + click.echo(f"No annotations in {Path(input).name}") + return + + click.echo(f"\nAnnotations in {Path(input).name}:") + for i, ann in enumerate(recording.annotations): + # Parse type from comment JSON + try: + comment_data = json.loads(ann.comment) + ann_type = comment_data.get("type", "unknown") + user_comment = comment_data.get("user_comment", "") + except (json.JSONDecodeError, TypeError): + ann_type = "unknown" + user_comment = ann.comment or "" + + # Basic info + freq_range = f"{format_frequency(ann.freq_lower_edge)} - {format_frequency(ann.freq_upper_edge)}" + click.echo( + f" [{i}] Samples {format_sample_count(ann.sample_start)}-" + f"{format_sample_count(ann.sample_start + ann.sample_count)}: {ann.label}" + ) + click.echo(f" Type: {ann_type}") + + if verbose: + if user_comment: + click.echo(f" Comment: {user_comment}") + click.echo(f" Frequency: {freq_range}") + if ann.detail: + click.echo(f" Detail: {ann.detail}") + + click.echo(f"\nTotal: {len(recording.annotations)} annotation(s)") + + +# ============================================================================ +# Add subcommand +# ============================================================================ + + +@annotate.command(context_settings={"max_content_width": 200}) +@click.argument("input", type=click.Path(exists=True)) +@click.option("--start", type=int, required=True, help="Start sample index") +@click.option("--count", type=int, required=True, help="Sample count") +@click.option("--label", type=str, required=True, help="Annotation label") +@click.option("--freq-lower", type=float, help="Lower frequency edge (Hz)") +@click.option("--freq-upper", type=float, help="Upper frequency edge (Hz)") +@click.option("--comment", type=str, help="Human-readable comment") +@click.option( + "--type", + "annotation_type", + type=click.Choice(["standalone", "parallel", "intersection"]), + default="standalone", + help="Annotation type", +) +@click.option("--output", "-o", type=click.Path(), help="Output file path") +@click.option("--overwrite", is_flag=True, help="Overwrite input file (non-SigMF only)") +@click.option("--quiet", is_flag=True, help="Quiet mode") +def add(input, start, count, label, freq_lower, freq_upper, comment, annotation_type, output, overwrite, quiet): + """Add a manual annotation. + + \b + Examples: + utils annotate add file.npy --start 1000 --count 500 --label wifi + utils annotate add signal.sigmf-data --start 0 --count 1000 --label burst --comment "Strong signal" + """ + try: + recording = load_recording(input) + if not quiet: + click.echo(f"Loaded: {input}") + except Exception as e: + raise click.ClickException(f"Failed to load recording: {e}") + + # Validate sample range + n_samples = len(recording.data[0]) + if start < 0: + raise click.ClickException(f"--start must be >= 0, got {start}") + if count <= 0: + raise click.ClickException(f"--count must be > 0, got {count}") + if start + count > n_samples: + raise click.ClickException( + f"Invalid annotation range:\n" + f" Start: {start:,}\n" + f" Count: {count:,}\n" + f" End: {start + count:,}\n" + f"Recording only has {n_samples:,} samples" + ) + + # Handle frequency bounds + freq_lower, freq_upper, freq_default = determine_frequency_bounds( + recording=recording, freq_lower=freq_lower, freq_upper=freq_upper + ) + + # Build comment JSON + comment_data = {"type": annotation_type} + if comment: + comment_data["user_comment"] = comment + + # Create annotation + ann = Annotation( + sample_start=start, + sample_count=count, + freq_lower_edge=freq_lower, + freq_upper_edge=freq_upper, + label=label, + comment=json.dumps(comment_data), + detail={}, + ) + + recording._annotations.append(ann) + + if not quiet: + click.echo("\nAdding annotation:") + click.echo(f" Start: {format_sample_count(start)}") + click.echo(f" Count: {format_sample_count(count)} samples") + freq_str = ( + "full bandwidth" if freq_default else f"{format_frequency(freq_lower)} - {format_frequency(freq_upper)}" + ) + click.echo(f" Frequency: {freq_str}") + click.echo(f" Label: {label}") + click.echo(f" Type: {annotation_type}") + if comment: + click.echo(f" Comment: {comment}") + + try: + save_recording_auto(recording, output, input, quiet, overwrite) + if not quiet: + click.echo(" ✓ Saved") + except Exception as e: + raise click.ClickException(f"Failed to save: {e}") + + +# ============================================================================ +# Remove subcommand +# ============================================================================ + + +@annotate.command(context_settings={"max_content_width": 200}) +@click.argument("input", type=click.Path(exists=True)) +@click.argument("index", type=int) +@click.option("--output", "-o", type=click.Path(), help="Output file path") +@click.option("--overwrite", is_flag=True, help="Overwrite input file (non-SigMF only)") +@click.option("--quiet", is_flag=True, help="Quiet mode") +def remove(input, index, output, overwrite, quiet): + """Remove annotation by index. + + Use 'utils annotate list' to see annotation indices. + + \b + Examples: + utils annotate remove signal.sigmf-data 2 + utils annotate remove file.npy 0 + """ + try: + recording = load_recording(input) + if not quiet: + click.echo(f"Loaded: {input}") + except Exception as e: + raise click.ClickException(f"Failed to load recording: {e}") + + if index < 0 or index >= len(recording.annotations): + raise click.ClickException( + f"Cannot remove annotation at index {index}\n" + f"Recording has {len(recording.annotations)} annotation(s) (indices 0-{len(recording.annotations)-1})" + ) + + removed_ann = recording.annotations[index] + recording._annotations.pop(index) + + if not quiet: + click.echo(f"\nRemoving annotation [{index}]:") + click.echo( + f" Removed: samples {format_sample_count(removed_ann.sample_start)}-" + f"{format_sample_count(removed_ann.sample_start + removed_ann.sample_count)} ({removed_ann.label})" + ) + + try: + save_recording_auto(recording, output_path=input, input_path=input, quiet=quiet, overwrite=True) + if not quiet: + click.echo(" ✓ Saved") + except Exception as e: + raise click.ClickException(f"Failed to save: {e}") + + +# ============================================================================ +# Clear subcommand +# ============================================================================ + + +@annotate.command(context_settings={"max_content_width": 175}) +@click.argument("input", type=click.Path(exists=True)) +@click.option("--output", "-o", type=click.Path(), help="Output file path") +@click.option("--overwrite", is_flag=True, help="Overwrite input file (non-SigMF only)") +@click.option("--force", is_flag=True, help="Skip confirmation") +@click.option("--quiet", is_flag=True, help="Quiet mode") +def clear(input, output, overwrite, force, quiet): + """Clear all annotations. + + \b + Examples: + utils annotate clear signal.sigmf-data + utils annotate clear file.npy --force + """ + try: + recording = load_recording(input) + if not quiet: + click.echo(f"Loaded: {input}") + except Exception as e: + raise click.ClickException(f"Failed to load recording: {e}") + + count_before = len(recording.annotations) + + if count_before == 0: + if not quiet: + click.echo("No annotations to clear") + return + + # Confirm unless --force + if not force and not quiet: + click.echo(f"\nWarning: This will remove all {count_before} annotation(s)") + click.confirm("Continue?", abort=True) + + recording._annotations = [] + + if not quiet: + click.echo(f"\nCleared {count_before} annotation(s)") + + recording._annotations = [] + + try: + save_recording_auto(recording, output_path=input, input_path=input, quiet=quiet, overwrite=True) + if not quiet: + click.echo(" ✓ Saved") + except Exception as e: + raise click.ClickException(f"Failed to save: {e}") + + +# ============================================================================ +# Energy detection subcommand +# ============================================================================ + + +@annotate.command(context_settings={"max_content_width": 200}) +@click.argument("input", type=click.Path(exists=True)) +@click.option("--label", type=str, default="signal", help="Annotation label") +@click.option("--threshold", type=float, default=1.2, help="Threshold multiplier above noise floor") +@click.option("--segments", type=int, default=10, help="Number of segments for noise estimation") +@click.option("--window-size", type=int, default=200, help="Smoothing window size") +@click.option("--min-distance", type=int, default=5000, help="Min distance between detections") +@click.option( + "--freq-method", + type=click.Choice(["nbw", "obw", "full-detected", "full-bandwidth"]), + default="nbw", + help="Frequency bounding method", +) +@click.option("--nfft", type=int, default=None, help="FFT size for frequency calculation") +@click.option("--obw-power", type=float, default=0.99, help="Power percentage for OBW/NBW (0.98-0.9999)") +@click.option( + "--type", + "annotation_type", + type=click.Choice(["standalone", "parallel", "intersection"]), + default="standalone", + help="Annotation type", +) +@click.option("--output", "-o", type=click.Path(), help="Output file path") +@click.option("--overwrite", is_flag=True, help="Overwrite input file (non-SigMF only)") +@click.option("--quiet", is_flag=True, help="Quiet mode") +def energy( + input, + label, + threshold, + segments, + window_size, + min_distance, + freq_method, + nfft, + obw_power, + annotation_type, + output, + overwrite, + quiet, +): + """Auto-detect signals using energy-based method. + + Detects bursts based on energy above noise floor. Best for bursty signals + and intermittent transmissions. + + \b + Frequency Bounding Methods: + nbw - Nominal bandwidth (default, best for real signals) + obw - Occupied bandwidth (more conservative, includes sidelobes) + full-detected - Lowest to highest spectral component + full-bandwidth - Entire Nyquist span + + \b + Examples: + utils annotate energy capture.sigmf-data --label burst + utils annotate energy signal.npy --threshold 1.5 --min-distance 10000 + utils annotate energy signal.sigmf-data --freq-method obw + utils annotate energy signal.sigmf-data --freq-method full-detected + + """ + try: + recording = load_recording(input) + if not quiet: + click.echo(f"Loaded: {input}") + except Exception as e: + raise click.ClickException(f"Failed to load recording: {e}") + + if not quiet: + click.echo("\nDetecting signals using energy-based method...") + click.echo(" Time detection:") + click.echo(f" Segments: {segments}") + click.echo(f" Threshold: {threshold}x noise floor") + click.echo(f" Window size: {window_size} samples") + click.echo(f" Min distance: {min_distance} samples") + click.echo(f" Frequency bounds: {freq_method}") + + try: + initial_count = len(recording.annotations) + recording = detect_signals_energy( + recording, + k=segments, + threshold_factor=threshold, + window_size=window_size, + min_distance=min_distance, + label=label, + annotation_type=annotation_type, + freq_method=freq_method, + nfft=nfft, + obw_power=obw_power, + ) + added = len(recording.annotations) - initial_count + + if not quiet: + click.echo(f" ✓ Added {added} annotation(s)") + + save_recording_auto(recording, output, input, quiet, overwrite) + if not quiet: + click.echo(" ✓ Saved") + except Exception as e: + raise click.ClickException(f"Energy detection failed: {e}") + + +# ============================================================================ +# CUSUM detection subcommand +# ============================================================================ + + +@annotate.command() +@click.argument("input", type=click.Path(exists=True)) +@click.option("--label", type=str, default="segment", help="Annotation label") +@click.option("--min-duration", type=float, default=5.0, help="Min duration in ms (prevents over-segmentation)") +@click.option("--window-size", type=int, default=1, help="Smoothing window size") +@click.option("--tolerance", type=int, default=-1, help="Sample tolerance for merging") +@click.option( + "--type", + "annotation_type", + type=click.Choice(["standalone", "parallel", "intersection"]), + default="standalone", + help="Annotation type", +) +@click.option("--output", "-o", type=click.Path(), help="Output file path") +@click.option("--overwrite", is_flag=True, help="Overwrite input file (non-SigMF only)") +@click.option("--quiet", is_flag=True, help="Quiet mode") +def cusum(input, label, min_duration, window_size, tolerance, annotation_type, output, overwrite, quiet): + """Auto-detect segments using CUSUM method. + + Detects signal state changes (on/off, amplitude transitions). Best for + segmenting continuous signals. + + IMPORTANT: Always specify --min-duration to prevent excessive segmentation. + + \b + Examples: + utils annotate cusum signal.sigmf-data --min-duration 5.0 + utils annotate cusum data.npy --min-duration 10.0 --label state + """ + try: + recording = load_recording(input) + if not quiet: + click.echo(f"Loaded: {input}") + except Exception as e: + raise click.ClickException(f"Failed to load recording: {e}") + + if not quiet: + click.echo("\nDetecting segments using CUSUM...") + click.echo(f" Min duration: {min_duration} ms") + if window_size != 1: + click.echo(f" Window size: {window_size} samples") + + try: + initial_count = len(recording.annotations) + recording = annotate_with_cusum( + recording, + label=label, + window_size=window_size, + min_duration=min_duration, + tolerance=tolerance, + annotation_type=annotation_type, + ) + added = len(recording.annotations) - initial_count + + if not quiet: + click.echo(f" ✓ Added {added} annotation(s)") + + save_recording_auto(recording, output, input, quiet, overwrite) + if not quiet: + click.echo(" ✓ Saved") + except Exception as e: + raise click.ClickException(f"CUSUM detection failed: {e}") + + +# ============================================================================ +# Threshold detection subcommand +# ============================================================================ + + +@annotate.command() +@click.argument("input", type=click.Path(exists=True)) +@click.option("--threshold", type=float, required=True, help="Threshold (0.0-1.0, fraction of max magnitude)") +@click.option("--label", type=str, default=None, help="Annotation label") +@click.option("--window-size", type=int, default=1024, help="Smoothing window size") +@click.option( + "--type", + "annotation_type", + type=click.Choice(["standalone", "parallel", "intersection"]), + default="standalone", + help="Annotation type", +) +@click.option("--output", "-o", type=click.Path(), help="Output file path") +@click.option("--overwrite", is_flag=True, help="Overwrite input file (non-SigMF only)") +@click.option("--quiet", is_flag=True, help="Quiet mode") +def threshold(input, threshold, label, window_size, annotation_type, output, overwrite, quiet): + """Auto-detect signals using threshold method. + + Detects samples above a percentage of maximum magnitude. Best for simple + power-based detection. + + \b + Examples: + utils annotate threshold signal.sigmf-data --threshold 0.7 --label wifi + utils annotate threshold data.npy --threshold 0.5 --window-size 2048 + """ + if not (0.0 <= threshold <= 1.0): + raise click.ClickException(f"--threshold must be between 0.0 and 1.0, got {threshold}") + + try: + recording = load_recording(input) + if not quiet: + click.echo(f"Loaded: {input}") + except Exception as e: + raise click.ClickException(f"Failed to load recording: {e}") + + if not quiet: + click.echo("\nDetecting signals using threshold qualifier...") + click.echo(f" Threshold: {threshold * 100:.1f}% of max magnitude") + click.echo(f" Window size: {window_size} samples") + + try: + initial_count = len(recording.annotations) + recording = threshold_qualifier( + recording, + threshold=threshold, + window_size=window_size, + label=label, + annotation_type=annotation_type, + ) + added = len(recording.annotations) - initial_count + + if not quiet: + click.echo(f" ✓ Added {added} annotation(s)") + + save_recording_auto(recording, output, input, quiet, overwrite) + if not quiet: + click.echo(" ✓ Saved") + except Exception as e: + raise click.ClickException(f"Threshold detection failed: {e}") + + +# ============================================================================ +# Separate subcommand (Phase 2: Parallel signal separation) +# ============================================================================ + + +@annotate.command() +@click.argument("input", type=click.Path(exists=True)) +@click.option("--indices", type=str, help="Comma-separated annotation indices to split (default: all)") +@click.option("--nfft", type=int, default=65536, help="FFT size for spectral analysis") +@click.option("--noise-threshold-db", type=float, help="Noise floor threshold in dB (auto-estimated if not specified)") +@click.option("--min-component-bw", type=float, default=50e3, help="Min component bandwidth in Hz") +@click.option("--output", "-o", type=click.Path(), help="Output file path") +@click.option("--overwrite", is_flag=True, help="Overwrite input file (non-SigMF only)") +@click.option("--quiet", is_flag=True, help="Quiet mode") +@click.option("--verbose", is_flag=True, help="Verbose output (show detected components)") +def separate(input, indices, nfft, noise_threshold_db, min_component_bw, output, overwrite, quiet, verbose): + """ + Auto-detect parallel frequency-offset signals and split into sub-bands. + + Provides methods to detect and separate overlapping frequency-domain signals + that occupy the same time window but different frequency bands. + + Detects multiple frequency components within single annotations and splits + them into separate annotations. Uses spectral peak detection with dual + bandwidth estimation. + + \b + Key Features: + - Spectral peak detection for frequency components + - Auto noise floor estimation (or user-specified) + - Dual bandwidth estimation: -3dB primary, cumulative power fallback + - Handles narrowband and wide signals (OFDM) + + \b + Examples: + utils annotate separate capture.sigmf-data + utils annotate separate signal.npy --indices 0,1,2 + utils annotate separate data.sigmf-data --noise-threshold-db -70 + utils annotate separate signal.npy --min-component-bw 100000 + + """ + try: + recording = load_recording(input) + if not quiet: + click.echo(f"Loaded: {input}") + except Exception as e: + raise click.ClickException(f"Failed to load recording: {e}") + + # Parse indices if specified + indices_list = get_indices_list(indices=indices, recording=recording) + + if len(recording.annotations) == 0: + if not quiet: + click.echo("No annotations to split") + return + + if not quiet: + click.echo("\nSplitting annotations by frequency components...") + click.echo(f" Input annotations: {len(recording.annotations)}") + if indices_list: + click.echo(f" Splitting indices: {indices_list}") + click.echo(f" FFT size: {nfft}") + if noise_threshold_db is not None: + click.echo(f" Noise threshold: {noise_threshold_db} dB") + else: + click.echo(" Noise threshold: auto-estimated") + click.echo(f" Min component BW: {format_frequency(min_component_bw)}") + + try: + initial_count = len(recording.annotations) + + recording = split_recording_annotations( + recording, + indices=indices_list, + nfft=nfft, + noise_threshold_db=noise_threshold_db, + min_component_bw=min_component_bw, + ) + + final_count = len(recording.annotations) + added = final_count - initial_count + + if not quiet: + click.echo(f" ✓ Output annotations: {final_count} ({'+' if added >= 0 else ''}{added} change)") + if verbose and added > 0: + click.echo("\n Details:") + for i in range(initial_count, final_count): + ann = recording.annotations[i] + freq_range = f"{format_frequency(ann.freq_lower_edge)} - {format_frequency(ann.freq_upper_edge)}" + click.echo( + f" [{i}] samples {format_sample_count(ann.sample_start)}-" + f"{format_sample_count(ann.sample_start + ann.sample_count)}: {freq_range}" + ) + + save_recording_auto(recording, output, input, quiet, overwrite) + if not quiet: + click.echo(" ✓ Saved") + except Exception as e: + raise click.ClickException(f"Spectral separation failed: {e}")