khodowany
105 lines
5.1 KiB
Python
105 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sqlite3
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
APPLICATION_ID = 0x56414D50
|
|
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,
|
|
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 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 not isinstance(document, dict) or not isinstance(document.get('music'), dict) or not isinstance(document.get('presentation'), dict):
|
|
raise ValueError('score bundle requires music and presentation objects')
|
|
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')
|
|
key = meta['key_signature']
|
|
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,
|
|
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};
|
|
CREATE TABLE songs ({COMMON_COLUMNS}, part TEXT NOT NULL) STRICT;
|
|
CREATE TABLE practice_exercises ({COMMON_COLUMNS}, exercise_id TEXT NOT NULL, part 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(exercise_id, id);
|
|
''')
|
|
practice_parts = {}
|
|
for category, table, depth in (('songs', 'songs', 1), ('practice', 'practice_exercises', 2)):
|
|
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 = parts[0]
|
|
if path.stem not in ('acoustic', 'electric', 'vocal') or row['part'] != path.stem.capitalize():
|
|
raise ValueError('practice requires an acoustic, electric, or vocal part')
|
|
identity_metadata = (row['title'], row['composer'], row['key_signature'])
|
|
group = practice_parts.setdefault(identity, (identity_metadata, set()))
|
|
if group[0] != identity_metadata:
|
|
raise ValueError(f'practice parts disagree on metadata: {identity}')
|
|
group[1].add(path.stem)
|
|
row.update(id=identity + '-' + path.stem, exercise_id=identity)
|
|
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
|
|
for identity, (_, parts) in practice_parts.items():
|
|
if parts != {'acoustic', 'electric', 'vocal'}:
|
|
raise ValueError(f'practice exercise requires all three parts: {identity}')
|
|
connection.commit()
|
|
if connection.execute('PRAGMA integrity_check').fetchall() != [('ok',)]:
|
|
raise ValueError('catalog integrity check failed')
|
|
|
|
def build_catalog(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 / 'catalog.db'
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
with tempfile.TemporaryDirectory(prefix='.vamp-catalog-', dir=destination.parent) as temporary:
|
|
staged = Path(temporary) / 'catalog.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 catalog.')
|
|
parser.add_argument('assets', type=Path)
|
|
parser.add_argument('destination', type=Path, nargs='?')
|
|
arguments = parser.parse_args()
|
|
build_catalog(arguments.assets, arguments.destination)
|