Skip to content
Open

Mypy #70

Show file tree
Hide file tree
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
26 changes: 26 additions & 0 deletions mypy.ini
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
[mypy]
# strict checking
strict = True
implicit_reexport = True

# Minimum version supported
python_version = 3.11

packages=pyobs_gui
exclude = docs/

# Allows Type[T] to refer to abstract classes, which is not otherwise supported.
# See https://github.com/python/mypy/issues/4717
disable_error_code = type-abstract

[mypy-pyobs_gui.qt.*]
ignore_errors = true

[mypy-astropy.*]
ignore_missing_imports = True

[mypy-astroplan.*]
ignore_missing_imports = True

[mypy-qasync.*]
ignore_missing_imports = True
85 changes: 43 additions & 42 deletions pyobs_gui/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,17 +3,7 @@
import asyncio
import logging
from collections.abc import Coroutine
from typing import (
List,
Dict,
Tuple,
Any,
Union,
TypeVar,
Optional,
Callable,
)

from typing import Any, TypeVar, Callable, Type
from PyQt5 import QtWidgets, QtGui, QtCore
from astroplan import Observer

Expand All @@ -33,18 +23,26 @@
class BaseWindow:
def __init__(self) -> None:
"""Base class for MainWindow and all widgets."""
self.modules: List[Proxy] = []
self.comm: Optional[Comm] = None
self.observer: Optional[Observer] = None
self.vfs: Optional[Union[VirtualFileSystem, Dict[str, Any]]] = None
self._base_widgets: List[BaseWidget] = []
self.modules: list[Proxy] = []
self._comm: Comm | None = None
self.observer: Observer | None = None
self.vfs: VirtualFileSystem | dict[str, Any] | None = None
self._base_widgets: list[BaseWidget] = []

@property
def comm(self) -> Comm:
if self._comm is None:
raise ValueError("No comm object.")
return self._comm

@property
def module(self) -> Optional[Proxy]:
def module(self) -> Proxy:
"""Returns the first module in the list or None, if list is empty"""
return self.modules[0] if len(self.modules) > 0 else None
if len(self.modules) == 0:
raise ValueError("No module.")
return self.modules[0]

def module_by_name(self, name: str) -> Optional[Proxy]:
def module_by_name(self, name: str) -> Proxy | None:
"""Return the module with the given name or None, if not exists.

Args:
Expand All @@ -62,7 +60,7 @@ def module_by_name(self, name: str) -> Optional[Proxy]:
# nothing found
return None

def modules_by_interface(self, interface: Any) -> List[Proxy]:
def modules_by_interface(self, interface: Any) -> list[Proxy]:
"""Returns all modules that implement the given interface.

Args:
Expand All @@ -73,7 +71,7 @@ def modules_by_interface(self, interface: Any) -> List[Proxy]:
"""
return list(filter(lambda m: isinstance(m, interface), self.modules))

def module_by_interface(self, interface: Any) -> Optional[Proxy]:
def module_by_interface(self, interface: Any) -> Proxy | None:
"""Returns first modules that implement the given interface, or None, if no exist.

Args:
Expand All @@ -85,7 +83,7 @@ def module_by_interface(self, interface: Any) -> Optional[Proxy]:
modules = self.modules_by_interface(interface)
return None if len(modules) == 0 else modules[0]

