Files
UnheardCity/tools/build_catalog.py

182 lines
8.8 KiB
Python

"""Build data/catalog/ from the installed (unmodified!) game and the curated character lists.
python tools/build_catalog.py [--game DIR]
Outputs
data/catalog/characters.csv name -> verified speaker, from data/catalog/source/<lang>/*.txt
data/catalog/<ARCHIVE>.csv every sound of one archive (schema: data/catalog/README.md)
data/catalog/BANK.csv in-memory sounds (soundbanks in SFX/*.bnk and the .upk packages)
data/catalog/BANK_locations.csv.gz every stored copy of them + the prefetch heads of speech lines
data/catalog/SUB_<LANG>.csv subtitles of the speech lines (one file per language)
data/catalog/SUB_locations.csv.gz every stored copy of each subtitle string
data/catalog/index.json fingerprints the CSVs were built from + counts
Run it only on clean game files (`ackit status` must say nothing is modified).
"""
import argparse
import collections
import csv
import hashlib
import json
import os
import sys
import time
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from ackit import banks, settings, subtitles # noqa: E402
from ackit.catalog import (apply_line_names, character_of, hash_names, scan_archive, scan_text, # noqa: E402
write_csv)
from ackit.classify import USER_LABELS # noqa: E402
from ackit.game import Game, default_game_dir, wad_fingerprint # noqa: E402
from ackit.inject import read_target # noqa: E402
from ackit.paths import catalog_dir # noqa: E402
from ackit.wad import Wad # noqa: E402
SKIP = {"files.txt", "stats.txt", "checkpoint.txt"}
def read_user_lists(src_dir):
"""{user file name without extension (lower): set(display character)}"""
out = collections.defaultdict(set)
for f in sorted(os.listdir(src_dir)):
if not f.lower().endswith(".txt") or f.lower() in SKIP:
continue
label = os.path.splitext(f)[0]
who = USER_LABELS.get(label, label)
for line in open(os.path.join(src_dir, f), encoding="utf-8", errors="replace"):
n = line.strip()
while os.path.splitext(n)[1].lower() in (".wem", ".wav"):
n = os.path.splitext(n)[0]
if n:
out[n.lower()].add(who)
return out
def build_characters(sounds, src_root):
"""Map the user's wemscan-style names onto catalog names (name or alias)."""
by_alias = collections.defaultdict(list)
for s in sounds:
for a in s.names():
by_alias[a.lower()].append(s)
rows, unmatched, ambiguous = {}, [], []
for lang in sorted(os.listdir(src_root)):
d = os.path.join(src_root, lang)
if not os.path.isdir(d):
continue
for uname, who in read_user_lists(d).items():
hits = list({id(s): s for s in by_alias.get(uname, [])}.values())
if not hits:
unmatched.append(uname)
continue
if len(hits) > 1:
ambiguous.append(uname) # wemscan name collision: both candidates get the label
for s in hits:
cur = rows.setdefault(s.name, set())
cur |= who
return rows, unmatched, ambiguous
def file_sha1(path, target, wad):
return hashlib.sha1(read_target(path, target, wad)).hexdigest()[:16]
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--game")
a = ap.parse_args()
game = Game(a.game or default_game_dir(settings.load()))
out_dir = catalog_dir()
src_root = os.path.join(out_dir, "source")
hash_names()
# 1. every package once: in-memory sounds, prefetch heads, subtitles, events (a few minutes)
t0 = time.time()
bank_snd, locs, sc, texts, sublocs, sub_issues, line_names = scan_text(
game, progress=lambda k, n: print(" banks + subtitles: %d/%d files" % (k, n), flush=True)
if k % 500 == 0 else None, characters={})
print("packages scanned (%.0fs); %d speech sounds named by their game line" % (time.time() - t0, len(line_names)))
# 2. characters from the curated lists. They name ITA sounds by position (wemscan); the position of a
# variation differs between languages, so labels go through the game's line name (valid everywhere);
# sounds without one keep an ITA-only label.
ref_lang = "ITA" if "ITA" in game.archives else game.languages[0]
ref = scan_archive(ref_lang, game.wad(ref_lang), characters={})
chars, unmatched, ambiguous = build_characters(ref, src_root)
by_pos = {s.name: s for s in ref}
rows = collections.defaultdict(set) # (name, language) -> characters
for pos_name, who in chars.items():
s = by_pos[pos_name]
if s.key in line_names:
rows[(line_names[s.key], "")] |= who
else:
rows[(pos_name, ref_lang)] |= who
with open(os.path.join(out_dir, "characters.csv"), "w", encoding="utf-8", newline="") as f:
w = csv.writer(f, lineterminator="\n")
w.writerow(["name", "character", "language"])
for (n, lg) in sorted(rows, key=lambda r: (r[0].lower(), r[1])):
w.writerow([n, "; ".join(sorted(rows[(n, lg)])), lg])
print("characters.csv: %d sounds labelled (%d by line name, %d %s-only); %d unmatched names %s; %d ambiguous %s"
% (len(rows), sum(1 for _, lg in rows if not lg), sum(1 for _, lg in rows if lg), ref_lang,
len(unmatched), unmatched[:3], len(ambiguous), ambiguous[:3]))
def charmap(lang):
return {n.lower(): "; ".join(sorted(c)) for (n, lg), c in rows.items() if not lg or lg == lang}
index = {"built_from": "unmodified game files", "archives": {}}
for arch, path in game.archives.items():
cm = charmap(arch)
sounds = apply_line_names(scan_archive(arch, path, characters={}), line_names)
for s in sounds:
s.character = character_of(s, cm)
wad = Wad(path)
for s in sounds:
s.sha1 = file_sha1(path, s.target, wad)
write_csv(os.path.join(out_dir, "%s.csv" % arch), sounds)
g = collections.Counter(s.group for s in sounds)
index["archives"][arch] = {"wad": game.rel(path), "fingerprint": wad_fingerprint(path),
"sounds": len(sounds), "groups": dict(g),
"with_character": sum(1 for s in sounds if s.character)}
print("%s.csv: %d sounds %s" % (arch, len(sounds), dict(g)))
# in-memory sounds (scanned in step 1); localized speech lines carry the game's line names too
sounds = bank_snd
cm = charmap("") # only labels valid in every language (game line names)
for s in sounds:
s.character = character_of(s, cm)
write_csv(os.path.join(out_dir, "%s.csv" % banks.ARCHIVE), sounds)
banks.write_locations(os.path.join(out_dir, banks.LOCATIONS), locs)
g = collections.Counter(s.group for s in sounds)
heads = [l for l in locs if l.kind == banks.PREFETCH_HEAD]
index["archives"][banks.ARCHIVE] = {
"fingerprint": banks.game_fingerprint(game.cooked), "sounds": len(sounds), "groups": dict(g),
"copies": sum(1 for l in locs if l.kind == banks.MEMORY), "files": len(sc.files),
"named_by_event": sum(1 for s in sounds if not s.name.lower().startswith(s.bank.lower() + "_")),
"prefetch_heads": {"lines": len({l.media for l in heads}), "copies": len(heads),
"unmapped": len({l.media for l in heads if not l.sound})},
"issues": ["%s: %s" % i for i in sc.issues]}
print("%s.csv: %d sounds %s, %d copies in %d files, %d prefetch heads"
% (banks.ARCHIVE, len(sounds), dict(g), len(locs) - len(heads), len(sc.files), len(heads)))
# subtitles (dialogue objects in the packages), one CSV per language + every stored copy
for f in os.listdir(out_dir):
if f.startswith("SUB_") and f.endswith(".csv"):
os.remove(os.path.join(out_dir, f))
for lang, rows in texts.items():
subtitles.write_texts(os.path.join(out_dir, "SUB_%s.csv" % lang), rows)
subtitles.write_locations(os.path.join(out_dir, subtitles.LOCATIONS), sublocs)
any_lang = next(iter(texts.values()), [])
index["archives"]["SUB"] = {
"fingerprint": banks.game_fingerprint(game.cooked),
"lines": {lang: len(rows) for lang, rows in sorted(texts.items())},
"speech_lines": len({s.speech for s in any_lang}),
"linked": {lang: sum(1 for s in rows if s.sound) for lang, rows in sorted(texts.items())},
"copies": len(sublocs), "issues": ["%s: %s" % i for i in sub_issues]}
print("SUB_*.csv: %d languages, %d subtitle strings, %d copies"
% (len(texts), sum(len(r) for r in texts.values()), len(sublocs)))
# flatten for catalog.load_archive
flat = {a: v for a, v in index["archives"].items()}
json.dump(dict(flat, _info=index["built_from"]), open(os.path.join(out_dir, "index.json"), "w", newline="\n"), indent=1)
if __name__ == "__main__":
main()