234 lines
12 KiB
Python
234 lines
12 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
|
|
(speech sounds carry the game's line identity and one name in every language: ackit/lines.py)
|
|
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). It stops before writing
|
|
anything if a sound differs from the current catalog (a modified game); after a game update use --accept-changes.
|
|
"""
|
|
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_lines, character_of, read_csv, 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(game, src_root, final):
|
|
"""Map the curated lists onto the catalog. Each list folder (source/<lang>/) names sounds of THAT language's
|
|
WAD by position (wemscan names: NAME = single entry, NAME_i = i-th WEM of the entry). Positions differ
|
|
between languages, so a listed recording is found by its position in that language and the label goes to
|
|
its line (`final`: {key: Sound} with line identities and names), i.e. to the same recording in every
|
|
language; a recording without a line keeps the label in that language only.
|
|
-> ({(name, language or ''): set(characters)}, unmatched names, ambiguous names, skipped folders)"""
|
|
rows = collections.defaultdict(set)
|
|
unmatched, ambiguous, skipped = [], [], []
|
|
for folder in sorted(os.listdir(src_root)):
|
|
d = os.path.join(src_root, folder)
|
|
if not os.path.isdir(d):
|
|
continue
|
|
lang = folder.upper()
|
|
if lang not in game.archives:
|
|
skipped.append(folder) # no speech WAD of that language: positions cannot be resolved
|
|
continue
|
|
ref = scan_archive(lang, game.wad(lang), characters={}) # position names of this language
|
|
by_alias = collections.defaultdict(list)
|
|
for s in ref:
|
|
for a in s.names():
|
|
by_alias[a.lower()].append(s)
|
|
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:
|
|
f = final[s.key]
|
|
rows[(f.name, "" if f.line else lang)] |= who
|
|
return rows, unmatched, ambiguous, skipped
|
|
|
|
|
|
def changed_since_catalog(game, out_dir, bank_snd):
|
|
"""Keys whose original checksum in the existing catalog differs from the game files now."""
|
|
out = []
|
|
for arch, path in game.archives.items():
|
|
p = os.path.join(out_dir, "%s.csv" % arch)
|
|
if not os.path.exists(p):
|
|
continue
|
|
wad = Wad(path)
|
|
for s in read_csv(p):
|
|
if s.sha1 and s.sha1 != file_sha1(path, s.target, wad):
|
|
out.append(s.key)
|
|
p = os.path.join(out_dir, "%s.csv" % banks.ARCHIVE)
|
|
if os.path.exists(p):
|
|
old = {s.key: s.sha1 for s in read_csv(p)}
|
|
out += [s.key for s in bank_snd if old.get(s.key) and s.sha1 and old[s.key] != s.sha1]
|
|
return out
|
|
|
|
|
|
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")
|
|
ap.add_argument("--accept-changes", action="store_true",
|
|
help="the game was updated: accept sounds whose bytes differ from the current catalog")
|
|
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")
|
|
|
|
# 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, lines = scan_text(
|
|
game, progress=lambda k, n: print(" banks + subtitles: %d/%d files" % (k, n), flush=True)
|
|
if k % 500 == 0 else None, characters={})
|
|
st = lines.stats
|
|
print("packages scanned (%.0fs); %d recordings in %d lines, %d of them in every language (%s); "
|
|
"%d recordings without a line; %d relation issues"
|
|
% (time.time() - t0, st["recordings"], st["lines"], st["lines_in_every_language"], "/".join(lines.langs),
|
|
len(st["without_line"]), len(lines.issues)))
|
|
for k, d in lines.issues[:10]:
|
|
print(" issue:", k, d)
|
|
|
|
# safety: the catalog records ORIGINAL bytes. A sound that differs from the current catalog means a modified
|
|
# game (an applied change, another tool) unless the game itself was updated -> stop before writing anything.
|
|
changed = changed_since_catalog(game, out_dir, bank_snd)
|
|
if changed and not a.accept_changes:
|
|
print("STOP: %d sound(s) differ from the current catalog, e.g. %s." % (len(changed), changed[:5]))
|
|
print("Revert all changes (ackit status / ackit revert --all) and run again; after a game update use "
|
|
"--accept-changes.")
|
|
sys.exit(2)
|
|
|
|
# 2. every speech sound with its line identity and name, then the curated speaker lists through them
|
|
wad_sounds = {}
|
|
for arch, path in game.archives.items():
|
|
sounds = scan_archive(arch, path, characters={})
|
|
wad_sounds[arch] = apply_lines(sounds, lines, lines.names) if arch in lines.langs else sounds
|
|
final = {s.key: s for ss in wad_sounds.values() for s in ss}
|
|
rows, unmatched, ambiguous, skipped = build_characters(game, src_root, final)
|
|
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 language-only); %d unmatched names %s; "
|
|
"%d ambiguous %s; skipped list folders without a speech WAD: %s"
|
|
% (len(rows), sum(1 for _, lg in rows if not lg), sum(1 for _, lg in rows if lg),
|
|
len(unmatched), unmatched[:3], len(ambiguous), ambiguous[:3], skipped))
|
|
|
|
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 = wad_sounds[arch]
|
|
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]
|
|
played = {m for ms in banks.event_media(sc).values() for m in ms} # reached by a named event
|
|
by_event = sum(1 for s in sounds if (s.line not in lines.unnamed if s.line else int(s.target, 16) in played))
|
|
index["archives"][banks.ARCHIVE] = {
|
|
"fingerprint": banks.game_fingerprint(game.cooked), "sounds": len(sounds), "groups": dict(g),
|
|
"named_by_event": by_event, "named_by_bank": len(sounds) - by_event,
|
|
"copies": sum(1 for l in locs if l.kind == banks.MEMORY), "files": len(sc.files),
|
|
# bank copies of WAD recordings: prefetch heads and whole in-memory copies (catalog._whole_recordings)
|
|
"prefetch_heads": {"recordings": 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 heads of WAD recordings"
|
|
% (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), **sc.subtitle_objects, "issues": ["%s: %s" % i for i in sub_issues]}
|
|
wad_lines = {lid for lid, per in lines.members.items() if any(not k.startswith("BANK:") for k in per.values())}
|
|
index["archives"]["LINES"] = {
|
|
"languages": lines.langs, "recordings": st["recordings"], "lines": st["lines"],
|
|
"wad_lines": len(wad_lines), "wad_lines_named_by_event": len(wad_lines - lines.unnamed),
|
|
"wad_lines_named_after_int_recording": len(wad_lines & lines.unnamed),
|
|
"lines_in_every_language": st["lines_in_every_language"],
|
|
"lines_missing_a_language": st["lines_missing_a_language"], "without_line": st["without_line"],
|
|
"issues": ["%s: %s" % i for i in lines.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()
|