Source code for prsctrl_gui.utility.config

"""Static app configuration.


"""
import os
from typing import Optional, overload, Union
from devctrl.utility.settings import SettingsClass, dataclass, field
from devctrl.utility.email import SmtpSettings
from ..version import version
from ..measurement._new_power_calibration import NewPowerCalibration


[docs] @dataclass class AppCache(SettingsClass): geometry_height: int = field( default=700, metadata={"doc": "Height of the application window."} ) geometry_width: int = field( default=1000, metadata={"doc": "Width of the application window."} ) geometry_x: int = field( default=100, metadata={"doc": "X position of the application window."} ) geometry_y: int = field( default=100, metadata={"doc": "Y position of the application window."} ) last_measurement_load_dir: Optional[str] = field( default=None, metadata={"doc": "Directory from which the user last loaded measurement data."} ) last_measurement_save_dir: Optional[str] = field( default=None, metadata={"doc": "Directory to which the user last saved measurement data."} ) last_measurement_save_ext: str = field( default=".hdf5", metadata={"doc": "File extension used for the last saved measurement data."} ) last_power_calibration_settings: Optional[NewPowerCalibration] = field( default=None, metadata={ "doc": "Last inputs of the PowerCalibration widget.", "group": "runtime" } ) last_task_dir: str = field( default="~/", metadata={"doc": "Path where a task was last saved/loaded.", "group": "runtime"} ) last_tasklist_dir: str = field( default="~/", metadata={"doc": "Path where a tasklist was last saved/loaded.", "group": "runtime"} ) global_metadata_input: dict[str, str] = field( default_factory=dict, metadata={"doc": "Input of the global metadata widget."} )
[docs] @dataclass class AppSettings(SettingsClass): app_language: str = field( default="en", metadata={"doc": "Language of the application. Requires a restart to apply.", "valid": ["en", "de", "de_BV"]} ) system_path_append: list[str] = field( default_factory=list, metadata={"doc": "List of paths to append to the system path.\nThis is required to locate proprietary binaries, like the Bentham SKD or the Cleware Power Switch executable."} ) console_history_path: str = field( default="<CACHE_DIR>/console-history.txt", metadata={"doc": "Path to the console command history file."} ) data_dir: str|list[str] = field( default="~/prsctrl-gui-data", metadata={"doc": "Directory/directories to put the final data in after measurements.", "group": "data"} ) data_dir_archive: str|list[str] = field( default="~/prsctrl-gui-archive", metadata={"doc": "Directory/directories to put the archived data (single-file zip) after measurements.", "group": "data"} ) archive_file_count_threshold: int = field( default=20, metadata={"doc": "Minimum number of measurement points (wavelengths) the data must have to be archived.", "group": "data"} ) data_dir_cache: str = field( default="<CACHE_DIR>/data", metadata={"doc": "Directory for storing temporary data during measurements.", "group": "data"} ) reference_data_dir: Optional[str] = field( default=None, metadata={"doc": "Directory containing the DC measurement for calculating absorbance spectra (read-only).", "group": "data"} ) log_dir: str = field( default="<CACHE_DIR>/logs", metadata={"doc": "Directory for storing log files."} ) plot_max_data_points: int = field( default=20000, metadata={"doc": "Maximum number of data points to plot."} ) plot_update_interval_ms: int = field( default=10, metadata={"doc": "Interval in milliseconds between plot updates."} ) device_auto_reconnect: bool = field( default=True, metadata={"doc": "Whether to automatically reconnect to the last connected devices.", "group": "device"} ) tasklist_path: str = field( default="<CACHE_DIR>/tasklist.yaml", metadata={"doc": "Path to the task list YAML file."} ) power_switch_autoconnect_devices_timeout_s: int = field( default=10, metadata={"doc": "Timeout in seconds to wait with auto-reconnect after powering on devices.", "group": "device"} ) idle_update_interval_s: int = field( default=10, metadata={"doc": "Interval in seconds to check for device connectivity while idling.", "group": "device"} ) error_stack_trace: bool = field( default=True, metadata={"doc": "Whether to show stack traces when errors occur.", "group": "error"} ) error_show_dialog: bool = field( default=True, metadata={"doc": "Whether to show error dialogs when errors occur.", "group": "error"} ) compress_data: bool = field( default=True, metadata={"doc": "Whether to compress data when saving to disk.", "group": "data"} ) power_switch_exe: Optional[str] = field( default=None, metadata={"doc": "Path to the power switch executable.", "group": "device"} ) smtp_settings: Optional[SmtpSettings] = field( default=None, metadata={"doc": "SMTP configuration for sending emails."} ) sample_db: str = field( default="<CONFIG_DIR>/samples.yaml", metadata={"doc": "Path to the sample database file.", "group": "runtime"} ) power_calibration_db: str = field( default="<CONFIG_DIR>/power-calibration.yaml", metadata={"doc": "Path to the pump-power-calibration database file.", "group": "runtime"} ) setup_info_db: str = field( default="<CONFIG_DIR>/setup-info.yaml", metadata={"doc": "Path to the setup-info database file.", "group": "runtime"} ) task_preset_dir: str = field( default="<CONFIG_DIR>/task-presets", metadata={"doc": "Path to the task-preset directory.", "group": "runtime"} ) device_connection_infos_path: str = field( default="<CACHE_DIR>/devices.yaml", metadata={"doc": "File to store which devices were connected.", "group": "runtime"} )
[docs] class AppConfig: """Configuration collection """ APP_NAME: str = "prsctrl-gui" APP_VERSION: str = version CONFIG_DIR: str = os.path.expanduser("~/.config/prsctrl-gui") CACHE_DIR: str = os.path.expanduser("~/.cache/prsctrl-gui") MAIN_CFG: AppSettings = AppSettings() MAIN_CFG_PATH: str = os.path.expanduser("~/.config/prsctrl-gui/prsctrl-gui.yaml") CACHE_CFG: AppCache = AppCache() CACHE_CFG_PATH: str = os.path.expanduser("~/.cache/prsctrl-gui/prsctrl-gui.yaml")
[docs] @classmethod def initialize(cls) -> None: """Initialize configuration Perform any necessary initializations here, e.g.: - Set correct configuration directory (high to low priority): 1. $PRSCTRL_GUI_CONFIG 2. $XDG_CONFIG_HOME/prsctrl-gui 3. ~/.config/prsctrl-gui - Set correct cache directory (high to low priority): 1. $PRSCTRL_GUI_CACHE 2. $XDG_CACHE_HOME/prsctrl-gui 3. ~/.cache/prsctrl-gui - Loading settings from a file: """ # Main CFG if 'XDG_CONFIG_HOME' in os.environ: AppConfig.CONFIG_DIR = os.path.join(os.environ["XDG_CONFIG_HOME"], "prsctrl-gui") if 'PRSCTRL_GUI_CONFIG' in os.environ: AppConfig.CONFIG_DIR = os.environ["PRSCTRL_GUI_CONFIG"] if not os.path.isdir(AppConfig.CONFIG_DIR): os.makedirs(AppConfig.CONFIG_DIR) AppConfig.MAIN_CFG_PATH = os.path.join(AppConfig.CONFIG_DIR, "prsctrl-gui.yaml") main_cfg_exists = os.path.isfile(AppConfig.MAIN_CFG_PATH) # log not set up at this point -> use print print(f"MAIN_CFG_PATH = '{AppConfig.MAIN_CFG_PATH}'. Exists? {main_cfg_exists}") if main_cfg_exists: loaded_main = AppSettings.load_yaml(AppConfig.MAIN_CFG_PATH) AppConfig.MAIN_CFG |= loaded_main # Cache CFG if 'XDG_CACHE_HOME' in os.environ: AppConfig.CACHE_DIR = os.path.join(os.environ["XDG_CACHE_HOME"], "prsctrl-gui") if 'PRSCTRL_GUI_CACHE' in os.environ: AppConfig.CACHE_DIR = os.environ["PRSCTRL_GUI_CACHE"] if not os.path.isdir(AppConfig.CACHE_DIR): os.makedirs(AppConfig.CACHE_DIR) AppConfig.CACHE_CFG_PATH = os.path.join(AppConfig.CACHE_DIR, "prsctrl-gui.yaml") cache_cfg_exists = os.path.isfile(AppConfig.CACHE_CFG_PATH) print(f"CACHE_CFG_PATH = '{AppConfig.CACHE_CFG_PATH}'. Exists? {cache_cfg_exists}") if cache_cfg_exists: loaded_cache = AppCache.load_yaml(AppConfig.CACHE_CFG_PATH) AppConfig.CACHE_CFG |= loaded_cache # Get the Version from importlib # This will always show the version of the installed package, not the local development version from importlib.metadata import version, PackageNotFoundError try: AppConfig.APP_VERSION = version("prsctrl-gui") except PackageNotFoundError: pass
[docs] @classmethod def finalize(cls) -> None: """Write configuration to files. """ AppConfig.MAIN_CFG.save_yaml(AppConfig.MAIN_CFG_PATH) AppConfig.CACHE_CFG.save_yaml(AppConfig.CACHE_CFG_PATH)
@classmethod @overload def get_path(cls, path: None, check_can_read: bool = False, check_can_write: bool = False) -> None: ... @classmethod @overload def get_path(cls, path: str, check_can_read: bool = False, check_can_write: bool = False) -> str: ... @classmethod @overload def get_path(cls, path: list[str], check_can_read: bool = False, check_can_write: bool = False) -> list[str]: ...
[docs] @classmethod def get_path(cls, path: str | list[str] | None, check_can_read: bool = False, check_can_write: bool = False) -> str | list[str] | None: """Substitute a string for a path Substitutes: - ~ - <CONFIG_DIR> for ``CONFIG_DIR`` - <CACHE_DIR> for ``CACHE_DIR`` :param path: String to apply substitutions to. May be None or a list of paths/None as well. :param check_can_read: If True, check that the path exists and can be read. If not, raise an exception. :param check_can_write: If True, check that the path exists and can be written. If not, raise an exception. :raises ValueError: If the path cannot be read or written (only when check_can_read or check_can_write is True) :return: The path with substitutions applied. If path is None, None is returned. If list of paths was given, a list is returned. """ def _get_path(path): if path is None: if check_can_read: raise ValueError("No path given, path is not readable.") if check_can_write: raise ValueError("No path given, path is not writable.") return path expanded_path = os.path.expanduser(path) expanded_path = expanded_path.replace("<CONFIG_DIR>", cls.CONFIG_DIR) expanded_path = expanded_path.replace("<CACHE_DIR>", cls.CACHE_DIR) if check_can_read: if not os.path.exists(expanded_path): raise ValueError(f"Path does not exist: '{expanded_path}'") if not os.access(expanded_path, os.R_OK): raise ValueError(f"Path cannot be read: '{expanded_path}'") # Check if path exists and can be written if requested if check_can_write: if not os.path.exists(expanded_path): # If path doesn't exist, check if parent directory is writable parent_dir = os.path.dirname(expanded_path) or "." if not os.access(parent_dir, os.W_OK): raise ValueError(f"Parent directory is not writable: '{parent_dir}'") else: if not os.access(expanded_path, os.W_OK): raise ValueError(f"Path cannot be written: '{expanded_path}'") return expanded_path if isinstance(path, list): return [_get_path(p) for p in path] else: return _get_path(path)