210 lines
8.4 KiB
Python
210 lines
8.4 KiB
Python
"""Step 1: from the installed game to work/notes.json + the EN originals as 24 kHz WAV.
|
|
|
|
- kfc-parser unpacks the descriptors we need (voiceover, sound, localisation, journal);
|
|
- localisation tags of EN and RU are read straight from the .dat content (24-byte tag records);
|
|
- every journal entry with a voiceover joins its EN sound and its EN/RU text;
|
|
- the RU text is turned into the text for the voice (tags off, dashes to commas, numbers in words);
|
|
- the EN audio (frameless QOA, 48 kHz) is decoded to work/en24/<k>.wav with its loudness.
|
|
The game files are only read. Keys n000..n524 follow the EN sound names sorted.
|
|
|
|
A game with this mod (or any EML mod) applied has a PATCHED enshrouded.kfc: its index then points to the
|
|
Russian takes. EML appends new content and keeps the original index as enshrouded.kfc.bak, so the
|
|
originals are read through work/orig: a copy of the .bak as enshrouded.kfc plus hard links to the .dat
|
|
and .kfc_resources files (the work folder must then be on the game's drive; otherwise run
|
|
`tools/emm.exe restore` first — the mod is applied again at the next game launch).
|
|
"""
|
|
import glob
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import wave
|
|
from multiprocessing import Pool
|
|
|
|
import numpy as np
|
|
|
|
from .config import load
|
|
from .kfc import KFC
|
|
from .qoa import decode
|
|
from .ru_num import normalize
|
|
|
|
TYPES = ["VoiceoverResource", "SoundContainerResource", "SoundResource",
|
|
"LocaTagCollectionResource", "JournalRegistryResource"]
|
|
|
|
|
|
def source(cfg):
|
|
"""The game folder with the ORIGINAL index (see the module docstring)."""
|
|
bak = cfg.game / "enshrouded.kfc.bak"
|
|
if not bak.exists():
|
|
return cfg.game
|
|
orig = cfg.w("orig", "x").parent
|
|
shutil.copyfile(bak, orig / "enshrouded.kfc")
|
|
for f in list(cfg.game.glob("enshrouded_*.dat")) + [cfg.game / "enshrouded.kfc_resources"]:
|
|
dst = orig / f.name
|
|
if dst.exists():
|
|
dst.unlink()
|
|
try:
|
|
os.link(f, dst)
|
|
except OSError:
|
|
raise SystemExit(f"the game has EML mods applied and {orig} cannot hard-link {f.name} "
|
|
f"(another drive?): put paths.work on the game's drive, or run "
|
|
f"'{cfg.tools / 'emm.exe'} restore -g {cfg.game}' and extract again")
|
|
print("reading the original index from enshrouded.kfc.bak (the game has EML mods applied)")
|
|
return orig
|
|
|
|
|
|
def unpack(cfg, src):
|
|
out = cfg.w("unpack")
|
|
if all(glob.glob(str(out / t / "*.json")) for t in TYPES):
|
|
print("unpack: already done"); return out
|
|
exe = cfg.tools / "kfc-parser.exe"
|
|
flt = ",".join(f"tkeen::{t}" for t in TYPES)
|
|
subprocess.run([str(exe), "unpack", "-g", str(src), "-o", str(out), "-f", flt], check=True)
|
|
return out
|
|
|
|
|
|
def by_guid(d, t):
|
|
return {os.path.basename(f).split("_")[0]: f for f in glob.glob(str(d / t / "*.json"))}
|
|
|
|
|
|
def loca(cfg, kfc, d):
|
|
res = json.load(open(glob.glob(str(d / "LocaTagCollectionResource" / "*.json"))[0], encoding="utf-8"))
|
|
langs = {l["language"]: l["data"] for l in res["languages"]}
|
|
out = {}
|
|
for name in ("En_Us", "Ru_Ru"):
|
|
data = kfc.read_hash(langs[name]["hash"])
|
|
off, cnt = struct.unpack_from("<II", data, 0)
|
|
tags = {}
|
|
for i in range(cnt):
|
|
p = off + i * 24
|
|
tid, so, sl = struct.unpack_from("<III", data, p)
|
|
tags[str(tid)] = data[p + 4 + so:p + 4 + so + sl].decode("utf-8", "replace")
|
|
out[name] = tags
|
|
print("loca", name, len(tags), "tags")
|
|
return out
|
|
|
|
|
|
def voice_index(d):
|
|
sc, sr = by_guid(d, "SoundContainerResource"), by_guid(d, "SoundResource")
|
|
rows = []
|
|
for f in glob.glob(str(d / "VoiceoverResource" / "*.json")):
|
|
vo = json.load(open(f, encoding="utf-8"))
|
|
en = vo["languages"][0] # fixed slots EN/DE/FR/ES/ZH
|
|
if en not in sc:
|
|
continue
|
|
for e in json.load(open(sc[en], encoding="utf-8"))["entries"]:
|
|
s = e["$value"].get("sound")
|
|
if s in sr:
|
|
r = json.load(open(sr[s], encoding="utf-8"))
|
|
rows.append(dict(vo=vo["$guid"], sound=s, name=r["debugName"],
|
|
sec=r["frameCount"] / r["framesPerSecond"], h=r["streamedData"]["hash"]))
|
|
return rows
|
|
|
|
|
|
def journal(d):
|
|
j = json.load(open(glob.glob(str(d / "JournalRegistryResource" / "*.json"))[0], encoding="utf-8"))
|
|
m = {}
|
|
|
|
def walk(o):
|
|
if isinstance(o, dict):
|
|
if o.get("voiceover") and "text" in o:
|
|
m[o["voiceover"]] = (o["name"]["value"], o["text"]["value"])
|
|
for v in o.values():
|
|
walk(v)
|
|
elif isinstance(o, list):
|
|
for v in o:
|
|
walk(v)
|
|
walk(j)
|
|
return m
|
|
|
|
|
|
def screen(t):
|
|
return re.sub(r"<[^>]+>", "", t or "")
|
|
|
|
|
|
YO = {"еще": "ещё", "все равно": "всё равно", "трясется": "трясётся", "придется": "придётся", "намек": "намёк",
|
|
"идет": "идёт", "ее ": "её ", "черный": "чёрный", "темн": "тёмн", "зеленый": "зелёный", "живет": "живёт",
|
|
"дает": "даёт", "берет": "берёт"}
|
|
|
|
|
|
def voice_text(t):
|
|
"""The RU screen text -> the text for the voice (the subtitles keep the screen text)."""
|
|
t = screen(t)
|
|
t = re.sub(r"[ \t]+\n", "\n", t); t = re.sub(r"\n[ \t]+", "\n", t)
|
|
t = re.sub(r"^\s*[—–-]\s*", "", t, flags=re.M) # list dashes at a line start
|
|
t = re.sub(r"\s+[—–]\s+", ", ", t) # a dash makes a long pause: a comma
|
|
t = re.sub(r"(\w)-(то|ка|нибудь|либо)\b", r"\1 \2", t)
|
|
for a, b in YO.items():
|
|
t = re.sub(r"\b" + a, b, t)
|
|
t = re.sub(r"\n{3,}", "\n\n", t).strip()
|
|
t = normalize(t).strip()
|
|
return re.sub(r"\.{4,}", "...", t)
|
|
|
|
|
|
_FIR = None
|
|
|
|
|
|
def _en_job(args):
|
|
game, k, r, out = args
|
|
global _FIR
|
|
if _FIR is None:
|
|
n = np.arange(-48, 49)
|
|
_FIR = np.sinc(n * 0.46) * np.hamming(len(n)); _FIR /= _FIR.sum()
|
|
s = np.asarray(decode(KFC(game).read_hash(r["h"]), int(r["en_sec"] * 48000)), np.float64)
|
|
v = s[::5]; v = v[np.abs(v) > 200]
|
|
rms = float(np.sqrt(np.mean(v * v))) if len(v) else 1000.0
|
|
y = np.convolve(s, _FIR, mode="same")[::2]
|
|
with wave.open(out, "wb") as w:
|
|
w.setnchannels(1); w.setsampwidth(2); w.setframerate(24000)
|
|
w.writeframes(np.clip(y, -32768, 32767).astype("<i2").tobytes())
|
|
return k, rms
|
|
|
|
|
|
def main():
|
|
cfg = load()
|
|
src = source(cfg)
|
|
d = unpack(cfg, src)
|
|
kfc = KFC(src)
|
|
L = loca(cfg, kfc, d)
|
|
rows = voice_index(d)
|
|
byvo = {}
|
|
for r in rows:
|
|
byvo.setdefault(r["vo"], []).append(r)
|
|
notes = []
|
|
for vo, (nm, tx) in journal(d).items():
|
|
for r in byvo.get(vo, []):
|
|
ru = L["Ru_Ru"].get(str(tx))
|
|
if not ru:
|
|
continue
|
|
notes.append(dict(name=r["name"], vo=vo, sound=r["sound"], en_sec=r["sec"], h=r["h"],
|
|
title_en=screen(L["En_Us"].get(str(nm))), en=L["En_Us"].get(str(tx)),
|
|
ru_screen=screen(ru), text=voice_text(ru)))
|
|
notes.sort(key=lambda x: x["name"])
|
|
for i, n in enumerate(notes):
|
|
n["k"] = f"n{i:03d}"
|
|
os.makedirs(cfg.w("en24", "x").parent, exist_ok=True)
|
|
jobs = [(str(src), n["k"], n, str(cfg.w("en24", f"{n['k']}.wav"))) for n in notes
|
|
if not os.path.exists(cfg.w("en24", f"{n['k']}.wav"))]
|
|
rms = {}
|
|
if os.path.exists(cfg.w("en_rms.json")):
|
|
rms = json.load(open(cfg.w("en_rms.json")))
|
|
with Pool(max(1, (os.cpu_count() or 2) - 1)) as p:
|
|
for k, v in p.imap_unordered(_en_job, jobs, chunksize=1):
|
|
rms[k] = v
|
|
json.dump(rms, open(cfg.w("en_rms.json"), "w"))
|
|
for n in notes:
|
|
n["en_rms"] = rms.get(n["k"])
|
|
json.dump(notes, open(cfg.w("notes.json"), "w", encoding="utf-8"), ensure_ascii=False, indent=1)
|
|
print("notes", len(notes), "EN minutes", round(sum(n["en_sec"] for n in notes) / 60),
|
|
"RU chars", sum(len(n["text"]) for n in notes))
|
|
if len(notes) != 525:
|
|
print("WARNING: expected 525 voiced notes (game 1.0, 2026-10); a game update may have changed them",
|
|
file=sys.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|