diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml new file mode 100644 index 0000000..8badc77 --- /dev/null +++ b/.github/workflows/release.yml @@ -0,0 +1,40 @@ +name: Release + +on: + push: + tags: + - '*' + workflow_dispatch: + inputs: + tag: + description: 'Existing tag to build and release' + required: true + +permissions: + contents: write + +jobs: + build: + runs-on: windows-latest + steps: + - uses: actions/checkout@v4 + with: + ref: ${{ github.event_name == 'workflow_dispatch' && inputs.tag || github.ref }} + + - uses: actions/setup-python@v5 + with: + python-version: '3.11' + + - name: Build + run: ./build.ps1 + + - name: Draft release + uses: softprops/action-gh-release@v2 + with: + tag_name: ${{ github.event_name == 'workflow_dispatch' && inputs.tag || github.ref_name }} + name: ${{ github.event_name == 'workflow_dispatch' && inputs.tag || github.ref_name }} + body: "" + draft: true + files: | + dist/hd2-repatcher.exe + dist/hd2-repatcher-cli.exe diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml new file mode 100644 index 0000000..7126984 --- /dev/null +++ b/.github/workflows/test.yml @@ -0,0 +1,21 @@ +name: Test + +on: + push: + pull_request: + +jobs: + test: + runs-on: windows-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: '3.11' + + - name: Install dependencies + run: pip install -e .[test] + + - name: Run tests + run: pytest diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..bb547ed --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +__pycache__/ +*.pyc +.venv/ +*.egg-info/ +.pytest_cache/ + +# PyInstaller build output +build/ +dist/ +*.spec diff --git a/README.md b/README.md new file mode 100644 index 0000000..07e89e0 --- /dev/null +++ b/README.md @@ -0,0 +1,148 @@ +# HD2 Repatcher + +Repatches Helldivers II unit mods after a game update. + +When Helldivers II updates, unit mod `.patch` files can go out of sync with the +game's unit data, causing them to fail to load. This tool scans a folder of +patch files, finds the ones containing unit resources, and updates those +resources in place using the current game data. + +## Requirements + +- A Helldivers II install (specifically its `data` folder) +- Windows: no extra requirements — use the prebuilt executable below +- Linux/macOS: no prebuilt executable is provided, so run from source with + Python 3.10+ (see below) + +## Installation + +### Option 1: Prebuilt executable (Windows) + +Download `hd2-repatcher.exe` (GUI, no console) or `hd2-repatcher-cli.exe` +(console/CLI) from the [Releases](../../releases) page. No Python required. +These are built by CI (`.github/workflows/release.yml`) on `windows-latest`. + +### Option 2: Run from source (Windows, Linux, macOS) + +Clone the repo, then: + +```powershell +python -m venv .venv +.venv\Scripts\Activate.ps1 +pip install -e . +``` + +(On Linux/macOS, activate with `source .venv/bin/activate` instead.) + +### Why two builds? + +`hd2-repatcher.exe` is built by PyInstaller in **windowed** mode: it has no +console and no stdout/stderr at all, so double-clicking it never pops up a +window beyond its own dialogs. + +> **Caveat:** dragging a mod folder onto the windowed `hd2-repatcher.exe` +> still processes it CLI-style — but with no console and no dialogs, so it +> runs (or fails) with zero feedback. Use `hd2-repatcher-cli.exe` for +> drag-and-drop. + +`hd2-repatcher-cli.exe` is built in **console** mode instead. Run from an +already-open terminal, it just prints to that terminal like any other console +program. But double-click it (or drag a folder onto it), and since there's no +terminal for it to attach to, Windows pops up a brand new console window to +show the result — and the tool waits for a keypress before exiting, so the +window doesn't vanish before you can read it. + +If you don't need CLI usage or drag-and-drop feedback, download the regular +windowed `hd2-repatcher.exe` so it doesn't pop up a console at you. + +## Usage + +### GUI + +Double-click `hd2-repatcher.exe`, or from a source install run: + +```powershell +hd2-repatcher +``` + +You'll be prompted to select your Helldivers II `data` folder (once — it's +cached for future runs), then the folder containing the patch files you want +to fix. + +### CLI + +```powershell +hd2-repatcher-cli --game "C:\Program Files (x86)\Steam\steamapps\common\Helldivers 2\data" C:\path\to\mods\SomeMod +``` + +- `-g`/`--game PATH` — path to the Helldivers II `data` folder. Only needs to + be passed once; it's cached for future runs. Requires at least one + `PATCH_FOLDER` in the same invocation. +- `--no-game-path-caching` — don't save or overwrite the cached game data + path when `-g`/`--game` is given. +- `PATCH_FOLDER [PATCH_FOLDER ...]` — one or more folders containing patch + files to update. + +A source install (`pip install -e .`) puts both `hd2-repatcher` and +`hd2-repatcher-cli` on your PATH; they're the same program, named to mirror +the two prebuilt executables. + +Once the game data path is cached, you can also just drag and drop one or more +mod folders directly onto `hd2-repatcher-cli.exe` (or a shortcut to it) — +Windows passes the dropped folder(s) as arguments, so the tool processes them +immediately instead of prompting, with a console window showing the result +(see [Why two builds?](#why-two-builds) for why this only pops up for the CLI +build). The window stays open until you press Enter. + +Once the game path is cached, you can omit `-g`: + +```powershell +hd2-repatcher-cli C:\path\to\mods\SomeMod C:\path\to\mods\AnotherMod +``` + +Exit code is non-zero if any corrupted patch files were found. + +If you're integrating this into a mod manager, always pass the game data path +explicitly via `-g`/`--game` on every invocation rather than relying on the +cache — the cache is a convenience for interactive/manual use, and a mod +manager shouldn't assume a previous run (by itself or another tool) already +set it. + +## Settings + +The game data path chosen via the GUI or `-g`/`--game` is cached in +`%LOCALAPPDATA%\hd2-repatcher\settings.json`. Delete that file to +reset it, or pass `-g`/`--game` again to overwrite it. + +## Testing + +Unit tests live in `tests/` and run in CI on every push and pull request +(`.github/workflows/test.yml`). To run them locally: + +```powershell +pip install -e ".[test]" +pytest +``` + +There's no automated end-to-end test against real game files (the game data +isn't available in CI), so changes that touch the patching logic should be +verified manually against a mod that's actually broken by a game update: + +1. Find a mod that's known to break after updates, e.g. + [Invisible supply pack](https://www.nexusmods.com/helldivers2/mods/7308?tab=files), + and download an **older** file version — recent-enough game updates should + have desynced it from current unit data. +2. Install it with a mod manager (or manually) and confirm in-game that it's + broken (fails to load / crashes / doesn't apply). +3. Run it through the repatcher (GUI, drag-and-drop, or CLI) and confirm it + reports the patch as updated. +4. Redeploy the mod and confirm it now loads correctly in-game. + +## Building + +```powershell +./build.ps1 +``` + +Builds both `dist\hd2-repatcher.exe` (windowed) and +`dist\hd2-repatcher-cli.exe` (console) via PyInstaller. diff --git a/build.ps1 b/build.ps1 new file mode 100644 index 0000000..547f247 --- /dev/null +++ b/build.ps1 @@ -0,0 +1,20 @@ +# Builds standalone executables into dist\ (gitignored). +# Usage: ./build.ps1 +$ErrorActionPreference = "Stop" + +if (-not (Test-Path .venv)) { + python -m venv .venv +} +. .venv\Scripts\Activate.ps1 + +pip install -e ".[build]" -q + +# GUI build: no console window, for double-click use +pyinstaller --onefile --windowed --name hd2-repatcher --clean --specpath build cli.py + +# CLI build: normal console app, for use from a terminal +pyinstaller --onefile --name hd2-repatcher-cli --clean --specpath build cli.py + +Write-Host "Build complete:" +Write-Host " dist\hd2-repatcher.exe (GUI, double-click)" +Write-Host " dist\hd2-repatcher-cli.exe (CLI, run from a terminal)" diff --git a/cli.py b/cli.py new file mode 100644 index 0000000..1c334cc --- /dev/null +++ b/cli.py @@ -0,0 +1,153 @@ +import argparse +import os +import sys + +if os.name == "nt": + import ctypes + +from settings import get_cached_game_data_path, set_cached_game_data_path +from update_unit_mods import ( + LEGACY_MARKER_FILE, + SLIM_MARKER_FILE, + PatchResult, + init_game_resources, + is_valid_game_data_path, + process_patch_folder, +) + +def print_cli_result(directory: str, result: PatchResult): + print(f"\n{directory}") + if result.patches_found == 0: + print(" No patch files found.") + return + print(f" Checked {result.patches_found} patch file(s)") + print(f" Updated {len(result.updated)} patch file(s) containing unit resources") + if result.no_units: + print(f" Skipped {len(result.no_units)} patch file(s) with no unit resources") + if result.corrupted_files: + print(f" Found {len(result.corrupted_files)} corrupted patch file(s):", file=sys.stderr) + for name in result.corrupted_files: + print(f" {os.path.normpath(name)}", file=sys.stderr) + +def _process_image_path(pid): + PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 + handle = ctypes.windll.kernel32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid) + if not handle: + return None + try: + buf = ctypes.create_unicode_buffer(260) + size = ctypes.c_uint32(260) + if ctypes.windll.kernel32.QueryFullProcessImageNameW(handle, 0, buf, ctypes.byref(size)): + return buf.value + return None + finally: + ctypes.windll.kernel32.CloseHandle(handle) + +def pause_if_owns_console(): + ''' + When the console build is launched by double-click or drag-and-drop, + Windows spawns a fresh console that closes the instant this process + exits, taking the output with it. If every other process attached to + the console is either us or (for the --onefile build) the bootloader + process that re-executed itself as us, hold the window open so the + result can be read. If a real shell is attached (run from an existing + terminal), or in the windowed build (which has no console at all), + this is a no-op. + ''' + if os.name != "nt": + return + process_ids = (ctypes.c_uint32 * 8)() + count = ctypes.windll.kernel32.GetConsoleProcessList(process_ids, 8) + if count == 0 or count > 8: + return + our_pid = os.getpid() + our_path = os.path.normcase(sys.executable) + for pid in process_ids[:count]: + if pid == our_pid: + continue + image_path = _process_image_path(pid) + if image_path is None or os.path.normcase(image_path) != our_path: + return + if sys.stdin is None or not sys.stdin.isatty(): + return + try: + input("\nPress Enter to exit...") + except EOFError: + pass + +def exit_cli(code): + pause_if_owns_console() + sys.exit(code) + +def run_cli(game_path, patch_dirs): + if game_path is None: + cached = get_cached_game_data_path() + if cached and is_valid_game_data_path(cached): + game_path = cached + else: + print("error: no game data directory configured; pass -g/--game ", file=sys.stderr) + exit_cli(1) + + print(f"Loading game resources from: {game_path}") + init_game_resources(game_path) + + exit_code = 0 + for patch_dir in patch_dirs: + patch_dir = os.path.abspath(patch_dir) + if not os.path.isdir(patch_dir): + print(f"error: '{patch_dir}' is not a directory", file=sys.stderr) + exit_code = 1 + continue + result = process_patch_folder(patch_dir) + print_cli_result(patch_dir, result) + if result.corrupted_files: + exit_code = 1 + exit_cli(exit_code) + +def parse_args(): + parser = argparse.ArgumentParser(description="Update unit resources in Helldivers II patch files.") + parser.add_argument("-g", "--game", metavar="PATH", + help="path to the Helldivers II game data folder; also cached for future runs") + parser.add_argument("--no-game-path-caching", action="store_true", + help="do not save or overwrite the cached game data path") + parser.add_argument("patches", nargs="*", metavar="PATCH_FOLDER", + help="folder(s) containing patch files to update") + args = parser.parse_args() + if args.game and not args.patches: + parser.error("at least one PATCH_FOLDER is required with -g/--game") + return args + +def setup_console_io(): + ''' + The windowed build has no console, so sys.stdout/stderr are None; guard + stray print() calls from crashing it. The console build already has real + stdio and this is a no-op there. + ''' + if sys.stdout is None: + sys.stdout = open(os.devnull, "w") + if sys.stderr is None: + sys.stderr = open(os.devnull, "w") + +def main(): + setup_console_io() + args = parse_args() + + game_path = None + if args.game: + game_path = os.path.abspath(args.game) + if not is_valid_game_data_path(game_path): + print(f"error: '{args.game}' does not look like a Helldivers II data folder " + f"(expected to find `{LEGACY_MARKER_FILE}` or `{SLIM_MARKER_FILE}` inside it)", file=sys.stderr) + exit_cli(1) + if not args.no_game_path_caching: + set_cached_game_data_path(game_path) + print(f"Game data directory set to: {game_path}") + + if args.patches: + run_cli(game_path, args.patches) + else: + from gui import run_gui + run_gui() + +if __name__ == "__main__": + main() diff --git a/gui.py b/gui.py new file mode 100644 index 0000000..7bb7413 --- /dev/null +++ b/gui.py @@ -0,0 +1,86 @@ +''' +Tkinter GUI: folder-picker dialogs around the patching engine in +update_unit_mods. Only imported (by cli.main) when the tool is started with +no arguments, so CLI runs never load tkinter. +''' +import os +import sys +import tkinter as tk +from tkinter import filedialog, messagebox + +from settings import get_cached_game_data_path, set_cached_game_data_path +from update_unit_mods import init_game_resources, is_valid_game_data_path, process_patch_folder + +def select_folder(): + d = filedialog.askdirectory(title="Select folder containing patch files") + if d: + if not os.path.exists(d): + messagebox.showwarning(message="No valid folder selected!") + return False + else: + return None + return d + +def select_data_folder(): + d = filedialog.askdirectory(title="Select folder containing game data") + if d: + if not os.path.exists(d): + messagebox.showwarning(message="No valid folder selected!") + return False + if not is_valid_game_data_path(d): + messagebox.showwarning(message="Unable to find Helldivers II game data at this location; make sure you select the `data` folder in your Helldivers II install") + return False + else: + return None + return d + +def update_all_gui(directory: str): + result = process_patch_folder(directory) + if result.patches_found == 0: + messagebox.showwarning(message="No patch files found in folder!") + return + if result.corrupted_files: + m = f"Found {len(result.corrupted_files)} corrupted patch file(s)!" + for name in result.corrupted_files: + m += f"\n{os.path.normpath(name)}" + messagebox.showerror(message=m) + m = (f"Update Complete!\nChecked {result.patches_found} patch file(s).\n" + f"Updated {len(result.updated)} patch file(s) that contained unit resources.") + if result.no_units: + m += f"\n{len(result.no_units)} patch file(s) did not contain any unit resources and were skipped." + messagebox.showinfo(message=m) + +def run_gui(): + root = tk.Tk() + root.withdraw() + + print("fixing unit mods...") + + game_data_path = "" + cached = get_cached_game_data_path() + if cached and is_valid_game_data_path(cached): + game_data_path = cached + init_game_resources(game_data_path) + + while True: + + if not game_data_path: + selection = select_data_folder() + if selection is False: + continue + if selection is None: + if messagebox.askyesnocancel(message="Would you like to quit?"): + sys.exit() + continue + game_data_path = os.path.abspath(selection) + set_cached_game_data_path(game_data_path) + init_game_resources(game_data_path) + + directory = select_folder() + if directory is False: + continue + if directory is None: + if messagebox.askyesnocancel(message="Would you like to quit?"): + sys.exit() + continue + update_all_gui(directory) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..a987595 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,29 @@ +[project] +name = "hd2-repatcher" +version = "0.3.0" +description = "Repatches Helldivers II unit mods after game updates" +requires-python = ">=3.10" +dependencies = [ + "lz4==4.4.5", + "platformdirs==4.9.6", +] + +[project.optional-dependencies] +build = ["pyinstaller>=6.10"] +test = ["pytest>=8"] + +# Two names for the same entry point, mirroring the two prebuilt exes: a +# source install gets both `hd2-repatcher` and `hd2-repatcher-cli`. +[project.scripts] +hd2-repatcher = "cli:main" +hd2-repatcher-cli = "cli:main" + +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[tool.setuptools] +py-modules = ["update_unit_mods", "slim", "settings", "cli", "gui"] + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/requirements.txt b/requirements.txt deleted file mode 100644 index 9340b32..0000000 Binary files a/requirements.txt and /dev/null differ diff --git a/settings.py b/settings.py new file mode 100644 index 0000000..7fd163a --- /dev/null +++ b/settings.py @@ -0,0 +1,34 @@ +import json +from pathlib import Path +from platformdirs import user_config_dir + +APP_NAME = "hd2-repatcher" + +def _settings_path() -> Path: + return Path(user_config_dir(APP_NAME)) / "settings.json" + + +def load_settings() -> dict: + path = _settings_path() + if not path.exists(): + return {} + try: + return json.loads(path.read_text()) + except (json.JSONDecodeError, OSError): + return {} + + +def save_settings(settings: dict) -> None: + path = _settings_path() + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(settings, indent=2)) + + +def get_cached_game_data_path() -> str | None: + return load_settings().get("game_data_path") + + +def set_cached_game_data_path(path: str) -> None: + settings = load_settings() + settings["game_data_path"] = path + save_settings(settings) diff --git a/slim.py b/slim.py index 8852a8f..2ba86cd 100644 --- a/slim.py +++ b/slim.py @@ -1,412 +1,412 @@ -import struct -import os -import sys -from lz4 import block - -def read_int(file): - return int.from_bytes(file.read(4), "little") - -def read_long(file): - return int.from_bytes(file.read(8), "little") - -def read_short(file): - return int.from_bytes(file.read(2), "little") - -def read_char(file): - return int.from_bytes(file.read(1), "little") - -def to_int(byte_data): - return int.from_bytes(byte_data, "little") - - -# chunk type flags -CONTINUE = 0x04 -START = 0x02 -UNK = 0x01 - -# compression -UNCOMPRESSED = 0x00 -COMPRESSED = 0x03 - -# package type -LEGACY = 3 -BUNDLED = 2 -DSAR = 1 -UNKNOWN = 0 - -done_init = False -package_contents = {} -bundle_offsets = {} -file_handles = {} - -# optimization stuff -START_OFFSET = 1 -BUNDLE_INDEX = 2 -ORIGINAL_ARCHIVE_OFFSET = 0 -SIZE = 0 -ENTRIES = 1 - -game_data_folder = "" - -def slim_init(file_path: str): - global game_data_folder - game_data_folder = file_path - if is_slim_version(): - init_bundle_mapping() - -def is_slim_version(): - return not os.path.exists(os.path.join(game_data_folder, "9ba626afa44a3aa3")) - -def get_file_handle(file_path): - file_path = os.path.normpath(file_path) - if file_path in file_handles: - f = file_handles[file_path] - f.seek(0) - return f - else: - f = open(file_path, 'rb') - file_handles[file_path] = f - return f - -def close_file_handles(): - global file_handles - for f in file_handles.values(): - f.close() - file_handles = {} - -def decompress_dsar(file_path): - - # decompresses entire bundle file - - bundle = open(file_path, 'rb') - - num_chunks = num_chunks = struct.unpack("<8xI20x", bundle.read(0x20))[0] # num data chunks - data = [] - file_count = 0 - chunk_data = struct.unpack(f"<{'QQIIBB6x'*num_chunks}", bundle.read(0x20*num_chunks)) - - for i in range(num_chunks): - uncompressed_offset, compressed_offset, uncompressed_size, compressed_size, compression_type, chunk_type = chunk_data[6*i:6*(i+1)] - - bundle.seek(compressed_offset) - - # read and decompress data - temp_data = bundle.read(compressed_size) - if compression_type == COMPRESSED: - temp_data = block.decompress(temp_data, uncompressed_size=uncompressed_size) - data.append(temp_data) - - bundle.close() - - return b"".join(data) - -def get_resource_from_bundle(bundle_path: str, resource_file_offset: int): - - # returns resource from bundle file; resource determined by file offset in uncompressed bundle - # handles resources split into multiple compressed chunks to return complete resource - - bundle = open(bundle_path, 'rb') - num_chunks = struct.unpack("<8xI", bundle.read(12))[0] # num data chunks - data = [] - - global bundle_offsets - chunk_num = bundle_offsets[os.path.basename(bundle_path)][resource_file_offset] - - while True: - bundle.seek(0x20 + 0x20 * chunk_num) - uncompressed_offset, compressed_offset, uncompressed_size, compressed_size, compression_type, chunk_type = struct.unpack(" 0: - bundle.close() - return b"".join(data) - - # read and decompress data - bundle.seek(compressed_offset) - temp_data = bundle.read(compressed_size) - if compression_type == COMPRESSED: - temp_data = block.decompress(temp_data, uncompressed_size=uncompressed_size) - data.append(temp_data) - - if chunk_num == num_chunks - 1: - bundle.close() - return b"".join(data) - - chunk_num += 1 - - bundle.close() - -class Package: - - def __init__(self): - self.size = 0 - self.entries = [] - -class BundleEntry: - - def __init__(self): - self.start_offset = self.bundle_index = self.original_archive_offset = 0 - -def get_resource_from_package(package_name: str, resource_file_offset: int, resource_size: int = 0): - - global package_contents - - package_name = os.path.basename(package_name) - - full_path = os.path.join(game_data_folder, package_name) - - package_type = 0 - - if os.path.exists(full_path): - with open(full_path, 'rb') as f: - magic = int.from_bytes(f.read(4), "little") - if magic == 1380012868: # compressed DSAR file - package_type = DSAR - else: - package_type = LEGACY - else: - package_type = BUNDLED - - if package_type == BUNDLED: - - try: - package = package_contents[package_name] - except KeyError: - # print(f"Unable to get package {package_name}") - return bytearray() - - # how to convert file offset in package into file offset in bundle? - - for entry in reversed(package[ENTRIES]): - if entry[ORIGINAL_ARCHIVE_OFFSET] <= resource_file_offset: - return get_resource_from_bundle(os.path.join(game_data_folder, f"bundles.{entry[BUNDLE_INDEX]:02d}.nxa"), entry[START_OFFSET] + (resource_file_offset - entry[ORIGINAL_ARCHIVE_OFFSET])) - - return bytearray() - - elif package_type == DSAR: - - return get_resource_from_bundle(full_path, resource_file_offset) - - elif package_type == LEGACY: - - package_file = open(full_path, 'rb') - bin_data = b"" - bin_data = package_file.read(12) - magic, numTypes, numFiles = struct.unpack(" []") - sys.exit() - game_data_folder = sys.argv[1] - package_name = sys.argv[2] - if len(sys.argv) == 3: - output_folder = "." - else: - output_folder = sys.argv[3] - slim_init(game_data_folder) - content = reconstruct_package_from_bundles(package_name) - if content: - with open(os.path.join(output_folder, package_name), 'wb') as f: - f.write(content) - - content = reconstruct_package_from_bundles(f"{package_name}.gpu_resources") - if content: - with open(os.path.join(output_folder, f"{package_name}.gpu_resources"), 'wb') as f: - f.write(content) - - content = reconstruct_package_from_bundles(f"{package_name}.stream") - if content: - with open(os.path.join(output_folder, f"{package_name}.stream"), 'wb') as f: - f.write(content) +import struct +import os +import sys +from lz4 import block + +def read_int(file): + return int.from_bytes(file.read(4), "little") + +def read_long(file): + return int.from_bytes(file.read(8), "little") + +def read_short(file): + return int.from_bytes(file.read(2), "little") + +def read_char(file): + return int.from_bytes(file.read(1), "little") + +def to_int(byte_data): + return int.from_bytes(byte_data, "little") + + +# chunk type flags +CONTINUE = 0x04 +START = 0x02 +UNK = 0x01 + +# compression +UNCOMPRESSED = 0x00 +COMPRESSED = 0x03 + +# package type +LEGACY = 3 +BUNDLED = 2 +DSAR = 1 +UNKNOWN = 0 + +done_init = False +package_contents = {} +bundle_offsets = {} +file_handles = {} + +# optimization stuff +START_OFFSET = 1 +BUNDLE_INDEX = 2 +ORIGINAL_ARCHIVE_OFFSET = 0 +SIZE = 0 +ENTRIES = 1 + +game_data_folder = "" + +def slim_init(file_path: str): + global game_data_folder + game_data_folder = file_path + if is_slim_version(): + init_bundle_mapping() + +def is_slim_version(): + return not os.path.exists(os.path.join(game_data_folder, "9ba626afa44a3aa3")) + +def get_file_handle(file_path): + file_path = os.path.normpath(file_path) + if file_path in file_handles: + f = file_handles[file_path] + f.seek(0) + return f + else: + f = open(file_path, 'rb') + file_handles[file_path] = f + return f + +def close_file_handles(): + global file_handles + for f in file_handles.values(): + f.close() + file_handles = {} + +def decompress_dsar(file_path): + + # decompresses entire bundle file + + bundle = open(file_path, 'rb') + + num_chunks = num_chunks = struct.unpack("<8xI20x", bundle.read(0x20))[0] # num data chunks + data = [] + file_count = 0 + chunk_data = struct.unpack(f"<{'QQIIBB6x'*num_chunks}", bundle.read(0x20*num_chunks)) + + for i in range(num_chunks): + uncompressed_offset, compressed_offset, uncompressed_size, compressed_size, compression_type, chunk_type = chunk_data[6*i:6*(i+1)] + + bundle.seek(compressed_offset) + + # read and decompress data + temp_data = bundle.read(compressed_size) + if compression_type == COMPRESSED: + temp_data = block.decompress(temp_data, uncompressed_size=uncompressed_size) + data.append(temp_data) + + bundle.close() + + return b"".join(data) + +def get_resource_from_bundle(bundle_path: str, resource_file_offset: int): + + # returns resource from bundle file; resource determined by file offset in uncompressed bundle + # handles resources split into multiple compressed chunks to return complete resource + + bundle = open(bundle_path, 'rb') + num_chunks = struct.unpack("<8xI", bundle.read(12))[0] # num data chunks + data = [] + + global bundle_offsets + chunk_num = bundle_offsets[os.path.basename(bundle_path)][resource_file_offset] + + while True: + bundle.seek(0x20 + 0x20 * chunk_num) + uncompressed_offset, compressed_offset, uncompressed_size, compressed_size, compression_type, chunk_type = struct.unpack(" 0: + bundle.close() + return b"".join(data) + + # read and decompress data + bundle.seek(compressed_offset) + temp_data = bundle.read(compressed_size) + if compression_type == COMPRESSED: + temp_data = block.decompress(temp_data, uncompressed_size=uncompressed_size) + data.append(temp_data) + + if chunk_num == num_chunks - 1: + bundle.close() + return b"".join(data) + + chunk_num += 1 + + bundle.close() + +class Package: + + def __init__(self): + self.size = 0 + self.entries = [] + +class BundleEntry: + + def __init__(self): + self.start_offset = self.bundle_index = self.original_archive_offset = 0 + +def get_resource_from_package(package_name: str, resource_file_offset: int, resource_size: int = 0): + + global package_contents + + package_name = os.path.basename(package_name) + + full_path = os.path.join(game_data_folder, package_name) + + package_type = 0 + + if os.path.exists(full_path): + with open(full_path, 'rb') as f: + magic = int.from_bytes(f.read(4), "little") + if magic == 1380012868: # compressed DSAR file + package_type = DSAR + else: + package_type = LEGACY + else: + package_type = BUNDLED + + if package_type == BUNDLED: + + try: + package = package_contents[package_name] + except KeyError: + # print(f"Unable to get package {package_name}") + return bytearray() + + # how to convert file offset in package into file offset in bundle? + + for entry in reversed(package[ENTRIES]): + if entry[ORIGINAL_ARCHIVE_OFFSET] <= resource_file_offset: + return get_resource_from_bundle(os.path.join(game_data_folder, f"bundles.{entry[BUNDLE_INDEX]:02d}.nxa"), entry[START_OFFSET] + (resource_file_offset - entry[ORIGINAL_ARCHIVE_OFFSET])) + + return bytearray() + + elif package_type == DSAR: + + return get_resource_from_bundle(full_path, resource_file_offset) + + elif package_type == LEGACY: + + package_file = open(full_path, 'rb') + bin_data = b"" + bin_data = package_file.read(12) + magic, numTypes, numFiles = struct.unpack(" []") + sys.exit() + game_data_folder = sys.argv[1] + package_name = sys.argv[2] + if len(sys.argv) == 3: + output_folder = "." + else: + output_folder = sys.argv[3] + slim_init(game_data_folder) + content = reconstruct_package_from_bundles(package_name) + if content: + with open(os.path.join(output_folder, package_name), 'wb') as f: + f.write(content) + + content = reconstruct_package_from_bundles(f"{package_name}.gpu_resources") + if content: + with open(os.path.join(output_folder, f"{package_name}.gpu_resources"), 'wb') as f: + f.write(content) + + content = reconstruct_package_from_bundles(f"{package_name}.stream") + if content: + with open(os.path.join(output_folder, f"{package_name}.stream"), 'wb') as f: + f.write(content) close_file_handles() \ No newline at end of file diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..588beed --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,194 @@ +import sys + +import pytest + +import cli +import gui +from update_unit_mods import PatchResult + + +@pytest.fixture(autouse=True) +def no_pause(monkeypatch): + # Never block on the drag-and-drop "Press Enter to exit" pause in tests. + monkeypatch.setattr(cli, "pause_if_owns_console", lambda: None) + + +class TestPrintCliResult: + def test_reports_no_patches_found(self, capsys): + cli.print_cli_result("somedir", PatchResult(patches_found=0)) + assert "No patch files found." in capsys.readouterr().out + + def test_reports_updated_and_skipped_counts(self, capsys): + result = PatchResult(patches_found=3, updated=["a"], no_units=["b"]) + cli.print_cli_result("somedir", result) + out = capsys.readouterr().out + assert "Checked 3 patch file(s)" in out + assert "Updated 1 patch file(s)" in out + assert "Skipped 1 patch file(s)" in out + + def test_reports_corrupted_files_to_stderr(self, capsys): + result = PatchResult(patches_found=1, corrupted_files=["bad.patch_0"]) + cli.print_cli_result("somedir", result) + err = capsys.readouterr().err + assert "Found 1 corrupted patch file(s)" in err + assert "bad.patch_0" in err + + +class TestRunCli: + def test_exits_with_error_when_no_game_path_configured(self, monkeypatch, capsys): + monkeypatch.setattr(cli, "get_cached_game_data_path", lambda: None) + + with pytest.raises(SystemExit) as exc: + cli.run_cli(None, ["somedir"]) + + assert exc.value.code == 1 + assert "no game data directory configured" in capsys.readouterr().err + + def test_uses_cached_path_when_none_given(self, monkeypatch, tmp_path): + inited = {} + monkeypatch.setattr(cli, "get_cached_game_data_path", lambda: str(tmp_path)) + monkeypatch.setattr(cli, "is_valid_game_data_path", lambda p: True) + monkeypatch.setattr(cli, "init_game_resources", lambda p: inited.setdefault("path", p)) + monkeypatch.setattr(cli, "process_patch_folder", lambda d: PatchResult(directory=d)) + patch_dir = tmp_path / "patches" + patch_dir.mkdir() + + with pytest.raises(SystemExit) as exc: + cli.run_cli(None, [str(patch_dir)]) + + assert exc.value.code == 0 + assert inited["path"] == str(tmp_path) + + def test_reports_error_for_missing_patch_directory(self, monkeypatch, tmp_path, capsys): + monkeypatch.setattr(cli, "init_game_resources", lambda p: None) + + with pytest.raises(SystemExit) as exc: + cli.run_cli(str(tmp_path), [str(tmp_path / "missing")]) + + assert exc.value.code == 1 + assert "is not a directory" in capsys.readouterr().err + + def test_exit_code_reflects_corrupted_files(self, monkeypatch, tmp_path): + monkeypatch.setattr(cli, "init_game_resources", lambda p: None) + monkeypatch.setattr( + cli, + "process_patch_folder", + lambda d: PatchResult(directory=d, patches_found=1, corrupted_files=["bad.patch_0"]), + ) + patch_dir = tmp_path / "patches" + patch_dir.mkdir() + + with pytest.raises(SystemExit) as exc: + cli.run_cli(str(tmp_path), [str(patch_dir)]) + + assert exc.value.code == 1 + + +class TestParseArgs: + def test_parses_game_and_patches(self, monkeypatch): + monkeypatch.setattr(sys, "argv", ["prog", "-g", "C:\\game", "dir1", "dir2"]) + args = cli.parse_args() + assert args.game == "C:\\game" + assert args.patches == ["dir1", "dir2"] + + def test_defaults_when_no_args_given(self, monkeypatch): + monkeypatch.setattr(sys, "argv", ["prog"]) + args = cli.parse_args() + assert args.game is None + assert args.patches == [] + + def test_game_without_patches_is_an_error(self, monkeypatch, capsys): + monkeypatch.setattr(sys, "argv", ["prog", "-g", "C:\\game"]) + + with pytest.raises(SystemExit) as exc: + cli.parse_args() + + assert exc.value.code == 2 + assert "PATCH_FOLDER" in capsys.readouterr().err + + def test_no_game_path_caching_defaults_to_false(self, monkeypatch): + monkeypatch.setattr(sys, "argv", ["prog"]) + args = cli.parse_args() + assert args.no_game_path_caching is False + + def test_parses_no_game_path_caching_flag(self, monkeypatch): + monkeypatch.setattr(sys, "argv", ["prog", "-g", "C:\\game", "--no-game-path-caching", "dir1"]) + args = cli.parse_args() + assert args.no_game_path_caching is True + + +class TestSetupConsoleIo: + def test_replaces_none_stdout_and_stderr(self, monkeypatch): + monkeypatch.setattr(sys, "stdout", None) + monkeypatch.setattr(sys, "stderr", None) + + cli.setup_console_io() + + assert sys.stdout is not None + assert sys.stderr is not None + + def test_leaves_existing_stdio_untouched(self, monkeypatch, capsys): + original_stdout = sys.stdout + original_stderr = sys.stderr + + cli.setup_console_io() + + assert sys.stdout is original_stdout + assert sys.stderr is original_stderr + + +class TestMain: + def test_dispatches_to_run_cli_when_patches_given(self, monkeypatch, tmp_path): + monkeypatch.setattr(sys, "argv", ["prog", str(tmp_path)]) + called = {} + monkeypatch.setattr( + cli, "run_cli", lambda game_path, patch_dirs: called.setdefault("run_cli", (game_path, patch_dirs)) + ) + monkeypatch.setattr(gui, "run_gui", lambda: called.setdefault("run_gui", True)) + + cli.main() + + assert called["run_cli"] == (None, [str(tmp_path)]) + assert "run_gui" not in called + + def test_dispatches_to_run_gui_when_no_args(self, monkeypatch): + monkeypatch.setattr(sys, "argv", ["prog"]) + called = {} + monkeypatch.setattr(gui, "run_gui", lambda: called.setdefault("run_gui", True)) + + cli.main() + + assert called.get("run_gui") is True + + def test_caches_game_path_before_running_cli(self, monkeypatch, tmp_path): + (tmp_path / "bundles.nxa").touch() + monkeypatch.setattr(sys, "argv", ["prog", "-g", str(tmp_path), str(tmp_path)]) + called = {} + monkeypatch.setattr(cli, "set_cached_game_data_path", lambda p: called.setdefault("cached", p)) + monkeypatch.setattr(cli, "run_cli", lambda g, p: called.setdefault("run_cli", (g, p))) + + cli.main() + + assert called["cached"] == str(tmp_path) + assert called["run_cli"] == (str(tmp_path), [str(tmp_path)]) + + def test_exits_with_error_for_invalid_game_path(self, monkeypatch, tmp_path, capsys): + monkeypatch.setattr(sys, "argv", ["prog", "-g", str(tmp_path), str(tmp_path)]) + + with pytest.raises(SystemExit) as exc: + cli.main() + + assert exc.value.code == 1 + assert "does not look like a Helldivers II data folder" in capsys.readouterr().err + + def test_no_game_path_caching_skips_saving_path(self, monkeypatch, tmp_path): + (tmp_path / "bundles.nxa").touch() + monkeypatch.setattr(sys, "argv", ["prog", "-g", str(tmp_path), "--no-game-path-caching", str(tmp_path)]) + called = {} + monkeypatch.setattr(cli, "set_cached_game_data_path", lambda p: called.setdefault("cached", p)) + monkeypatch.setattr(cli, "run_cli", lambda g, p: called.setdefault("run_cli", (g, p))) + + cli.main() + + assert "cached" not in called + assert called["run_cli"] == (str(tmp_path), [str(tmp_path)]) diff --git a/tests/test_settings.py b/tests/test_settings.py new file mode 100644 index 0000000..9091d94 --- /dev/null +++ b/tests/test_settings.py @@ -0,0 +1,50 @@ +import json + +import settings + + +def _use_temp_settings_path(monkeypatch, tmp_path): + path = tmp_path / "settings.json" + monkeypatch.setattr(settings, "_settings_path", lambda: path) + return path + + +def test_load_settings_returns_empty_dict_when_file_missing(monkeypatch, tmp_path): + _use_temp_settings_path(monkeypatch, tmp_path) + assert settings.load_settings() == {} + + +def test_load_settings_returns_empty_dict_on_invalid_json(monkeypatch, tmp_path): + path = _use_temp_settings_path(monkeypatch, tmp_path) + path.write_text("{not valid json") + assert settings.load_settings() == {} + + +def test_save_settings_then_load_settings_round_trips(monkeypatch, tmp_path): + _use_temp_settings_path(monkeypatch, tmp_path) + settings.save_settings({"game_data_path": "C:\\game"}) + assert settings.load_settings() == {"game_data_path": "C:\\game"} + + +def test_save_settings_creates_parent_directories(monkeypatch, tmp_path): + path = tmp_path / "nested" / "dir" / "settings.json" + monkeypatch.setattr(settings, "_settings_path", lambda: path) + settings.save_settings({"a": 1}) + assert path.exists() + assert json.loads(path.read_text()) == {"a": 1} + + +def test_get_cached_game_data_path_returns_none_when_unset(monkeypatch, tmp_path): + _use_temp_settings_path(monkeypatch, tmp_path) + assert settings.get_cached_game_data_path() is None + + +def test_set_cached_game_data_path_preserves_other_keys(monkeypatch, tmp_path): + path = _use_temp_settings_path(monkeypatch, tmp_path) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"other_key": "keep me"})) + + settings.set_cached_game_data_path("D:\\HD2\\data") + + assert settings.get_cached_game_data_path() == "D:\\HD2\\data" + assert settings.load_settings()["other_key"] == "keep me" diff --git a/tests/test_update_unit_mods.py b/tests/test_update_unit_mods.py new file mode 100644 index 0000000..343c30f --- /dev/null +++ b/tests/test_update_unit_mods.py @@ -0,0 +1,81 @@ +import os + +import update_unit_mods as uum + + +class TestIsValidGameDataPath: + def test_false_when_directory_missing(self, tmp_path): + assert uum.is_valid_game_data_path(str(tmp_path / "missing")) is False + + def test_false_when_neither_marker_present(self, tmp_path): + assert uum.is_valid_game_data_path(str(tmp_path)) is False + + def test_true_when_legacy_marker_present(self, tmp_path): + (tmp_path / uum.LEGACY_MARKER_FILE).touch() + assert uum.is_valid_game_data_path(str(tmp_path)) is True + + def test_true_when_slim_marker_present(self, tmp_path): + (tmp_path / uum.SLIM_MARKER_FILE).touch() + assert uum.is_valid_game_data_path(str(tmp_path)) is True + + +class TestFindPatchFiles: + def test_finds_patch_files_recursively(self, tmp_path): + (tmp_path / "a.patch_0").touch() + (tmp_path / "sub").mkdir() + (tmp_path / "sub" / "b.patch_1").touch() + (tmp_path / "c.txt").touch() + + found = uum.find_patch_files(str(tmp_path)) + + assert sorted(os.path.basename(f) for f in found) == ["a.patch_0", "b.patch_1"] + + def test_empty_list_when_no_patch_files(self, tmp_path): + (tmp_path / "c.txt").touch() + assert uum.find_patch_files(str(tmp_path)) == [] + + +class TestProcessPatchFiles: + def test_aggregates_results_by_status_code(self, monkeypatch): + def fake_update(path): + return { + "u.patch": (uum.UPDATE_SUCCESS, "u.patch"), + "n.patch": (uum.NO_UNIT_FILES, "n.patch"), + "c.patch": (uum.CORRUPTED_FILE, "c.patch"), + }[path] + + monkeypatch.setattr(uum, "update_patch_file", fake_update) + + result = uum.process_patch_files(["u.patch", "n.patch", "c.patch"]) + + assert result.patches_found == 3 + assert result.updated == ["u.patch"] + assert result.no_units == ["n.patch"] + assert result.corrupted_files == ["c.patch"] + + def test_empty_patch_list_returns_empty_result(self): + assert uum.process_patch_files([]) == uum.PatchResult() + + +class TestProcessPatchFolder: + def test_includes_directory_and_patch_count(self, tmp_path, monkeypatch): + (tmp_path / "a.patch_0").touch() + monkeypatch.setattr(uum, "update_patch_file", lambda p: (uum.UPDATE_SUCCESS, p)) + + result = uum.process_patch_folder(str(tmp_path)) + + assert result.directory == str(tmp_path) + assert result.patches_found == 1 + assert len(result.updated) == 1 + + +class TestInitGameResources: + def test_sets_path_and_loads_resources(self, monkeypatch): + called = {} + monkeypatch.setattr(uum, "slim_init", lambda p: called.setdefault("slim_init", p)) + monkeypatch.setattr(uum, "load_game_resources", lambda: called.setdefault("loaded", True)) + + uum.init_game_resources("C:\\game\\data") + + assert uum.game_resource_path == "C:\\game\\data" + assert called == {"slim_init": "C:\\game\\data", "loaded": True} diff --git a/update_unit_mods.py b/update_unit_mods.py index 86062be..4c8e2d7 100644 --- a/update_unit_mods.py +++ b/update_unit_mods.py @@ -1,439 +1,403 @@ -import os -import struct -import sys -import concurrent.futures -import tkinter as tk -from tkinter import filedialog -from tkinter import messagebox -from pathlib import Path -from slim import slim_init, is_slim_version, load_package, get_package_toc, get_resource_from_bundle, get_resource_from_package - -game_resource_mapping = {} -game_resource_path = "" -directory = "" - -root = tk.Tk() -root.withdraw() - -UPDATE_SUCCESS = 0 -NO_UNIT_FILES = 1 -CORRUPTED_FILE = 2 - -print("fixing unit mods...") - -def select_folder(): - d = filedialog.askdirectory(title="Select folder containing patch files") - if d: - if not os.path.exists(d): - messagebox.showwarning(message="No valid folder selected!") - return False - else: - return None - return d - -def select_data_folder(): - d = filedialog.askdirectory(title="Select folder containing game data") - if d: - if not os.path.exists(d): - messagebox.showwarning(message="No valid folder selected!") - return False - if not os.path.exists(os.path.join(d, "9ba626afa44a3aa3")) and not os.path.exists(os.path.join(d, "bundles.nxa")): - messagebox.showwarning(message="Unable to find Helldivers II game data at this location; make sure you select the `data` folder in your Helldivers II install") - return False - else: - return None - return d - -class TocHeader: - - def __init__(self): - pass - - def from_bytes(self, bytes): - (self.file_id, - self.type_id, - self.toc_data_offset, - self.stream_file_offset, - self.gpu_resource_offset, - self.unknown1, - self.unknown2, - self.toc_data_size, - self.stream_size, - self.gpu_resource_size, - self.unknown3, - self.unknown4, - self.entry_index) = struct.unpack(" len(self.data): - missing_bytes = self.location - len(self.data) - self.data += bytearray(missing_bytes) - - def tell(self): # Get Position In Stream - return self.location - - def read(self, length=-1): # read Bytes From Stream - if length == -1: - length = len(self.data) - self.location - if self.location + length > len(self.data): - raise Exception("reading past end of stream") - - newData = self.data[self.location:self.location+length] - self.location += length - return bytearray(newData) - - def advance(self, offset): - self.location += offset - if self.location < 0: - self.location = 0 - if self.location > len(self.data): - missing_bytes = self.location - len(self.data) - self.data += bytearray(missing_bytes) - - def insert(self, length): - self.data[self.location:self.location] = bytearray(length) - - def delete(self, length): - self.data[self.location:self.location+length] = b'' - - def write(self, bytes): # Write Bytes To Stream - length = len(bytes) - if self.location + length > len(self.data): - missing_bytes = (self.location + length) - len(self.data) - self.data += bytearray(missing_bytes) - self.data[self.location:self.location+length] = bytearray(bytes) - self.location += length - - def read_format(self, format, size): - format = self.endian+format - return struct.unpack(format, self.read(size))[0] - - def bytes(self, value, size = -1): - if size == -1: - size = len(value) - if len(value) != size: - value = bytearray(size) - - if self.is_reading(): - return bytearray(self.read(size)) - elif self.is_writing(): - self.write(value) - return bytearray(value) - return value - - def int8_read(self): - return self.read_format('b', 1) - - def uint8_read(self): - return self.read_format('B', 1) - - def int16_read(self): - return self.read_format('h', 2) - - def uint16_read(self): - return self.read_format('H', 2) - - def int32_read(self): - return self.read_format('i', 4) - - def uint32_read(self): - return self.read_format('I', 4) - - def int64_read(self): - return self.read_format('q', 8) - - def uint64_read(self): - return self.read_format('Q', 8) - - def float32_read(self): - return self.read_format('f', 4) - -def get_data_from_original_file(unit_id: int): - if is_slim_version(): - unit_data = get_resource_from_package(*game_resource_mapping[unit_id]) - unit_version = unit_data[0x2C:0x30] - lod_group_offset, joint_list_offset = struct.unpack_from(" file_size: - return (CORRUPTED_FILE, file_path) - headers.append([tocHeader, tocStart+n*80]) - stream.seek(tocStart) - header_offset_adjustment = 0 - temp_headers = [] - for header in headers: - header[1] += header_offset_adjustment - header_data, header_offset = header - if header_data.file_id not in game_resource_mapping and header_data.type_id == 16187218042980615487: - stream.seek(header_offset) - stream.delete(80) - numFiles -= 1 - num_resources -= 1 - header_offset_adjustment -= 80 - else: - temp_headers.append(header) - headers = temp_headers - for header in headers: - header[0].toc_data_offset += header_offset_adjustment - headers.sort(key=lambda h: h[0].toc_data_offset) - stream.seek(8) - stream.write(struct.pack(" 16: - stream.advance(-4) - stream.write(struct.pack(" 0: - stream.insert(size_difference) - else: - stream.delete(-size_difference) - # update offsets - stream.seek(header_data.toc_data_offset + size_offset + 0x34) - for _ in range(16): - offset = stream.uint32_read() - if offset != 0 and offset > lod_group_offset: - stream.advance(-4) - stream.write((offset + size_difference).to_bytes(4, "little")) - stream.seek(header_data.toc_data_offset + size_offset + lod_group_offset) - stream.write(lod_group_data) - size_offset += (size_difference) - tocFile.seek(0) - tocFile.write(stream.data) - tocFile.close() - return (UPDATE_SUCCESS, file_path) - -def update_all(): - futures = [] - executor = concurrent.futures.ThreadPoolExecutor() - patches = [] - no_units = [] - corrupted_files = [] - for root, dirs, files in os.walk(directory): - for file in files: - if "patch" in os.path.splitext(file)[1]: - patches.append(os.path.join(root, file)) - if len(patches) == 0: - messagebox.showwarning(message="No patch files found in folder!") - return - else: - messagebox.showinfo(message=f"Checking {len(patches)} patch files...") - for patch in patches: - futures.append(executor.submit(update_patch_file, patch)) - for index, future in enumerate(futures): - result = future.result() - if result[0] == CORRUPTED_FILE: - corrupted_files.append(result[1]) - if result[0] == NO_UNIT_FILES: - no_units.append(result[1]) - executor.shutdown() - patch_files_updated = len(patches) - len(no_units) - len(corrupted_files) - if len(corrupted_files) > 0: - m = f"Found {len(corrupted_files)} corrupted patch file(s)!" - for name in corrupted_files: - m += f"\n{os.path.normpath(name)}" - messagebox.showerror(message=m) - m = f"Update Complete!\nUpdated {patch_files_updated} patch file(s) that contained unit resources." - if len(no_units) > 0: - m += f"\n{len(no_units)} patch file(s) did not contain any unit resources and were skipped." - messagebox.showinfo(message=m) - -while True: - - if not game_resource_path: - game_resource_path = select_data_folder() - print(game_resource_path) - if game_resource_path == False: continue - if game_resource_path is None: - do_exit = messagebox.askyesnocancel(message="Would you like to quit?") - if do_exit: - sys.exit() - else: - continue - slim_init(game_resource_path) - load_game_resources() - - directory = select_folder() - if directory == False: continue - if directory is None: - do_exit = messagebox.askyesnocancel(message="Would you like to quit?") - if do_exit: - sys.exit() - else: - continue - update_all() \ No newline at end of file +import os +import struct +import concurrent.futures +from dataclasses import dataclass, field +from pathlib import Path +from slim import slim_init, is_slim_version, load_package, get_package_toc, get_resource_from_bundle, get_resource_from_package + +game_resource_mapping = {} +game_resource_path = "" + +UPDATE_SUCCESS = 0 +NO_UNIT_FILES = 1 +CORRUPTED_FILE = 2 + +# Marker files that identify a Helldivers II `data` folder: legacy installs +# contain the `9ba626afa44a3aa3` bundle, "slim" installs contain `bundles.nxa`. +LEGACY_MARKER_FILE = "9ba626afa44a3aa3" +SLIM_MARKER_FILE = "bundles.nxa" + +def is_valid_game_data_path(path: str) -> bool: + return os.path.isdir(path) and ( + os.path.exists(os.path.join(path, LEGACY_MARKER_FILE)) or + os.path.exists(os.path.join(path, SLIM_MARKER_FILE)) + ) + +class TocHeader: + + def __init__(self): + pass + + def from_bytes(self, bytes): + (self.file_id, + self.type_id, + self.toc_data_offset, + self.stream_file_offset, + self.gpu_resource_offset, + self.unknown1, + self.unknown2, + self.toc_data_size, + self.stream_size, + self.gpu_resource_size, + self.unknown3, + self.unknown4, + self.entry_index) = struct.unpack(" len(self.data): + missing_bytes = self.location - len(self.data) + self.data += bytearray(missing_bytes) + + def tell(self): # Get Position In Stream + return self.location + + def read(self, length=-1): # read Bytes From Stream + if length == -1: + length = len(self.data) - self.location + if self.location + length > len(self.data): + raise Exception("reading past end of stream") + + newData = self.data[self.location:self.location+length] + self.location += length + return bytearray(newData) + + def advance(self, offset): + self.location += offset + if self.location < 0: + self.location = 0 + if self.location > len(self.data): + missing_bytes = self.location - len(self.data) + self.data += bytearray(missing_bytes) + + def insert(self, length): + self.data[self.location:self.location] = bytearray(length) + + def delete(self, length): + self.data[self.location:self.location+length] = b'' + + def write(self, bytes): # Write Bytes To Stream + length = len(bytes) + if self.location + length > len(self.data): + missing_bytes = (self.location + length) - len(self.data) + self.data += bytearray(missing_bytes) + self.data[self.location:self.location+length] = bytearray(bytes) + self.location += length + + def read_format(self, format, size): + format = self.endian+format + return struct.unpack(format, self.read(size))[0] + + def bytes(self, value, size = -1): + if size == -1: + size = len(value) + if len(value) != size: + value = bytearray(size) + + if self.is_reading(): + return bytearray(self.read(size)) + elif self.is_writing(): + self.write(value) + return bytearray(value) + return value + + def int8_read(self): + return self.read_format('b', 1) + + def uint8_read(self): + return self.read_format('B', 1) + + def int16_read(self): + return self.read_format('h', 2) + + def uint16_read(self): + return self.read_format('H', 2) + + def int32_read(self): + return self.read_format('i', 4) + + def uint32_read(self): + return self.read_format('I', 4) + + def int64_read(self): + return self.read_format('q', 8) + + def uint64_read(self): + return self.read_format('Q', 8) + + def float32_read(self): + return self.read_format('f', 4) + +def get_data_from_original_file(unit_id: int): + if is_slim_version(): + unit_data = get_resource_from_package(*game_resource_mapping[unit_id]) + unit_version = unit_data[0x2C:0x30] + lod_group_offset, joint_list_offset = struct.unpack_from(" file_size: + return (CORRUPTED_FILE, file_path) + headers.append([tocHeader, tocStart+n*80]) + stream.seek(tocStart) + header_offset_adjustment = 0 + temp_headers = [] + for header in headers: + header[1] += header_offset_adjustment + header_data, header_offset = header + if header_data.file_id not in game_resource_mapping and header_data.type_id == 16187218042980615487: + stream.seek(header_offset) + stream.delete(80) + numFiles -= 1 + num_resources -= 1 + header_offset_adjustment -= 80 + else: + temp_headers.append(header) + headers = temp_headers + for header in headers: + header[0].toc_data_offset += header_offset_adjustment + headers.sort(key=lambda h: h[0].toc_data_offset) + stream.seek(8) + stream.write(struct.pack(" 16: + stream.advance(-4) + stream.write(struct.pack(" 0: + stream.insert(size_difference) + else: + stream.delete(-size_difference) + # update offsets + stream.seek(header_data.toc_data_offset + size_offset + 0x34) + for _ in range(16): + offset = stream.uint32_read() + if offset != 0 and offset > lod_group_offset: + stream.advance(-4) + stream.write((offset + size_difference).to_bytes(4, "little")) + stream.seek(header_data.toc_data_offset + size_offset + lod_group_offset) + stream.write(lod_group_data) + size_offset += (size_difference) + tocFile.seek(0) + tocFile.write(stream.data) + tocFile.close() + return (UPDATE_SUCCESS, file_path) + +def find_patch_files(directory: str): + patches = [] + for root, dirs, files in os.walk(directory): + for file in files: + if "patch" in os.path.splitext(file)[1]: + patches.append(os.path.join(root, file)) + return patches + +@dataclass +class PatchResult: + directory: str = "" + patches_found: int = 0 + updated: list = field(default_factory=list) + no_units: list = field(default_factory=list) + corrupted_files: list = field(default_factory=list) + +def process_patch_files(patches: list) -> PatchResult: + result = PatchResult(patches_found=len(patches)) + futures = [] + executor = concurrent.futures.ThreadPoolExecutor() + for patch in patches: + futures.append(executor.submit(update_patch_file, patch)) + for future in futures: + code, path = future.result() + if code == CORRUPTED_FILE: + result.corrupted_files.append(path) + elif code == NO_UNIT_FILES: + result.no_units.append(path) + else: + result.updated.append(path) + executor.shutdown() + return result + +def process_patch_folder(directory: str) -> PatchResult: + result = process_patch_files(find_patch_files(directory)) + result.directory = directory + return result \ No newline at end of file