(archives.id_software)(#191) Pak refactor

bsp_tool/archives/cdrom.py

@@ -8,8 +8,8 @@
 import os
 from typing import List
 
-from . import base
 from ..utils import binary
+from . import base
 
 
 # NOTE: Logical Block Address -> Byte Address:

bsp_tool/archives/id_software.py

@@ -3,18 +3,29 @@
 import struct
 from typing import Dict, List
 
+from ..utils import binary
 from . import base
 from . import pkware
 
 
+# NOTE: can't use branches.base.Struct without making a circular import
 class PakFileEntry:
-    def __init__(self, filepath: str, offset: int, size: int):
-        self.filepath = filepath
+    filename: bytes  # paintext
+    offset: int
+    length: int
+    __slots__ = ["filename", "offset", "length"]
+    _format = "56s2I"
+
+    def __init__(self, filename: str, offset: int, length: int):
+        self.filename = filename
         self.offset = offset
-        self.size = size
+        self.length = length
 
     def __repr__(self) -> str:
-        return f"PakFileEntry(filepath={self.filepath!r}, offset={self.offset}, size={self.size})"
+        attrs = ", ".join(
+            f"{attr}={getattr(self, attr)!r}"
+            for attr in ("filename, offset, length"))
+        return f"PakFileEntry({attrs})"
 
     @classmethod
     def from_stream(cls, stream: io.BytesIO) -> PakFileEntry:
@@ -24,40 +35,42 @@ def from_stream(cls, stream: io.BytesIO) -> PakFileEntry:
 class Pak(base.Archive):
     # https://quakewiki.org/wiki/.pak
     ext = "*.pak"
-    files: Dict[str, PakFileEntry]
+    _file: io.BytesIO
+    entries: Dict[str, PakFileEntry]
 
     def __init__(self):
-        self.files = dict()
+        self.entries = dict()
 
     def __repr__(self) -> str:
-        return f"<{self.__class__.__name__} {len(self.files)} files @ 0x{id(self):016X}>"
+        descriptor = f"{len(self.entries)} files"
+        return f"<{self.__class__.__name__} {descriptor} @ 0x{id(self):016X}>"
 
     def read(self, filepath: str) -> bytes:
-        assert filepath in self.files
-        entry = self.files[filepath]
-        self.file.seek(entry.offset)
-        return self.file.read(entry.size)
+        assert filepath in self.entries
+        entry = self.entries[filepath]
+        self._file.seek(entry.offset)
+        return self._file.read(entry.length)
 
     def namelist(self) -> List[str]:
-        return sorted(self.files.keys())
+        return sorted(self.entries.keys())
 
     @classmethod
     def from_stream(cls, stream: io.BytesIO) -> Pak:
         out = cls()
-        out.file = stream
-        assert out.file.read(4) == b"PACK", "not a .pak file"
+        out._file = stream
+        assert out._file.read(4) == b"PACK", "not a .pak file"
         # file table
-        offset, size = struct.unpack("2I", out.file.read(8))
-        assert size % 64 == 0, "unexpected file table size"
-        out.file.seek(offset)
-        out.files = {
-            entry.filepath.split(b"\0")[0].decode(): entry
+        offset, length = binary.read_struct(out._file, "2I")
+        sizeof_entry = 64
+        assert length % sizeof_entry == 0, "unexpected file table size"
+        out._file.seek(offset)
+        out.entries = {
+            entry.filename.partition(b"\0")[0].decode("ascii"): entry
             for entry in [
-                PakFileEntry.from_stream(out.file)
-                for i in range(size // 64)]}
+                PakFileEntry.from_stream(out._file)
+                for i in range(length // sizeof_entry)]}
         return out
 
 
 class Pk3(pkware.Zip):
-    """IdTech .bsps are stored in .pk3 files, which are basically .zip archives"""
     ext = "*.pk3"

bsp_tool/branches/valve/source.py

@@ -986,7 +986,7 @@ def from_bytes(cls, raw_lump: bytes):
                 "WORLD_LIGHTS_HDR":          {0: WorldLight}}
 
 SPECIAL_LUMP_CLASSES = {"ENTITIES":                 {0: shared.Entities},
-                        "PAKFILE":                  {0: archives.id_software.Pk3},
+                        "PAKFILE":                  {0: archives.pkware.Zip},
                         # "PHYSICS_COLLIDE":          {0: physics.CollideLump},  # BROKEN .as_bytes()
                         "PHYSICS_DISPLACEMENT":     {0: physics.Displacement},
                         "TEXTURE_DATA_STRING_DATA": {0: TextureDataStringData},

tests/archives/id_software/test_Pak.py

@@ -0,0 +1,69 @@
+import fnmatch
+import os
+
+import pytest
+
+from bsp_tool.archives import id_software
+
+from ... import utils
+
+
+paks = dict()
+archive = utils.archive_dirs()
+if archive.steam_dir is not None:
+    # {"top_dir", {"game": {"mod": ["pak_dirs"]}}}
+    pak_dirs = {
+        archive.steam_dir: {
+            "Hexen 2": {
+                "Core Game": ["data1"]},
+            "Quake": {
+                "Alkaline": ["alk1.1", "alkaline", "alkaline_dk"],
+                "Arcane Dimensions": ["ad"],
+                "Copper": ["copper"],  # Underdark Overbright
+                "Core Game": ["hipnotic", "Id1", "rogue"],
+                "Prototype Jam #3": ["sm220"]},
+            "Quake/rerelease": {
+                "Core Game": ["hipnotic", "id1", "rogue"],
+                "Dimension of the Past": ["dopa"],
+                "New": ["ctf", "mg1"]},
+            "Quake 2": {
+                "Core Game": ["baseq2", "ctf", "rogue", "xatrix", "zaero"]},
+            "Quake 2/rerelease": {
+                "Core Game": ["baseq2"]}},
+        archive.gog_dir: {
+            "Soldier of Fortune": {
+                "Core Game": ["base"]}}}
+    if None in pak_dirs:  # skip top_dirs not in this ArchivistStash
+        pak_dirs.pop(None)
+    # NOTE: "mod" is to give clear names to each group of mod folders
+    # -- not part of the path
+    # NOTE: Daikatana uses a different .pak format
+    # TODO: Heretic II
+    # TODO: Quake64 in %USERPROFILE%
+
+    def paks_of(top_dir: str, game: str, pak_dir: str):
+        full_dir = os.path.join(top_dir, game, pak_dir)
+        pak_filenames = fnmatch.filter(os.listdir(full_dir), "*.pak")
+        pak_full_paths = [
+            os.path.join(full_dir, pak_filename)
+            for pak_filename in pak_filenames]
+        return list(zip(pak_filenames, pak_full_paths))
+
+    paks = {
+        f"{game} | {mod} ({pak_dir}) | {pak_filename}": pak_full_path
+        for top_dir, games in pak_dirs.items()
+        for game, mods in games.items()
+        for mod, pak_dirs in mods.items()
+        for pak_dir in pak_dirs
+        for pak_filename, pak_full_path in paks_of(top_dir, game, pak_dir)}
+    # TODO: if a game is not installed on this device, skip it
+
+
+@pytest.mark.parametrize("filename", paks.values(), ids=paks.keys())
+def test_from_file(filename: str):
+    pak = id_software.Pak.from_file(filename)
+    namelist = pak.namelist()
+    assert isinstance(namelist, list), ".namelist() failed"
+    if len(namelist) != 0:
+        first_file = pak.read(namelist[0])
+        assert isinstance(first_file, bytes), ".read() failed"

tests/archives/valve/test_Vpk.py

@@ -8,26 +8,23 @@
 from ... import utils
 
 
-steam_common = "D:/SteamLibrary/steamapps/common/"
-vpk_dirs = {
-    "Black Mesa": "bms/",
-    "Bloody Good Time": "vpks/",
-    "Contagion": "vpks/",
-    "Dark Messiah Might and Magic Single Player": "vpks/",
-    "Dark Messiah Might and Magic Multi-Player": "vpks/",
-    "SiN Episodes Emergence": "vpks/",
-    "The Ship": "vpks/",
-    "The Ship Single Player": "vpks/",
-    "The Ship Tutorial": "vpks/"}
-# NOTE: Dark Messiah Singleplayer | depot_2109_dir.vpk is empty
-# -- https://steamdb.info/depot/2109/ (mm_media)
-# NOTE: The Ship | depot_2402_dir.vpk contains "umlaut e" b"\xEB" (latin_1)
-# -- https://steamdb.info/depot/2402/ (The Ship Common)
-
-
 vpks = dict()
 archive = utils.archive_dirs()
 if archive.steam_dir is not None:
+    vpk_dirs = {
+        "Black Mesa": "bms/",
+        "Bloody Good Time": "vpks/",
+        "Contagion": "vpks/",
+        "Dark Messiah Might and Magic Single Player": "vpks/",
+        "Dark Messiah Might and Magic Multi-Player": "vpks/",
+        "SiN Episodes Emergence": "vpks/",
+        "The Ship": "vpks/",
+        "The Ship Single Player": "vpks/",
+        "The Ship Tutorial": "vpks/"}
+    # NOTE: Dark Messiah Singleplayer | depot_2109_dir.vpk is empty
+    # -- https://steamdb.info/depot/2109/ (mm_media)
+    # NOTE: The Ship | depot_2402_dir.vpk contains "umlaut e" b"\xEB" (latin_1)
+    # -- https://steamdb.info/depot/2402/ (The Ship Common)
     for game, vpk_dir in vpk_dirs.items():
         vpk_dir = os.path.join(archive.steam_dir, game, vpk_dir)
         for vpk_filename in fnmatch.filter(os.listdir(vpk_dir), "*_dir.vpk"):

tests/maplist.py

@@ -164,9 +164,9 @@
                  "update101.pk3/maps",  # 4 maps | 22 MB | .pk3
                  "update102.pk3/maps",  # 5 maps | 51 MB | .pk3
                  "update103.pk3/maps"],  # 4 maps | 31 MB | .pk3
-        "StarTrekEliteForce": ["pak0/maps",  # 67 maps | 334 MB | .pak
-                               "pak1/maps",  # 4 maps | 20 MB | .pak
-                               "pak3/maps"],  # 22 maps | 107 MB | .pak
+        "StarTrekEliteForce": ["pak0/maps",  # 67 maps | 334 MB | .pk3
+                               "pak1/maps",  # 4 maps | 20 MB | .pk3
+                               "pak3/maps"],  # 22 maps | 107 MB | .pk3
         "QuakeII": ["pak0/maps",  # 39 maps | 89 MB | .pak
                     "pak1/maps",  # 8 maps | 10 MB | .pak
                     "zaero/pak0/maps"],  # 14 maps | 36 MB | .pak

tests/utils.py

@@ -15,27 +15,31 @@
 # TODO: we can't safely assume cwd() is tests/../
 # -- use __file__ to get the test maps dir instead
 
-ArchiveDirs = collections.namedtuple("ArchiveDirs", ["steam_dir", "mod_dir"])
+ArchivistStash = collections.namedtuple(
+    "ArchivistStash", [
+        "steam_dir", "mod_dir", "gog_dir"])
 
 archivist_aliases = {"Jared@ITANI_WAYSOUND": "bikkie"}
 
 archivists = {
     # Windows Desktop
-    ("bikkie", "ITANI_WAYSOUND"): ArchiveDirs(
+    ("bikkie", "ITANI_WAYSOUND"): ArchivistStash(
         "D:/SteamLibrary/steamapps/common/",
-        "E:/Mod/"),
+        "E:/Mod/",
+        "D:/GoG Galaxy/Games"),
     # Linux Laptop
-    ("bikkie", "coplandbentokom-9876"): ArchiveDirs(
+    ("bikkie", "coplandbentokom-9876"): ArchivistStash(
         None,
-        "/media/bikkie/3964-39352/Mod/")}
+        "/media/bikkie/3964-39352/Mod/",
+        None)}
 
 
 def archive_available() -> bool:
     return archivist_login() in archivists
 
 
-def archive_dirs() -> ArchiveDirs:
-    return archivists.get(archivist_login(), ArchiveDirs(*[None] * 2))
+def archive_dirs() -> ArchivistStash:
+    return archivists.get(archivist_login(), ArchivistStash(*[None] * 3))
 
 
 def archivist_login() -> (str, str):