util: Replace file collection lambdas with file entry objects.

Author: heinezen

openage/cabextract/cab.py

@@ -1,4 +1,4 @@
-# Copyright 2015-2023 the openage authors. See copying.md for legal info.
+# Copyright 2015-2024 the openage authors. See copying.md for legal info.
 
 """
 Provides CABFile, an extractor for the MSCAB format.
@@ -18,7 +18,7 @@
 from ..util.filelike.readonly import PosSavingReadOnlyFileLikeObject
 from ..util.filelike.stream import StreamFragment
 from ..util.files import read_guaranteed, read_nullterminated_string
-from ..util.fslike.filecollection import FileCollection
+from ..util.fslike.filecollection import FileCollection, FileEntry
 from ..util.math import INF
 from ..util.strings import try_decode
 from ..util.struct import NamedStruct, Flags
@@ -216,6 +216,28 @@ def verify_checksum(self) -> Union[None, NoReturn]:
             raise ValueError("checksum error in MSCAB data block")
 
 
+class CABEntry(FileEntry):
+    """
+    Entry in a CAB file.
+    """
+
+    def __init__(self, fileobj: CFFile):
+        self.fileobj = fileobj
+
+    def open_r(self):
+        return StreamFragment(
+            self.fileobj.folder.plain_stream,
+            self.fileobj.pos,
+            self.fileobj.size
+        )
+
+    def size(self) -> int:
+        return self.fileobj.size
+
+    def mtime(self) -> float:
+        return self.fileobj.timestamp
+
+
 class CABFile(FileCollection):
     """
     The actual file system-like CAB object.
@@ -275,20 +297,9 @@ def __init__(self, cab: FileLikeObject, offset: int = 0):
                     "CABFile has multiple entries with the same path: " +
                     b'/'.join(fileobj.path).decode())
 
-            def open_r(fileobj=fileobj):
-                """ Returns a opened ('rb') file-like object for fileobj. """
-                return StreamFragment(
-                    fileobj.folder.plain_stream,
-                    fileobj.pos,
-                    fileobj.size
-                )
-
-            self.add_fileentry(fileobj.path, (
-                open_r,
-                None,
-                lambda fileobj=fileobj: fileobj.size,
-                lambda fileobj=fileobj: fileobj.timestamp
-            ))
+            file_entry = CABEntry(fileobj)
+
+            self.add_fileentry(fileobj.path, file_entry)
 
     def __repr__(self):
         return "CABFile"

openage/convert/value_object/read/media/drs.py

@@ -1,4 +1,4 @@
-# Copyright 2013-2022 the openage authors. See copying.md for legal info.
+# Copyright 2013-2024 the openage authors. See copying.md for legal info.
 
 """
 Code for reading Genie .DRS archives.
@@ -12,7 +12,7 @@
 
 from .....log import spam, dbg
 from .....util.filelike.stream import StreamFragment
-from .....util.fslike.filecollection import FileCollection
+from .....util.fslike.filecollection import FileCollection, FileEntry
 from .....util.strings import decode_until_null
 from .....util.struct import NamedStruct
 
@@ -87,6 +87,23 @@ class DRSFileInfo(NamedStruct):
     file_size        = "i"
 
 
+class DRSEntry(FileEntry):
+    """
+    Entry in a DRS archive.
+    """
+
+    def __init__(self, fileobj: GuardedFile, offset: int, size: int):
+        self.fileobj = fileobj
+        self.offset = offset
+        self.entry_size = size
+
+    def open_r(self):
+        return StreamFragment(self.fileobj, self.offset, self.entry_size)
+
+    def size(self) -> int:
+        return self.entry_size
+
+
 class DRS(FileCollection):
     """
     represents a file archive in DRS format.
@@ -133,14 +150,9 @@ def __init__(self, fileobj: GuardedFile, game_version: GameVersion):
             self.tables.append(table_header)
 
         for filename, offset, size in self.read_tables():
