Source code for melody_features.pipeline.processing

"""Per-melody and parallel processing helpers."""

from __future__ import annotations

import logging
import time
from typing import Dict, List, Optional, Tuple

from tqdm import tqdm

from ..core.representations import Melody
from ..utils.warnings import suppress_common_melody_warnings
from .config import Config
from .timing import TIMING_STAT_CATEGORIES, _init_timing_stats


def _features_module():
    from melody_features import features as features_module

    return features_module


def _melody_id_to_num_map(melody_data_list: List[dict]) -> Dict[str, object]:
    """Map melody ID strings to melody_num for IDyOM and export lookups."""
    return {str(m["ID"]): m.get("melody_num") for m in melody_data_list}


def _build_feature_row(
    melody_num: object,
    melody_id: object,
    melody_features: Dict,
    headers: List[str],
    idyom_results_dict: Dict[str, dict],
) -> List:
    """Build one CSV row from computed features and IDyOM batch results."""
    row = [melody_num, melody_id]
    for header in headers[2:]:
        if header.startswith("idyom_"):
            prefix, feature_name = header.split(".", 1)
            idyom_name = prefix[len("idyom_") : -len("_features")]
            value = (
                idyom_results_dict.get(idyom_name, {})
                .get(str(melody_num), {})
                .get(feature_name, 0.0)
            )
            row.append(value)
        else:
            category, feature_name = header.split(".", 1)
            value = melody_features.get(category, {}).get(feature_name, 0.0)
            row.append(value)
    return row


def _record_timing_result(
    result: Tuple,
    headers: List[str],
    id_to_num: Dict[str, object],
    idyom_results_dict: Dict[str, dict],
    all_features: List[List],
    timing_stats: Dict[str, List[float]],
) -> None:
    """Append one melody result row and timing samples."""
    melody_id, melody_features, timings = result
    melody_num = id_to_num.get(str(melody_id))
    row = _build_feature_row(
        melody_num, melody_id, melody_features, headers, idyom_results_dict
    )
    all_features.append(row)
    for category, duration in timings.items():
        timing_stats[category].append(duration)


