Source code for spacr.qt.screens.parameter_sweep

"""Interactive regression parameter sweeps.

The screen runs a regression under multiple defensible settings combinations
and compares the resulting hits and control ranks. Each trial writes to its
own folder, and the results table is updated as trials finish so interrupted
sweeps can be resumed.

Worker recommendations account for available memory and trials run at low
CPU and I/O priority. On systems that support ``systemd-run --user``, each
trial also receives kernel-enforced resource limits. The Containment row
states whether those limits are active; memory-based scheduling alone cannot
guarantee containment.
"""
from __future__ import annotations

import os
from ..widgets.sortable_table import install_sorting, table_item

APP_KEY = "parameter_sweep"
APP_NAME = "Parameter Sweep"
APP_DESCRIPTION = (
    "Run the regression under many settings combinations and compare what "
    "each one concludes")
APP_INTRO = (
    "Point this at the same score and count CSVs the regression uses, tick "
    "the settings you want varied, and start. Each trial runs in its own "
    "folder and the table fills in as they finish: model family, correction, "
    "analysis unit, how many hits were called, and the rank of your positive "
    "control in each. The spread across settings is the result — a screen "
    "whose hit count swings from 2 to 400 depending on the correction has "
    "not been analysed, it has been chosen. Workers are sized from available "
    "memory and run at low priority. The Containment row states whether "
    "kernel-enforced per-trial limits are active on this machine.")
APP_TRANSLATIONS = (
    "Parametersvep", "Parameter-Sweep", "Barrido de parámetros",
    "参数扫描", "Varredura de parâmetros", "पैरामीटर स्वीप",
    "매개변수 스윕", "Færibreytusveip", "Balayage de paramètres")
#: Why there is no ``spacr-run parameter_sweep``. Reaches
#: :data:`spacr.cli.INTERACTIVE_ONLY`, which is what the CLI prints instead of
#: "unknown module".
#:
#: The SWEEP is fully headless -- :mod:`spacr.parameter_sweep` is the engine
#: this screen drives, and a cluster is where a few hundred trials belong. It
#: is the WORKBENCH that cannot be batched: the table, and double-clicking a
#: row to get that exact regression back with its figures. So the note points
#: at the engine rather than claiming the feature has no headless path.
APP_CLI_NOTE = (
    "Parameter Sweep is the interactive workbench for a sweep -- the table of "
    "trials, and double-clicking a row to get that exact regression back with "
    "its figures. The sweep itself is headless: from spacr.parameter_sweep "
    "import run_sweep, summarise_sweep; run_sweep(base_settings, destination) "
    "writes the same results table this screen reads, and summarise_sweep / "
    "rank_trials give the comparison it draws.")

__all__ = ["APP_KEY", "APP_NAME", "APP_DESCRIPTION", "APP_INTRO",
           "APP_CLI_NOTE", "build_parameter_sweep_card", "sweepable"]


