Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
202 changes: 16 additions & 186 deletions src/opm_v2/_opm_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,120 +6,46 @@
2025/02: Initial version of the script.
"""
from __future__ import annotations
# TODO AO getting values from gui instead of json.
import argparse
import importlib
import importlib.util
import os
import sys
import traceback
from contextlib import suppress
from pathlib import Path
from typing import TYPE_CHECKING, cast
from datetime import datetime
from PyQt6.QtCore import pyqtSignal, Qt
from PyQt6.QtGui import QIcon
from PyQt6.QtWidgets import QApplication, QDockWidget
from superqt.utils import WorkerBase

from pymmcore_gui import MicroManagerGUI,WidgetAction, __version__

import json
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING

import numpy as np
from pymmcore_gui import WidgetAction, create_mmgui
from PyQt6.QtCore import Qt
from PyQt6.QtWidgets import QApplication, QDockWidget

from opm_v2.hardware.OPMNIDAQ import OPMNIDAQ
from opm_v2.hardware.AOMirror import AOMirror
from opm_v2.hardware.ElveFlow import OB1Controller
from opm_v2.hardware.PicardShutter import PicardShutter
from opm_v2.engine.opm_engine import OPMEngine
from opm_v2._update_config_widget import OPMSettings
from opm_v2.engine.opm_engine import OPMEngine
from opm_v2.engine.setup_events import (
setup_stagescan,
setup_mirrorscan,
setup_optimizenow,
setup_projection,
setup_stagescan,
setup_optimizenow,
setup_timelapse
)
from opm_v2.hardware.AOMirror import AOMirror
from opm_v2.hardware.ElveFlow import OB1Controller
from opm_v2.hardware.OPMNIDAQ import OPMNIDAQ
from opm_v2.hardware.PicardShutter import PicardShutter

DEBUGGING = True
if TYPE_CHECKING:
from collections.abc import Sequence
from types import TracebackType

ExcTuple = tuple[type[BaseException], BaseException, TracebackType | None]


APP_NAME = "Micro-Manager GUI"
APP_VERSION = __version__
ORG_NAME = "pymmcore-plus"
ORG_DOMAIN = "pymmcore-plus"
APP_ID = f"{ORG_DOMAIN}.{ORG_NAME}.{APP_NAME}.{APP_VERSION}"
RESOURCES = Path(__file__).parent / "resources"
ICON = RESOURCES / ("icon.ico" if sys.platform.startswith("win") else "logo.png")
IS_FROZEN = getattr(sys, "frozen", False)


class MMQApplication(QApplication):
exceptionRaised = pyqtSignal(BaseException)

def __init__(self, argv: list[str]) -> None:
if sys.platform == "darwin" and not argv[0].endswith("mmgui"):
# Make sure the app name in the Application menu is `mmgui`
# which is taken from the basename of sys.argv[0]; we use
# a copy so the original value is still available at sys.argv
argv[0] = "napari"

super().__init__(argv)
self.setApplicationName("Micro-Manager GUI")
self.setWindowIcon(QIcon(str(ICON)))

self.setApplicationName(APP_NAME)
self.setApplicationVersion(APP_VERSION)
self.setOrganizationName(ORG_NAME)
self.setOrganizationDomain(ORG_DOMAIN)
if os.name == "nt" and not IS_FROZEN:
import ctypes

ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID(APP_ID) # type: ignore

self.aboutToQuit.connect(WorkerBase.await_workers)


def parse_args(args: Sequence[str] = ()) -> argparse.Namespace:
if not args:
args = sys.argv[1:]

parser = argparse.ArgumentParser(description="Enter string")
parser.add_argument(
"-c",
"--config",
type=str,
default=None,
help="Config file to load",
nargs="?",
)
return parser.parse_args(args)
DEBUGGING = True


def main() -> None:
"""Run the Micro-Manager GUI."""
args = parse_args()

app = MMQApplication(sys.argv)
_install_excepthook()

win = MicroManagerGUI()
win.setWindowIcon(QIcon(str(ICON)))
win.showMaximized()
win.show()

splsh = "_PYI_SPLASH_IPC" in os.environ and importlib.util.find_spec("pyi_splash")
if splsh: # pragma: no cover
import pyi_splash # pyright: ignore [reportMissingModuleSource]

pyi_splash.update_text("UI Loaded ...")
pyi_splash.close()
win = create_mmgui(exec_app=False)

# --------------------------------------------------------------------------------
# --------------------------------------------------------------------------------
Expand Down Expand Up @@ -502,101 +428,5 @@ def custom_execute_mda(output: Path | str | object | None) -> None:
# --------------------------------------------------------------------------------
# --------------------------------------------------------------------------------

app = QApplication.instance()
app.exec()


# ------------------- Custom excepthook -------------------

def _install_excepthook() -> None:
"""Install a custom excepthook that does not raise sys.exit().

This is necessary to prevent the application from closing when an exception
is raised.
"""
if hasattr(sys, "_original_excepthook_"):
return
sys._original_excepthook_ = sys.excepthook # type: ignore
sys.excepthook = ndv_excepthook


def rich_print_exception(
exc_type: type[BaseException],
exc_value: BaseException,
exc_traceback: TracebackType | None,
) -> None:
import psygnal
from rich.console import Console
from rich.traceback import Traceback

tb = Traceback.from_exception(
exc_type,
exc_value,
exc_traceback,
suppress=[psygnal],
max_frames=100 if IS_FROZEN else 10,
show_locals=True,
)
Console(stderr=True).print(tb)


def _print_exception(
exc_type: type[BaseException],
exc_value: BaseException,
exc_traceback: TracebackType | None,
) -> None:
try:
rich_print_exception(exc_type, exc_value, exc_traceback)
except ImportError:
traceback.print_exception(exc_type, value=exc_value, tb=exc_traceback)


# This log list is used by the ExceptionLog widget
# Be aware that it's currently possible for that widget to clear this list.
# If an immutable record of exceptions is needed, additional logic will be required.
EXCEPTION_LOG: list[ExcTuple] = []


def ndv_excepthook(
exc_type: type[BaseException], exc_value: BaseException, tb: TracebackType | None
) -> None:
EXCEPTION_LOG.append((exc_type, exc_value, tb))
_print_exception(exc_type, exc_value, tb)
if sig := getattr(QApplication.instance(), "exceptionRaised", None):
sig.emit(exc_value)
if not tb:
return

# if we're running in a vscode debugger, let the debugger handle the exception
if (
(debugpy := sys.modules.get("debugpy"))
and debugpy.is_client_connected()
and ("pydevd" in sys.modules)
): # pragma: no cover
with suppress(Exception):
import threading

import pydevd # pyright: ignore [reportMissingImports]

if (py_db := pydevd.get_global_debugger()) is None:
return

py_db = cast("pydevd.PyDB", py_db)
thread = threading.current_thread()
additional_info = py_db.set_additional_thread_info(thread)
additional_info.is_tracing += 1

try:
arg = (exc_type, exc_value, tb)
py_db.stop_on_unhandled_exception(py_db, thread, additional_info, arg)
finally:
additional_info.is_tracing -= 1
# otherwise, if MMGUI_DEBUG_EXCEPTIONS is set, drop into pdb
elif os.getenv("MMGUI_DEBUG_EXCEPTIONS"):
import pdb

pdb.post_mortem(tb)

# after handling the exception, exit if MMGUI_EXIT_ON_EXCEPTION is set
if os.getenv("MMGUI_EXIT_ON_EXCEPTION"):
print("\nMMGUI_EXIT_ON_EXCEPTION is set, exiting.")
sys.exit(1)