lindbergh: squashfs support (via an overlay with write access)

Author: Dancin Forever

batocera-Changelog.md

@@ -29,7 +29,7 @@
 - Steering wheel support added for :
   - Logitech: G923 (Xbox), PRO Racing Wheel
   - Speedlink: 4in1 Leather Power Feedback Wheel
-- Sega Lindbergh loader
+- Sega Lindbergh loader with .squashfs support
 - Variable Refresh Rate (VRR) support for modern AMD gpus
 - Support of Shanwan Twin USB Joystick (new revision)
 - Libretro-PS2 core

package/batocera/core/batocera-configgen/configgen/configgen/emulatorlauncher.py

@@ -44,7 +44,8 @@
 def main(args: argparse.Namespace, maxnbplayers: int) -> int:
     # squashfs roms if squashed
     if args.rom.suffix == ".squashfs":
-        with squashfs_rom(args.rom) as rom:
+        writable_rom = (args.system == "lindbergh")
+        with squashfs_rom(args.rom, overlay=writable_rom) as rom:
             return start_rom(args, maxnbplayers, rom, args.rom)
     else:
         return start_rom(args, maxnbplayers, args.rom, args.rom)

package/batocera/core/batocera-configgen/configgen/configgen/generators/lindbergh/lindberghGenerator.py

@@ -67,6 +67,11 @@ def getHotkeysContext(self) -> HotkeysContext:
         }
 
     def generate(self, system, rom, playersControllers, metadata, guns, wheels, gameResolution):
+        # .squashfs returns mount point so rewire rom to first .game file found
+        # https://stackoverflow.com/a/43669828/9983389
+        if (rom.is_dir()):
+            rom = next(rom.rglob("*.game"))
+
         romDir = rom.parent
         romName = rom.name
         _logger.debug("ROM path: %s", romDir)

package/batocera/core/batocera-configgen/configgen/configgen/utils/squashfs.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import logging
+import os
+import shutil
 import subprocess
 from contextlib import contextmanager
 from pathlib import Path
@@ -16,28 +18,63 @@
 
 _SQUASHFS_DIR: Final = Path("/var/run/squashfs/")
 
+def _squashfs_removemount(mount_point: Path):
+    if mount_point.exists():
+        if mount_point.is_mount():
+            return_code = subprocess.call(["umount", mount_point])
+            if return_code != 0:
+                _logger.debug("squashfs_rom: unmounting %s failed", mount_point)
+                raise BatoceraException(f"Unable to unmount the file {mount_point}")
+        mount_point.rmdir()
+
+def _squashfs_rom_cleanup(mount_point: Path, overlay: bool):
+    _logger.debug("squashfs_rom: cleaning up %s", mount_point)
+    os.chdir(_SQUASHFS_DIR)
+    _squashfs_removemount(mount_point)
+
+    if overlay:
+        overlay_root = mount_point.parent
+        mount_point_rw = overlay_root / "overlay"
+
+        if overlay_root.exists():
+            _logger.debug("squashfs_rom: cleaning up overlay root %s", overlay_root)
+            _squashfs_removemount(mount_point_rw)
+            shutil.rmtree(overlay_root)
 
 @contextmanager
-def squashfs_rom(rom: Path, /) -> Iterator[Path]:
+def squashfs_rom(rom: Path, overlay: bool) -> Iterator[Path]:
     _logger.debug("squashfs_rom(%s)", rom)
-    mount_point = _SQUASHFS_DIR / rom.stem
 
     mkdir_if_not_exists(_SQUASHFS_DIR)
 
