cleanup what I just did

This commit is contained in:
2026-06-28 09:09:06 -07:00
parent 617daa0e9b
commit f10eb64332
16 changed files with 77 additions and 64 deletions
@@ -1,21 +1,23 @@
import os
from PySide6.QtWidgets import QComboBox, QDialog, QGridLayout, QHBoxLayout, QMessageBox, QPushButton, QScrollArea, QWidget, QVBoxLayout, QLabel
from PySide6.QtWidgets import QComboBox, QDialog, QGridLayout, QHBoxLayout, QPushButton, QScrollArea, QWidget, QVBoxLayout, QLabel, QMessageBox
from PySide6.QtCore import QSize
from src.shared.s_flaresbasewidget import FlaresBaseWidget, ParameterInputDialog
from src.shared.flaresbasewidget import FlaresBaseWidget, ParameterInputDialog
from src.shared.shareddata import APP_NAME
class GroupFunctionalConnectivityWidget(FlaresBaseWidget):
def __init__(self, haemo_dict, group, config_dict):
super().__init__("GroupFunctionalConnectivityWidget")
self.setWindowTitle("FLARES Group Viewer")
self.setWindowTitle(f"{APP_NAME.upper} Group Viewer")
self.haemo_dict = haemo_dict
self.group = group
self.config_dict = config_dict
self.show_all_events = True
self._updating_checkstates = False
QMessageBox.warning(self, "Warning - FLARES", f"Functional Connectivity is still in development and the results should currently be taken with a grain of salt. "
QMessageBox.warning(self, f"Warning - {APP_NAME.upper}", f"Functional Connectivity is still in development and the results should currently be taken with a grain of salt. "
"By clicking OK, you accept that the images generated may not be factual.")
@@ -87,7 +89,7 @@ class GroupFunctionalConnectivityWidget(FlaresBaseWidget):
def show_brain_images(self):
import src.flares as flares
import flares as flares
selected_event = self.event_dropdown.currentText()
if selected_event == "<None Selected>":
@@ -2,18 +2,19 @@ import os
from pathlib import Path
from datetime import datetime
from PySide6.QtWidgets import QComboBox, QFrame, QGridLayout, QHBoxLayout, QListView, QMessageBox, QPushButton, QScrollArea, QSpinBox, QWidget, QVBoxLayout, QLabel, QLineEdit
from PySide6.QtCore import QThread, Signal, Qt, QTimer, QEvent, QSize, QPoint, QUrl
from PySide6.QtGui import QAction, QDesktopServices, QKeySequence, QIcon, QIntValidator, QDoubleValidator, QPixmap, QStandardItemModel, QStandardItem, QImage
from PySide6.QtWidgets import QGridLayout, QHBoxLayout, QMessageBox, QPushButton, QScrollArea, QWidget, QVBoxLayout, QLabel
from PySide6.QtCore import Qt, QSize
from PySide6.QtGui import QPixmap
from src.shared.s_flaresbasewidget import ClickableLabel, FlaresBaseWidget
from src.shared.flaresbasewidget import ClickableLabel, FlaresBaseWidget
from src.shared.shareddata import APP_NAME
class ParticipantViewerWidget(FlaresBaseWidget):
def __init__(self, haemo_dict, fig_bytes_dict):
super().__init__("ParticipantViewer")
self.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose)
self.setWindowTitle("FLARES Participant Viewer")
self.setWindowTitle(f"{APP_NAME.upper} Participant Viewer")
self.haemo_dict = haemo_dict
self.fig_bytes_dict = fig_bytes_dict
@@ -1,15 +1,16 @@
import os
from PySide6.QtWidgets import QComboBox, QDialog, QFrame, QGridLayout, QHBoxLayout, QListView, QMessageBox, QPushButton, QScrollArea, QSpinBox, QWidget, QVBoxLayout, QLabel, QLineEdit
from PySide6.QtCore import QThread, Signal, Qt, QTimer, QEvent, QSize, QPoint, QUrl
from PySide6.QtWidgets import QComboBox, QDialog, QGridLayout, QHBoxLayout, QPushButton, QScrollArea, QWidget, QVBoxLayout, QLabel
from PySide6.QtCore import QSize
from src.shared.s_flaresbasewidget import FlaresBaseWidget, ParameterInputDialog
from src.shared.flaresbasewidget import FlaresBaseWidget, ParameterInputDialog
from src.shared.shareddata import APP_NAME
class ParticipantBrainViewerWidget(FlaresBaseWidget):
def __init__(self, haemo_dict, cha_dict):
super().__init__("ParticipantBrainViewer")
self.setWindowTitle("FLARES Participant Brain Viewer")
self.setWindowTitle(f"{APP_NAME.upper} Participant Brain Viewer")
self.haemo_dict = haemo_dict
self.cha_dict = cha_dict
@@ -67,7 +68,7 @@ class ParticipantBrainViewerWidget(FlaresBaseWidget):
def show_brain_images(self):
import src.flares as flares
import flares as flares
selected_event = self.event_dropdown.currentText()
if selected_event == "<None Selected>":
@@ -1,18 +1,20 @@
import os
from PySide6.QtWidgets import QComboBox, QDialog, QFrame, QGridLayout, QHBoxLayout, QListView, QMessageBox, QPushButton, QScrollArea, QSpinBox, QWidget, QVBoxLayout, QLabel, QLineEdit
from PySide6.QtCore import QThread, Signal, Qt, QTimer, QEvent, QSize, QPoint, QUrl
from PySide6.QtWidgets import QComboBox, QDialog, QGridLayout, QHBoxLayout, QPushButton, QScrollArea, QWidget, QVBoxLayout, QLabel, QMessageBox
from PySide6.QtCore import QSize
from src.shared.flaresbasewidget import FlaresBaseWidget, ParameterInputDialog
from src.shared.shareddata import APP_NAME
from src.shared.s_flaresbasewidget import FlaresBaseWidget, ParameterInputDialog
class ParticipantFunctionalConnectivityWidget(FlaresBaseWidget):
def __init__(self, haemo_dict, epochs_dict):
super().__init__("FunctionalConnectivityWidget")
self.setWindowTitle("FLARES Functional Connectivity Viewer [BETA]")
self.setWindowTitle(f"{APP_NAME.upper} Functional Connectivity Viewer [BETA]")
self.haemo_dict = haemo_dict
self.epochs_dict = epochs_dict
QMessageBox.warning(self, "Warning - FLARES", f"Functional Connectivity is still in development and the results should currently be taken with a grain of salt. "
QMessageBox.warning(self, f"Warning - {APP_NAME.upper}", f"Functional Connectivity is still in development and the results should currently be taken with a grain of salt. "
"By clicking OK, you accept that the images generated may not be factual.")
# Create mappings: file_path -> participant label and dropdown display text
@@ -69,7 +71,7 @@ class ParticipantFunctionalConnectivityWidget(FlaresBaseWidget):
def show_brain_images(self):
import src.flares as flares
import flares as flares
selected_event = self.event_dropdown.currentText()
if selected_event == "<None Selected>":
-4763
View File
File diff suppressed because it is too large Load Diff
-256
View File
@@ -1,256 +0,0 @@
"""
Filename: flares_updater.py
Description: FLARES updater executable
Author: Tyler de Zeeuw
License: GPL-3.0
"""
# Built-in imports
import os
import sys
import time
import shlex
import psutil
import shutil
import platform
import subprocess
from datetime import datetime
PLATFORM_NAME = platform.system().lower()
APP_NAME = "flares"
if PLATFORM_NAME == 'darwin':
LOG_FILE = os.path.join(os.path.dirname(sys.executable), f"../../../{APP_NAME}_updater.log")
else:
LOG_FILE = os.path.join(os.getcwd(), f"{APP_NAME}_updater.log")
def log(msg):
with open(LOG_FILE, "a", encoding="utf-8") as f:
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
f.write(f"{timestamp} - {msg}\n")
def kill_all_processes_by_executable(exe_path):
terminated_any = False
exe_path = os.path.realpath(exe_path)
if PLATFORM_NAME == 'windows':
for proc in psutil.process_iter(['pid', 'exe']):
try:
proc_exe = proc.info.get('exe')
if proc_exe and os.path.samefile(os.path.realpath(proc_exe), exe_path):
log(f"Terminating process: PID {proc.pid}")
_terminate_process(proc)
terminated_any = True
except Exception as e:
log(f"Error terminating process (Windows): {e}")
elif PLATFORM_NAME == 'linux':
for proc in psutil.process_iter(['pid', 'cmdline']):
try:
cmdline = proc.info.get('cmdline', [])
if cmdline:
proc_cmd = os.path.realpath(cmdline[0])
if os.path.samefile(proc_cmd, exe_path):
log(f"Terminating process: PID {proc.pid}")
_terminate_process(proc)
terminated_any = True
except Exception as e:
log(f"Error terminating process (Linux): {e}")
if not terminated_any:
log(f"No running processes found for {exe_path}")
return terminated_any
def _terminate_process(proc):
try:
proc.terminate()
proc.wait(timeout=10)
log(f"Process {proc.pid} terminated gracefully.")
except psutil.TimeoutExpired:
log(f"Process {proc.pid} did not terminate in time. Killing forcefully.")
proc.kill()
proc.wait(timeout=5)
log(f"Process {proc.pid} killed.")
def wait_for_unlock(path, timeout=100):
start_time = time.time()
while time.time() - start_time < timeout:
try:
if os.path.isdir(path):
shutil.rmtree(path)
else:
os.remove(path)
log(f"Deleted (after wait): {path}")
return
except Exception as e:
log(f"Still locked: {path} - {e}")
time.sleep(1)
log(f"Failed to delete after wait: {path}")
def delete_path(path):
if os.path.exists(path):
try:
if os.path.isdir(path):
shutil.rmtree(path)
log(f"Deleted directory: {path}")
else:
os.remove(path)
log(f"Deleted file: {path}")
except Exception as e:
log(f"Error deleting {path}: {e}")
def copy_update_files(src_folder, dest_folder, updater_name):
for item in os.listdir(src_folder):
if item.lower() == updater_name.lower():
log(f"Skipping updater executable: {item}")
continue
s = os.path.join(src_folder, item)
d = os.path.join(dest_folder, item)
delete_path(d)
try:
if os.path.isdir(s):
shutil.copytree(s, d)
log(f"Copied folder: {s} -> {d}")
else:
shutil.copy2(s, d)
log(f"Copied file: {s} -> {d}")
except Exception as e:
log(f"Error copying {s} -> {d}: {e}")
def copy_update_files_darwin(src_folder, dest_folder, updater_name):
updater_name = updater_name + ".app"
for item in os.listdir(src_folder):
if item.lower() == updater_name.lower():
log(f"Skipping updater executable: {item}")
continue
s = os.path.join(src_folder, item)
d = os.path.join(dest_folder, item)
delete_path(d)
try:
if os.path.isdir(s):
subprocess.check_call(["ditto", s, d])
log(f"Copied folder with ditto: {s} -> {d}")
else:
shutil.copy2(s, d)
log(f"Copied file: {s} -> {d}")
except Exception as e:
log(f"Error copying {s} -> {d}: {e}")
def remove_quarantine(app_path):
script = f'''
do shell script "xattr -d -r com.apple.quarantine {shlex.quote(app_path)}" with administrator privileges with prompt "{APP_NAME} needs privileges to finish the update. (1/2)"
'''
try:
subprocess.run(['osascript', '-e', script], check=True)
print("✅ Quarantine attribute removed.")
except subprocess.CalledProcessError as e:
print("❌ Failed to remove quarantine attribute.")
print(e)
def main():
try:
log(f"[Updater] sys.argv: {sys.argv}")
if len(sys.argv) != 3:
log(f"Invalid arguments. Usage: {APP_NAME}_updater <update_folder> <main_app_executable>")
sys.exit(1)
update_folder = sys.argv[1]
main_exe = sys.argv[2]
# Interesting naming convention
parent_dir = os.path.dirname(os.path.abspath(main_exe))
pparent_dir = os.path.dirname(parent_dir)
ppparent_dir = os.path.dirname(pparent_dir)
pppparent_dir = os.path.dirname(ppparent_dir)
updater_name = os.path.basename(sys.argv[0])
log("Updater started.")
log(f"Update folder: {update_folder}")
log(f"Main EXE: {main_exe}")
log(f"Updater EXE: {updater_name}")
if PLATFORM_NAME == 'darwin':
log(f"Main App Folder: {ppparent_dir}")
# Kill all instances of main app
kill_all_processes_by_executable(main_exe)
# Wait until main_exe process is fully gone (polling)
for _ in range(20): # wait max 10 seconds
running = False
for proc in psutil.process_iter(['exe', 'cmdline']):
try:
if PLATFORM_NAME == 'windows':
proc_exe = proc.info.get('exe')
if proc_exe and os.path.samefile(os.path.realpath(proc_exe), os.path.realpath(main_exe)):
running = True
break
elif PLATFORM_NAME == 'linux':
cmdline = proc.info.get('cmdline', [])
if cmdline:
proc_cmd = os.path.realpath(cmdline[0])
if os.path.samefile(proc_cmd, os.path.realpath(main_exe)):
running = True
break
except Exception as e:
log(f"Polling error: {e}")
if not running:
break
time.sleep(0.5)
else:
log("Warning: main executable still running after wait timeout.")
# Delete old version files
if PLATFORM_NAME == 'darwin':
log(f'Attempting to delete {ppparent_dir}')
delete_path(ppparent_dir)
update_folder = os.path.join(sys.argv[1], f"{APP_NAME}-darwin")
copy_update_files_darwin(update_folder, pppparent_dir, updater_name)
else:
delete_path(main_exe)
wait_for_unlock(os.path.join(parent_dir, "_internal"))
# Copy new files excluding the updater itself
copy_update_files(update_folder, parent_dir, updater_name)
except Exception as e:
log(f"Something went wrong: {e}")
# Relaunch main app
try:
if PLATFORM_NAME == 'linux':
os.chmod(main_exe, 0o755)
log("Added executable bit")
if PLATFORM_NAME == 'darwin':
os.chmod(ppparent_dir, 0o755)
log("Added executable bit")
remove_quarantine(ppparent_dir)
log(f"Removed the quarantine flag on {ppparent_dir}")
subprocess.Popen(['open', ppparent_dir, "--args", "--finish-update"])
else:
subprocess.Popen([main_exe, "--finish-update"], cwd=parent_dir)
log("Relaunched main app.")
except Exception as e:
log(f"Failed to relaunch main app: {e}")
log("Updater completed. Exiting.")
sys.exit(0)
if __name__ == "__main__":
main()
-4922
View File
File diff suppressed because it is too large Load Diff
@@ -1,9 +1,13 @@
import os
from PySide6.QtWidgets import QApplication, QComboBox, QDialog, QHBoxLayout, QLabel, QLineEdit, QListView, QMessageBox, QPushButton, QVBoxLayout, QWidget
from PySide6.QtGui import QStandardItemModel, QStandardItem, QPixmap
from PySide6.QtWidgets import QApplication, QComboBox, QDialog, QHBoxLayout, QLabel, QLineEdit, QListView, QMessageBox, QPushButton, QVBoxLayout, QWidget, QFrame, QSpinBox
from PySide6.QtGui import QStandardItemModel, QStandardItem, QPixmap, QIntValidator, QDoubleValidator
from PySide6.QtCore import QEvent, Qt
from src.shared.shareddata import APP_NAME
from src.shared.flaresbasewidget import FullClickComboBox
class ClickableLabel(QLabel):
def __init__(self, full_pixmap: QPixmap, thumbnail_pixmap: QPixmap):
super().__init__()
@@ -178,14 +182,6 @@ class FullClickComboBox(QComboBox):
return super().eventFilter(obj, event)
from PySide6.QtWidgets import QComboBox, QFrame, QHBoxLayout, QListView, QMessageBox, QPushButton, QSpinBox, QWidget, QVBoxLayout, QLabel, QLineEdit
from PySide6.QtGui import QIntValidator, QDoubleValidator, QStandardItemModel, QStandardItem
from PySide6.QtCore import Qt
from src.shared.s_shared import APP_NAME
from src.shared.s_flaresbasewidget import FullClickComboBox
class ParamSection(QWidget):
"""
A widget section that dynamically creates labeled input fields from parameter metadata.
@@ -1027,4 +1023,4 @@ class FlaresBaseWidget(QWidget):
model.appendRow(item)
self._connect_select_all_toggle(toggle_ref, model)
self.update_participant_dropdown_label(combo=target_combo)
self.update_participant_dropdown_label(combo=target_combo)
-550
View File
@@ -1,550 +0,0 @@
"""
Filename: updater.py
Description: Generic updater file
Author: Tyler de Zeeuw
License: GPL-3.0
"""
# Built-in imports
import os
import re
import sys
import time
import shlex
import shutil
import zipfile
import traceback
import subprocess
import configparser
# External library imports
import psutil
import requests
from PySide6.QtWidgets import QMessageBox
from PySide6.QtCore import QThread, Signal, QObject
class UpdateDownloadThread(QThread):
"""
Thread that downloads and extracts an update package and emits a signal on completion or error.
Args:
download_url (str): URL of the update zip file to download.
latest_version (str): Version string of the latest update.
"""
update_ready = Signal(str, str)
error_occurred = Signal(str)
def __init__(self, download_url, latest_version, platform_name, app_name):
super().__init__()
self.download_url = download_url
self.latest_version = latest_version
self.platform_name = platform_name
self.app_name = app_name
def run(self):
try:
local_filename = os.path.basename(self.download_url)
if self.platform_name == 'darwin':
tmp_dir = f'/tmp/{self.app_name}tempupdate'
os.makedirs(tmp_dir, exist_ok=True)
local_path = os.path.join(tmp_dir, local_filename)
else:
local_path = os.path.join(os.getcwd(), local_filename)
# Download the file
with requests.get(self.download_url, stream=True, timeout=15) as r:
r.raise_for_status()
with open(local_path, 'wb') as f:
for chunk in r.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
# Extract folder name (remove .zip)
if self.platform_name == 'darwin':
extract_folder = os.path.splitext(local_filename)[0]
extract_path = os.path.join(tmp_dir, extract_folder)
else:
extract_folder = os.path.splitext(local_filename)[0]
extract_path = os.path.join(os.getcwd(), extract_folder)
# Create the folder if not exists
os.makedirs(extract_path, exist_ok=True)
# Extract the zip file contents
if self.platform_name == 'darwin':
subprocess.run(['ditto', '-xk', local_path, extract_path], check=True)
else:
with zipfile.ZipFile(local_path, 'r') as zip_ref:
zip_ref.extractall(extract_path)
# Remove the zip once extracted and emit a signal
os.remove(local_path)
self.update_ready.emit(self.latest_version, extract_path)
except Exception as e:
# Emit a signal signifying failure
self.error_occurred.emit(str(e))
class UpdateCheckThread(QThread):
"""
Thread that checks for updates by querying the API and emits a signal based on the result.
Signals:
download_requested(str, str): Emitted with (download_url, latest_version) when an update is available.
no_update_available(): Emitted when no update is found or current version is up to date.
error_occurred(str): Emitted with an error message if the update check fails.
"""
download_requested = Signal(str, str)
no_update_available = Signal()
error_occurred = Signal(str)
def __init__(self, api_url, api_url_sec, current_version, platform_name, app_name):
super().__init__()
self.api_url = api_url
self.api_url_sec = api_url_sec
self.current_version = current_version
self.platform_name = platform_name
self.app_name = app_name
def run(self):
# if not getattr(sys, 'frozen', False):
# self.error_occurred.emit("Application is not frozen (Development mode).")
# return
try:
latest_version, download_url = self.get_latest_release_for_platform()
if not latest_version:
self.no_update_available.emit()
return
if not download_url:
self.error_occurred.emit(f"No download available for platform '{self.platform_name}'")
return
if self.version_compare(latest_version, self.current_version) > 0:
self.download_requested.emit(download_url, latest_version)
else:
self.no_update_available.emit()
except Exception as e:
self.error_occurred.emit(f"Update check failed: {e}")
def version_compare(self, v1, v2):
def normalize(v): return [int(x) for x in v.split(".")]
return (normalize(v1) > normalize(v2)) - (normalize(v1) < normalize(v2))
def get_latest_release_for_platform(self):
urls = [self.api_url, self.api_url_sec]
for url in urls:
try:
response = requests.get(url, timeout=5)
response.raise_for_status()
releases = response.json()
if not releases:
continue
latest = next((r for r in releases if not r.get("prerelease") and not r.get("draft")), None)
if not latest:
continue
tag = latest["tag_name"].lstrip("v")
for asset in latest.get("assets", []):
if self.platform_name in asset["name"].lower():
return tag, asset["browser_download_url"]
return tag, None
except (requests.RequestException, ValueError) as e:
continue
return None, None
class LocalPendingUpdateCheckThread(QThread):
"""
Thread that checks for locally pending updates by scanning the download directory and emits a signal accordingly.
Args:
current_version (str): Current application version.
platform_suffix (str): Platform-specific suffix to identify update folders.
"""
pending_update_found = Signal(str, str)
no_pending_update = Signal()
def __init__(self, current_version, platform_suffix, platform_name, app_name):
super().__init__()
self.current_version = current_version
self.platform_suffix = platform_suffix
self.platform_name = platform_name
self.app_name = app_name
def version_compare(self, v1, v2):
def normalize(v): return [int(x) for x in v.split(".")]
return (normalize(v1) > normalize(v2)) - (normalize(v1) < normalize(v2))
def run(self):
if self.platform_name == 'darwin':
cwd = f'/tmp/{self.app_name}tempupdate'
else:
cwd = os.getcwd()
pattern = re.compile(r".*-(\d+\.\d+\.\d+)" + re.escape(self.platform_suffix) + r"$")
found = False
try:
for item in os.listdir(cwd):
folder_path = os.path.join(cwd, item)
if os.path.isdir(folder_path) and item.endswith(self.platform_suffix):
match = pattern.match(item)
if match:
folder_version = match.group(1)
if self.version_compare(folder_version, self.current_version) > 0:
self.pending_update_found.emit(folder_version, folder_path)
found = True
break
except:
pass
if not found:
self.no_pending_update.emit()
class UpdateManager(QObject):
"""
Orchestrates the update process.
Main apps should instantiate this and call check_for_updates().
"""
def __init__(self, main_window, api_url, api_url_sec, current_version, platform_name, platform_suffix, app_name):
super().__init__()
self.parent = main_window
self.api_url = api_url
self.api_url_sec = api_url_sec
self.current_version = current_version
self.platform_name = platform_name
self.platform_suffix = platform_suffix
self.app_name = app_name
self.pending_update_version = None
self.pending_update_path = None
def manual_check_for_updates(self):
self.local_check_thread = LocalPendingUpdateCheckThread(self.current_version, self.platform_suffix, self.platform_name, self.app_name)
self.local_check_thread.pending_update_found.connect(self.on_pending_update_found)
self.local_check_thread.no_pending_update.connect(self.on_no_pending_update)
self.local_check_thread.start()
def on_pending_update_found(self, version, folder_path):
self.parent.statusBar().showMessage(f"Pending update found: version {version}")
self.pending_update_version = version
self.pending_update_path = folder_path
self.show_pending_update_popup()
def on_no_pending_update(self):
# No pending update found locally, start server check directly
self.parent.statusBar().showMessage("No pending local update found. Checking server...")
self.start_update_check_thread()
def show_pending_update_popup(self):
msg_box = QMessageBox(self.parent)
msg_box.setWindowTitle("Pending Update Found")
msg_box.setText(f"A previously downloaded update for {self.app_name.upper()} (version {self.pending_update_version}) is available at:\n{self.pending_update_path}\nWould you like to install it now?")
install_now_button = msg_box.addButton("Install Now", QMessageBox.ButtonRole.AcceptRole)
install_later_button = msg_box.addButton("Install Later", QMessageBox.ButtonRole.RejectRole)
msg_box.exec()
if msg_box.clickedButton() == install_now_button:
self.install_update(self.pending_update_path)
else:
self.parent.statusBar().showMessage("Pending update available. Install later.")
# After user dismisses, still check the server for new updates
self.start_update_check_thread()
def start_update_check_thread(self):
self.check_thread = UpdateCheckThread(self.api_url, self.api_url_sec, self.current_version, self.platform_name, self.app_name)
self.check_thread.download_requested.connect(self.on_server_update_requested)
self.check_thread.no_update_available.connect(self.on_server_no_update)
self.check_thread.error_occurred.connect(self.on_error)
self.check_thread.start()
def on_server_no_update(self):
self.parent.statusBar().showMessage("No new updates found on server.", 5000)
def on_server_update_requested(self, download_url, latest_version):
if self.pending_update_version:
cmp = self.version_compare(latest_version, self.pending_update_version)
if cmp > 0:
# Server version is newer than pending update
self.parent.statusBar().showMessage(f"Newer version {latest_version} available on server. Removing old pending update...")
try:
shutil.rmtree(self.pending_update_path)
self.parent.statusBar().showMessage(f"Deleted old update folder: {self.pending_update_path}")
except Exception as e:
self.parent.statusBar().showMessage(f"Failed to delete old update folder: {e}")
# Clear pending update info so new download proceeds
self.pending_update_version = None
self.pending_update_path = None
# Download the new update
self.download_update(download_url, latest_version)
elif cmp == 0:
# Versions equal, no download needed
self.parent.statusBar().showMessage(f"Pending update version {self.pending_update_version} is already latest. No download needed.")
else:
# Server version older than pending? Unlikely but just keep pending update
self.parent.statusBar().showMessage(f"Pending update version {self.pending_update_version} is newer than server version. No action.")
else:
# No pending update, just download
self.download_update(download_url, latest_version)
def download_update(self, download_url, latest_version):
self.parent.statusBar().showMessage("Downloading update...")
self.download_thread = UpdateDownloadThread(download_url, latest_version, self.platform_name, self.app_name)
self.download_thread.update_ready.connect(self.on_update_ready)
self.download_thread.error_occurred.connect(self.on_error)
self.download_thread.start()
def on_update_ready(self, latest_version, extract_folder):
self.parent.statusBar().showMessage("Update downloaded and extracted.")
msg_box = QMessageBox(self.parent)
msg_box.setWindowTitle("Update Ready")
msg_box.setText(f"Version {latest_version} has been downloaded and extracted to:\n{extract_folder}\nWould you like to install it now?")
install_now_button = msg_box.addButton("Install Now", QMessageBox.ButtonRole.AcceptRole)
install_later_button = msg_box.addButton("Install Later", QMessageBox.ButtonRole.RejectRole)
msg_box.exec()
if msg_box.clickedButton() == install_now_button:
self.install_update(extract_folder)
else:
self.parent.statusBar().showMessage("Update ready. Install later.")
def install_update(self, extract_folder):
# Path to updater executable
if self.platform_name == 'windows':
updater_path = os.path.join(os.getcwd(), f"{self.app_name}_updater.exe")
elif self.platform_name == 'darwin':
if getattr(sys, 'frozen', False):
updater_path = os.path.join(os.path.dirname(sys.executable), f"../../../{self.app_name}_updater.app")
else:
updater_path = os.path.join(os.getcwd(), f"../{self.app_name}_updater.app")
elif self.platform_name == 'linux':
updater_path = os.path.join(os.getcwd(), f"{self.app_name}_updater")
else:
updater_path = os.getcwd()
if not os.path.exists(updater_path):
QMessageBox.critical(self.parent, "Error", f"Updater not found at:\n{updater_path}. The absolute path was {os.path.abspath(updater_path)}")
return
# Launch updater with extracted folder path as argument
try:
# Pass current app's executable path for updater to relaunch
main_app_executable = os.path.abspath(sys.argv[0])
print(f'Launching updater with: "{updater_path}" "{extract_folder}" "{main_app_executable}"')
if self.platform_name == 'darwin':
subprocess.Popen(['open', updater_path, '--args', extract_folder, main_app_executable])
else:
subprocess.Popen([updater_path, f'{extract_folder}', f'{main_app_executable}'], cwd=os.path.dirname(updater_path))
# Close the current app so updater can replace files
sys.exit(0)
except Exception as e:
QMessageBox.critical(self.parent, "Error", f"[Updater Launch Failed]\n{str(e)}\n{traceback.format_exc()}")
def on_error(self, message):
# print(f"Error: {message}")
self.parent.statusBar().showMessage(f"Error occurred during update process. {message}")
def version_compare(self, v1, v2):
def normalize(v): return [int(x) for x in v.split(".")]
return (normalize(v1) > normalize(v2)) - (normalize(v1) < normalize(v2))
def wait_for_process_to_exit(process_name, timeout=10):
"""
Waits for a process with the specified name to exit within a timeout period.
Args:
process_name (str): Name (or part of the name) of the process to wait for.
timeout (int, optional): Maximum time to wait in seconds. Defaults to 10.
Returns:
bool: True if the process exited before the timeout, False otherwise.
"""
print(f"Waiting for {process_name} to exit...")
deadline = time.time() + timeout
while time.time() < deadline:
still_running = False
for proc in psutil.process_iter(['name']):
try:
if proc.info['name'] and process_name.lower() in proc.info['name'].lower():
still_running = True
print(f"Still running: {proc.info['name']} (PID: {proc.pid})")
break
except (psutil.NoSuchProcess, psutil.AccessDenied):
continue
if not still_running:
print(f"{process_name} has exited.")
return True
time.sleep(0.5)
print(f"{process_name} did not exit in time.")
return False
def finish_update_if_needed(platform_name, app_name, cfg_path):
"""
Completes a pending application update if '--finish-update' is present in the command-line arguments.
"""
if "--finish-update" in sys.argv:
print("Finishing update...")
update_cfg = configparser.ConfigParser()
try:
update_cfg.read(cfg_path)
update_cfg.set("Options", "show_welcome_dialog", "true")
with open(cfg_path, "w") as f:
update_cfg.write(f)
print("Welcome dialog flag successfully reset to 'true' for next run.")
except Exception as e:
print(f"Warning: Could not update welcome dialog preference flag: {e}")
if platform_name == 'darwin':
app_dir = f'/tmp/{app_name}tempupdate'
else:
app_dir = os.getcwd()
# 1. Find update folder
update_folder = None
for entry in os.listdir(app_dir):
entry_path = os.path.join(app_dir, entry)
if os.path.isdir(entry_path) and entry.startswith(f"{app_name}-") and entry.endswith("-" + platform_name):
update_folder = os.path.join(app_dir, entry)
break
if update_folder is None:
print("No update folder found. Skipping update steps.")
return
if platform_name == 'darwin':
update_folder = os.path.join(update_folder, f"{app_name}-darwin")
# 2. Wait for updater to exit
print(f"Waiting for {app_name}_updater to exit...")
for proc in psutil.process_iter(['pid', 'name']):
if proc.info['name'] and f"{app_name}_updater" in proc.info['name'].lower():
try:
proc.wait(timeout=5)
except psutil.TimeoutExpired:
print(f"Force killing lingering {app_name}_updater")
proc.kill()
# 3. Replace the updater
if platform_name == 'windows':
new_updater = os.path.join(update_folder, f"{app_name}_updater.exe")
dest_updater = os.path.join(app_dir, f"{app_name}_updater.exe")
elif platform_name == 'darwin':
new_updater = os.path.join(update_folder, f"{app_name}_updater.app")
dest_updater = os.path.abspath(os.path.join(sys.executable, f"../../../../{app_name}_updater.app"))
elif platform_name == 'linux':
new_updater = os.path.join(update_folder, f"{app_name}_updater")
dest_updater = os.path.join(app_dir, f"{app_name}_updater")
else:
print("Unknown Platform")
new_updater = os.getcwd()
dest_updater = os.getcwd()
print(f"New updater is {new_updater}")
print(f"Dest updater is {dest_updater}")
print("Writable?", os.access(dest_updater, os.W_OK))
print("Executable path:", sys.executable)
print("Trying to copy:", new_updater, "->", dest_updater)
if os.path.exists(new_updater):
try:
if os.path.exists(dest_updater):
if platform_name == 'darwin':
try:
if os.path.isdir(dest_updater):
shutil.rmtree(dest_updater)
print(f"Deleted directory: {dest_updater}")
else:
os.remove(dest_updater)
print(f"Deleted file: {dest_updater}")
except Exception as e:
print(f"Error deleting {dest_updater}: {e}")
else:
os.remove(dest_updater)
if platform_name == 'darwin':
wait_for_process_to_exit(f"{app_name}_updater", timeout=10)
subprocess.check_call(["ditto", new_updater, dest_updater])
else:
shutil.copy2(new_updater, dest_updater)
if platform_name in ('linux', 'darwin'):
os.chmod(dest_updater, 0o755)
if platform_name == 'darwin':
remove_quarantine(dest_updater, app_name)
print(f"{app_name}_updater replaced.")
except Exception as e:
print(f"Failed to replace {app_name}_updater: {e}")
# 4. Delete the update folder
try:
if platform_name == 'darwin':
shutil.rmtree(app_dir)
else:
shutil.rmtree(update_folder)
except Exception as e:
print(f"Failed to delete update folder: {e}")
sys.argv.remove("--finish-update")
def remove_quarantine(app_path, app_name):
"""
Removes the macOS quarantine attribute from the specified application path.
"""
script = f'''
do shell script "xattr -d -r com.apple.quarantine {shlex.quote(app_path)}" with administrator privileges with prompt "{app_name.upper()} needs privileges to finish the update. (2/2)"
'''
try:
subprocess.run(['osascript', '-e', script], check=True)
print("✅ Quarantine attribute removed.")
except subprocess.CalledProcessError as e:
print("❌ Failed to remove quarantine attribute.")
print(e)
@@ -1,7 +1,7 @@
from PySide6.QtWidgets import QWidget, QVBoxLayout, QLabel
from PySide6.QtCore import Qt
from src.shared.s_shared import APP_NAME, CURRENT_VERSION
from src.shared.shareddata import APP_NAME, CURRENT_VERSION
class AboutWindow(QWidget):
"""
@@ -1,7 +1,8 @@
from PySide6.QtWidgets import QWidget, QVBoxLayout, QTextEdit, QLineEdit
from PySide6.QtCore import Qt
from src.shared.s_shared import APP_NAME, CURRENT_VERSION
from src.shared.shareddata import APP_NAME, CURRENT_VERSION
class TerminalWindow(QWidget):
def __init__(self, parent=None):
@@ -1,22 +1,25 @@
import os
import json
from enum import Enum, auto
from datetime import datetime
import numpy as np
from PySide6.QtWidgets import QWidget, QVBoxLayout, QLabel, QLineEdit, QPushButton, QComboBox, QHBoxLayout, QMessageBox, QFileDialog
from PySide6.QtCore import Qt
from enum import Enum, auto
import json
import os
from mne import Annotations
from mne.io import read_raw_snirf
from mne_nirs.io import write_raw_snirf
import numpy as np
from datetime import datetime
from mne import Annotations
from src.shared.s_shared import APP_NAME
from src.shared.shareddata import APP_NAME
class EventUpdateMode(Enum):
WRITE_SNIRF = auto() # destructive
WRITE_JSON = auto() # non-destructive
class UpdateEventsWindow(QWidget):
def __init__(self, parent=None, mode=EventUpdateMode.WRITE_SNIRF, caller=None):
super().__init__(parent, Qt.WindowType.Window)
@@ -1,12 +1,18 @@
import os
from pathlib import Path
import pandas as pd
import numpy as np
from mne.io import read_raw_snirf
from mne_nirs.io import write_raw_snirf
from PySide6.QtWidgets import QWidget, QVBoxLayout, QLabel, QHBoxLayout, QMessageBox, QLineEdit, QPushButton, QFileDialog
from PySide6.QtCore import Qt
from mne.io import read_raw_snirf
from mne_nirs.io import write_raw_snirf
from src.shared.shareddata import APP_NAME
class UpdateOptodesWindow(QWidget):
def __init__(self, parent=None):
@@ -130,7 +136,7 @@ class UpdateOptodesWindow(QWidget):
def show_help_popup(self, text):
msg = QMessageBox(self)
msg.setWindowTitle("Parameter Info - FLARES")
msg.setWindowTitle(f"Parameter Info - {APP_NAME.upper}")
msg.setText(text)
msg.exec()
@@ -1,7 +1,7 @@
from PySide6.QtWidgets import QWidget, QVBoxLayout, QLabel
from PySide6.QtCore import Qt
from src.shared.s_shared import APP_NAME, PIPELINE_STAGES
from src.shared.shareddata import APP_NAME, PIPELINE_STAGES
class UserGuideWindow(QWidget):
@@ -3,7 +3,8 @@ from PySide6.QtGui import QDesktopServices, QIcon
from PySide6.QtCore import QUrl
from PySide6.QtNetwork import QNetworkAccessManager, QNetworkRequest
from src.shared.s_shared import APP_NAME, CURRENT_VERSION, resource_path
from src.shared.shareddata import APP_NAME, CURRENT_VERSION, resource_path
class WelcomeDialog(QDialog):
def __init__(self, parent=None, direct=True):