./generate.zsh \
--model-dir keras-gabc-model-deep \
--epochs 50 \
--patience 8 \
--context 64 \
--embedding 256 \
--mode-embedding 48 \
--units 768 \
--dropout 0.20 \
--batch-size 128 \
--learning-rate 0.0005./generative.zsh \
keras-gabc-model-deep \
--output-dir generated-gabc-deep \
--min-complex-neumes 10 \
--min-torculus 5 \
--complexity-bias 1.5 \
--torculus-bias 1.75./assemble-generated-pages.zsh| #!/usr/bin/env zsh | |
| # Assemble the individual Keras-generated melodies into one complete GABC | |
| # anthology source for each Gregorian mode. | |
| # | |
| # Usage: | |
| # ./assemble-generated-pages.zsh [--overwrite] [generated-gabc-directory] | |
| # | |
| # The default source directory is generated-gabc beside this script. Output is | |
| # written to its pages/ subdirectory. Each mode page contains exactly ten | |
| # source melodies and 500 sung notes. | |
| emulate -L zsh | |
| setopt err_exit no_unset pipe_fail | |
| readonly SCRIPT_DIR=${0:A:h} | |
| readonly PYTHON_BIN=${PYTHON:-python3} | |
| if ! command -v "$PYTHON_BIN" >/dev/null 2>&1; then | |
| print -u2 -- "error: Python interpreter not found: $PYTHON_BIN" | |
| exit 1 | |
| fi | |
| exec "$PYTHON_BIN" - "$SCRIPT_DIR" "$@" <<'PYTHON' | |
| from __future__ import annotations | |
| import argparse | |
| import os | |
| import re | |
| import sys | |
| from pathlib import Path | |
| SCRIPT_DIR = Path(sys.argv.pop(1)).resolve() | |
| CLEF_RE = re.compile(r"[cf]b?[1-4]") | |
| GROUP_RE = re.compile(r"\(([^()]*)\)", re.DOTALL) | |
| ROMAN_MODES = { | |
| 1: "I", | |
| 2: "II", | |
| 3: "III", | |
| 4: "IV", | |
| 5: "V", | |
| 6: "VI", | |
| 7: "VII", | |
| 8: "VIII", | |
| } | |
| def parse_arguments() -> argparse.Namespace: | |
| parser = argparse.ArgumentParser( | |
| prog="assemble-generated-pages.zsh", | |
| description="Assemble ten generated melodies into a titled GABC page for each mode.", | |
| ) | |
| parser.add_argument( | |
| "source_dir", | |
| nargs="?", | |
| type=Path, | |
| default=SCRIPT_DIR / "generated-gabc", | |
| help="directory containing mode-N-NN.gabc files (default: %(default)s)", | |
| ) | |
| parser.add_argument( | |
| "--overwrite", | |
| action="store_true", | |
| help="replace existing assembled page sources", | |
| ) | |
| return parser.parse_args() | |
| def split_document(document: str, path: Path) -> tuple[str, str]: | |
| parts = re.split(r"^\s*%%\s*$", document, maxsplit=1, flags=re.MULTILINE) | |
| if len(parts) != 2: | |
| raise SystemExit(f"error: missing GABC header delimiter in {path}") | |
| return parts[0], parts[1].strip() | |
| def header_value(header: str, key: str) -> str | None: | |
| match = re.search( | |
| rf"^{re.escape(key)}\s*:\s*([^;\r\n]+)", | |
| header, | |
| flags=re.MULTILINE | re.IGNORECASE, | |
| ) | |
| return match.group(1).strip() if match else None | |
| def count_notes(music: str) -> int: | |
| music = music.split("|", 1)[0] | |
| music = re.sub(r"\[[^]]*\]", "", music) | |
| music = re.sub(r"\{[^}]*\}", "", music) | |
| music = re.sub(r"<[^>]*>", "", music) | |
| count = 0 | |
| index = 0 | |
| while index < len(music): | |
| clef = CLEF_RE.match(music, index) | |
| if clef: | |
| index = clef.end() | |
| continue | |
| pitch = music[index] | |
| if pitch not in "abcdefghijklmABCDEFGHIJKLM": | |
| index += 1 | |
| continue | |
| following = music[index + 1 :] | |
| if following[:1] in {"x", "y", "#", "+"}: | |
| index += 2 | |
| continue | |
| repetitions = re.match(r"[sv]+", following) | |
| if repetitions: | |
| count += len(repetitions.group(0)) | |
| index += 1 + len(repetitions.group(0)) | |
| else: | |
| count += 1 | |
| index += 1 | |
| return count | |
| def document_note_count(body: str) -> int: | |
| return sum(count_notes(match.group(1)) for match in GROUP_RE.finditer(body)) | |
| def normalize_melody(body: str, number: int, source: Path) -> str: | |
| # Layout breaks in the individual files were chosen for standalone output. | |
| # Let Gregorio reflow the combined score, retaining only a forced staff end | |
| # between complete melodies. | |
| body = re.sub(r"\([zZ](?:[+-])?\)", "", body) | |
| # Model tokens may contain a double bar either as their entire group or at | |
| # the end of a note group. Remove every such internal bar, cleaning a cut | |
| # left dangling at the end, and add one canonical double bar below. | |
| def without_double_bar(match: re.Match[str]) -> str: | |
| music = match.group(1).replace("::", "") | |
| music = re.sub(r"[!/@]+$", "", music) | |
| return f"({music})" if music else "" | |
| body = GROUP_RE.sub(without_double_bar, body) | |
| body = re.sub(r"[ \t]+", " ", body) | |
| body = re.sub(r" *\n *", " ", body).strip() | |
| opening_clef = re.match(r"^\(([cf]b?[1-4])\)\s*", body) | |
| if not opening_clef: | |
| raise SystemExit(f"error: generated melody has no opening clef: {source}") | |
| normalized = ( | |
| f"({opening_clef.group(1)}) <alt>{number}</alt> " | |
| + body[opening_clef.end() :] | |
| ) | |
| return normalized.rstrip() + " (::)" | |
| def assemble_mode(source_dir: Path, mode: int) -> str: | |
| melodies = [] | |
| for number in range(1, 11): | |
| source = source_dir / f"mode-{mode}-{number:02d}.gabc" | |
| if not source.is_file(): | |
| raise SystemExit(f"error: missing generated melody: {source}") | |
| header, body = split_document(source.read_text(encoding="utf-8"), source) | |
| source_mode = header_value(header, "mode") | |
| if source_mode != str(mode): | |
| raise SystemExit( | |
| f"error: {source} declares mode {source_mode!r}, expected {mode}" | |
| ) | |
| notes = document_note_count(body) | |
| if notes != 50: | |
| raise SystemExit(f"error: {source} has {notes} sung notes, expected 50") | |
| melodies.append(normalize_melody(body, number, source)) | |
| body = " (z)\n\n".join(melodies) | |
| result = ( | |
| f"name:genmode{mode};\n" | |
| "supertitle:_;\n" | |
| f"title:Generative Melodies in Mode {ROMAN_MODES[mode]};\n" | |
| "subtitle:_;\n" | |
| f"mode:{mode};\n" | |
| "initial-style:0;\n" | |
| "commentary:Ten mode-conditioned Keras melodies, 50 notes each;\n" | |
| "%%\n" | |
| f"{body}\n" | |
| ) | |
| total = document_note_count(body) | |
| if total != 500: | |
| raise SystemExit( | |
| f"error: assembled mode {mode} has {total} sung notes, expected 500" | |
| ) | |
| return result | |
| def main() -> None: | |
| args = parse_arguments() | |
| source_dir = args.source_dir.expanduser().resolve() | |
| if not source_dir.is_dir(): | |
| raise SystemExit(f"error: generated GABC directory not found: {source_dir}") | |
| output_dir = source_dir / "pages" | |
| destinations = [ | |
| output_dir / f"genmode{mode}.gabc" | |
| for mode in range(1, 9) | |
| ] | |
| existing = [path for path in destinations if path.exists()] | |
| if existing and not args.overwrite: | |
| raise SystemExit( | |
| f"error: assembled page already exists: {existing[0]}; use --overwrite" | |
| ) | |
| documents = [assemble_mode(source_dir, mode) for mode in range(1, 9)] | |
| output_dir.mkdir(parents=True, exist_ok=True) | |
| for destination, document in zip(destinations, documents): | |
| temporary = destination.with_suffix(".gabc.tmp") | |
| temporary.write_text(document, encoding="utf-8") | |
| os.replace(temporary, destination) | |
| print(f"Wrote {destination}") | |
| if __name__ == "__main__": | |
| main() | |
| PYTHON |
| #!/usr/bin/env zsh | |
| # Train a mode-conditioned Keras language model on a GregoBase GABC corpus. | |
| # | |
| # The model operates on complete parenthesized GABC music groups rather than | |
| # individual characters. This makes every generated note token an observed, | |
| # syntactically valid GABC glyph/neume while still letting the recurrent model | |
| # learn melodic succession, grouping, rhythmic signs, and mode. | |
| # | |
| # Usage: | |
| # ./generate.zsh [options] [corpus] | |
| # | |
| # The corpus defaults to gregobasecorpus-v0 beside this script. It may be the | |
| # corpus root (containing gabc/) or a directory containing .gabc files itself. | |
| # Run with --help for training and model-size options. | |
| emulate -L zsh | |
| setopt err_exit no_unset pipe_fail | |
| readonly SCRIPT_DIR=${0:A:h} | |
| readonly PYTHON_BIN=${PYTHON:-python3} | |
| if ! command -v "$PYTHON_BIN" >/dev/null 2>&1; then | |
| print -u2 -- "error: Python interpreter not found: $PYTHON_BIN" | |
| exit 1 | |
| fi | |
| # Keras 3 uses TensorFlow by default but respects an explicitly selected Keras | |
| # backend. TensorFlow is the best-tested backend for this trainer. | |
| export KERAS_BACKEND=${KERAS_BACKEND:-tensorflow} | |
| typeset -i CHECK_DEPENDENCIES=1 | |
| for argument in "$@"; do | |
| if [[ $argument == -h || $argument == --help ]]; then | |
| CHECK_DEPENDENCIES=0 | |
| break | |
| fi | |
| done | |
| if (( CHECK_DEPENDENCIES )) && ! "$PYTHON_BIN" -c 'import keras, numpy' >/dev/null 2>&1; then | |
| print -u2 -- "error: Keras and NumPy are required for training." | |
| print -u2 -- "Install them in a virtual environment, for example:" | |
| print -u2 -- " $PYTHON_BIN -m pip install 'keras>=3' tensorflow numpy" | |
| print -u2 -- "You may also set PYTHON=/path/to/venv/bin/python." | |
| exit 1 | |
| fi | |
| exec "$PYTHON_BIN" - "$SCRIPT_DIR" "$@" <<'PYTHON' | |
| from __future__ import annotations | |
| import argparse | |
| import hashlib | |
| import json | |
| import math | |
| import os | |
| import random | |
| import re | |
| import sys | |
| from collections import Counter | |
| from pathlib import Path | |
| SCRIPT_DIR = Path(sys.argv.pop(1)).resolve() | |
| SPECIAL_TOKENS = ("<PAD>", "<UNK>", "<START>", "<END>") | |
| PAD, UNK, START, END = range(len(SPECIAL_TOKENS)) | |
| CLEF_RE = re.compile(r"[cf]b?[1-4]") | |
| CONTROL_RE = re.compile(r"[cf]b?[1-4]|::|:|;|,") | |
| MODE_RE = re.compile(r"^mode\s*:\s*([^;\r\n]+)", re.MULTILINE | re.IGNORECASE) | |
| GROUP_RE = re.compile(r"\(([^()]*)\)", re.DOTALL) | |
| def positive_int(value: str) -> int: | |
| number = int(value) | |
| if number < 1: | |
| raise argparse.ArgumentTypeError("must be a positive integer") | |
| return number | |
| def probability(value: str) -> float: | |
| number = float(value) | |
| if not 0.0 <= number < 1.0: | |
| raise argparse.ArgumentTypeError("must be at least 0 and less than 1") | |
| return number | |
| def parse_arguments() -> argparse.Namespace: | |
| parser = argparse.ArgumentParser( | |
| prog="generate.zsh", | |
| description=( | |
| "Train a Keras next-glyph model on every GABC score in a corpus, " | |
| "conditioning predictions on Gregorian mode." | |
| ), | |
| ) | |
| parser.add_argument( | |
| "corpus", | |
| nargs="?", | |
| type=Path, | |
| default=SCRIPT_DIR / "gregobasecorpus-v0", | |
| help="corpus root or GABC directory (default: %(default)s)", | |
| ) | |
| parser.add_argument( | |
| "-o", | |
| "--model-dir", | |
| type=Path, | |
| default=SCRIPT_DIR / "keras-gabc-model", | |
| help="directory for model.keras and metadata.json (default: %(default)s)", | |
| ) | |
| parser.add_argument("--epochs", type=positive_int, default=20) | |
| parser.add_argument("--batch-size", type=positive_int, default=256) | |
| parser.add_argument("--context", type=positive_int, default=32) | |
| parser.add_argument("--embedding", type=positive_int, default=192) | |
| parser.add_argument("--mode-embedding", type=positive_int, default=24) | |
| parser.add_argument("--units", type=positive_int, default=384) | |
| parser.add_argument("--dropout", type=probability, default=0.20) | |
| parser.add_argument("--validation-split", type=probability, default=0.05) | |
| parser.add_argument("--patience", type=positive_int, default=3) | |
| parser.add_argument("--learning-rate", type=float, default=0.001) | |
| parser.add_argument( | |
| "--min-frequency", | |
| type=positive_int, | |
| default=2, | |
| help="minimum corpus frequency for a directly generated glyph (default: 2)", | |
| ) | |
| parser.add_argument( | |
| "--max-vocab", | |
| type=positive_int, | |
| default=12000, | |
| help="maximum vocabulary size including four special tokens (default: 12000)", | |
| ) | |
| parser.add_argument("--seed", type=int, default=2026) | |
| parser.add_argument( | |
| "--overwrite", | |
| action="store_true", | |
| help="replace model.keras and metadata.json if they already exist", | |
| ) | |
| args = parser.parse_args() | |
| if args.max_vocab <= len(SPECIAL_TOKENS): | |
| parser.error(f"--max-vocab must exceed {len(SPECIAL_TOKENS)}") | |
| if args.learning_rate <= 0: | |
| parser.error("--learning-rate must be positive") | |
| return args | |
| def split_document(document: str) -> tuple[str, str] | None: | |
| parts = re.split(r"^\s*%%\s*$", document, maxsplit=1, flags=re.MULTILINE) | |
| if len(parts) != 2: | |
| return None | |
| return parts[0], parts[1] | |
| def canonical_mode(header: str) -> int: | |
| match = MODE_RE.search(header) | |
| if not match: | |
| return 0 | |
| value = match.group(1).strip() | |
| return int(value) if value in {str(mode) for mode in range(1, 9)} else 0 | |
| def count_notes(music: str) -> int: | |
| """Count primary-voice sung notes using Gregorio's compact s/v rules.""" | |
| count = 0 | |
| index = 0 | |
| while index < len(music): | |
| clef = CLEF_RE.match(music, index) | |
| if clef: | |
| index = clef.end() | |
| continue | |
| pitch = music[index] | |
| if pitch not in "abcdefghijklmABCDEFGHIJKLM": | |
| index += 1 | |
| continue | |
| following = music[index + 1 :] | |
| if following[:1] in {"x", "y", "#", "+"}: | |
| index += 2 | |
| continue | |
| repetitions = re.match(r"[sv]+", following) | |
| if repetitions: | |
| count += len(repetitions.group(0)) | |
| index += 1 + len(repetitions.group(0)) | |
| else: | |
| count += 1 | |
| index += 1 | |
| return count | |
| def clean_group(raw_group: str) -> list[str]: | |
| """Return one note token or normalized structural tokens from a group.""" | |
| # NABC following | and braced secondary voices use different alphabets. | |
| music = raw_group.split("|", 1)[0] | |
| music = re.sub(r"\[[^]]*\]", "", music) | |
| music = re.sub(r"\{[^}]*\}", "", music) | |
| music = re.sub(r"<[^>]*>", "", music) | |
| music = re.sub(r"\s+", "", music) | |
| # Source layout instructions and manual custodes do not belong to a newly | |
| # laid-out melody. Removing them also prevents their pitch letters from | |
| # being mistaken for sung notes. | |
| music = re.sub(r"[zZ](?:0|[+-])?", "", music) | |
| music = re.sub(r"[a-mA-M]\+", "", music) | |
| if not music: | |
| return [] | |
| if count_notes(music): | |
| return [music] | |
| # Canonicalizing control-only groups sharply reduces vocabulary size while | |
| # retaining their original order (e.g. a double bar followed by a clef). | |
| return [match.group(0) for match in CONTROL_RE.finditer(music)] | |
| def is_control_token(token: str) -> bool: | |
| return bool(re.fullmatch(r"[cf]b?[1-4]|::|:|;|,", token)) | |
| def read_corpus(corpus: Path): | |
| gabc_dir = corpus / "gabc" if (corpus / "gabc").is_dir() else corpus | |
| if not gabc_dir.is_dir(): | |
| raise SystemExit(f"error: GABC directory not found: {gabc_dir}") | |
| paths = sorted(gabc_dir.rglob("*.gabc")) | |
| if not paths: | |
| raise SystemExit(f"error: no .gabc files found in: {gabc_dir}") | |
| sequences: list[tuple[int, list[str]]] = [] | |
| token_counts: Counter[str] = Counter() | |
| score_modes: Counter[int] = Counter() | |
| digest = hashlib.sha256() | |
| malformed = 0 | |
| empty = 0 | |
| for path in paths: | |
| raw = path.read_bytes() | |
| relative = path.relative_to(gabc_dir).as_posix().encode("utf-8") | |
| digest.update(len(relative).to_bytes(4, "big")) | |
| digest.update(relative) | |
| digest.update(len(raw).to_bytes(8, "big")) | |
| digest.update(raw) | |
| document = raw.decode("utf-8", errors="replace") | |
| split = split_document(document) | |
| if split is None: | |
| malformed += 1 | |
| continue | |
| header, body = split | |
| mode = canonical_mode(header) | |
| tokens: list[str] = [] | |
| for match in GROUP_RE.finditer(body): | |
| tokens.extend(clean_group(match.group(1))) | |
| if not tokens: | |
| empty += 1 | |
| continue | |
| score_modes[mode] += 1 | |
| token_counts.update(tokens) | |
| sequences.append((mode, tokens)) | |
| if not sequences: | |
| raise SystemExit("error: no trainable GABC token sequences were found") | |
| return gabc_dir, paths, sequences, token_counts, score_modes, digest.hexdigest(), malformed, empty | |
| def make_vocabulary( | |
| counts: Counter[str], min_frequency: int, max_vocab: int | |
| ) -> tuple[list[str], dict[str, int]]: | |
| # Preserve all safe structural controls even if they are rare. Fill the | |
| # remaining capacity by corpus frequency, using lexical order as a stable | |
| # tie-breaker so identical input always receives identical token IDs. | |
| controls = sorted( | |
| (token for token in counts if is_control_token(token)), | |
| key=lambda token: (-counts[token], token), | |
| ) | |
| frequent = sorted( | |
| (token for token, count in counts.items() if count >= min_frequency), | |
| key=lambda token: (-counts[token], token), | |
| ) | |
| selected: list[str] = [] | |
| seen = set(SPECIAL_TOKENS) | |
| for token in controls + frequent: | |
| if token in seen: | |
| continue | |
| if len(selected) + len(SPECIAL_TOKENS) >= max_vocab: | |
| break | |
| selected.append(token) | |
| seen.add(token) | |
| vocabulary = list(SPECIAL_TOKENS) + selected | |
| return vocabulary, {token: index for index, token in enumerate(vocabulary)} | |
| def vectorize( | |
| sequences: list[tuple[int, list[str]]], token_to_id: dict[str, int], context: int | |
| ): | |
| sample_count = sum(len(tokens) + 1 for _, tokens in sequences) | |
| contexts = np.zeros((sample_count, context), dtype=np.int32) | |
| modes = np.zeros(sample_count, dtype=np.int32) | |
| targets = np.zeros(sample_count, dtype=np.int32) | |
| row = 0 | |
| for mode, tokens in sequences: | |
| ids = [START] + [token_to_id.get(token, UNK) for token in tokens] + [END] | |
| for target_position in range(1, len(ids)): | |
| previous = ids[max(0, target_position - context) : target_position] | |
| # Right padding preserves chronological order and permits Keras to | |
| # use its optimized recurrent implementation with mask_zero=True. | |
| contexts[row, : len(previous)] = previous | |
| modes[row] = mode | |
| targets[row] = ids[target_position] | |
| row += 1 | |
| return contexts, modes, targets | |
| def balanced_mode_weights(modes: np.ndarray) -> tuple[np.ndarray, dict[int, int]]: | |
| counts = {mode: int(np.count_nonzero(modes == mode)) for mode in range(9)} | |
| present_canonical = [mode for mode in range(1, 9) if counts[mode]] | |
| canonical_samples = sum(counts[mode] for mode in present_canonical) | |
| weights_by_mode = {0: 1.0} | |
| for mode in present_canonical: | |
| weights_by_mode[mode] = canonical_samples / ( | |
| len(present_canonical) * counts[mode] | |
| ) | |
| weights = np.asarray([weights_by_mode.get(int(mode), 1.0) for mode in modes], dtype=np.float32) | |
| weights /= float(weights.mean()) | |
| return weights, counts | |
| def build_model(args: argparse.Namespace, vocabulary_size: int) -> keras.Model: | |
| token_input = keras.Input(shape=(args.context,), dtype="int32", name="tokens") | |
| mode_input = keras.Input(shape=(), dtype="int32", name="mode") | |
| token_vectors = keras.layers.Embedding( | |
| vocabulary_size, | |
| args.embedding, | |
| mask_zero=True, | |
| name="token_embedding", | |
| )(token_input) | |
| mode_vector = keras.layers.Embedding( | |
| 9, args.mode_embedding, name="mode_embedding" | |
| )(mode_input) | |
| initial_state = keras.layers.Dense( | |
| args.units, activation="tanh", name="mode_initial_state" | |
| )(mode_vector) | |
| sequence_vector = keras.layers.GRU( | |
| args.units, dropout=args.dropout, name="glyph_gru" | |
| )(token_vectors, initial_state=[initial_state]) | |
| combined = keras.layers.Concatenate(name="mode_conditioned_state")( | |
| [sequence_vector, mode_vector] | |
| ) | |
| combined = keras.layers.Dropout(args.dropout, name="output_dropout")(combined) | |
| output = keras.layers.Dense( | |
| vocabulary_size, activation="softmax", name="next_glyph" | |
| )(combined) | |
| model = keras.Model([token_input, mode_input], output, name="mode_sensitive_gabc") | |
| model.compile( | |
| optimizer=keras.optimizers.Adam(learning_rate=args.learning_rate), | |
| loss=keras.losses.SparseCategoricalCrossentropy(), | |
| metrics=[keras.metrics.SparseCategoricalAccuracy(name="accuracy")], | |
| ) | |
| return model | |
| def main() -> None: | |
| args = parse_arguments() | |
| # Imports are intentionally delayed until after argument parsing so | |
| # --help remains available before a user installs the training runtime. | |
| global keras, np | |
| import keras | |
| import numpy as np | |
| if int(keras.__version__.split(".", 1)[0]) < 3: | |
| raise SystemExit("error: Keras 3 or newer is required") | |
| corpus = args.corpus.expanduser().resolve() | |
| model_dir = args.model_dir.expanduser().resolve() | |
| model_path = model_dir / "model.keras" | |
| metadata_path = model_dir / "metadata.json" | |
| existing = [path for path in (model_path, metadata_path) if path.exists()] | |
| if existing and not args.overwrite: | |
| joined = ", ".join(str(path) for path in existing) | |
| raise SystemExit(f"error: training output already exists: {joined}; use --overwrite") | |
| random.seed(args.seed) | |
| np.random.seed(args.seed) | |
| keras.utils.set_random_seed(args.seed) | |
| print(f"Reading GABC corpus: {corpus}", flush=True) | |
| ( | |
| gabc_dir, | |
| paths, | |
| sequences, | |
| token_counts, | |
| score_modes, | |
| corpus_digest, | |
| malformed, | |
| empty, | |
| ) = read_corpus(corpus) | |
| vocabulary, token_to_id = make_vocabulary( | |
| token_counts, args.min_frequency, args.max_vocab | |
| ) | |
| represented = sum(token_counts[token] for token in vocabulary[len(SPECIAL_TOKENS) :]) | |
| total_tokens = sum(token_counts.values()) | |
| print( | |
| f"Read {len(paths):,} files; retained {len(sequences):,} scores and " | |
| f"{total_tokens:,} music groups.", | |
| flush=True, | |
| ) | |
| if malformed or empty: | |
| print( | |
| f"Skipped {malformed:,} malformed and {empty:,} musically empty scores.", | |
| flush=True, | |
| ) | |
| print( | |
| "Scores by mode: " | |
| + ", ".join( | |
| f"{mode if mode else 'other/unknown'}={score_modes[mode]:,}" | |
| for mode in sorted(score_modes) | |
| ), | |
| flush=True, | |
| ) | |
| print( | |
| f"Vocabulary: {len(vocabulary):,} tokens; direct glyph coverage " | |
| f"{represented / total_tokens:.2%} ({represented:,}/{total_tokens:,}).", | |
| flush=True, | |
| ) | |
| contexts, modes, targets = vectorize(sequences, token_to_id, args.context) | |
| sample_weights, example_modes = balanced_mode_weights(modes) | |
| rng = np.random.default_rng(args.seed) | |
| order = rng.permutation(len(targets)) | |
| contexts = contexts[order] | |
| modes = modes[order] | |
| targets = targets[order] | |
| sample_weights = sample_weights[order] | |
| print(f"Training examples: {len(targets):,}", flush=True) | |
| model = build_model(args, len(vocabulary)) | |
| model.summary() | |
| callbacks: list[keras.callbacks.Callback] = [] | |
| if args.validation_split: | |
| callbacks.append( | |
| keras.callbacks.EarlyStopping( | |
| monitor="val_loss", | |
| patience=args.patience, | |
| restore_best_weights=True, | |
| verbose=1, | |
| ) | |
| ) | |
| history = model.fit( | |
| {"tokens": contexts, "mode": modes}, | |
| targets, | |
| sample_weight=sample_weights, | |
| batch_size=args.batch_size, | |
| epochs=args.epochs, | |
| validation_split=args.validation_split, | |
| callbacks=callbacks, | |
| shuffle=True, | |
| verbose=2, | |
| ) | |
| model_dir.mkdir(parents=True, exist_ok=True) | |
| model.save(model_path) | |
| note_counts = [ | |
| 0 if token in SPECIAL_TOKENS else count_notes(token) for token in vocabulary | |
| ] | |
| control_ids = [ | |
| index for index, token in enumerate(vocabulary) if is_control_token(token) | |
| ] | |
| clef_ids = [ | |
| index | |
| for index, token in enumerate(vocabulary) | |
| if re.fullmatch(r"[cf]b?[1-4]", token) | |
| ] | |
| metadata = { | |
| "format_version": 1, | |
| "model_type": "mode-conditioned GABC music-group language model", | |
| "keras_version": keras.__version__, | |
| "keras_backend": keras.backend.backend(), | |
| "corpus_directory": str(gabc_dir), | |
| "corpus_file_count": len(paths), | |
| "corpus_sha256": corpus_digest, | |
| "trained_score_count": len(sequences), | |
| "score_modes": {str(mode): count for mode, count in sorted(score_modes.items())}, | |
| "training_example_modes": { | |
| str(mode): count for mode, count in sorted(example_modes.items()) | |
| }, | |
| "total_music_groups": total_tokens, | |
| "direct_vocabulary_coverage": represented / total_tokens, | |
| "context_length": args.context, | |
| "vocabulary": vocabulary, | |
| "note_counts": note_counts, | |
| "control_token_ids": control_ids, | |
| "clef_token_ids": clef_ids, | |
| "special_token_ids": { | |
| "pad": PAD, | |
| "unknown": UNK, | |
| "start": START, | |
| "end": END, | |
| }, | |
| "training": { | |
| "epochs_requested": args.epochs, | |
| "epochs_completed": len(history.history.get("loss", [])), | |
| "batch_size": args.batch_size, | |
| "embedding": args.embedding, | |
| "mode_embedding": args.mode_embedding, | |
| "units": args.units, | |
| "dropout": args.dropout, | |
| "validation_split": args.validation_split, | |
| "learning_rate": args.learning_rate, | |
| "min_frequency": args.min_frequency, | |
| "max_vocab": args.max_vocab, | |
| "seed": args.seed, | |
| "history": { | |
| key: [float(value) for value in values] | |
| for key, values in history.history.items() | |
| }, | |
| }, | |
| } | |
| temporary_metadata = metadata_path.with_suffix(".json.tmp") | |
| with temporary_metadata.open("w", encoding="utf-8") as handle: | |
| json.dump(metadata, handle, ensure_ascii=False, indent=2) | |
| handle.write("\n") | |
| os.replace(temporary_metadata, metadata_path) | |
| print(f"Saved Keras model: {model_path}") | |
| print(f"Saved vocabulary and corpus metadata: {metadata_path}") | |
| print(f"Next: {SCRIPT_DIR / 'generative.zsh'} {model_dir}") | |
| if __name__ == "__main__": | |
| main() | |
| PYTHON |
| #!/usr/bin/env zsh | |
| # Generate complete GABC melodies from the mode-conditioned Keras model trained | |
| # by generate.zsh. By default this writes 10 distinct 50-note melodies for each | |
| # of the eight Gregorian modes (80 standalone .gabc files plus a manifest). | |
| # Music is emitted as bare parenthesized GABC glyph groups with no lyric text. | |
| # | |
| # Usage: | |
| # ./generative.zsh [options] [model-directory] | |
| # | |
| # Run with --help for sampling, reproducibility, and output options. | |
| emulate -L zsh | |
| setopt err_exit no_unset pipe_fail | |
| readonly SCRIPT_DIR=${0:A:h} | |
| readonly PYTHON_BIN=${PYTHON:-python3} | |
| if ! command -v "$PYTHON_BIN" >/dev/null 2>&1; then | |
| print -u2 -- "error: Python interpreter not found: $PYTHON_BIN" | |
| exit 1 | |
| fi | |
| export KERAS_BACKEND=${KERAS_BACKEND:-tensorflow} | |
| typeset -i CHECK_DEPENDENCIES=1 | |
| for argument in "$@"; do | |
| if [[ $argument == -h || $argument == --help ]]; then | |
| CHECK_DEPENDENCIES=0 | |
| break | |
| fi | |
| done | |
| if (( CHECK_DEPENDENCIES )) && ! "$PYTHON_BIN" -c 'import keras, numpy' >/dev/null 2>&1; then | |
| print -u2 -- "error: Keras and NumPy are required for generation." | |
| print -u2 -- "Install them in a virtual environment, for example:" | |
| print -u2 -- " $PYTHON_BIN -m pip install 'keras>=3' tensorflow numpy" | |
| print -u2 -- "You may also set PYTHON=/path/to/venv/bin/python." | |
| exit 1 | |
| fi | |
| exec "$PYTHON_BIN" - "$SCRIPT_DIR" "$@" <<'PYTHON' | |
| from __future__ import annotations | |
| import argparse | |
| import csv | |
| import json | |
| import math | |
| import os | |
| import re | |
| import sys | |
| from collections import Counter | |
| from dataclasses import dataclass, field | |
| from pathlib import Path | |
| SCRIPT_DIR = Path(sys.argv.pop(1)).resolve() | |
| MODE_NAMES = { | |
| 1: "Dorian authentic", | |
| 2: "Hypodorian", | |
| 3: "Phrygian authentic", | |
| 4: "Hypophrygian", | |
| 5: "Lydian authentic", | |
| 6: "Hypolydian", | |
| 7: "Mixolydian authentic", | |
| 8: "Hypomixolydian", | |
| } | |
| CLEF_RE = re.compile(r"[cf]b?[1-4]") | |
| def positive_int(value: str) -> int: | |
| number = int(value) | |
| if number < 1: | |
| raise argparse.ArgumentTypeError("must be a positive integer") | |
| return number | |
| def nonnegative_int(value: str) -> int: | |
| number = int(value) | |
| if number < 0: | |
| raise argparse.ArgumentTypeError("must be nonnegative") | |
| return number | |
| def positive_float(value: str) -> float: | |
| number = float(value) | |
| if number <= 0: | |
| raise argparse.ArgumentTypeError("must be greater than zero") | |
| return number | |
| def nonnegative_float(value: str) -> float: | |
| number = float(value) | |
| if number < 0: | |
| raise argparse.ArgumentTypeError("must be nonnegative") | |
| return number | |
| def parse_arguments() -> argparse.Namespace: | |
| parser = argparse.ArgumentParser( | |
| prog="generative.zsh", | |
| description=( | |
| "Generate mode-conditioned GABC melodies from the Keras model " | |
| "created by generate.zsh." | |
| ), | |
| ) | |
| parser.add_argument( | |
| "model_dir", | |
| nargs="?", | |
| type=Path, | |
| default=SCRIPT_DIR / "keras-gabc-model", | |
| help="directory containing model.keras and metadata.json (default: %(default)s)", | |
| ) | |
| parser.add_argument( | |
| "-o", | |
| "--output-dir", | |
| type=Path, | |
| default=SCRIPT_DIR / "generated-gabc", | |
| help="output directory (default: %(default)s)", | |
| ) | |
| parser.add_argument( | |
| "-n", | |
| "--count-per-mode", | |
| type=positive_int, | |
| default=10, | |
| help="melodies generated for each mode (default: 10)", | |
| ) | |
| parser.add_argument( | |
| "--notes", | |
| type=positive_int, | |
| default=50, | |
| help="exact sung-note count of every melody (default: 50)", | |
| ) | |
| parser.add_argument("--temperature", type=positive_float, default=0.90) | |
| parser.add_argument("--top-k", type=positive_int, default=40) | |
| parser.add_argument( | |
| "--repetition-penalty", | |
| type=nonnegative_float, | |
| default=0.35, | |
| help="log-probability penalty per previous use of a glyph (default: 0.35)", | |
| ) | |
| parser.add_argument( | |
| "--min-unique-glyphs", | |
| type=positive_int, | |
| default=6, | |
| help="reject and resample melodies with less glyph variety (default: 6)", | |
| ) | |
| parser.add_argument( | |
| "--min-complex-neumes", | |
| type=nonnegative_int, | |
| default=6, | |
| help="minimum connected neumes of at least three notes per melody (default: 6)", | |
| ) | |
| parser.add_argument( | |
| "--min-torculus", | |
| type=nonnegative_int, | |
| default=3, | |
| help="minimum three-note rising-then-falling torculi per melody (default: 3)", | |
| ) | |
| parser.add_argument( | |
| "--complexity-bias", | |
| type=nonnegative_float, | |
| default=1.0, | |
| help="log-probability bonus for learned complex neumes (default: 1.0)", | |
| ) | |
| parser.add_argument( | |
| "--torculus-bias", | |
| type=nonnegative_float, | |
| default=1.25, | |
| help="additional log-probability bonus for learned torculi (default: 1.25)", | |
| ) | |
| parser.add_argument( | |
| "--line-notes", | |
| type=positive_int, | |
| default=10, | |
| help="approximate sung notes per engraved staff (default: 10)", | |
| ) | |
| parser.add_argument("--seed", type=int, default=20260831) | |
| parser.add_argument( | |
| "--max-rounds", | |
| type=positive_int, | |
| default=20, | |
| help="maximum diversity/duplicate resampling rounds per mode (default: 20)", | |
| ) | |
| parser.add_argument( | |
| "--overwrite", | |
| action="store_true", | |
| help="replace only the melody files and manifest this run would create", | |
| ) | |
| args = parser.parse_args() | |
| required_complex_notes = 3 * max(args.min_complex_neumes, args.min_torculus) | |
| if required_complex_notes > args.notes: | |
| parser.error( | |
| "complex-neume quotas are impossible for the requested note count: " | |
| f"need at least {required_complex_notes} notes" | |
| ) | |
| return args | |
| @dataclass | |
| class GenerationState: | |
| seed: int | |
| rng: np.random.Generator | |
| ids: list[int] | |
| emitted: list[int] = field(default_factory=list) | |
| note_total: int = 0 | |
| control_streak: int = 0 | |
| complex_neumes: int = 0 | |
| torculi: int = 0 | |
| failed: bool = False | |
| def pitch_components(token: str) -> list[list[int]]: | |
| """Extract connected pitch sequences, respecting GABC neume boundaries.""" | |
| components: list[list[int]] = [] | |
| current: list[int] = [] | |
| def finish_component() -> None: | |
| nonlocal current | |
| if current: | |
| components.append(current) | |
| current = [] | |
| index = 0 | |
| while index < len(token): | |
| clef = CLEF_RE.match(token, index) | |
| if clef: | |
| finish_component() | |
| index = clef.end() | |
| continue | |
| character = token[index] | |
| # Slash is a neumatic cut; ! is a zero-width glyph boundary. Bars also | |
| # terminate any connected note figure. @ is deliberately not included | |
| # because GABC uses it to fuse notes into one figure. | |
| if character in "/!,:;": | |
| finish_component() | |
| index += 1 | |
| continue | |
| if character not in "abcdefghijklmABCDEFGHIJKLM": | |
| index += 1 | |
| continue | |
| following = token[index + 1 :] | |
| if following[:1] in {"x", "y", "#", "+"}: | |
| index += 2 | |
| continue | |
| pitch = ord(character.lower()) - ord("a") | |
| repetitions = re.match(r"[sv]+", following) | |
| if repetitions: | |
| current.extend([pitch] * len(repetitions.group(0))) | |
| index += 1 + len(repetitions.group(0)) | |
| else: | |
| current.append(pitch) | |
| index += 1 | |
| finish_component() | |
| return components | |
| def classify_vocabulary(vocabulary: list[str]): | |
| """Count complex connected figures and true three-note torculus contours.""" | |
| complex_counts = np.zeros(len(vocabulary), dtype=np.int32) | |
| torculus_counts = np.zeros(len(vocabulary), dtype=np.int32) | |
| for token_id, token in enumerate(vocabulary): | |
| for component in pitch_components(token): | |
| if len(component) >= 3: | |
| complex_counts[token_id] += 1 | |
| if ( | |
| len(component) == 3 | |
| and component[0] < component[1] | |
| and component[1] > component[2] | |
| ): | |
| torculus_counts[token_id] += 1 | |
| return complex_counts, torculus_counts | |
| def load_artifacts(model_dir: Path): | |
| model_path = model_dir / "model.keras" | |
| metadata_path = model_dir / "metadata.json" | |
| if not model_path.is_file(): | |
| raise SystemExit(f"error: Keras model not found: {model_path}") | |
| if not metadata_path.is_file(): | |
| raise SystemExit(f"error: model metadata not found: {metadata_path}") | |
| with metadata_path.open(encoding="utf-8") as handle: | |
| metadata = json.load(handle) | |
| if metadata.get("format_version") != 1: | |
| raise SystemExit( | |
| f"error: unsupported metadata format: {metadata.get('format_version')!r}" | |
| ) | |
| model = keras.models.load_model(model_path, compile=False) | |
| return model, metadata, model_path | |
| def validate_metadata(model: keras.Model, metadata: dict): | |
| vocabulary = metadata.get("vocabulary") | |
| note_counts = metadata.get("note_counts") | |
| special = metadata.get("special_token_ids", {}) | |
| if not isinstance(vocabulary, list) or not vocabulary: | |
| raise SystemExit("error: metadata has no vocabulary") | |
| if not isinstance(note_counts, list) or len(note_counts) != len(vocabulary): | |
| raise SystemExit("error: metadata note_counts does not match its vocabulary") | |
| required_special = {"pad", "unknown", "start", "end"} | |
| if not required_special.issubset(special): | |
| raise SystemExit("error: metadata is missing special token IDs") | |
| if int(model.output_shape[-1]) != len(vocabulary): | |
| raise SystemExit("error: model output size does not match metadata vocabulary") | |
| context = int(metadata.get("context_length", 0)) | |
| if context < 1: | |
| raise SystemExit("error: invalid context length in metadata") | |
| clef_ids = [int(value) for value in metadata.get("clef_token_ids", [])] | |
| if not clef_ids: | |
| raise SystemExit("error: model vocabulary contains no clef tokens") | |
| control_ids = {int(value) for value in metadata.get("control_token_ids", [])} | |
| return vocabulary, np.asarray(note_counts, dtype=np.int32), special, context, clef_ids, control_ids | |
| def context_array(states: list[GenerationState], context: int, pad_id: int) -> np.ndarray: | |
| batch = np.full((len(states), context), pad_id, dtype=np.int32) | |
| for row, state in enumerate(states): | |
| previous = state.ids[-context:] | |
| batch[row, : len(previous)] = previous | |
| return batch | |
| def sample_token( | |
| probabilities: np.ndarray, | |
| allowed: np.ndarray, | |
| state: GenerationState, | |
| complex_counts: np.ndarray, | |
| torculus_counts: np.ndarray, | |
| temperature: float, | |
| top_k: int, | |
| repetition_penalty: float, | |
| complexity_bias: float, | |
| torculus_bias: float, | |
| ) -> int: | |
| candidate_ids = np.flatnonzero(allowed) | |
| if not len(candidate_ids): | |
| raise RuntimeError("sampling constraints left no eligible GABC tokens") | |
| logits = np.log(np.clip(probabilities[candidate_ids], 1e-12, 1.0)) | |
| logits /= temperature | |
| # Apply these bonuses before top-k truncation so less-frequent complex | |
| # figures can compete with the corpus's extremely common puncta. | |
| logits += complexity_bias * np.minimum(complex_counts[candidate_ids], 3) | |
| logits += torculus_bias * np.minimum(torculus_counts[candidate_ids], 2) | |
| if repetition_penalty: | |
| previous_counts = Counter(state.emitted) | |
| logits -= np.asarray( | |
| [repetition_penalty * previous_counts[int(token)] for token in candidate_ids], | |
| dtype=np.float64, | |
| ) | |
| if len(candidate_ids) > top_k: | |
| keep = np.argpartition(logits, -top_k)[-top_k:] | |
| candidate_ids = candidate_ids[keep] | |
| logits = logits[keep] | |
| logits -= float(np.max(logits)) | |
| weights = np.exp(logits) | |
| total = float(weights.sum()) | |
| if not math.isfinite(total) or total <= 0: | |
| weights = np.full(len(candidate_ids), 1.0 / len(candidate_ids)) | |
| else: | |
| weights /= total | |
| return int(state.rng.choice(candidate_ids, p=weights)) | |
| def generate_batch( | |
| model: keras.Model, | |
| mode: int, | |
| seeds: list[int], | |
| note_goal: int, | |
| context: int, | |
| note_counts: np.ndarray, | |
| complex_counts: np.ndarray, | |
| torculus_counts: np.ndarray, | |
| double_bar_tokens: np.ndarray, | |
| special: dict, | |
| clef_ids: list[int], | |
| control_ids: set[int], | |
| temperature: float, | |
| top_k: int, | |
| repetition_penalty: float, | |
| min_complex_neumes: int, | |
| min_torculus: int, | |
| complexity_bias: float, | |
| torculus_bias: float, | |
| ) -> list[GenerationState]: | |
| pad_id = int(special["pad"]) | |
| start_id = int(special["start"]) | |
| forbidden = { | |
| int(special["pad"]), | |
| int(special["unknown"]), | |
| int(special["start"]), | |
| int(special["end"]), | |
| } | |
| states = [ | |
| GenerationState(seed=seed, rng=np.random.default_rng(seed), ids=[start_id]) | |
| for seed in seeds | |
| ] | |
| max_steps = note_goal * 3 + 16 | |
| for _ in range(max_steps): | |
| active = [state for state in states if state.note_total < note_goal and not state.failed] | |
| if not active: | |
| break | |
| token_batch = context_array(active, context, pad_id) | |
| mode_batch = np.full(len(active), mode, dtype=np.int32) | |
| prediction = model( | |
| {"tokens": token_batch, "mode": mode_batch}, training=False | |
| ) | |
| prediction = np.asarray(keras.ops.convert_to_numpy(prediction)) | |
| for row, state in enumerate(active): | |
| allowed = np.zeros(len(note_counts), dtype=bool) | |
| if not state.emitted: | |
| allowed[clef_ids] = True | |
| else: | |
| remaining = note_goal - state.note_total | |
| allowed |= (note_counts > 0) & (note_counts <= remaining) | |
| # At most one non-note group in succession prevents wandering | |
| # control sequences while still allowing predicted bars/clefs. | |
| if state.control_streak == 0: | |
| for token_id in control_ids: | |
| allowed[token_id] = True | |
| for token_id in forbidden: | |
| allowed[token_id] = False | |
| # Preserve enough unspent notes to make both quotas reachable. | |
| # Torculi count toward the general complex-neume quota, so the | |
| # lower bound is the maximum of the two missing counts. | |
| complex_missing_after = np.maximum( | |
| 0, | |
| min_complex_neumes | |
| - (state.complex_neumes + complex_counts), | |
| ) | |
| torculus_missing_after = np.maximum( | |
| 0, | |
| min_torculus - (state.torculi + torculus_counts), | |
| ) | |
| minimum_future_notes = 3 * np.maximum( | |
| complex_missing_after, torculus_missing_after | |
| ) | |
| allowed &= ( | |
| (note_counts == 0) | |
| | (remaining - note_counts >= minimum_future_notes) | |
| ) | |
| # Distribute required forms through the melody. When a quota | |
| # milestone has passed, the next sung token is constrained to | |
| # a qualifying figure selected from the model distribution. | |
| torculus_deadline = ( | |
| math.floor( | |
| (state.torculi + 1) | |
| * note_goal | |
| / (min_torculus + 1) | |
| ) | |
| if state.torculi < min_torculus | |
| else note_goal + 1 | |
| ) | |
| complex_deadline = ( | |
| math.floor( | |
| (state.complex_neumes + 1) | |
| * note_goal | |
| / (min_complex_neumes + 1) | |
| ) | |
| if state.complex_neumes < min_complex_neumes | |
| else note_goal + 1 | |
| ) | |
| if state.note_total >= torculus_deadline: | |
| forced = allowed & (torculus_counts > 0) | |
| if np.any(forced): | |
| allowed = forced | |
| elif state.note_total >= complex_deadline: | |
| forced = allowed & (complex_counts > 0) | |
| if np.any(forced): | |
| allowed = forced | |
| # The renderer appends exactly one final (::); never sample a token | |
| # containing another double bar within the melody. | |
| allowed[double_bar_tokens] = False | |
| try: | |
| token_id = sample_token( | |
| prediction[row], | |
| allowed, | |
| state, | |
| complex_counts, | |
| torculus_counts, | |
| temperature, | |
| top_k, | |
| repetition_penalty, | |
| complexity_bias, | |
| torculus_bias, | |
| ) | |
| except RuntimeError: | |
| state.failed = True | |
| continue | |
| state.ids.append(token_id) | |
| state.emitted.append(token_id) | |
| token_notes = int(note_counts[token_id]) | |
| state.note_total += token_notes | |
| state.control_streak = 0 if token_notes else state.control_streak + 1 | |
| state.complex_neumes += int(complex_counts[token_id]) | |
| state.torculi += int(torculus_counts[token_id]) | |
| for state in states: | |
| if ( | |
| state.note_total != note_goal | |
| or state.complex_neumes < min_complex_neumes | |
| or state.torculi < min_torculus | |
| ): | |
| state.failed = True | |
| return states | |
| def render_gabc( | |
| mode: int, | |
| number: int, | |
| state: GenerationState, | |
| vocabulary: list[str], | |
| note_counts: np.ndarray, | |
| note_goal: int, | |
| line_notes: int, | |
| temperature: float, | |
| top_k: int, | |
| ) -> str: | |
| body_parts: list[str] = [] | |
| running_notes = 0 | |
| next_break = line_notes | |
| for token_id in state.emitted: | |
| token = vocabulary[token_id] | |
| token_notes = int(note_counts[token_id]) | |
| body_parts.append(f"({token})") | |
| if token_notes: | |
| running_notes += token_notes | |
| if running_notes >= next_break and running_notes < note_goal: | |
| body_parts.append("(z)\n") | |
| while next_break <= running_notes: | |
| next_break += line_notes | |
| body_parts.append("(::)") | |
| body = " ".join(body_parts) | |
| # Do not leave a space before physical newlines introduced with (z). | |
| body = body.replace("(z)\n ", "(z)\n") | |
| return ( | |
| f"name:Keras melody mode {mode}, number {number:02d};\n" | |
| f"mode:{mode};\n" | |
| "initial-style:0;\n" | |
| f"commentary:{note_goal} notes; mode-conditioned sampling; " | |
| f"{state.complex_neumes} complex neumes; {state.torculi} torculi; " | |
| f"temperature {temperature:g}; top-k {top_k}; seed {state.seed};\n" | |
| "%%\n" | |
| f"{body}\n" | |
| ) | |
| def main() -> None: | |
| args = parse_arguments() | |
| # Delay heavyweight imports so --help works even before runtime setup. | |
| global keras, np | |
| import keras | |
| import numpy as np | |
| if int(keras.__version__.split(".", 1)[0]) < 3: | |
| raise SystemExit("error: Keras 3 or newer is required") | |
| model_dir = args.model_dir.expanduser().resolve() | |
| output_dir = args.output_dir.expanduser().resolve() | |
| model, metadata, model_path = load_artifacts(model_dir) | |
| ( | |
| vocabulary, | |
| note_counts, | |
| special, | |
| context, | |
| clef_ids, | |
| control_ids, | |
| ) = validate_metadata(model, metadata) | |
| complex_counts, torculus_counts = classify_vocabulary(vocabulary) | |
| double_bar_tokens = np.asarray( | |
| ["::" in token for token in vocabulary], dtype=bool | |
| ) | |
| if args.min_complex_neumes and not np.any(complex_counts): | |
| raise SystemExit("error: model vocabulary contains no complex neumes") | |
| if args.min_torculus and not np.any(torculus_counts): | |
| raise SystemExit("error: model vocabulary contains no torculus figures") | |
| planned_files = [ | |
| output_dir / f"mode-{mode}-{number:02d}.gabc" | |
| for mode in range(1, 9) | |
| for number in range(1, args.count_per_mode + 1) | |
| ] | |
| manifest_path = output_dir / "manifest.tsv" | |
| existing = [path for path in planned_files + [manifest_path] if path.exists()] | |
| if existing and not args.overwrite: | |
| preview = ", ".join(str(path) for path in existing[:3]) | |
| more = f" (and {len(existing) - 3} more)" if len(existing) > 3 else "" | |
| raise SystemExit( | |
| f"error: generation output already exists: {preview}{more}; use --overwrite" | |
| ) | |
| print(f"Loaded model: {model_path}") | |
| print( | |
| f"Generating {args.count_per_mode} × 8 mode-conditioned melodies, " | |
| f"exactly {args.notes} notes each." | |
| ) | |
| accepted_by_mode: dict[int, list[GenerationState]] = {} | |
| all_signatures: set[tuple[int, ...]] = set() | |
| for mode in range(1, 9): | |
| accepted: list[GenerationState] = [] | |
| candidate_serial = 0 | |
| for _round in range(args.max_rounds): | |
| needed = args.count_per_mode - len(accepted) | |
| if needed <= 0: | |
| break | |
| # Generate a few spare candidates after the first rejection so | |
| # diversity checks do not force unnecessary single-item batches. | |
| batch_size = needed if candidate_serial == 0 else max(needed, 2) | |
| seeds = [] | |
| for _ in range(batch_size): | |
| candidate_serial += 1 | |
| seeds.append(args.seed + mode * 1_000_000 + candidate_serial) | |
| candidates = generate_batch( | |
| model=model, | |
| mode=mode, | |
| seeds=seeds, | |
| note_goal=args.notes, | |
| context=context, | |
| note_counts=note_counts, | |
| complex_counts=complex_counts, | |
| torculus_counts=torculus_counts, | |
| double_bar_tokens=double_bar_tokens, | |
| special=special, | |
| clef_ids=clef_ids, | |
| control_ids=control_ids, | |
| temperature=args.temperature, | |
| top_k=args.top_k, | |
| repetition_penalty=args.repetition_penalty, | |
| min_complex_neumes=args.min_complex_neumes, | |
| min_torculus=args.min_torculus, | |
| complexity_bias=args.complexity_bias, | |
| torculus_bias=args.torculus_bias, | |
| ) | |
| for state in candidates: | |
| if state.failed: | |
| continue | |
| glyph_ids = [ | |
| token_id for token_id in state.emitted if note_counts[token_id] > 0 | |
| ] | |
| if len(set(glyph_ids)) < args.min_unique_glyphs: | |
| continue | |
| if state.complex_neumes < args.min_complex_neumes: | |
| continue | |
| if state.torculi < args.min_torculus: | |
| continue | |
| signature = tuple(state.emitted) | |
| if signature in all_signatures: | |
| continue | |
| all_signatures.add(signature) | |
| accepted.append(state) | |
| if len(accepted) == args.count_per_mode: | |
| break | |
| if len(accepted) != args.count_per_mode: | |
| raise SystemExit( | |
| f"error: mode {mode} produced only {len(accepted)} diverse melodies " | |
| f"after {args.max_rounds} rounds; lower a minimum quota or " | |
| "raise --max-rounds" | |
| ) | |
| accepted_by_mode[mode] = accepted | |
| print(f" Mode {mode} ({MODE_NAMES[mode]}): {len(accepted)} accepted") | |
| output_dir.mkdir(parents=True, exist_ok=True) | |
| manifest_rows = [] | |
| for mode in range(1, 9): | |
| for number, state in enumerate(accepted_by_mode[mode], start=1): | |
| filename = f"mode-{mode}-{number:02d}.gabc" | |
| destination = output_dir / filename | |
| gabc = render_gabc( | |
| mode, | |
| number, | |
| state, | |
| vocabulary, | |
| note_counts, | |
| args.notes, | |
| args.line_notes, | |
| args.temperature, | |
| args.top_k, | |
| ) | |
| temporary = destination.with_suffix(".gabc.tmp") | |
| temporary.write_text(gabc, encoding="utf-8") | |
| os.replace(temporary, destination) | |
| glyph_ids = [ | |
| token_id for token_id in state.emitted if note_counts[token_id] > 0 | |
| ] | |
| clef = next( | |
| ( | |
| vocabulary[token_id] | |
| for token_id in state.emitted | |
| if CLEF_RE.fullmatch(vocabulary[token_id]) | |
| ), | |
| "unknown", | |
| ) | |
| manifest_rows.append( | |
| { | |
| "file": filename, | |
| "mode": mode, | |
| "mode_name": MODE_NAMES[mode], | |
| "notes": state.note_total, | |
| "glyph_groups": len(glyph_ids), | |
| "unique_glyphs": len(set(glyph_ids)), | |
| "complex_neumes": state.complex_neumes, | |
| "torculi": state.torculi, | |
| "initial_clef": clef, | |
| "seed": state.seed, | |
| } | |
| ) | |
| temporary_manifest = manifest_path.with_suffix(".tsv.tmp") | |
| with temporary_manifest.open("w", encoding="utf-8", newline="") as handle: | |
| writer = csv.DictWriter( | |
| handle, | |
| fieldnames=[ | |
| "file", | |
| "mode", | |
| "mode_name", | |
| "notes", | |
| "glyph_groups", | |
| "unique_glyphs", | |
| "complex_neumes", | |
| "torculi", | |
| "initial_clef", | |
| "seed", | |
| ], | |
| delimiter="\t", | |
| lineterminator="\n", | |
| ) | |
| writer.writeheader() | |
| writer.writerows(manifest_rows) | |
| os.replace(temporary_manifest, manifest_path) | |
| print(f"Wrote {len(manifest_rows)} GABC melodies to: {output_dir}") | |
| print(f"Manifest: {manifest_path}") | |
| if __name__ == "__main__": | |
| main() | |
| PYTHON |