# Copyright (c) 2025-2026 Contributors to the AdsorPy project.
# SPDX-License-Identifier: MIT
"""GUI module of adsorpy.""" # TODO: Make a new repo for this!
from __future__ import annotations
# __lazy_modules__ = ["json", "multiprocessing", "pickle", "zipfile", "h5py", "matplotlib", "pandas", "seaborn", "dask"]
# """Modules that are imported lazily (only when needed) for 3.15+, but regularly for older Python. Place at top."""
import inspect
import io
import json
import multiprocessing
import pickle
import re
import sys
import textwrap
import webbrowser
import zipfile
from collections import defaultdict
if sys.version_info >= (3, 11):
from datetime import UTC, datetime # For datetime stamping and seed generation.
from typing import Unpack
else:
from datetime import datetime
from typing_extensions import Unpack
if sys.version_info >= (3, 12):
from typing import TypedDict, override
else:
from typing_extensions import TypedDict, override
from itertools import count
from pathlib import Path
from typing import (
TYPE_CHECKING,
ClassVar,
Generic,
Literal,
ParamSpec,
TypeAlias,
TypeGuard,
TypeVar,
cast,
get_origin,
get_type_hints,
)
import matplotlib as mpl
import matplotlib.pyplot as plt
import numpy as np
from matplotlib import patches
from pydantic import (
ConfigDict,
NonNegativeInt,
PositiveFloat,
PositiveInt,
TypeAdapter,
ValidationError,
with_config,
)
from pydantic_core import core_schema
from PySide6.QtCore import (
QByteArray,
QMarginsF,
QObject,
QRect,
QRegularExpression,
QRunnable,
QSettings,
Qt,
QThreadPool,
Signal,
Slot,
)
from PySide6.QtGui import (
QAction,
QDoubleValidator,
QDropEvent,
QGuiApplication,
QIcon,
QIntValidator,
QPageLayout,
QPageSize,
QPainter,
QPdfWriter,
QPixmap,
QRegularExpressionValidator,
QResizeEvent,
QWheelEvent,
)
from PySide6.QtSvg import QSvgGenerator, QSvgRenderer
from PySide6.QtSvgWidgets import QSvgWidget
from PySide6.QtWidgets import (
QAbstractItemView,
QApplication,
QCheckBox,
QComboBox,
QDoubleSpinBox,
QFileDialog,
QFrame,
QGridLayout,
QGroupBox,
QHBoxLayout,
QLabel,
QLayout,
QLayoutItem,
QLineEdit,
QListWidget,
QListWidgetItem,
QMainWindow,
QMessageBox,
QProgressBar,
QPushButton,
QScrollArea,
QSpinBox,
QSplitter,
QTabWidget,
QVBoxLayout,
QWidget,
)
from shapely import Polygon, from_geojson
from shapely.geometry import mapping
from shiboken6 import Shiboken
try:
import h5py
import pandas as pd
import seaborn as sns
from dask.delayed import delayed
from dask.distributed import Client, Future, as_completed
except ImportError as e:
print("\n[Error] GUI requirements are missing!", file=sys.stderr)
print("Please install the GUI sub-requirements using:\n", file=sys.stderr)
print(" pip install adsorpy[gui-deps]\n", file=sys.stderr)
print(f"Details: {e}", file=sys.stderr)
raise
from adsorpy import __version__, molecule_lib
from adsorpy.run_simulation import run_simulation, show_surface
T_qobj = TypeVar("T_qobj", bound=QObject)
T_inv = TypeVar("T_inv", bool, int, str, float)
P_mol = ParamSpec("P_mol") # Helps with static type checkers.
P = ParamSpec("P")
R = TypeVar("R")
# T_widg = TypeVar("T_widg", bound=QWidget)
if TYPE_CHECKING:
from collections.abc import Callable, ItemsView, ValuesView
from dask.delayed import Delayed
from numpy.random import Generator
from adsorpy.randomsequentialadsorption import Simulator
from adsorpy.rsa_config import RsaConfig
from adsorpy.types import BoolArray, DistArray, FloatArray, GeoArray, IdxArray
InputWidget: TypeAlias = QSpinBox | QDoubleSpinBox | QLineEdit | "FilePickerWidget"
RunResult: TypeAlias = tuple[DistArray, DistArray, DistArray]
[docs]
def set_content(
widget: QSpinBox | QDoubleSpinBox | QLineEdit | FilePickerWidget,
content: str | float | list[str],
) -> None:
"""Set content of widget by matched content type.
:param widget: The widget being edited.
:param content: The content of the widget being edited.
:raises ValueError: If the content does not match the widget.
"""
match widget, content:
case (QSpinBox(), int()):
widget.setValue(content)
case (QDoubleSpinBox(), float()):
widget.setValue(content)
case (QLineEdit() | FilePickerWidget(), str()):
widget.setText(content)
case (QLineEdit(), list()):
widget.setText(",".join(content))
case _:
errmsg = f"Widget and content mismatch: {type(widget).__name__} and {type(content).__name__}"
raise ValueError(errmsg)
[docs]
def from_geojson_str_to_polygon(geojson_str: str) -> Polygon:
"""Convert from GeoJSON string to Polygon and validate geometry.
:param geojson_str: GeoJSON string to convert to Polygon.
:returns: Shapely Polygon.
:raises TypeError: If the string does not generate a Polygon.
:raises ValueError: If the generated Polygon is invalid.
:raises GEOSException: If the string cannot be parsed as a geojson.
"""
polygon = from_geojson(geojson_str)
if isinstance(polygon, Polygon):
if polygon.is_valid:
return polygon
errmsg = f"Polygon is invalid. Exterior coordinates: {polygon.exterior.coords}"
raise ValueError(errmsg)
errmsg = f"Geometry is of wrong type: {type(polygon).__name__}"
raise TypeError(errmsg)
[docs]
def validate_polygon(pol: Polygon | str | dict[str, str | list[list[list[float]]]]) -> Polygon:
"""Convert the GeoJSON dict data into a real Shapely Polygon or pass the data if it is already a Polygon.
:param pol: Polygon or GeoJSON format.
:returns: Polygon.
:raises TypeError: if the type cannot be converted to Polygon.
"""
if isinstance(pol, Polygon):
return pol
if isinstance(pol, dict):
# Convert the python dictionary to a valid JSON string first
json_str = json.dumps(pol)
return from_geojson_str_to_polygon(json_str)
if isinstance(pol, str):
# Turns {"type": "Polygon", "coordinates": ...} into a Shapely object
return from_geojson_str_to_polygon(pol)
errmsg = f"Cannot convert {type(pol)} to a Shapely Polygon"
raise TypeError(errmsg)
[docs]
class SimplePolygonDict(TypedDict):
"""Concise representation of a GeoJSON Polygon dictionary."""
type: Literal["Polygon"]
coordinates: list[list[list[float]]]
[docs]
class PydanticPolygon(Polygon):
"""A Pydantic-native wrapper type for a Shapely Polygon."""
@classmethod
def __get_pydantic_core_schema__(
cls,
_source_type: object,
_handler: Callable[[object], core_schema.CoreSchema],
) -> core_schema.CoreSchema:
"""Tell Pydantic exactly how to validate and serialise a Shapely Polygon.
Uses American spelling of 'serialize' to be compliant with most programming conventions.
:param _source_type: The source type to use for validation.
:param _handler: The handler function to use for validation.
:returns: The Pydantic core schema.
"""
def validate(value: Polygon | str | dict[str, str | list[list[list[float]]]]) -> Polygon:
return validate_polygon(value)
def serialize(instance: Polygon) -> SimplePolygonDict:
return cast("SimplePolygonDict", mapping(instance))
return core_schema.no_info_before_validator_function(
validate,
core_schema.any_schema(),
serialization=core_schema.plain_serializer_function_ser_schema(
serialize,
return_schema=core_schema.dict_schema(),
),
)
ParamName = Literal[
"radius",
"distance",
"x_offset",
"y_offset",
"quad_segs",
"scale",
"verts",
"roundedness",
"file_name",
"ignore_atoms",
"roll",
"pitch",
"yaw",
"z_trim",
"reference_lattice_spacing",
]
[docs]
def is_valid_param(name: str) -> TypeGuard[ParamName]:
"""Check if a parameter name is valid.
:param name: Name of the parameter.
:returns: Boolean denoting validity.
"""
return name in ParamWidgets.__annotations__
[docs]
@with_config(ConfigDict(arbitrary_types_allowed=True))
class MoleculeParameters(TypedDict):
"""Molecule parameters dataclass.
:ivar index: Index of the molecule parameters configuration.
:ivar label: Label of the molecule parameters configuration, guaranteed to be unique.
:ivar function_name: Function name of the molecule.
:ivar refl_sym: Reflection symmetry.
:ivar rot_sym: Rotation symmetry.
:ivar rot_cnt: Rotation count (before accounting for reflection/rotation symmetry).
:ivar polygon: 2D polygon representation of the molecule.
:ivar settings: Function input of the molecule. Defaults to an empty dictionary.
"""
index: NonNegativeInt
label: str
function_name: str
refl_sym: bool
rot_sym: NonNegativeInt
rot_cnt: PositiveInt
polygon: PydanticPolygon
settings: dict[str, float | int | str]
[docs]
class SurfaceParameters(TypedDict, total=False):
"""Surface parameters dataclass.
:ivar lattice_type: Surface lattice type.
:ivar site_count: Site count of the surface.
:ivar lattice_a: Lattice spacing of the surface.
:ivar seed: RNG seed.
"""
lattice_type: Literal["hexagonal", "triangular", "honeycomb", "square"]
site_count: PositiveInt
lattice_a: PositiveFloat | None
seed: int | None
[docs]
class MiscParameters(TypedDict):
"""Miscellaneous parameters dataclass.
:ivar seed: RNG seed.
:ivar timestep_limit: Maximum allowed step count of the simulation.
"""
seed: NonNegativeInt | None
timestep_limit: NonNegativeInt | None
[docs]
class AppState(QObject, metaclass=AutoStateMeta): # pyright: ignore[reportMissingTypeArgument]
"""AppState class to communicate between tabs.
This class maintains synchronised states across the user interface. Changes
to any property automatically emit a corresponding ``<property>Changed`` signal.
:ivar seed_input: The Qt input widget holding the seed value.
:ivar step_limit: The maximum allowable processing steps.
:ivar misc_params: Miscellaneous parameters.
:ivar molecule_param_list: Settings of the molecule(s).
:ivar surface_params: Settings of the surface.
:ivar coverages: Coverage of simulation results.
:ivar fraction_of_covered_area: Fraction of covered area of simulation results.
:ivar gap_size_distribution: Gap size distribution of simulation results.
"""
seed_input: QLineEdit
step_limit: QSpinBox
misc_params: MiscParameters | None
molecule_param_list: list[MoleculeParameters] | None
surface_params: SurfaceParameters | None
coverages: tuple[DistArray, ...] | None
fraction_of_covered_area: tuple[DistArray, ...] | None
gap_size_distribution: DistArray | None
[docs]
class AdsorpyGUI(QMainWindow):
"""Main window application shell for the AdsorPy simulation engine framework.
Coordinates the primary window frame, top level configuration menu bars,
and hooks up the shared global data state across tab layout frames.
:cvar window_resized: Signal of (width, height) emitted when the main application window dimensions are modified.
"""
window_resized: Signal = Signal(int, int)
[docs]
def __init__(self) -> None:
"""Initialise frame parameters, global context caches, and child windows.
This is the main window of the AdsorPy simulation application.
"""
super().__init__()
self.setWindowTitle("AdsorPy Simulation GUI")
self.state = AppState()
"""Shared application runtime cache synchronised across all view frames."""
self._settings = QSettings(type(self).__name__)
"""Persistent platform configuration handle cached between user runtime sessions."""
# Delegate initialisation to helper workflows
self._init_menu_bar()
self._init_tabs()
[docs]
def _init_tabs(self) -> None:
"""Assemble the central tab frame layout and register sub-dashboards."""
self.tabs = QTabWidget()
"""Primary navigation container organizing distinct module windows."""
# Instantiate separate view models sharing the single source of truth state
self.tabs.addTab(GeneralSettings(self.state), "General")
self.tabs.addTab(SurfaceGeneration(self.state), "Surface")
self.tabs.addTab(MoleculeGeneration(self.state), "Molecule(s)")
self.setCentralWidget(self.tabs)
[docs]
def _save_settings_json(self) -> None:
"""Save settings to JSON file."""
# Validate seed
file_path, _ = QFileDialog.getSaveFileName(
self,
"Save Settings",
self._fetch_setting("last_visited_directory", default=""),
"JSON Files (*.json);;All Files (*)",
)
if not file_path:
return
seed_text = self.state.seed_input.text().strip()
step_limit_val = self.state.step_limit.value()
# Convert empty fields to None, or parse them to integers
seed_val = int(seed_text) if seed_text else None
misc_settings = MiscParameters(seed=seed_val, timestep_limit=step_limit_val)
surf_settings: SurfaceParameters = cast("SurfaceParameters", self.state.surface_params)
molecule_settings: list[MoleculeParameters] = cast("list[MoleculeParameters]", self.state.molecule_param_list)
try:
misc_adapter = TypeAdapter(MiscParameters)
surf_adapter = TypeAdapter(SurfaceParameters)
mol_adapter = TypeAdapter(list[MoleculeParameters])
misc_dump = misc_adapter.dump_python(misc_settings)
surf_dump = surf_adapter.dump_python(surf_settings)
mol_dump = mol_adapter.dump_python(molecule_settings)
misc_adapter.validate_python(misc_settings)
surf_adapter.validate_python(surf_settings)
mol_adapter.validate_python(molecule_settings)
except ValidationError as e:
QMessageBox.critical(
self,
"Error Saving File",
f"Failed to save settings. Structure or type constraints were broken:\n{e}",
)
return
combined_data = {
"adsorpy_version": __version__,
"miscellaneous_parameters": misc_dump,
"surface_parameters": surf_dump,
"molecule_parameters": mol_dump,
}
with Path(file_path).open("w", encoding="utf-8") as f:
json.dump(combined_data, f, indent=4)
QMessageBox.information(
self,
"Save Successful",
"Your simulation configuration settings have been successfully saved!",
)
[docs]
def _load_settings_json(self) -> None:
"""Load, validate, and version-check simulation settings profiles."""
file_path, _ = QFileDialog.getOpenFileName(
self,
"Open Settings",
self._fetch_setting("last_visited_directory", default=""),
"JSON Files (*.json);;All Files (*)",
)
if not file_path:
return # User cancelled the file selection dialogue
with Path(file_path).open("rb") as f:
json_bytes = f.read()
try:
raw_structure = TypeAdapter(dict).validate_json(json_bytes)
misc_adapter = TypeAdapter(MiscParameters)
surf_adapter = TypeAdapter(SurfaceParameters)
mol_adapter = TypeAdapter(list[MoleculeParameters])
# Validate and re-hydrate fields directly into application state
self.state.misc_params = misc_adapter.validate_python(raw_structure["miscellaneous_parameters"])
self.state.surface_params = surf_adapter.validate_python(raw_structure["surface_parameters"])
self.state.molecule_param_list = mol_adapter.validate_python(raw_structure["molecule_parameters"])
# Synchronise GUI with the newly loaded state
misc = self.state.misc_params
self.state.seed_input.setText(getattr(misc, "seed", ""))
# self.log("Settings successfully loaded and validated.")
except (KeyError, ValidationError) as e:
QMessageBox.critical(
self,
"Error Loading File",
f"Failed to parse settings file. Structure or type constraints were broken:\n{e}",
)
[docs]
def _fetch_setting(self, name: str, default: T_inv, return_type: type[T_inv] | None = None) -> T_inv:
"""Fetch settings by checking if they exist followed by their value.
:param name: The name of the setting to fetch.
:param default: The default value to return if the setting does not exist.
:param return_type: The default return type if the setting exists. If not given, type(default) is used.
:returns: The setting value if it exists, or else the default.
"""
check_type = type(default) if return_type is None else return_type
return cast("T_inv", self._settings.value(name, defaultValue=default, type=check_type))
[docs]
@override # This decorator is used to indicate a method overrides a method of the base class.
def resizeEvent(self, event: QResizeEvent) -> None:
"""Trigger automatically whenever the window size changes.
:param event: QResizeEvent, an event changing the window size.
"""
# Get the new size from the event object
new_size = event.size()
width: int = new_size.width()
height: int = new_size.height()
self.window_resized.emit(width, height)
super().resizeEvent(event)
[docs]
class GeneralSettings(QWidget):
"""General simulation configuration dashboard tab view.
Provides inputs for setting the execution step boundaries, absolute pseudo-random
number generator seeds, and renders real-time structural vector tracking maps.
"""
[docs]
def __init__(self, state: AppState) -> None:
"""Initialise validation engines and build structural control modules.
:param state: AppState object for communication between tab widgets.
"""
super().__init__() # Inherit from the super() class (in this case: AppState).
self._settings = QSettings(type(self).__name__)
"""Persistent platform configuration handle cached between user runtime sessions."""
self.state = state
"""App state object for communication between tab widgets."""
self.bg_signals = BackgroundTaskSignals()
"""Signals for the simulation background tasks."""
# Run UI Initialisation steps
self._init_validators()
# Extract widgets/layouts from the initialisation helpers
# (Assuming _init_controls sets up fields like seed, step limits, etc.)
controls_layout = self._init_controls()
svg_widget = self._init_svg_view()
# Create the Left Panel: Wrap controls layout inside a clean container QWidget
left_panel = QWidget()
left_panel.setLayout(controls_layout)
# Create the Centre Panel: Wrap SVG view in a QScrollArea for responsiveness
centre_scroll = QScrollArea()
centre_scroll.setWidgetResizable(True)
centre_scroll.setWidget(svg_widget)
self.state.surface_paramsChanged.connect(self._on_surface_changed) # pyright: ignore[reportAttributeAccessIssue]
self.state.molecule_param_listChanged.connect(self._on_molecules_changed) # pyright: ignore[reportAttributeAccessIssue]
self.input_metadata: BatchSimulationInput = BatchSimulationInput()
"""Dict of input values, to be stored as metadata."""
# Clean up scroll area borders to integrate smoothly with the splitter look
# centre_scroll.setFrameShape(QScrollArea.FrameShape.NoFrame)
# Create the Right Panel: Create a panel for listing generated arrays/molecules
# right_panel = self._init_management_panel() # Or: right_panel = QWidget()
# right_panel = QWidget()
# Unify sub-panels using exact splitter framework layout method
self._assemble_layout(left=left_panel, center=centre_scroll)
[docs]
def _fetch_setting(self, name: str, default: T_inv, return_type: type[T_inv] | None = None) -> T_inv:
"""Fetch settings by checking if they exist followed by their value.
:param name: The name of the setting to fetch.
:param default: The default value to return if the setting does not exist.
:param return_type: The default return type if the setting exists. If not given, type(default) is used.
:returns: The setting value if it exists, or else the default.
"""
check_type: type[T_inv] = type(default) if return_type is None else return_type
return cast("T_inv", self._settings.value(name, defaultValue=default, type=check_type))
[docs]
def _init_validators(self) -> None:
"""Instantiate validation models for text constraint processing."""
self._seed_validator = QRegularExpressionValidator()
"""Restricts string parameters strictly to absolute positive digits."""
self._seed_validator.setRegularExpression(QRegularExpression(r"^\d+$"))
[docs]
def _init_controls(self) -> QVBoxLayout:
"""Assemble environment settings selectors and connect state triggers.
:return: A populated vertical layout holding runtime widgets.
"""
layout = QVBoxLayout()
# Pseudo-random generator state seed tracking
layout.addWidget(QLabel("Optional Seed (positive int):"))
self.seed_input = QLineEdit()
"""Input widget capturing custom random generation bounds."""
self.seed_input.setValidator(self._seed_validator)
self.seed_input.setPlaceholderText("e.g. 23")
self.seed_input.setToolTip("RNG seed for the simulation. If empty, defaults to datetime in microseconds.")
layout.addWidget(self.seed_input)
# Sync the specific text field reference directly to global state tracking
self.state.seed_input = self.seed_input
# Absolute execution cycle step limit constraints
layout.addWidget(QLabel("Step limit (optional, > 0 int):"))
self.step_limit = QSpinBox()
"""Input widget restricting maximum sequential process cycles."""
self.step_limit.setToolTip("The maximum step limit of the simulation. Stops when done or when limit reached.")
# self.step_limit.setPlaceholderText("e.g. 1")
# self.step_limit.setValidator(self._gt_one_validator)
self.step_limit.setMinimum(0)
self.step_limit.setMaximum(100000000)
self.step_limit.setValue(cast("int", self.get_run_sim_default("timestep_limit")))
self.step_limit.setAccelerated(True)
self.step_limit.setStepType(QSpinBox.StepType.AdaptiveDecimalStepType)
self.state.step_limit = self.step_limit
layout.addWidget(self.step_limit)
layout.addWidget(_make_horizontal_line())
layout.addLayout(self._init_feedback_textboxes())
self.run_group = QGroupBox()
"""Simulation run group box."""
run_grid = QGridLayout(self.run_group)
self.run_button = QPushButton("Run Simulation (1x)")
"""Trigger execution wrapper for adsorpy run."""
self.run_button.setToolTip("Runs the random sequential adsorption simulation.")
run_grid.addWidget(self.run_button, 0, 1)
self.run_button.clicked.connect(self.run_simulation)
self.repeat_count = QSpinBox()
"""Repeat count value."""
self.repeat_count.setToolTip("Number of times to repeat. 100 is plenty for most purposes.")
self.repeat_count.setMinimum(1)
self.repeat_count.setMaximum(100000)
self.repeat_count.setValue(self._fetch_setting("repeat_count", 10))
self.repeat_count.setAccelerated(True)
self.repeat_count.valueChanged.connect(self._change_bulk_run_value)
run_grid.addWidget(self.repeat_count, 1, 0)
self.bulk_run_button = QPushButton(f"Bulk Run ({self.repeat_count.value()}x)")
"""Trigger execution wrapper for adsorpy bulk run."""
self.bulk_run_button.setToolTip("Runs the simulation multiple times in parallel.")
self.bulk_run_button.clicked.connect(self.run_batch_simulation)
run_grid.addWidget(self.bulk_run_button, 1, 1)
layout.addWidget(self.run_group)
layout.addWidget(_make_horizontal_line())
self.progress_bar = QProgressBar()
"""Progress bar for simulations."""
self.progress_bar.setToolTip("Simulation progress. 'Are we there yet?'")
self.progress_bar.setRange(0, 100) # Maps perfectly to percentages (0 to 100)
self.progress_bar.setValue(0) # Start empty
self.progress_bar.hide()
layout.addWidget(self.progress_bar, stretch=1, alignment=Qt.AlignmentFlag.AlignTop)
self.coverage_label = QLabel("")
"""Coverage value."""
self.coverage_label.hide()
self.coverage_label.setToolTip("Fraction of surface sites consumed by molecules.")
layout.addWidget(self.coverage_label, alignment=Qt.AlignmentFlag.AlignTop)
self.covered_area_label = QLabel("")
"""Fraction of covered area value."""
self.coverage_label.setToolTip("Fraction of surface area covered by molecule footprints.")
self.covered_area_label.hide()
layout.addWidget(self.covered_area_label, alignment=Qt.AlignmentFlag.AlignTop)
self.export_results_button = QPushButton("Export Results")
"""Button to export the results."""
self.export_results_button.setToolTip("Export results by format of choice.")
self.export_results_button.hide()
self.export_results_button.clicked.connect(self.export_results)
layout.addWidget(self.export_results_button, alignment=Qt.AlignmentFlag.AlignTop)
layout.addStretch()
return layout
[docs]
def _change_bulk_run_value(self, run_count: int) -> None:
"""Change the bulk run button tooltip.
:param run_count: The number of times to repeat the simulation.
"""
self.bulk_run_button.setText(f"Bulk Run ({run_count}x)")
self._settings.setValue("repeat_count", run_count)
[docs]
def _init_feedback_textboxes(self) -> QGridLayout:
"""Provide text to show the user whether data has been loaded."""
grid_layout = QGridLayout()
self.initiated_surface_label = QLabel("Surface:")
"""Surface label."""
self.initiated_surface_textbox = QLabel("Default.")
"""What kind of surface has been loaded."""
self.initiated_molecules_label = QLabel("Molecule(s):")
"""Molecule label."""
self.initiated_molecules_textbox = QLabel("Default.")
"""How many molecules has been loaded."""
grid_layout.addWidget(self.initiated_surface_label, 0, 0)
grid_layout.addWidget(self.initiated_surface_textbox, 0, 1)
grid_layout.addWidget(self.initiated_molecules_label, 1, 0)
grid_layout.addWidget(self.initiated_molecules_textbox, 1, 1)
return grid_layout
[docs]
def _on_surface_changed(self, params: SurfaceParameters | None) -> None:
"""Fire instantly when surface_params changes in another tab."""
if params is not None:
self.initiated_surface_textbox.setText("User-defined.")
else:
self.initiated_surface_textbox.setText("Default.")
[docs]
def _on_molecules_changed(self, mol_list: list[MoleculeParameters] | None) -> None:
"""Fire instantly when molecule_param_list changes in another tab."""
if mol_list: # Checks if list exists and is not empty
count: int = len(mol_list)
self.initiated_molecules_textbox.setText(f"{count} molecule{'s' * bool(count - 1)} defined by user.")
else:
self.initiated_molecules_textbox.setText("Default.")
[docs]
def _init_svg_view(self) -> QSvgWidget:
"""Construct the graphics frame and isolate structural canvas layouts.
:return: An isolated vector viewport container canvas.
"""
self.svg_widget = ZoomableSvgWidget()
"""Custom render context displaying loaded vector data files."""
self.svg_widget.renderer().setAspectRatioMode(Qt.AspectRatioMode.KeepAspectRatio)
return self.svg_widget
[docs]
def _assemble_layout(self, left: QWidget, center: QScrollArea) -> None:
"""Unify sub-panels inside the scalable horizontal splitter framework.
:param left: QWidget to place sub-panels inside.
:param center: QScrollArea to place sub-panels inside.
"""
self.main_splitter = QSplitter(Qt.Orientation.Horizontal)
"""Main splitter to dynamically divide the window."""
self.main_splitter.addWidget(left)
self.main_splitter.addWidget(center)
self.main_splitter.setStretchFactor(0, 1)
self.main_splitter.setStretchFactor(1, 3)
root_layout = QVBoxLayout(self)
root_layout.addWidget(self.main_splitter)
self.setLayout(root_layout) # Formally registers root_layout to this QWidget
[docs]
@staticmethod
def get_run_sim_default(name: str) -> str | int | float | None:
"""Get the default value of a function.
:param name: Name of the parameter.
:returns: Default value of the parameter.
:raises ValueError: If the parameter has no default value.
:raises KeyError: If the parameter does not exist.
"""
sig: inspect.Signature = inspect.signature(run_simulation)
param: inspect.Parameter = sig.parameters[name]
if param.default is inspect.Parameter.empty:
errmsg: str = f"{name} has no default"
raise ValueError(errmsg)
return cast("str | int | float | None", param.default)
[docs]
def run_simulation(self) -> None:
"""Run exactly one instance of the simulation engine."""
inputs = self._prepare_simulation_inputs()
self.input_metadata = inputs
# if inputs is None:
# errmsg = "Simulation input is empty."
# QMessageBox.critical(self, "Input Error", errmsg)
# return
self.run_group.setEnabled(False)
self.progress_bar.show()
self.progress_bar.setValue(0)
# self.run_group.setText("Computing...")
sim_input = RunSimulationInput(**{key: value for key, value in inputs.items() if key != "repeats"}) # type: ignore[typeddict-item]
task = BackgroundTask(run_simulation, **sim_input)
task.signals.finished.connect(self._on_simulation_complete)
task.signals.error.connect(self._on_simulation_error)
QThreadPool.globalInstance().start(task)
[docs]
def run_batch_simulation(self) -> None:
"""Run N parallel instances using Dask with safe child-spawned seeds."""
n_instances = self.repeat_count.value()
inputs = self._prepare_simulation_inputs()
self.input_metadata = inputs.copy()
self.input_metadata["repeats"] = n_instances
# if inputs is None:
# return
self.run_group.setEnabled(False)
self.progress_bar.show()
self.progress_bar.setValue(0)
def execute_dask_batch(
base_inputs: RunSimulationInput,
total_runs: int,
task_ref: BackgroundTask | None = None, # type: ignore[type-arg]
) -> list[RunResult]:
tasks: list[Delayed] = []
parent_seed: int = cast("int", base_inputs.get("seed"))
child_seeds = np.random.SeedSequence(parent_seed).spawn(total_runs)
def wrap_run_func(**kwargs: Unpack[RunSimulationInput]) -> RunResult:
output = run_simulation(**kwargs)[-1]
return output.coverage, output.fraction_of_covered_area, output.analyse_gap_size()
for seed in child_seeds:
run_inputs = base_inputs.copy()
run_inputs["seed"] = seed.generate_state(n_words=1, dtype=np.uint32)[0]
tasks.append(delayed(wrap_run_func)(**run_inputs))
workers = max(1, multiprocessing.cpu_count() - 1)
with Client(n_workers=workers, threads_per_worker=1, processes=True) as client:
futures: list[Future[RunResult]] = client.compute(tasks) # pyright: ignore[reportAssignmentType]
for idx, _ in enumerate(as_completed(futures), start=1):
if task_ref is not None:
percentage = int((idx / total_runs) * 100)
task_ref.signals.progress.emit(percentage)
results: tuple[RunResult] = client.gather(futures) # pyright: ignore[reportAssignmentType]
return list(results)
# Instantiate task and pass 'task' itself into the execution function so it can access signals
task = BackgroundTask(
execute_dask_batch,
base_inputs=inputs,
total_runs=n_instances,
) # typing: ignore[arg-type]
task.kwargs["task_ref"] = task # Dynamically inject the task reference into kwargs
if hasattr(self, "progress_bar"):
def set_bar(val: int) -> None:
self.progress_bar.setValue(val)
task.signals.progress.connect(
set_bar if Shiboken.isValid(self.progress_bar) else None,
)
task.signals.finished.connect(self._on_batch_simulation_complete)
task.signals.error.connect(self._on_simulation_error)
QThreadPool.globalInstance().start(task)
[docs]
def _on_simulation_complete(
self,
simulation_outputs: tuple[list[int], DistArray, int | Generator, tuple[IdxArray, ...], IdxArray, Simulator],
) -> None:
"""Run lightweight plotting pipeline back on the UI thread.
:returns: Simulation outputs.
"""
try:
self.progress_bar.setValue(100)
output = simulation_outputs[-1]
# Generate SVG elements in memory
svg_buffer = io.BytesIO()
dark_mode_bool: bool = QGuiApplication.styleHints().colorScheme() == Qt.ColorScheme.Dark
output.svgplot_covered_grid(filename=svg_buffer, dark_mode_bool=dark_mode_bool)
svg_data = svg_buffer.getvalue()
# Render UI updates directly
self.svg_widget.load(svg_data)
self.svg_widget.renderer().setAspectRatioMode(Qt.AspectRatioMode.KeepAspectRatio)
# Populate numeric output displays
self.state.coverages = (output.coverage,)
self.coverage_label.setText(f"Coverage: {np.sum(output.coverage):.4f}")
self.coverage_label.show()
self.state.fraction_of_covered_area = (output.fraction_of_covered_area,)
frac_of_covered_area = np.sum(output.fraction_of_covered_area)
self.state.gap_size_distribution = output.analyse_gap_size()
self.covered_area_label.setText(f"Fraction of covered area: {frac_of_covered_area:.4f}")
self.covered_area_label.show()
except (ValueError, TypeError, ValidationError) as e:
self.error(f"Error preparing visual plot data:\n{e}")
finally:
# Always unlock button when processing completes
self.run_group.setEnabled(True)
self.progress_bar.hide()
# self.run_group.setText("Run Simulation")
[docs]
def _on_batch_simulation_complete(self, batch_outputs: list[RunResult]) -> None:
"""Process multiple parallel output tuples sent back from the dask pool cluster.
:param batch_outputs: list of output values.
"""
try:
if not batch_outputs:
return
coverages: tuple[DistArray, ...]
fraction_of_cov_ar: tuple[DistArray, ...]
gapsize_dist: tuple[DistArray, ...]
coverages, fraction_of_cov_ar, gapsize_dist = zip(*batch_outputs, strict=True)
mpl.use("Agg")
fig = plt.figure(figsize=(8, 6))
gs = fig.add_gridspec(2, 2)
coverages_arr = np.array(coverages)
fraction_arr = np.array(fraction_of_cov_ar)
gapsize_distribution: DistArray = np.hstack(gapsize_dist)
missing_coverage = 1.0 - np.sum(coverages_arr, axis=1)
missing_fraction = 1.0 - np.sum(fraction_arr, axis=1)
coverages_final: FloatArray = np.column_stack((coverages_arr, missing_coverage)).tolist()
fraction_final: FloatArray = np.column_stack((fraction_arr, missing_fraction)).tolist()
cov = [np.mean(x) for x in zip(*coverages_final, strict=True)] # typing: ignore[explicit-any, assignment]
frac_of_cov_ar = [
np.mean(x) for x in zip(*fraction_final, strict=True)
] # typing: ignore[explicit-any, assignment]
self.coverage_label.setText(f"Coverage: {(1.0 - cov[-1]):.4f}")
self.coverage_label.show()
self.covered_area_label.setText(f"Fraction of covered area: {(1.0 - frac_of_cov_ar[-1]):.4f}")
self.covered_area_label.show()
self.export_results_button.show()
colors = [f"C{ii}" for ii in range(len(cov))]
colors[-1] = "none"
# Top left plot (Row 0, Column 0)
ax1 = fig.add_subplot(gs[0, 0])
ax1.set_title("Coverage")
ax1.pie(cov, colors=colors)
# Add outer circle to ax1
circle1 = patches.Circle((0, 0), 1, facecolor="none", edgecolor="black", linewidth=1.5)
ax1.add_patch(circle1)
# Top right plot (Row 0, Column 1)
ax2 = fig.add_subplot(gs[1, 0])
ax2.set_title("Frac. cov. area")
ax2.pie(frac_of_cov_ar, colors=colors)
# Add outer circle to ax2
circle2 = patches.Circle((0, 0), 1, facecolor="none", edgecolor="black", linewidth=1.5)
ax2.add_patch(circle2)
# Bottom double-length plot (Row 1, spans both Columns 0 and 1)
ax3 = fig.add_subplot(gs[:, 1])
ax3.set_title("Gap size distribution")
ax3.set_ylabel("Gap size (Å)")
sns.violinplot(gapsize_distribution, ax=ax3, color="0.8")
svg_buffer = io.BytesIO()
plt.savefig(svg_buffer, format="svg", bbox_inches="tight")
plt.close(fig) # Clear memory
self.state.coverages = tuple(col for col in coverages_arr.T)
self.state.fraction_of_covered_area = tuple(col for col in fraction_arr.T)
self.state.gap_size_distribution = gapsize_distribution
svg_data = svg_buffer.getvalue()
self.svg_widget.load(svg_data)
self.svg_widget.renderer().setAspectRatioMode(Qt.AspectRatioMode.KeepAspectRatio)
except (ValueError, TypeError, ValidationError) as e:
self.error(f"Error processing compiled batch metrics:\n{e}")
finally:
self.run_group.setEnabled(True)
self.progress_bar.hide()
# self.run_group.setText("Run Simulation")
[docs]
def _on_simulation_error(self, exception: Exception) -> None:
"""Fallback callback handling background core crashes safely.
:param exception: Exception raised during simulation.
"""
self.error(f"Simulation engine error:\n{exception}")
self.run_group.setEnabled(True)
self.progress_bar.hide()
# self.run_group.setText("Run Simulation")
[docs]
def export_results(self) -> None:
"""Export the simulation results to JSON, HDF5, Pickle, or zipped CSVs."""
if self.state.gap_size_distribution is None:
QMessageBox.warning(self, "Export Warning", "No valid simulation data found to export.")
return
file_filters = (
"Hierarchical Data Format (*.h5);;"
"JSON Data Interchange (*.json);;"
"Python Pickle Binary (*.pkl);;"
"Zipped Comma Separated Values (*.zip)"
)
chosen_path_str, selected_filter = QFileDialog.getSaveFileName(
self,
"Export Simulation Results As",
"",
file_filters,
)
if not chosen_path_str:
return # User cancelled out of the file dialogue
file_path = Path(chosen_path_str)
# Enforce file extension strings if skipped by the user
if not file_path.suffix:
if "json" in selected_filter:
ext = ".json"
elif "pkl" in selected_filter:
ext = ".pkl"
elif "zip" in selected_filter:
ext = ".zip"
elif "h5" in selected_filter:
ext = ".h5"
else:
errmsg: str = "Incorrect file extension."
QMessageBox.warning(self, "Error", errmsg)
return
file_path = file_path.with_suffix(ext)
covs = cast("tuple[DistArray]", self.state.coverages)
fracs = cast("tuple[DistArray]", self.state.fraction_of_covered_area)
gaps = cast("DistArray", self.state.gap_size_distribution)
suffix_lower = file_path.suffix.lower()
meta = self.input_metadata.copy()
if "molecules_list" in meta and meta["molecules_list"] is not None:
meta["molecules_list"] = [str(mol) for mol in meta["molecules_list"] if mol is not None] # pyright: ignore[reportGeneralTypeIssues]
try:
file_path.parent.mkdir(parents=True, exist_ok=True)
# -------------------------------------------------------------
# OPTION 1: HDF5 (.h5) -> Variable-length dataset trees
# -------------------------------------------------------------
if ".h5" in suffix_lower:
with h5py.File(str(file_path), "w") as f:
for key, val in meta.items():
f.attrs[key] = str(val)
grp_a = f.create_group("Coverage")
for idx, arr in enumerate(covs):
grp_a.create_dataset(f"col_{idx}", data=arr, compression="gzip")
grp_b = f.create_group("Fraction_of_covered_area")
for idx, arr in enumerate(fracs):
grp_b.create_dataset(f"col_{idx}", data=arr, compression="gzip")
f.create_dataset("Gap_size_distribution", data=gaps, compression="gzip")
# -------------------------------------------------------------
# OPTION 2: JSON (.json) -> Plaintext serialisation format
# -------------------------------------------------------------
elif ".json" in suffix_lower:
payload = {
"metadata": meta,
"Coverage": [arr.tolist() for arr in covs],
"Fraction_of_covered_area": [arr.tolist() for arr in fracs],
"Gap_size_distribution": gaps.tolist(),
}
with file_path.open("w", encoding="utf-8") as f:
json.dump(payload, f, indent=4)
# -------------------------------------------------------------
# OPTION 3: PICKLE (.pkl) -> High-speed memory state dump
# -------------------------------------------------------------
elif ".pkl" in suffix_lower:
payload = {
"metadata": meta,
"Coverage": covs,
"Fraction_of_covered_area": fracs,
"Gap_size_distribution": gaps,
}
with file_path.open("wb") as f:
pickle.dump(payload, f, protocol=pickle.HIGHEST_PROTOCOL)
# -------------------------------------------------------------
# OPTION 4: ZIPPED CSV (.csv.gz) -> Tabular with NaN padding
# -------------------------------------------------------------
elif ".zip" in suffix_lower:
# Open a compressed zip file archive stream wrapper directly
with zipfile.ZipFile(file_path, mode="w", compression=zipfile.ZIP_DEFLATED) as zf:
meta_io = io.StringIO()
for key, val in meta.items():
meta_io.write(f"{key}: {val}\n")
zf.writestr("metadata.txt", meta_io.getvalue())
coverage_dict = {f"col_{idx}": arr for idx, arr in enumerate(covs)}
coverage_io = io.StringIO()
pd.DataFrame(coverage_dict).to_csv(coverage_io, index=False)
zf.writestr("Coverage.csv", coverage_io.getvalue())
frac_cov_dict = {f"col_{idx}": arr for idx, arr in enumerate(fracs)}
frac_cov_io = io.StringIO()
pd.DataFrame(frac_cov_dict).to_csv(frac_cov_io, index=False)
zf.writestr("Fraction_of_covered_area.csv", frac_cov_io.getvalue())
gapsize_io = io.StringIO()
pd.DataFrame({"gap_size_distribution": gaps}).to_csv(gapsize_io, index=False)
zf.writestr("Gap_size_distribution.csv", gapsize_io.getvalue())
QMessageBox.information(self, "Success", f"Results successfully exported to:\n{file_path.name}")
except (OSError, ValueError, NotADirectoryError) as e:
QMessageBox.critical(self, "Export Failed", f"An error occurred while compiling your export:\n{e!s}")
[docs]
def error(self, msg: str) -> None:
"""Handle the errors.
:param msg: Error message to display in a new window.
"""
QMessageBox.critical(self, "Input Error", msg)
[docs]
class MoleculeGeneration(QWidget):
"""Molecule layout configuration dashboard tab view.
Handles dynamic generation of geometric molecule polygon shapes via reflective
library lookups, updates parameters on the fly, and lists them inside a tracking layout.
"""
[docs]
def __init__(self, state: AppState) -> None:
"""Initialise settings storage engines and compile separate view columns.
:param state: AppState instance for communication between tabs.
"""
super().__init__()
self.param_widgets: ParamWidgets = {}
""""Parameter widgets derived from molecule function signatures."""
self.opt_checkboxes: dict[str, QCheckBox] = {}
""""Optional checkbox widgets derived from molecule function signatures."""
self._settings = QSettings(type(self).__name__)
"""Persistent platform configuration handle cached between user runtime sessions."""
self.state = state
"""Shared application state cache container."""
# Initialise data storage metrics
self._init_data_storage()
# Build the three core panel containers
left_container = self._build_left_panel()
scroll_area = self._build_center_panel()
right_container = self._build_right_panel()
# Assemble components into the splitter layout interface
self._assemble_layout(left_container, scroll_area, right_container)
[docs]
def _init_data_storage(self) -> None:
"""Initialise internal state tracking arrays and counting iterations."""
self.mol_list_counter: count[int] = count()
"""Thread-safe sequential index iterator generating unique molecule instance identifier tags."""
self.mol_params_list: list[MoleculeParameters] = []
"""List of MoleculeParameters dataclasses."""
[docs]
def _build_left_panel(self) -> QWidget:
"""Construct the left parameters control dashboard and link active list triggers.
The left panel is populated by the parameters of a molecule selected from ``adsorpy.mol_lib``.
The molecules are kept in a dropdown menu based on a filtered list of the molecule functions.
Parameters are taken from type hints. Type hints determine whether a box is a spinbox, textbox, optional, etc.
:return: A populated structural container pane acting as the configuration panel.
"""
container = QWidget()
self.controls_layout = QVBoxLayout(container)
"""Layout frame coordinating selection toggles and molecule configuration property fields."""
self.func_dropdown = QComboBox()
"""Selection field populated with valid introspected molecule generator workflows."""
self.controls_layout.addWidget(QLabel("Select molecule"), alignment=Qt.AlignmentFlag.AlignTop)
self.controls_layout.addWidget(self.func_dropdown, alignment=Qt.AlignmentFlag.AlignTop)
# Discover generators using introspective library lookups
self.generators = cast("dict[str, Callable[P_mol, Polygon]]", self._discover_molecule_generators()) # pyright: ignore[reportGeneralTypeIssues]
"""Registry cache linking user-facing label text keys directly to underlying library callables."""
self.func_dropdown.addItems(list(self.generators.keys()))
self.func_dropdown.currentTextChanged.connect(self._update_func_dropdown)
# Parameter group layout setup
mol_param_group = QGroupBox("Parameters")
mol_param_layout = QVBoxLayout(mol_param_group)
mol_param_layout.addWidget(QLabel("Mouse over parameter for tooltip."), alignment=Qt.AlignmentFlag.AlignTop)
self.param_layout = QVBoxLayout()
"""Parameter layout."""
mol_param_layout.addLayout(self.param_layout)
self.controls_layout.addWidget(mol_param_group, alignment=Qt.AlignmentFlag.AlignTop)
target_index = self._fetch_setting("current_molecule", 0)
# Handle the index 0 edge case manually if it matches the default initial index
if target_index == 0:
# Force execution since setCurrentIndex(0) won't trigger a change event
self.build_param_inputs(self.func_dropdown.currentText())
self.plot_molecule()
else:
self.func_dropdown.setCurrentIndex(target_index)
self.output_label = QLabel("")
"""Status indicator updating real-time compilation info or syntax exceptions."""
self.controls_layout.addWidget(self.output_label)
return container
[docs]
def _update_func_dropdown(self, name: str) -> None:
"""Update when func dropdown changes.
:param name: The name of the dropdown option.
"""
self.build_param_inputs(name)
self.plot_molecule()
[docs]
@staticmethod
def _discover_molecule_generators() -> dict[str, Callable[P_mol, Polygon]]:
"""Isolate reflection logic filtering usable library structural definitions.
:return: A sorted lookup dict mapping valid function names to execution references.
"""
temp_generators: dict[str, Callable[P_mol, Polygon]] = {
name: func
for name, func in molecule_lib.__dict__.items()
if inspect.isfunction(func)
and not name.startswith("_")
and func.__module__ == molecule_lib.__name__
and inspect.signature(func).return_annotation in {"Polygon", "dict[str, str | float | list[str] | None]"}
}
return dict(sorted(temp_generators.items()))
[docs]
def _build_center_panel(self) -> QScrollArea:
"""Construct the viewport frame area housing the centered vector graphics.
:return: A scroll area wrapper managing the interactive central viewport.
"""
scroll_area = QScrollArea()
scroll_area.setWidgetResizable(True)
scroll_container = QWidget()
container_layout = QGridLayout(scroll_container)
container_layout.setContentsMargins(0, 0, 0, 0)
self.svg_widget = ZoomableSvgWidget()
"""Custom structural viewport rendering vector polygon outlines."""
self.svg_widget.setMinimumSize(600, 600)
self.svg_widget.renderer().setAspectRatioMode(Qt.AspectRatioMode.KeepAspectRatio)
container_layout.addWidget(self.svg_widget, 0, 0, Qt.AlignmentFlag.AlignCenter)
scroll_area.setWidget(scroll_container)
return scroll_area
[docs]
def _build_right_panel(self) -> QWidget:
"""Construct the right tracking grid columns managing existing records.
:return: A secondary control panel listing items added to the current system context.
"""
container = QWidget()
right_col = QVBoxLayout(container)
group = QGroupBox("Molecules")
group_layout = QVBoxLayout(group)
self.molecule_list_widget = ReorderableListWidget()
"""List widget selection tool indicating current added molecule configurations."""
self.molecule_list_widget.currentItemChanged.connect(self.show_molecule_settings)
self.molecule_list_widget.itemsMoved.connect(self.sync_list_order)
group_layout.addWidget(self.molecule_list_widget)
self.delete_btn = QPushButton("Delete Selected Molecule")
"""Delete selected molecule from list of molecules."""
self.delete_btn.clicked.connect(self.delete_molecule)
self.delete_btn.setToolTip("Delete selected molecule.")
group_layout.addWidget(self.delete_btn)
right_col.addWidget(group)
return container
[docs]
def _assemble_layout(self, left: QWidget, center: QScrollArea, right: QWidget) -> None:
"""Unify sub-panels inside the scalable horizontal splitter framework.
:param left: Parameter selection pane widget.
:param center: Scroll pane holding vector outputs.
:param right: Management column listing generated arrays.
"""
self.main_splitter = QSplitter(Qt.Orientation.Horizontal)
"""Scalable divider framework managing responsive interface margins."""
self.main_splitter.addWidget(left)
self.main_splitter.addWidget(center)
self.main_splitter.addWidget(right)
self.main_splitter.setStretchFactor(0, 1)
self.main_splitter.setStretchFactor(1, 3)
self.main_splitter.setStretchFactor(2, 1)
root_layout = QVBoxLayout(self)
root_layout.addWidget(self.main_splitter)
[docs]
def _fetch_setting(self, name: str, default: T_inv, return_type: type[T_inv] | None = None) -> T_inv:
"""Fetch settings by checking if they exist followed by their value.
:param name: The name of the setting to fetch.
:param default: The default value to return if the setting does not exist.
:param return_type: The default return type if the setting exists. If not given, type(default) is used.
:returns: The setting value if it exists, or else the default.
"""
check_type: type[T_inv] = type(default) if return_type is None else return_type
return cast("T_inv", self._settings.value(name, defaultValue=default, type=check_type))
[docs]
def _delete_previous_layout(self) -> None:
"""Recursively delete the layout of the previous molecule parameters."""
def clear_layout(layout: QLayout | None) -> None:
"""Clear the layout by deleting its widgets or traversing its child layouts.
:param layout: The layout to clear.
"""
if layout is None:
return
widget: QWidget | None
child_layout: QLayout | None
while layout.count():
item = cast("QLayoutItem", layout.takeAt(0))
# Use structural pattern matching to safely handle the item type
match item.widget(), item.layout():
case (widget, _) if widget is not None:
# It is a widget container item
widget.deleteLater()
case (_, child_layout) if child_layout is not None:
# It is a nested layout item; recurse down first, then delete it
clear_layout(child_layout)
child_layout.deleteLater()
case _:
# It is a spacer item or an empty container
del item
clear_layout(self.param_layout)
[docs]
def sync_list_order(self, old_index: int, new_index: int) -> None:
"""Take the row transformation from list A and applies it programmatically to list B.
The ReorderableListWidget allows for items to be drag/dropped. This function links the reordering.
:param old_index: The original index of the item changing position.
:param new_index: The new index to which the item is moved.
"""
# Pop the item out of its old position in List B
taken_item = self.mol_params_list.pop(old_index)
# Insert it into the exact same new position
if taken_item:
self.mol_params_list.insert(new_index, taken_item)
self.state.molecule_param_list = self.mol_params_list
[docs]
def _build_symmetry_controls(self) -> None:
"""Assemble the geometric shape matrix transformation property grid layouts."""
self.refl_sym_default: bool = False
"""Default value of reflection symmetry."""
self.rot_sym_default: int = 1
"""Default value of rotation symmetry."""
self.rot_cnt_default: int = 360
"""Default value of rotation count."""
symmetry_options_layout = QGridLayout()
refl_sym_label = QLabel("Reflection symmetry")
refl_sym_tooltip_text = "Set checked if the molecule has reflection symmetry (symmetric group Cn → Dn)."
refl_sym_label.setToolTip(refl_sym_tooltip_text)
self.refl_sym = QCheckBox()
"""Reflection symmetry checkbox, corresponding to True (checked) or False (unchecked)."""
self.refl_sym.setChecked(self.refl_sym_default)
self.refl_sym.setToolTip(refl_sym_tooltip_text)
self.rot_sym_label = QLabel("Rotation symmetry")
"""Rotation symmetry label."""
self.rot_sym = QSpinBox()
"""Rotation symmetry spinbox, for non-negative integers."""
self.rot_sym.setMinimum(0)
self.rot_sym.setValue(self.rot_sym_default)
self._update_symmetry_tooltip(self.refl_sym_default)
self.refl_sym.toggled.connect(self._update_symmetry_tooltip)
rot_cnt_label = QLabel("Rotation count")
rot_cnt_tooltip_text = "Number of rotations to be used for the molecule. The step size is 360/n°."
rot_cnt_label.setToolTip(rot_cnt_tooltip_text)
self.rot_cnt = QSpinBox()
"""Rotation count spinbox, for positive (non-zero) integers."""
self.rot_cnt.setMinimum(1)
self.rot_cnt.setMaximum(99999)
self.rot_cnt.setValue(self.rot_cnt_default)
self.rot_cnt.setToolTip(rot_cnt_tooltip_text)
self.symmetry_widgets = SymmetryWidgets(rot_sym=self.rot_sym, refl_sym=self.refl_sym, rot_cnt=self.rot_cnt)
symmetry_options_layout.addWidget(refl_sym_label, 0, 0)
symmetry_options_layout.addWidget(self.refl_sym, 0, 1)
symmetry_options_layout.addWidget(self.rot_sym_label, 1, 0)
symmetry_options_layout.addWidget(self.rot_sym, 1, 1)
symmetry_options_layout.addWidget(rot_cnt_label, 2, 0)
symmetry_options_layout.addWidget(self.rot_cnt, 2, 1)
self.param_layout.addLayout(symmetry_options_layout)
[docs]
def launch_first_time_loader(self) -> None:
"""Launch the first time loader from molecule_lib.
If no file path has been provided, prompt the user to add one before running the first time loader.
"""
if "file_name" not in self.param_widgets:
errmsg = "Parameter file_name not found in widget."
QMessageBox.critical(self, "Key Error", errmsg)
return
if not self.param_widgets["file_name"].text():
self.param_widgets["file_name"].browse_button.click()
output = molecule_lib.first_time_loader(Path(self.param_widgets["file_name"].text()))
first_time_key: ParamName
first_time_value: str | float | list[str] | None
for first_time_key, first_time_value in output.items(): # pyright: ignore[reportAssignmentType]
if not (is_valid_param(first_time_key) or first_time_key in self.param_widgets):
errmsg = f"Not a valid key: {first_time_key}"
QMessageBox.critical(self, "Key Error", errmsg)
return
if first_time_value is not None:
set_content(self.param_widgets[first_time_key], first_time_value) # pyright: ignore[reportTypedDictNotRequiredAccess]
if first_time_key in self.opt_checkboxes:
self.opt_checkboxes[first_time_key].setChecked(True)
[docs]
def get_param_values(self) -> dict[str, float | int | str]:
"""Extract current user inputs from widgets back into a data dictionary.
:return: Dictionary containing the key-value pairs of the parameters.
"""
values: dict[str, float | int | str] = {}
for name, widget in cast("ItemsView[str, InputWidget]", self.param_widgets.items()):
# If the widget is disabled, the optional checkbox was unchecked -> value is None
if not widget.isEnabled():
continue
# Extract value based on the PySide6/PyQt6 widget type
# Ignore the error because this match is exhaustive!
match widget: # type: ignore[exhaustive-match]
case QSpinBox() | QDoubleSpinBox():
values[name] = widget.value()
case QLineEdit() | FilePickerWidget():
values[name] = widget.text()
return values
[docs]
def error(self, msg: str) -> None:
"""Handle the errors.
Please open a ticket if this happens when it should not.
:param msg: Error message.
"""
QMessageBox.critical(self, "Input Error", msg)
[docs]
def plot_molecule(self) -> None:
"""Plot the molecule."""
if not self.show_molecule_checkbox.isChecked():
return
molecule_func = self.generators[self.func_dropdown.currentText()]
molecule_dict = self.get_param_values()
if molecule_func.__name__ == "first_time_loader":
molecule_func = cast("Callable[P_mol, Polygon]", molecule_lib.xyz_reader) # pyright: ignore[reportGeneralTypeIssues]
try:
svg_io = io.BytesIO()
molecule_lib.save_molecule_svg(molecule_func(**molecule_dict), filename=svg_io) # pyright: ignore[reportCallIssue]
svg_data = svg_io.getvalue()
self.svg_widget.load(svg_data)
except ValueError as e:
self.error(str(e))
[docs]
def add_molecule(self) -> None:
"""Add a molecule to the list of molecules to use."""
current_func_name = self.func_dropdown.currentText()
current_func_name = current_func_name if current_func_name != "first_time_loader" else "xyz_reader"
molecule_func = self.generators[current_func_name]
molecule_dict = self.get_param_values()
try:
result = molecule_func(**molecule_dict) # pyright: ignore[reportCallIssue]
except ValidationError as e:
self.error(str(e))
return
name = (
self.func_dropdown.currentText()
if "file_name" not in molecule_dict
else Path(cast("str", molecule_dict["file_name"])).name
)
# Update dropdown
index = next(self.mol_list_counter)
label = f"{name} #{index}"
self.molecule_list_widget.addItem(label)
mol_params = MoleculeParameters(
index=index,
function_name=current_func_name,
label=label,
polygon=PydanticPolygon(result),
settings=molecule_dict,
refl_sym=self.refl_sym.isChecked(),
rot_sym=self.rot_sym.value(),
rot_cnt=self.rot_cnt.value(),
)
self.mol_params_list.append(mol_params)
self.state.molecule_param_list = self.mol_params_list
self.output_label.setText(f"Added: {name}")
[docs]
def delete_molecule(self) -> None:
"""Delete the current selected molecule."""
idx = self.molecule_list_widget.currentRow()
if idx < 0: # Hitting the delete button without a selection results in -1.
return
del self.mol_params_list[idx]
self.molecule_list_widget.takeItem(idx)
self.state.molecule_param_list = self.mol_params_list
self.output_label.setText("Molecule deleted")
self.molecule_list_widget.clearSelection()
self.molecule_list_widget.setCurrentItem(QListWidgetItem(None))
[docs]
def show_molecule_settings(self) -> None:
"""Show the settings of this molecule."""
current_idx: int = self.molecule_list_widget.currentRow()
if current_idx < 0 or current_idx >= len(self.mol_params_list):
return
current_name: str = self.mol_params_list[current_idx]["function_name"]
match_idx: int = self.func_dropdown.findText(current_name, Qt.MatchFlag.MatchExactly)
self.func_dropdown.setCurrentIndex(match_idx)
for key, val in self.mol_params_list[current_idx]["settings"].items():
if not is_valid_param(key) or key not in self.param_widgets:
errmsg: str = f"Key does not exist: {key}"
raise KeyError(errmsg)
current_param: InputWidget = self.param_widgets[key] # pyright: ignore[reportTypedDictNotRequiredAccess]
set_content(current_param, val)
if key in self.opt_checkboxes:
self.opt_checkboxes[key].setChecked(True)
for symmetry_name, symmetry_widget in self.symmetry_widgets.items():
sym_val: int | bool = self.mol_params_list[current_idx][
cast("Literal['rot_sym', 'refl_sym', 'rot_cnt']", symmetry_name)
]
if isinstance(sym_val, bool): # Must check bool, not int, because bool is a subtype of int for Python.
cast("QCheckBox", symmetry_widget).setChecked(sym_val)
else:
cast("QSpinBox", symmetry_widget).setValue(sym_val)
self.show_molecule_checkbox.setChecked(True)
[docs]
class SurfaceGeneration(QWidget):
"""Surface generation dashboard tab view.
Provides control inputs for generating geometric lattice surfaces and
displays the resulting surface within an interactive, centered viewer.
"""
[docs]
def __init__(self, state: AppState) -> None:
"""Initialise user widgets and assemble geometric layout wrappers.
:param state: AppState object to share information between tabs.
"""
super().__init__()
self.state = state
"""Shared application state cache container."""
self.surface_count: int = 50
"""Default surface site count."""
self.real_surface_count: int = 50
"""Default computed surface site count."""
self.stored_params: SurfaceParameters | None = None
"""Parameters of the surface, to be communicated between tabs."""
self._init_validators()
# Initialise panels
left_container = self._build_left_panel()
scroll_area = self._init_svg_view()
# Assemble components directly inside the splitter framework
self.main_splitter = QSplitter(Qt.Orientation.Horizontal)
"""Main splitter of the window."""
self.main_splitter.addWidget(left_container)
self.main_splitter.addWidget(scroll_area)
# Configure layout stretching rules (1 unit left panel, 3 units centre viewport)
self.main_splitter.setStretchFactor(0, 1)
self.main_splitter.setStretchFactor(1, 3)
# Clean up root assignment
root_layout = QHBoxLayout(self)
root_layout.addWidget(self.main_splitter)
[docs]
def _init_validators(self) -> None:
"""Instantiate validation models for text constraint processing."""
self._gt_one_validator = QIntValidator()
"""Validator to ensure an int >= 1"""
self._gt_one_validator.setBottom(1)
self._pos_float_validator = QDoubleValidator()
"""Validator to ensure a float >= 0.0"""
self._pos_float_validator.setBottom(0.0)
[docs]
def _build_left_panel(self) -> QWidget:
"""Construct the left controls container pane layout.
:return: A populated structural layout container.
"""
container = QWidget()
layout = QVBoxLayout(container)
# Surface configuration selector.
layout.addWidget(QLabel("Surface Type:"), alignment=Qt.AlignmentFlag.AlignTop)
self.surface_dropdown = QComboBox()
"""Selection box for geometry presets."""
self.surface_dropdown.addItems(sorted(["hexagonal", "square", "honeycomb"]))
self.surface_dropdown.setToolTip("Select surface type.")
layout.addWidget(self.surface_dropdown, alignment=Qt.AlignmentFlag.AlignTop)
# Theoretical count constraints input.
layout.addWidget(QLabel("Optional surface site count (positive int):"), alignment=Qt.AlignmentFlag.AlignTop)
self.site_count_input = QLineEdit()
"""Numeric entry field for requested surface site count."""
self.site_count_input.setValidator(self._gt_one_validator)
self.site_count_input.setPlaceholderText("e.g. 42")
layout.addWidget(self.site_count_input, alignment=Qt.AlignmentFlag.AlignTop)
# Evaluated real layout node calculation trackers.
layout.addWidget(QLabel("Real surface site count:"), alignment=Qt.AlignmentFlag.AlignTop)
self.real_site_count = QLabel()
"""Text label reflecting processed actual surface site count."""
self.real_site_count.setText("50")
self.real_site_count.setToolTip("The actual site count computed from the input count and the surface type.")
layout.addWidget(self.real_site_count, alignment=Qt.AlignmentFlag.AlignTop)
# Physical spacing distance parameters.
layout.addWidget(QLabel("Lattice Spacing (optional, > 0 float):"), alignment=Qt.AlignmentFlag.AlignTop)
self.lattice_input = QDoubleSpinBox()
"""Numeric entry field for lattice spacing."""
self.lattice_input.setMinimum(0.0)
self.lattice_input.setValue(1.0)
self.lattice_input.setDecimals(2)
self.lattice_input.setSingleStep(0.01)
self.lattice_input.setAccelerated(True)
self.lattice_input.setSuffix(" Å")
self.lattice_input.setToolTip("Numeric entry field for lattice spacing.")
layout.addWidget(self.lattice_input, alignment=Qt.AlignmentFlag.AlignTop)
# Control trigger processing elements.
self.generate_surface_button = QPushButton("Generate Surface")
"""Trigger execution pipeline for layout generation code."""
self.generate_surface_button.setToolTip("Plot and store the surface.")
layout.addWidget(self.generate_surface_button, alignment=Qt.AlignmentFlag.AlignTop)
# Establish signalling loops.
self.site_count_input.textChanged.connect(self._get_real_surface_site_count)
self.surface_dropdown.currentIndexChanged.connect(self._get_real_surface_site_count)
self.generate_surface_button.clicked.connect(self.generate_surface)
# Force components to stick tight to the top boundary layout.
layout.addStretch()
return container
[docs]
def _init_svg_view(self) -> QScrollArea:
"""Construct the graphics frame and isolate structural canvas centering.
:return: A scroll container managing the viewport window frame.
"""
self.svg_widget = ZoomableSvgWidget()
"""Custom render context displaying loaded vector data."""
self.svg_widget.renderer().setAspectRatioMode(Qt.AspectRatioMode.KeepAspectRatio)
# Create a container widget with a Grid Layout to centre the SVG
container = QWidget()
container_layout = QGridLayout(container)
container_layout.addWidget(self.svg_widget, 0, 0, Qt.AlignmentFlag.AlignCenter)
self.scroll_area = QScrollArea()
"""Interactive bounding box containing the centered layout viewport."""
self.scroll_area.setWidgetResizable(True)
self.scroll_area.setWidget(container)
return self.scroll_area
[docs]
def _get_real_surface_site_count(self) -> None:
"""Get the real surface site count."""
default_count: int = 50
surface_type: str = self.surface_dropdown.currentText()
temp_count: str = self.site_count_input.text().strip()
self.surface_count = default_count if not temp_count else int(temp_count)
self.real_surface_count = self.surface_count * self.surface_count
if surface_type == "hexagonal":
self.real_surface_count *= 2
elif surface_type == "honeycomb":
self.real_surface_count *= 4
self.real_site_count.setText(str(self.real_surface_count))
[docs]
def generate_surface(self) -> None:
"""Generate an example surface."""
seed_text = self.state.seed_input.text().strip()
seed: int | None = None
if seed_text:
if not seed_text.isnumeric() or int(seed_text) < 0:
self.error("Seed must be a positive integer")
return
seed = int(seed_text)
# Validate lattice spacing
lattice_text = self.lattice_input.value()
lattice: float | None = None
if lattice_text:
try:
lattice = float(lattice_text)
except ValueError as e:
self.error(str(e))
return
lattice_type = cast(
"Literal['hexagonal', 'triangular', 'honeycomb', 'square']",
self.surface_dropdown.currentText(),
)
app = cast("QGuiApplication", QGuiApplication.instance())
dark_mode_bool = app.styleHints().colorScheme() == Qt.ColorScheme.Dark
svg_buffer = io.BytesIO()
surf_params: SurfaceParameters = {
"lattice_a": lattice,
"lattice_type": lattice_type,
"seed": seed,
"site_count": self.surface_count,
}
show_surface(
**surf_params,
filepath=svg_buffer,
svg_flag=True,
dark_mode_bool=dark_mode_bool,
)
svg_data = svg_buffer.getvalue()
self.svg_widget.load(svg_data)
self.svg_widget.renderer().setAspectRatioMode(Qt.AspectRatioMode.KeepAspectRatio)
self.stored_params = SurfaceParameters(**surf_params)
self.state.surface_params = self.stored_params
[docs]
def error(self, msg: str) -> None:
"""Handle the errors.
:param msg: Error message to display in a new window.
"""
QMessageBox.critical(self, "Input Error", msg)
[docs]
class BackgroundTaskSignals(QObject):
"""Signals for the generic background worker.
:cvar finished: Emits the raw output data package.
:cvar progress: Emits the current simulation progress as a percentage integer.
:cvar progress: Emits the integer percentage (0 to 100).
"""
finished = Signal(object)
error = Signal(Exception)
progress = Signal(int)
[docs]
class BackgroundTask(QRunnable, Generic[P, R]):
"""Executes a single blocking function call in the background thread pool."""
[docs]
def __init__(
self,
func: Callable[P, R],
*args: P.args,
**kwargs: P.kwargs,
) -> None:
"""Initialise the BackgroundTask.
:param func: Function to be executed.
:param args: Positional arguments to be passed to the function.
:param kwargs: Keyword arguments to be passed to the function.
"""
super().__init__()
self.func: Callable[P, R] = func
self.args = args
self.kwargs = kwargs
self.signals = BackgroundTaskSignals()
[docs]
@override
def run(self) -> None:
"""Run the background task."""
try:
result = self.func(*self.args, **self.kwargs)
self.signals.finished.emit(result)
except (ValueError, TypeError, OSError) as e:
self.signals.error.emit(e)
[docs]
def _make_horizontal_line() -> QFrame:
"""Create a horizontal line widget using a QFrame object.
:returns: A horizontal line widget.
"""
hline = QFrame()
hline.setFrameShape(QFrame.Shape.HLine)
hline.setFrameShadow(QFrame.Shadow.Sunken)
return hline
[docs]
def main() -> int:
"""Launch the adsorpy GUI.
:returns: Return code.
"""
app = QApplication(sys.argv)
app.setOrganizationName("adsorpy")
app.setApplicationName("adsorpy-app")
gui = AdsorpyGUI()
gui.resize(1600, 900)
gui.show()
return cast("int", app.exec())
if __name__ == "__main__":
sys.exit(main())