python/steamutil,steamsync,protontool: Remove space on empty lines

master
Taeyeon Mori 4 years ago
parent a9ec47f54a
commit 368f1a1e24
  1. 24
      bin/protontool
  2. 34
      lib/python/steamsync.py
  3. 80
      lib/python/steamutil.py
  4. 52
      lib/python/vdfparser.py

@ -59,7 +59,7 @@ class ProtonTool:
self.app = apps[0] self.app = apps[0]
print_info("Found game: %s" % self.app.name) print_info("Found game: %s" % self.app.name)
@steamutil.CachedProperty @steamutil.CachedProperty
def user(self) -> steamutil.LoginUser: def user(self) -> steamutil.LoginUser:
if "userid" in self.args: if "userid" in self.args:
@ -83,7 +83,6 @@ class ProtonTool:
print(" ", path) print(" ", path)
parser.exit(0) parser.exit(0)
@classmethod @classmethod
def parse_args(cls, args, prog=None): def parse_args(cls, args, prog=None):
parser = argparse.ArgumentParser(prog=prog) parser = argparse.ArgumentParser(prog=prog)
@ -133,12 +132,12 @@ class ProtonTool:
@steamutil.CachedProperty @steamutil.CachedProperty
def appinfo(self): def appinfo(self):
return self.steam.appinfo return self.steam.appinfo
@property @property
def compat_tools(self) -> dict: def compat_tools(self) -> dict:
self.steam.steamplay_manifest = self.appinfo[891390]["appinfo"] self.steam.steamplay_manifest = self.appinfo[891390]["appinfo"]
return self.steam.compat_tools return self.steam.compat_tools
@property @property
def compat_tool(self) -> dict: def compat_tool(self) -> dict:
return self.app.compat_tool return self.app.compat_tool
@ -147,9 +146,9 @@ class ProtonTool:
def _format_tool_info(cls, steam, tool, toolinfo=None) -> (str, str): def _format_tool_info(cls, steam, tool, toolinfo=None) -> (str, str):
if toolinfo is None: if toolinfo is None:
toolinfo = steam.compat_tools.get(tool["name"], None) toolinfo = steam.compat_tools.get(tool["name"], None)
proton = [] proton = []
if not toolinfo: if not toolinfo:
proton.append("\033[31mNot found:\033[0m") proton.append("\033[31mNot found:\033[0m")
@ -181,7 +180,7 @@ class ProtonTool:
def cmd_info(self): def cmd_info(self):
uc = self.user.get_app_config(self.app) uc = self.user.get_app_config(self.app)
if self.app.installed: if self.app.installed:
installed = "[%s] %s" % (self.app.language, format_size(self.app.declared_install_size)) installed = "[%s] %s" % (self.app.language, format_size(self.app.declared_install_size))
@ -206,7 +205,7 @@ class ProtonTool:
print(" @", proton_path) print(" @", proton_path)
if self.app.is_proton_app: if self.app.is_proton_app:
print("\033[1;35mProtonDrv.\033[0m:", self.app.compat_drive) print("\033[1;35mProtonDrv.\033[0m:", self.app.compat_drive)
def setup_proton_env(self, tool, wine=False): def setup_proton_env(self, tool, wine=False):
info = self.compat_tools[tool["name"]] info = self.compat_tools[tool["name"]]
if not info: if not info:
@ -232,9 +231,9 @@ class ProtonTool:
tool = self.compat_tool tool = self.compat_tool
self.setup_proton_env(tool, wine=True) self.setup_proton_env(tool, wine=True)
return os.execvp(self.args.cmdline[0], self.args.cmdline) return os.execvp(self.args.cmdline[0], self.args.cmdline)
def cmd_run(self): def cmd_run(self):
""" Run a game executable directly, possibly using a compat tool """ """ Run a game executable directly, possibly using a compat tool """
# Obviously has to be installed # Obviously has to be installed
@ -249,7 +248,7 @@ class ProtonTool:
tool = self.compat_tool tool = self.compat_tool
else: else:
tool = None tool = None
# Default to native oslist # Default to native oslist
if sys.platform.startswith("linux"): if sys.platform.startswith("linux"):
target_oslist = "linux" target_oslist = "linux"
@ -304,6 +303,7 @@ class ProtonTool:
else: else:
print_error("No matching launch configuration in appinfo") print_error("No matching launch configuration in appinfo")
return 51 return 51
def cmd_proton(self): def cmd_proton(self):
# proton and wine subcommands: # proton and wine subcommands:
# Run proton respective wine with arguments # Run proton respective wine with arguments
@ -361,7 +361,7 @@ class ProtonTool:
print() print()
self._pretty_dump(info.dict) self._pretty_dump(info.dict)
def _pretty_dump(self, d: dict, level=0): def _pretty_dump(self, d: dict, level=0):
for key, value in d.items(): for key, value in d.items():
print(" " * level, "\033[1m", key, "\033[0m", end="", sep="") print(" " * level, "\033[1m", key, "\033[0m", end="", sep="")

