| """PLUS core: нотная модель, чтение/запись MIDI, темп и общие утилиты. |
| |
| Всё остальное в пакете опирается на этот файл, поэтому здесь нет зависимостей |
| кроме pretty_midi и numpy. Ни одна функция не требует GPU. |
| """ |
| from __future__ import annotations |
|
|
| import json |
| import os |
| from dataclasses import dataclass, field, asdict |
| from typing import Iterable, Iterator, Sequence |
|
|
| import numpy as np |
|
|
| __version__ = "1.0" |
|
|
| DRUM_CHANNEL = 9 |
|
|
|
|
| @dataclass |
| class Note: |
| """Нота в секундах. pitch 0..127, velocity 1..127.""" |
|
|
| start: float |
| dur: float |
| pitch: int |
| velocity: int = 80 |
| channel: int = 0 |
| track: str = "main" |
|
|
| @property |
| def end(self) -> float: |
| return self.start + self.dur |
|
|
| def shifted(self, delta: float) -> "Note": |
| return Note(self.start + delta, self.dur, self.pitch, self.velocity, self.channel, self.track) |
|
|
| def as_tuple(self) -> tuple[float, float, int, int, int]: |
| return (self.start, self.dur, self.pitch, self.channel, self.velocity) |
|
|
|
|
| @dataclass |
| class Piece: |
| """Набор нот плюс карта темпа и метаданные.""" |
|
|
| notes: list[Note] = field(default_factory=list) |
| tempo_changes: list[tuple[float, float]] = field(default_factory=lambda: [(0.0, 120.0)]) |
| time_signature: tuple[int, int] = (4, 4) |
| source: str = "" |
| meta: dict = field(default_factory=dict) |
|
|
| def copy(self) -> "Piece": |
| return Piece([Note(**asdict(n)) for n in self.notes], |
| list(self.tempo_changes), self.time_signature, self.source, dict(self.meta)) |
|
|
| |
| @property |
| def duration(self) -> float: |
| return max((n.end for n in self.notes), default=0.0) |
|
|
| @property |
| def bpm(self) -> float: |
| return self.tempo_changes[0][1] if self.tempo_changes else 120.0 |
|
|
| def pitches(self) -> np.ndarray: |
| return np.array([n.pitch for n in self.notes], dtype=int) |
|
|
| def starts(self) -> np.ndarray: |
| return np.array([n.start for n in self.notes], dtype=float) |
|
|
| def velocities(self) -> np.ndarray: |
| return np.array([n.velocity for n in self.notes], dtype=int) |
|
|
| def stats(self) -> dict: |
| if not self.notes: |
| return {"notes": 0, "duration": 0.0} |
| dur = self.duration or 1e-9 |
| return { |
| "notes": len(self.notes), |
| "duration": round(dur, 3), |
| "notes_per_second": round(len(self.notes) / dur, 2), |
| "pitch_min": int(min(n.pitch for n in self.notes)), |
| "pitch_max": int(max(n.pitch for n in self.notes)), |
| "velocity_min": int(min(n.velocity for n in self.notes)), |
| "velocity_max": int(max(n.velocity for n in self.notes)), |
| "bpm": round(self.bpm, 2), |
| } |
|
|
|
|
| |
| def _pretty_midi(): |
| try: |
| import pretty_midi |
| except Exception as exc: |
| raise SystemExit("нужен pretty_midi: pip install pretty_midi (%s)" % exc) |
| return pretty_midi |
|
|
|
|
| def load_midi(path: str) -> Piece: |
| """MIDI-файл как Piece. Дорожки сохраняются по именам инструментов.""" |
| pm = _pretty_midi() |
| midi = pm.PrettyMIDI(path) |
| notes: list[Note] = [] |
| for instrument in midi.instruments: |
| name = instrument.name or ("drums" if instrument.is_drum else "track%d" % instrument.program) |
| for n in instrument.notes: |
| notes.append(Note(float(n.start), max(1e-3, float(n.end - n.start)), |
| int(n.pitch), int(n.velocity) or 80, |
| int(instrument.program), name)) |
| notes.sort(key=lambda n: (n.start, n.pitch)) |
| times, bpms = midi.get_tempo_changes() |
| tempo = [(float(t), float(b)) for t, b in zip(times, bpms)] or [(0.0, 120.0)] |
| ts = midi.time_signature_changes[0] if midi.time_signature_changes else None |
| piece = Piece(notes, tempo, (ts.numerator, ts.denominator) if ts else (4, 4), |
| source=os.path.basename(path)) |
| return piece |
|
|
|
|
| def save_midi(piece: Piece, path: str, split_tracks: bool = True) -> str: |
| """Piece в MIDI. При split_tracks каждая дорожка идёт своим инструментом.""" |
| pm = _pretty_midi() |
| midi = pm.PrettyMIDI(initial_tempo=piece.bpm or 120.0) |
| groups: dict[str, list[Note]] = {} |
| for n in piece.notes: |
| groups.setdefault(n.track if split_tracks else "main", []).append(n) |
| for index, (name, group) in enumerate(sorted(groups.items())): |
| is_drum = name.lower() in ("drums", "drum") |
| instrument = pm.Instrument(program=0 if is_drum else group[0].channel % 128, |
| is_drum=is_drum, name=name) |
| for n in sorted(group, key=lambda x: x.start): |
| instrument.notes.append(pm.Note(int(max(1, min(127, int(n.velocity)))), |
| int(max(0, min(127, int(n.pitch)))), |
| float(n.start), |
| float(n.end))) |
| midi.instruments.append(instrument) |
| os.makedirs(os.path.dirname(os.path.abspath(path)) or ".", exist_ok=True) |
| midi.write(path) |
| return path |
|
|
|
|
| |
| def explained_fraction(starts: Sequence[float], bpm: float, subdivisions: int = 4, |
| tolerance: float | None = None) -> float: |
| """Доля онсетов, попадающих на сетку 1/subdivisions доли в пределах допуска.""" |
| onsets = np.asarray(list(starts), dtype=float) |
| if onsets.size == 0 or bpm <= 0: |
| return 0.0 |
| step = 60.0 / bpm / subdivisions |
| tol = tolerance if tolerance is not None else max(0.012, step * 0.22) |
| residual = np.abs(onsets - np.round(onsets / step) * step) |
| return float((residual <= tol).mean()) |
|
|
|
|
| def tempo_from_onsets(starts: Sequence[float], lo: float = 50.0, hi: float = 220.0, |
| center: float | None = None, spread: float = 1.25) -> float: |
| """Темп, на чью сетку ложится больше всего онсетов. |
| |
| С бинарными онсетами чистые восьмые неотличимы от четвертей вдвое быстрее, |
| поэтому без center выбирается самый МЕДЛЕННЫЙ темп с объяснением >= 0.92 |
| (он же самый правдоподобный для нотного текста), а ambiguity_of() показывает |
| кратные варианты. С center поиск сужается до center*[1/spread, spread]. |
| """ |
| onsets = np.asarray(list(starts), dtype=float) |
| if onsets.size < 4: |
| return float(center or 120.0) |
| low, high = (center / spread, center * spread) if center else (lo, hi) |
| candidates = np.arange(max(20.0, low), min(400.0, high) + 0.01, 0.5) |
| scored = [(explained_fraction(onsets, bpm), bpm) for bpm in candidates] |
| best_fraction = max(fraction for fraction, _bpm in scored) |
| if best_fraction < 0.5: |
| return round(float(center or 120.0), 2) |
| solid = [bpm for fraction, bpm in scored if fraction >= max(0.92, best_fraction - 0.03)] |
| chosen = min(solid) if center is None else min(solid, key=lambda b: abs(b - center)) |
| return round(float(chosen), 2) |
|
|
|
|
| def ambiguity_of(starts: Sequence[float], bpm: float) -> list[dict]: |
| """Кратные варианты темпа с их объяснением: честная оценка неоднозначности.""" |
| out = [] |
| for factor in (0.5, 1.0, 2.0, 3.0): |
| candidate = bpm * factor |
| if 20 <= candidate <= 400: |
| out.append({"bpm": round(candidate, 2), |
| "explained": round(explained_fraction(starts, candidate), 3)}) |
| return out |
|
|
|
|
| def grid_times(piece: Piece, bpm: float | None = None, division: int = 4) -> np.ndarray: |
| """Сетка долей (division долей на четверть) на всю длительность.""" |
| bpm = bpm or piece.bpm |
| beat = 60.0 / bpm |
| step = beat / division |
| total = max(piece.duration + beat, step) |
| return np.arange(0.0, total, step) |
|
|
|
|
| def nearest_grid(times: Iterable[float], grid: np.ndarray) -> np.ndarray: |
| times = np.asarray(list(times), dtype=float) |
| if grid.size == 0: |
| return times |
| index = np.clip(np.searchsorted(grid, times), 1, grid.size - 1) |
| left, right = grid[index - 1], grid[index] |
| return np.where(np.abs(times - left) <= np.abs(right - times), left, right) |
|
|
|
|
| |
| def step(number: int, total: int, text: str) -> None: |
| """Единый формат прогресса: [этап] сообщение.""" |
| print("[%d/%d] %s" % (number, total, text), flush=True) |
|
|
|
|
| def json_default(obj): |
| """numpy-скаляры в JSON: np.bool_ / np.float32 не сериализуются как есть.""" |
| if hasattr(obj, "item"): |
| try: |
| return obj.item() |
| except Exception: |
| pass |
| if isinstance(obj, (set, tuple)): |
| return list(obj) |
| return str(obj) |
|
|
|
|
| def dump_json(payload: dict, path: str) -> str: |
| os.makedirs(os.path.dirname(os.path.abspath(path)) or ".", exist_ok=True) |
| with open(path, "w", encoding="utf-8") as handle: |
| json.dump(payload, handle, ensure_ascii=False, indent=2, |
| default=json_default) |
| return path |
|
|
|
|
| def pitch_name(pitch: int) -> str: |
| names = ["C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"] |
| return "%s%d" % (names[int(pitch) % 12], int(pitch) // 12 - 1) |
|
|
|
|
|
|
