tools/assemble_sound.py

   1 #!/usr/bin/env python3
   2 from collections import namedtuple, OrderedDict
   3 from json import JSONDecoder
   4 import os
   5 import re
   6 import struct
   7 import subprocess
   8 import sys
   9
  10 TYPE_CTL = 1
  11 TYPE_TBL = 2
  12
  13 STACK_TRACES = False
  14 DUMP_INDIVIDUAL_BINS = False
  15 ENDIAN_MARKER = ">"
  16 WORD_BYTES = 4
  17
  18 orderedJsonDecoder = JSONDecoder(object_pairs_hook=OrderedDict)
  19
  20
  21 class Aifc:
  22     def __init__(self, name, fname, data, sample_rate, book, loop):
  23         self.name = name
  24         self.fname = fname
  25         self.data = data
  26         self.sample_rate = sample_rate
  27         self.book = book
  28         self.loop = loop
  29         self.used = False
  30         self.offset = None
  31
  32
  33 class SampleBank:
  34     def __init__(self, name, entries):
  35         self.name = name
  36         self.uses = []
  37         self.entries = entries
  38         self.name_to_entry = {}
  39         for e in entries:
  40             self.name_to_entry[e.name] = e
  41
  42
  43 Book = namedtuple("Book", ["order", "npredictors", "table"])
  44 Loop = namedtuple("Loop", ["start", "end", "count", "state"])
  45 Bank = namedtuple("Bank", ["name", "sample_bank", "json"])
  46
  47
  48 def align(val, al):
  49     return (val + (al - 1)) & -al
  50
  51
  52 def fail(msg):
  53     print(msg, file=sys.stderr)
  54     if STACK_TRACES:
  55         raise Exception("re-raising exception")
  56     sys.exit(1)
  57
  58
  59 def validate(cond, msg, forstr=""):
  60     if not cond:
  61         if forstr:
  62             msg += " for " + forstr
  63         raise Exception(msg)
  64
  65
  66 def strip_comments(string):
  67     string = re.sub(re.compile("/\*.*?\*/", re.DOTALL), "", string)
  68     return re.sub(re.compile("//.*?\n"), "", string)
  69
  70
  71 def pack(fmt, *args):
  72     if WORD_BYTES == 4:
  73         fmt = fmt.replace('P', 'I').replace('X', '')
  74     else:
  75         fmt = fmt.replace('P', 'Q').replace('X', 'xxxx')
  76     return struct.pack(ENDIAN_MARKER + fmt, *args)
  77
  78
  79 def to_bcd(num):
  80     assert num >= 0
  81     shift = 0
  82     ret = 0
  83     while num:
  84         ret |= (num % 10) << shift
  85         shift += 4
  86         num //= 10
  87     return ret
  88
  89
  90 def parse_f80(data):
  91     exp_bits, mantissa_bits = struct.unpack(">HQ", data)
  92     sign_bit = exp_bits & 2 ** 15
  93     exp_bits ^= sign_bit
  94     sign = -1 if sign_bit else 1
  95     if exp_bits == mantissa_bits == 0:
  96         return sign * 0.0
  97     validate(exp_bits != 0, "sample rate is a denormal")
  98     validate(exp_bits != 0x7FFF, "sample rate is infinity/nan")
  99     mant = float(mantissa_bits) / 2 ** 63
 100     return sign * mant * pow(2, exp_bits - 0x3FFF)
 101
 102
 103 def parse_aifc_loop(data):
 104     validate(len(data) == 48, "loop chunk size should be 48")
 105     version, nloops, start, end, count = struct.unpack(">HHIIi", data[:16])
 106     validate(version == 1, "loop version doesn't match")
 107     validate(nloops == 1, "only one loop is supported")
 108     state = []
 109     for i in range(16, len(data), 2):
 110         state.append(struct.unpack(">h", data[i : i + 2])[0])
 111     return Loop(start, end, count, state)
 112
 113
 114 def parse_aifc_book(data):
 115     version, order, npredictors = struct.unpack(">hhh", data[:6])
 116     validate(version == 1, "codebook version doesn't match")
 117     validate(
 118         len(data) == 6 + 16 * order * npredictors,
 119         "predictor book chunk size doesn't match",
 120     )
 121     table = []
 122     for i in range(6, len(data), 2):
 123         table.append(struct.unpack(">h", data[i : i + 2])[0])
 124     return Book(order, npredictors, table)
 125
 126
 127 def parse_aifc(data, name, fname):
 128     validate(data[:4] == b"FORM", "must start with FORM")
 129     validate(data[8:12] == b"AIFC", "format must be AIFC")
 130     i = 12
 131     sections = []
 132     while i < len(data):
 133         tp = data[i : i + 4]
 134         le, = struct.unpack(">I", data[i + 4 : i + 8])
 135         i += 8
 136         sections.append((tp, data[i : i + le]))
 137         i = align(i + le, 2)
 138
 139     audio_data = None
 140     vadpcm_codes = None
 141     vadpcm_loops = None
 142     sample_rate = None
 143
 144     for (tp, data) in sections:
 145         if tp == b"APPL" and data[:4] == b"stoc":
 146             plen = data[4]
 147             tp = data[5 : 5 + plen]
 148             data = data[align(5 + plen, 2) :]
 149             if tp == b"VADPCMCODES":
 150                 vadpcm_codes = data
 151             elif tp == b"VADPCMLOOPS":
 152                 vadpcm_loops = data
 153         elif tp == b"SSND":
 154             audio_data = data[8:]
 155         elif tp == b"COMM":
 156             sample_rate = parse_f80(data[8:18])
 157
 158     validate(sample_rate is not None, "no COMM section")
 159     validate(audio_data is not None, "no SSND section")
 160     validate(vadpcm_codes is not None, "no VADPCM table")
 161
 162     book = parse_aifc_book(vadpcm_codes)
 163     loop = parse_aifc_loop(vadpcm_loops) if vadpcm_loops is not None else None
 164     return Aifc(name, fname, audio_data, sample_rate, book, loop)
 165
 166
 167 class ReserveSerializer:
 168     def __init__(self):
 169         self.parts = []
 170         self.sizes = []
 171         self.size = 0
 172
 173     def add(self, part):
 174         assert isinstance(part, (bytes, list))
 175         self.parts.append(part)
 176         self.sizes.append(len(part))
 177         self.size += len(part)
 178
 179     def reserve(self, space):
 180         li = []
 181         self.parts.append(li)
 182         self.sizes.append(space)
 183         self.size += space
 184         return li
 185
 186     def align(self, alignment):
 187         new_size = (self.size + alignment - 1) & -alignment
 188         self.add((new_size - self.size) * b"\0")
 189
 190     def finish(self):
 191         flat_parts = []
 192         for (li, si) in zip(self.parts, self.sizes):
 193             if isinstance(li, list):
 194                 li = b"".join(li)
 195             assert (
 196                 len(li) == si
 197             ), "unfulfilled reservation of size {}, only got {}".format(si, len(li))
 198             flat_parts.append(li)
 199         return b"".join(flat_parts)
 200
 201
 202 class GarbageSerializer:
 203     def __init__(self):
 204         self.garbage_bufs = [[]]
 205         self.parts = []
 206         self.size = 0
 207         self.garbage_pos = 0
 208
 209     def reset_garbage_pos(self):
 210         self.garbage_bufs.append([])
 211         self.garbage_pos = 0
 212
 213     def add(self, part):
 214         assert isinstance(part, bytes)
 215         self.parts.append(part)
 216         self.garbage_bufs[-1].append((self.garbage_pos, part))
 217         self.size += len(part)
 218         self.garbage_pos += len(part)
 219
 220     def align(self, alignment):
 221         new_size = (self.size + alignment - 1) & -alignment
 222         self.add((new_size - self.size) * b"\0")
 223
 224     def garbage_at(self, pos):
 225         # Find the last write to position pos & 0xffff, assuming a cyclic
 226         # buffer of size 0x10000 where the write position is reset to 0 on
 227         # each call to reset_garbage_pos.
 228         pos &= 0xFFFF
 229         for bufs in self.garbage_bufs[::-1]:
 230             for (bpos, buf) in bufs[::-1]:
 231                 q = ((bpos + len(buf) - 1 - pos) & ~0xFFFF) + pos
 232                 if q >= bpos:
 233                     return buf[q - bpos]
 234         return 0
 235
 236     def align_garbage(self, alignment):
 237         while self.size % alignment != 0:
 238             self.add(bytes([self.garbage_at(self.garbage_pos)]))
 239
 240     def finish(self):
 241         return b"".join(self.parts)
 242
 243
 244 def validate_json_format(json, fmt, forstr=""):
 245     constructor_to_name = {
 246         str: "a string",
 247         dict: "an object",
 248         int: "an integer",
 249         float: "a floating point number",
 250         list: "an array",
 251     }
 252     for key, tp in fmt.items():
 253         validate(key in json, 'missing key "' + key + '"', forstr)
 254         if isinstance(tp, list):
 255             validate_int_in_range(json[key], tp[0], tp[1], '"' + key + '"', forstr)
 256         else:
 257             validate(
 258                 isinstance(json[key], tp)
 259                 or (tp == float and isinstance(json[key], int)),
 260                 '"{}" must be {}'.format(key, constructor_to_name[tp]),
 261                 forstr,
 262             )
 263
 264
 265 def validate_int_in_range(val, lo, hi, msg, forstr=""):
 266     validate(isinstance(val, int), "{} must be an integer".format(msg), forstr)
 267     validate(
 268         lo <= val <= hi, "{} must be in range {} to {}".format(msg, lo, hi), forstr
 269     )
 270
 271
 272 def validate_sound(json, sample_bank, forstr=""):
 273     validate_json_format(json, {"sample": str}, forstr)
 274     if "tuning" in json:
 275         validate_json_format(json, {"tuning": float}, forstr)
 276     validate(
 277         json["sample"] in sample_bank.name_to_entry,
 278         "reference to sound {} which isn't found in sample bank {}".format(
 279             json["sample"], sample_bank.name
 280         ),
 281         forstr,
 282     )
 283
 284
 285 def validate_bank_toplevel(json):
 286     validate(isinstance(json, dict), "must have a top-level object")
 287     validate_json_format(
 288         json,
 289         {
 290             "envelopes": dict,
 291             "sample_bank": str,
 292             "instruments": dict,
 293             "instrument_list": list,
 294         },
 295     )
 296
 297
 298 def normalize_sound_json(json):
 299     # Convert {"sound": "str"} into {"sound": {"sample": "str"}}
 300     fixup = []
 301     for inst in json["instruments"].values():
 302         if isinstance(inst, list):
 303             for drum in inst:
 304                 fixup.append((drum, "sound"))
 305         else:
 306             fixup.append((inst, "sound_lo"))
 307             fixup.append((inst, "sound"))
 308             fixup.append((inst, "sound_hi"))
 309     for (obj, key) in fixup:
 310         if isinstance(obj, dict) and isinstance(obj.get(key, None), str):
 311             obj[key] = {"sample": obj[key]}
 312
 313
 314 def validate_bank(json, sample_bank):
 315     if "date" in json:
 316         validate(
 317             isinstance(json["date"], str)
 318             and re.match(r"[0-9]{4}-[0-9]{2}-[0-9]{2}\Z", json["date"]),
 319             "date must have format yyyy-mm-dd",
 320         )
 321
 322     for key, env in json["envelopes"].items():
 323         validate(isinstance(env, list), 'envelope "' + key + '" must be an array')
 324         last_fine = False
 325         for entry in env:
 326             if entry in ["stop", "hang", "restart"]:
 327                 last_fine = True
 328             else:
 329                 validate(
 330                     isinstance(entry, list) and len(entry) == 2,
 331                     'envelope entry in "'
 332                     + key
 333                     + '" must be a list of length 2, or one of stop/hang/restart',
 334                 )
 335                 if entry[0] == "goto":
 336                     validate_int_in_range(
 337                         entry[1], 0, len(env) - 2, "envelope goto target out of range:"
 338                     )
 339                     last_fine = True
 340                 else:
 341                     validate_int_in_range(
 342                         entry[0], 1, 2 ** 16 - 4, "envelope entry's first part"
 343                     )
 344                     validate_int_in_range(
 345                         entry[1], 0, 2 ** 16 - 1, "envelope entry's second part"
 346                     )
 347                     last_fine = False
 348         validate(
 349             last_fine, 'envelope "{}" must end with stop/hang/restart/goto'.format(key)
 350         )
 351
 352     drums = []
 353     instruments = []
 354     instrument_names = set()
 355     for name, inst in json["instruments"].items():
 356         if name == "percussion":
 357             validate(isinstance(inst, list), "drums entry must be a list")
 358             drums = inst
 359         else:
 360             validate(isinstance(inst, dict), "instrument entry must be an object")
 361             instruments.append((name, inst))
 362             instrument_names.add(name)
 363
 364     for drum in drums:
 365         validate(isinstance(drum, dict), "drum entry must be an object")
 366         validate_json_format(
 367             drum,
 368             {"release_rate": [0, 255], "pan": [0, 128], "envelope": str, "sound": dict},
 369         )
 370         validate_sound(drum["sound"], sample_bank)
 371         validate(
 372             drum["envelope"] in json["envelopes"],
 373             "reference to non-existent envelope " + drum["envelope"],
 374             "drum",
 375         )
 376
 377     no_sound = {}
 378
 379     for name, inst in instruments:
 380         forstr = "instrument " + name
 381         for lohi in ["lo", "hi"]:
 382             nr = "normal_range_" + lohi
 383             so = "sound_" + lohi
 384             if nr in inst:
 385                 validate(so in inst, nr + " is specified, but not " + so, forstr)
 386             if so in inst:
 387                 validate(nr in inst, so + " is specified, but not " + nr, forstr)
 388             else:
 389                 inst[so] = no_sound
 390         if "normal_range_lo" not in inst:
 391             inst["normal_range_lo"] = 0
 392         if "normal_range_hi" not in inst:
 393             inst["normal_range_hi"] = 127
 394
 395         validate_json_format(
 396             inst,
 397             {
 398                 "release_rate": [0, 255],
 399                 "envelope": str,
 400                 "normal_range_lo": [0, 127],
 401                 "normal_range_hi": [0, 127],
 402                 "sound_lo": dict,
 403                 "sound": dict,
 404                 "sound_hi": dict,
 405             },
 406             forstr,
 407         )
 408
 409         if "ifdef" in inst:
 410             validate(
 411                 isinstance(inst["ifdef"], list)
 412                 and all(isinstance(x, str) for x in inst["ifdef"]),
 413                 '"ifdef" must be an array of strings',
 414             )
 415
 416         validate(
 417             inst["normal_range_lo"] <= inst["normal_range_hi"],
 418             "normal_range_lo > normal_range_hi",
 419             forstr,
 420         )
 421         validate(
 422             inst["envelope"] in json["envelopes"],
 423             "reference to non-existent envelope " + inst["envelope"],
 424             forstr,
 425         )
 426         for key in ["sound_lo", "sound", "sound_hi"]:
 427             if inst[key] is no_sound:
 428                 del inst[key]
 429             else:
 430                 validate_sound(inst[key], sample_bank, forstr)
 431
 432     seen_instruments = set()
 433     for inst in json["instrument_list"]:
 434         if inst is None:
 435             continue
 436         validate(
 437             isinstance(inst, str),
 438             "instrument list should contain only strings and nulls",
 439         )
 440         validate(
 441             inst in instrument_names, "reference to non-existent instrument " + inst
 442         )
 443         validate(
 444             inst not in seen_instruments, inst + " occurs twice in the instrument list"
 445         )
 446         seen_instruments.add(inst)
 447
 448     for inst in instrument_names:
 449         validate(inst in seen_instruments, "unreferenced instrument " + inst)
 450
 451
 452 def apply_version_diffs(json, defines):
 453     if "VERSION_EU" in defines and isinstance(json.get("date", None), str):
 454         json["date"] = json["date"].replace("1996-03-19", "1996-06-24")
 455
 456     ifdef_removed = set()
 457     for key, inst in json["instruments"].items():
 458         if (
 459             isinstance(inst, dict)
 460             and isinstance(inst.get("ifdef", None), list)
 461             and all(d not in defines for d in inst["ifdef"])
 462         ):
 463             ifdef_removed.add(key)
 464     for key in ifdef_removed:
 465         del json["instruments"][key]
 466         json["instrument_list"].remove(key)
 467
 468
 469 def mark_sample_bank_uses(bank):
 470     bank.sample_bank.uses.append(bank)
 471
 472     def mark_used(name):
 473         bank.sample_bank.name_to_entry[name].used = True
 474
 475     for inst in bank.json["instruments"].values():
 476         if isinstance(inst, list):
 477             for drum in inst:
 478                 mark_used(drum["sound"]["sample"])
 479         else:
 480             if "sound_lo" in inst:
 481                 mark_used(inst["sound_lo"]["sample"])
 482             mark_used(inst["sound"]["sample"])
 483             if "sound_hi" in inst:
 484                 mark_used(inst["sound_hi"]["sample"])
 485
 486
 487 def serialize_ctl(bank, base_ser):
 488     json = bank.json
 489
 490     drums = []
 491     instruments = []
 492     for inst in json["instruments"].values():
 493         if isinstance(inst, list):
 494             drums = inst
 495         else:
 496             instruments.append(inst)
 497
 498     y, m, d = map(int, json.get("date", "0000-00-00").split("-"))
 499     date = y * 10000 + m * 100 + d
 500     base_ser.add(
 501         pack(
 502             "IIII",
 503             len(json["instrument_list"]),
 504             len(drums),
 505             1 if len(bank.sample_bank.uses) > 1 else 0,
 506             to_bcd(date),
 507         )
 508     )
 509
 510     ser = ReserveSerializer()
 511     if drums:
 512         drum_pos_buf = ser.reserve(WORD_BYTES)
 513     else:
 514         ser.add(b"\0" * WORD_BYTES)
 515         drum_pos_buf = None
 516
 517     inst_pos_buf = ser.reserve(WORD_BYTES * len(json["instrument_list"]))
 518     ser.align(16)
 519
 520     used_samples = []
 521     for inst in json["instruments"].values():
 522         if isinstance(inst, list):
 523             for drum in inst:
 524                 used_samples.append(drum["sound"]["sample"])
 525         else:
 526             if "sound_lo" in inst:
 527                 used_samples.append(inst["sound_lo"]["sample"])
 528             used_samples.append(inst["sound"]["sample"])
 529             if "sound_hi" in inst:
 530                 used_samples.append(inst["sound_hi"]["sample"])
 531
 532     sample_name_to_addr = {}
 533     for name in used_samples:
 534         if name in sample_name_to_addr:
 535             continue
 536         sample_name_to_addr[name] = ser.size
 537         aifc = bank.sample_bank.name_to_entry[name]
 538         sample_len = len(aifc.data)
 539
 540         # Sample
 541         ser.add(pack("PP", 0, aifc.offset))
 542         loop_addr_buf = ser.reserve(WORD_BYTES)
 543         book_addr_buf = ser.reserve(WORD_BYTES)
 544         ser.add(pack("I", align(sample_len, 2)))
 545         ser.align(16)
 546
 547         # Book
 548         book_addr_buf.append(pack("P", ser.size))
 549         ser.add(pack("ii", aifc.book.order, aifc.book.npredictors))
 550         for x in aifc.book.table:
 551             ser.add(pack("h", x))
 552         ser.align(16)
 553
 554         # Loop
 555         loop_addr_buf.append(pack("P", ser.size))
 556         if aifc.loop is None:
 557             assert sample_len % 9 in [0, 1]
 558             end = sample_len // 9 * 16 + (sample_len % 2) + (sample_len % 9)
 559             ser.add(pack("IIiI", 0, end, 0, 0))
 560         else:
 561             ser.add(pack("IIiI", aifc.loop.start, aifc.loop.end, aifc.loop.count, 0))
 562             assert aifc.loop.count != 0
 563             for x in aifc.loop.state:
 564                 ser.add(pack("h", x))
 565         ser.align(16)
 566
 567     env_name_to_addr = {}
 568     for name, env in json["envelopes"].items():
 569         env_name_to_addr[name] = ser.size
 570         for entry in env:
 571             if entry == "stop":
 572                 entry = [0, 0]
 573             elif entry == "hang":
 574                 entry = [2 ** 16 - 1, 0]
 575             elif entry == "restart":
 576                 entry = [2 ** 16 - 3, 0]
 577             elif entry[0] == "goto":
 578                 entry[0] = 2 ** 16 - 2
 579             # Envelopes are always written as big endian, to match sequence files
 580             # which are byte blobs and can embed envelopes.
 581             ser.add(struct.pack(">HH", *entry))
 582         ser.align(16)
 583
 584     def ser_sound(sound):
 585         sample_addr = (
 586             0 if sound["sample"] is None else sample_name_to_addr[sound["sample"]]
 587         )
 588         if "tuning" in sound:
 589             tuning = sound["tuning"]
 590         else:
 591             aifc = bank.sample_bank.name_to_entry[sound["sample"]]
 592             tuning = aifc.sample_rate / 32000
 593         ser.add(pack("PfX", sample_addr, tuning))
 594
 595     no_sound = {"sample": None, "tuning": 0.0}
 596
 597     inst_name_to_pos = {}
 598     for name, inst in json["instruments"].items():
 599         if isinstance(inst, list):
 600             continue
 601         inst_name_to_pos[name] = ser.size
 602         env_addr = env_name_to_addr[inst["envelope"]]
 603         ser.add(
 604             pack(
 605                 "BBBBXP",
 606                 0,
 607                 inst.get("normal_range_lo", 0),
 608                 inst.get("normal_range_hi", 127),
 609                 inst["release_rate"],
 610                 env_addr,
 611             )
 612         )
 613         ser_sound(inst.get("sound_lo", no_sound))
 614         ser_sound(inst["sound"])
 615         ser_sound(inst.get("sound_hi", no_sound))
 616     ser.align(16)
 617
 618     for name in json["instrument_list"]:
 619         if name is None:
 620             inst_pos_buf.append(pack("P", 0))
 621             continue
 622         inst_pos_buf.append(pack("P", inst_name_to_pos[name]))
 623
 624     if drums:
 625         drum_poses = []
 626         for drum in drums:
 627             drum_poses.append(ser.size)
 628             ser.add(pack("BBBBX", drum["release_rate"], drum["pan"], 0, 0))
 629             ser_sound(drum["sound"])
 630             env_addr = env_name_to_addr[drum["envelope"]]
 631             ser.add(pack("P", env_addr))
 632         ser.align(16)
 633
 634         drum_pos_buf.append(pack("P", ser.size))
 635         for pos in drum_poses:
 636             ser.add(pack("P", pos))
 637         ser.align(16)
 638
 639     base_ser.add(ser.finish())
 640
 641
 642 def serialize_tbl(sample_bank, ser):
 643     ser.reset_garbage_pos()
 644     base_addr = ser.size
 645     for aifc in sample_bank.entries:
 646         if not aifc.used:
 647             continue
 648         ser.align(16)
 649         aifc.offset = ser.size - base_addr
 650         ser.add(aifc.data)
 651     ser.align(2)
 652     ser.align_garbage(16)
 653
 654
 655 def serialize_seqfile(entries, serialize_entry, entry_list, magic, extra_padding=True):
 656     ser = ReserveSerializer()
 657     ser.add(pack("HHX", magic, len(entry_list)))
 658     table = ser.reserve(len(entry_list) * 2 * WORD_BYTES)
 659     ser.align(16)
 660     data_start = ser.size
 661
 662     ser2 = GarbageSerializer()
 663     entry_offsets = []
 664     entry_lens = []
 665     for entry in entries:
 666         entry_offsets.append(ser2.size)
 667         serialize_entry(entry, ser2)
 668         entry_lens.append(ser2.size - entry_offsets[-1])
 669     ser.add(ser2.finish())
 670     if extra_padding:
 671         ser.add(b"\0")
 672     ser.align(64)
 673
 674     for ent in entry_list:
 675         table.append(pack("P", entry_offsets[ent] + data_start))
 676         table.append(pack("IX", entry_lens[ent]))
 677     return ser.finish()
 678
 679
 680 def validate_and_normalize_sequence_json(json, bank_names, defines):
 681     validate(isinstance(json, dict), "must have a top-level object")
 682     if "comment" in json:
 683         del json["comment"]
 684     for key, seq in json.items():
 685         if isinstance(seq, dict):
 686             validate_json_format(seq, {"ifdef": list, "banks": list}, key)
 687             validate(
 688                 all(isinstance(x, str) for x in seq["ifdef"]),
 689                 '"ifdef" must be an array of strings',
 690                 key,
 691             )
 692             if all(d not in defines for d in seq["ifdef"]):
 693                 seq = None
 694             else:
 695                 seq = seq["banks"]
 696             json[key] = seq
 697         if isinstance(seq, list):
 698             for x in seq:
 699                 validate(
 700                     isinstance(x, str), "bank list must be an array of strings", key
 701                 )
 702                 validate(
 703                     x in bank_names, "reference to non-existing sound bank " + x, key
 704                 )
 705         else:
 706             validate(seq is None, "bad JSON type, expected null, array or object", key)
 707
 708
 709 def write_sequences(
 710     inputs, out_filename, out_bank_sets, sound_bank_dir, seq_json, defines
 711 ):
 712     bank_names = sorted(
 713         [os.path.splitext(os.path.basename(x))[0] for x in os.listdir(sound_bank_dir)]
 714     )
 715
 716     try:
 717         with open(seq_json, "r") as inf:
 718             data = inf.read()
 719             data = strip_comments(data)
 720             json = orderedJsonDecoder.decode(data)
 721             validate_and_normalize_sequence_json(json, bank_names, defines)
 722
 723     except Exception as e:
 724         fail("failed to parse " + str(seq_json) + ": " + str(e))
 725
 726     inputs.sort(key=lambda f: os.path.basename(f))
 727     name_to_fname = {}
 728     for fname in inputs:
 729         name = os.path.splitext(os.path.basename(fname))[0]
 730         if name in name_to_fname:
 731             fail(
 732                 "Files "
 733                 + fname
 734                 + " and "
 735                 + name_to_fname[name]
 736                 + " conflict. Remove one of them."
 737             )
 738         name_to_fname[name] = fname
 739         if name not in json:
 740             fail(
 741                 "Sequence file " + fname + " is not mentioned in sequences.json. "
 742                 "Either assign it a list of sound banks, or set it to null to "
 743                 "explicitly leave it out from the build."
 744             )
 745
 746     for key, seq in json.items():
 747         if key not in name_to_fname and seq is not None:
 748             fail(
 749                 "sequences.json assigns sound banks to "
 750                 + key
 751                 + ", but there is no such sequence file. Either remove the entry (or "
 752                 "set it to null), or create sound/sequences/" + key + ".m64."
 753             )
 754
 755     ind_to_name = []
 756     for key in json:
 757         ind = int(key.split("_")[0], 16)
 758         while len(ind_to_name) <= ind:
 759             ind_to_name.append(None)
 760         if ind_to_name[ind] is not None:
 761             fail(
 762                 "Sequence files "
 763                 + key
 764                 + " and "
 765                 + ind_to_name[ind]
 766                 + " have the same index. Renumber or delete one of them."
 767             )
 768         ind_to_name[ind] = key
 769
 770     while ind_to_name and json.get(ind_to_name[-1], None) is None:
 771         ind_to_name.pop()
 772
 773     def serialize_file(name, ser):
 774         if json.get(name, None) is None:
 775             return
 776         ser.reset_garbage_pos()
 777         with open(name_to_fname[name], "rb") as f:
 778             ser.add(f.read())
 779         ser.align_garbage(16)
 780
 781     with open(out_filename, "wb") as f:
 782         n = range(len(ind_to_name))
 783         f.write(serialize_seqfile(ind_to_name, serialize_file, n, 3, False))
 784
 785     with open(out_bank_sets, "wb") as f:
 786         ser = ReserveSerializer()
 787         table = ser.reserve(len(ind_to_name) * 2)
 788         for name in ind_to_name:
 789             bank_set = json.get(name, None)
 790             if bank_set is None:
 791                 bank_set = []
 792             table.append(pack("H", ser.size))
 793             ser.add(bytes([len(bank_set)]))
 794             for bank in bank_set[::-1]:
 795                 ser.add(bytes([bank_names.index(bank)]))
 796         ser.align(16)
 797         f.write(ser.finish())
 798
 799
 800 def main():
 801     global STACK_TRACES
 802     global ENDIAN_MARKER
 803     global WORD_BYTES
 804     need_help = False
 805     skip_next = 0
 806     cpp_command = None
 807     print_samples = False
 808     sequences_out_file = None
 809     defines = []
 810     args = []
 811     for i, a in enumerate(sys.argv[1:], 1):
 812         if skip_next > 0:
 813             skip_next -= 1
 814             continue
 815         if a == "--help" or a == "-h":
 816             need_help = True
 817         elif a == "--cpp":
 818             cpp_command = sys.argv[i + 1]
 819             skip_next = 1
 820         elif a == "-D":
 821             defines.append(sys.argv[i + 1])
 822             skip_next = 1
 823         elif a == "--endian":
 824             endian = sys.argv[i + 1]
 825             if endian == "big":
 826                 ENDIAN_MARKER = ">"
 827             elif endian == "little":
 828                 ENDIAN_MARKER = "<"
 829             elif endian == "native":
 830                 ENDIAN_MARKER = "="
 831             else:
 832                 fail("--endian takes argument big, little or native")
 833             skip_next = 1
 834         elif a == "--bitwidth":
 835             bitwidth = sys.argv[i + 1]
 836             if bitwidth == 'native':
 837                 WORD_BYTES = struct.calcsize('P')
 838             else:
 839                 if bitwidth not in ['32', '64']:
 840                     fail("--bitwidth takes argument 32, 64 or native")
 841                 WORD_BYTES = int(bitwidth) // 8
 842             skip_next = 1
 843         elif a.startswith("-D"):
 844             defines.append(a[2:])
 845         elif a == "--stack-trace":
 846             STACK_TRACES = True
 847         elif a == "--print-samples":
 848             print_samples = True
 849         elif a == "--sequences":
 850             sequences_out_file = sys.argv[i + 1]
 851             bank_sets_out_file = sys.argv[i + 2]
 852             sound_bank_dir = sys.argv[i + 3]
 853             sequence_json = sys.argv[i + 4]
 854             skip_next = 4
 855         elif a.startswith("-"):
 856             print("Unrecognized option " + a)
 857             sys.exit(1)
 858         else:
 859             args.append(a)
 860
 861     defines_set = {d.split("=")[0] for d in defines}
 862
 863     if sequences_out_file is not None and not need_help:
 864         write_sequences(
 865             args,
 866             sequences_out_file,
 867             bank_sets_out_file,
 868             sound_bank_dir,
 869             sequence_json,
 870             defines_set,
 871         )
 872         sys.exit(0)
 873
 874     if need_help or len(args) != 4:
 875         print(
 876             "Usage: {} <samples dir> <sound bank dir>"
 877             " <out .ctl file> <out .tbl file>"
 878             " [--cpp <preprocessor>]"
 879             " [-D <symbol>]"
 880             " [--stack-trace]"
 881             " | --sequences <out sequence .bin> <out bank sets .bin> <sound bank dir> "
 882             "<sequences.json> <inputs...>".format(sys.argv[0])
 883         )
 884         sys.exit(0 if need_help else 1)
 885
 886     sample_bank_dir = args[0]
 887     sound_bank_dir = args[1]
 888     ctl_data_out = args[2]
 889     tbl_data_out = args[3]
 890
 891     banks = []
 892     sample_banks = []
 893     name_to_sample_bank = {}
 894
 895     sample_bank_names = sorted(os.listdir(sample_bank_dir))
 896     for name in sample_bank_names:
 897         dir = os.path.join(sample_bank_dir, name)
 898         if not os.path.isdir(dir):
 899             continue
 900         entries = []
 901         for f in sorted(os.listdir(dir)):
 902             fname = os.path.join(dir, f)
 903             if not f.endswith(".aifc"):
 904                 continue
 905             try:
 906                 with open(fname, "rb") as inf:
 907                     data = inf.read()
 908                     entries.append(parse_aifc(data, f[:-5], fname))
 909             except Exception as e:
 910                 fail("malformed AIFC file " + fname + ": " + str(e))
 911         if entries:
 912             sample_bank = SampleBank(name, entries)
 913             sample_banks.append(sample_bank)
 914             name_to_sample_bank[name] = sample_bank
 915
 916     bank_names = sorted(os.listdir(sound_bank_dir))
 917     for f in bank_names:
 918         fname = os.path.join(sound_bank_dir, f)
 919         if not f.endswith(".json"):
 920             continue
 921
 922         try:
 923             if cpp_command:
 924                 data = subprocess.run(
 925                     [cpp_command, fname] + ["-D" + x for x in defines],
 926                     stdout=subprocess.PIPE,
 927                     check=True,
 928                 ).stdout.decode()
 929             else:
 930                 with open(fname, "r") as inf:
 931                     data = inf.read()
 932                 data = strip_comments(data)
 933             bank_json = orderedJsonDecoder.decode(data)
 934
 935             validate_bank_toplevel(bank_json)
 936             apply_version_diffs(bank_json, defines_set)
 937             normalize_sound_json(bank_json)
 938
 939             sample_bank_name = bank_json["sample_bank"]
 940             validate(
 941                 sample_bank_name in name_to_sample_bank,
 942                 "sample bank " + sample_bank_name + " not found",
 943             )
 944             sample_bank = name_to_sample_bank[sample_bank_name]
 945
 946             validate_bank(bank_json, sample_bank)
 947
 948             bank = Bank(f[:-5], sample_bank, bank_json)
 949             mark_sample_bank_uses(bank)
 950             banks.append(bank)
 951
 952         except Exception as e:
 953             fail("failed to parse bank " + fname + ": " + str(e))
 954
 955     sample_banks = [b for b in sample_banks if b.uses]
 956     sample_banks.sort(key=lambda b: b.uses[0].name)
 957     sample_bank_index = {}
 958     for sample_bank in sample_banks:
 959         sample_bank_index[sample_bank] = len(sample_bank_index)
 960
 961     with open(tbl_data_out, "wb") as out:
 962         out.write(
 963             serialize_seqfile(
 964                 sample_banks,
 965                 serialize_tbl,
 966                 [sample_bank_index[x.sample_bank] for x in banks],
 967                 TYPE_TBL,
 968             )
 969         )
 970
 971     with open(ctl_data_out, "wb") as out:
 972         if DUMP_INDIVIDUAL_BINS:
 973             # Debug logic, may simplify diffing
 974             os.makedirs("ctl/", exist_ok=True)
 975             for b in banks:
 976                 with open("ctl/" + b.name + ".bin", "wb") as f:
 977                     ser = GarbageSerializer()
 978                     serialize_ctl(b, ser)
 979                     f.write(ser.finish())
 980             print("wrote to ctl/")
 981
 982         out.write(
 983             serialize_seqfile(banks, serialize_ctl, list(range(len(banks))), TYPE_CTL)
 984         )
 985
 986     if print_samples:
 987         for sample_bank in sample_banks:
 988             for entry in sample_bank.entries:
 989                 if entry.used:
 990                     print(entry.fname)
 991
 992
 993 if __name__ == "__main__":
 994     main()