import logging import os import pathlib import shutil import typing import tempfile import fs import fs.base import fs.copy import fs.osfs from . import core class GameVFSHandler(object): vfs = None log = logging.getLogger().getChild("vfs") proto_handlers = {} def _osfs_handle(self, path): if self.osfs_cwd.exists(path): return self.osfs_cwd.opendir(path) elif self.osfs_appdir.exists(path): return self.osfs_appdir.opendir(path) else: raise FileNotFoundError def __init__(self) -> None: self.vfs = fs.open_fs("mem://") self.proto_handlers["osfs"] = self._osfs_handle self.osfs_cwd = fs.osfs.OSFS(os.getcwd()) self.osfs_appdir = fs.osfs.OSFS(core.path_appdir.as_posix()) self.pth_temp = pathlib.Path(tempfile.mkdtemp()) self.osfs_temp = fs.osfs.OSFS(self.pth_temp.as_posix()) def __del__(self): if self.pth_temp and self.pth_temp.exists(): shutil.rmtree(self.pth_temp) def __getattr__(self, name: str) -> typing.Any: try: return getattr(self.vfs, name) except: raise def load_data_source(self, source: pathlib.Path | fs.base.FS | typing.Text, proto="osfs"): print(f"data source: {source} ({proto})") assert self.vfs is not None if isinstance(source, pathlib.Path): assert isinstance(source, pathlib.Path) # for linter self.log.info(f"loading vfs source: {source.as_posix()}") if proto in self.proto_handlers: fs_source = self.proto_handlers[proto](source.as_posix()) fs.copy.copy_fs(fs_source, self.vfs) else: fs_source = fs.open_fs(f"{proto}:/{source.as_posix()}") fs.copy.copy_fs(fs_source, self.vfs) else: if proto in self.proto_handlers: fs_source = self.proto_handlers[proto](source) else: fs_source = fs.open_fs(f"{proto}://{source}") self.log.info(f"loading vfs source: {fs_source} (pyfilesystem2 handler)") fs.copy.copy_fs(fs_source, self.vfs) def copy_out(self, filepath, dest=None): if not dest: self.osfs_temp.makedirs(pathlib.Path(filepath).parent.as_posix(), recreate=True) fs.copy.copy_file(self.vfs, filepath, self.osfs_temp, filepath) return self.pth_temp / filepath else: pth_dest = pathlib.Path(dest) pth_file = pathlib.Path(filepath) osfs_dest = fs.osfs.OSFS(dest) osfs_dest.makedirs(pth_file.parent.as_posix(), recreate=True) fs.copy.copy_file(self.vfs, filepath, dest, filepath) return (pth_dest / filepath).as_posix() vfs = GameVFSHandler()