869 lines
33 KiB
Python
869 lines
33 KiB
Python
"""
|
|
Filename: ssh_remote_plugin.py
|
|
Description: ssh-remote-pack plugin (PyInstaller Single-Binary Edition with
|
|
file-based result transfer)
|
|
|
|
Author: (you)
|
|
License: GPL-3.0
|
|
"""
|
|
|
|
import base64
|
|
import json
|
|
import os
|
|
import pickle
|
|
import shlex
|
|
import tempfile
|
|
import time
|
|
import uuid
|
|
from multiprocessing import Queue
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import paramiko
|
|
|
|
from PySide6.QtCore import QObject, QThread, QTimer, Signal
|
|
from PySide6.QtGui import QAction
|
|
from PySide6.QtWidgets import (
|
|
QApplication, QCheckBox, QFormLayout, QGroupBox, QHBoxLayout, QLabel,
|
|
QLineEdit, QMenu, QMessageBox, QPushButton, QSpinBox, QVBoxLayout, QWidget,
|
|
)
|
|
|
|
from src.shared.shareddata import DATA_SCHEMA
|
|
|
|
|
|
CONFIG_DIR = Path.home() / ".config" / "ssh_remote_pack"
|
|
CONFIG_PATH = CONFIG_DIR / "connection.json"
|
|
|
|
|
|
def load_connection_config() -> dict:
|
|
if CONFIG_PATH.is_file():
|
|
try:
|
|
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
pass
|
|
return {"host": "", "port": 22, "username": "", "remote_executable": ""}
|
|
|
|
|
|
def save_connection_config(host: str, port: int, username: str, remote_executable: str = "") -> None:
|
|
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
|
|
with open(CONFIG_PATH, "w", encoding="utf-8") as f:
|
|
json.dump({
|
|
"host": host,
|
|
"port": port,
|
|
"username": username,
|
|
"remote_executable": remote_executable,
|
|
}, f, indent=2)
|
|
|
|
|
|
class SshSession:
|
|
"""Holds SSH connection details for the current process only."""
|
|
|
|
def __init__(self) -> None:
|
|
self.host: str = ""
|
|
self.port: int = 22
|
|
self.username: str = ""
|
|
self.password: str = ""
|
|
self.remote_executable: str = ""
|
|
|
|
def is_configured(self) -> bool:
|
|
return bool(self.host and self.username and self.password and self.remote_executable)
|
|
|
|
def clear_password(self) -> None:
|
|
self.password = ""
|
|
|
|
|
|
SESSION = SshSession()
|
|
|
|
|
|
def _connect(session: SshSession, timeout: int = 10) -> paramiko.SSHClient:
|
|
client = paramiko.SSHClient()
|
|
client.load_system_host_keys()
|
|
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
client.connect(
|
|
session.host,
|
|
port=session.port,
|
|
username=session.username,
|
|
password=session.password,
|
|
timeout=timeout,
|
|
look_for_keys=False,
|
|
allow_agent=False,
|
|
)
|
|
return client
|
|
|
|
|
|
class _FakeRemoteProcess:
|
|
def __init__(self) -> None:
|
|
self._alive = False
|
|
self.exitcode: Optional[int] = 0
|
|
self.pid: int = 99999
|
|
|
|
def mark_alive(self, value: bool) -> None:
|
|
self._alive = value
|
|
self.exitcode = None if value else 0
|
|
|
|
def is_alive(self) -> bool:
|
|
return self._alive
|
|
|
|
def terminate(self) -> None:
|
|
self._alive = False
|
|
self.exitcode = -15
|
|
|
|
def join(self, timeout: Optional[float] = None) -> None:
|
|
pass
|
|
|
|
|
|
class RemoteTaskWorker(QObject):
|
|
"""Uploads the job to the remote server, launches the PyInstaller
|
|
worker binary over SSH, and streams results back.
|
|
|
|
Small messages (heartbeats, progress ticks, completion, errors) come
|
|
back inline, pickled and base64-encoded, over the SSH channel's stdout.
|
|
Anything carrying a real per-participant result (MNE objects,
|
|
DataFrames, figure bytes, etc.) is written to a file on the remote box
|
|
instead - the worker sends a small pointer message, and this class
|
|
fetches that one file over the still-open SFTP session. That avoids
|
|
both the size overhead of inlining large pickles as base64 text and the
|
|
lossy JSON/str() conversion the old transport used.
|
|
|
|
Everything ends up on `result_queue` - the same queue
|
|
check_for_pipeline_results already polls - including progress ticks,
|
|
which the old version sent to `progress_queue` where nothing in the app
|
|
was actually reading them.
|
|
"""
|
|
|
|
finished = Signal()
|
|
log_signal = Signal(str)
|
|
|
|
def __init__(
|
|
self,
|
|
session: SshSession,
|
|
config: dict,
|
|
result_queue: "Queue",
|
|
progress_queue: Optional["Queue"] = None,
|
|
remote_workdir: str = "/tmp/flares_remote",
|
|
) -> None:
|
|
super().__init__()
|
|
self.session = session
|
|
self.config = config
|
|
self.result_queue = result_queue
|
|
# Kept for backward compatibility with callers that still pass one
|
|
# in, but nothing is routed here anymore - see class docstring.
|
|
self.progress_queue = progress_queue if progress_queue is not None else result_queue
|
|
self.remote_workdir = remote_workdir.rstrip("/")
|
|
self._channel = None
|
|
self._client: Optional[paramiko.SSHClient] = None
|
|
self._sftp = None
|
|
self._remote_pid: Optional[str] = None
|
|
self._remote_job_dir: Optional[str] = None
|
|
self._cancelled = False
|
|
|
|
def cancel(self) -> None:
|
|
self._cancelled = True
|
|
# No process group here (setsid was removed - see run()), so this
|
|
# targets the worker's own PID directly. If it has spawned
|
|
# multiprocessing children by the time this fires, killing the
|
|
# parent won't necessarily reap them immediately; SIGTERM to the
|
|
# main process is still the right first move; escalate manually on
|
|
# the server if orphans turn out to be a real problem.
|
|
if self._remote_pid and self._client is not None:
|
|
try:
|
|
self._client.exec_command(f"kill -TERM {self._remote_pid}")
|
|
except Exception:
|
|
pass
|
|
if self._channel is not None:
|
|
try:
|
|
self._channel.close()
|
|
except Exception:
|
|
pass
|
|
|
|
def run(self) -> None:
|
|
client = None
|
|
try:
|
|
self.log_signal.emit(f"Connecting to {self.session.username}@{self.session.host}:{self.session.port}...")
|
|
client = _connect(self.session)
|
|
self._client = client
|
|
sftp = client.open_sftp()
|
|
self._sftp = sftp
|
|
|
|
job_id = uuid.uuid4().hex[:8]
|
|
remote_job_dir = f"{self.remote_workdir}/{job_id}"
|
|
self._remote_job_dir = remote_job_dir
|
|
self.log_signal.emit(f"Creating remote workdir: {remote_job_dir}")
|
|
self._mkdir_p(sftp, remote_job_dir)
|
|
|
|
snirf_files = self.config.get("SNIRF_FILES", [])
|
|
remote_config = dict(self.config)
|
|
remote_files = []
|
|
path_map: dict[str, str] = {}
|
|
|
|
for i, local_path in enumerate(snirf_files, 1):
|
|
if self._cancelled:
|
|
return
|
|
fname = Path(local_path).name
|
|
remote_path = f"{remote_job_dir}/{fname}"
|
|
self.log_signal.emit(f"SFTP Upload ({i}/{len(snirf_files)}): {fname}")
|
|
sftp.put(local_path, remote_path)
|
|
remote_files.append(remote_path)
|
|
path_map[remote_path] = local_path
|
|
remote_config["SNIRF_FILES"] = remote_files
|
|
# This is what the worker's remap_payload() reads to translate
|
|
# its own abspaths back to paths the GUI recognizes.
|
|
remote_config["PATH_MAP"] = path_map
|
|
|
|
self.log_signal.emit("Uploading config.pkl...")
|
|
with tempfile.NamedTemporaryFile(suffix=".pkl", delete=False) as tmp:
|
|
pickle.dump(remote_config, tmp)
|
|
local_config_path = tmp.name
|
|
remote_config_path = f"{remote_job_dir}/config.pkl"
|
|
sftp.put(local_config_path, remote_config_path)
|
|
os.remove(local_config_path)
|
|
# sftp deliberately stays open (self._sftp) past this point -
|
|
# result files get fetched through it as they're reported.
|
|
|
|
transport = client.get_transport()
|
|
channel = transport.open_session()
|
|
self._channel = channel
|
|
|
|
# $0/$1 keep the actual paths out of the single-quoted script
|
|
# text, avoiding nested-quoting headaches. No setsid here -
|
|
# depending on the util-linux version, setsid can fork and let
|
|
# its immediate parent return almost instantly while the real
|
|
# work continues detached, which reads to the SSH channel as
|
|
# "process exited with code 0" right after the PID line, long
|
|
# before the worker has done anything. exec keeps this shell's
|
|
# PID as the worker's real PID, which is enough for cancel().
|
|
pid_script = 'echo __PID__:$$; exec "$0" "$1"'
|
|
command = (
|
|
f"sh -c {shlex.quote(pid_script)} "
|
|
f"{shlex.quote(self.session.remote_executable)} {shlex.quote(remote_config_path)}"
|
|
)
|
|
self.log_signal.emit(f"Executing: {command}")
|
|
channel.exec_command(command)
|
|
|
|
stdout_buf = ""
|
|
stderr_buf = ""
|
|
|
|
while True:
|
|
if self._cancelled:
|
|
break
|
|
|
|
if channel.recv_ready():
|
|
chunk = channel.recv(4096).decode("utf-8", errors="replace")
|
|
stdout_buf += chunk
|
|
while "\n" in stdout_buf:
|
|
line, stdout_buf = stdout_buf.split("\n", 1)
|
|
self._handle_line(line)
|
|
|
|
if channel.recv_stderr_ready():
|
|
chunk = channel.recv_stderr(4096).decode("utf-8", errors="replace")
|
|
stderr_buf += chunk
|
|
while "\n" in stderr_buf:
|
|
line, stderr_buf = stderr_buf.split("\n", 1)
|
|
self.log_signal.emit(f"[Remote Stderr] {line}")
|
|
|
|
if channel.exit_status_ready() and not channel.recv_ready() and not channel.recv_stderr_ready():
|
|
if stdout_buf:
|
|
self._handle_line(stdout_buf)
|
|
if stderr_buf:
|
|
self.log_signal.emit(f"[Remote Stderr] {stderr_buf}")
|
|
break
|
|
|
|
time.sleep(0.05)
|
|
|
|
if not self._cancelled:
|
|
exit_status = channel.recv_exit_status()
|
|
self.log_signal.emit(f"Remote process completed with exit code: {exit_status}")
|
|
if exit_status != 0:
|
|
self.result_queue.put({
|
|
"success": False,
|
|
"error": f"Remote executable exited with code {exit_status}.",
|
|
})
|
|
|
|
except Exception as e:
|
|
self.log_signal.emit(f"Remote execution error: {e}")
|
|
self.result_queue.put({
|
|
"success": False,
|
|
"error": f"Remote execution failed: {e}",
|
|
})
|
|
finally:
|
|
# Best-effort cleanup: the uploaded SNIRFs, config.pkl, and any
|
|
# result files that weren't already removed after download all
|
|
# live under this one directory, so a single rm -rf covers it.
|
|
# Runs regardless of how the job ended (success, failure, or
|
|
# cancel) as long as we still have a connection to use.
|
|
if self._remote_job_dir and client is not None:
|
|
try:
|
|
self.log_signal.emit(f"Cleaning up remote workdir: {self._remote_job_dir}")
|
|
_, cleanup_stdout, _ = client.exec_command(
|
|
f"rm -rf {shlex.quote(self._remote_job_dir)}"
|
|
)
|
|
cleanup_stdout.channel.recv_exit_status() # wait so this finishes before we close the connection
|
|
except Exception as e:
|
|
self.log_signal.emit(f"Non-fatal: failed to clean up remote workdir: {e}")
|
|
if self._sftp is not None:
|
|
try:
|
|
self._sftp.close()
|
|
except Exception:
|
|
pass
|
|
if client is not None:
|
|
try:
|
|
client.close()
|
|
except Exception as e:
|
|
self.log_signal.emit(f"Non-fatal error closing SSH connection: {e}")
|
|
self.log_signal.emit("Worker execution finished.")
|
|
self.finished.emit()
|
|
|
|
def _handle_line(self, line: str) -> None:
|
|
clean_line = line.strip()
|
|
if not clean_line:
|
|
return
|
|
|
|
if clean_line.startswith("__PID__:"):
|
|
self._remote_pid = clean_line.split(":", 1)[1].strip()
|
|
self.log_signal.emit(f"Remote PID: {self._remote_pid} (process group leader)")
|
|
return
|
|
|
|
if clean_line.startswith("FLARES_B64:"):
|
|
payload = clean_line[len("FLARES_B64:"):].strip()
|
|
try:
|
|
msg = pickle.loads(base64.b64decode(payload))
|
|
except Exception as e:
|
|
self.log_signal.emit(f"Failed to unpickle FLARES_B64 payload: {e}")
|
|
return
|
|
self._route_message(msg)
|
|
return
|
|
|
|
if clean_line.startswith("FLARES_RESULT_FILE:"):
|
|
payload = clean_line[len("FLARES_RESULT_FILE:"):].strip()
|
|
try:
|
|
meta = pickle.loads(base64.b64decode(payload))
|
|
except Exception as e:
|
|
self.log_signal.emit(f"Failed to unpickle result-file metadata: {e}")
|
|
return
|
|
self._fetch_and_route_result(meta)
|
|
return
|
|
|
|
if clean_line.startswith("FLARES_REMOTE_ERROR:"):
|
|
err_msg = clean_line[len("FLARES_REMOTE_ERROR:"):].strip()
|
|
self.log_signal.emit(f"[Remote Error] {err_msg}")
|
|
# This used to only be logged, never reaching the GUI as an
|
|
# actual message - only the generic exit-code fallback did.
|
|
self.result_queue.put({"success": False, "error": err_msg})
|
|
return
|
|
|
|
self.log_signal.emit(f"[Remote Output] {clean_line}")
|
|
|
|
def _route_message(self, msg) -> None:
|
|
"""Puts a fully-materialized message onto result_queue, in the
|
|
shape check_for_pipeline_results already expects."""
|
|
if isinstance(msg, dict) and msg.get("type") == "progress":
|
|
self.result_queue.put(("progress", msg.get("file"), msg.get("step")))
|
|
return
|
|
self.result_queue.put(msg)
|
|
|
|
def _fetch_and_route_result(self, meta: dict) -> None:
|
|
"""meta is the message minus its "result" key, plus a
|
|
"result_path" pointing at the pickle file the worker wrote on the
|
|
remote box. Downloads it, unpickles it, reattaches it as "result",
|
|
and routes the completed message like any other."""
|
|
remote_result_path = meta.pop("result_path", None)
|
|
if not remote_result_path or self._sftp is None:
|
|
self.log_signal.emit("Result-file message missing a path or SFTP session - dropping the result payload.")
|
|
self._route_message(meta)
|
|
return
|
|
|
|
local_tmp_path = None
|
|
try:
|
|
with tempfile.NamedTemporaryFile(suffix=".pkl", delete=False) as tmp:
|
|
local_tmp_path = tmp.name
|
|
self.log_signal.emit(f"SFTP Download: {os.path.basename(remote_result_path)}")
|
|
self._sftp.get(remote_result_path, local_tmp_path)
|
|
with open(local_tmp_path, "rb") as f:
|
|
meta["result"] = pickle.load(f)
|
|
try:
|
|
self._sftp.remove(remote_result_path)
|
|
except Exception:
|
|
pass # best-effort cleanup, not worth failing the run over
|
|
except Exception as e:
|
|
self.log_signal.emit(f"Failed to download result file: {e}")
|
|
meta["success"] = False
|
|
meta["error"] = f"Failed to download remote result: {e}"
|
|
finally:
|
|
if local_tmp_path and os.path.exists(local_tmp_path):
|
|
try:
|
|
os.remove(local_tmp_path)
|
|
except Exception:
|
|
pass
|
|
|
|
self._route_message(meta)
|
|
|
|
@staticmethod
|
|
def _mkdir_p(sftp, remote_dir: str) -> None:
|
|
parts = remote_dir.strip("/").split("/")
|
|
path = ""
|
|
for part in parts:
|
|
path += "/" + part
|
|
try:
|
|
sftp.mkdir(path)
|
|
except IOError:
|
|
pass
|
|
|
|
|
|
class RemoteExecutionController(QObject):
|
|
def __init__(self, main_window: QWidget) -> None:
|
|
super().__init__()
|
|
self.main_window = main_window
|
|
self._active = False
|
|
self._original_clicked_slot = None
|
|
self._original_text: Optional[str] = None
|
|
self._thread: Optional[QThread] = None
|
|
self._worker: Optional[RemoteTaskWorker] = None
|
|
|
|
@property
|
|
def active(self) -> bool:
|
|
return self._active
|
|
|
|
def enable(self) -> None:
|
|
mw = self.main_window
|
|
if self._active:
|
|
return
|
|
if not hasattr(mw, "button1"):
|
|
raise RuntimeError("Main window has no 'button1' to hook into.")
|
|
if mw.button1.text() != "Process":
|
|
raise RuntimeError("Can't switch to remote mode while a task is running.")
|
|
|
|
self._original_text = mw.button1.text()
|
|
try:
|
|
mw.button1.clicked.disconnect(mw.on_run_task)
|
|
self._original_clicked_slot = mw.on_run_task
|
|
except (TypeError, RuntimeError):
|
|
self._original_clicked_slot = None
|
|
|
|
mw.button1.clicked.connect(self._on_run_task_remote)
|
|
mw.button1.setText("Process (Remote)")
|
|
self._active = True
|
|
|
|
def disable(self) -> None:
|
|
mw = self.main_window
|
|
if not self._active:
|
|
return
|
|
try:
|
|
mw.button1.clicked.disconnect(self._on_run_task_remote)
|
|
except (TypeError, RuntimeError):
|
|
pass
|
|
if self._original_clicked_slot is not None:
|
|
mw.button1.clicked.connect(self._original_clicked_slot)
|
|
mw.button1.setText(self._original_text or "Process")
|
|
self._active = False
|
|
|
|
def _on_run_task_remote(self) -> None:
|
|
mw = self.main_window
|
|
|
|
if not SESSION.is_configured():
|
|
QMessageBox.warning(mw, "Not Connected", "Configure and test the SSH connection first.")
|
|
return
|
|
|
|
if not getattr(mw, "analysis_clearing_bypass", False):
|
|
if mw.button3.isVisible():
|
|
msg = QMessageBox(mw)
|
|
msg.setWindowTitle("Confirm - FLARES")
|
|
msg.setText(
|
|
"Processing new data will clear the current analysis and close all "
|
|
"other windows. Continue? (If you do not want this dialog box to "
|
|
"appear, toggle 'Analysis Clearing Bypass' from the Preferences menu.)"
|
|
)
|
|
msg.setStandardButtons(QMessageBox.StandardButton.Ok | QMessageBox.StandardButton.Cancel)
|
|
msg.setDefaultButton(QMessageBox.StandardButton.Cancel)
|
|
if msg.exec() != QMessageBox.StandardButton.Ok:
|
|
return
|
|
|
|
mw.button3.setVisible(False)
|
|
|
|
for item in DATA_SCHEMA:
|
|
setattr(mw, item["key"], {})
|
|
|
|
for bubble in mw.bubble_widgets.values():
|
|
bubble.reset()
|
|
|
|
for widget in QApplication.topLevelWidgets():
|
|
if widget is not mw and widget.isVisible():
|
|
widget.close()
|
|
|
|
try:
|
|
mw.button1.clicked.disconnect(self._on_run_task_remote)
|
|
except (TypeError, RuntimeError):
|
|
pass
|
|
mw.button1.setText("Cancel")
|
|
mw.button1.clicked.connect(self._cancel_remote_task)
|
|
|
|
mw.first_run = False
|
|
|
|
snirf_files: list[str] = []
|
|
if getattr(mw, "selected_paths", None):
|
|
for path in mw.selected_paths:
|
|
p = Path(path)
|
|
if p.is_dir():
|
|
snirf_files += [str(f) for f in p.glob("*.snirf")]
|
|
elif p.is_file() and p.suffix == ".snirf":
|
|
snirf_files.append(str(p))
|
|
elif getattr(mw, "selected_path", None):
|
|
p = Path(mw.selected_path)
|
|
if p.is_dir():
|
|
snirf_files += [str(f) for f in p.glob("*.snirf")]
|
|
elif p.is_file() and p.suffix == ".snirf":
|
|
snirf_files.append(str(p))
|
|
else:
|
|
QMessageBox.critical(mw, "Error", "No file(s) selected")
|
|
self._reset_button_to_idle()
|
|
return
|
|
|
|
if not snirf_files:
|
|
QMessageBox.critical(mw, "Error", "No .snirf files found in selection")
|
|
self._reset_button_to_idle()
|
|
return
|
|
|
|
mw.files_total = len(snirf_files)
|
|
mw.files_done = set()
|
|
mw.files_failed = set()
|
|
mw.files_results = {}
|
|
|
|
all_params = {}
|
|
for section_widget in mw.param_sections:
|
|
all_params.update(section_widget.get_param_values())
|
|
if getattr(mw, "folding_bypass", False):
|
|
all_params["FOLDING_BYP"] = True
|
|
|
|
collected_data = {
|
|
"SNIRF_FILES": snirf_files,
|
|
"PARAMS": all_params,
|
|
"METADATA": mw.get_all_metadata(),
|
|
}
|
|
|
|
mw.result_queue = Queue()
|
|
mw.progress_queue = Queue()
|
|
mw.ack_queue = Queue()
|
|
mw.result_process = _FakeRemoteProcess()
|
|
mw.result_process.mark_alive(True)
|
|
|
|
self._thread = QThread()
|
|
self._worker = RemoteTaskWorker(
|
|
SESSION,
|
|
collected_data,
|
|
result_queue=mw.result_queue,
|
|
progress_queue=mw.progress_queue,
|
|
)
|
|
self._worker.moveToThread(self._thread)
|
|
|
|
self._worker.log_signal.connect(mw.statusbar.showMessage)
|
|
self._worker.log_signal.connect(lambda msg: print(f"[SSH Plugin] {msg}"))
|
|
|
|
self._thread.started.connect(self._worker.run)
|
|
self._worker.finished.connect(self._on_remote_task_finished)
|
|
self._worker.finished.connect(self._thread.quit)
|
|
self._worker.finished.connect(self._worker.deleteLater)
|
|
self._thread.finished.connect(self._thread.deleteLater)
|
|
|
|
self._thread.start()
|
|
|
|
mw.result_timer = QTimer()
|
|
mw.result_timer.timeout.connect(mw.check_for_pipeline_results)
|
|
mw.result_timer.start()
|
|
|
|
def _on_remote_task_finished(self) -> None:
|
|
mw = self.main_window
|
|
if hasattr(mw, "result_process") and mw.result_process is not None:
|
|
mw.result_process.mark_alive(False)
|
|
|
|
QTimer.singleShot(300, self._finalize_remote_ui)
|
|
|
|
def _finalize_remote_ui(self) -> None:
|
|
"""On a normal completion, the real FINISHED_SUCCESSFULLY message
|
|
from flares.gui_entry() now flows through result_queue exactly like
|
|
a local run, so check_for_pipeline_results (untouched app code)
|
|
already stopped the timer, revealed button3, and reset button1 -
|
|
we just need to put button1 back into remote mode, since that
|
|
handler unconditionally reconnects it to local on_run_task.
|
|
|
|
The one case that handler doesn't cover: a fatal-crash message
|
|
(success=False with no "type") stops the timer but never resets
|
|
button1 off "Cancel" - catch that here as a safety net.
|
|
"""
|
|
mw = self.main_window
|
|
|
|
if mw.button1.text() == "Cancel":
|
|
try:
|
|
mw.button1.clicked.disconnect()
|
|
except (TypeError, RuntimeError):
|
|
pass
|
|
mw.button1.clicked.connect(self._on_run_task_remote)
|
|
mw.button1.setText("Process (Remote)")
|
|
else:
|
|
self._reapply_hijack_if_needed()
|
|
|
|
mw.statusbar.showMessage("Remote execution completed.")
|
|
|
|
def _cancel_remote_task(self) -> None:
|
|
mw = self.main_window
|
|
if self._worker is not None:
|
|
self._worker.cancel()
|
|
mw.statusbar.showMessage("Cancelling remote task...")
|
|
self._reset_button_to_idle()
|
|
|
|
def _reset_button_to_idle(self) -> None:
|
|
mw = self.main_window
|
|
try:
|
|
mw.button1.clicked.disconnect()
|
|
except (TypeError, RuntimeError):
|
|
pass
|
|
if self._active:
|
|
mw.button1.setText("Process (Remote)")
|
|
mw.button1.clicked.connect(self._on_run_task_remote)
|
|
else:
|
|
mw.button1.setText("Process")
|
|
if self._original_clicked_slot is not None:
|
|
mw.button1.clicked.connect(self._original_clicked_slot)
|
|
|
|
def _reapply_hijack_if_needed(self) -> None:
|
|
if not self._active:
|
|
return
|
|
mw = self.main_window
|
|
try:
|
|
mw.button1.clicked.disconnect(mw.on_run_task)
|
|
except (TypeError, RuntimeError):
|
|
pass
|
|
try:
|
|
mw.button1.clicked.disconnect(self._on_run_task_remote)
|
|
except (TypeError, RuntimeError):
|
|
pass
|
|
mw.button1.clicked.connect(self._on_run_task_remote)
|
|
mw.button1.setText("Process (Remote)")
|
|
|
|
|
|
class ConnectionTestWorker(QObject):
|
|
finished = Signal()
|
|
success = Signal(str)
|
|
failure = Signal(str)
|
|
|
|
def __init__(self, session: SshSession) -> None:
|
|
super().__init__()
|
|
self.session = session
|
|
|
|
def run(self) -> None:
|
|
try:
|
|
client = _connect(self.session)
|
|
try:
|
|
transport = client.get_transport()
|
|
banner = transport.remote_version if transport else "connected"
|
|
self.success.emit(banner)
|
|
finally:
|
|
client.close()
|
|
except Exception as e:
|
|
self.failure.emit(str(e))
|
|
finally:
|
|
self.finished.emit()
|
|
|
|
|
|
class SshRemoteSettingsWidget(QWidget):
|
|
def __init__(self, controller: Optional[RemoteExecutionController] = None, parent: QWidget | None = None) -> None:
|
|
super().__init__(parent)
|
|
self.controller = controller
|
|
self.setWindowTitle("SSH Remote Processing Settings")
|
|
self.resize(460, 260)
|
|
|
|
self._thread: QThread | None = None
|
|
self._worker: QObject | None = None
|
|
|
|
self._setup_ui()
|
|
self._load_existing_config()
|
|
|
|
def _setup_ui(self) -> None:
|
|
main_layout = QVBoxLayout(self)
|
|
|
|
conn_box = QGroupBox("Remote Server", self)
|
|
form = QFormLayout(conn_box)
|
|
|
|
self.txt_host = QLineEdit(self)
|
|
self.txt_host.setPlaceholderText("e.g. 192.168.1.50 or myserver.example.com")
|
|
|
|
self.spin_port = QSpinBox(self)
|
|
self.spin_port.setRange(1, 65535)
|
|
self.spin_port.setValue(22)
|
|
|
|
self.txt_username = QLineEdit(self)
|
|
self.txt_username.setPlaceholderText("remote username")
|
|
|
|
self.txt_password = QLineEdit(self)
|
|
self.txt_password.setEchoMode(QLineEdit.EchoMode.Password)
|
|
self.txt_password.setPlaceholderText("required each app session - never saved to disk")
|
|
|
|
self.txt_remote_executable = QLineEdit(self)
|
|
self.txt_remote_executable.setPlaceholderText("e.g. /usr/local/bin/flares_worker or /home/user/flares_bin")
|
|
|
|
form.addRow("Host / IP:", self.txt_host)
|
|
form.addRow("Port:", self.spin_port)
|
|
form.addRow("Username:", self.txt_username)
|
|
form.addRow("Password:", self.txt_password)
|
|
form.addRow("Remote Executable Path:", self.txt_remote_executable)
|
|
|
|
main_layout.addWidget(conn_box)
|
|
|
|
btn_layout = QHBoxLayout()
|
|
self.btn_apply = QPushButton("Save && Apply", self)
|
|
self.btn_apply.clicked.connect(self._on_apply)
|
|
|
|
self.btn_test = QPushButton("Test Connection", self)
|
|
self.btn_test.clicked.connect(self._on_test_connection)
|
|
|
|
self.btn_forget = QPushButton("Forget Password", self)
|
|
self.btn_forget.clicked.connect(self._on_forget_password)
|
|
|
|
btn_layout.addWidget(self.btn_apply)
|
|
btn_layout.addWidget(self.btn_test)
|
|
btn_layout.addWidget(self.btn_forget)
|
|
main_layout.addLayout(btn_layout)
|
|
|
|
self.chk_remote = QCheckBox("Route processing (button1) through this remote server", self)
|
|
self.chk_remote.toggled.connect(self._on_toggle_remote)
|
|
main_layout.addWidget(self.chk_remote)
|
|
|
|
self.lbl_status = QLabel("", self)
|
|
self.lbl_status.setWordWrap(True)
|
|
main_layout.addWidget(self.lbl_status)
|
|
main_layout.addStretch()
|
|
|
|
def _load_existing_config(self) -> None:
|
|
cfg = load_connection_config()
|
|
self.txt_host.setText(cfg.get("host", ""))
|
|
self.spin_port.setValue(int(cfg.get("port", 22)))
|
|
self.txt_username.setText(cfg.get("username", ""))
|
|
self.txt_remote_executable.setText(cfg.get("remote_executable", ""))
|
|
SESSION.remote_executable = cfg.get("remote_executable", "")
|
|
if self.controller is not None:
|
|
self.chk_remote.setChecked(self.controller.active)
|
|
|
|
def _on_apply(self) -> None:
|
|
host = self.txt_host.text().strip()
|
|
port = self.spin_port.value()
|
|
username = self.txt_username.text().strip()
|
|
password = self.txt_password.text()
|
|
remote_executable = self.txt_remote_executable.text().strip()
|
|
|
|
if not host or not username:
|
|
QMessageBox.warning(self, "Missing Fields", "Host and username are required.")
|
|
return
|
|
if not remote_executable:
|
|
QMessageBox.warning(self, "Missing Fields", "Remote Executable Path is required.")
|
|
return
|
|
if not password and not SESSION.password:
|
|
QMessageBox.warning(self, "Missing Password", "Enter the password for this session.")
|
|
return
|
|
|
|
save_connection_config(host, port, username, remote_executable)
|
|
|
|
SESSION.host = host
|
|
SESSION.port = port
|
|
SESSION.username = username
|
|
SESSION.remote_executable = remote_executable
|
|
if password:
|
|
SESSION.password = password
|
|
|
|
self.lbl_status.setText(f"Applied settings for {username}@{host}:{port} (password held in memory only).")
|
|
|
|
def _on_forget_password(self) -> None:
|
|
SESSION.clear_password()
|
|
self.txt_password.clear()
|
|
if self.controller is not None and self.controller.active:
|
|
self.controller.disable()
|
|
self.chk_remote.setChecked(False)
|
|
self.lbl_status.setText("Password cleared from memory.")
|
|
|
|
def _on_test_connection(self) -> None:
|
|
self._on_apply()
|
|
if not SESSION.is_configured():
|
|
return
|
|
|
|
self.btn_test.setEnabled(False)
|
|
self.lbl_status.setText("Connecting...")
|
|
|
|
self._thread = QThread()
|
|
self._worker = ConnectionTestWorker(SESSION)
|
|
self._worker.moveToThread(self._thread)
|
|
|
|
self._thread.started.connect(self._worker.run)
|
|
self._worker.success.connect(self._on_test_success)
|
|
self._worker.failure.connect(self._on_test_failure)
|
|
self._worker.finished.connect(self._thread.quit)
|
|
self._worker.finished.connect(self._worker.deleteLater)
|
|
self._thread.finished.connect(self._thread.deleteLater)
|
|
self._thread.finished.connect(lambda: self.btn_test.setEnabled(True))
|
|
|
|
self._thread.start()
|
|
|
|
def _on_test_success(self, banner: str) -> None:
|
|
self.lbl_status.setText(f"Connected successfully. Server: {banner}")
|
|
|
|
def _on_test_failure(self, error: str) -> None:
|
|
self.lbl_status.setText(f"Connection failed: {error}")
|
|
QMessageBox.critical(self, "Connection Failed", error)
|
|
|
|
def _on_toggle_remote(self, checked: bool) -> None:
|
|
if self.controller is None:
|
|
return
|
|
if checked:
|
|
if not SESSION.is_configured():
|
|
QMessageBox.warning(self, "Not Connected", "Save settings and test the connection first.")
|
|
self.chk_remote.setChecked(False)
|
|
return
|
|
try:
|
|
self.controller.enable()
|
|
self.lbl_status.setText("Remote processing is ON - button1 now runs jobs on the remote server.")
|
|
except Exception as e:
|
|
QMessageBox.critical(self, "Couldn't Enable", str(e))
|
|
self.chk_remote.setChecked(False)
|
|
else:
|
|
self.controller.disable()
|
|
self.lbl_status.setText("Remote processing is OFF - button1 runs locally again.")
|
|
|
|
|
|
class Plugin:
|
|
def __init__(self, main_window: QWidget) -> None:
|
|
self.main_window = main_window
|
|
self.name = "SSH Remote Processing"
|
|
self.widget_instance: SshRemoteSettingsWidget | None = None
|
|
self.controller = RemoteExecutionController(main_window)
|
|
|
|
def register_menu(self, plugin_menu: QMenu) -> None:
|
|
open_action = QAction("Configure SSH Remote...", self.main_window)
|
|
open_action.triggered.connect(self.show_widget)
|
|
|
|
about_action = QAction("About SSH Remote Processing", self.main_window)
|
|
about_action.triggered.connect(self.show_about)
|
|
|
|
plugin_menu.addAction(open_action)
|
|
plugin_menu.addAction(about_action)
|
|
|
|
def show_widget(self) -> None:
|
|
if self.widget_instance is None or not self.widget_instance.isVisible():
|
|
self.widget_instance = SshRemoteSettingsWidget(controller=self.controller)
|
|
self.widget_instance.show()
|
|
else:
|
|
self.widget_instance.raise_()
|
|
self.widget_instance.activateWindow()
|
|
|
|
def show_about(self) -> None:
|
|
title = "SSH Remote Processing"
|
|
version = "0.3"
|
|
author = "Unknown"
|
|
description = (
|
|
"Runs the app's flares processing job on a remote server using a "
|
|
"standalone PyInstaller executable over SSH, and streams results "
|
|
"back - small messages inline, per-participant results via SFTP."
|
|
)
|
|
|
|
message = (
|
|
f"<b>{title}</b><br>"
|
|
f"<b>Version:</b> {version} | <b>Author:</b> {author}<br><br>"
|
|
f"{description}"
|
|
)
|
|
QMessageBox.about(self.main_window, f"About {title}", message) |