-    # first, try to clean an empty remaining directory (for example because of a crash)
-    if mount_point.exists() and mount_point.is_dir():
-        _logger.debug("squashfs_rom: %s already exists", mount_point)
-        # try to remove an empty directory, else, run the directory, ignoring the .squashfs
-        try:
-            mount_point.rmdir()
-        except (FileNotFoundError, OSError):
-            _logger.debug("squashfs_rom: failed to rmdir %s", mount_point)
-            yield mount_point
-            # No cleanup is necessary
-            return
+    if overlay:
+        overlay_root   = _SQUASHFS_DIR / rom.stem
+        mount_point    = overlay_root / "rom"
+        overlay_delta  = overlay_root / "delta"
+        overlay_tmp    = overlay_root / "tmp"
+        mount_point_rw = overlay_root / "overlay"
+
+        if overlay_root.exists() and overlay_root.is_dir():
+            _logger.debug("squashfs_rom: %s overlay root already exists", overlay_root)
+            _squashfs_rom_cleanup(mount_point, overlay)
+    else:
+        mount_point    = _SQUASHFS_DIR / rom.stem
+
+        # first, try to clean an empty remaining directories (for example because of a crash)
+        if mount_point.exists() and mount_point.is_dir():
+            _logger.debug("squashfs_rom: %s already exists", mount_point)
+
+            # try to remove an empty directory, else, run the directory, ignoring the .squashfs
+            try:
+                mount_point.rmdir()
+            except (FileNotFoundError, OSError):
+                _logger.debug("squashfs_rom: failed to rmdir %s", mount_point)
+                yield mount_point
+                # No cleanup is necessary
+                return
 
     # ok, the base directory doesn't exist, let's create it and mount the squashfs on it
-    mount_point.mkdir()
+    mount_point.mkdir(parents=True)
 
     return_code = subprocess.call(["mount", rom, mount_point])
     if return_code != 0:
@@ -48,28 +85,33 @@ def squashfs_rom(rom: Path, /) -> Iterator[Path]:
             pass
         raise BatoceraException(f"Unable to mount the file {rom}")
 
+    if overlay:
+        for dir in [overlay_delta, overlay_tmp, mount_point_rw]: dir.mkdir(exist_ok=True)
+        return_code = subprocess.call(["mount",
+				       "-t", "overlay",
+				       "overlay",
+				       "-o", f"lowerdir={mount_point},upperdir={overlay_delta},workdir={overlay_tmp}",
+				       mount_point_rw])
+        if return_code != 0:
+            _logger.debug("squashfs_rom: mounting overlay %s failed", mount_point_rw)
+            _squashfs_rom_cleanup(mount_point, overlay)
+            raise BatoceraException(f"Unable to mount the file {rom}")
+
     try:
+        working_mount_point = mount_point_rw if overlay else mount_point
+
         # if the squashfs contains a single file with the same name, take it as the rom file
-        rom_single = mount_point / rom.stem
+        rom_single = working_mount_point / rom.stem
         if len(list(mount_point.iterdir())) == 1 and rom_single.exists():
             _logger.debug("squashfs: single rom %s", rom_single)
             yield rom_single
         else:
             try:
-                rom_linked = (mount_point / ".ROM").resolve(strict=True)
+                rom_linked = (working_mount_point / ".ROM").resolve(strict=True)
             except OSError:
-                yield mount_point
+                yield working_mount_point
             else:
                 _logger.debug("squashfs: linked rom %s", rom_linked)
                 yield rom_linked
     finally:
-        _logger.debug("squashfs_rom: cleaning up %s", mount_point)
-
-        # unmount
-        return_code = subprocess.call(["umount", mount_point])
-        if return_code != 0:
-            _logger.debug("squashfs_rom: unmounting %s failed", mount_point)
-            raise BatoceraException(f"Unable to unmount the file {mount_point}")
-
-        # cleaning the empty directory
-        mount_point.rmdir()
+        _squashfs_rom_cleanup(mount_point, overlay)

package/batocera/emulationstation/batocera-es-system/es_systems.yml

@@ -5651,7 +5651,7 @@ lindbergh:
   manufacturer: Sega
   release: 2005
   hardware: arcade
-  extensions: [game]
+  extensions: [game, squashfs]
   platform: lindbergh, arcade
   emulators:
     lindbergh-loader: