[517] Hand Over Config to New Format

This commit is contained in:
David Sangrey committed 2025-11-16 19:08:19 -05:00
1 parent 2d3373843e
commit 567d42e4b3
6 files changed
+358 -598

No files matched your search

+3 -1
View File
@@ -51,7 +51,7 @@ from threading import get_native_id as thread_native_id
from time import gmtime
from traceback import print_exc
from typing import TYPE_CHECKING, cast
from config import appcmdname, appname, config, trace_on
from config import appcmdname, appname, config, trace_on, config_logger
# TODO: Tests:
#
@@ -535,4 +535,6 @@ if not loglevel:
base_logger_name = appcmdname if os.getenv('EDMC_NO_UI') else appname
edmclogger = Logger(base_logger_name, loglevel=loglevel)
for h in config_logger.handlers[:]:
config_logger.removeHandler(h)
logger: 'LoggerMixin' = edmclogger.get_logger()
+2 -2
View File
@@ -38,7 +38,7 @@ from monitor import monitor
from EDMCLogging import get_main_logger
def get_sys_report(active_config: config.AbstractConfig) -> str:
def get_sys_report(active_config: config.Config) -> str:
"""Gather system information about Elite, the Host Computer, and EDMC."""
# Calculate Requested Information
plt = platform.uname()
@@ -108,7 +108,7 @@ def copy_sys_report(root: tk.Tk, report: str) -> None:
messagebox.showinfo("System Profiler", "System Report copied to Clipboard", parent=root)
def main(active_config: config.AbstractConfig) -> None:
def main(active_config: config.Config) -> None:
"""Entry Point for the System Profiler."""
# Now Let's Begin
root: tk.Tk = tk.Tk()
+224 -134
View File
@@ -12,27 +12,27 @@ from __future__ import annotations
__all__ = [
# defined in the order they appear in the file
'GITVERSION_FILE',
'appname',
'applongname',
'appcmdname',
'copyright',
'update_interval',
'debug_senders',
'trace_on',
'capi_pretend_down',
'capi_debug_access_token',
'logger',
'git_shorthash_from_head',
'appversion',
'user_agent',
'appversion_nobuild',
'AbstractConfig',
'config',
'get_update_feed',
"GITVERSION_FILE",
"appname",
"applongname",
"appcmdname",
"copyright",
"update_interval",
"debug_senders",
"trace_on",
"capi_pretend_down",
"capi_debug_access_token",
"logger",
"git_shorthash_from_head",
"appversion",
"user_agent",
"appversion_nobuild",
"Config",
"config",
"get_update_feed",
"config_logger",
]
import abc
import contextlib
import logging
import os
@@ -40,23 +40,25 @@ import pathlib
import re
import subprocess
import sys
from abc import abstractmethod
import tomllib
import tomli_w
from time import gmtime
from typing import Any, Callable, Type, TypeVar
import semantic_version
from constants import GITVERSION_FILE, applongname, appname
# Any of these may be imported by plugins
appcmdname = 'EDMC'
appcmdname = "EDMC"
# appversion **MUST** follow Semantic Versioning rules:
# <https://semver.org/#semantic-versioning-specification-semver>
# Major.Minor.Patch(-prerelease)(+buildmetadata)
# NB: Do *not* import this, use the functions appversion() and appversion_nobuild()
_static_appversion = '6.0.0-alpha0'
_static_appversion = "6.0.0-alpha0"
_cached_version: semantic_version.Version | None = None
copyright = '© 2015-2019 Jonathan Harris, 2020-2025 EDCD'
copyright = "© 2015-2019 Jonathan Harris, 2020-2025 EDCD"
update_interval = 8*60*60 # 8 Hours
update_interval = 8 * 60 * 60 # 8 Hours
# Providers marked to be in debug mode. Generally this is expected to switch to sending data to a log file
debug_senders: list[str] = []
# TRACE logging code that should actually be used. Means not spamming it
@@ -67,10 +69,14 @@ capi_pretend_down: bool = False
capi_debug_access_token: str | None = None
# This must be done here in order to avoid an import cycle with EDMCLogging.
# Other code should use EDMCLogging.get_main_logger
logger = logging.getLogger(appcmdname) if os.getenv("EDMC_NO_UI") else logging.getLogger(appname)
logger = (
logging.getLogger(appcmdname)
if os.getenv("EDMC_NO_UI")
else logging.getLogger(appname)
)
_T = TypeVar('_T')
_T = TypeVar("_T")
def git_shorthash_from_head() -> str | None:
@@ -94,14 +100,14 @@ def git_shorthash_from_head() -> str | None:
return None
if not re.fullmatch(r"[0-9a-f]{7,}", shorthash):
logger.error(f"'{shorthash}' doesn't look like a valid git short hash, forcing to None")
logger.error(
f"'{shorthash}' doesn't look like a valid git short hash, forcing to None"
)
return None
with contextlib.suppress(Exception):
diff_result = subprocess.run(
["git", "diff", "--stat", "HEAD"],
capture_output=True,
check=True
["git", "diff", "--stat", "HEAD"], capture_output=True, check=True
)
if diff_result.stdout:
shorthash += ".DIRTY"
@@ -121,10 +127,12 @@ def appversion() -> semantic_version.Version:
if _cached_version is not None:
return _cached_version
if getattr(sys, 'frozen', False):
if getattr(sys, "frozen", False):
# Running frozen, so we should have a .gitversion file
# Yes, .parent because if frozen we're inside library.zip
with open(pathlib.Path(sys.path[0]).parent / GITVERSION_FILE, encoding='utf-8') as gitv:
with open(
pathlib.Path(sys.path[0]).parent / GITVERSION_FILE, encoding="utf-8"
) as gitv:
shorthash: str | None = gitv.read()
else:
@@ -134,16 +142,18 @@ def appversion() -> semantic_version.Version:
shorthash = git_shorthash_from_head()
if shorthash is None:
if pathlib.Path(sys.path[0] + "/" + GITVERSION_FILE).exists():
with open(pathlib.Path(sys.path[0] + "/" + GITVERSION_FILE), encoding='utf-8') as gitv:
with open(
pathlib.Path(sys.path[0] + "/" + GITVERSION_FILE), encoding="utf-8"
) as gitv:
shorthash = gitv.read()
else:
shorthash = 'UNKNOWN'
shorthash = "UNKNOWN"
_cached_version = semantic_version.Version(f'{_static_appversion}+{shorthash}')
_cached_version = semantic_version.Version(f"{_static_appversion}+{shorthash}")
return _cached_version
user_agent = f'EDCD-{appname}-{appversion()}'
user_agent = f"EDCD-{appname}-{appversion()}"
def appversion_nobuild() -> semantic_version.Version:
@@ -155,12 +165,12 @@ def appversion_nobuild() -> semantic_version.Version:
:return: App version without any build meta data.
"""
return appversion().truncate('prerelease')
return appversion().truncate("prerelease")
class AbstractConfig(abc.ABC):
class Config:
"""
Abstract root class of all platform specific Config implementations.
Platform-unified Config class for 6.0+.
Commented lines are no longer supported or replaced.
"""
@@ -197,13 +207,113 @@ class AbstractConfig(abc.ABC):
__eddn_tracking_ui = False # Show EDDN tracking UI ?
__skip_timecheck = False # Skip checking event timestamps?
def __init__(self) -> None:
def __init__(self, app_path) -> None:
self.home_path = pathlib.Path.home()
# Set Needed Platform Var for app_dir_path
self.app_dir_path = app_path
self.toml_path: pathlib.Path = self.app_dir_path / "config.toml"
self.generated: str | None = None
self.source: str | None = None
self.settings: dict[str, Any] = {}
self._load()
self.default_plugin_dir_path = self.app_dir_path / "plugins"
if (plugdir_str := self.get_str("plugin_dir")) is None or not pathlib.Path(
plugdir_str
).is_dir():
self.set("plugin_dir", str(self.default_plugin_dir_path))
plugdir_str = self.default_plugin_dir
self.plugin_dir_path = pathlib.Path(plugdir_str)
self.plugin_dir_path.mkdir(exist_ok=True)
# Call the rest of the platform var helpers
self._init_platform()
def _init_platform(self):
if sys.platform == "win32":
from .windows import win_helper
win_helper(self)
elif sys.platform == "linux":
from .linux import linux_helper
linux_helper(self)
else:
raise RuntimeError(f"Unsupported platform: {sys.platform}")
def set_shutdown(self):
"""Set flag denoting we're in the shutdown sequence."""
self.__in_shutdown = True
def _load(self):
"""Load TOML from disk and store fields."""
if not self.toml_path.exists():
raise FileNotFoundError(f"Config file not found: {self.toml_path}")
with self.toml_path.open("rb") as f:
data = tomllib.load(f)
# Capture metadata
self.generated = data.get("generated")
self.source = data.get("source")
# Settings dict created by write_registry_to_toml()
raw_settings = data.get("settings", {})
self.settings = {}
self.settings = dict(raw_settings)
def get(self, key: str, default=None):
"""Return raw stored value."""
return self.settings.get(key, default)
def get_str(self, key: str, default="") -> str:
"""Return string value."""
val = self.get(key, default)
return str(val) if val is not None else default
def get_int(self, key: str, default=0) -> int:
"""Adaptive int (handles booleans stored as ints)."""
val = self.get(key)
if isinstance(val, int):
return val
try:
return int(val)
except (ValueError, TypeError):
return default
def get_bool(self, key: str, default=False) -> bool:
"""
Adaptive boolean reader.
- Accepts ints 0/1
- Accepts strings "true"/"false"/"1"/"0"
- Accepts real booleans
"""
val = self.get(key)
if isinstance(val, bool):
return val
if isinstance(val, int):
return val != 0
if isinstance(val, str):
v = val.strip().lower()
if v in ("1", "true", "yes", "on"):
return True
if v in ("0", "false", "no", "off"):
return False
return default
def get_list(self, key: str, default=None):
"""Return the list referred to by the given key if it exists, or the default."""
val = self.get(key)
return (
val if isinstance(val, list) else (default if default is not None else [])
)
@property
def shutting_down(self) -> bool:
"""
@@ -315,8 +425,10 @@ class AbstractConfig(abc.ABC):
@staticmethod
def _suppress_call(
func: Callable[..., _T], exceptions: Type[BaseException] | list[Type[BaseException]] = Exception,
*args: Any, **kwargs: Any
func: Callable[..., _T],
exceptions: Type[BaseException] | list[Type[BaseException]] = Exception,
*args: Any,
**kwargs: Any,
) -> _T | None:
if exceptions is None:
exceptions = [Exception]
@@ -329,96 +441,54 @@ class AbstractConfig(abc.ABC):
return None
@abstractmethod
def get_list(self, key: str, *, default: list | None = None) -> list:
"""
Return the list referred to by the given key if it exists, or the default.
Implements :meth:`AbstractConfig.get_list`.
"""
raise NotImplementedError
@abstractmethod
def get_str(self, key: str, *, default: str | None = None) -> str:
"""
Return the string referred to by the given key if it exists, or the default.
:param key: The key data is being requested for.
:param default: Default to return if the key does not exist, defaults to None.
:raises ValueError: If an internal error occurs getting or converting a value.
:raises OSError: On Windows, if a Registry error occurs.
:return: The requested data or the default.
"""
raise NotImplementedError
@abstractmethod
def get_bool(self, key: str, *, default: bool | None = None) -> bool:
"""
Return the bool referred to by the given key if it exists, or the default.
:param key: The key data is being requested for.
:param default: Default to return if the key does not exist, defaults to None
:raises ValueError: If an internal error occurs getting or converting a value
:raises OSError: On Windows, if a Registry error occurs.
:return: The requested data or the default
"""
raise NotImplementedError
@abstractmethod
def get_int(self, key: str, *, default: int = 0) -> int:
"""
Return the int referred to by key if it exists in the config.
For legacy reasons, the default is 0 and not None.
:param key: The key data is being requested for.
:param default: Default to return if the key does not exist, defaults to 0.
:raises ValueError: If the internal representation of this key cannot be converted to an int.
:raises OSError: On Windows, if a Registry error occurs.
:return: The requested data or the default.
"""
raise NotImplementedError
@abstractmethod
def set(self, key: str, val: int | str | list[str] | bool) -> None:
"""
Set the given key's data to the given value.
:param key: The key to set the value on.
:param val: The value to set the key's data to.
:raises ValueError: On an invalid type.
:raises OSError: On Windows, if a Registry error occurs.
"""
raise NotImplementedError
@abstractmethod
def delete(self, key: str, *, suppress=False) -> None:
"""
Delete the given key from the config.
"""Delete the given key from the config."""
try:
self.settings.pop(key, None)
except (KeyError, IndexError):
if suppress:
return
raise
self.save()
:param key: The key to delete.
:param suppress: bool - Whether to suppress any errors. Useful in case
code to migrate settings is blindly removing an old key.
:raises OSError: On Windows, if a registry error occurs.
"""
raise NotImplementedError
def set(self, key: str, value: Any):
"""Modify a setting and save to disk."""
self.settings[key] = value
self.save()
@abstractmethod
def save(self) -> None:
"""
Save the current configuration.
def save(self):
"""Write updated config back to TOML."""
data = {
"generated": self.generated,
"source": self.source,
"settings": self.settings,
}
:raises OSError: On Windows, if a Registry error occurs.
"""
raise NotImplementedError
with self.toml_path.open("wb") as f:
tomli_w.dump(data, f)
@abstractmethod
def close(self) -> None:
"""Close this config and release any associated resources."""
raise NotImplementedError
"""Save config changes before closing."""
self.save()
def get_config(*args, **kwargs) -> AbstractConfig:
def get_appdirpath():
"""Grab the Application Directory early."""
app_dir_path = None
if sys.platform == "win32":
base = pathlib.Path(os.getenv("LOCALAPPDATA")) # type: ignore
app_dir_path = base / appname
if sys.platform == "linux":
xdg_data_home = pathlib.Path(
os.getenv("XDG_DATA_HOME", default="~/.local/share")
).expanduser()
app_dir_path = xdg_data_home / appname
if app_dir_path is None:
raise ValueError
return app_dir_path
def get_config(*args, **kwargs) -> Config:
"""
Get the appropriate config class for the current platform.
@@ -426,25 +496,45 @@ def get_config(*args, **kwargs) -> AbstractConfig:
:param kwargs: Args to be passed through to implementation.
:return: Instance of the implementation.
"""
app_dir_path = get_appdirpath()
config = None
if pathlib.Path.exists(app_dir_path / "config.toml"):
return Config(app_path=app_dir_path)
if sys.platform == "win32": # pragma: sys-platform-win32
from .windows import WinConfig
return WinConfig(*args, **kwargs)
from .windows import WinConfigMinimal
config = WinConfigMinimal()
if sys.platform == "linux": # pragma: sys-platform-linux
from .linux import LinuxConfig
return LinuxConfig(*args, **kwargs)
from .linux import LinuxConfigMinimal
config = LinuxConfigMinimal()
raise ValueError(f'Unknown platform: {sys.platform=}')
if config:
config.write_to_toml(f"{app_dir_path}/config.toml") # type: ignore
return Config(app_path=app_dir_path)
raise ValueError(f"Unknown platform: {sys.platform=}")
# Set internal Config logger, because config is set up before main logger.
config_logger = logging.getLogger("pre_config")
config_logger.setLevel(logging.DEBUG) # Or INFO
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
formatter = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")
formatter.converter = gmtime # Optional: match your main logger's UTC timestamps
ch.setFormatter(formatter)
config_logger.addHandler(ch)
config = get_config()
if sys.platform == "win32":
config.write_registry_to_toml(f"{config.app_dir_path}/config.toml") # type: ignore
# Wiki: https://github.com/EDCD/EDMarketConnector/wiki/Participating-in-Open-Betas-of-EDMC
def get_update_feed() -> str:
"""Select the proper update feed for the current update track."""
if config.get_bool('beta_optin'):
return 'https://raw.githubusercontent.com/EDCD/EDMarketConnector/releases/edmarketconnector-beta.xml'
return 'https://raw.githubusercontent.com/EDCD/EDMarketConnector/releases/edmarketconnector.xml'
if config.get_bool("beta_optin"):
return "https://raw.githubusercontent.com/EDCD/EDMarketConnector/releases/edmarketconnector-beta.xml"
return "https://raw.githubusercontent.com/EDCD/EDMarketConnector/releases/edmarketconnector.xml"
+63 -228
View File
@@ -5,252 +5,87 @@ Copyright (c) EDCD, All Rights Reserved
Licensed under the GNU General Public License.
See LICENSE file.
"""
from __future__ import annotations
import os
import pathlib
import sys
from configparser import ConfigParser
from config import AbstractConfig, appname, logger
from config import config_logger
if sys.platform != 'linux':
if sys.platform != "linux":
raise EnvironmentError("This file is for Linux only.")
class LinuxConfig(AbstractConfig):
"""Linux implementation of AbstractConfig."""
import pathlib
from configparser import ConfigParser
import datetime
import tomli_w
from typing import Any, TYPE_CHECKING
SECTION = 'config'
# TODO: I dislike this, would rather use a sane config file format. But here we are.
__unescape_lut = {'\\': '\\', 'n': '\n', ';': ';', 'r': '\r', '#': '#'}
__escape_lut = {'\\': '\\', '\n': 'n', ';': ';', '\r': 'r'}
if TYPE_CHECKING:
from . import Config
class LinuxConfigMinimal:
"""Minimal Linux config loader that dumps existing config to TOML."""
SECTION = "config"
def __init__(self, filename: str | None = None) -> None:
super().__init__()
# http://standards.freedesktop.org/basedir-spec/latest/ar01s03.html
xdg_data_home = pathlib.Path(os.getenv('XDG_DATA_HOME', default='~/.local/share')).expanduser()
self.app_dir_path = xdg_data_home / appname
self.app_dir_path.mkdir(exist_ok=True, parents=True)
self.default_plugin_dir_path = self.app_dir_path / 'plugins'
self.respath_path = pathlib.Path(__file__).parent.parent
self.internal_plugin_dir_path = self.respath_path / 'plugins'
self.default_journal_dir_path = None # type: ignore
self.identifier = f'uk.org.marginal.{appname.lower()}' # TODO: Unused?
config_home = pathlib.Path(os.getenv('XDG_CONFIG_HOME', default='~/.config')).expanduser()
self.filename = config_home / appname / f'{appname}.ini'
if filename is not None:
self.filename = pathlib.Path(filename)
# Determine config path
xdg_config_home = pathlib.Path(
os.getenv("XDG_CONFIG_HOME", "~/.config")
).expanduser()
self.filename = (
pathlib.Path(filename)
if filename
else xdg_config_home / "EDMarketConnector" / "EDMarketConnector.ini"
)
self.filename.parent.mkdir(exist_ok=True, parents=True)
self.config: ConfigParser | None = ConfigParser(comment_prefixes=('#',), interpolation=None)
self.config.read(self.filename) # read() ignores files that dont exist
# Ensure that our section exists. This is here because configparser will happily create files for us, but it
# does not magically create sections
try:
self.config[self.SECTION].get("this_does_not_exist", fallback=None)
except KeyError:
logger.info("Config section not found. Backing up existing file (if any) and readding a section header")
if self.filename.exists():
(self.filename.parent / f'{appname}.ini.backup').write_bytes(self.filename.read_bytes())
# Load INI
self.config = ConfigParser(interpolation=None)
self.config.read(self.filename)
# Ensure section exists
if self.SECTION not in self.config:
self.config.add_section(self.SECTION)
if (plugdir_str := self.get_str('plugin_dir')) is None or not pathlib.Path(plugdir_str).is_dir():
self.set("plugin_dir", str(self.default_plugin_dir_path))
plugdir_str = self.default_plugin_dir
self.plugin_dir_path = pathlib.Path(plugdir_str)
self.plugin_dir_path.mkdir(exist_ok=True)
def _get_settings_dict(self) -> dict[str, Any]:
"""Return all keys/values from SECTION as a dict of strings."""
return {
key: self.config[self.SECTION][key] for key in self.config[self.SECTION]
}
if (outdir := self.get_str('outdir')) is None or not pathlib.Path(outdir).is_dir():
self.set('outdir', self.home)
def write_to_toml(self, toml_path: str) -> None:
"""Dump existing config to TOML file."""
config_logger.debug("Generating New Config File")
config_data: dict[str, Any] = {
"generated": datetime.datetime.now(datetime.timezone.utc).isoformat(),
"source": "linux_ini",
"section": self.SECTION,
"settings": self._get_settings_dict(),
}
def __escape(self, s: str) -> str:
"""
Escape a string using self.__escape_lut.
This does NOT support multi-character escapes.
:param s: str - String to be escaped.
:return: str - The escaped string.
"""
out = ""
for c in s:
if c not in self.__escape_lut:
out += c
continue
out += '\\' + self.__escape_lut[c]
return out
def __unescape(self, s: str) -> str:
"""
Unescape a string.
:param s: str - The string to unescape.
:return: str - The unescaped string.
"""
out: list[str] = []
i = 0
while i < len(s):
c = s[i]
if c != '\\':
out.append(c)
i += 1
continue
# We have a backslash, check what its escaping
if i == len(s)-1:
raise ValueError('Escaped string has unescaped trailer')
unescaped = self.__unescape_lut.get(s[i+1])
if unescaped is None:
raise ValueError(f'Unknown escape: \\ {s[i+1]}')
out.append(unescaped)
i += 2
return "".join(out)
def __raw_get(self, key: str) -> str | None:
"""
Get a raw data value from the config file.
:param key: str - The key data is being requested for.
:return: str - The raw data, if found.
"""
if self.config is None:
raise ValueError('Attempt to use a closed config')
return self.config[self.SECTION].get(key)
def get_str(self, key: str, *, default: str | None = None) -> str:
"""
Return the string referred to by the given key if it exists, or the default.
Implements :meth:`AbstractConfig.get_str`.
"""
data = self.__raw_get(key)
if data is None:
return default # type: ignore # Yes it could be None, but we're _assuming_ that people gave us a default
if '\n' in data:
raise ValueError('asked for string, got list')
return self.__unescape(data)
def get_list(self, key: str, *, default: list | None = None) -> list:
"""
Return the list referred to by the given key if it exists, or the default.
Implements :meth:`AbstractConfig.get_list`.
"""
data = self.__raw_get(key)
if data is None:
return default # type: ignore # Yes it could be None, but we're _assuming_ that people gave us a default
split = data.split('\n')
if split[-1] != ';':
raise ValueError('Encoded list does not have trailer sentinel')
return list(map(self.__unescape, split[:-1]))
def get_int(self, key: str, *, default: int = 0) -> int:
"""
Return the int referred to by key if it exists in the config.
Implements :meth:`AbstractConfig.get_int`.
"""
data = self.__raw_get(key)
if data is None:
return default
try:
return int(data)
except ValueError as e:
raise ValueError(f'requested {key=} as int cannot be converted to int') from e
def get_bool(self, key: str, *, default: bool | None = None) -> bool:
"""
Return the bool referred to by the given key if it exists, or the default.
Implements :meth:`AbstractConfig.get_bool`.
"""
if self.config is None:
raise ValueError('attempt to use a closed config')
data = self.__raw_get(key)
if data is None:
return default # type: ignore # Yes it could be None, but we're _assuming_ that people gave us a default
return bool(int(data))
def set(self, key: str, val: int | str | list[str]) -> None:
"""
Set the given key's data to the given value.
Implements :meth:`AbstractConfig.set`.
"""
if self.config is None:
raise ValueError('attempt to use a closed config')
to_set: str | None = None
if isinstance(val, bool):
to_set = str(int(val))
elif isinstance(val, str):
to_set = self.__escape(val)
elif isinstance(val, int):
to_set = str(val)
elif isinstance(val, list):
to_set = '\n'.join([self.__escape(s) for s in val] + [';'])
else:
raise ValueError(f'Unexpected type for value {type(val)=}')
self.config.set(self.SECTION, key, to_set)
self.save()
def delete(self, key: str, *, suppress=False) -> None:
"""
Delete the given key from the config.
Implements :meth:`AbstractConfig.delete`.
"""
if self.config is None:
raise ValueError('attempt to use a closed config')
self.config.remove_option(self.SECTION, key)
self.save()
def save(self) -> None:
"""
Save the current configuration.
Implements :meth:`AbstractConfig.save`.
"""
if self.config is None:
raise ValueError('attempt to use a closed config')
with open(self.filename, 'w', encoding='utf-8') as f:
self.config.write(f)
path_obj = pathlib.Path(toml_path)
path_obj.parent.mkdir(parents=True, exist_ok=True)
config_logger.debug("Writing Config File")
with path_obj.open("wb") as f:
tomli_w.dump(config_data, f)
def close(self) -> None:
"""
Close this config and release any associated resources.
Implements :meth:`AbstractConfig.close`.
"""
self.save()
"""Release resources (stub)."""
self.config = None
def linux_helper(config: Config) -> Config:
"""Set Environment Specific Variables for Linux Config."""
config_logger.debug("Linux environment detected. Setting platform-specific variables.")
config.respath_path = pathlib.Path(__file__).parent.parent
config.internal_plugin_dir_path = config.respath_path / "plugins"
config.default_journal_dir_path = None # type: ignore
if (outdir := config.get_str("outdir")) is None or not pathlib.Path(
outdir
).is_dir():
config.set("outdir", config.home)
return config
+65 -233
View File
@@ -5,21 +5,23 @@ Copyright (c) EDCD, All Rights Reserved
Licensed under the GNU General Public License.
See LICENSE file.
"""
from __future__ import annotations
import functools
import pathlib
import sys
import uuid
import winreg
import datetime
import tomli_w
import base64
from typing import Literal, Any
from config import AbstractConfig, applongname, appname, logger
from config import config_logger
from win32comext.shell import shell
from typing import Any, TYPE_CHECKING
if sys.platform != 'win32':
if TYPE_CHECKING:
from . import Config
if sys.platform != "win32":
raise EnvironmentError("This file is for Windows only.")
REG_RESERVED_ALWAYS_ZERO = 0
@@ -30,254 +32,84 @@ def known_folder_path(guid: uuid.UUID) -> str | None:
return shell.SHGetKnownFolderPath(guid, 0, 0)
class WinConfig(AbstractConfig):
"""Implementation of AbstractConfig for Windows."""
class WinConfigMinimal:
"""Minimal Windows config for exporting registry values to TOML."""
REGISTRY_SUBKEY = r"Software\Marginal\EDMarketConnector"
def __init__(self) -> None:
super().__init__()
self.REGISTRY_SUBKEY = r'Software\Marginal\EDMarketConnector'
create_key_defaults = functools.partial(
winreg.CreateKeyEx,
key=winreg.HKEY_CURRENT_USER,
# Open or create the registry key
self.__reg_handle = winreg.CreateKeyEx(
winreg.HKEY_CURRENT_USER,
self.REGISTRY_SUBKEY,
access=winreg.KEY_ALL_ACCESS | winreg.KEY_WOW64_64KEY,
)
try:
self.__reg_handle: winreg.HKEYType = create_key_defaults(sub_key=self.REGISTRY_SUBKEY)
except OSError:
logger.exception('Could not create required registry keys')
raise
if local_appdata := known_folder_path(shell.FOLDERID_LocalAppData):
self.app_dir_path = pathlib.Path(local_appdata) / appname
self.app_dir_path.mkdir(exist_ok=True)
self.default_plugin_dir_path = self.app_dir_path / 'plugins'
if (plugdir_str := self.get_str('plugin_dir')) is None or not pathlib.Path(plugdir_str).is_dir():
self.set("plugin_dir", str(self.default_plugin_dir_path))
plugdir_str = self.default_plugin_dir
self.plugin_dir_path = pathlib.Path(plugdir_str)
self.plugin_dir_path.mkdir(exist_ok=True)
if getattr(sys, 'frozen', False):
self.respath_path = pathlib.Path(sys.executable).parent
self.internal_plugin_dir_path = self.respath_path / 'plugins'
else:
self.respath_path = pathlib.Path(__file__).parent.parent
self.internal_plugin_dir_path = self.respath_path / 'plugins'
self.home_path = pathlib.Path.home()
journal_dir_path = pathlib.Path(
known_folder_path(shell.FOLDERID_SavedGames)) / 'Frontier Developments' / 'Elite Dangerous' # type: ignore
self.default_journal_dir_path = journal_dir_path if journal_dir_path.is_dir() else None # type: ignore
self.identifier = applongname
if (outdir_str := self.get_str('outdir')) is None or not pathlib.Path(outdir_str).is_dir():
docs = known_folder_path(shell.FOLDERID_Documents)
self.set("outdir", docs if docs is not None else self.home)
def _convert_reg_value_for_toml(self, data, data_type):
"""Convert Windows registry values into TOML-native types for 6.0's global config."""
# Plain string
def _convert_reg_value_for_toml(self, data, data_type) -> Any:
"""Convert Windows registry values into TOML-native types."""
if data_type == winreg.REG_SZ:
return str(data)
# Expandable string: keep as string
if data_type == winreg.REG_EXPAND_SZ:
try:
expanded = winreg.ExpandEnvironmentStrings(data)
return str(winreg.ExpandEnvironmentStrings(data))
except Exception:
expanded = data
return str(expanded)
# DWORD or QWORD: store as integer
return str(data)
if data_type in (winreg.REG_DWORD, winreg.REG_QWORD):
return int(data)
# Multi-string: store as list
if data_type == winreg.REG_MULTI_SZ:
return list(data)
# Binary: store as base64 string
if data_type == winreg.REG_BINARY:
return base64.b64encode(data).decode("ascii")
# Else, force to string
# Fallback: force to string
return str(data)
def write_registry_to_toml(self, toml_path: str) -> None:
"""
Export all registry values under self.REGISTRY_SUBKEY into a global TOML config.
Values are normalized into TOML-native types (str, int, list, base64).
Booleans are left as ints (0/1) and should be interpreted adaptively by get_bool.
"""
def write_to_toml(self, toml_path: str) -> None:
"""Export all registry values under REGISTRY_SUBKEY to TOML."""
config_logger.debug("Generating New Config File")
key = self.__reg_handle
_, num_values, _ = winreg.QueryInfoKey(key)
try:
_, num_values, _ = winreg.QueryInfoKey(key)
config_data: dict[str, Any] = {
"generated": datetime.datetime.now(datetime.timezone.utc).isoformat(),
"source": "windows_registry",
"settings": {},
}
# Write config, including writing where and when we got this from.
config_data: dict[str, Any] = {
"generated": datetime.datetime.now(datetime.timezone.utc).isoformat(),
"source": "windows_registry",
"subkey": self.REGISTRY_SUBKEY,
"settings": {}
}
for i in range(num_values):
name, data, data_type = winreg.EnumValue(key, i)
config_data["settings"][name] = self._convert_reg_value_for_toml(
data, data_type
)
settings = config_data["settings"]
for i in range(num_values):
name, data, data_type = winreg.EnumValue(key, i)
settings[name] = self._convert_reg_value_for_toml(data, data_type)
# Write TOML to file
path_obj = pathlib.Path(toml_path)
path_obj.parent.mkdir(parents=True, exist_ok=True)
with path_obj.open("wb") as f:
tomli_w.dump(config_data, f)
logger.info(f"Registry exported to TOML: {toml_path}")
except OSError:
logger.exception(f"Could not export registry key {self.REGISTRY_SUBKEY}")
raise
def __get_regentry(self, key: str) -> None | list | str | int:
"""Access the Registry for the raw entry."""
try:
value, _type = winreg.QueryValueEx(self.__reg_handle, key)
except FileNotFoundError:
# Key doesn't exist
return None
# For programmers who want to actually know what is going on
if _type == winreg.REG_SZ:
return str(value)
if _type == winreg.REG_DWORD:
return int(value)
if _type == winreg.REG_MULTI_SZ:
return list(value)
logger.warning(f'Registry key {key=} returned unknown type {_type=} {value=}')
return None
def get_str(self, key: str, *, default: str | None = None) -> str:
"""
Return the string referred to by the given key if it exists, or the default.
Implements :meth:`AbstractConfig.get_str`.
"""
res = self.__get_regentry(key)
if res is None:
return default # type: ignore # Yes it could be None, but we're _assuming_ that people gave us a default
if not isinstance(res, str):
raise ValueError(f'Data from registry is not a string: {type(res)=} {res=}')
return res
def get_list(self, key: str, *, default: list | None = None) -> list:
"""
Return the list referred to by the given key if it exists, or the default.
Implements :meth:`AbstractConfig.get_list`.
"""
res = self.__get_regentry(key)
if res is None:
return default # type: ignore # Yes it could be None, but we're _assuming_ that people gave us a default
if not isinstance(res, list):
raise ValueError(f'Data from registry is not a list: {type(res)=} {res}')
return res
def get_int(self, key: str, *, default: int = 0) -> int:
"""
Return the int referred to by key if it exists in the config.
Implements :meth:`AbstractConfig.get_int`.
"""
res = self.__get_regentry(key)
if res is None:
return default
if not isinstance(res, int):
raise ValueError(f'Data from registry is not an int: {type(res)=} {res}')
return res
def get_bool(self, key: str, *, default: bool | None = None) -> bool:
"""
Return the bool referred to by the given key if it exists, or the default.
Implements :meth:`AbstractConfig.get_bool`.
"""
res = self.get_int(key, default=default) # type: ignore
if res is None:
return default # Yes it could be None, but we're _assuming_ that people gave us a default
return bool(res)
def set(self, key: str, val: int | str | list[str] | bool) -> None:
"""
Set the given key's data to the given value.
Implements :meth:`AbstractConfig.set`.
"""
# These are the types that winreg.REG_* below resolve to.
reg_type: Literal[1] | Literal[4] | Literal[7]
if isinstance(val, str):
reg_type = winreg.REG_SZ
elif isinstance(val, int):
reg_type = winreg.REG_DWORD
elif isinstance(val, list):
reg_type = winreg.REG_MULTI_SZ
elif isinstance(val, bool):
reg_type = winreg.REG_DWORD
val = int(val)
else:
raise ValueError(f'Unexpected type for value {type(val)=}')
winreg.SetValueEx(self.__reg_handle, key, REG_RESERVED_ALWAYS_ZERO, reg_type, val) # type: ignore
def delete(self, key: str, *, suppress=False) -> None:
"""
Delete the given key from the config.
'key' is relative to the base Registry path we use.
Implements :meth:`AbstractConfig.delete`.
"""
try:
winreg.DeleteValue(self.__reg_handle, key)
except OSError:
if suppress:
return
raise
def save(self) -> None:
"""
Save the configuration.
Not required for WinConfig as Registry keys are flushed on write.
"""
pass
path_obj = pathlib.Path(toml_path)
path_obj.parent.mkdir(parents=True, exist_ok=True)
config_logger.debug("Writing Config File")
with path_obj.open("wb") as f:
tomli_w.dump(config_data, f)
def close(self):
"""
Close this config and release any associated resources.
Implements :meth:`AbstractConfig.close`.
"""
"""Close registry handle."""
self.__reg_handle.Close()
def win_helper(config: Config) -> Config:
"""Set Environment Specific Variables for Windows Config."""
config_logger.debug("Windows environment detected. Setting platform-specific variables.")
if getattr(sys, "frozen", False):
config.respath_path = pathlib.Path(sys.executable).parent
config.internal_plugin_dir_path = config.respath_path / "plugins"
else:
config.respath_path = pathlib.Path(__file__).parent.parent
config.internal_plugin_dir_path = config.respath_path / "plugins"
journal_dir_path = (
pathlib.Path(known_folder_path(shell.FOLDERID_SavedGames)) # type: ignore
/ "Frontier Developments"
/ "Elite Dangerous"
)
config.default_journal_dir_path = journal_dir_path if journal_dir_path.is_dir() else None # type: ignore
if (outdir_str := config.get_str("outdir")) is None or not pathlib.Path(
outdir_str
).is_dir():
docs = known_folder_path(shell.FOLDERID_Documents)
config.set("outdir", docs if docs is not None else config.home)
return config
+1
View File
@@ -6,3 +6,4 @@ semantic-version==2.10.0
# For manipulating folder permissions and the like.
pywin32==311; sys_platform == 'win32'
psutil==7.1.3
tomli-w==1.2.0