def _make_screen(app_key=None, host=None):
    """Build the screen lazily -- it pulls in the sweep engine and pandas."""
    from PySide6.QtCore import Qt
    from PySide6.QtWidgets import (
        QCheckBox, QComboBox, QFormLayout, QGroupBox, QHBoxLayout, QLabel,
        QLineEdit, QMessageBox, QProgressBar, QPushButton, QScrollArea,
        QSpinBox, QTableWidget, QTableWidgetItem, QVBoxLayout,
        QWidget,
    )
    from ..i18n import set_translatable_text
    from ..widgets.collapsible_splitter import EDGE, CollapsibleSplitter

    from ..job_runner import JobRunner
    from ..widgets.file_list import FilePathListWidget
    from ...parameter_sweep import (
        DEFAULT_SWEEP_SPACE, SweepSpace, _recommended_worker_budget,
        build_trials,
    )

    class ParameterSweepScreen(QWidget):
        """The parameter sweep, as the registry builds it.

        :param host: the main window, passed by the registry for navigation.
            NOT a Qt parent -- the registry's host is not always a QWidget,
            so it is kept as an attribute and never handed to
            ``QWidget.__init__``.
        """

        def __init__(self, host=None):
            """Build the sweep screen. ``host`` is the window, not a Qt parent."""
            super().__init__()
            self.host = host
            self._results = None
            self._runner = JobRunner(self, app_key=APP_KEY)

            outer = QVBoxLayout(self)
            splitter = CollapsibleSplitter(
                Qt.Horizontal, self, persist_key=f"{APP_KEY}::body")
            self._body = splitter
            outer.addWidget(splitter)

            left = QScrollArea(self)
            left.setWidgetResizable(True)
            left.setMinimumWidth(420)
            panel = QWidget()
            form = QVBoxLayout(panel)

            inputs = QGroupBox("Inputs", panel)
            inputs_form = QFormLayout(inputs)
            self.score_data = FilePathListWidget(
                kind="table", title="Choose per-object score CSVs")
            self.count_data = FilePathListWidget(
                kind="table", title="Choose gRNA count CSVs (one per plate)")
            self.destination = QLineEdit(panel)
            self.destination.setPlaceholderText(
                "Folder for the trial folders and the results table")
            self.dependent_variable = QLineEdit("pred", panel)
            inputs_form.addRow("Score CSVs", self.score_data)
            inputs_form.addRow("Count CSVs", self.count_data)
            inputs_form.addRow("Response column", self.dependent_variable)
            inputs_form.addRow("Output folder", self.destination)
            form.addWidget(inputs)

            axes = QGroupBox("Settings to sweep", panel)
            axes_layout = QVBoxLayout(axes)
            axes_layout.addWidget(QLabel(
                "Tick a setting to vary it. Unticked settings keep the value "
                "on the right, applied to every trial.", panel))
            self._axis_rows = {}
            grid = QFormLayout()
            for key, values in DEFAULT_SWEEP_SPACE.items():
                row = QWidget(panel)
                row_layout = QHBoxLayout(row)
                row_layout.setContentsMargins(0, 0, 0, 0)
                include = QCheckBox(row)
                include.setChecked(key not in (
                    "fraction_threshold", "min_cells_per_well", "fdr_alpha",
                    "min_observations_per_hit", "outlier_detection", "threshold_method"))
                editor = QLineEdit(
                    ", ".join("None" if v is None else str(v) for v in values),
                    row)
                editor.setToolTip(
                    "Enter comma-separated values for this setting. If its "
                    "checkbox is clear, the first value is used for every "
                    "trial.")
                row_layout.addWidget(include)
                row_layout.addWidget(editor, 1)
                grid.addRow(key, row)
                self._axis_rows[key] = (include, editor)
            axes_layout.addLayout(grid)
            form.addWidget(axes)

            budget = QGroupBox("Budget", panel)
            budget_form = QFormLayout(budget)
            self.max_trials = QSpinBox(panel)
            self.max_trials.setRange(1, 100000)
            self.max_trials.setValue(500)
            self.mode = QComboBox(panel)
            self.mode.addItems(["random", "grid"])
            self.mode.setToolTip(
                "random samples combinations uniformly without replacement; "
                "grid walks the Cartesian product in order and truncates at "
                "the trial limit.")
            self.seed = QSpinBox(panel)
            self.seed.setRange(0, 2 ** 31 - 1)
            self.seed.setValue(20260815)
            self.workers = QSpinBox(panel)
            self.workers.setRange(1, 32)
            worker_budget = _recommended_worker_budget()
            suggested = int(worker_budget["workers"])
            self.workers.setValue(suggested)
            self.workers.setToolTip(
                "Requested worker count. The sweep clamps it according to "
                "available memory; see Containment below for whether "
                "kernel-enforced per-trial limits are active.")
            self.worker_note = QLabel("", panel)
            self.worker_note.setWordWrap(True)
            self._set_worker_note(worker_budget)
            from ...parameter_sweep import (
                TRIAL_CPU_QUOTA, TRIAL_MEMORY_MAX, containment_available,
            )

            self.containment = QLabel("", panel)
            if containment_available():
                set_translatable_text(
                    self.containment,
                    "Kernel containment is active for each trial: memory "
                    "{memory}, swap disabled, and CPU quota {cpu}. If a trial "
                    "exceeds a limit, only that trial is stopped and recorded "
                    "as 'killed'; the sweep continues.",
                    memory=TRIAL_MEMORY_MAX,
                    cpu=TRIAL_CPU_QUOTA,
                )
            else:
                set_translatable_text(
                    self.containment,
                    "Kernel containment is unavailable because systemd-run "
                    "--user --scope could not be started. This is common in "
                    "containers and SSH sessions without a user manager. "
                    "Thread limits and the free-memory check still apply, "
                    "but they cannot prevent a single trial from exhausting "
                    "system memory. Reduce the worker count or run the sweep "
                    "from a systemd user session before using a large search "
                    "space.",
                )
            self.containment.setWordWrap(True)
            if not containment_available():
                self.containment.setObjectName("DangerLabel")
                try:
                    from ..theme import active_palette
                    self.containment.setStyleSheet(
                        f"color: {active_palette()['error']};")
                except Exception:                        # noqa: BLE001
                    pass
            budget_form.addRow("Maximum trials", self.max_trials)
            budget_form.addRow("Sampling", self.mode)
            budget_form.addRow("Seed", self.seed)
            budget_form.addRow("Workers", self.workers)
            budget_form.addRow("", self.worker_note)
            budget_form.addRow("Containment", self.containment)
            form.addWidget(budget)

            buttons = QHBoxLayout()
            self.estimate_button = QPushButton("Estimate", panel)
            self.estimate_button.setToolTip(
                "Count the valid parameter combinations and estimate runtime "
                "without starting a trial.")
            self.estimate_button.clicked.connect(self.estimate)
            self.start_button = QPushButton("Start sweep", panel)
            self.start_button.clicked.connect(self.start)
            buttons.addWidget(self.estimate_button)
            buttons.addWidget(self.start_button)
            buttons.addStretch(1)
            form.addLayout(buttons)
            form.addStretch(1)
            left.setWidget(panel)
            splitter.add_pane(left, "Sweep settings", mode=EDGE, stretch=1,
                              fold_key=f"{APP_KEY}/Sweep settings")

            right = QWidget(self)
            right_layout = QVBoxLayout(right)
            self.status = QLabel("Nothing running.", right)
            self.status.setWordWrap(True)
            right_layout.addWidget(self.status)
            self.progress = QProgressBar(right)
            self.progress.setTextVisible(False)
            self.progress.setVisible(False)
            right_layout.addWidget(self.progress)
            output = CollapsibleSplitter(
                Qt.Vertical, right, persist_key=f"{APP_KEY}::results")
            self._output = output
            trials = QWidget(right)
            trials_layout = QVBoxLayout(trials)
            trials_layout.setContentsMargins(0, 0, 0, 0)
            self.table = QTableWidget(0, 0, trials)
            install_sorting(self.table)
            self.table.setSelectionBehavior(QTableWidget.SelectRows)
            self.table.setEditTriggers(QTableWidget.NoEditTriggers)
            self.table.setSortingEnabled(True)
            self.table.doubleClicked.connect(self._on_row_activated)
            self.table.setToolTip(
                "Double-click a row to re-run that trial and draw its "
                "figures below. They are live figures: right-click one to "
                "restyle it.")
            trials_layout.addWidget(self.table, 1)

            row_buttons = QHBoxLayout()
            self.refresh_button = QPushButton("Refresh results", trials)
            self.refresh_button.clicked.connect(self.load_results)
            row_buttons.addWidget(self.refresh_button)
            self.show_button = QPushButton("Show figures for selected row",
                                           trials)
            self.show_button.clicked.connect(self._on_row_activated)
            self.show_button.setEnabled(False)
            row_buttons.addWidget(self.show_button, 1)
            trials_layout.addLayout(row_buttons)
            self.table.itemSelectionChanged.connect(
                lambda: self.show_button.setEnabled(
                    self.table.currentRow() >= 0))

            self.trial_status = QLabel("", trials)
            self.trial_status.setWordWrap(True)
            trials_layout.addWidget(self.trial_status)
            self.trials_section = output.add_section(
                trials, "Trials", persist_key=f"{APP_KEY}/Trials",
                stretch=1)

            from ..widgets.regression_results import RegressionResultsPanel
            self.results = RegressionResultsPanel(right)
            self.results.setMinimumHeight(320)
            self.results_section = output.add_section(
                self.results, "Regression results",
                persist_key=f"{APP_KEY}/Regression results", stretch=1)

            from ..widgets.figure_queue import FigureQueue
            self.figures = FigureQueue(parent=right)
            self.figures.setMinimumHeight(200)
            self.figures.hide()
            self.figures_section = output.add_section(
                self.figures, "Figures", persist_key=f"{APP_KEY}/Figures",
                stretch=1)
            right_layout.addWidget(output, 1)
            splitter.add_pane(right, "Sweep results", stretch=2)

            from ..dnd import install_for
            install_for(self, APP_KEY, self)


        def space(self):
            """Build the sweep space from the ticked axes.

            An unticked axis is not dropped -- it is PINNED to its first
            value and applied to every trial, so the settings a sweep ran
            under are always fully recorded rather than left to defaults that
            might change.
            """
            import ast

            def parse(text):
                """Read one comma-separated field as a list of values."""
                values = []
                for chunk in str(text).split(","):
                    chunk = chunk.strip()
                    if not chunk:
                        continue
                    try:
                        values.append(ast.literal_eval(chunk))
                    except (ValueError, SyntaxError):
                        values.append(chunk)
                return values

            axes, fixed = {}, {}
            for key, (include, editor) in self._axis_rows.items():
                values = parse(editor.text())
                if not values:
                    continue
                if include.isChecked() and len(values) > 1:
                    axes[key] = values
                else:
                    fixed[key] = values[0]
            return SweepSpace(axes=axes, fixed=fixed)

        def apply_settings(self, settings):
            """Seed the sweep from the module's settings panel.

            Opening the sweep should not mean retyping the inputs that are
            already on screen. The score/count CSVs, the response column and
            the output folder come straight across; every swept axis is
            additionally PINNED to the value the user currently has, so an
            unticked axis reproduces their run rather than a default.
            """
            if not isinstance(settings, dict):
                return
            for key, widget in (("score_data", self.score_data),
                                ("count_data", self.count_data)):
                value = settings.get(key)
                if value and hasattr(widget, "set_value"):
                    try:
                        widget.set_value(list(value))
                    except Exception:
                        pass
            response = settings.get("dependent_variable")
            if response:
                self.dependent_variable.setText(str(response))
            source = settings.get("src")
            if source and not self.destination.text().strip():
                self.destination.setText(os.path.join(str(source), "sweep"))
            for key, (include, editor) in self._axis_rows.items():
                if key in settings and settings[key] is not None:
                    current = settings[key]
                    text = "None" if current is None else str(current)
                    if not include.isChecked():
                        editor.setText(text)
                    elif text not in editor.text():
                        editor.setText(f"{text}, {editor.text()}")

        def base_settings(self):
            """The settings every trial in the sweep starts from."""
            return {
                "score_data": self.score_data.get_value(),
                "count_data": self.count_data.get_value(),
                "dependent_variable": self.dependent_variable.text().strip(),
                "annotation_source": "",
                "verbose": False,
            }


        def _set_worker_note(self, budget):
            """Show the worker calculation through translatable templates."""
            values = dict(budget)
            if values.get("available") is None:
                set_translatable_text(
                    self.worker_note,
                    "Available memory could not be measured, so the worker "
                    "count is limited to {workers}.",
                    workers=values["workers"],
                )
                return
            requested = values.get("requested")
            if requested and int(values["workers"]) < int(requested):
                set_translatable_text(
                    self.worker_note,
                    "{available:.0f} GiB available; allowing about "
                    "{per_trial:.1f} GiB per trial and "
                    "{budget_fraction:.0%} of available memory gives a "
                    "worker count of {workers}. The requested count was "
                    "{requested}.",
                    **values,
                )
                return
            set_translatable_text(
                self.worker_note,
                "{available:.0f} GiB available; allowing about "
                "{per_trial:.1f} GiB per trial and {budget_fraction:.0%} of "
                "available memory gives a worker count of {workers}.",
                **values,
            )

        def estimate(self):
            """Say how many trials the current space would run, without running them."""
            space = self.space()
            trials = build_trials(
                space, mode=self.mode.currentText(),
                max_trials=int(self.max_trials.value()),
                seed=int(self.seed.value()))
            worker_budget = _recommended_worker_budget(
                requested=int(self.workers.value()))
            workers = int(worker_budget["workers"])
            self._set_worker_note(worker_budget)
            assumed_seconds_per_trial = 60
            minutes = (len(trials) * assumed_seconds_per_trial
                       / max(workers, 1) / 60)
            self.status.setText(
                f"{space.size():,} raw combinations; {len(trials)} valid "
                f"trials would run on {workers} worker(s) — roughly "
                f"{minutes:.0f} minutes if each trial takes about one "
                f"minute. Invalid combinations are excluded before the "
                f"sweep starts.")
            return len(trials)

        def start(self):
            """Refuse an incomplete design, else run the sweep off the GUI thread."""
            base = self.base_settings()
            if not base["score_data"] or not base["count_data"]:
                QMessageBox.warning(self, "Nothing to sweep",
                                    "Choose at least one score CSV and one "
                                    "count CSV.")
                return
            destination = self.destination.text().strip()
            if not destination:
                QMessageBox.warning(self, "No output folder",
                                    "Choose a folder for the trial folders "
                                    "and the results table.")
                return
            space = self.space()
            mode = self.mode.currentText()
            max_trials = int(self.max_trials.value())
            seed = int(self.seed.value())
            workers = int(self.workers.value())
            self.start_button.setEnabled(False)
            self.progress.setVisible(True)
            self.progress.setRange(0, 0)
            self.status.setText("Sweeping…")

            def job():
                """Run the whole sweep. Called on a worker thread."""
                from ...parameter_sweep import run_sweep_parallel
                return run_sweep_parallel(
                    base, destination, space, mode=mode,
                    max_trials=max_trials, seed=seed, n_jobs=workers,
                    controls={"positive": str(base.get("positive_control_id",
                                                       "239740"))})

            self._runner.submit(job, self._sweep_finished)

        def _sweep_finished(self, results):
            """Show the results and give the controls back."""
            self.progress.setVisible(False)
            self.start_button.setEnabled(True)
            self._results = results
            if results is None or not len(results):
                self.status.setText("The sweep produced no trials.")
                return
            ok = int((results["status"] == "ok").sum())
            self.status.setText(
                f"{len(results)} trials, {ok} succeeded. "
                f"Results in {self.destination.text().strip()}")
            self._show(results)

        def load_results(self):
            """Read the table from disk, so a running sweep can be watched."""
            import pandas as pd
            path = os.path.join(self.destination.text().strip(),
                                "sweep_results.csv")
            if not os.path.exists(path):
                self.status.setText(f"No results table at {path} yet.")
                return
            frame = pd.read_csv(path)
            self._results = frame
            self.status.setText(f"{len(frame)} trials recorded so far.")
            self._show(frame)

        def _on_row_activated(self, *_args):
            """Re-run the selected trial and show its figures, editable.

            Off the GUI thread: this is a full regression, not a lookup. The
            row keeps its own settings, so what comes back is that trial and
            not a fresh one built from whatever the controls happen to say
            now -- which is the point of being able to compare conditions.
            """
            if self._results is None or not len(self._results):
                return
            row_index = self.table.currentRow()
            if row_index < 0:
                return
            key_item = self.table.item(row_index, 0)
            frame = self._results
            record = None
            if key_item is not None and "trial_id" in frame.columns:
                try:
                    match = frame[frame["trial_id"].astype(str)
                                  == key_item.text()]
                    if len(match):
                        record = match.iloc[0].to_dict()
                except Exception:
                    record = None
            if record is None and row_index < len(frame):
                record = frame.iloc[row_index].to_dict()
            if record is None:
                return
            if str(record.get("status", "ok")) != "ok":
                QMessageBox.information(
                    self, "That trial failed",
                    "This trial did not produce a regression:\n\n"
                    f"{record.get('error_type', '')}: "
                    f"{record.get('error', 'no reason recorded')}")
                return

            folder = record.get("folder")
            if folder and self.results.load(folder):
                self.trial_status.setText(
                    f"Trial {record.get('trial_id', '?')} loaded from disk "
                    f"({folder}). Nothing was re-fitted.")
                return

            base = self.base_settings()
            self.show_button.setEnabled(False)
            self.trial_status.setText(
                f"Trial {record.get('trial_id', '?')} has no saved results; "
                f"re-fitting it to draw them…")

            def job():
                """Re-run one trial from the table. Called on a worker thread."""
                from ...parameter_sweep import rerun_trial
                return rerun_trial(base, record)

            self._runner.submit(job, self._trial_figures_ready)

        def _trial_figures_ready(self, payload):
            """Put a re-run trial's figures on screen. On the GUI thread."""
            self.show_button.setEnabled(self.table.currentRow() >= 0)
            if not isinstance(payload, dict):
                self.trial_status.setText(
                    "That trial did not come back. See the console.")
                return
            figures = payload.get("figures") or []
            for figure in figures:
                try:
                    self.figures.add_figure(figure)
                except Exception:
                    pass
            if figures:
                self.figures.show()
            output = payload.get("output") or {}
            settings = payload.get("settings") or {}
            try:
                results = output.get("results")
                if results is not None and len(results):
                    self.results.set_frame(
                        results, source=str(settings.get("src", "")))
            except Exception:
                pass
            settings = payload.get("settings") or {}
            described = ", ".join(
                f"{key}={settings.get(key)!r}" for key in (
                    "regression_type", "inference", "analysis_unit",
                    "multiple_testing_method", "min_cells_per_well",
                    "fraction_threshold") if key in settings)
            self.trial_status.setText(
                f"{len(figures)} figure(s) from {described or 'that trial'}. "
                f"Right-click a figure to restyle it."
                if figures else
                "That trial produced no figures.")

        def _show(self, frame):
            """Put the results in the table, useful columns first.

            A VIEW AND NOT A FILTER: every column is still in the CSV. The order is
            settings, then what went in, then what came out -- a hit count means
            little without the size of the design it came from, and two trials
            differing only by a filtration cutoff can fit completely different data.
            """
            preferred = [c for c in (
                "trial_id", "status", "regression_type", "inference",
                "analysis_unit", "agg_type", "transform",
                "multiple_testing_method", "fdr_alpha",
                "fraction_threshold", "min_cells_per_well",
                "n_rows_fitted_grna", "n_wells_grna",
                "n_rows_fitted_gene", "n_wells_gene",
                "n_wells", "n_guides", "n_cells", "n_rows_fitted",
                "n_rows_prepared", "n_wells_prepared",
                "n_guides_prepared", "n_genes_prepared",
                "n_results", "n_below_alpha", "positive_rank",
                "seconds", "error_type") if c in frame.columns]
            columns = preferred or list(frame.columns)[:12]
            self.table.setColumnCount(len(columns))
            self.table.setHorizontalHeaderLabels(columns)
            self.table.setRowCount(min(len(frame), 2000))
            for row in range(self.table.rowCount()):
                for column, name in enumerate(columns):
                    value = frame.iloc[row][name]
                    self.table.setItem(row, column,
                                       table_item(str(value)))

    screen = ParameterSweepScreen(host=host)
    from ..theme import clear_container_surfaces, make_transparent
    make_transparent(screen)
    clear_container_surfaces(screen)
    return screen




