Merge branch 'main' into pre-commit-ci-update-config

.github/workflows/python.yml

@@ -101,7 +101,7 @@ jobs:
 
       - name: Export
         run:
-          venv/bin/python -m open_samus_returns_rando --input-path "$SAMUS_RETURNS_PATH" --output-path export/ --input-json tests/test_files/starter_preset_patcher.json
+          venv/bin/python -m open_samus_returns_rando --input-path "$SAMUS_RETURNS_PATH_ROMS"/MSR.3ds --output-path export/ --input-json tests/test_files/starter_preset_patcher.json
 
   mypy:
     runs-on: 'ubuntu-latest'

pyproject.toml

@@ -16,7 +16,7 @@ classifiers = [
 ]
 requires-python = ">=3.10"
 dependencies = [
-    "mercury-engine-data-structures>=0.31.1",
+    "mercury-engine-data-structures>=0.33.0",
     "jsonschema>=4.0.0",
     "ips.py>=0.1.2",
 ]
@@ -59,7 +59,7 @@ lint.select = [
 ]
 
 # Version to target for generated code.
-target-version = "py39"
+target-version = "py310"
 
 [tool.ruff.lint.mccabe]
 # Flag errors (`C901`) whenever the complexity level exceeds 25.
@@ -73,10 +73,31 @@ typing = [
     "mypy"
 ]
 
+test = [
+    "pytest"
+]
+
 [tool.mypy]
 files = [
     "src/"
 ]
 follow_imports = "silent"
 disallow_untyped_defs = true
 local_partial_types = true
+
+[[tool.mypy.overrides]]
+module = "tests.*"
+disallow_untyped_defs = false
+check_untyped_defs = true
+disable_error_code = "method-assign"
+
+[tool.pytest.ini_options]
+minversion = "6.0"
+testpaths = [
+    "tests",
+]
+xfail_strict = true
+filterwarnings = [
+    "error",
+    "ignore::DeprecationWarning",
+]

src/open_samus_returns_rando/__init__.py

@@ -3,9 +3,9 @@
 from .patch_util import patch_with_status_update
 
 
-def patch(input_path: Path, input_exheader: Path | None, output_path: Path, configuration: dict) -> None:
+def patch(input_path: Path, output_path: Path, configuration: dict) -> None:
     from .samus_returns_patcher import patch_extracted
