"""Configure, submit and monitor remote/distributed spaCR jobs.
Every network and scheduler call runs through :func:`make_thread`; the GUI
thread only updates controls and renders already-returned records.
"""
from __future__ import annotations
import json
import logging
import os
import threading
from pathlib import Path
from typing import Callable, List, Optional
from PySide6.QtCore import QObject, Qt, QThread, QTimer, QUrl, Signal
from PySide6.QtGui import QColor, QDesktopServices
from PySide6.QtWidgets import (
QAbstractItemView,
QApplication,
QCheckBox,
QComboBox,
QDialog,
QDialogButtonBox,
QFileDialog,
QFormLayout,
QHBoxLayout,
QHeaderView,
QInputDialog,
QLabel,
QLineEdit,
QMessageBox,
QPlainTextEdit,
QPushButton,
QSpinBox,
QTableWidget,
QTableWidgetItem,
QVBoxLayout,
QWidget,
)
from ...remote_execution import (
_SSH_SESSIONS,
ACTIVE_STATES,
ExecutionProfile,
RemoteExecutionError,
RemoteJob,
RemoteJobManager,
_discover_ssh_settings,
_keychain_account,
_keychain_available,
_keychain_password,
_remote_argv,
_run_command,
)
from ..bridge import make_thread
from ..i18n import tr
from ..iconset import icon
from ..theme import SPACING, active_palette
from ..widgets import Card, Divider
from ..widgets.collapsible_splitter import CollapsibleSplitter
from ..widgets.sortable_table import (
_retire_table_cells, _spare_table_item, install_sorting, table_item,
)
from .settings_model import attach_api_tooltip
LOG = logging.getLogger(__name__)
APP_KEY = "distributed_jobs"
APP_NAME = "Distributed Jobs"
APP_SECTION = "Data"
APP_INTRO = (
"Submit a normal spaCR settings file to another workstation, Slurm, or "
"a cloud/HPC command; monitor logs and cancel it without blocking spaCR."
)
_COLUMNS = (
"Job", "Status", "Module", "Profile", "Remote ID", "Submitted", "Updated",
)
def _item(value) -> QTableWidgetItem:
"""Return a non-editable table item."""
item = table_item("" if value is None else str(value))
item.setFlags(item.flags() & ~Qt.ItemIsEditable)
return item
class _SSHAuthBroker(QObject):
"""Marshal ephemeral SSH challenges onto the application GUI thread."""
requested = Signal(object)
def __init__(self, app):
"""Hold pending SSH challenges and connect GUI delivery and shutdown."""
super().__init__(app)
self._notices = []
self._workers = []
self._pending = []
self.requested.connect(self._show, Qt.QueuedConnection)
app.aboutToQuit.connect(self._shutdown)
def _request(self, title, instructions, prompts):
"""Wait only in the SSH worker; Qt retains control of the window."""
if QThread.currentThread() == self.thread():
raise RemoteExecutionError(tr("SSH authentication must run in the background."))
request = {"title": title, "instructions": instructions, "prompts": prompts,
"event": threading.Event(), "answers": None, "cancelled": False}
self._pending.append(request)
self.requested.emit(request)
try:
if not request["event"].wait(185):
request["cancelled"] = True
return None
return request["answers"]
finally:
self._pending.remove(request)
def _shutdown(self):
"""Cancel challenges before closing transports off the GUI thread."""
for request in list(self._pending):
request["cancelled"] = True
request["event"].set()
threading.Thread(target=_SSH_SESSIONS.close, daemon=True).start()
def _show(self, request):
"""Show server text verbatim, masking prompts whose echo flag is false."""
if request["cancelled"]:
return
answers = []
for notice in self._notices:
notice.close()
self._notices.clear()
if not request["prompts"] and request["instructions"]:
notice = QMessageBox(QApplication.activeWindow())
notice.setWindowTitle(str(request["title"]) or tr("SSH authentication"))
notice.setTextFormat(Qt.PlainText)
notice.setText(str(request["instructions"]))
notice.show()
self._notices.append(notice)
for prompt, echo in request["prompts"]:
dialog = QInputDialog(QApplication.activeWindow())
dialog.setWindowTitle(str(request["title"]) or tr("SSH authentication"))
text = str(request["instructions"])
dialog.setLabelText((text + "\n\n" if text else "") + str(prompt))
for label in dialog.findChildren(QLabel):
label.setTextFormat(Qt.PlainText)
dialog.setTextEchoMode(QLineEdit.Normal if echo else QLineEdit.Password)
timer = QTimer(dialog)
timer.setSingleShot(True)
timer.timeout.connect(dialog.reject)
timer.start(180000)
accepted = dialog.exec() == QDialog.Accepted
timer.stop()
if not accepted or request["cancelled"]:
answers.clear()
request["event"].set()
return
answers.append(dialog.textValue())
dialog.setTextValue("")
if not request["cancelled"]:
request["answers"] = answers
request["event"].set()
def _ensure_ssh_broker():
"""Keep the authentication bridge alive after a profile editor closes."""
app = QApplication.instance()
broker = getattr(app, "_spacr_ssh_auth_broker", None)
if broker is None:
broker = _SSHAuthBroker(app)
app._spacr_ssh_auth_broker = broker
_SSH_SESSIONS.prompt = broker._request
return broker
[docs]
class ExecutionProfileDialog(QDialog):
"""Create or edit one distributed execution profile.
:param parent: parent widget.
:param profile: the profile to edit. ``None`` creates one -- the same
dialog handles both cases.
"""
def __init__(
self,
parent=None,
profile: Optional[ExecutionProfile] = None,
):
"""Build the execution-profile editor.
:param parent: parent widget, or ``None``.
:param profile: the profile to edit; ``None`` starts a new one. Its
original name is kept so a rename can be told from a new profile.
"""
super().__init__(parent)
self.setWindowTitle(tr("Execution profile"))
from ..preferences import scaled_px
self.setMinimumWidth(scaled_px(650))
self._original_name = profile.name if profile else ""
self._ssh_options = {}
self._loaded_keychain = (profile.host, dict(profile.ssh_options)) if profile is not None and profile.ssh_keychain else None
self._connection_worker = None
_ensure_ssh_broker()
self._build_ui()
if profile is not None:
self._load(profile)
self._sync_backend()
from .settings_model import retarget_field_tooltips
retarget_field_tooltips(self)
def _build_ui(self) -> None:
"""Build the backend-independent profile editor."""
outer = QVBoxLayout(self)
intro = QLabel(
tr(
"Profiles store connection settings, never passwords or OTPs. SSH prompts stay in this session."
),
self,
)
intro.setWordWrap(True)
intro.setObjectName("Muted")
outer.addWidget(intro)
form = QFormLayout()
self._name = QLineEdit(self)
self._name.setObjectName("ProfileName")
self._name.setToolTip(
"A local display name. API: spacr.remote_execution.ExecutionProfile"
)
self._add_profile_row(
form, "Profile name", self._name, "profile_name"
)
self._backend = QComboBox(self)
self._backend.setObjectName("ProfileBackend")
self._backend.addItem(tr("SSH workstation"), "ssh")
self._backend.addItem(tr("Slurm cluster"), "slurm")
self._backend.addItem(tr("Cloud / custom command"), "command")
self._backend.setToolTip(
"SSH runs a durable background process; Slurm uses sbatch/squeue/"
"sacct; custom executes argument templates without shell=True."
)
self._backend.currentIndexChanged.connect(self._sync_backend)
self._add_profile_row(form, "Backend", self._backend, "backend")
self._host = QLineEdit(self)
self._host.setObjectName("ProfileHost")
self._host.setPlaceholderText("user@workstation or cluster-login")
self._host.setToolTip(
"OpenSSH target. Leave blank only for Slurm commands installed "
"on this computer. Configure key authentication outside spaCR."
)
self._add_profile_row(form, "SSH host", self._host, "host")
discovery = QHBoxLayout()
self._find_ssh_button = QPushButton(tr("Find my SSH settings"), self)
self._find_ssh_button.setObjectName("FindSSHSettings")
self._find_ssh_button.setToolTip(tr("Reads your SSH config file and the files it includes, the system-wide SSH config, the .ssh folder in your Windows user profile, and known_hosts. Read-only: spaCR never changes SSH files, never reads private keys and never runs ssh -G."))
self._find_ssh_button.clicked.connect(lambda: self._find_ssh())
self._choose_ssh_button = QPushButton(tr("Choose folder…"), self)
self._choose_ssh_button.setObjectName("ChooseSSHFolder")
self._choose_ssh_button.setToolTip(tr("Choose another folder that holds an SSH config or known_hosts file."))
self._choose_ssh_button.clicked.connect(self._choose_ssh_folder)
discovery.addWidget(self._find_ssh_button)
discovery.addWidget(self._choose_ssh_button)
form.addRow(discovery)
self._ssh_aliases = QComboBox(self)
self._ssh_aliases.setObjectName("SSHDiscoveredAliases")
self._ssh_aliases.currentIndexChanged.connect(self._select_ssh)
form.addRow(tr("SSH settings"), self._ssh_aliases)
self._ssh_hostname = QLineEdit(self)
self._ssh_user = QLineEdit(self)
self._ssh_port = QSpinBox(self)
self._ssh_port.setRange(0, 65535)
self._ssh_port.setSpecialValueText(tr("From SSH settings"))
self._ssh_key = QLineEdit(self)
self._ssh_jump = QLineEdit(self)
for label, widget, key in (("SSH host name", self._ssh_hostname, "ssh_hostname"), ("SSH user", self._ssh_user, "ssh_user"), ("SSH port", self._ssh_port, "ssh_port"),
("SSH identity file", self._ssh_key, "ssh_identity_file"), ("SSH jump host", self._ssh_jump, "ssh_jump_host")):
widget.setObjectName(key)
widget.setToolTip(tr("Non-secret SSH settings. Review imported paths and routing before connecting."))
self._add_profile_row(form, label, widget, key)
self._ssh_interactive = QCheckBox(tr("Authenticate in spaCR (keyboard-interactive / MFA)"), self)
self._ssh_interactive.setObjectName("SSHInteractiveAuthentication")
self._ssh_interactive.setToolTip(tr("Great Lakes may ask for a Kerberos password and Okta verification. spaCR shows server prompts verbatim and stores no password or OTP."))
form.addRow(self._ssh_interactive)
self._ssh_keychain = QCheckBox(tr("Remember my password in the system keychain"), self)
self._ssh_keychain.setObjectName("SSHKeychainPassword")
self._ssh_keychain.setChecked(False)
self._ssh_keychain.setToolTip(
tr("Off by default. Stores only the login password in the operating-system keychain, never one-time codes or push answers.")
if _keychain_available() else tr("Install the keyring package to store the password in the system keychain."))
self._ssh_interactive.toggled.connect(self._sync_backend)
form.addRow(self._ssh_keychain)
connection = QHBoxLayout()
self._test_ssh = QPushButton(tr("Test connection"), self)
self._test_ssh.setObjectName("SSHTestConnection")
self._test_ssh.clicked.connect(self._test_connection)
self._retry_ssh = QPushButton(tr("Retry sign-in"), self)
self._retry_ssh.setObjectName("SSHRetryConnection")
self._retry_ssh.clicked.connect(self._test_connection)
self._retry_ssh.setVisible(False)
connection.addWidget(self._test_ssh)
connection.addWidget(self._retry_ssh)
form.addRow(connection)
self._workdir = QLineEdit(self)
self._workdir.setObjectName("ProfileWorkdir")
self._workdir.setPlaceholderText("/shared/project")
self._workdir.setToolTip(
"Absolute directory on the execution host. Small settings and log "
"files are stored below <workdir>/spacr-jobs/."
)
self._add_profile_row(
form, "Remote work directory", self._workdir, "workdir"
)
self._local_root = QLineEdit(self)
self._local_root.setObjectName("ProfileLocalRoot")
self._local_root.setPlaceholderText("/local/mount/project")
self._local_root.setToolTip(
"Optional local prefix for shared/mirrored input data. spaCR maps "
"paths but deliberately does not copy entire image datasets."
)
self._add_profile_row(
form, "Local dataset root", self._local_root, "local_root"
)
self._remote_root = QLineEdit(self)
self._remote_root.setObjectName("ProfileRemoteRoot")
self._remote_root.setPlaceholderText("/cluster/mount/project")
self._remote_root.setToolTip(
"Remote prefix representing the same data as Local dataset root."
)
self._add_profile_row(
form, "Remote dataset root", self._remote_root, "remote_root"
)
self._runner = QLineEdit(self)
self._runner.setObjectName("ProfileRunner")
self._runner.setText("spacr-run")
self._runner.setToolTip(
"Installed headless spaCR executable on the remote target."
)
self._add_profile_row(
form, "spaCR runner", self._runner, "runner"
)
self._slurm = QLineEdit(self)
self._slurm.setObjectName("ProfileSlurmOptions")
self._slurm.setPlaceholderText("--partition=gpu --gres=gpu:1 --time=12:00:00")
self._slurm.setToolTip(
"Additional sbatch arguments. They are parsed as arguments, not "
"executed as a shell command."
)
self._add_profile_row(
form, "Slurm options", self._slurm, "scheduler_options"
)
self._submit_command = QLineEdit(self)
self._submit_command.setObjectName("ProfileSubmitCommand")
self._submit_command.setPlaceholderText(
"cloud-submit --module {module} --settings {settings}"
)
self._submit_command.setToolTip(
"Argument template. Must print a job ID and contain {settings}. "
"Also supports {job_id}, {module} and {profile}."
)
self._add_profile_row(
form, "Submit command", self._submit_command, "submit_command"
)
self._status_command = QLineEdit(self)
self._status_command.setObjectName("ProfileStatusCommand")
self._status_command.setPlaceholderText(
"cloud-status {external_id}"
)
self._status_command.setToolTip(
"Must print a conventional state such as PENDING, RUNNING, "
"SUCCEEDED, FAILED or CANCELLED."
)
self._add_profile_row(
form, "Status command", self._status_command, "status_command"
)
self._cancel_command = QLineEdit(self)
self._cancel_command.setObjectName("ProfileCancelCommand")
self._cancel_command.setPlaceholderText(
"cloud-cancel {external_id}"
)
self._cancel_command.setToolTip(
"Argument template used to request cancellation. Must contain "
"{external_id}."
)
self._add_profile_row(
form, "Cancel command", self._cancel_command, "cancel_command"
)
self._log_command = QLineEdit(self)
self._log_command.setObjectName("ProfileLogCommand")
self._log_command.setPlaceholderText("cloud-logs {external_id}")
self._log_command.setToolTip(
"Optional argument template that prints the job log. Use "
"{external_id}; credentials stay in the external CLI."
)
self._add_profile_row(
form, "Log command (optional)", self._log_command, "log_command"
)
self._job_pattern = QLineEdit(self)
self._job_pattern.setObjectName("ProfileJobPattern")
self._job_pattern.setPlaceholderText(
r'"jobId":\s*"(?P<id>[A-Za-z0-9-]+)"'
)
self._job_pattern.setToolTip(
"Optional regular expression extracting a job ID from submit "
"output. Use a named group (?P<id>...)."
)
self._add_profile_row(
form, "Job-ID pattern (optional)", self._job_pattern,
"job_id_pattern",
)
self._poll = QSpinBox(self)
self._poll.setObjectName("ProfilePollSeconds")
self._poll.setRange(2, 3600)
self._poll.setValue(10)
self._poll.setSuffix(tr(" seconds"))
self._poll.setToolTip(
"Suggested local polling interval. Manual refresh is always "
"available."
)
self._add_profile_row(
form, "Poll interval", self._poll, "poll_seconds"
)
outer.addLayout(form)
self._error = QLabel("", self)
self._error.setObjectName("InlineError")
self._error.setWordWrap(True)
outer.addWidget(self._error)
buttons = QDialogButtonBox(
QDialogButtonBox.Save | QDialogButtonBox.Cancel, parent=self
)
save = buttons.button(QDialogButtonBox.Save)
if save is not None:
save.setObjectName("PrimaryButton")
cancel = buttons.button(QDialogButtonBox.Cancel)
if cancel is not None:
cancel.setObjectName("DangerButton")
buttons.accepted.connect(self._validate_and_accept)
buttons.rejected.connect(self.reject)
outer.addWidget(buttons)
def _add_profile_row(
self,
form: QFormLayout,
source_label: str,
field: QWidget,
api_key: str,
) -> None:
"""Put this field's hover help, API link and all, on its label.
NOTHING IS DRAWN BESIDE THE LABEL. A teal API dot used to be, and
the settings forms had already dropped theirs: a column of dots
reads as texture rather than as one affordance per setting. The
link is the last line of the help, where the reader was already
finding it.
"""
help_text = field.toolTip() or (
f"Controls {source_label.casefold()} for distributed execution."
)
label = QLabel(tr(source_label), self)
label.setObjectName("SettingsLabel")
attach_api_tooltip(
label, "distributed_jobs", api_key, help_text
)
field.setToolTip("")
label.setCursor(Qt.WhatsThisCursor)
label.setProperty("settingHelpLabel", True)
self._install_help_filter(label)
form.addRow(label, field)
def _install_help_filter(self, label: QWidget) -> None:
"""Give ``label`` the clickable sticky popup the settings forms use.
A native Qt tooltip cannot be walked into, so its API link cannot be
clicked -- the pointer leaving the label takes the link with it.
One filter per dialog: Qt keeps a LIST of event filters, so a second
object would show two popups for one hover.
"""
filter_ = getattr(self, "_api_tooltip_filter", None)
if filter_ is None:
from .settings_model import _ApiTooltipFilter
filter_ = _ApiTooltipFilter(self)
self._api_tooltip_filter = filter_
filter_.watch(label)
def _sync_backend(self, *_args) -> None:
"""Show only controls meaningful to the selected backend."""
backend = self._backend.currentData()
ssh_or_slurm = backend in {"ssh", "slurm"}
for widget in (
self._host, self._workdir, self._local_root, self._remote_root,
self._runner, self._ssh_hostname, self._choose_ssh_button, self._ssh_user, self._ssh_port, self._ssh_key, self._ssh_jump,
self._ssh_aliases, self._find_ssh_button, self._ssh_interactive, self._test_ssh,
):
widget.setEnabled(ssh_or_slurm)
self._ssh_keychain.setEnabled(ssh_or_slurm and self._ssh_interactive.isChecked() and _keychain_available())
self._slurm.setEnabled(backend == "slurm")
for widget in (
self._submit_command, self._status_command, self._cancel_command,
self._log_command, self._job_pattern,
):
widget.setEnabled(backend == "command")
def _load(self, profile: ExecutionProfile) -> None:
"""Populate controls from an existing profile."""
self._name.setText(profile.name)
index = self._backend.findData(profile.backend)
self._backend.setCurrentIndex(max(0, index))
self._host.setText(profile.host)
self._ssh_hostname.setText(profile.ssh_options.get("hostname", ""))
self._ssh_options = dict(profile.ssh_options)
self._ssh_user.setText(profile.ssh_options.get("user", ""))
self._ssh_port.setValue(int(profile.ssh_options.get("port", "0")))
self._ssh_key.setText(profile.ssh_options.get("identityfile", ""))
self._ssh_jump.setText(profile.ssh_options.get("proxyjump", ""))
self._ssh_interactive.setChecked(profile.ssh_interactive)
self._ssh_keychain.setChecked(profile.ssh_keychain)
self._workdir.setText(profile.workdir)
self._local_root.setText(profile.local_root)
self._remote_root.setText(profile.remote_root)
self._runner.setText(profile.runner)
self._slurm.setText(profile.scheduler_options)
self._submit_command.setText(profile.submit_command)
self._status_command.setText(profile.status_command)
self._cancel_command.setText(profile.cancel_command)
self._log_command.setText(profile.log_command)
self._job_pattern.setText(profile.job_id_pattern)
self._poll.setValue(profile.poll_seconds)
[docs]
def profile(self) -> ExecutionProfile:
"""Return the currently entered, validated profile."""
return ExecutionProfile(
name=self._name.text(),
backend=str(self._backend.currentData()),
host=self._host.text(),
workdir=self._workdir.text(),
local_root=self._local_root.text(),
remote_root=self._remote_root.text(),
runner=self._runner.text(),
scheduler_options=self._slurm.text(),
submit_command=self._submit_command.text(),
status_command=self._status_command.text(),
cancel_command=self._cancel_command.text(),
log_command=self._log_command.text(),
job_id_pattern=self._job_pattern.text(),
poll_seconds=self._poll.value(),
ssh_options=self._connection_options(),
ssh_interactive=self._ssh_interactive.isChecked(),
ssh_keychain=self._ssh_interactive.isChecked() and self._ssh_keychain.isChecked(),
).validate()
def _find_ssh(self, folder=None):
"""Populate aliases only after an explicit read-only discovery action."""
try:
rows = _discover_ssh_settings(folder)
except (OSError, ValueError, RemoteExecutionError):
self._error.setText(tr("SSH settings could not be read. Choose a folder containing a readable config file."))
return
self._ssh_aliases.clear()
for row in rows:
self._ssh_aliases.addItem(f'{row["alias"]} → {row["hostname"]}', row)
if not rows:
self._error.setText(tr("No SSH aliases or public host names were found."))
else:
self._error.setText(tr("SSH settings imported. Conditional Match blocks are not evaluated; review the connection before testing."))
self._select_ssh(0)
def _choose_ssh_folder(self):
"""Read a user-selected configuration folder without copying keys."""
folder = QFileDialog.getExistingDirectory(self, tr("Choose SSH settings folder"))
if folder:
self._find_ssh(folder)
def _select_ssh(self, index):
"""Apply one alias and its explicit non-secret connection settings."""
row = self._ssh_aliases.itemData(index)
if not row:
return
try:
port = int(row.get("port", "22"))
if not 1 <= port <= 65535:
raise ValueError
except ValueError:
self._error.setText(tr("SSH port must be between 1 and 65535."))
return
self._host.setText(row["alias"])
self._ssh_options = {key: value for key, value in row.items() if key != "alias"}
self._ssh_hostname.setText(row.get("hostname", ""))
self._ssh_user.setText(row.get("user", ""))
self._ssh_port.setValue(port)
self._ssh_key.setText(row.get("identityfile", ""))
self._ssh_jump.setText(row.get("proxyjump", ""))
self._ssh_interactive.setChecked(True)
def _connection_options(self):
"""Collect paths and routing fields, never authentication responses."""
options = dict(self._ssh_options)
for key, value in (("hostname", self._ssh_hostname.text()), ("user", self._ssh_user.text()), ("port", str(self._ssh_port.value()) if self._ssh_port.value() else ""),
("identityfile", self._ssh_key.text()), ("proxyjump", self._ssh_jump.text())):
if value.strip():
options[key] = value.strip()
else:
options.pop(key, None)
if options.get("proxyjump"):
options.pop("proxycommand", None)
return options
def _test_connection(self):
"""Run hostname, then Slurm discovery, on a background SSH channel."""
if self._connection_worker is not None:
return
try:
profile = ExecutionProfile(name="connection-test", backend="slurm" if self._backend.currentData() == "slurm" else "ssh", host=self._host.text(), workdir="/tmp",
ssh_options=self._connection_options(), ssh_interactive=self._ssh_interactive.isChecked(),
ssh_keychain=self._ssh_interactive.isChecked() and self._ssh_keychain.isChecked()).validate()
except RemoteExecutionError as exc:
self._error.setText(str(exc))
return
self._connection_error = ""
slurm = self._backend.currentData() == "slurm"
self._test_ssh.setEnabled(False)
self._retry_ssh.setVisible(False)
self._error.setText(tr("Testing SSH connection…"))
def operation(_settings):
"""Check the remote hostname before querying Slurm on the worker thread."""
stages = [("hostname", ["hostname"])]
if slurm:
stages.append(("sinfo", ["sinfo", "--noheader"]))
for stage, argv in stages:
try:
result = _run_command(_remote_argv(profile, argv), timeout=60)
if result.returncode:
raise RemoteExecutionError(tr("Remote command returned a non-zero exit status."))
except RemoteExecutionError as exc:
self._connection_error = tr("{stage} failed: {error}").format(stage=stage, error=exc)
break
thread, worker = make_thread(operation, {}, app_key="ssh_connection_test", journal=False, capture_figures=False)
self._connection_worker = (thread, worker)
_ensure_ssh_broker()._workers.append((self, thread, worker))
worker.finished.connect(self._connection_finished)
thread.finished.connect(self._connection_retired)
thread.start()
def _connection_finished(self, ok):
"""Report the precise failed stage without retaining secret responses."""
self._test_ssh.setEnabled(True)
failed = bool(self._connection_error) or not ok
self._retry_ssh.setVisible(failed)
self._error.setText(self._connection_error or (tr("SSH connection test passed.") if ok else tr("SSH connection test failed.")))
def _connection_retired(self):
"""Release a finished worker while keeping a running worker owned."""
self._connection_worker = None
broker = _ensure_ssh_broker()
broker._workers[:] = [entry for entry in broker._workers if entry[0] is not self]
def _validate_and_accept(self) -> None:
"""Keep the dialog open and show validation errors inline."""
try:
self.profile()
except RemoteExecutionError as exc:
self._error.setText(str(exc))
return
self._error.clear()
if self._loaded_keychain and not self._ssh_keychain.isChecked():
self._forget_password(*self._loaded_keychain)
self.accept()
def _forget_password(self, host, options):
"""Remove a remembered password off the GUI thread once storage is unticked."""
try:
account = _keychain_account(host, options)
except (OSError, RemoteExecutionError):
return
threading.Thread(target=_keychain_password, args=("delete", account), daemon=True).start()
[docs]
class DistributedJobsScreen(QWidget):
"""Submit and monitor distributed spaCR jobs without blocking Qt.
:param parent: parent widget.
:param manager: the job manager to submit through. One is created when
none is given, so a test can supply a manager that reaches no cluster.
:param threaded: whether submissions and polls run off the GUI thread.
False runs them inline, which is what makes a test deterministic.
:param auto_poll: whether the screen starts its own refresh timer. False
leaves the job list to be refreshed by hand, so a test is not racing
a timer it did not start.
"""
def __init__(
self,
parent=None,
*,
manager: Optional[RemoteJobManager] = None,
threaded: bool = True,
auto_poll: bool = True,
):
"""Build the screen and arm its drop zone and poll timer.
A store that cannot be read reports itself on the status line and leaves
the screen usable with no jobs listed, rather than failing to build --
the profiles half still works and is how the store gets fixed.
:param parent: parent widget, or ``None``.
:param manager: the remote job manager; ``None`` opens the default one.
:param threaded: run submissions and polls on a worker thread.
:param auto_poll: start the refresh timer. Off in tests, which drive
``refresh`` themselves.
"""
super().__init__(parent)
self.manager = manager or RemoteJobManager()
self._threaded = bool(threaded)
self._auto_poll = bool(auto_poll)
self._jobs: List[RemoteJob] = []
self._workers: List[tuple] = []
self._busy = False
self._pending_result = None
self._pending_error = ""
self._pending_callback: Optional[Callable] = None
self._settings_snapshot: Optional[dict] = None
self._snapshot_module = ""
self.setAcceptDrops(True)
self._build_ui()
self._reload_profiles()
try:
initial_jobs = self.manager.jobs.list()
except Exception as exc:
initial_jobs = []
self._set_status(
f"Could not load distributed job records: "
f"{type(exc).__name__}: {exc}",
error=True,
)
self._render_jobs(initial_jobs)
self._timer = QTimer(self)
self._timer.timeout.connect(self.refresh)
self._update_poll_interval()
from ..dnd import install_for
install_for(self, "distributed_jobs")
from .settings_model import retarget_field_tooltips
retarget_field_tooltips(self)
def _build_ui(self) -> None:
"""Construct profile, submission, table, and detail controls."""
outer = QVBoxLayout(self)
outer.setContentsMargins(
SPACING["lg"], SPACING["lg"], SPACING["lg"], SPACING["lg"]
)
outer.setSpacing(SPACING["md"])
title = QLabel(APP_NAME, self)
title.setObjectName("DisplayHeading")
outer.addWidget(title)
intro = QLabel(APP_INTRO, self)
intro.setObjectName("Muted")
intro.setWordWrap(True)
outer.addWidget(intro)
outer.addWidget(Divider())
submit_card = Card(title="Submission")
profile_row = QHBoxLayout()
self._profile = QComboBox(self)
self._profile.setObjectName("ExecutionProfileChoice")
self._profile.setProperty("i18nSkipItems", True)
self._profile.setToolTip(
"Saved target from spacr.remote_execution.ProfileStore."
)
self._new_profile = QPushButton("New profile…", self)
self._new_profile.clicked.connect(self._create_profile)
self._edit_profile = QPushButton("Edit profile…", self)
self._edit_profile.clicked.connect(self._edit_selected_profile)
self._delete_profile = QPushButton("Delete profile", self)
self._delete_profile.setObjectName("DangerButton")
self._delete_profile.clicked.connect(self._delete_selected_profile)
profile_row.addWidget(QLabel("Execution target", self))
profile_row.addWidget(self._profile, 1)
profile_row.addWidget(self._new_profile)
profile_row.addWidget(self._edit_profile)
profile_row.addWidget(self._delete_profile)
submit_card.body_layout.addLayout(profile_row)
job_row = QHBoxLayout()
self._module = QComboBox(self)
self._module.setObjectName("DistributedModule")
self._module.setProperty("i18nSkipItems", True)
from ...cli import MODULES
for key in sorted(MODULES):
self._module.addItem(key, key)
self._settings_path = QLineEdit(self)
self._settings_path.setObjectName("DistributedSettingsPath")
self._settings_path.setPlaceholderText(
"Drop or choose a spaCR settings CSV/JSON"
)
self._settings_path.setToolTip(
"A settings CSV exported by a module or settings.json from a run "
"manifest. It is fully resolved before submission."
)
self._settings_path.textEdited.connect(self._clear_settings_snapshot)
browse = QPushButton("Browse…", self)
browse.clicked.connect(self._browse_settings)
from ..i18n import set_translatable_text
self._save_snapshot = QPushButton(self)
set_translatable_text(self._save_snapshot, "Save settings CSV…")
self._save_snapshot.setObjectName("DistributedSaveSnapshot")
self._save_snapshot.setToolTip(tr(
"Save the settings snapshot sent by Submit remote… as a settings "
"CSV, to copy to a cluster and run with spacr-run --settings."))
self._save_snapshot.setEnabled(False)
self._save_snapshot.clicked.connect(self._save_snapshot_csv)
self._submit = QPushButton("Submit", self)
self._submit.setObjectName("PrimaryButton")
self._submit.clicked.connect(self.submit)
job_row.addWidget(QLabel("Module", self))
job_row.addWidget(self._module)
job_row.addWidget(self._settings_path, 1)
job_row.addWidget(browse)
job_row.addWidget(self._save_snapshot)
job_row.addWidget(self._submit)
submit_card.body_layout.addLayout(job_row)
self._allocated_gpus = QCheckBox(
tr("Mask generation: segment batches on every GPU allocated to the job"),
self)
self._allocated_gpus.setObjectName("DistributedAllocatedGpus")
self._allocated_gpus.setToolTip(tr(
"Submits mask_parallel on with a blank mask_gpu_indices, so the "
"job uses exactly the GPUs its scheduler allocates, one model per "
"GPU. Request them in the execution profile, for example "
"--gres=gpu:2. A job given one GPU runs on that one."))
submit_card.body_layout.addWidget(self._allocated_gpus)
self._module.currentIndexChanged.connect(self._sync_allocated_gpus)
self._sync_allocated_gpus()
outer.addWidget(submit_card)
actions = QHBoxLayout()
self._refresh = QPushButton("Refresh", self)
self._refresh.setObjectName("PrimaryButton")
self._refresh.setIcon(icon("redo"))
self._refresh.clicked.connect(self.refresh)
self._cancel = QPushButton("Cancel job", self)
self._cancel.setObjectName("DangerButton")
self._cancel.clicked.connect(self.cancel_selected)
self._logs = QPushButton("Refresh log", self)
self._logs.clicked.connect(self.refresh_log)
self._open_local = QPushButton("Open local record", self)
self._open_local.setIcon(icon("folder"))
self._open_local.clicked.connect(self._open_local_record)
actions.addWidget(self._refresh)
actions.addWidget(self._cancel)
actions.addWidget(self._logs)
actions.addWidget(self._open_local)
actions.addStretch(1)
outer.addLayout(actions)
splitter = CollapsibleSplitter(Qt.Vertical, self,
persist_key="distributed_jobs::body")
self._table = QTableWidget(0, len(_COLUMNS))
install_sorting(self._table)
self._table.setHorizontalHeaderLabels(list(_COLUMNS))
self._table.setSelectionBehavior(QAbstractItemView.SelectRows)
self._table.setSelectionMode(QAbstractItemView.SingleSelection)
self._table.setEditTriggers(QAbstractItemView.NoEditTriggers)
self._table.verticalHeader().setVisible(False)
self._table.setAlternatingRowColors(True)
header = self._table.horizontalHeader()
header.setSectionResizeMode(QHeaderView.ResizeToContents)
header.setSectionResizeMode(2, QHeaderView.Stretch)
self._table.itemSelectionChanged.connect(self._show_selection)
splitter.add_section(self._table, "Jobs",
persist_key="distributed_jobs/Jobs", stretch=2)
self._detail = QPlainTextEdit()
self._detail.setReadOnly(True)
self._detail.setLineWrapMode(QPlainTextEdit.NoWrap)
self._detail.setAccessibleName("Distributed job details and log")
splitter.add_section(self._detail, "Job details and log",
persist_key="distributed_jobs/Job details and log",
stretch=1)
outer.addWidget(splitter, 1)
self._body_splitter = splitter
self._status = QLabel("", self)
self._status.setObjectName("Muted")
self._status.setWordWrap(True)
outer.addWidget(self._status)
def _sync_allocated_gpus(self, *_args) -> None:
"""Offer allocated-GPU mask batches only for the mask module."""
self._allocated_gpus.setEnabled(
str(self._module.currentData() or "") == "mask")
def _clear_settings_snapshot(self, _text: str = "") -> None:
"""Switch back to file mode when the user edits the path field."""
self._settings_snapshot = None
self._save_snapshot.setEnabled(False)
def _save_snapshot_csv(self) -> None:
"""Save the settings snapshot as a CSV that ``spacr-run`` reads.
Submitting is unchanged: the snapshot stays what Submit sends. The
file is the one Export settings… writes on the module's own screen.
"""
if self._settings_snapshot is None:
return
from .app_screen import default_settings_csv_path, export_settings_csv
module = self._snapshot_module
default = default_settings_csv_path(
module, self._settings_snapshot.get("src"))
try:
written = export_settings_csv(self, self._settings_snapshot,
default)
except Exception as exc:
self._set_status(
tr("Could not save the settings: {error}").format(error=exc),
error=True)
return
if written:
self._set_status(
tr("Saved the settings to {path}. On the cluster run: "
"spacr-run {module} --settings {name}").format(
path=written[0], module=module,
name=os.path.basename(written[0])))
def _reload_profiles(self, selected: str = "") -> None:
"""Reload the profile combo from persistent storage."""
current = selected or str(self._profile.currentData() or "")
try:
profiles = self.manager.profiles.list()
except Exception as exc:
profiles = []
self._set_status(
f"Could not load execution profiles: "
f"{type(exc).__name__}: {exc}",
error=True,
)
self._profile.clear()
for profile in profiles:
self._profile.addItem(profile.name, profile.name)
index = self._profile.findData(current)
self._profile.setCurrentIndex(index if index >= 0 else 0)
available = bool(profiles)
self._edit_profile.setEnabled(available)
self._delete_profile.setEnabled(available)
self._submit.setEnabled(available and not self._busy)
self._update_poll_interval(profiles)
def _update_poll_interval(
self, profiles: Optional[List[ExecutionProfile]] = None
) -> None:
"""Use the fastest configured profile interval for local monitoring."""
timer = getattr(self, "_timer", None)
if timer is None:
return
profiles = profiles if profiles is not None else self.manager.profiles.list()
seconds = min(
(profile.poll_seconds for profile in profiles), default=10
)
timer.setInterval(max(2, int(seconds)) * 1000)
def _create_profile(self) -> None:
"""Open an empty profile editor and persist accepted values."""
dialog = ExecutionProfileDialog(self)
if dialog.exec() == QDialog.Accepted:
profile = dialog.profile()
try:
self.manager.profiles.save(profile)
except Exception as exc:
LOG.exception("Could not save execution profile")
self._set_status(
f"Could not save profile: {type(exc).__name__}: {exc}",
error=True,
)
return
self._reload_profiles(profile.name)
def _edit_selected_profile(self) -> None:
"""Edit the selected profile snapshot."""
name = str(self._profile.currentData() or "")
if not name:
return
try:
existing = self.manager.profiles.get(name)
except RemoteExecutionError as exc:
self._set_status(str(exc), error=True)
return
dialog = ExecutionProfileDialog(self, existing)
if dialog.exec() == QDialog.Accepted:
profile = dialog.profile()
try:
self.manager.profiles.save(profile)
if profile.name.casefold() != existing.name.casefold():
self.manager.profiles.delete(existing.name)
except Exception as exc:
LOG.exception("Could not update execution profile")
self._set_status(
f"Could not update profile: {type(exc).__name__}: {exc}",
error=True,
)
return
self._reload_profiles(profile.name)
def _delete_selected_profile(self) -> None:
"""Delete a profile after explicit confirmation."""
name = str(self._profile.currentData() or "")
if not name:
return
answer = QMessageBox.question(
self,
tr("Delete profile"),
tr("Delete execution profile '{name}'?").format(name=name),
)
if answer == QMessageBox.Yes:
try:
self.manager.profiles.delete(name)
except Exception as exc:
LOG.exception("Could not delete execution profile")
self._set_status(
f"Could not delete profile: {type(exc).__name__}: {exc}",
error=True,
)
return
self._reload_profiles()
def _browse_settings(self) -> None:
"""Choose a normal spaCR settings export."""
path, _ = QFileDialog.getOpenFileName(
self,
tr("Choose spaCR settings"),
"",
"spaCR settings (*.csv *.json);;All files (*)",
)
if path:
self._settings_path.setText(path)
def _start_task(self, label: str, operation: Callable, callback: Callable) -> None:
"""Run one blocking manager operation off the GUI thread."""
if self._busy:
self._set_status(tr("Another distributed-job operation is running."))
return
self._busy = True
self._pending_result = None
self._pending_error = ""
self._pending_callback = callback
self._set_busy(True)
self._set_status(label)
def _work(_settings):
"""Run the operation on the job runner's thread."""
try:
self._pending_result = operation()
except Exception as exc:
LOG.exception("Distributed job operation failed")
self._pending_error = f"{type(exc).__name__}: {exc}"
if not self._threaded:
_work({})
self._finish_task(not bool(self._pending_error))
return
thread, worker = make_thread(
_work, {}, app_key="distributed_jobs_io", journal=False
)
self._workers.append((thread, worker))
worker.finished.connect(self._finish_task)
thread.finished.connect(self._retire_workers)
thread.start()
def _finish_task(self, ok: bool) -> None:
"""Apply one worker result on the GUI thread."""
self._busy = False
self._set_busy(False)
if not ok or self._pending_error:
self._set_status(
self._pending_error or tr("Distributed operation failed."),
error=True,
)
return
callback = self._pending_callback
if callback is not None:
callback(self._pending_result)
def _retire_workers(self) -> None:
"""Release ownership pairs whose QThread has stopped.
A bare ``isRunning()`` filter leaked every pair: by the time this
queued slot runs, ``thread.finished -> deleteLater`` has reaped the
QThread's C++ half and ``isRunning()`` raises ``RuntimeError`` out
of the slot, so the assignment never happens. See
:func:`spacr.qt.bridge.prune_job_pairs`.
"""
from ..bridge import prune_job_pairs
self._workers = prune_job_pairs(self._workers, self.sender())
def _set_busy(self, busy: bool) -> None:
"""Update semantic action states while a network call is active."""
for button in (self._submit, self._refresh, self._cancel, self._logs):
button.setEnabled(not busy)
for button in (self._submit, self._refresh):
button.setProperty("buttonActionBusy", bool(busy))
button.style().unpolish(button)
button.style().polish(button)
if not busy:
self._reload_profiles()
[docs]
def submit(self) -> None:
"""Resolve and submit the selected settings file."""
module_name = str(self._module.currentData() or "")
profile_name = str(self._profile.currentData() or "")
settings_path = self._settings_path.text().strip()
if not profile_name:
self._set_status(tr("Create an execution profile first."), error=True)
return
if (
self._settings_snapshot is None
and (not settings_path or not Path(settings_path).is_file())
):
self._set_status(tr("Choose an existing settings CSV/JSON."), error=True)
return
from copy import deepcopy
settings_snapshot = (
deepcopy(self._settings_snapshot)
if self._settings_snapshot is not None else None
)
use_allocated_gpus = (module_name == "mask"
and self._allocated_gpus.isChecked())
def _submit():
"""Resolve the module and its settings, then submit. Off-thread."""
from ...cli import resolve_module, resolve_settings
module = resolve_module(module_name)
if module is None:
raise RemoteExecutionError(
f"Unknown spaCR module: {module_name}"
)
settings = (
settings_snapshot
if settings_snapshot is not None
else resolve_settings(module, settings_path, [])
)
if "model_metadata" in settings:
from ...model_zoo import _model_input_kinds, _validate_model_metadata
_validate_model_metadata(
settings, model_kinds=_model_input_kinds(module, settings))
if use_allocated_gpus:
settings = dict(settings, mask_parallel=True,
mask_gpu_indices="")
return self.manager.submit(module.key, settings, profile_name)
def _done(job: RemoteJob):
"""Redraw the job list with the new job selected."""
self._render_jobs(self.manager.jobs.list(), select=job.job_id)
self._set_status(
tr("Submitted {module} as {job}.").format(
module=job.module, job=job.job_id
)
)
self._start_task(tr("Submitting distributed job…"), _submit, _done)
[docs]
def refresh(self) -> None:
"""Poll all non-terminal jobs in a worker thread."""
def _done(jobs):
"""Redraw the job list and report how many are still active."""
self._render_jobs(jobs)
active = sum(
job.status in ACTIVE_STATES or job.status == "unknown"
for job in jobs
)
self._set_status(
tr("Loaded {count} jobs; {active} active.").format(
count=len(jobs), active=active
)
)
self._start_task(
tr("Polling remote jobs…"),
lambda: self.manager.refresh_all(include_logs=False),
_done,
)
def _selected_job(self) -> Optional[RemoteJob]:
"""Return the job attached to the selected table row."""
row = self._table.currentRow()
if row < 0:
return None
item = self._table.item(row, 0)
job_id = item.data(Qt.UserRole) if item is not None else ""
return next((job for job in self._jobs if job.job_id == job_id), None)
[docs]
def cancel_selected(self) -> None:
"""Request cancellation for the selected active job."""
job = self._selected_job()
if job is None:
self._set_status(tr("Select a job first."), error=True)
return
if job.status not in ACTIVE_STATES and job.status != "unknown":
self._set_status(
tr("That job is already {status}.").format(status=job.status)
)
return
answer = QMessageBox.question(
self,
tr("Cancel job"),
tr("Request cancellation for job {job}?").format(
job=job.job_id[:12]
),
)
if answer != QMessageBox.Yes:
return
def _done(updated):
"""Redraw with the cancelled job still selected.
The selection is kept so the list does not jump to the top after
a cancel and the user's row stays selected.
"""
self._render_jobs(self.manager.jobs.list(), select=updated.job_id)
self._set_status(
tr("Cancellation requested for {job}.").format(
job=updated.job_id
)
)
self._start_task(
tr("Cancelling remote job…"),
lambda: self.manager.cancel(job.job_id),
_done,
)
[docs]
def refresh_log(self) -> None:
"""Fetch the selected job's remote log tail."""
job = self._selected_job()
if job is None:
self._set_status(tr("Select a job first."), error=True)
return
def _done(text):
"""Show the fetched log, re-reading the job in case it moved on."""
updated = self.manager.jobs.get(job.job_id)
self._render_jobs(self.manager.jobs.list(), select=updated.job_id)
self._detail.setPlainText(self._job_detail(updated))
self._set_status(tr("Log refreshed."))
self._start_task(
tr("Retrieving remote log…"),
lambda: self.manager.logs(job.job_id, 500),
_done,
)
def _render_jobs(
self,
jobs: List[RemoteJob],
*,
select: str = "",
) -> None:
"""Render newest-first persistent jobs and restore selection."""
previous = select
current = self._selected_job()
if not previous and current is not None:
previous = current.job_id
self._jobs = list(jobs)
_retire_table_cells(self._table)
palette = active_palette()
colours = {
"success": palette["success"],
"failed": palette["error"],
"cancelled": palette["error"],
"running": palette["accent_hi"],
"queued": palette["warning"],
"pending": palette["warning"],
"unknown": palette["fg_dim"],
}
selected_row = -1
for job in self._jobs:
row = self._table.rowCount()
self._table.insertRow(row)
values = (
job.job_id[:12], job.status, job.module, job.profile_name,
job.external_id or "—", job.created_utc, job.updated_utc,
)
for column, value in enumerate(values):
cell = _spare_table_item(self._table, "" if value is None else str(value), make=_item)
cell.setFlags(cell.flags() & ~Qt.ItemIsEditable)
if column == 0:
cell.setData(Qt.UserRole, job.job_id)
if column == 1:
cell.setForeground(
QColor(colours.get(job.status, palette["fg_dim"]))
)
self._table.setItem(row, column, cell)
if job.job_id == previous:
selected_row = row
if selected_row >= 0:
self._table.selectRow(selected_row)
elif self._jobs:
self._table.selectRow(0)
else:
self._detail.setPlainText(
tr("No distributed jobs have been submitted.")
)
self._show_selection()
@staticmethod
def _job_detail(job: RemoteJob) -> str:
"""Format a complete, copyable job record plus retained log."""
record = job.to_dict()
log = record.pop("log_tail", "")
profile = dict(record.get("profile") or {})
for key in (
"submit_command", "status_command", "cancel_command", "log_command",
):
if profile.get(key):
profile[key] = "<configured command>"
record["profile"] = profile
text = json.dumps(record, indent=2, sort_keys=True, default=str)
if log:
text += "\n\n--- remote log tail ---\n" + log
return text
def _show_selection(self) -> None:
"""Render selected details and update action availability."""
job = self._selected_job()
if job is None:
self._cancel.setEnabled(False)
self._logs.setEnabled(False)
self._open_local.setEnabled(False)
return
self._detail.setPlainText(self._job_detail(job))
self._cancel.setEnabled(
not self._busy
and (job.status in ACTIVE_STATES or job.status == "unknown")
)
self._logs.setEnabled(not self._busy)
self._open_local.setEnabled(True)
def _open_local_record(self) -> None:
"""Open the selected local job record directory."""
job = self._selected_job()
if job is None:
return
QDesktopServices.openUrl(
QUrl.fromLocalFile(str(Path(job.settings_path).parent))
)
def _set_status(self, text: str, error: bool = False) -> None:
"""Set the status line with theme-aware severity."""
self._status.setText(str(text))
palette = active_palette()
self._status.setStyleSheet(
f"color: {palette['error' if error else 'fg_dim']};"
)
[docs]
def showEvent(self, event) -> None: # noqa: N802 - Qt override
"""Poll only while this screen is visible.
:param event: the show event; passed on unchanged to the base class
before polling restarts (when auto-polling is on).
"""
super().showEvent(event)
if self._auto_poll and not self._timer.isActive():
self._timer.start()
[docs]
def hideEvent(self, event) -> None: # noqa: N802 - Qt override
"""Stop background polling when another module is open.
:param event: the hide event; passed on unchanged to the base class.
"""
self._timer.stop()
super().hideEvent(event)
[docs]
def closeEvent(self, event) -> None: # noqa: N802 - Qt override
"""Stop polling and drain live workers before the screen goes away.
``requestInterruption`` alone has no effect — ``_work`` never polls
it — and would leave the screen's REST calls running with no owner. An
ownerless job stays in the process-wide run registry, which is what
``MainWindow.closeEvent`` consults when it decides whether the
application may quit.
:param event: the close event; it is passed on unchanged to the
base-class handler after the workers are drained.
"""
from ..bridge import drain_thread
self._timer.stop()
for thread, worker in list(self._workers):
if worker is not None:
try:
worker.request_cancel("distributed-jobs screen closed")
except Exception:
pass
try:
thread.requestInterruption()
except Exception:
pass
drain_thread(thread, worker, timeout_ms=3000)
self._workers.clear()
super().closeEvent(event)