Signal processing

class song_phenotyping.signal.ProcessingResult(specs, waveforms, spec_times, valid_indices, tempos=None, phase_features=None)[source]

Bases: object

Standardized return type for processing operations.

Parameters:
phase_features: List[dict] | None = None
spec_times: List[ndarray]
specs: List[ndarray]
tempos: Dict[str, Any] | None = None
valid_indices: List[int]
waveforms: List[ndarray]
song_phenotyping.signal.add_tempo_to_h5(h5_path, min_freq=500.0, max_freq=8000.0, overwrite=False)[source]

Compute and append tempo fields to an existing Stage-A HDF5 file in place.

Reads the stored audio_filename, reloads and preprocesses the audio identically to get_song_specs, then writes the 8 tempo_* arrays. The rest of the file is untouched.

Parameters:

  • h5_path (str) – Path to a Stage-A syllable HDF5 file.

  • min_freq (float) – Bandpass filter cutoffs — should match the SpectrogramParams used when the file was originally generated.

  • max_freq (float) – Bandpass filter cutoffs — should match the SpectrogramParams used when the file was originally generated.

  • overwrite (bool) – If False (default), skip files that already contain tempo data.

Returns:

‘written’, ‘skipped’ (already present, overwrite=False), or ‘failed’.

Return type:

str

song_phenotyping.signal.backfill_tempo_for_bird(bird_root, run_name, overwrite=False)[source]

Add tempo fields to all Stage-A HDF5 files for a bird that don’t have them yet. Reads min_freq and max_freq from the run’s run_config.json automatically.

Parameters:

  • bird_root (str) – Bird output directory, e.g. /data/pipeline_runs/bk1bk3.

  • run_name (str) – Run name / hash, e.g. 'd5dfde49'.

  • overwrite (bool) – Re-compute and overwrite tempo fields even if already present.

Return type:

dict with keys ‘written’, ‘skipped’, ‘failed’ and integer counts.

song_phenotyping.signal.backfill_tempo_for_run(save_path, run_name, overwrite=False, birds=None)[source]

Add tempo fields to all Stage-A HDF5 files across every bird in an existing run.

Discovers birds by looking for <save_path>/<bird>/<run_name>/stages/01_specs/ directories that contain at least one HDF5 file. Reads min_freq and max_freq from each bird’s run_config.json automatically.

Parameters:

  • save_path (str) – Root output directory passed to the pipeline (cfg.pipeline.save_path).

  • run_name (str) – Run name / hash shared by all birds, e.g. 'd5dfde49'.

  • overwrite (bool) – Re-compute and overwrite tempo fields even if already present.

  • birds (list of str, optional) – Restrict to these bird IDs. None processes every bird found.

Returns:

'total' counts across all birds plus per-bird breakdown under 'per_bird'.

Return type:

dict

song_phenotyping.signal.create_output_paths(save_path, bird, run_name='default')[source]

Create and return standardized output directory paths.

Parameters:
Return type:

Dict[str, str]

song_phenotyping.signal.define_slice_on_off(onsets, offsets, time_bin_size)[source]

Create an array of time bin onsets and corresponding offsets

Parameters:
song_phenotyping.signal.downsample_spec(spec, params)[source]

Downsample spectrogram time axis to target shape.

Parameters:

spec (ndarray)

Return type:

ndarray

song_phenotyping.signal.extract_compact_phase_features(complex_spec_ds)[source]

Extract minimal phase features from already-downsampled complex spectrogram.

song_phenotyping.signal.extract_minimal_phase_feature(complex_spec_ds)[source]

Extract single most informative phase feature.

song_phenotyping.signal.extract_phase_features(Sx, params)[source]

Extract compact phase-derived features.

song_phenotyping.signal.generate_syllable_hashes(base_identifier, indices)[source]

Generate unique hash IDs for each syllable. Consolidated from A_spec_saving.py

Parameters:
Return type:

List[str]

song_phenotyping.signal.get_memory_usage()[source]

Get current memory usage in MB

song_phenotyping.signal.get_song_spec(t1, t2, audio, params, fs=32000, fill_value=-999999999.9999999, downsample=True)[source]

