264 lines
10 KiB
Python
264 lines
10 KiB
Python
"""Index of every sound in the game's WAD archives, with names, categories and characters
|
|
(sounds stored inside soundbanks are not indexed yet: docs/formats/banks.md).
|
|
|
|
Sources, in order:
|
|
1. data/catalog/<ARCHIVE>.csv -- shipped catalog (instant) when the WAD fingerprint matches;
|
|
2. otherwise a header scan of the WAD (a few seconds, cached in the user cache), classified with
|
|
`classify` and labelled with data/catalog/characters.csv (speaker per sound name).
|
|
Names follow the community hash list (ackit/data/hashlist.txt): an entry with one WEM is NAME
|
|
(alias NAME_1), entries with n variations are NAME_1..NAME_n (alias NAME#i); stream-pack media are
|
|
named by their ID or <Pack>_<ID>.
|
|
"""
|
|
import csv
|
|
import difflib
|
|
import json
|
|
import os
|
|
import re
|
|
import struct
|
|
from dataclasses import dataclass, asdict
|
|
|
|
from . import classify
|
|
from .game import wad_fingerprint
|
|
from .inject import pack_bnk_path
|
|
from .pack import Pack
|
|
from .paths import catalog_dir, package_data, user_dir
|
|
from .wad import Wad
|
|
|
|
FIELDS = ["archive", "target", "name", "aliases", "group", "category", "character",
|
|
"characters_inferred", "seconds", "rate", "channels", "size", "sha1", "notes"]
|
|
SCAN_VERSION = 1
|
|
|
|
|
|
@dataclass
|
|
class Sound:
|
|
archive: str
|
|
target: str
|
|
name: str
|
|
aliases: tuple = ()
|
|
group: str = "Unknown"
|
|
category: str = "Unknown"
|
|
character: str = "" # verified speaker(s), '; '-separated (curated catalog)
|
|
characters_inferred: str = "" # characters/roles mentioned by the name (hint)
|
|
seconds: float = 0.0
|
|
rate: int = 0
|
|
channels: int = 0
|
|
size: int = 0 # size of the ORIGINAL WEM (bytes) when known from the catalog
|
|
sha1: str = "" # first 16 hex digits of the original WEM's SHA-1 ('' if unknown)
|
|
notes: str = ""
|
|
|
|
@property
|
|
def key(self):
|
|
return "%s:%s" % (self.archive, self.target)
|
|
|
|
@property
|
|
def is_pack_media(self):
|
|
return "/" in self.target
|
|
|
|
@property
|
|
def limit(self):
|
|
"""Max replacement size in bytes (None = any size)."""
|
|
return None if self.is_pack_media else self.size
|
|
|
|
def names(self):
|
|
return (self.name,) + tuple(self.aliases)
|
|
|
|
|
|
# --- static data ---------------------------------------------------------------------------
|
|
_names = None
|
|
|
|
|
|
def hash_names():
|
|
global _names
|
|
if _names is None:
|
|
_names = {}
|
|
for ln in open(package_data("hashlist.txt"), encoding="latin-1"):
|
|
p = ln.rstrip("\r\n").split("\t")
|
|
if len(p) >= 2 and p[1]:
|
|
try:
|
|
_names[int(p[0], 16)] = p[1]
|
|
except ValueError:
|
|
pass
|
|
return _names
|
|
|
|
|
|
def load_characters(path=None):
|
|
"""{sound name (lower): 'Speaker' or 'A; B'} from data/catalog/characters.csv."""
|
|
path = path or os.path.join(catalog_dir(), "characters.csv")
|
|
out = {}
|
|
if os.path.exists(path):
|
|
for row in csv.DictReader(open(path, encoding="utf-8")):
|
|
out[row["name"].lower()] = row["character"]
|
|
return out
|
|
|
|
|
|
def read_csv(path):
|
|
out = []
|
|
for r in csv.DictReader(open(path, encoding="utf-8", newline="")):
|
|
out.append(Sound(archive=r["archive"], target=r["target"], name=r["name"],
|
|
aliases=tuple(a for a in r["aliases"].split("|") if a), group=r["group"],
|
|
category=r["category"], character=r["character"],
|
|
characters_inferred=r["characters_inferred"], seconds=float(r["seconds"] or 0),
|
|
rate=int(r["rate"] or 0), channels=int(r["channels"] or 0), size=int(r["size"] or 0),
|
|
sha1=r["sha1"], notes=r["notes"]))
|
|
return out
|
|
|
|
|
|
def write_csv(path, sounds):
|
|
with open(path, "w", encoding="utf-8", newline="") as f:
|
|
w = csv.DictWriter(f, FIELDS, lineterminator="\n")
|
|
w.writeheader()
|
|
for s in sounds:
|
|
d = asdict(s)
|
|
d["aliases"] = "|".join(s.aliases)
|
|
d["seconds"] = "%.3f" % s.seconds
|
|
w.writerow({k: d[k] for k in FIELDS})
|
|
|
|
|
|
# --- scanning ------------------------------------------------------------------------------
|
|
def _hdr(b):
|
|
"""(rate, channels, seconds) from the first bytes of a WEM (fmt is always the first chunk)."""
|
|
if len(b) >= 0x30 and b[:4] == b"RIFF" and b[12:16] == b"fmt ":
|
|
ch, rate = struct.unpack_from("<HI", b, 22)
|
|
frames = struct.unpack_from("<I", b, 44)[0]
|
|
return rate, ch, (frames / float(rate) if rate else 0.0)
|
|
return 0, 0, 0.0
|
|
|
|
|
|
def _walk_entry(f, offset, size):
|
|
"""Offsets/lengths of the WEMs inside one WAD entry, reading only headers."""
|
|
out = []
|
|
pos = 0
|
|
while pos + 12 <= size:
|
|
f.seek(offset + pos)
|
|
h = f.read(0x30)
|
|
if h[:4] != b"RIFF" or h[8:12] != b"WAVE":
|
|
break
|
|
ln = 8 + struct.unpack_from("<I", h, 4)[0]
|
|
out.append((pos, ln, h))
|
|
# next WEM starts at the first non-zero byte after this one (zero padding in between)
|
|
cand = pos + ln
|
|
f.seek(offset + cand)
|
|
nxt = None
|
|
while cand < size and nxt is None:
|
|
chunk = f.read(min(0x10000, size - cand))
|
|
if not chunk:
|
|
break
|
|
stripped = chunk.lstrip(b"\0")
|
|
if stripped:
|
|
nxt = cand + len(chunk) - len(stripped)
|
|
cand += len(chunk)
|
|
if nxt is None:
|
|
break
|
|
pos = nxt
|
|
return out
|
|
|
|
|
|
def scan_archive(archive, wad_path, characters=None, progress=None):
|
|
"""Header scan of one WAD -> [Sound]."""
|
|
names = hash_names()
|
|
characters = characters if characters is not None else load_characters()
|
|
wad = Wad(wad_path)
|
|
out = []
|
|
with open(wad_path, "rb") as f:
|
|
for n, e in enumerate(wad.entries):
|
|
if progress and n % 200 == 0:
|
|
progress(n, len(wad.entries))
|
|
bnk = pack_bnk_path(wad_path, e.hash)
|
|
if bnk and open(bnk, "rb").read(4) == b"BKPK":
|
|
pk = Pack(open(bnk, "rb").read())
|
|
pname = names.get(e.hash, "%08X" % e.hash)
|
|
for mid, size, off in pk.entries:
|
|
f.seek(e.offset + off)
|
|
rate, ch, secs = _hdr(f.read(0x30))
|
|
out.append(Sound(archive, "%08X/%08X" % (e.hash, mid), names.get(mid, "%s_%08X" % (pname, mid)),
|
|
seconds=secs, rate=rate, channels=ch, size=size))
|
|
out[-1].group, out[-1].category = classify.classify(out[-1].name, pack=pname)
|
|
continue
|
|
wems = _walk_entry(f, e.offset, e.size)
|
|
base = names.get(e.hash, "%08X" % e.hash)
|
|
for i, (off, ln, h) in enumerate(wems):
|
|
rate, ch, secs = _hdr(h)
|
|
if len(wems) == 1:
|
|
s = Sound(archive, "%08X" % e.hash, base, (base + "_1",))
|
|
else:
|
|
s = Sound(archive, "%08X:%d" % (e.hash, i), "%s_%d" % (base, i + 1), ("%s#%d" % (base, i + 1),))
|
|
s.seconds, s.rate, s.channels, s.size = secs, rate, ch, ln
|
|
s.group, s.category = classify.classify(s.name)
|
|
out.append(s)
|
|
for s in out:
|
|
s.characters_inferred = "; ".join(classify.infer_characters(s.name))
|
|
s.character = next((characters[a.lower()] for a in s.names() if a.lower() in characters), "")
|
|
return out
|
|
|
|
|
|
# --- catalog -------------------------------------------------------------------------------
|
|
def _shipped_index():
|
|
p = os.path.join(catalog_dir(), "index.json")
|
|
try:
|
|
return json.load(open(p))
|
|
except (OSError, ValueError):
|
|
return {}
|
|
|
|
|
|
def load_archive(game, archive, progress=None):
|
|
"""[Sound] for one archive of a game (shipped CSV, user-cache scan, or fresh scan)."""
|
|
path = game.wad(archive)
|
|
fp = wad_fingerprint(path)
|
|
shipped = _shipped_index().get(archive, {})
|
|
csv_path = os.path.join(catalog_dir(), "%s.csv" % archive)
|
|
if shipped.get("fingerprint") == fp and os.path.exists(csv_path):
|
|
return read_csv(csv_path)
|
|
cache = os.path.join(user_dir("cache"), "scan_%s_%s_v%d.csv" % (archive, fp, SCAN_VERSION))
|
|
if os.path.exists(cache):
|
|
return read_csv(cache)
|
|
sounds = scan_archive(archive, path, progress=progress)
|
|
# unknown build: no original checksums (states then rely on the originals store only)
|
|
write_csv(cache, sounds)
|
|
return sounds
|
|
|
|
|
|
class Catalog:
|
|
"""All sounds of the selected archives (default: the game's speech language + SFX)."""
|
|
|
|
def __init__(self, game, archives=None, progress=None):
|
|
self.game = game
|
|
self.archives = archives or [a for a in (game.language(), "SFX") if a in game.archives]
|
|
self.sounds = []
|
|
for a in self.archives:
|
|
self.sounds += load_archive(game, a, progress)
|
|
self.by_key = {s.key: s for s in self.sounds}
|
|
self._by_name = {}
|
|
for s in self.sounds:
|
|
for n in s.names():
|
|
self._by_name.setdefault(n.lower(), []).append(s)
|
|
|
|
def get(self, key):
|
|
return self.by_key[key]
|
|
|
|
def resolve(self, name):
|
|
"""One sound by name, alias, key ('ITA:B49CDF48:11') or file name ('X_3.wav')."""
|
|
if name in self.by_key:
|
|
return self.by_key[name]
|
|
key = os.path.splitext(os.path.basename(name))[0].lower() if re.search(r"\.\w{2,4}$", name) else name.lower()
|
|
hits = self._by_name.get(key, [])
|
|
uniq = list({id(s): s for s in hits}.values())
|
|
if len(uniq) == 1:
|
|
return uniq[0]
|
|
if len(uniq) > 1:
|
|
raise LookupError("'%s' is ambiguous; use one of: %s" % (name, ", ".join(
|
|
next((a for a in s.names() if a.lower() != key), s.key) for s in uniq)))
|
|
near = [s.name for s in self.search(key, limit=8)] or self.suggest(key)
|
|
raise LookupError("no sound named '%s'.%s" % (name, ("\n did you mean: " + ", ".join(near)) if near else ""))
|
|
|
|
def search(self, text, limit=None):
|
|
"""All words must appear in the name, category or character (case-insensitive)."""
|
|
words = [w for w in re.split(r"[\s_]+", text.lower()) if w]
|
|
res = [s for s in self.sounds
|
|
if all(w in s.name.lower() or w in s.category.lower() or w in s.character.lower() for w in words)]
|
|
res.sort(key=lambda s: (len(s.name), s.name))
|
|
return res[:limit] if limit else res
|
|
|
|
def suggest(self, text, n=8):
|
|
names = {s.name.lower(): s.name for s in self.sounds}
|
|
return [names[k] for k in difflib.get_close_matches(text.lower(), list(names), n=n, cutoff=0.6)]
|