Spaces:
Running on Zero
Running on Zero
| from __future__ import annotations | |
| import copy | |
| import bisect | |
| import io | |
| import json | |
| import math | |
| import re | |
| import tempfile | |
| import unicodedata | |
| import zipfile | |
| import xml.etree.ElementTree as ElementTree | |
| from dataclasses import dataclass | |
| from fractions import Fraction | |
| from pathlib import Path | |
| from statistics import median | |
| from typing import Any | |
| QUANTIZATION_DIVISORS = { | |
| "1/8": 2, | |
| "1/16": 4, | |
| "1/32": 8, | |
| } | |
| # music21 accepts several candidate grids at once and selects the closest one | |
| # note by note. Keeping a ternary divisor beside each binary resolution is a | |
| # much better fit for performances containing triplets or compound meter than | |
| # forcing every onset onto a single straight grid. | |
| QUANTIZATION_GRIDS = { | |
| "1/8": (2, 3), | |
| "1/16": (4, 3, 6), | |
| "1/32": (8, 6, 12), | |
| } | |
| CLEANUP_PROFILES = { | |
| "faithful": { | |
| "label": "Faithful", | |
| "merge_gap_beats": 0.025, | |
| "minimum_note_beats": 0.035, | |
| "drop_ghost_beats": 0.0, | |
| }, | |
| "balanced": { | |
| "label": "Balanced", | |
| "merge_gap_beats": 0.07, | |
| "minimum_note_beats": 0.09, | |
| "drop_ghost_beats": 0.0, | |
| }, | |
| "readable": { | |
| "label": "Readable", | |
| "merge_gap_beats": 0.12, | |
| "minimum_note_beats": 0.14, | |
| "drop_ghost_beats": 0.045, | |
| }, | |
| } | |
| AMBIGUOUS_INSTRUMENT_GROUPS = { | |
| "brass_section", | |
| "chromatic_percussion", | |
| "flutes", | |
| "orchestra_hit", | |
| "soprano_and_alto_sax", | |
| "string_ensemble", | |
| "strings", | |
| "synth_lead", | |
| "synth_pad", | |
| "synth_strings", | |
| "voice", | |
| } | |
| KEY_PITCH_NAMES = { | |
| "major": ("C", "D♭", "D", "E♭", "E", "F", "F♯", "G", "A♭", "A", "B♭", "B"), | |
| "minor": ("C", "C♯", "D", "E♭", "E", "F", "F♯", "G", "G♯", "A", "B♭", "B"), | |
| } | |
| SOLFEGE_NAMES = ("Do", "Do♯", "Ré", "Mi♭", "Mi", "Fa", "Fa♯", "Sol", "La♭", "La", "Si♭", "Si") | |
| class NotationPartResult: | |
| track_id: str | |
| track_key: str | |
| name: str | |
| slug: str | |
| musicxml: Path | |
| pdf: Path | None | |
| svg_pages: tuple[Path, ...] | |
| render_error: str | |
| class NotationResult: | |
| musicxml: Path | |
| pdf: Path | None | |
| bundle: Path | |
| svg_pages: tuple[Path, ...] | |
| preview_svg: str | |
| parts: tuple[NotationPartResult, ...] | |
| warnings: tuple[str, ...] | |
| export_stem: str | |
| report: Path | |
| diagnostics: dict[str, Any] | |
| def _default_score_analysis(onset_count: int = 0) -> dict[str, Any]: | |
| return { | |
| "analysis_version": 3, | |
| "tempo_bpm": 120.0, | |
| "time_signature": "4/4", | |
| "quantization": "1/16", | |
| "key_signature": "C major", | |
| "key_confidence": 0.0, | |
| "confidence": "low", | |
| "onset_count": int(onset_count), | |
| "tempo_confidence": 0.0, | |
| "quantization_fit": 0.0, | |
| "beat_phase_seconds": 0.0, | |
| "first_downbeat_seconds": 0.0, | |
| "pickup_beats": 0.0, | |
| "beat_quarter_length": 1.0, | |
| "beat_times_seconds": [], | |
| "beat_origin_index": 0, | |
| "grid_family": "binary + ternary", | |
| "analysis_source": "symbolic notes", | |
| "review_flags": [ | |
| "Tempo, meter and key need manual confirmation.", | |
| "Dynamics are not predicted by MuScriptor and are intentionally left neutral.", | |
| ], | |
| } | |
| def _estimate_key_signature( | |
| tracks: list[dict[str, Any]], | |
| np: Any, | |
| ) -> tuple[str, float]: | |
| """Estimate a concert key with Krumhansl-Schmuckler profiles. | |
| This is deliberately a suggestion, not a promise: chromatic music and | |
| short clips can make relative major/minor keys mathematically ambiguous. | |
| """ | |
| histogram = np.zeros(12, dtype=float) | |
| for track in tracks: | |
| if track.get("is_drum"): | |
| continue | |
| for item in track.get("notes") or []: | |
| pitch = max(0, min(127, int(item.get("pitch", 60)))) | |
| start = max(0.0, float(item.get("start", 0.0))) | |
| end = max(start + 0.015, float(item.get("end", start + 0.1))) | |
| # Long pedal tails should not dominate the harmonic analysis. | |
| histogram[pitch % 12] += min(2.0, end - start) ** 0.72 | |
| if float(histogram.sum()) <= 0: | |
| return "C major", 0.0 | |
| major_profile = np.asarray( | |
| [6.35, 2.23, 3.48, 2.33, 4.38, 4.09, 2.52, 5.19, 2.39, 3.66, 2.29, 2.88], | |
| dtype=float, | |
| ) | |
| minor_profile = np.asarray( | |
| [6.33, 2.68, 3.52, 5.38, 2.60, 3.53, 2.54, 4.75, 3.98, 2.69, 3.34, 3.17], | |
| dtype=float, | |
| ) | |
| scores: list[tuple[float, int, str]] = [] | |
| for root in range(12): | |
| for mode, profile in (("major", major_profile), ("minor", minor_profile)): | |
| shifted = np.roll(profile, root) | |
| if float(histogram.std()) <= 1e-9: | |
| score = 0.0 | |
| else: | |
| score = float(np.corrcoef(histogram, shifted)[0, 1]) | |
| scores.append((score, root, mode)) | |
| scores.sort(reverse=True) | |
| best, second = scores[0], scores[1] | |
| separation = max(0.0, min(1.0, (best[0] - second[0]) / 0.24)) | |
| return f"{KEY_PITCH_NAMES[best[2]][best[1]]} {best[2]}", round(separation, 3) | |
| def estimate_score_settings(tracks: list[dict[str, Any]]) -> dict[str, Any]: | |
| """Estimate notation presets from the decoded note onsets. | |
| The estimate intentionally runs after transcription: instrument-separated | |
| note onsets are substantially cleaner than a broadband audio onset curve, | |
| and this keeps the Space free from another heavyweight audio model. | |
| """ | |
| try: | |
| import numpy as np | |
| except ImportError: # pragma: no cover - numpy is an app dependency | |
| return _default_score_analysis() | |
| events: list[tuple[float, float, bool]] = [] | |
| durations: list[float] = [] | |
| for track in tracks: | |
| is_drum = bool(track.get("is_drum")) | |
| for item in track.get("notes") or []: | |
| start = max(0.0, float(item.get("start", 0.0))) | |
| end = max(start + 0.015, float(item.get("end", start + 0.1))) | |
| velocity = max(1.0, min(127.0, float(item.get("velocity", 90.0)))) / 127.0 | |
| events.append((start, velocity * (1.35 if is_drum else 1.0), is_drum)) | |
| durations.append(end - start) | |
| if len(events) < 4: | |
| analysis = _default_score_analysis(len(events)) | |
| key_signature, key_confidence = _estimate_key_signature(tracks, np) | |
| analysis.update( | |
| { | |
| "key_signature": key_signature, | |
| "key_confidence": key_confidence, | |
| } | |
| ) | |
| return analysis | |
| # Collapse chords and near-simultaneous detections into one accented onset. | |
| events.sort(key=lambda event: event[0]) | |
| clusters: list[list[tuple[float, float, bool]]] = [] | |
| for event in events: | |
| if not clusters or event[0] - clusters[-1][-1][0] > 0.035: | |
| clusters.append([event]) | |
| else: | |
| clusters[-1].append(event) | |
| onsets = np.asarray( | |
| [sum(event[0] * event[1] for event in cluster) / sum(event[1] for event in cluster) for cluster in clusters], | |
| dtype=float, | |
| ) | |
| accents = np.asarray( | |
| [max(event[1] for event in cluster) + 0.18 * sum(event[1] for event in cluster[1:]) for cluster in clusters], | |
| dtype=float, | |
| ) | |
| bpm_axis = np.arange(45.0, 221.0, 0.5) | |
| tempo_scores = np.zeros_like(bpm_axis) | |
| beat_hypotheses = ((0.25, 0.22), (0.5, 0.82), (1.0, 1.0), (1.5, 0.62), (2.0, 0.7), (3.0, 0.42), (4.0, 0.3)) | |
| for left in range(len(onsets) - 1): | |
| for right in range(left + 1, min(len(onsets), left + 9)): | |
| interval = onsets[right] - onsets[left] | |
| if interval > 4.5: | |
| break | |
| if interval < 0.08: | |
| continue | |
| pair_weight = math.sqrt(accents[left] * accents[right]) / math.sqrt(right - left) | |
| for beats, hypothesis_weight in beat_hypotheses: | |
| bpm = 60.0 * beats / interval | |
| if bpm_axis[0] <= bpm <= bpm_axis[-1]: | |
| distance = (bpm_axis - bpm) / 1.15 | |
| tempo_scores += pair_weight * hypothesis_weight * np.exp(-0.5 * distance * distance) | |
| tempo_prior = 0.72 + 0.28 * np.exp(-0.5 * ((bpm_axis - 112.0) / 52.0) ** 2) | |
| tempo_scores *= tempo_prior | |
| best_index = int(np.argmax(tempo_scores)) | |
| best_bpm = float(bpm_axis[best_index]) | |
| onset_span = max(1e-6, float(onsets[-1] - onsets[0])) | |
| onset_rate_hz = len(onsets) / onset_span | |
| tempo_octave_adjusted = False | |
| if best_bpm < 74 and best_bpm * 2 <= bpm_axis[-1]: | |
| double_index = int(np.argmin(np.abs(bpm_axis - best_bpm * 2))) | |
| # Dense arrangements are very often detected at half-time. Prefer the | |
| # musician-friendly octave when the double-tempo candidate has useful | |
| # evidence, while retaining genuinely sparse adagios at their slow BPM. | |
| double_threshold = 0.58 if onset_rate_hz >= 1.35 else 0.74 | |
| if tempo_scores[double_index] >= tempo_scores[best_index] * double_threshold: | |
| best_index = double_index | |
| best_bpm = float(bpm_axis[best_index]) | |
| tempo_octave_adjusted = True | |
| if best_bpm > 178: | |
| half_index = int(np.argmin(np.abs(bpm_axis - best_bpm / 2))) | |
| if tempo_scores[half_index] >= tempo_scores[best_index] * 0.88: | |
| best_index = half_index | |
| best_bpm = float(bpm_axis[best_index]) | |
| neighborhood = np.abs(bpm_axis - best_bpm) <= 2.0 | |
| if float(tempo_scores[neighborhood].sum()) > 0: | |
| best_bpm = float(np.average(bpm_axis[neighborhood], weights=tempo_scores[neighborhood])) | |
| beat_seconds = 60.0 / max(1.0, best_bpm) | |
| strong_threshold = float(np.quantile(accents, 0.55)) | |
| strong = accents >= strong_threshold | |
| phase_candidates = np.linspace(0.0, beat_seconds, 96, endpoint=False) | |
| phase_scores = [] | |
| for phase in phase_candidates: | |
| distance = np.abs(((onsets[strong] - phase + beat_seconds / 2) % beat_seconds) - beat_seconds / 2) | |
| phase_scores.append(float(np.sum(accents[strong] * np.exp(-0.5 * (distance / (beat_seconds * 0.11)) ** 2)))) | |
| beat_phase = float(phase_candidates[int(np.argmax(phase_scores))]) | |
| beat_indices = np.rint((onsets - beat_phase) / beat_seconds).astype(int) | |
| minimum_beat = int(beat_indices.min()) | |
| beat_indices -= minimum_beat | |
| beat_energy = np.zeros(int(beat_indices.max()) + 1, dtype=float) | |
| for index, accent in zip(beat_indices, accents): | |
| beat_energy[index] += accent | |
| meter_scores: dict[int, float] = {} | |
| meter_phases: dict[int, int] = {} | |
| # Common time is deliberately the tie-breaker: sparse arrangements often | |
| # make 2- and 4-beat periodicities mathematically indistinguishable. | |
| meter_priors = {2: 0.0, 3: 0.08, 4: 0.32} | |
| energy_mean = float(beat_energy.mean()) + 1e-9 | |
| for meter_value in (2, 3, 4): | |
| best_meter_score = -1e9 | |
| for bar_phase in range(meter_value): | |
| downbeats = beat_energy[bar_phase::meter_value] | |
| others = np.asarray( | |
| [value for index, value in enumerate(beat_energy) if index % meter_value != bar_phase], | |
| dtype=float, | |
| ) | |
| contrast = (float(downbeats.mean()) - float(others.mean() if len(others) else 0.0)) / energy_mean | |
| periodicity = 0.0 | |
| if len(beat_energy) > meter_value * 2: | |
| left = beat_energy[:-meter_value] | |
| right = beat_energy[meter_value:] | |
| if float(left.std()) > 1e-6 and float(right.std()) > 1e-6: | |
| periodicity = float(np.corrcoef(left, right)[0, 1]) | |
| best_meter_score = max(best_meter_score, contrast * 0.72 + periodicity * 0.28) | |
| if best_meter_score == contrast * 0.72 + periodicity * 0.28: | |
| meter_phases[meter_value] = bar_phase | |
| meter_scores[meter_value] = best_meter_score + meter_priors[meter_value] | |
| meter = max(meter_scores, key=meter_scores.get) | |
| bar_phase = int(meter_phases.get(meter, 0)) | |
| relative = (onsets - beat_phase) / beat_seconds | |
| offbeat = np.abs(relative - np.rint(relative)) > 0.08 | |
| compound = False | |
| if int(np.count_nonzero(offbeat)) >= max(3, len(onsets) // 8): | |
| triple_error = np.abs(relative[offbeat] * 3 - np.rint(relative[offbeat] * 3)) | |
| binary_error = np.minimum( | |
| np.abs(relative[offbeat] * 2 - np.rint(relative[offbeat] * 2)), | |
| np.abs(relative[offbeat] * 4 - np.rint(relative[offbeat] * 4)), | |
| ) | |
| compound = float(np.median(triple_error)) + 0.035 < float(np.median(binary_error)) | |
| if meter == 3: | |
| time_signature = "3/4" | |
| elif compound and meter == 2: | |
| time_signature = "6/8" | |
| elif compound and meter == 4: | |
| time_signature = "12/8" | |
| else: | |
| time_signature = f"{meter}/4" | |
| # Use chord-collapsed onsets here. Small inter-player and chord-note | |
| # offsets should not force a 1/32 grid merely because the source was a | |
| # human performance rather than step-entered notation. | |
| starts = onsets | |
| note_durations = np.asarray(durations, dtype=float) | |
| quantization_fits: dict[str, float] = {} | |
| for label, divisors in QUANTIZATION_GRIDS.items(): | |
| start_error = np.ones_like(starts) | |
| duration_error = np.ones_like(note_durations) | |
| for divisor in divisors: | |
| grid = beat_seconds / divisor | |
| start_units = (starts - beat_phase) / grid | |
| duration_units = note_durations / grid | |
| start_error = np.minimum(start_error, np.abs(start_units - np.rint(start_units))) | |
| duration_error = np.minimum(duration_error, np.abs(duration_units - np.rint(duration_units))) | |
| # Note-off timing is expressive in real performances; onset placement | |
| # is the much stronger signal for choosing an engraving grid. | |
| fit = 0.9 * float(np.mean(start_error <= 0.2)) + 0.1 * float(np.mean(duration_error <= 0.22)) | |
| quantization_fits[label] = fit | |
| if quantization_fits["1/8"] >= 0.82: | |
| quantization = "1/8" | |
| elif quantization_fits["1/16"] >= 0.72: | |
| quantization = "1/16" | |
| elif ( | |
| quantization_fits["1/32"] >= 0.82 | |
| and quantization_fits["1/32"] >= quantization_fits["1/16"] + 0.07 | |
| ): | |
| quantization = "1/32" | |
| else: | |
| # False precision is substantially harder for a musician to correct | |
| # than a few deliberately simplified ornaments. | |
| quantization = "1/16" | |
| peak = float(tempo_scores[best_index]) | |
| baseline = float(np.quantile(tempo_scores, 0.75)) + 1e-9 | |
| tempo_separation = max(0.0, min(1.0, (peak - baseline) / (peak + 1e-9))) | |
| confidence = "high" if len(onsets) >= 24 and tempo_separation >= 0.55 else "medium" if len(onsets) >= 10 else "low" | |
| # Preserve a lightweight symbolic tempo map. Onsets close to an integer | |
| # pulse gently move that pulse; missing pulses retain the global estimate. | |
| # This lets the engraving follow moderate rubato without letting every | |
| # ornamental note distort the bar structure. | |
| raw_beat_indices = np.rint((onsets - beat_phase) / beat_seconds).astype(int) | |
| first_index = int(raw_beat_indices.min()) - 1 | |
| last_index = int(raw_beat_indices.max()) + 2 | |
| beat_times: list[float] = [] | |
| for beat_index in range(first_index, last_index + 1): | |
| predicted = beat_phase + beat_index * beat_seconds | |
| distances = np.abs(onsets - predicted) | |
| nearby = distances <= beat_seconds * 0.16 | |
| if int(np.count_nonzero(nearby)): | |
| local_weights = accents[nearby] * np.exp( | |
| -0.5 * (distances[nearby] / (beat_seconds * 0.09)) ** 2 | |
| ) | |
| observed = float(np.average(onsets[nearby], weights=local_weights)) | |
| beat_time = predicted * 0.38 + observed * 0.62 | |
| else: | |
| beat_time = predicted | |
| if beat_times: | |
| beat_time = max(beat_times[-1] + beat_seconds * 0.35, beat_time) | |
| beat_times.append(round(beat_time, 6)) | |
| first_downbeat_index = minimum_beat + bar_phase | |
| first_onset = float(onsets[0]) | |
| bar_seconds = meter * beat_seconds | |
| while beat_phase + first_downbeat_index * beat_seconds < first_onset - beat_seconds * 0.18: | |
| first_downbeat_index += meter | |
| candidate_downbeat = beat_phase + first_downbeat_index * beat_seconds | |
| previous_downbeat = candidate_downbeat - bar_seconds | |
| # Avoid labelling almost a whole opening bar as a pickup when the clip | |
| # simply starts just after a downbeat. | |
| if candidate_downbeat - first_onset > bar_seconds * 0.62: | |
| candidate_downbeat = previous_downbeat | |
| first_downbeat_index -= meter | |
| pickup_beats = max(0.0, (candidate_downbeat - first_onset) / beat_seconds) | |
| pickup_beats = round(pickup_beats * 4) / 4 | |
| if pickup_beats < 0.25 or pickup_beats >= meter: | |
| pickup_beats = 0.0 | |
| while candidate_downbeat > first_onset + beat_seconds * 0.18: | |
| candidate_downbeat -= bar_seconds | |
| first_downbeat_index -= meter | |
| beat_quarter_length = 1.5 if time_signature in {"6/8", "9/8", "12/8"} else 1.0 | |
| key_signature, key_confidence = _estimate_key_signature(tracks, np) | |
| review_flags: list[str] = [] | |
| if confidence == "low": | |
| review_flags.append("Tempo and meter have low confidence; verify them against the recording.") | |
| if key_confidence < 0.34: | |
| review_flags.append("The detected key is ambiguous; confirm pitch spelling and accidentals.") | |
| if quantization_fits[quantization] < 0.68: | |
| review_flags.append( | |
| "The rhythmic grid has low confidence; 1/16 was preferred over unreadable false precision." | |
| ) | |
| ambiguous = sorted( | |
| { | |
| str(track.get("key") or "") | |
| for track in tracks | |
| if str(track.get("key") or "") in AMBIGUOUS_INSTRUMENT_GROUPS | |
| } | |
| ) | |
| if ambiguous: | |
| review_flags.append( | |
| "Broad instrument groups need an exact player/instrument assignment: " | |
| + ", ".join(name.replace("_", " ") for name in ambiguous) | |
| + "." | |
| ) | |
| review_flags.append( | |
| "Dynamics are not predicted by MuScriptor and are intentionally left neutral." | |
| ) | |
| return { | |
| "analysis_version": 3, | |
| "tempo_bpm": round(best_bpm, 1), | |
| "time_signature": time_signature, | |
| "quantization": quantization, | |
| "key_signature": key_signature, | |
| "key_confidence": key_confidence, | |
| "confidence": confidence, | |
| "onset_count": int(len(onsets)), | |
| "tempo_confidence": round(tempo_separation, 3), | |
| "tempo_octave_adjusted": tempo_octave_adjusted, | |
| "onset_rate_hz": round(onset_rate_hz, 3), | |
| "quantization_fit": round(quantization_fits[quantization], 3), | |
| "beat_phase_seconds": round(beat_phase, 6), | |
| "first_downbeat_seconds": round(candidate_downbeat, 6), | |
| "pickup_beats": pickup_beats, | |
| "beat_quarter_length": beat_quarter_length, | |
| "beat_times_seconds": beat_times, | |
| "beat_origin_index": int(first_downbeat_index - first_index), | |
| "grid_family": "binary + ternary", | |
| "analysis_source": "symbolic notes", | |
| "review_flags": review_flags, | |
| } | |
| _INVALID_XML_CHARACTERS = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f]") | |
| _MUSICXML_NUMERIC_VOICE = re.compile(r"(<voice>)(\d+)(</voice>)") | |
| def _clean_display_text(value: Any, fallback: str) -> str: | |
| cleaned = _INVALID_XML_CHARACTERS.sub("", str(value or "")) | |
| cleaned = " ".join(cleaned.split()) | |
| return cleaned or fallback | |
| def safe_filename_stem(value: str, fallback: str = "muscriptor-score") -> str: | |
| ascii_value = ( | |
| unicodedata.normalize("NFKD", str(value or "")) | |
| .encode("ascii", "ignore") | |
| .decode("ascii") | |
| .lower() | |
| ) | |
| return re.sub(r"[^a-z0-9]+", "-", ascii_value).strip("-") or fallback | |
| def _safe_slug(value: str) -> str: | |
| return safe_filename_stem(value, "instrument") | |
| def _parse_time_signature(value: str) -> tuple[int, int]: | |
| try: | |
| numerator, denominator = value.split("/", 1) | |
| parsed = int(numerator), int(denominator) | |
| except (AttributeError, TypeError, ValueError): | |
| return 4, 4 | |
| if parsed[0] <= 0 or parsed[1] not in {1, 2, 4, 8, 16, 32}: | |
| return 4, 4 | |
| return parsed | |
| def _beat_quarter_length(time_signature: str) -> float: | |
| numerator, denominator = _parse_time_signature(time_signature) | |
| return 1.5 if denominator == 8 and numerator in {6, 9, 12} else 1.0 | |
| def _quantization_grid(label: str) -> tuple[int, ...]: | |
| return QUANTIZATION_GRIDS.get(str(label), QUANTIZATION_GRIDS["1/16"]) | |
| def _nearest_grid_value( | |
| value: float, | |
| divisors: tuple[int, ...], | |
| *, | |
| for_duration: bool = False, | |
| ) -> float: | |
| value = max(0.0, float(value)) | |
| candidates: list[float] = [] | |
| for divisor in divisors: | |
| numerator = max(0, round(value * divisor)) | |
| fraction = Fraction(numerator, divisor) | |
| remainder = fraction - int(fraction) | |
| # Durations such as 5/6 or 7/8 of a quarter are technically | |
| # representable but produce distracting 6:5 / 8:7 tuplets. Keep | |
| # ternary positions for onsets while favouring musician-readable note | |
| # values and ties for durations. | |
| if for_duration and remainder and remainder.numerator > 4: | |
| continue | |
| candidates.append(float(fraction)) | |
| if not candidates: | |
| candidates = [round(value * QUANTIZATION_DIVISORS["1/16"]) / QUANTIZATION_DIVISORS["1/16"]] | |
| return min(candidates, key=lambda candidate: (abs(candidate - value), -candidate)) | |
| def _key_for_label(value: str, key_module: Any) -> Any: | |
| text = _clean_display_text(value, "C major").replace("♭", "-").replace("b", "-").replace("♯", "#") | |
| match = re.fullmatch(r"([A-Ga-g])([#-]?)[ ]+(major|minor)", text, flags=re.IGNORECASE) | |
| if not match: | |
| return key_module.Key("C", "major") | |
| tonic = match.group(1).upper() + match.group(2) | |
| return key_module.Key(tonic, match.group(3).lower()) | |
| def _normalized_key_label(value: str) -> str: | |
| text = _clean_display_text(value, "C major").replace("♭", "-").replace("b", "-").replace("♯", "#") | |
| match = re.fullmatch(r"([A-Ga-g])([#-]?)[ ]+(major|minor)", text, flags=re.IGNORECASE) | |
| if not match: | |
| return "C major" | |
| accidental = {"-": "♭", "#": "♯", "": ""}[match.group(2)] | |
| return f"{match.group(1).upper()}{accidental} {match.group(3).lower()}" | |
| class _ScoreTiming: | |
| bpm: float | |
| beat_quarter_length: float | |
| first_downbeat_seconds: float | |
| pickup_beats: float | |
| beat_times_seconds: tuple[float, ...] | |
| beat_origin_index: int | |
| use_tempo_map: bool | |
| def beat_seconds(self) -> float: | |
| return 60.0 / self.bpm | |
| def _pulse_position(self, seconds: float) -> float: | |
| if not self.use_tempo_map or len(self.beat_times_seconds) < 2: | |
| return (seconds - self.first_downbeat_seconds) / self.beat_seconds | |
| values = self.beat_times_seconds | |
| index = bisect.bisect_right(values, seconds) - 1 | |
| if index < 0: | |
| spacing = max(1e-6, values[1] - values[0]) | |
| position = (seconds - values[0]) / spacing | |
| elif index >= len(values) - 1: | |
| spacing = max(1e-6, values[-1] - values[-2]) | |
| position = len(values) - 1 + (seconds - values[-1]) / spacing | |
| else: | |
| spacing = max(1e-6, values[index + 1] - values[index]) | |
| position = index + (seconds - values[index]) / spacing | |
| return position - self.beat_origin_index | |
| def offset_quarters(self, seconds: float) -> float: | |
| pulses = self._pulse_position(seconds) + self.pickup_beats | |
| return max(0.0, pulses * self.beat_quarter_length) | |
| def duration_quarters(self, start: float, end: float) -> float: | |
| start_pulse = self._pulse_position(start) | |
| end_pulse = self._pulse_position(end) | |
| return max(0.0, (end_pulse - start_pulse) * self.beat_quarter_length) | |
| def _score_timing( | |
| tracks: list[dict[str, Any]], | |
| *, | |
| bpm: float, | |
| time_signature: str, | |
| pickup_beats: float, | |
| timing_analysis: dict[str, Any] | None, | |
| ) -> _ScoreTiming: | |
| note_starts = [ | |
| max(0.0, float(item.get("start", 0.0))) | |
| for track in tracks | |
| for item in track.get("notes") or [] | |
| ] | |
| first_note = min(note_starts, default=0.0) | |
| beat_ql = _beat_quarter_length(time_signature) | |
| pickup = max(0.0, float(pickup_beats or 0.0)) | |
| numerator, denominator = _parse_time_signature(time_signature) | |
| bar_quarters = numerator * 4.0 / denominator | |
| pickup = min(pickup, max(0.0, bar_quarters / beat_ql - 0.25)) | |
| analysis = timing_analysis or {} | |
| suggested_bpm = float(analysis.get("tempo_bpm") or 0.0) | |
| suggested_pickup = float(analysis.get("pickup_beats") or 0.0) | |
| analysis_matches = ( | |
| suggested_bpm > 0 | |
| and abs(suggested_bpm - bpm) / bpm <= 0.04 | |
| and str(analysis.get("time_signature") or "") == time_signature | |
| and abs(suggested_pickup - pickup) <= 0.13 | |
| ) | |
| if analysis_matches: | |
| first_downbeat = float(analysis.get("first_downbeat_seconds") or first_note) | |
| beat_times = tuple( | |
| float(value) | |
| for value in analysis.get("beat_times_seconds") or [] | |
| if math.isfinite(float(value)) | |
| ) | |
| origin = int(analysis.get("beat_origin_index") or 0) | |
| valid_map = ( | |
| len(beat_times) >= 2 | |
| and 0 <= origin < len(beat_times) | |
| and all(right > left for left, right in zip(beat_times, beat_times[1:])) | |
| ) | |
| if valid_map: | |
| first_downbeat = beat_times[origin] | |
| else: | |
| first_downbeat = first_note + pickup * (60.0 / bpm) | |
| beat_times = () | |
| origin = 0 | |
| valid_map = False | |
| if pickup <= 0 and first_downbeat > first_note + 60.0 / bpm * 0.25: | |
| first_downbeat = first_note | |
| valid_map = False | |
| return _ScoreTiming( | |
| bpm=bpm, | |
| beat_quarter_length=beat_ql, | |
| first_downbeat_seconds=first_downbeat, | |
| pickup_beats=pickup, | |
| beat_times_seconds=beat_times, | |
| beat_origin_index=origin, | |
| use_tempo_map=valid_map, | |
| ) | |
| def _prepare_tracks( | |
| tracks: list[dict[str, Any]], | |
| *, | |
| bpm: float, | |
| time_signature: str, | |
| cleanup_profile: str, | |
| ) -> tuple[list[dict[str, Any]], dict[str, Any]]: | |
| profile_key = cleanup_profile if cleanup_profile in CLEANUP_PROFILES else "readable" | |
| profile = CLEANUP_PROFILES[profile_key] | |
| beat_seconds = 60.0 / bpm | |
| merge_gap = float(profile["merge_gap_beats"]) * beat_seconds | |
| minimum_duration = float(profile["minimum_note_beats"]) * beat_seconds | |
| ghost_duration = float(profile["drop_ghost_beats"]) * beat_seconds | |
| totals = { | |
| "input_notes": 0, | |
| "output_notes": 0, | |
| "merged_notes": 0, | |
| "lengthened_notes": 0, | |
| "dropped_notes": 0, | |
| "trimmed_overlaps": 0, | |
| } | |
| prepared: list[dict[str, Any]] = [] | |
| track_reports: list[dict[str, Any]] = [] | |
| for track in tracks: | |
| source_notes = list(track.get("notes") or []) | |
| totals["input_notes"] += len(source_notes) | |
| is_drum = bool(track.get("is_drum")) | |
| by_pitch: dict[int, list[dict[str, Any]]] = {} | |
| for item in source_notes: | |
| pitch = max(0, min(127, int(item.get("pitch", 60)))) | |
| start = max(0.0, float(item.get("start", 0.0))) | |
| end = max(start + 0.01, float(item.get("end", start + 0.1))) | |
| normalized = dict(item) | |
| normalized.update({"pitch": pitch, "start": start, "end": end}) | |
| by_pitch.setdefault(pitch, []).append(normalized) | |
| cleaned: list[dict[str, Any]] = [] | |
| track_merged = 0 | |
| track_lengthened = 0 | |
| track_dropped = 0 | |
| track_trimmed = 0 | |
| for pitch_notes in by_pitch.values(): | |
| pitch_notes.sort(key=lambda item: (item["start"], item["end"])) | |
| merged: list[dict[str, Any]] = [] | |
| for item in pitch_notes: | |
| if merged and item["start"] <= merged[-1]["end"] + merge_gap: | |
| merged[-1]["end"] = max(merged[-1]["end"], item["end"]) | |
| track_merged += 1 | |
| else: | |
| merged.append(item) | |
| for item in merged: | |
| duration = item["end"] - item["start"] | |
| if not is_drum and ghost_duration and duration < ghost_duration: | |
| track_dropped += 1 | |
| continue | |
| if not is_drum and duration < minimum_duration: | |
| item["end"] = item["start"] + minimum_duration | |
| track_lengthened += 1 | |
| cleaned.append(item) | |
| cleaned.sort(key=lambda item: (item["start"], item["pitch"], item["end"])) | |
| distinct_onsets = sorted({float(item["start"]) for item in cleaned}) | |
| trim_tolerance = max(0.02, merge_gap * 2.0) | |
| for item in cleaned: | |
| next_index = bisect.bisect_right(distinct_onsets, float(item["start"]) + 1e-9) | |
| if next_index >= len(distinct_onsets): | |
| continue | |
| next_onset = distinct_onsets[next_index] | |
| overlap = float(item["end"]) - next_onset | |
| if 0 < overlap <= trim_tolerance: | |
| item["end"] = max(float(item["start"]) + 0.01, next_onset) | |
| track_trimmed += 1 | |
| updated_track = dict(track) | |
| updated_track["notes"] = cleaned | |
| updated_track["note_count"] = len(cleaned) | |
| prepared.append(updated_track) | |
| totals["output_notes"] += len(cleaned) | |
| totals["merged_notes"] += track_merged | |
| totals["lengthened_notes"] += track_lengthened | |
| totals["dropped_notes"] += track_dropped | |
| totals["trimmed_overlaps"] += track_trimmed | |
| track_reports.append( | |
| { | |
| "track": str(track.get("name") or track.get("key") or "Instrument"), | |
| "input_notes": len(source_notes), | |
| "output_notes": len(cleaned), | |
| "merged_notes": track_merged, | |
| "lengthened_notes": track_lengthened, | |
| "dropped_notes": track_dropped, | |
| "trimmed_overlaps": track_trimmed, | |
| } | |
| ) | |
| return prepared, { | |
| "profile": profile_key, | |
| "profile_label": profile["label"], | |
| **totals, | |
| "tracks": track_reports, | |
| } | |
| def _split_keyboard_hands(notes: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: | |
| """Split piano notes with chord-aware, range-aware hand continuity.""" | |
| ordered = sorted(notes, key=lambda item: (float(item.get("start", 0.0)), int(item.get("pitch", 60)))) | |
| groups: list[list[dict[str, Any]]] = [] | |
| for item in ordered: | |
| if not groups or float(item.get("start", 0.0)) - float(groups[-1][0].get("start", 0.0)) > 0.045: | |
| groups.append([item]) | |
| else: | |
| groups[-1].append(item) | |
| lower: list[dict[str, Any]] = [] | |
| upper: list[dict[str, Any]] = [] | |
| lower_center = 48.0 | |
| upper_center = 72.0 | |
| for group in groups: | |
| group = sorted(group, key=lambda item: int(item.get("pitch", 60))) | |
| pitches = [int(item.get("pitch", 60)) for item in group] | |
| best_split = 0 | |
| best_cost = float("inf") | |
| for split in range(len(group) + 1): | |
| low_pitches = pitches[:split] | |
| high_pitches = pitches[split:] | |
| cost = 0.0 | |
| cost += sum(max(0, pitch - 64) ** 2 * 0.45 for pitch in low_pitches) | |
| cost += sum(max(0, 55 - pitch) ** 2 * 0.45 for pitch in high_pitches) | |
| if low_pitches: | |
| cost += abs(sum(low_pitches) / len(low_pitches) - lower_center) * 0.55 | |
| elif min(pitches) < 58: | |
| cost += 18 | |
| if high_pitches: | |
| cost += abs(sum(high_pitches) / len(high_pitches) - upper_center) * 0.55 | |
| elif max(pitches) > 64: | |
| cost += 18 | |
| if low_pitches and high_pitches: | |
| gap = high_pitches[0] - low_pitches[-1] | |
| cost -= min(8, max(0, gap)) * 0.6 | |
| if cost < best_cost: | |
| best_cost = cost | |
| best_split = split | |
| low_group = group[:best_split] | |
| high_group = group[best_split:] | |
| lower.extend(low_group) | |
| upper.extend(high_group) | |
| if low_group: | |
| lower_center = lower_center * 0.62 + median([int(item["pitch"]) for item in low_group]) * 0.38 | |
| if high_group: | |
| upper_center = upper_center * 0.62 + median([int(item["pitch"]) for item in high_group]) * 0.38 | |
| # A true grand staff needs usable material in both hands. The caller can | |
| # fall back to one staff when a strongly one-sided performance is found. | |
| return upper, lower | |
| def _instrument_for_track(track: dict[str, Any], instrument_module: Any, interval_module: Any) -> Any: | |
| if track.get("is_drum"): | |
| result = instrument_module.UnpitchedPercussion() | |
| result.midiChannel = 9 | |
| else: | |
| program = max(0, min(127, int(track.get("program") or 0))) | |
| try: | |
| result = instrument_module.instrumentFromMidiProgram(program) | |
| except Exception: | |
| result = instrument_module.Instrument() | |
| result.midiProgram = program | |
| octave_transposers = tuple( | |
| cls | |
| for cls in ( | |
| getattr(instrument_module, "Guitar", None), | |
| getattr(instrument_module, "AcousticGuitar", None), | |
| getattr(instrument_module, "ElectricGuitar", None), | |
| getattr(instrument_module, "AcousticBass", None), | |
| getattr(instrument_module, "ElectricBass", None), | |
| getattr(instrument_module, "Contrabass", None), | |
| ) | |
| if cls is not None | |
| ) | |
| if octave_transposers and isinstance(result, octave_transposers): | |
| result.transposition = interval_module.Interval("P-8") | |
| name = str(track.get("name") or "Instrument") | |
| result.partName = name | |
| result.instrumentName = name | |
| return result | |
| def _clef_for_track( | |
| track: dict[str, Any], | |
| midi_instrument: Any, | |
| pitches: list[int], | |
| clef_module: Any, | |
| instrument_module: Any, | |
| staff_role: str | None = None, | |
| ) -> Any: | |
| if track.get("is_drum"): | |
| return clef_module.PercussionClef() | |
| if staff_role == "upper": | |
| return clef_module.TrebleClef() | |
| if staff_role == "lower": | |
| return clef_module.BassClef() | |
| if isinstance(midi_instrument, instrument_module.Viola): | |
| return clef_module.AltoClef() | |
| bass_classes = tuple( | |
| cls | |
| for cls in ( | |
| getattr(instrument_module, "AcousticBass", None), | |
| getattr(instrument_module, "ElectricBass", None), | |
| getattr(instrument_module, "Contrabass", None), | |
| getattr(instrument_module, "Violoncello", None), | |
| getattr(instrument_module, "Bassoon", None), | |
| getattr(instrument_module, "Trombone", None), | |
| getattr(instrument_module, "BassTrombone", None), | |
| getattr(instrument_module, "Tuba", None), | |
| ) | |
| if cls is not None | |
| ) | |
| if bass_classes and isinstance(midi_instrument, bass_classes): | |
| return clef_module.BassClef() | |
| guitar_classes = tuple( | |
| cls | |
| for cls in ( | |
| getattr(instrument_module, "Guitar", None), | |
| getattr(instrument_module, "AcousticGuitar", None), | |
| getattr(instrument_module, "ElectricGuitar", None), | |
| ) | |
| if cls is not None | |
| ) | |
| if guitar_classes and isinstance(midi_instrument, guitar_classes): | |
| return clef_module.Treble8vbClef() | |
| return clef_module.BassClef() if median(pitches) < 52 else clef_module.TrebleClef() | |
| def _drum_note(pitch: int, duration: float, note_module: Any, instrument_module: Any) -> Any: | |
| if pitch in {35, 36}: | |
| display, class_name = "F4", "BassDrum" | |
| elif pitch in {37, 38, 39, 40}: | |
| display, class_name = "C5", "SnareDrum" | |
| elif pitch in {41, 43, 45, 47, 48, 50}: | |
| display, class_name = "A4", "TomTom" | |
| elif pitch in {42, 44, 46}: | |
| display, class_name = "G5", "HiHatCymbal" | |
| elif pitch in {49, 51, 52, 53, 55, 57, 59}: | |
| display, class_name = "A5", "Cymbals" | |
| elif pitch == 54: | |
| display, class_name = "E5", "Tambourine" | |
| elif pitch == 56: | |
| display, class_name = "D5", "Cowbell" | |
| elif pitch in {60, 61}: | |
| display, class_name = "F5", "BongoDrums" | |
| elif pitch in {62, 63, 64}: | |
| display, class_name = "E5", "CongaDrum" | |
| else: | |
| display, class_name = "B4", "UnpitchedPercussion" | |
| stored_instrument = getattr(instrument_module, class_name)() | |
| return note_module.Unpitched( | |
| displayName=display, | |
| storedInstrument=stored_instrument, | |
| quarterLength=duration, | |
| ) | |
| def _build_music21_staff( | |
| track: dict[str, Any], | |
| notes: list[dict[str, Any]], | |
| *, | |
| staff_id: str, | |
| staff_role: str | None, | |
| display_name: str, | |
| numerator: int, | |
| denominator: int, | |
| divisors: tuple[int, ...], | |
| timing: _ScoreTiming, | |
| key_signature: str, | |
| show_solfege: bool, | |
| add_tempo: bool, | |
| modules: dict[str, Any], | |
| ) -> Any: | |
| chord = modules["chord"] | |
| clef = modules["clef"] | |
| duration_module = modules["duration"] | |
| instrument = modules["instrument"] | |
| interval = modules["interval"] | |
| key_module = modules["key"] | |
| meter = modules["meter"] | |
| note = modules["note"] | |
| percussion = modules["percussion"] | |
| stream = modules["stream"] | |
| tempo = modules["tempo"] | |
| part_class = stream.PartStaff if staff_role else stream.Part | |
| part = part_class(id=staff_id) | |
| part.partName = display_name | |
| midi_instrument = _instrument_for_track(track, instrument, interval) | |
| midi_instrument.partName = display_name | |
| part.insert(0, midi_instrument) | |
| if add_tempo: | |
| part.insert( | |
| 0, | |
| tempo.MetronomeMark( | |
| number=timing.bpm, | |
| referent=duration_module.Duration(timing.beat_quarter_length), | |
| ), | |
| ) | |
| part.insert(0, meter.TimeSignature(f"{numerator}/{denominator}")) | |
| part.insert(0, _key_for_label(key_signature, key_module)) | |
| pitches = [max(0, min(127, int(item.get("pitch", 60)))) for item in notes] | |
| part.insert(0, _clef_for_track(track, midi_instrument, pitches, clef, instrument, staff_role)) | |
| grouped: dict[float, list[tuple[int, float]]] = {} | |
| bar_quarters = numerator * 4.0 / denominator | |
| pickup_quarters = timing.pickup_beats * timing.beat_quarter_length | |
| pickup_padding = max(0.0, bar_quarters - pickup_quarters) if pickup_quarters else 0.0 | |
| minimum_duration = 1.0 / max(divisors) | |
| for item in notes: | |
| start_seconds = max(0.0, float(item.get("start", 0.0))) | |
| end_seconds = max(start_seconds + 0.03, float(item.get("end", start_seconds + 0.1))) | |
| raw_offset = timing.offset_quarters(start_seconds) + pickup_padding | |
| raw_duration = timing.duration_quarters(start_seconds, end_seconds) | |
| offset = _nearest_grid_value(raw_offset, divisors) | |
| duration = max( | |
| minimum_duration, | |
| _nearest_grid_value(raw_duration, divisors, for_duration=True), | |
| ) | |
| grouped.setdefault(round(offset, 6), []).append( | |
| ( | |
| max(0, min(127, int(item.get("pitch", 60)))), | |
| round(duration, 6), | |
| ) | |
| ) | |
| for offset, values in sorted(grouped.items()): | |
| # A performed chord rarely has identical note-off times. Group by the | |
| # quantized onset and use a representative duration instead of asking | |
| # makeVoices() to create one notated voice per key release. | |
| durations_by_pitch: dict[int, float] = {} | |
| for pitch, duration in values: | |
| durations_by_pitch[pitch] = max(durations_by_pitch.get(pitch, 0.0), duration) | |
| group_pitches = list(durations_by_pitch) | |
| duration = max(minimum_duration, float(median(durations_by_pitch.values()))) | |
| if track.get("is_drum"): | |
| unpitched = [_drum_note(pitch, duration, note, instrument) for pitch in group_pitches] | |
| musical_item = ( | |
| unpitched[0] | |
| if len(unpitched) == 1 | |
| else percussion.PercussionChord(unpitched, quarterLength=duration) | |
| ) | |
| elif len(group_pitches) == 1: | |
| musical_item = note.Note(group_pitches[0], quarterLength=duration) | |
| if show_solfege: | |
| musical_item.addLyric(SOLFEGE_NAMES[group_pitches[0] % 12]) | |
| else: | |
| unique_pitches = list(dict.fromkeys(group_pitches)) | |
| musical_item = chord.Chord(unique_pitches, quarterLength=duration) | |
| if show_solfege: | |
| musical_item.addLyric(" ".join(SOLFEGE_NAMES[pitch % 12] for pitch in unique_pitches)) | |
| part.insert(offset, musical_item) | |
| part.quantize( | |
| quarterLengthDivisors=divisors, | |
| processOffsets=True, | |
| processDurations=True, | |
| inPlace=True, | |
| ) | |
| voiced = part.makeVoices(inPlace=False, fillGaps=False) | |
| measured = voiced.makeMeasures(inPlace=False) | |
| measured.makeNotation(inPlace=True) | |
| if pickup_quarters: | |
| first_measure = measured.measure(1) | |
| if first_measure is not None: | |
| containers = list(first_measure.voices) or [first_measure] | |
| for container in containers: | |
| leading_rests = [ | |
| item | |
| for item in container.notesAndRests | |
| if isinstance(item, note.Rest) and float(item.offset) < pickup_padding | |
| ] | |
| if leading_rests: | |
| container.remove(leading_rests) | |
| for item in container.notesAndRests: | |
| item.offset = max(0.0, float(item.offset) - pickup_padding) | |
| first_measure.paddingLeft = pickup_padding | |
| first_measure.number = 0 | |
| first_measure.showNumber = stream.enums.ShowNumber.NEVER | |
| # MuScriptor pitches are concert/sounding pitches. Converting here gives | |
| # clarinet, saxophone, trumpet and horn players proper written parts while | |
| # keeping MusicXML playback at the original sounding pitch. | |
| measured.atSoundingPitch = True | |
| measured.toWrittenPitch(inPlace=True) | |
| return measured | |
| def _music21_score( | |
| tracks: list[dict[str, Any]], | |
| *, | |
| title: str, | |
| tempo_bpm: float, | |
| time_signature: str, | |
| quantization: str, | |
| show_solfege: bool, | |
| key_signature: str = "C major", | |
| pickup_beats: float = 0.0, | |
| cleanup_profile: str = "readable", | |
| timing_analysis: dict[str, Any] | None = None, | |
| tracks_prepared: bool = False, | |
| ) -> tuple[Any, list[dict[str, Any]]]: | |
| try: | |
| from music21 import ( | |
| chord, | |
| clef, | |
| duration, | |
| instrument, | |
| interval, | |
| key, | |
| layout, | |
| metadata, | |
| meter, | |
| note, | |
| percussion, | |
| stream, | |
| tempo, | |
| ) | |
| except ImportError as exc: # pragma: no cover - deployment configuration | |
| raise RuntimeError("La dépendance music21 est nécessaire pour générer la partition.") from exc | |
| modules = { | |
| "chord": chord, | |
| "clef": clef, | |
| "duration": duration, | |
| "instrument": instrument, | |
| "interval": interval, | |
| "key": key, | |
| "meter": meter, | |
| "note": note, | |
| "percussion": percussion, | |
| "stream": stream, | |
| "tempo": tempo, | |
| } | |
| bpm = max(20.0, min(300.0, float(tempo_bpm))) | |
| numerator, denominator = _parse_time_signature(time_signature) | |
| divisors = _quantization_grid(quantization) | |
| if not tracks_prepared: | |
| tracks, _ = _prepare_tracks( | |
| tracks, | |
| bpm=bpm, | |
| time_signature=time_signature, | |
| cleanup_profile=cleanup_profile, | |
| ) | |
| timing = _score_timing( | |
| tracks, | |
| bpm=bpm, | |
| time_signature=time_signature, | |
| pickup_beats=pickup_beats, | |
| timing_analysis=timing_analysis, | |
| ) | |
| score = stream.Score(id="muscriptor-score") | |
| score.metadata = metadata.Metadata() | |
| score.metadata.title = _clean_display_text(title, "MuScriptor transcription") | |
| score.metadata.composer = "Transcription automatique MuScriptor" | |
| built_tracks: list[dict[str, Any]] = [] | |
| for track_index, track in enumerate(tracks): | |
| track_notes = list(track.get("notes") or []) | |
| if not track_notes: | |
| continue | |
| track_name = _clean_display_text(track.get("name"), f"Instrument {track_index + 1}") | |
| key = str(track.get("key") or "").lower() | |
| pitches = [int(item.get("pitch", 60)) for item in track_notes] | |
| grand_staff = ( | |
| any(token in key for token in ("piano", "keyboard", "organ", "harp")) | |
| and min(pitches) < 60 <= max(pitches) | |
| and len(track_notes) >= 4 | |
| ) | |
| parts: list[Any] = [] | |
| if grand_staff: | |
| upper_notes, lower_notes = _split_keyboard_hands(track_notes) | |
| staff_specs = ( | |
| ("upper", upper_notes, track_name), | |
| ("lower", lower_notes, ""), | |
| ) | |
| if not upper_notes or not lower_notes: | |
| staff_specs = ((None, track_notes, track_name),) | |
| else: | |
| staff_specs = ((None, track_notes, track_name),) | |
| for staff_number, (staff_role, staff_notes, display_name) in enumerate(staff_specs, start=1): | |
| if not staff_notes: | |
| continue | |
| measured = _build_music21_staff( | |
| track, | |
| staff_notes, | |
| staff_id=f"track-{track_index + 1}-staff-{staff_number}", | |
| staff_role=staff_role, | |
| display_name=display_name, | |
| numerator=numerator, | |
| denominator=denominator, | |
| divisors=divisors, | |
| timing=timing, | |
| key_signature=key_signature, | |
| show_solfege=show_solfege, | |
| add_tempo=staff_number == 1, | |
| modules=modules, | |
| ) | |
| score.insert(0, measured) | |
| parts.append(measured) | |
| if len(parts) > 1: | |
| score.insert( | |
| 0, | |
| layout.StaffGroup( | |
| *parts, | |
| name=track_name, | |
| abbreviation=track_name[:8], | |
| symbol="brace", | |
| barTogether=True, | |
| ), | |
| ) | |
| built_tracks.append({"track": track, "name": track_name, "parts": tuple(parts)}) | |
| if not score.parts: | |
| raise ValueError("Aucune note n’est disponible pour générer une partition.") | |
| return score, built_tracks | |
| def _write_musicxml(score: Any, output_path: Path) -> Path: | |
| """Write well-formed UTF-8 MusicXML and remove filename control chars. | |
| music21 preserves control characters found in an uploaded filename. Those | |
| characters make the document invalid XML for strict readers even though | |
| some Verovio builds accept them, so sanitize and validate every export. | |
| """ | |
| score.write("musicxml", fp=str(output_path)) | |
| xml_text = output_path.read_text(encoding="utf-8-sig") | |
| cleaned = _INVALID_XML_CHARACTERS.sub("", xml_text) | |
| # music21 numbers Voice objects from zero in some polyphonic exports. | |
| # MusicXML technically accepts a token here, but Verovio maps numeric | |
| # voices to one-based MEI layers and reports every <voice>0</voice> as a | |
| # missing layer. If a zero-based voice is present, shift the complete | |
| # numeric voice set together so voices remain distinct and portable. | |
| if "<voice>0</voice>" in cleaned: | |
| cleaned = _MUSICXML_NUMERIC_VOICE.sub( | |
| lambda match: ( | |
| f"{match.group(1)}{int(match.group(2)) + 1}{match.group(3)}" | |
| ), | |
| cleaned, | |
| ) | |
| if cleaned != xml_text: | |
| output_path.write_text(cleaned, encoding="utf-8") | |
| try: | |
| ElementTree.fromstring(cleaned.encode("utf-8")) | |
| except ElementTree.ParseError as exc: | |
| raise RuntimeError( | |
| f"The generated MusicXML is not well formed ({output_path.name}): {exc}" | |
| ) from exc | |
| return output_path | |
| def _verovio_options(layout_kind: str) -> dict[str, Any]: | |
| is_part = layout_kind == "part" | |
| return { | |
| "adjustPageHeight": False, | |
| "breaks": "auto", | |
| "footer": "none", | |
| "header": "none", | |
| # Full scores use A4 landscape; individual parts use A4 portrait. | |
| "pageHeight": 2970 if is_part else 2100, | |
| "pageWidth": 2100 if is_part else 2970, | |
| "pageMarginBottom": 80, | |
| "pageMarginLeft": 80, | |
| "pageMarginRight": 80, | |
| "pageMarginTop": 80, | |
| "mmOutput": True, | |
| "scale": 100, | |
| # Verovio's documented raster guidance uses about 9 units for parts | |
| # and a smaller staff for a multi-instrument conductor score. | |
| "unit": 9.0 if is_part else 6.875, | |
| "spacingStaff": 12 if is_part else 10, | |
| "spacingSystem": 18 if is_part else 15, | |
| "minLastJustification": 0.62, | |
| } | |
| def _verovio_toolkit(verovio_module: Any, layout_kind: str = "score") -> Any: | |
| """Create a toolkit with an explicit wheel resource path when available.""" | |
| module_file = getattr(verovio_module, "__file__", "") | |
| resource_path = Path(module_file).resolve().parent / "data" if module_file else None | |
| toolkit = None | |
| if resource_path and resource_path.is_dir(): | |
| try: | |
| toolkit = verovio_module.toolkit(False) | |
| if not toolkit.setResourcePath(str(resource_path)): | |
| toolkit = None | |
| except (AttributeError, TypeError): | |
| toolkit = None | |
| if toolkit is None: | |
| toolkit = verovio_module.toolkit() | |
| toolkit.setOptions(_verovio_options(layout_kind)) | |
| return toolkit | |
| def _verovio_log(toolkit: Any) -> str: | |
| try: | |
| raw_log = toolkit.getLog() | |
| except Exception: | |
| raw_log = "" | |
| lines = [line.strip() for line in str(raw_log or "").splitlines() if line.strip()] | |
| return " · ".join(lines[-3:])[-600:] if lines else "no parser details" | |
| def _load_verovio_musicxml( | |
| verovio_module: Any, | |
| musicxml: Path, | |
| xml_text: str, | |
| *, | |
| layout_kind: str = "score", | |
| ) -> Any: | |
| """Load MusicXML through both native Python routes before giving up. | |
| `loadFile` is Verovio's documented Python path and avoids copying a large | |
| score through the binding. `loadData` remains a useful independent retry | |
| for unusual temporary-file or filesystem behaviour on hosted Spaces. | |
| """ | |
| diagnostics: list[str] = [] | |
| for method in ("file", "data"): | |
| toolkit = _verovio_toolkit(verovio_module, layout_kind) | |
| try: | |
| loaded = ( | |
| toolkit.loadFile(str(musicxml.resolve())) | |
| if method == "file" | |
| else toolkit.loadData(xml_text) | |
| ) | |
| except Exception as exc: | |
| diagnostics.append(f"{method} loader: {type(exc).__name__}: {exc}") | |
| continue | |
| if loaded: | |
| return toolkit | |
| diagnostics.append(f"{method} loader: {_verovio_log(toolkit)}") | |
| root_name = ElementTree.fromstring(xml_text.encode("utf-8")).tag | |
| detail = " · ".join(diagnostics)[-1200:] | |
| raise RuntimeError( | |
| f"Verovio could not read {musicxml.name} after file and data loading " | |
| f"({root_name}, {len(xml_text.encode('utf-8'))} bytes): {detail}" | |
| ) | |
| def _render_svg_pages( | |
| musicxml: Path, | |
| output_dir: Path, | |
| *, | |
| page_stem: str, | |
| layout_kind: str = "score", | |
| ) -> tuple[tuple[Path, ...], str]: | |
| try: | |
| import verovio | |
| except ImportError as exc: # pragma: no cover - deployment configuration | |
| raise RuntimeError("La dépendance verovio est nécessaire pour afficher la partition.") from exc | |
| xml_text = _INVALID_XML_CHARACTERS.sub("", musicxml.read_text(encoding="utf-8-sig")) | |
| try: | |
| ElementTree.fromstring(xml_text.encode("utf-8")) | |
| except ElementTree.ParseError as exc: | |
| raise RuntimeError(f"Invalid MusicXML in {musicxml.name}: {exc}") from exc | |
| toolkit = _load_verovio_musicxml( | |
| verovio, | |
| musicxml, | |
| xml_text, | |
| layout_kind=layout_kind, | |
| ) | |
| pages: list[Path] = [] | |
| preview = "" | |
| for page_number in range(1, toolkit.getPageCount() + 1): | |
| svg = toolkit.renderToSVG(page_number) | |
| if page_number == 1: | |
| preview = svg | |
| path = output_dir / f"{page_stem}-page-{page_number:02d}.svg" | |
| path.write_text(svg, encoding="utf-8") | |
| pages.append(path) | |
| return tuple(pages), preview | |
| def _render_pdf(svg_pages: tuple[Path, ...], output_path: Path) -> Path | None: | |
| try: | |
| import cairosvg | |
| from pypdf import PdfReader, PdfWriter | |
| except ImportError: | |
| return None | |
| writer = PdfWriter() | |
| for svg in svg_pages: | |
| pdf_bytes = cairosvg.svg2pdf(bytestring=svg.read_bytes()) | |
| reader = PdfReader(io.BytesIO(pdf_bytes)) | |
| for page in reader.pages: | |
| writer.add_page(page) | |
| with output_path.open("wb") as handle: | |
| writer.write(handle) | |
| return output_path | |
| def generate_notation( | |
| tracks: list[dict[str, Any]], | |
| *, | |
| title: str, | |
| tempo_bpm: float = 120, | |
| time_signature: str = "4/4", | |
| quantization: str = "1/16", | |
| key_signature: str = "C major", | |
| pickup_beats: float = 0.0, | |
| cleanup_profile: str = "readable", | |
| show_solfege: bool = False, | |
| midi_files: list[str] | None = None, | |
| timing_analysis: dict[str, Any] | None = None, | |
| ) -> NotationResult: | |
| output_dir = Path(tempfile.mkdtemp(prefix="muscriptor-notation-")) | |
| clean_title = _clean_display_text(title, "MuScriptor transcription") | |
| export_stem = safe_filename_stem(clean_title) | |
| bpm = max(20.0, min(300.0, float(tempo_bpm))) | |
| numerator, denominator = _parse_time_signature(time_signature) | |
| time_signature = f"{numerator}/{denominator}" | |
| quantization = quantization if quantization in QUANTIZATION_GRIDS else "1/16" | |
| key_signature = _normalized_key_label(key_signature) | |
| profile_key = cleanup_profile if cleanup_profile in CLEANUP_PROFILES else "readable" | |
| prepared_tracks, cleanup_diagnostics = _prepare_tracks( | |
| tracks, | |
| bpm=bpm, | |
| time_signature=time_signature, | |
| cleanup_profile=profile_key, | |
| ) | |
| timing = _score_timing( | |
| prepared_tracks, | |
| bpm=bpm, | |
| time_signature=time_signature, | |
| pickup_beats=pickup_beats, | |
| timing_analysis=timing_analysis, | |
| ) | |
| score, built_tracks = _music21_score( | |
| prepared_tracks, | |
| title=clean_title, | |
| tempo_bpm=bpm, | |
| time_signature=time_signature, | |
| quantization=quantization, | |
| key_signature=key_signature, | |
| pickup_beats=pickup_beats, | |
| cleanup_profile=profile_key, | |
| timing_analysis=timing_analysis, | |
| tracks_prepared=True, | |
| show_solfege=show_solfege, | |
| ) | |
| musicxml = output_dir / f"{export_stem}-full-score.musicxml" | |
| _write_musicxml(score, musicxml) | |
| warnings: list[str] = [] | |
| svg_pages: tuple[Path, ...] = () | |
| preview_svg = "" | |
| pdf: Path | None = None | |
| try: | |
| svg_pages, preview_svg = _render_svg_pages( | |
| musicxml, | |
| output_dir, | |
| page_stem=f"{export_stem}-full-score", | |
| ) | |
| except Exception as exc: | |
| warnings.append(f"Full score preview: {type(exc).__name__}: {exc}") | |
| if svg_pages: | |
| try: | |
| pdf = _render_pdf(svg_pages, output_dir / f"{export_stem}-full-score.pdf") | |
| except Exception as exc: | |
| warnings.append(f"Full score PDF: {type(exc).__name__}: {exc}") | |
| parts_dir = output_dir / "parts" | |
| parts_dir.mkdir() | |
| from music21 import layout as music21_layout | |
| from music21 import stream as music21_stream | |
| part_results: list[NotationPartResult] = [] | |
| for part_index, built_track in enumerate(built_tracks, start=1): | |
| source_track = built_track["track"] | |
| part_name = _clean_display_text(built_track["name"], f"Instrument {part_index}") | |
| part_slug = f"{export_stem}-{part_index:02d}-{_safe_slug(part_name)}" | |
| part_score = music21_stream.Score(id=f"score-track-{part_index}") | |
| part_score.metadata = copy.deepcopy(score.metadata) | |
| copied_parts = [copy.deepcopy(part) for part in built_track["parts"]] | |
| for copied_part in copied_parts: | |
| part_score.insert(0, copied_part) | |
| if len(copied_parts) > 1: | |
| part_score.insert( | |
| 0, | |
| music21_layout.StaffGroup( | |
| *copied_parts, | |
| name=part_name, | |
| abbreviation=part_name[:8], | |
| symbol="brace", | |
| barTogether=True, | |
| ), | |
| ) | |
| part_musicxml = parts_dir / f"{part_slug}.musicxml" | |
| _write_musicxml(part_score, part_musicxml) | |
| part_svg_dir = parts_dir / f"{part_slug}-svg" | |
| part_svg_dir.mkdir() | |
| part_svg_pages: tuple[Path, ...] = () | |
| part_pdf: Path | None = None | |
| part_warnings: list[str] = [] | |
| try: | |
| part_svg_pages, _ = _render_svg_pages( | |
| part_musicxml, | |
| part_svg_dir, | |
| page_stem=part_slug, | |
| layout_kind="part", | |
| ) | |
| except Exception as exc: | |
| part_warnings.append(f"{part_name} preview: {type(exc).__name__}: {exc}") | |
| if part_svg_pages: | |
| try: | |
| part_pdf = _render_pdf(part_svg_pages, parts_dir / f"{part_slug}.pdf") | |
| except Exception as exc: | |
| part_warnings.append(f"{part_name} PDF: {type(exc).__name__}: {exc}") | |
| warnings.extend(part_warnings) | |
| source_id = source_track.get("id") | |
| if source_id is None: | |
| source_id = source_track.get("key") or part_index | |
| part_results.append( | |
| NotationPartResult( | |
| track_id=str(source_id), | |
| track_key=str(source_track.get("key") or source_track.get("id") or part_slug), | |
| name=part_name, | |
| slug=part_slug, | |
| musicxml=part_musicxml, | |
| pdf=part_pdf, | |
| svg_pages=part_svg_pages, | |
| render_error=" · ".join(part_warnings), | |
| ) | |
| ) | |
| review_flags = list(dict.fromkeys((timing_analysis or {}).get("review_flags") or [])) | |
| dynamics_flag = "Dynamics are not predicted by MuScriptor and are intentionally left neutral." | |
| if dynamics_flag not in review_flags: | |
| review_flags.append(dynamics_flag) | |
| if cleanup_diagnostics["dropped_notes"]: | |
| review_flags.append( | |
| f"{cleanup_diagnostics['dropped_notes']} very short note(s) were removed by the Readable profile." | |
| ) | |
| if cleanup_diagnostics["lengthened_notes"]: | |
| review_flags.append( | |
| f"{cleanup_diagnostics['lengthened_notes']} short note(s) were lengthened to remain engravable." | |
| ) | |
| if cleanup_diagnostics["trimmed_overlaps"]: | |
| review_flags.append( | |
| f"{cleanup_diagnostics['trimmed_overlaps']} small note-off overlap(s) were trimmed to avoid spurious voices." | |
| ) | |
| diagnostics: dict[str, Any] = { | |
| "format_version": 1, | |
| "title": clean_title, | |
| "settings": { | |
| "tempo_bpm": round(bpm, 3), | |
| "tempo_referent": ( | |
| "dotted quarter" if timing.beat_quarter_length == 1.5 else "quarter" | |
| ), | |
| "time_signature": time_signature, | |
| "key_signature": key_signature, | |
| "quantization": quantization, | |
| "quantization_divisors": list(_quantization_grid(quantization)), | |
| "grid_family": "binary + ternary", | |
| "pickup_beats": round(timing.pickup_beats, 3), | |
| "cleanup_profile": profile_key, | |
| "solfege_lyrics": bool(show_solfege), | |
| "dynamics": "neutral; MuScriptor does not predict note velocity", | |
| }, | |
| "timing": { | |
| "first_downbeat_seconds": round(timing.first_downbeat_seconds, 6), | |
| "symbolic_tempo_map_used": timing.use_tempo_map, | |
| "beat_anchor_count": len(timing.beat_times_seconds) if timing.use_tempo_map else 0, | |
| }, | |
| "cleanup": cleanup_diagnostics, | |
| "analysis": { | |
| key: value | |
| for key, value in (timing_analysis or {}).items() | |
| if key != "beat_times_seconds" | |
| }, | |
| "outputs": { | |
| "part_count": len(part_results), | |
| "full_score_pages": len(svg_pages), | |
| }, | |
| "review_flags": list(dict.fromkeys(review_flags)), | |
| "render_warnings": list(warnings), | |
| } | |
| report = output_dir / f"{export_stem}-score-preparation.json" | |
| report.write_text( | |
| json.dumps(diagnostics, ensure_ascii=False, indent=2, sort_keys=True), | |
| encoding="utf-8", | |
| ) | |
| bundle = output_dir / f"{export_stem}-exports.zip" | |
| with zipfile.ZipFile(bundle, "w", compression=zipfile.ZIP_DEFLATED) as archive: | |
| archive.write(musicxml, musicxml.name) | |
| archive.write(report, report.name) | |
| for part_result in part_results: | |
| archive.write(part_result.musicxml, f"parts/{part_result.musicxml.name}") | |
| if part_result.pdf: | |
| archive.write(part_result.pdf, f"parts/{part_result.pdf.name}") | |
| for path in part_result.svg_pages: | |
| archive.write(path, f"parts/svg/{part_result.slug}/{path.name}") | |
| for path in svg_pages: | |
| archive.write(path, f"svg/{path.name}") | |
| if pdf: | |
| archive.write(pdf, pdf.name) | |
| for value in midi_files or []: | |
| midi_path = Path(value) | |
| if midi_path.is_file(): | |
| archive.write(midi_path, f"midi/{midi_path.name}") | |
| return NotationResult( | |
| musicxml=musicxml, | |
| pdf=pdf, | |
| bundle=bundle, | |
| svg_pages=svg_pages, | |
| preview_svg=preview_svg, | |
| parts=tuple(part_results), | |
| warnings=tuple(warnings), | |
| export_stem=export_stem, | |
| report=report, | |
| diagnostics=diagnostics, | |
| ) | |
| def session_json(value: dict[str, Any]) -> str: | |
| return json.dumps(value, ensure_ascii=False, separators=(",", ":")) | |