#: Text on the toggle that reveals the card, and its hover help. Mirrors
#: spacr.qt.screens.hyperparam so the two searches read as the same feature in
#: two modules rather than as two different ones.
SWEEP_TOGGLE_TEXT = "Parameter sweep"
SWEEP_TOGGLE_TOOLTIP = (
    "Toggle the parameter sweep. When ON (blue), set a range for any "
    "regression setting -- model family, correction method, filtration "
    "cutoffs, wells vs cells -- run every legal combination, and double-click "
    "a row in the results to get that exact regression back with its figures."
)


[docs] def build_parameter_sweep_card(host): """Build the ``Parameter sweep`` card + panel pair. Mirrors :func:`spacr.qt.screens.hyperparam.build_hyperparam_card`: returns the pair without adding it to a layout, so the host puts it where it likes and starts it hidden behind the toggle. :param host: the ``AppScreen`` asking for the card. :returns: ``(panel, card)``. """ from ..widgets.card import Card card = Card(title="Parameter sweep") card.setMinimumHeight(320) holder = _lazy_sweep_panel(host) card.body_layout.addWidget(holder) return holder, card
_LAZY_SWEEP_PANEL_CLASS = None def _lazy_sweep_panel(host): """A stand-in that builds the real sweep panel the first time it is needed. :param host: the ``AppScreen`` the real panel will be built for. :returns: a ``LazySweepPanel`` holding nothing but a layout. """ return _lazy_sweep_panel_class()(host) def _lazy_sweep_panel_class(): """The ``LazySweepPanel`` class, defined once per process. The class is defined inside a function rather than at module scope because this module keeps PySide6 out of its import path on purpose -- see `_make_screen`. It is defined ONCE and cached, and the host is kept on the instance rather than captured by closure, because PySide6 never releases a QWidget subclass it has had to wrap: a class created per call, whose methods close over ``host``, pinned every regression ``AppScreen`` it was ever built for -- the screen's whole Python wrapper tree, long after Qt had deleted the widgets: about six MB and ~5,000 objects kept per regression screen built, and a serial ``pytest tests/qt`` builds hundreds of them. It forwards ``score_data`` and ``count_data`` because the drop handler finds the sweep by looking for that pair (see :func:`spacr.qt.dnd_handlers._sweep_panel`); asking for either builds the real panel, so a drop onto a card nobody has opened behaves exactly as it did when the panel was built up front. ``destination``, ``apply_settings``, and the panel's public ``space`` model are the only other interfaces its callers use. They are forwarded explicitly rather than through a catch-all ``__getattr__``: translation probes every widget for optional methods such as ``set_url`` and ``retranslate_dynamic_content``, and treating an introspection probe as a demand for the panel defeats the whole deferral. """ global _LAZY_SWEEP_PANEL_CLASS if _LAZY_SWEEP_PANEL_CLASS is not None: return _LAZY_SWEEP_PANEL_CLASS from PySide6.QtWidgets import QVBoxLayout, QWidget class LazySweepPanel(QWidget): """The sweep panel's place on screen, built the first time it shows. A REAL WIDGET THAT IS NOT THE PANEL, so the screen can be registered and laid out without paying for the panel's construction. The sweep panel pulls in the optimiser and its plotting stack, which is seconds of import on a cold interpreter, and a user who never opens Parameter Sweep should never pay it. It holds a layout and nothing else until `_ensure_panel` swaps the real one in, so the surrounding screen sizes correctly either way. """ def __init__(self, host): """Stand in for the panel without building it. :param host: the ``AppScreen`` the real panel is built for. """ super().__init__() self._host = host self._panel = None layout = QVBoxLayout(self) layout.setContentsMargins(0, 0, 0, 0) self._layout = layout def panel(self): """The real panel, built on the first ask.""" if self._panel is None: self._panel = _make_screen(host=self._host) self._layout.addWidget(self._panel) return self._panel def built(self) -> bool: """Whether the real panel exists yet. For tests and diagnostics.""" return self._panel is not None @property def score_data(self): """The real panel's score-file list, built on first use.""" return self.panel().score_data @property def count_data(self): """The real panel's count-file list, built on first use.""" return self.panel().count_data @property def destination(self): """The real panel's output-folder editor, built on first use.""" return self.panel().destination @property def space(self): """The real panel's sweep-space model, built on first use.""" return self.panel().space def apply_settings(self, settings): """Seed the real panel when the user opens the sweep card.""" return self.panel().apply_settings(settings) def showEvent(self, event): # noqa: N802 """Build the real panel the first time the screen is shown. That is the whole point of the placeholder: the sweep panel is expensive and a user who never opens this screen never pays for it. """ self.panel() super().showEvent(event) _LAZY_SWEEP_PANEL_CLASS = LazySweepPanel return LazySweepPanel
[docs] def sweepable(app_key: str) -> bool: """Whether a parameter sweep exists for ``app_key``. Only the regression module for now: the sweep axes, the legality filters and the row-to-regression round trip are all specific to it. :param app_key: the module's registry key; only ``"regression"`` gives ``True``. """ return app_key == "regression"