def create_widget(self, config: Union[Dict[str, Any], type], **kwargs: Any) -> "BaseWidget":
def create_widget(self, config: dict[str, Any] | type, **kwargs: Any) -> "BaseWidget":
"""Creates new widget.

Args:
Expand All @@ -112,37 +110,38 @@ def create_widget(self, config: Union[Dict[str, Any], type], **kwargs: Any) -> "

async def open(
self,
modules: Optional[List[Proxy]] = None,
comm: Optional[Comm] = None,
observer: Optional[Observer] = None,
vfs: Optional[Union[VirtualFileSystem, Dict[str, Any]]] = None,
modules: list[Proxy] | None = None,
comm: Comm | None = None,
observer: Observer | None = None,
vfs: VirtualFileSystem | dict[str, Any] | None = None,
) -> None:
# store
self.modules = [] if modules is None else modules
self.vfs = vfs
self.comm = comm
self._comm = comm
self.observer = observer

"""Open all widgets."""
for widget in self._base_widgets:
await self._open_child(widget)

async def _open_child(self, widget: BaseWidget):
async def _open_child(self, widget: BaseWidget) -> None:
await widget.open(modules=self.modules, vfs=self.vfs, comm=self.comm, observer=self.observer)


class BaseWidget(BaseWindow):
class BaseWidget(BaseWindow, QtWidgets.QWidget):
_show_error = QtCore.pyqtSignal(str)
_enable_buttons = QtCore.pyqtSignal(list, bool)

def __init__(
self,
update_func: Optional[Callable[[], Any]] = None,
update_func: Callable[[], Any] | None = None,
update_interval: float = 1,
*args: Any,
**kwargs: Any,
):
BaseWindow.__init__(self)
QtWidgets.QWidget.__init__(self)

# signals
self._show_error.connect(self.show_error)
Expand All @@ -151,15 +150,15 @@ def __init__(
# update
self._update_func = update_func
self._update_interval = update_interval
self._update_task: Optional[Any] = None
self._update_task: Any | None = None

# sidebar
self.sidebar_widgets: List[BaseWidget] = []
self.sidebar_layout: Optional[QtWidgets.QVBoxLayout] = None
self.sidebar_widgets: list[BaseWidget] = []
self.sidebar_layout: QtWidgets.QVBoxLayout | None = None

# button to extract to window
self.extract_window_button: Optional[QtWidgets.QToolButton] = None
self._windows: List[QtWidgets.QDialog] = []
self.extract_window_button: QtWidgets.QToolButton | None = None
self._windows: list[QtWidgets.QDialog] = []

# has it been initialized?
self._initialized = False
Expand All @@ -168,7 +167,7 @@ def resizeEvent(self, a0: QtGui.QResizeEvent) -> None:
if self.extract_window_button:
self.extract_window_button.move(self.width() - 20, 0)

def show_extract_button(self, klass, title):
def show_extract_button(self, klass: Type[Any], title: str) -> None:
# button to extract to window
self.extract_window_button = QtWidgets.QToolButton(self)
# self.extract_window_button.setText("X")
Expand All @@ -179,7 +178,7 @@ def show_extract_button(self, klass, title):
self.extract_window_button.raise_()

# method for creating new window
def create_window():
def create_window() -> None:
# create dialog and add widget
dialog = QtWidgets.QDialog()
dialog.setWindowTitle(title)
Expand All @@ -205,7 +204,8 @@ async def add_to_sidebar(self, widget: BaseWidget) -> None:
self.sidebar_layout.setContentsMargins(0, 0, 0, 0)
spacer_item = QtWidgets.QSpacerItem(20, 40, QtWidgets.QSizePolicy.Minimum, QtWidgets.QSizePolicy.Expanding)
self.sidebar_layout.addItem(spacer_item)
self.widgetSidebar.setLayout(self.sidebar_layout)
if hasattr(self, "widgetSidebar"):
self.widgetSidebar.setLayout(self.sidebar_layout)

# open it
await self._open_child(widget)
Expand Down Expand Up @@ -246,7 +246,7 @@ async def _update_loop(self) -> None:
while True:
try:
# get module state
if isinstance(self.module, IModule):
if self.module is not None and isinstance(self.module, IModule):
state = await self.module.get_state()
self.setEnabled(state == ModuleState.READY)
if state != ModuleState.READY:
Expand Down Expand Up @@ -292,11 +292,11 @@ async def show_error(self, exception: exc.PyObsError) -> None:
title, message = err.split(":") if ":" in err else ("Error", err)
await QAsyncMessageBox.warning(self, title, message)

def enable_buttons(self, widgets: List[QtWidgets.QWidget], enable: bool) -> None:
def enable_buttons(self, widgets: list[QtWidgets.QWidget], enable: bool) -> None:
for w in widgets:
w.setEnabled(enable)

def get_fits_headers(self, namespaces: Optional[List[str]] = None, **kwargs: Any) -> Dict[str, Tuple[Any, str]]:
def get_fits_headers(self, namespaces: list[str] | None = None, **kwargs: Any) -> dict[str, tuple[Any, str]]:
"""Returns FITS header for the current status of this module.