@ -23,12 +23,12 @@ class SyncPath:
Whereby target is the location being synched to and local is Whereby target is the location being synched to and local is
the prefix the data is synched from on the local machine. the prefix the data is synched from on the local machine.
Common has components common to both paths. Common has components common to both paths.
Usually, you'd set the local prefix and then the common part. Usually, you'd set the local prefix and then the common part.
e.g.: op.home.prefix(".my_game") / "Savegames" e.g.: op.home.prefix(".my_game") / "Savegames"
whereby "Savegames" is included in the resulting target whereby "Savegames" is included in the resulting target
path, but ".my_game" is not. path, but ".my_game" is not.
Note that SyncPath should be considered immutable. Relevant Note that SyncPath should be considered immutable. Relevant
methods return a new instance. methods return a new instance.
""" """
@ -52,16 +52,16 @@ class SyncPath:
""" Return a new SyncPath that nas a component added """ """ Return a new SyncPath that nas a component added """
return SyncPath(self.op, self.local, self.common / component) return SyncPath(self.op, self.local, self.common / component)
## Retrieve paths ## Retrieve paths
@property @property
def path(self) -> Path: def path(self) -> Path:
""" Get the local path """ """ Get the local path """
return self.local / self.common return self.local / self.common
def exists(self) -> bool: def exists(self) -> bool:
""" Chech whether local path exists """ """ Chech whether local path exists """
return self.path.exists() return self.path.exists()
@property @property
def target_path(self) -> Path: def target_path(self) -> Path:
""" Get the sync target path """ """ Get the sync target path """
@ -70,7 +70,7 @@ class SyncPath:
## Begin a SyncSet ## Begin a SyncSet
def __enter__(self) -> 'SyncSet': def __enter__(self) -> 'SyncSet':
return SyncSet(self) return SyncSet(self)
def __exit__(self, type, value, traceback): def __exit__(self, type, value, traceback):
# Todo: auto-commit? # Todo: auto-commit?
pass pass
@ -94,12 +94,12 @@ class SyncSet:
self.spath = path self.spath = path
self.local = {} self.local = {}
self.target = {} self.target = {}
@property @property
def path(self) -> Path: def path(self) -> Path:
""" The local path """ """ The local path """
return self.spath.path return self.spath.path
@property @property
def target_path(self) -> Path: def target_path(self) -> Path:
""" The target path """ """ The target path """
@ -123,11 +123,11 @@ class SyncSet:
self.local.update(self._collect_files(self.path, patterns)) self.local.update(self._collect_files(self.path, patterns))
self.target.update(self._collect_files(self.target_path, patterns)) self.target.update(self._collect_files(self.target_path, patterns))
self._inval() self._inval()
def __iadd__(self, pattern: str) -> 'SyncSet': def __iadd__(self, pattern: str) -> 'SyncSet':
self.add(pattern) self.add(pattern)
return self return self
# Calculate changes # Calculate changes
def _inval(self): def _inval(self):
for cache in "files_from_local", "files_from_target", "files_unmodified": for cache in "files_from_local", "files_from_target", "files_unmodified":
@ -143,19 +143,19 @@ class SyncSet:
for f, (_, sst) in src_files.items() for f, (_, sst) in src_files.items()
if f not in dst_files or sst.st_mtime > dst_files[f][1].st_mtime if f not in dst_files or sst.st_mtime > dst_files[f][1].st_mtime
} }
@CachedProperty @CachedProperty
def files_from_local(self) -> Set[Path]: def files_from_local(self) -> Set[Path]:
return self._sync_set(self.local, self.target) return self._sync_set(self.local, self.target)
@CachedProperty @CachedProperty
def files_from_target(self) -> Set[Path]: def files_from_target(self) -> Set[Path]:
return self._sync_set(self.target, self.local) return self._sync_set(self.target, self.local)
@CachedProperty @CachedProperty
def files_unmodified(self) -> Set[Path]: def files_unmodified(self) -> Set[Path]:
return (self.local.keys() | self.target.keys()) - (self.files_from_local | self.files_from_target) return (self.local.keys() | self.target.keys()) - (self.files_from_local | self.files_from_target)
def show_confirm(self, skip=True) -> bool: def show_confirm(self, skip=True) -> bool:
# XXX: move to SyncOp? # XXX: move to SyncOp?
print(" Local is newer: ", ", ".join(map(str, self.files_from_local))) print(" Local is newer: ", ", ".join(map(str, self.files_from_local)))
@ -182,7 +182,7 @@ class SyncSet:
if self.files_from_target: if self.files_from_target:
operations += [(self.target_path / p, self.path / p) for p in self.files_from_target] #pylint:disable=not-an-iterable operations += [(self.target_path / p, self.path / p) for p in self.files_from_target] #pylint:disable=not-an-iterable
return self.op._do_copy(operations) return self.op._do_copy(operations)
@ -342,7 +342,7 @@ class SteamSync:
return SteamSyncOp(self, app) return SteamSyncOp(self, app)
else: else:
return AppNotFound return AppNotFound
def by_name(self, pattern): def by_name(self, pattern):
""" Steam App by Name """ """ Steam App by Name """
pt = re.compile(fnmatch.translate(pattern).rstrip("\\Z"), re.IGNORECASE) pt = re.compile(fnmatch.translate(pattern).rstrip("\\Z"), re.IGNORECASE)
@ -355,7 +355,7 @@ class SteamSync:
if app is None: if app is None:
return AppNotFound return AppNotFound
return SteamSyncOp(self, app) return SteamSyncOp(self, app)
def generic(self, name, *, platform=None): def generic(self, name, *, platform=None):
""" Non-Steam App """ """ Non-Steam App """
if platform is not None and platform not in sys.platform: if platform is not None and platform not in sys.platform:

@ -39,7 +39,7 @@ class DictPathRoProperty:
self.path = path self.path = path
self.default = default self.default = default
self.type = type self.type = type
def __get__(self, obj, cls): def __get__(self, obj, cls):
if obj is None: if obj is None:
return self return self
@ -71,7 +71,7 @@ class DictPathProperty(DictPathRoProperty):
def __set__(self, obj, value): def __set__(self, obj, value):
self._get_create_parent(obj)[self.path[-1]] = value self._get_create_parent(obj)[self.path[-1]] = value
def __delete__(self, obj): def __delete__(self, obj):
del self._get_create_parent(obj)[self.path[-1]] del self._get_create_parent(obj)[self.path[-1]]
@ -88,7 +88,7 @@ class MalformedManifestError(Exception):
class AppInfo: class AppInfo:
steam: 'Steam' steam: 'Steam'
appid: int appid: int
def __init__(self, steam, appid, *, appinfo_data=None): def __init__(self, steam, appid, *, appinfo_data=None):
self.steam = steam self.steam = steam
self.appid = appid self.appid = appid
@ -99,7 +99,7 @@ class AppInfo:
return "<steamutil.AppInfo #%7d '%s' (%s)>" % (self.appid, self.name, self.install_dir) return "<steamutil.AppInfo #%7d '%s' (%s)>" % (self.appid, self.name, self.install_dir)
installed = False installed = False
# AppInfo # AppInfo
@CachedProperty @CachedProperty
def appinfo(self): def appinfo(self):
@ -115,11 +115,11 @@ class AppInfo:
install_dir = DictPathRoProperty("appinfo", ("appinfo", "config", "installdir"), default=None) install_dir = DictPathRoProperty("appinfo", ("appinfo", "config", "installdir"), default=None)
languages = DictPathRoProperty("appinfo", ("appinfo", "common", "supported_languages")) languages = DictPathRoProperty("appinfo", ("appinfo", "common", "supported_languages"))
gameid = DictPathRoProperty("appinfo", ("appinfo", "common", "gameid"), type=int) gameid = DictPathRoProperty("appinfo", ("appinfo", "common", "gameid"), type=int)
# Misc. # Misc.
def get_userdata_path(self, user_id: Union[int, 'LoginUser']) -> Path: def get_userdata_path(self, user_id: Union[int, 'LoginUser']) -> Path:
return self.steam.get_userdata_path(user_id) / str(self.appid) return self.steam.get_userdata_path(user_id) / str(self.appid)
@property @property
def is_native(self): def is_native(self):
return sys.platform in self.oslist return sys.platform in self.oslist
@ -166,11 +166,11 @@ class App(AppInfo):
if "AppState" not in self.manifest: if "AppState" not in self.manifest:
raise MalformedManifestError("App manifest doesn't have AppState key", self.manifest_path) raise MalformedManifestError("App manifest doesn't have AppState key", self.manifest_path)
super().__init__(libfolder.steam, int(self.manifest["AppState"]["appid"])) super().__init__(libfolder.steam, int(self.manifest["AppState"]["appid"]))
installed = True installed = True
def __repr__(self): def __repr__(self):
return "<steamutil.App #%7d '%s' @ \"%s\">" % (self.appid, self.name, self.install_path) return "<steamutil.App #%7d '%s' @ \"%s\">" % (self.appid, self.name, self.install_path)
@ -178,11 +178,11 @@ class App(AppInfo):
name = DictPathRoProperty("manifest", ("AppState", "name")) name = DictPathRoProperty("manifest", ("AppState", "name"))
language = DictPathRoProperty("manifest", ("AppState", "UserConfig", "language"), None) language = DictPathRoProperty("manifest", ("AppState", "UserConfig", "language"), None)
install_dir = DictPathRoProperty("manifest", ("AppState", "installdir")) install_dir = DictPathRoProperty("manifest", ("AppState", "installdir"))
@CachedProperty @CachedProperty
def install_path(self) -> Path: def install_path(self) -> Path:
return self.steamapps_path / "common" / self.install_dir return self.steamapps_path / "common" / self.install_dir
# Workshop # Workshop
# TODO # TODO
@CachedProperty @CachedProperty
@ -195,7 +195,7 @@ class App(AppInfo):
uc = self.manifest["AppState"].get("UserConfig") uc = self.manifest["AppState"].get("UserConfig")
if uc and "platform_override_source" in uc: if uc and "platform_override_source" in uc:
return uc["platform_override_source"] return uc["platform_override_source"]
@property @property
def is_proton_app(self): def is_proton_app(self):
uc = self.manifest["AppState"].get("UserConfig") uc = self.manifest["AppState"].get("UserConfig")
@ -205,14 +205,14 @@ class App(AppInfo):
@CachedProperty @CachedProperty
def compat_path(self) -> Path: def compat_path(self) -> Path:
return self.steamapps_path / "compatdata" / str(self.appid) return self.steamapps_path / "compatdata" / str(self.appid)
@CachedProperty @CachedProperty
def compat_drive(self) -> Path: def compat_drive(self) -> Path:
return self.compat_path / "pfx" / "drive_c" return self.compat_path / "pfx" / "drive_c"
# Install size # Install size
declared_install_size = DictPathRoProperty("manifest", ("AppState", "SizeOnDisk"), 0, type=int) declared_install_size = DictPathRoProperty("manifest", ("AppState", "SizeOnDisk"), 0, type=int)
def compute_install_size(self) -> int: def compute_install_size(self) -> int:
def sum_size(p: Path): def sum_size(p: Path):
acc = 0 acc = 0
@ -232,7 +232,7 @@ class LibraryFolder:
def __init__(self, steam: 'Steam', path: Path): def __init__(self, steam: 'Steam', path: Path):
self.steam = steam self.steam = steam
self.path = path self.path = path
def __repr__(self): def __repr__(self):
return "<steamutil.LibraryFolder @ \"%s\">" % self.path return "<steamutil.LibraryFolder @ \"%s\">" % self.path
@ -254,7 +254,7 @@ class LibraryFolder:
return found[0] return found[0]
# if none exists, return non-existant default name # if none exists, return non-existant default name
return steamapps return steamapps
@property @property
def common_path(self) -> Path: def common_path(self) -> Path:
return self.steamapps_path / "common" return self.steamapps_path / "common"
@ -295,26 +295,26 @@ class UserAppConfig:
def __init__(self, user, appid): def __init__(self, user, appid):
self.user = user self.user = user
self.appid = appid self.appid = appid
def __repr__(self): def __repr__(self):
return "<steamutil.UserAppConfig appid=%d for account %s>" % (self.appid, self.user.account_name) return "<steamutil.UserAppConfig appid=%d for account %s>" % (self.appid, self.user.account_name)
@property @property
def _data(self): def _data(self):
try: try:
return self.user.localconfig["UserLocalConfigStore"]["Software"]["Valve"]["Steam"]["Apps"][str(self.appid)] return self.user.localconfig["UserLocalConfigStore"]["Software"]["Valve"]["Steam"]["Apps"][str(self.appid)]
except KeyError: except KeyError:
return {} # TODO return {} # TODO
@property @property
def last_played(self) -> datetime.datetime: def last_played(self) -> datetime.datetime:
return datetime.datetime.fromtimestamp(int(self._data.get("LastPlayed", "0"))) return datetime.datetime.fromtimestamp(int(self._data.get("LastPlayed", "0")))
@property @property
def playtime(self) -> datetime.time: def playtime(self) -> datetime.time:
t = int(self._data.get("Playtime", "0")) t = int(self._data.get("Playtime", "0"))
return datetime.time(t // 60, t % 60) return datetime.time(t // 60, t % 60)
@property @property
def playtime_two_weeks(self) -> datetime.time: def playtime_two_weeks(self) -> datetime.time:
t = int(self._data.get("Playtime2wks", "0")) t = int(self._data.get("Playtime2wks", "0"))
@ -332,31 +332,31 @@ class LoginUser:
self.steam = steam self.steam = steam
self.id = id self.id = id
self.info = info self.info = info
def __repr__(self): def __repr__(self):
return "<steamutil.LoginUser %d %s '%s'>" % (self.id , self.account_name, self.username) return "<steamutil.LoginUser %d %s '%s'>" % (self.id , self.account_name, self.username)
@property @property
def account_id(self): def account_id(self):
""" 32-bit account ID """ """ 32-bit account ID """
return self.id & 0xffffffff return self.id & 0xffffffff
account_name = DictPathRoProperty("info", ("AccountName",)) account_name = DictPathRoProperty("info", ("AccountName",))
username = DictPathRoProperty("info", ("PersonaName",)) username = DictPathRoProperty("info", ("PersonaName",))
@CachedProperty @CachedProperty
def userdata_path(self) -> Path: def userdata_path(self) -> Path:
return self.steam.get_userdata_path(self) return self.steam.get_userdata_path(self)
@property @property
def localconfig_vdf(self) -> Path: def localconfig_vdf(self) -> Path:
return self.userdata_path / "config" / "localconfig.vdf" return self.userdata_path / "config" / "localconfig.vdf"
@CachedProperty @CachedProperty
def localconfig(self) -> DeepDict: def localconfig(self) -> DeepDict:
with open(self.localconfig_vdf, encoding="utf-8") as f: with open(self.localconfig_vdf, encoding="utf-8") as f:
return _vdf.parse(f) return _vdf.parse(f)
# Game config # Game config
def get_app_config(self, app: Union[int, App]) -> Optional[UserAppConfig]: def get_app_config(self, app: Union[int, App]) -> Optional[UserAppConfig]:
if isinstance(app, App): if isinstance(app, App):
@ -371,7 +371,7 @@ class Steam:
self.root = install_path if install_path is not None else self.find_install_path() self.root = install_path if install_path is not None else self.find_install_path()
if self.root is None: if self.root is None:
raise Exception("Could not find Steam") raise Exception("Could not find Steam")
def __repr__(self): def __repr__(self):
return "<steamutil.Steam @ \"%s\">" % self.root return "<steamutil.Steam @ \"%s\">" % self.root
@ -416,15 +416,15 @@ class Steam:
def libraryfolders_vdf(self) -> Path: def libraryfolders_vdf(self) -> Path:
""" The libraryfolders.vdf file listing all configured library locations """ """ The libraryfolders.vdf file listing all configured library locations """
return self.root / "steamapps" / "libraryfolders.vdf" return self.root / "steamapps" / "libraryfolders.vdf"
@property @property
def config_vdf(self) -> Path: def config_vdf(self) -> Path:
return self.root / "config" / "config.vdf" return self.root / "config" / "config.vdf"
@property @property
def loginusers_vdf(self) -> Path: def loginusers_vdf(self) -> Path:
return self.root / "config" / "loginusers.vdf" return self.root / "config" / "loginusers.vdf"
# Users # Users
@CachedProperty @CachedProperty
def most_recent_user(self) -> Optional[LoginUser]: def most_recent_user(self) -> Optional[LoginUser]:
@ -439,18 +439,18 @@ class Steam:
except KeyError: except KeyError:
pass pass
return None return None
def get_userdata_path(self, user_id: Union[int, LoginUser]) -> Path: def get_userdata_path(self, user_id: Union[int, LoginUser]) -> Path:
if isinstance(user_id, LoginUser): if isinstance(user_id, LoginUser):
user_id = user_id.account_id user_id = user_id.account_id
return self.root / "userdata" / str(user_id) return self.root / "userdata" / str(user_id)
# Config # Config
@CachedProperty @CachedProperty
def config(self) -> DeepDict: def config(self) -> DeepDict:
with open(self.config_vdf, encoding="utf-8") as f: with open(self.config_vdf, encoding="utf-8") as f:
return _vdf.parse(f) return _vdf.parse(f)
config_install_store = DictPathProperty("config", ("InstallConfigStore",)) config_install_store = DictPathProperty("config", ("InstallConfigStore",))
config_software_steam = DictPathProperty("config", ("InstallConfigStore", "Software", "Valve", "Steam")) config_software_steam = DictPathProperty("config", ("InstallConfigStore", "Software", "Valve", "Steam"))
compat_tool_mapping = DictPathProperty("config_software_steam", ("CompatToolMapping",)) compat_tool_mapping = DictPathProperty("config_software_steam", ("CompatToolMapping",))
@ -459,11 +459,11 @@ class Steam:
@CachedProperty @CachedProperty
def appinfo_vdf(self): def appinfo_vdf(self):
return self.root / "appcache" / "appinfo.vdf" return self.root / "appcache" / "appinfo.vdf"
@property @property
def appinfo(self) -> AppInfoFile: def appinfo(self) -> AppInfoFile:
return AppInfoFile.open(self.appinfo_vdf) return AppInfoFile.open(self.appinfo_vdf)
@CachedProperty @CachedProperty
def steamplay_manifest(self) -> DeepDict: def steamplay_manifest(self) -> DeepDict:
with self.appinfo as info: with self.appinfo as info:
@ -503,16 +503,16 @@ class Steam:
def library_folder_paths(self) -> List[Path]: def library_folder_paths(self) -> List[Path]:
with open(self.libraryfolders_vdf, encoding="utf-8") as f: with open(self.libraryfolders_vdf, encoding="utf-8") as f:
return [Path(v) for k,v in _vdf.parse(f)["LibraryFolders"].items() if k.isdigit()] return [Path(v) for k,v in _vdf.parse(f)["LibraryFolders"].items() if k.isdigit()]
@CachedProperty @CachedProperty
def library_folders(self) -> List[LibraryFolder]: def library_folders(self) -> List[LibraryFolder]:
return [LibraryFolder(self, self.root)] + [LibraryFolder(self, p) for p in self.library_folder_paths] #pylint:disable=not-an-iterable return [LibraryFolder(self, self.root)] + [LibraryFolder(self, p) for p in self.library_folder_paths] #pylint:disable=not-an-iterable
@property @property
def apps(self) -> Iterable[App]: def apps(self) -> Iterable[App]:
for lf in self.library_folders: #pylint:disable=not-an-iterable for lf in self.library_folders: #pylint:disable=not-an-iterable
yield from lf.apps yield from lf.apps
def get_app(self, id: int, installed=True) -> Optional[App]: def get_app(self, id: int, installed=True) -> Optional[App]:
for lf in self.library_folders: #pylint:disable=not-an-iterable for lf in self.library_folders: #pylint:disable=not-an-iterable
app = lf.get_app(id) app = lf.get_app(id)

@ -47,7 +47,7 @@ class VdfParser:
whitespace_chars = " \t\n" whitespace_chars = " \t\n"
comment_char = "/" comment_char = "/"
newline_char = "\n" newline_char = "\n"
def __init__(self, *, encoding=False, factory=dict, strict=True): def __init__(self, *, encoding=False, factory=dict, strict=True):
""" """
@brief Construct a VdfParser instance @brief Construct a VdfParser instance
@ -89,7 +89,7 @@ class VdfParser:
while True: while True:
c = fd.read(1) c = fd.read(1)
if not c: if not c:
finish() finish()
if len(tokens) / 2 != len(tokens) // 2: if len(tokens) / 2 != len(tokens) // 2:
@ -97,7 +97,7 @@ class VdfParser:
elif self.strict and (escape or quoted or inner): elif self.strict and (escape or quoted or inner):
raise ValueError("Unexpected EOF: EOF encountered while not processing outermost mapping") raise ValueError("Unexpected EOF: EOF encountered while not processing outermost mapping")
return self._make_map(tokens) return self._make_map(tokens)
if escape: if escape:
current.append(c) current.append(c)
escape = False escape = False
@ -110,11 +110,11 @@ class VdfParser:
finish(override=True) finish(override=True)
else: else:
current.append(c) current.append(c)
elif comment: elif comment:
if c == self.newline_char: if c == self.newline_char:
comment = False comment = False
else: else:
if c == self.escape_char: if c == self.escape_char:
escape = True escape = True
@ -154,11 +154,11 @@ class VdfParser:
return self.parse(io.BytesIO(content)) return self.parse(io.BytesIO(content))
else: else:
return self.parse(io.StringIO(content)) return self.parse(io.StringIO(content))
def _make_literal(self, lit): def _make_literal(self, lit):
# TODO # TODO
return "\"%s\"" % (str(lit).replace("\\", "\\\\").replace("\"", "\\\"")) return "\"%s\"" % (str(lit).replace("\\", "\\\\").replace("\"", "\\\""))
def _write_map(self, fd, dictionary, indent): def _write_map(self, fd, dictionary, indent):
if indent is None: if indent is None:
def write(str=None, i=False, d=False, nl=False): def write(str=None, i=False, d=False, nl=False):
@ -166,7 +166,7 @@ class VdfParser:
fd.write(str) fd.write(str)
if d: if d:
fd.write(" ") fd.write(" ")
else: else:
def write(str=None, i=False, d=False, nl=False): def write(str=None, i=False, d=False, nl=False):
if not str and nl: if not str and nl:
@ -180,7 +180,7 @@ class VdfParser:
fd.write("\n") fd.write("\n")
elif d: elif d:
fd.write("\t\t") fd.write("\t\t")
for k, v in dictionary.items(): for k, v in dictionary.items():
if isinstance(v, dict): if isinstance(v, dict):
write(self._make_literal(k), i=1, d=1, nl=1) write(self._make_literal(k), i=1, d=1, nl=1)
@ -191,7 +191,7 @@ class VdfParser:
write(self._make_literal(k), i=1, d=1) write(self._make_literal(k), i=1, d=1)
write(self._make_literal(v)) write(self._make_literal(v))
write(d=1, nl=1) write(d=1, nl=1)
def write(self, fd, dictionary: DeepDict, *, pretty=True): def write(self, fd, dictionary: DeepDict, *, pretty=True):
""" """
Write a dictionary to a file in VDF format Write a dictionary to a file in VDF format
@ -223,7 +223,7 @@ class BinaryVdfParser:
def __init__(self, factory=dict): def __init__(self, factory=dict):
self.factory = factory self.factory = factory
@staticmethod @staticmethod
def _read_until(fd: io.BufferedIOBase, delim: bytes) -> bytes: def _read_until(fd: io.BufferedIOBase, delim: bytes) -> bytes:
pieces = [] pieces = []
@ -238,7 +238,7 @@ class BinaryVdfParser:
end = buf.find(delim, 0, read) end = buf.find(delim, 0, read)
pieces.append(bytes(buf[:read if end < 0 else end])) pieces.append(bytes(buf[:read if end < 0 else end]))
fd.seek(end - read + len(delim), io.SEEK_CUR) fd.seek(end - read + len(delim), io.SEEK_CUR)
return b"".join(pieces) return b"".join(pieces)
@ -249,10 +249,10 @@ class BinaryVdfParser:
def _read_cstring(self, fd: io.BufferedIOBase) -> str: def _read_cstring(self, fd: io.BufferedIOBase) -> str:
return self._read_until(fd, b'\0').decode("utf-8", "replace") return self._read_until(fd, b'\0').decode("utf-8", "replace")
def _read_wstring(self, fd: io.BufferedIOBase) -> str: def _read_wstring(self, fd: io.BufferedIOBase) -> str:
return self._read_until(fd, b'\0\0').decode("utf-16") return self._read_until(fd, b'\0\0').decode("utf-16")
def _read_map(self, fd: io.BufferedIOBase) -> DeepDict: def _read_map(self, fd: io.BufferedIOBase) -> DeepDict:
map = self.factory() map = self.factory()
@ -264,10 +264,10 @@ class BinaryVdfParser:
if t in (self.T_END, self.T_END2): if t in (self.T_END, self.T_END2):
return map return map
key, value = self._read_item(fd, t) key, value = self._read_item(fd, t)
map[key] = value map[key] = value
def _read_item(self, fd: io.BufferedIOBase, t: int) -> (str, DeepDict): def _read_item(self, fd: io.BufferedIOBase, t: int) -> (str, DeepDict):
key = self._read_cstring(fd) key = self._read_cstring(fd)
@ -287,7 +287,7 @@ class BinaryVdfParser:
return key, self._read_struct(fd, self.S_FLT4)[0] return key, self._read_struct(fd, self.S_FLT4)[0]
else: else:
raise ValueError("Unknown data type", fd.tell(), t) raise ValueError("Unknown data type", fd.tell(), t)
def parse(self, fd: io.BufferedIOBase) -> DeepDict: def parse(self, fd: io.BufferedIOBase) -> DeepDict:
return self._read_map(fd) return self._read_map(fd)
@ -310,7 +310,7 @@ class AppInfoFile:
self._close_file = close self._close_file = close
self._universe = None self._universe = None
self._apps = None self._apps = None
def _load_map(self, offset: int) -> DeepDict: def _load_map(self, offset: int) -> DeepDict:
self.file.seek(offset, io.SEEK_SET) self.file.seek(offset, io.SEEK_SET)
return self.parser.parse(self.file) return self.parser.parse(self.file)
@ -329,19 +329,19 @@ class AppInfoFile:
self.appinfo = appinfo self.appinfo = appinfo
self.offset = offset self.offset = offset
self._data = None self._data = None
def __getitem__(self, key): def __getitem__(self, key):
if self._data is None: if self._data is None:
self._data = self.appinfo._load_map(self.offset) self._data = self.appinfo._load_map(self.offset)
return self._data[key] return self._data[key]
def __getattr__(self, attr): def __getattr__(self, attr):
if attr in dir(dict): if attr in dir(dict):
if self._data is None: if self._data is None:
self._data = self.appinfo._load_map(self.offset) self._data = self.appinfo._load_map(self.offset)
return getattr(self._data, attr) return getattr(self._data, attr)
raise AttributeError(attr) raise AttributeError(attr)
@property @property
def dict(self): def dict(self):
if self._data is None: if self._data is None:
@ -353,7 +353,7 @@ class AppInfoFile:
if len(cs) < s: if len(cs) < s:
raise EOFError() raise EOFError()
return cs return cs
def _read_int(self) -> int: def _read_int(self) -> int:
return self.S_INT4.unpack(self._read_exactly(self.S_INT4.size))[0] return self.S_INT4.unpack(self._read_exactly(self.S_INT4.size))[0]
@ -361,7 +361,7 @@ class AppInfoFile:
magic = self._read_exactly(4) magic = self._read_exactly(4)
if magic != b"\x27\x44\x56\x07": if magic != b"\x27\x44\x56\x07":
raise ValueError("Wrong appinfo.vdf magic") raise ValueError("Wrong appinfo.vdf magic")
self._universe = self._read_int() self._universe = self._read_int()
self._apps = {} self._apps = {}
@ -372,7 +372,7 @@ class AppInfoFile:
if read < 4: if read < 4:
raise EOFError() raise EOFError()
struct = self.S_APP_HEADER.unpack(buffer) struct = self.S_APP_HEADER.unpack(buffer)
appid, size, *_ = struct appid, size, *_ = struct
@ -402,7 +402,7 @@ class AppInfoFile:
if self._apps is None: if self._apps is None:
self._load() self._load()
return self._apps[key] return self._apps[key]
def __iter__(self): def __iter__(self):
if self._apps is None: if self._apps is None:
self._load() self._load()
@ -417,7 +417,7 @@ class AppInfoFile:
# Cleanup # Cleanup
def __enter__(self): def __enter__(self):
return self return self
def __exit__(self, exc, tp, tb): def __exit__(self, exc, tp, tb):
self.close() self.close()

Loading…
Cancel
Save