[docs] def process_melody(args): """Process a single melody and return its features. Parameters ---------- args : tuple Tuple containing (melody_data, corpus_stats, idyom_features, phrase_gap, max_ngram_order, key_estimation) Returns ------- tuple Tuple containing (melody_id, feature_dict, timings) """ suppress_common_melody_warnings() features_module = _features_module() start_total = time.time() melody_data, corpus_stats, _idyom_results_dict, phrase_gap, max_ngram_order, key_estimation = args mel = Melody(melody_data) timings: Dict[str, float] = {category: 0.0 for category in TIMING_STAT_CATEGORIES} start = time.time() pitch_features = features_module.get_pitch_features(mel) timings["absolute_pitch"] = time.time() - start start = time.time() pitch_class_features = features_module.get_pitch_class_features(mel) timings["pitch_class"] = time.time() - start start = time.time() interval_features = features_module.get_interval_features(mel) timings["pitch_interval"] = time.time() - start start = time.time() contour_features = features_module.get_contour_features(mel) timings["contour"] = time.time() - start rhythm_features, rhythm_timings = features_module.collect_rhythm_for_pipeline(mel) timings["timing"] = rhythm_timings["timing"] timings["inter_onset_interval"] = rhythm_timings["inter_onset_interval"] start = time.time() tonality_features = features_module.get_tonality_features(mel, key_estimation=key_estimation) timings["tonality"] = time.time() - start start = time.time() metre_features = features_module.get_metre_features(mel) timings["metre"] = time.time() - start start = time.time() expectation_features = features_module.get_expectation_features(mel) timings["expectation"] = time.time() - start start = time.time() complexity_features = features_module.get_complexity_features( mel, phrase_gap=phrase_gap, max_ngram_order=max_ngram_order ) timings["complexity"] = time.time() - start start = time.time() lexical_diversity_features = features_module.get_lexical_diversity_features( mel, phrase_gap=phrase_gap, max_ngram_order=max_ngram_order ) timings["lexical_diversity"] = time.time() - start melody_features = { "pitch_features": pitch_features, "pitch_class_features": pitch_class_features, "interval_features": interval_features, "contour_features": contour_features, "rhythm_features": rhythm_features, "tonality_features": tonality_features, "metre_features": metre_features, "expectation_features": expectation_features, "complexity_features": {**complexity_features, **lexical_diversity_features}, } if corpus_stats: start = time.time() melody_features["corpus_features"] = features_module.get_corpus_features( mel, corpus_stats, phrase_gap=phrase_gap, max_ngram_order=max_ngram_order ) timings["corpus"] = time.time() - start timings["total"] = time.time() - start_total return melody_data["ID"], melody_features, timings
def _setup_parallel_processing( melody_data_list: List[dict], corpus_stats: Optional[dict], idyom_results_dict: Dict[str, dict], config: Config, ) -> Tuple[List[str], List, Dict[str, List[float]]]: """Set up parallel processing arguments and headers.""" suppress_common_melody_warnings() features_module = _features_module() from multiprocessing import cpu_count mel = Melody(melody_data_list[0]) first_features = { "pitch_features": features_module.get_pitch_features(mel), "pitch_class_features": features_module.get_pitch_class_features(mel), "interval_features": features_module.get_interval_features(mel), "contour_features": features_module.get_contour_features(mel), "rhythm_features": features_module.get_rhythm_features(mel), "tonality_features": features_module.get_tonality_features( mel, key_estimation=config.key_estimation ), "metre_features": features_module.get_metre_features(mel), "expectation_features": features_module.get_expectation_features(mel), "complexity_features": features_module.get_complexity_feature_bundle( mel, phrase_gap=config.fantastic.phrase_gap, max_ngram_order=config.fantastic.max_ngram_order, ), } if corpus_stats: first_features["corpus_features"] = features_module.get_corpus_features( mel, corpus_stats, phrase_gap=config.fantastic.phrase_gap, max_ngram_order=config.fantastic.max_ngram_order, ) for idyom_name, idyom_results in idyom_results_dict.items(): if idyom_results: sample_id = next(iter(idyom_results)) for feature in idyom_results[sample_id].keys(): first_features[f"idyom_{idyom_name}_features.{feature}"] = None else: first_features[f"idyom_{idyom_name}_features.mean_information_content"] = None headers = ["melody_num", "melody_id"] for category, features in first_features.items(): if isinstance(features, dict): headers.extend(f"{category}.{feature}" for feature in features.keys()) elif features is None: headers.append(category) logger = logging.getLogger("melody_features") logger.info("Starting parallel processing...") n_cores = cpu_count() logger.info(f"Using {n_cores} CPU cores") melody_args = [ ( melody_data, corpus_stats, idyom_results_dict, config.fantastic.phrase_gap, config.fantastic.max_ngram_order, config.key_estimation, ) for melody_data in melody_data_list ] timing_stats = _init_timing_stats() return headers, melody_args, timing_stats def _process_melodies_parallel( melody_args: List, headers: List[str], melody_data_list: List[dict], idyom_results_dict: Dict[str, dict], timing_stats: Dict[str, List[float]], ) -> List[List]: """Process melodies in parallel and collect results.""" all_features = [] id_to_num = _melody_id_to_num_map(melody_data_list) n_melodies = len(melody_args) tqdm_kwargs = { "total": n_melodies, "unit": "melody", "dynamic_ncols": True, "mininterval": 0.15, "maxinterval": 1.0, "miniters": max(1, n_melodies // 200), "smoothing": 0.45, "bar_format": "{l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}, {rate_fmt}]", } logger = logging.getLogger("melody_features") try: from multiprocessing import Pool, cpu_count import multiprocessing as mp try: mp.set_start_method("fork", force=True) except RuntimeError: pass logger.info("Parallel processing initiated") n_cores = cpu_count() chunk_size = max(1, n_melodies // (n_cores * 16)) with Pool(n_cores) as pool: results_iter = pool.imap(process_melody, melody_args, chunksize=chunk_size) for result in tqdm(results_iter, desc="Processing melodies", **tqdm_kwargs): try: _record_timing_result( result, headers, id_to_num, idyom_results_dict, all_features, timing_stats, ) except Exception as e: logger.error("Error processing melody: %s", e) except Exception as e: logger.warning( "Parallel processing failed (%s), falling back to sequential processing", e, ) for i, args in tqdm( enumerate(melody_args), desc="Processing melodies (sequential)", **tqdm_kwargs, ): try: result = process_melody(args) _record_timing_result( result, headers, id_to_num, idyom_results_dict, all_features, timing_stats, ) except Exception as e: logger.error("Error processing melody %s: %s", i, e) return all_features