-            def open_r(offset=offset, size=size):
-                """ Returns a opened ('rb') file-like object for fileobj. """
-                return StreamFragment(self.fileobj, offset, size)
-
-            self.add_fileentry(
-                [filename.encode()],
-                (open_r, None, lambda size=size: size, None)
-            )
+            file_entry = DRSEntry(self.fileobj, offset, size)
+
+            self.add_fileentry([filename.encode()], file_entry)
 
     def read_tables(self) -> typing.Generator[tuple[str, str, str], None, None]:
         """

openage/util/fslike/filecollection.py

@@ -1,9 +1,11 @@
-# Copyright 2015-2022 the openage authors. See copying.md for legal info.
+# Copyright 2015-2024 the openage authors. See copying.md for legal info.
 
 """
 Provides Filecollection, a utility class for combining multiple file-like
 objects to a FSLikeObject.
 """
+from __future__ import annotations
+import typing
 
 from collections import OrderedDict
 from io import UnsupportedOperation
@@ -12,6 +14,9 @@
 from .abstract import FSLikeObject
 from .path import Path
 
+if typing.TYPE_CHECKING:
+    from openage.util.filelike.stream import StreamFragment
+
 
 class FileCollection(FSLikeObject):
     """
@@ -59,14 +64,12 @@ def get_direntries(self, parts=None, create: bool = False) -> tuple[OrderedDict,
 
         return entries
 
-    def add_fileentry(self, parts, fileentry):
+    def add_fileentry(self, parts, fileentry: FileEntry):
         """
         Adds a file entry (and parent directory entries, if needed).
 
         This method should not be called directly; instead, use the
         add_file method of Path objects that were obtained from this.
-
-        fileentry must be open_r, open_w, size, mtime.
         """
         if not parts:
             raise IsADirectoryError("FileCollection.root is a directory")
@@ -79,11 +82,9 @@ def add_fileentry(self, parts, fileentry):
 
         entries[0][name] = fileentry
 
-    def get_fileentry(self, parts):
+    def get_fileentry(self, parts) -> FileEntry:
         """
         Gets a file entry. Helper method for internal use.
-
-        Returns open_r, open_w, size, mtime
         """
         if not parts:
             raise IsADirectoryError(
@@ -101,45 +102,45 @@ def get_fileentry(self, parts):
 
         return entries[0][name]
 
-    def open_r(self, parts) -> None:
-        open_r, _, _, _ = self.get_fileentry(parts)
+    def open_r(self, parts: list[bytes]) -> StreamFragment:
+        entry = self.get_fileentry(parts)
+
+        open_r = entry.open_r()
 
         if open_r is None:
             raise UnsupportedOperation(
                 "not readable: " +
                 b"/".join(parts).decode(errors='replace'))
 
-        return open_r()
+        return open_r
+
+    def open_w(self, parts: list[bytes]):
+        entry = self.get_fileentry(parts)
 
-    def open_w(self, parts) -> None:
-        _, open_w, _, _ = self.get_fileentry(parts)
+        open_w = entry.open_w()
 
         if open_w is None:
             raise UnsupportedOperation(
                 "not writable: " +
                 b"/".join(parts).decode(errors='replace'))
 
+        return open_w
+
     def list(self, parts):
         fileentries, subdirs = self.get_direntries(parts)
 
         yield from subdirs
         yield from fileentries
 
     def filesize(self, parts) -> int:
-        _, _, filesize, _ = self.get_fileentry(parts)
+        entry = self.get_fileentry(parts)
 
-        if filesize is None:
-            return None
-
-        return filesize()
+        return entry.size()
 
     def mtime(self, parts) -> float:
-        _, _, _, mtime = self.get_fileentry(parts)
-
-        if mtime is None:
-            return None
+        entry = self.get_fileentry(parts)
 
-        return mtime()
+        return entry.mtime()
 
     def mkdirs(self, parts) -> None:
         self.get_direntries(parts, create=True)
@@ -248,3 +249,34 @@ def add_file_from_path(self, path: Path) -> None:
             open_w = None
 
         self.add_file(path.open_r, open_w, path.filesize, path.mtime)
+
+
+class FileEntry:
+    """
+    Entry in a file collection archive.
+    """
+    # pylint: disable=no-self-use
+
+    def open_r(self) -> StreamFragment:
+        """
+        Returns a file-like object for reading.
+        """
+        raise UnsupportedOperation("FileEntry.open_r")
+
+    def open_w(self):
+        """
+        Returns a file-like object for writing.
+        """
+        raise UnsupportedOperation("FileEntry.open_w")
+
+    def size(self) -> int:
+        """
+        Returns the size of the entr<.
+        """
+        raise UnsupportedOperation("FileEntry.size")
+
+    def mtime(self) -> float:
+        """
+        Returns the modification time of the entry.
+        """
+        raise UnsupportedOperation("FileEntry.mtime")