-    return patch_extracted(input_path, input_exheader, output_path, configuration)
+    return patch_extracted(input_path, output_path, configuration)
 
 
 __all__ = [

src/open_samus_returns_rando/cli.py

@@ -64,7 +64,6 @@ def main() -> None:
     start = time.time()
     samus_returns_patcher.patch_extracted(
         args.input_path,
-        args.input_exheader,
         args.output_path,
         configuration,
     )

src/open_samus_returns_rando/files/schema.json

@@ -623,15 +623,6 @@
             "type": "boolean",
             "default": false,
             "description": "When true, the game will start a socket that receives lua code to run"
-        },
-        "region": {
-            "type": "string",
-            "enum": [
-                "ntsc",
-                "pal"
-            ],
-            "default": "pal",
-            "description": "This is only required if enable_remote_lua is true to copy the NTSC or PAL bps patch."
         }
     },
     "required": [

src/open_samus_returns_rando/multiworld_integration.py

@@ -33,12 +33,13 @@ def get_lua_for_item(progression: list[list[dict[str, str | int]]]) -> str:
 
 
 def create_exefs_patches(
-        out_code: Path, out_exheader: Path, input_exheader: Path | None, enabled: bool, region: str
+        out_code: Path, out_exheader: Path, input_code: bytes | None, input_exheader: bytes, enabled: bool
     ) -> None:
-    if not enabled or input_exheader is None:
+    if not enabled:
         return
 
-    import shutil
+    if input_code is None:
+        raise ValueError("Could not get decompressed + decrypted code binary")
 
     import ips  # type: ignore
 
@@ -48,12 +49,18 @@ def create_exefs_patches(
     out_exheader.parent.mkdir(parents=True, exist_ok=True)
     with (
           Path.open(exheader_ips_path, "rb") as exheader_ips,
-          Path.open(input_exheader, "rb") as original,
           Path.open(out_exheader, "wb") as result
         ):
         content = exheader_ips.read()
         patch = ips.Patch.load(content)
-        patch.apply(original, result)
+        patch.apply(input_exheader, result)
 
-    # copy bps patch (don't ask me why the patch does not work as IPS format)
-    shutil.copyfile(files_path().joinpath("exefs_patches", f"code_{region}.bps"), out_code)
+    code_ips_path = files_path().joinpath("exefs_patches", "code.ips")
+    out_code.parent.mkdir(parents=True, exist_ok=True)
+    with (
+          Path.open(code_ips_path, "rb") as code_ips,
+          Path.open(out_code, "wb") as result
+        ):
+        content = code_ips.read()
+        patch = ips.Patch.load(content)
+        patch.apply(input_code, result)

src/open_samus_returns_rando/patch_util.py

@@ -5,7 +5,7 @@
 from open_samus_returns_rando.logger import LOG
 
 
-def patch_with_status_update(input_path: Path, input_exheader: Path | None, output_path: Path, configuration: dict,
+def patch_with_status_update(input_path: Path, output_path: Path, configuration: dict,
                              status_update: typing.Callable[[float, str], None]) -> None:
     from open_samus_returns_rando.samus_returns_patcher import patch_extracted
     # messages depends on the layout but it is a good approximation
@@ -32,7 +32,7 @@ def emit(self, record: logging.LogRecord) -> None:
         LOG.handlers.insert(0, new_handler)
         LOG.propagate = False
 
-        patch_extracted(input_path, input_exheader, output_path, configuration)
+        patch_extracted(input_path, output_path, configuration)
         if new_handler.count < total_logs:
             status_update(1, f"Done was {new_handler.count}")
 

src/open_samus_returns_rando/patcher_editor.py

@@ -1,13 +1,15 @@
 import copy
 import typing
-from pathlib import Path
 
 from construct import Container  # type: ignore[import-untyped]
+from mercury_engine_data_structures.base_resource import BaseResource
 from mercury_engine_data_structures.file_tree_editor import FileTreeEditor
-from mercury_engine_data_structures.formats import BaseResource, Bmsld
+from mercury_engine_data_structures.formats import Bmsld
 from mercury_engine_data_structures.game_check import Game
 
 from open_samus_returns_rando.constants import ALL_SCENARIOS, get_package_name
+from open_samus_returns_rando.romfs.packaged_romfs import PackagedRomFs
+from open_samus_returns_rando.romfs.rom3ds import Rom3DS
 
 T = typing.TypeVar("T")
 
@@ -19,8 +21,8 @@ def path_for_level(level_name: str) -> str:
 class PatcherEditor(FileTreeEditor):
     memory_files: dict[str, BaseResource]
 
-    def __init__(self, root: Path):
-        super().__init__(root, Game.SAMUS_RETURNS)
+    def __init__(self, parsed_rom: Rom3DS):
+        super().__init__(PackagedRomFs(parsed_rom), Game.SAMUS_RETURNS)
         self.memory_files = {}
 
     def get_file(self, path: str, type_hint: type[T] = BaseResource) -> T: # type: ignore

src/open_samus_returns_rando/romfs/lzss.py

@@ -0,0 +1,71 @@
+# Lzss structure was taken from GodMode9 https://github.com/d0k3/GodMode9/commit/a6a15eb70d66e3c96bbc164598f9482bb545b3f9
+
+from construct import Struct  # type: ignore
+from construct.core import Int32ul  # type: ignore
+
+LzssFooter = Struct(
+    "off_size_comp" / Int32ul,  # 0xOOSSSSSS, where O == reverse offset and S == size
+    "addsize_dec" / Int32ul,  # decompressed size - compressed size
+)
+
+
+def lzss_decompress(compressed: bytes) -> bytes | None:
+    # copy of citra's `LZSS_Decompress` function
+    # see: https://github.com/PabloMK7/citra/blob/7d00f47c5ead75db0a9f24d70aa4b609e85125d8/src/core/file_sys/ncch_container.cpp#L54
+    compressed_length = len(compressed)
+
+    lzss_footer = LzssFooter.parse(compressed[-8:])
+
+    decompressed_length = compressed_length + lzss_footer.addsize_dec
+    decompressed = bytearray(decompressed_length)
+
+    out = decompressed_length
+    index = compressed_length - ((lzss_footer.off_size_comp >> 24) & 0xFF)
+    stop_index = compressed_length - (lzss_footer.off_size_comp & 0xFFFFFF)
+
+    decompressed[0:compressed_length] = compressed
+    while index > stop_index:
+        index -= 1
+        control = compressed[index]
+
+        for i in range(8):
+            if index <= stop_index:
+                break
+            if index <= 0:
+                break
+            if out <= 0:
+                break
+
+            if control & 0x80:
+                # Check if compression is out of bounds
+                if index < 2:
+                    return None
+                index -= 2
+
+                segment_offset = compressed[index] | (compressed[index + 1] << 8)
+                segment_size = ((segment_offset >> 12) & 15) + 3
+                segment_offset &= 0x0FFF
+                segment_offset += 2
+
+                # Check if compression is out of bounds
+                if out < segment_size:
+                    return None
+
+                for j in range(segment_size):
+                    # Check if compression is out of bounds
+                    if out + segment_offset >= decompressed_length:
+                        return None
+
+                    data = decompressed[out + segment_offset]
+                    out -= 1
+                    decompressed[out] = data
+            else:
+                # Check if compression is out of bounds
+                if out < 1:
+                    return None
+                out -= 1
+                index -= 1
+                decompressed[out] = compressed[index]
+
+            control <<= 1
+    return decompressed

src/open_samus_returns_rando/romfs/packaged_romfs.py

@@ -0,0 +1,32 @@
+import io
+from collections.abc import Iterator
+from contextlib import contextmanager
+
+import construct  # type: ignore
+from mercury_engine_data_structures.romfs import RomFs
+
+from open_samus_returns_rando.romfs.rom3ds import Rom3DS
+
+
+class PackagedRomFs(RomFs):
+    def __init__(self, parsed_rom: Rom3DS):
+        self.parsed_rom = parsed_rom
+
+    @contextmanager
+    def get_pkg_stream(self, file_path: str) -> Iterator[io.BytesIO]:
+        file_stream = io.BytesIO(self.parsed_rom.get_file_binary(file_path))
+        try:
+            yield file_stream
+        finally:
+            file_stream.close()
+
+    def read_file_with_entry(self, file_path: str, entry: construct.Container) -> bytes:
+        with io.BytesIO(self.parsed_rom.get_file_binary(file_path)) as f:
+            f.seek(entry.start_offset)
+            return f.read(entry.end_offset - entry.start_offset)
+
+    def get_file(self, file_path: str) -> bytes:
+        return self.parsed_rom.get_file_binary(file_path)
+
+    def all_files(self) -> Iterator[str]:
+        yield from self.parsed_rom.file_name_to_entry.keys()