Extract and normalize spectrogram from a birdsong syllable segment. Updated with better logging and error handling.

Parameters:
Return type:

Tuple[ndarray, ndarray, ndarray]

song_phenotyping.signal.get_song_specs(audio_filename, onsets, offsets, params, split_syllables=False, tempo=False)[source]

Extract spectrograms from audio file for given onset/offset pairs. Updated to return ProcessingResult and use consolidated utilities.

Parameters:
Return type:

ProcessingResult

song_phenotyping.signal.label_slices(onsets, offsets, labels, time_bin_size)[source]

Label time slices based on syllable overlap. Keeping original implementation for compatibility.

Parameters:
Return type:

Tuple[ndarray, ndarray, ndarray, Dict[int, str]]

song_phenotyping.signal.load_and_validate_metadata(metadata_file_path, wseg_offset=0.0)[source]

Load metadata from .mat file and validate required fields. Consolidated version from A_spec_saving.py

Parameters:

  • metadata_file_path (str)

  • wseg_offset (float)

Return type:

Dict[str, Any]

song_phenotyping.signal.pad_waveforms_to_same_length(waveforms, time_arrays=None, pad_value=nan)[source]

Pad all waveforms (and optionally time arrays) to the same length. Consolidated from A_spec_saving.py

Parameters:
Return type:

Tuple[List[ndarray], List[ndarray] | None]

song_phenotyping.signal.parse_audio_filename(file_path)[source]

Parse audio filename to extract bird, date, and time components. Handles multiple filename formats including wseg patterns.

Parameters:

file_path (str)

Return type:

Dict[str, Any]

song_phenotyping.signal.read_metadata(metadata_file_path, read_songpath_from_metadata)[source]

Read metadata and return song path, fs, onsets, offsets, labels. Updated to use consolidated utilities.

Parameters:

  • metadata_file_path (str)

  • read_songpath_from_metadata (bool)

Return type:

Tuple[str, float, ndarray, ndarray, ndarray]

song_phenotyping.signal.rms_norm(array)[source]

RMS normalization with safe dtype handling

song_phenotyping.signal.save_segmented_audio_data(h5file_save_path, song_file_path, segmented_audio_data, save_manual=True)[source]

Save segmented audio data to HDF5 file. Updated with better logging and validation.

Parameters:

save_manual (bool)

song_phenotyping.signal.save_spec_slices(candidate_files, save_path, params, slice_length=None, read_songpath_from_metadata=True, verbose=False, prefer_local=True)[source]

Process fixed-length slice spectrograms from metadata files. Updated to handle file pairs and prefer_local parameter.

Parameters:
Return type:

None

song_phenotyping.signal.setup_logging(log_dir='logs')[source]

Setup centralized logging for song processing.

Parameters:

log_dir (str)

Return type:

Logger

song_phenotyping.signal.split_long_syllables_with_mapping(onsets, offsets, max_duration_ms)[source]

Split syllables longer than max_duration and create mapping to original indices. Consolidated from A_spec_saving.py

Parameters:
Return type:

Tuple[ndarray, ndarray, ndarray]

song_phenotyping.signal.tempo_estimates(audio_norm, fs, n_high=3, debug_dir=None, debug_label='')[source]

Estimate tempo from the power spectrum of the normalized autocorrelogram.

Returns a dict with:

high_band_freqs : (n_high,) top inter-syllable peak frequencies (>LOW_FREQ_THRESHOLD Hz), nan-padded high_band_powers : (n_high,) corresponding relative powers low_band_freq : scalar dominant inter-motif peak frequency (<LOW_FREQ_THRESHOLD Hz), nan if absent low_band_power : scalar relative power of dominant low-band peak, nan if absent filtered_freqs : full frequency axis (for debug) filtered_power : full normalized power spectrum (for debug) peak_freqs : all detected peak frequencies (for debug) peak_powers : all detected peak relative powers (for debug)

Parameters:
Return type:

Dict[str, Any]

song_phenotyping.signal.verify_save(filepath)[source]

Verify HDF5 file was saved correctly