91 lines
3.5 KiB
Python
91 lines
3.5 KiB
Python
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
|
|
|
|
## \brief Handler for virtual filesystem.
|
|
# \internal
|
|
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
|
|
|
|
## \brief Load files from a given data source.
|
|
# \param source The data source. This can be a pyfilesystem2 object, or any string or path-like
|
|
# object that can reasonably be interpreted as a directory or zip file.
|
|
# \param proto The <a href="https://pyfilesystem2.readthedocs.io/en/latest/builtin.html">PyFilesystem2 filesystem protocol</a> to use. Defaults to `"osfs"`, which loads directories.
|
|
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)
|
|
|
|
## \brief Copies a file out of VFS into the real filesystem.
|
|
# \param filepath The source file path to copy out.
|
|
# \param dest The path to write the file to. Defaults to temporary directory.
|
|
# \return The full destination file path.
|
|
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()
|
|
|
|
## \brief Primary VFS handler.
|
|
# \internal
|
|
vfs = GameVFSHandler() |