Args:
Expand All @@ -311,7 +311,8 @@ def get_fits_headers(self, namespaces: Optional[List[str]] = None, **kwargs: Any
hdr[k] = v
return hdr

def colorize_button(self, button: Any, background: Any, black_on_white: bool = True) -> None:
@staticmethod
def colorize_button(button: Any, background: Any, black_on_white: bool = True) -> None:
# get palette
pal = button.palette()

Expand Down
12 changes: 6 additions & 6 deletions pyobs_gui/camerawidget.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,14 +32,13 @@
log = logging.getLogger(__name__)


class CameraWidget(QtWidgets.QWidget, BaseWidget, Ui_CameraWidget):
class CameraWidget(BaseWidget, Ui_CameraWidget):
signal_update_gui = QtCore.pyqtSignal()
signal_new_image = QtCore.pyqtSignal(NewImageEvent, str)

def __init__(self, **kwargs: Any):
QtWidgets.QWidget.__init__(self)
BaseWidget.__init__(self, update_func=self._update, **kwargs)
self.setupUi(self)
self.setupUi(self) # type: ignore

# variables
self.new_image = False
Expand Down Expand Up @@ -88,7 +87,8 @@ async def open(
self.signal_update_gui.connect(self.update_gui)

# subscribe to events
await self.comm.register_event(ExposureStatusChangedEvent, self._on_exposure_status_changed)
if self.comm is not None:
await self.comm.register_event(ExposureStatusChangedEvent, self._on_exposure_status_changed)

# fill sidebar
await self.add_to_sidebar(self.create_widget(FitsHeadersWidget, module=self.module))
Expand Down Expand Up @@ -165,7 +165,7 @@ async def set_full_frame(self) -> None:

@QtCore.pyqtSlot(str, name="on_comboBinning_currentTextChanged")
def binning_changed(self, binning: str) -> None:
self.on_butFullFrame_clicked()
self._set_full_frame()

@QtCore.pyqtSlot(int, name="on_checkBroadcast_stateChanged")
def broadcast_changed(self, state: int) -> None:
Expand Down Expand Up @@ -383,7 +383,7 @@ async def _on_exposure_status_changed(self, event: Event, sender: str) -> bool:
"""

# ignore events from wrong sender
if self.module is None or sender != self.module.name or not isinstance(event, ExposureStatusChangedEvent):
if sender != self.module.name or not isinstance(event, ExposureStatusChangedEvent):
return False

# store new status
Expand Down
2 changes: 1 addition & 1 deletion pyobs_gui/commandinputwidget.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from PyQt5 import QtCore, QtWidgets, QtGui


class CommandInputWidget(QtWidgets.QLineEdit): # type: ignore
class CommandInputWidget(QtWidgets.QLineEdit):
commandExecuted = QtCore.pyqtSignal(str)

def __init__(self, *args: Any, **kwargs: Any) -> None:
Expand Down
13 changes: 8 additions & 5 deletions pyobs_gui/compassmovewidget.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,11 +10,10 @@
from .qt.compassmovewidget_ui import Ui_CompassMoveWidget


class CompassMoveWidget(QtWidgets.QWidget, BaseWidget, Ui_CompassMoveWidget):
class CompassMoveWidget(BaseWidget, Ui_CompassMoveWidget):
def __init__(self, parent: QtWidgets.QWidget, **kwargs: Any):
QtWidgets.QWidget.__init__(self, parent)
BaseWidget.__init__(self, **kwargs)
self.setupUi(self)
self.setupUi(self) # type: ignore

# button colors
self.colorize_button(self.buttonOffsetEast, QtCore.Qt.blue)
Expand All @@ -31,7 +30,11 @@ def _move_offset(self) -> None:

async def __move_offset(self, sender: QtWidgets.QWidget) -> None:
# get offsets
if isinstance(self.module, IOffsetsAltAz) and isinstance(self.module, IPointingAltAz):
if (
self.observer is not None
and isinstance(self.module, IOffsetsAltAz)
and isinstance(self.module, IPointingAltAz)
):
alt, az = await self.module.get_altaz()
altaz = SkyCoord(
alt=alt * u.degree,
Expand Down Expand Up @@ -63,7 +66,7 @@ async def __move_offset(self, sender: QtWidgets.QWidget) -> None:
# move
if isinstance(self.module, IOffsetsRaDec):
self.run_background(self.module.set_offsets_radec, off_ra, off_dec)
elif isinstance(self.module, IOffsetsAltAz):
elif isinstance(self.module, IOffsetsAltAz) and self.observer is not None:
off_alt, off_az = offset_radec_to_altaz(altaz.transform_to(ICRS()), off_ra, off_dec, self.observer.location)
self.run_background(self.module.set_offsets_altaz, off_alt, off_az)
else:
Expand Down
11 changes: 5 additions & 6 deletions pyobs_gui/coolingwidget.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import asyncio
import logging
from typing import Any

from PyQt5 import QtWidgets
from PyQt5.QtCore import pyqtSignal

from pyobs.interfaces import ICooling
Expand All @@ -12,16 +12,15 @@
log = logging.getLogger(__name__)


class CoolingWidget(QtWidgets.QWidget, BaseWidget, Ui_CoolingWidget):
class CoolingWidget(BaseWidget, Ui_CoolingWidget):
signal_update_gui = pyqtSignal()

def __init__(self, **kwargs):
QtWidgets.QWidget.__init__(self)
def __init__(self, **kwargs: Any):
BaseWidget.__init__(self, update_func=self._update, **kwargs)
self.setupUi(self)
self.setupUi(self) # type: ignore

# status
self._status = None
self._status: tuple[bool, float, float] | None = None

# connect signals
self.signal_update_gui.connect(self.update_gui)
Expand Down
Loading