@@ -0,0 +1,125 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sqlite3
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
APPLICATION_ID = 0x56414D50
|
||||
SCHEMA_VERSION = 1
|
||||
COMMON_COLUMNS = """
|
||||
id TEXT PRIMARY KEY NOT NULL CHECK(length(trim(id)) BETWEEN 1 AND 160),
|
||||
title TEXT NOT NULL CHECK(length(trim(title)) BETWEEN 1 AND 200),
|
||||
composer TEXT NOT NULL,
|
||||
key_signature TEXT NOT NULL,
|
||||
authored_key_signature TEXT NOT NULL,
|
||||
source_path TEXT NOT NULL UNIQUE CHECK(length(source_path) > 0),
|
||||
source_json TEXT NOT NULL CHECK(json_valid(source_json)),
|
||||
measure_count INTEGER NOT NULL CHECK(measure_count > 0),
|
||||
total_pages INTEGER NOT NULL CHECK(total_pages BETWEEN 1 AND 10000)
|
||||
"""
|
||||
|
||||
def human_label(value: str) -> str:
|
||||
result, capitalize = '', True
|
||||
for character in value:
|
||||
if character in '-_':
|
||||
result += ' '
|
||||
capitalize = True
|
||||
else:
|
||||
result += character.upper() if capitalize and 'a' <= character <= 'z' else character
|
||||
capitalize = False
|
||||
return result or 'Score'
|
||||
|
||||
def metadata(path: Path, root: Path) -> dict:
|
||||
if path.is_symlink() or not path.resolve().is_relative_to(root.resolve()):
|
||||
raise ValueError('score escapes the asset root')
|
||||
text = path.read_text(encoding='utf-8')
|
||||
document = json.loads(text)
|
||||
if document.get('schema_version') != 3:
|
||||
raise ValueError('expected a schema-v3 score bundle')
|
||||
music, presentation = document['music'], document['presentation']
|
||||
if len(music['staves']) != 1 or (presentation['width'], presentation['height']) != (976, 484):
|
||||
raise ValueError('expected one staff and 976x484 pages')
|
||||
staff, meta = music['staves'][0], music['metadata']
|
||||
for name in ('title', 'composer', 'key_signature', 'time_signature'):
|
||||
if not isinstance(meta[name], str):
|
||||
raise ValueError(f'{name} must be a string')
|
||||
authored = meta.get('authored_key_signature', '')
|
||||
if not isinstance(authored, str) or ('authored_key_signature' in meta and not authored):
|
||||
raise ValueError('authored_key_signature must be a nonempty string')
|
||||
key = meta['key_signature']
|
||||
if staff.get('arrangement_part_type') == 'practice' and key == 'No key' and 'variant_tonic' in meta:
|
||||
if not isinstance(meta['variant_tonic'], str):
|
||||
raise ValueError('variant_tonic must be a string')
|
||||
key = meta['variant_tonic'] + ' (no key)'
|
||||
if not isinstance(staff['name'], str):
|
||||
raise ValueError('staff name must be a string')
|
||||
relative = path.relative_to(root)
|
||||
return dict(id=path.stem, title=meta['title'], composer=meta['composer'], key_signature=key,
|
||||
authored_key_signature=authored, source_path=relative.as_posix(), source_json=text,
|
||||
measure_count=len(staff['measures']), total_pages=len(presentation['pages']), part=staff['name'])
|
||||
|
||||
def populate(connection: sqlite3.Connection, root: Path) -> None:
|
||||
if not root.is_dir():
|
||||
raise ValueError(f'{root}: score asset directory not found')
|
||||
connection.executescript(f'''
|
||||
PRAGMA application_id = {APPLICATION_ID};
|
||||
PRAGMA user_version = {SCHEMA_VERSION};
|
||||
CREATE TABLE songs ({COMMON_COLUMNS}, part TEXT NOT NULL) STRICT;
|
||||
CREATE TABLE practice_exercises ({COMMON_COLUMNS},
|
||||
exercise_id TEXT NOT NULL,
|
||||
practice_type_id TEXT NOT NULL, practice_type_title TEXT NOT NULL,
|
||||
lesson_group_id TEXT NOT NULL, lesson_group_title TEXT NOT NULL) STRICT;
|
||||
CREATE INDEX songs_menu ON songs(title COLLATE NOCASE, composer COLLATE NOCASE, part COLLATE NOCASE);
|
||||
CREATE INDEX practice_menu ON practice_exercises(practice_type_id COLLATE NOCASE, lesson_group_id COLLATE NOCASE, source_path COLLATE NOCASE);
|
||||
''')
|
||||
identities, authored_keys = {}, {}
|
||||
for category, table, depth in (('songs', 'songs', 1), ('practice', 'practice_exercises', 4)):
|
||||
folder = root / category
|
||||
if (folder.exists() or folder.is_symlink()) and not folder.is_dir():
|
||||
raise ValueError(f'{folder}: score asset directory not found')
|
||||
for path in sorted((root / category).rglob('*.json')):
|
||||
try:
|
||||
parts = path.relative_to(root / category).parts
|
||||
if len(parts) != depth:
|
||||
raise ValueError(f'unexpected {category} score path')
|
||||
row = metadata(path, root)
|
||||
if category == 'practice':
|
||||
identity = '-'.join(parts[:3])
|
||||
previous = identities.setdefault(identity, path.parent)
|
||||
if previous != path.parent:
|
||||
raise ValueError(f'practice exercise ID collision {identity!r}: {previous} and {path.parent}')
|
||||
if authored_keys.setdefault(identity, row['authored_key_signature']) != row['authored_key_signature']:
|
||||
raise ValueError(f'practice variants disagree on authored key: {identity}')
|
||||
row.pop('part')
|
||||
row.update(id=identity + '-' + path.stem, exercise_id=identity, practice_type_id=parts[0], practice_type_title=human_label(parts[0]) + ' Practice',
|
||||
lesson_group_id=parts[1], lesson_group_title=human_label(parts[1]))
|
||||
connection.execute(f'INSERT INTO {table} ({", ".join(row)}) VALUES ({", ".join("?" for _ in row)})', tuple(row.values()))
|
||||
except (ValueError, KeyError, TypeError, sqlite3.Error) as error:
|
||||
raise ValueError(f'{path}: {error}') from error
|
||||
connection.commit()
|
||||
if connection.execute('PRAGMA integrity_check').fetchall() != [('ok',)]:
|
||||
raise ValueError('catalogue integrity check failed')
|
||||
|
||||
def build_catalogue(root: Path, destination: Path | None = None) -> Path:
|
||||
if not root.is_dir():
|
||||
raise ValueError(f'{root}: score asset directory not found')
|
||||
destination = destination or root / 'catalogue.db'
|
||||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||||
with tempfile.TemporaryDirectory(prefix='.vamp-catalogue-', dir=destination.parent) as temporary:
|
||||
staged = Path(temporary) / 'catalogue.db'
|
||||
connection = sqlite3.connect(staged)
|
||||
try:
|
||||
populate(connection, root)
|
||||
finally:
|
||||
connection.close()
|
||||
staged.replace(destination)
|
||||
return destination
|
||||
|
||||
if __name__ == '__main__':
|
||||
parser = argparse.ArgumentParser(description='Compile generated score bundles into a read-only runtime catalogue.')
|
||||
parser.add_argument('assets', type=Path)
|
||||
parser.add_argument('destination', type=Path, nargs='?')
|
||||
arguments = parser.parse_args()
|
||||
build_catalogue(arguments.assets, arguments.destination)
|
||||
Reference in New Issue
Block a user