10568073180

Committed 26 Aug 2024 10:13PM UTC coverage: 47.538% (+0.7%) from 46.79%

Build # 10568073180

Build Type

Pull #711

github

Specific Base eeff82

Committed by

web-flow

Commit Message

Merge 599c9edfb into ad41db25f

Pull Request Pull Request #711: 8.23.2

Run Details

121 of 135 new or added lines in 8 files covered. (89.63%)

1025 existing lines in 22 files now uncovered.

4084 of 8591 relevant lines covered (47.54%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

33.76

/iblrig/gui/tools.py

import argparse
import subprocess
import sys
import traceback
from collections.abc import Callable
from inspect import signature
from pathlib import Path
from shutil import disk_usage
from typing import Any

import numpy as np
import pandas as pd
from PyQt5 import QtGui
from PyQt5.QtCore import (
    QAbstractTableModel,
    QModelIndex,
    QObject,
    QRunnable,
    Qt,
    QThreadPool,
    QVariant,
    pyqtProperty,
    pyqtSignal,
    pyqtSlot,
)
from PyQt5.QtGui import QStandardItem, QStandardItemModel
from PyQt5.QtWidgets import QListView, QProgressBar

from iblrig.constants import BASE_PATH
from iblrig.net import get_remote_devices
from iblrig.pydantic_definitions import RigSettings
from iblutil.util import dir_size


def convert_uis():
    """A wrapper for PyQt5's pyuic5 and pyrcc5, set up for development on iblrig."""
    parser = argparse.ArgumentParser()
    parser.add_argument('pattern', nargs='?', default='*.*', type=str)
    args = parser.parse_args()

    gui_path = BASE_PATH.joinpath('iblrig', 'gui')
    files = set([f for f in gui_path.glob(args.pattern)])

    for filename_in in files.intersection(gui_path.glob('*.qrc')):
        rel_path_in = filename_in.relative_to(BASE_PATH)
        rel_path_out = rel_path_in.with_stem(rel_path_in.stem + '_rc').with_suffix('.py')
        args = ['pyrcc5', str(rel_path_in), '-o', str(rel_path_out)]
        print(' '.join(args))
        subprocess.check_output(args, cwd=BASE_PATH)

    for filename_in in files.intersection(gui_path.glob('*.ui')):
        rel_path_in = filename_in.relative_to(BASE_PATH)
        rel_path_out = rel_path_in.with_suffix('.py')
        args = ['pyuic5', str(rel_path_in), '-o', str(rel_path_out), '-x', '--import-from=iblrig.gui']
        print(' '.join(args))
        subprocess.check_output(args, cwd=BASE_PATH)


class WorkerSignals(QObject):
    """
    Signals used by the Worker class to communicate with the main thread.

    Attributes
    ----------
    finished : pyqtSignal
        Signal emitted when the worker has finished its task.

    error : pyqtSignal(tuple)
        Signal emitted when an error occurs. The signal carries a tuple with the exception type,
        exception value, and the formatted traceback.

    result : pyqtSignal(Any)
        Signal emitted when the worker has successfully completed its task. The signal carries
        the result of the task.

    progress : pyqtSignal(int)
        Signal emitted to report progress during the task. The signal carries an integer value.
    """

    finished = pyqtSignal()
    error = pyqtSignal(tuple)
    result = pyqtSignal(object)
    progress = pyqtSignal(int)


class DiskSpaceIndicator(QProgressBar):
    """A custom progress bar widget that indicates the disk space usage of a specified directory."""

    def __init__(self, *args, directory: Path | None, percent_threshold: int = 90, **kwargs):
        """
        Initialize the DiskSpaceIndicator with the specified directory and threshold percentage.

        Parameters
        ----------
        *args : tuple
            Variable length argument list (passed to QProgressBar).
        directory : Path or None
            The directory path to monitor for disk space usage.
        percent_threshold : int, optional
            The threshold percentage at which the progress bar changes color to red. Default is 90.
        **kwargs : dict
            Arbitrary keyword arguments (passed to QProgressBar).
        """
        super().__init__(*args, **kwargs)
        self._directory = directory
        self._percent_threshold = percent_threshold
        self._percent_full = float('nan')
        self.setEnabled(False)
        if self._directory is not None:
            self.update_data()

    def update_data(self):
        """Update the disk space information."""
        worker = Worker(self._get_size)
        worker.signals.result.connect(self._on_get_size_result)
        QThreadPool.globalInstance().start(worker)

    @property
    def critical(self) -> bool:
        """True if the disk space usage is above the given threshold percentage."""
        return self._percent_full > self._percent_threshold

    def _get_size(self):
        """Get the disk usage information for the specified directory."""
        usage = disk_usage(self._directory.anchor)
        self._percent_full = usage.used / usage.total * 100
        self._gigs_dir = dir_size(self._directory) / 1024**3
        self._gigs_free = usage.free / 1024**3

    def _on_get_size_result(self, result):
        """Handle the result of getting disk usage information and update the progress bar accordingly."""
        self.setEnabled(True)
        self.setValue(round(self._percent_full))
        if self.critical:
            p = self.palette()
            p.setColor(QtGui.QPalette.Highlight, QtGui.QColor('red'))
            self.setPalette(p)
        self.setStatusTip(f'{self._directory}: {self._gigs_dir:.1f} GB  •  ' f'available space: {self._gigs_free:.1f} GB')


class Worker(QRunnable):
    """
    A generic worker class for executing functions concurrently in a separate thread.

    This class is designed to run functions concurrently in a separate thread and emit signals
    to communicate the results or errors back to the main thread.

    Adapted from: https://www.pythonguis.com/tutorials/multithreading-pyqt-applications-qthreadpool/

    Attributes
    ----------
    fn : Callable
        The function to be executed concurrently.

    args : tuple
        Positional arguments for the function.

    kwargs : dict
        Keyword arguments for the function.

    signals : WorkerSignals
        An instance of WorkerSignals used to emit signals.

    Methods
    -------
    run() -> None
        The main entry point for running the worker. Executes the provided function and
        emits signals accordingly.
    """

    def __init__(self, fn: Callable[..., Any], *args: Any, **kwargs: Any):
        """
        Initialize the Worker instance.

        Parameters
        ----------
        fn : Callable
            The function to be executed concurrently.

        *args : tuple
            Positional arguments for the function.

        **kwargs : dict
            Keyword arguments for the function.
        """
        super().__init__()
        self.fn = fn
        self.args = args
        self.kwargs = kwargs
        self.signals: WorkerSignals = WorkerSignals()
        if 'progress_callback' in signature(fn).parameters:
            self.kwargs['progress_callback'] = self.signals.progress

    def run(self) -> None:
        """
        Execute the provided function and emit signals accordingly.

        This method is the main entry point for running the worker. It executes the provided
        function and emits signals to communicate the results or errors back to the main thread.

        Returns
        -------
        None
        """
        try:
            result = self.fn(*self.args, **self.kwargs)
        except:  # noqa: E722
            # Handle exceptions and emit error signal with exception details
            traceback.print_exc()
            exctype, value = sys.exc_info()[:2]
            self.signals.error.emit((exctype, value, traceback.format_exc()))
        else:
            # Emit result signal with the result of the task
            self.signals.result.emit(result)
        finally:
            # Emit the finished signal to indicate completion
            self.signals.finished.emit()


class DataFrameTableModel(QAbstractTableModel):
    def __init__(self, *args, df: pd.DataFrame, **kwargs):
        super().__init__(*args, **kwargs)
        self._dataFrame = df

    def dataFrame(self):
        return self._dataFrame

    def setDataFrame(self, data_frame: pd.DataFrame):
        self.beginResetModel()
        self._dataFrame = data_frame.copy()
        self.endResetModel()

    dataFrame = pyqtProperty(pd.DataFrame, fget=dataFrame, fset=setDataFrame)

    def headerData(self, section, orientation, role=...):
        """
        Get the header data for the specified section.

        Parameters
        ----------
        section : int
            The section index.
        orientation : Qt.Orientation
            The orientation of the header.
        role : int, optional
            The role of the header data.

        Returns
        -------
        QVariant
            The header data.
        """
        if role == Qt.DisplayRole:
            if orientation == Qt.Horizontal:
                return str(self._dataFrame.columns[section])
            else:
                return str(self._dataFrame.index[section])

    def rowCount(self, parent=...):
        """
        Get the number of rows in the model.

        Parameters
        ----------
        parent : QModelIndex, optional
            The parent index.

        Returns
        -------
        int
            The number of rows.
        """
        if isinstance(parent, QModelIndex) and parent.isValid():
            return 0
        return self.dataFrame.shape[0]

    def columnCount(self, parent=...):
        """
        Get the number of columns in the model.

        Parameters
        ----------
        parent : QModelIndex, optional
            The parent index.

        Returns
        -------
        int
            The number of columns.
        """
        if isinstance(parent, QModelIndex) and parent.isValid():
            return 0
        return self.dataFrame.shape[1]

    def data(self, index, role=...):
        """
        Get the data for the specified index.

        Parameters
        ----------
        index : QModelIndex
            The index of the data.
        role : int, optional
            The role of the data.

        Returns
        -------
        QVariant
            The data for the specified index.
        """
        if index.isValid():
            row = self._dataFrame.index[index.row()]
            col = self._dataFrame.columns[index.column()]
            dat = self._dataFrame.iloc[row][col]
            if role == Qt.DisplayRole:
                if isinstance(dat, np.generic):
                    return dat.item()
                return dat
        return QVariant()

    def sort(self, column, order=...):
        """
        Sort the data based on the specified column and order.

        Parameters
        ----------
        column : int
            The column index to sort by.
        order : Qt.SortOrder, optional
            The sort order.
        """
        self.layoutAboutToBeChanged.emit()
        col_name = self._dataFrame.columns.values[column]
        self._dataFrame.sort_values(by=col_name, ascending=order == Qt.AscendingOrder, inplace=True)
        self._dataFrame.reset_index(inplace=True, drop=True)
        self.layoutChanged.emit()

    def setData(self, index, value, role=Qt.DisplayRole):
        """
        Set data at the specified index with the given value.

        Parameters
        ----------
        index : QModelIndex
            The index where the data will be set.
        value : Any
            The new value to be set at the specified index.
        role : int, optional
            The role of the data. Default is Qt.DisplayRole.
        """
        if index.isValid():
            row = self._dataFrame.index[index.row()]
            col = self._dataFrame.columns[index.column()]
            self._dataFrame.at[row, col] = value
            self.dataChanged.emit(index, index, [role])


class RemoteDevicesListView(QListView):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.setMouseTracking(True)  # needed for status tips

    def getDevices(self):
        out = []
        for idx in self.selectedIndexes():
            out.append(self.model().itemData(idx)[Qt.UserRole])
        return out


class RemoteDevicesItemModel(QStandardItemModel):
    def __init__(self, *args, iblrig_settings: RigSettings, **kwargs):
        super().__init__(*args, **kwargs)
        self.remote_devices = get_remote_devices(iblrig_settings=iblrig_settings)
        self.update()

    @pyqtSlot()
    def update(self):
        self.clear()
        for device_name, device_address in self.remote_devices.items():
            item = QStandardItem(device_name)
            item.setStatusTip(f'Remote Device "{device_name}" - {device_address}')
            item.setData(device_name, Qt.UserRole)
            self.appendRow(item)

1	import argparse	2✔
2	import subprocess	2✔
3	import sys	2✔
4	import traceback	2✔
5	from collections.abc import Callable	2✔
6	from inspect import signature	2✔
7	from pathlib import Path	2✔
8	from shutil import disk_usage	2✔
9	from typing import Any	2✔
10
11	import numpy as np	2✔
12	import pandas as pd	2✔
13	from PyQt5 import QtGui	2✔
14	from PyQt5.QtCore import (	2✔
15	QAbstractTableModel,
16	QModelIndex,
17	QObject,
18	QRunnable,
19	Qt,
20	QThreadPool,
21	QVariant,
22	pyqtProperty,
23	pyqtSignal,
24	pyqtSlot,
25	)
26	from PyQt5.QtGui import QStandardItem, QStandardItemModel	2✔
27	from PyQt5.QtWidgets import QListView, QProgressBar	2✔
28
29	from iblrig.constants import BASE_PATH	2✔
30	from iblrig.net import get_remote_devices	2✔
31	from iblrig.pydantic_definitions import RigSettings	2✔
32	from iblutil.util import dir_size	2✔
33
34
35	def convert_uis():	2✔
36	"""A wrapper for PyQt5's pyuic5 and pyrcc5, set up for development on iblrig."""
NEW 37	parser = argparse.ArgumentParser()	×
NEW 38	parser.add_argument('pattern', nargs='?', default='.', type=str)	×
39	args = parser.parse_args()	×
40
41	gui_path = BASE_PATH.joinpath('iblrig', 'gui')	×
42	files = set([f for f in gui_path.glob(args.pattern)])	×
43
44	for filename_in in files.intersection(gui_path.glob('*.qrc')):	×
45	rel_path_in = filename_in.relative_to(BASE_PATH)	×
46	rel_path_out = rel_path_in.with_stem(rel_path_in.stem + '_rc').with_suffix('.py')	×
47	args = ['pyrcc5', str(rel_path_in), '-o', str(rel_path_out)]	×
48	print(' '.join(args))	×
49	subprocess.check_output(args, cwd=BASE_PATH)	×
50
51	for filename_in in files.intersection(gui_path.glob('*.ui')):	×
52	rel_path_in = filename_in.relative_to(BASE_PATH)	×
53	rel_path_out = rel_path_in.with_suffix('.py')	×
54	args = ['pyuic5', str(rel_path_in), '-o', str(rel_path_out), '-x', '--import-from=iblrig.gui']	×
55	print(' '.join(args))	×
56	subprocess.check_output(args, cwd=BASE_PATH)	×
57
58
59	class WorkerSignals(QObject):	2✔
60	"""
61	Signals used by the Worker class to communicate with the main thread.
62
63	Attributes
64	----------
65	finished : pyqtSignal
66	Signal emitted when the worker has finished its task.
67
68	error : pyqtSignal(tuple)
69	Signal emitted when an error occurs. The signal carries a tuple with the exception type,
70	exception value, and the formatted traceback.
71
72	result : pyqtSignal(Any)
73	Signal emitted when the worker has successfully completed its task. The signal carries
74	the result of the task.
75
76	progress : pyqtSignal(int)
77	Signal emitted to report progress during the task. The signal carries an integer value.
78	"""
79
80	finished = pyqtSignal()	2✔
81	error = pyqtSignal(tuple)	2✔
82	result = pyqtSignal(object)	2✔
83	progress = pyqtSignal(int)	2✔
84
85
86	class DiskSpaceIndicator(QProgressBar):	2✔
87	"""A custom progress bar widget that indicates the disk space usage of a specified directory."""
88
89	def __init__(self, args, directory: Path \| None, percent_threshold: int = 90, *kwargs):	2✔
90	"""
91	Initialize the DiskSpaceIndicator with the specified directory and threshold percentage.
92
93	Parameters
94	----------
95	*args : tuple
96	Variable length argument list (passed to QProgressBar).
97	directory : Path or None
98	The directory path to monitor for disk space usage.
99	percent_threshold : int, optional
100	The threshold percentage at which the progress bar changes color to red. Default is 90.
101	**kwargs : dict
102	Arbitrary keyword arguments (passed to QProgressBar).
103	"""
104	super().__init__(args, *kwargs)	×
105	self._directory = directory	×
106	self._percent_threshold = percent_threshold	×
107	self._percent_full = float('nan')	×
108	self.setEnabled(False)	×
109	if self._directory is not None:	×
110	self.update_data()	×
111
112	def update_data(self):	2✔
113	"""Update the disk space information."""
114	worker = Worker(self._get_size)	×
115	worker.signals.result.connect(self._on_get_size_result)	×
116	QThreadPool.globalInstance().start(worker)	×
117
118	@property	2✔
119	def critical(self) -> bool:	2✔
120	"""True if the disk space usage is above the given threshold percentage."""
121	return self._percent_full > self._percent_threshold	×
122
123	def _get_size(self):	2✔
124	"""Get the disk usage information for the specified directory."""
125	usage = disk_usage(self._directory.anchor)	×
126	self._percent_full = usage.used / usage.total * 100	×
127	self._gigs_dir = dir_size(self._directory) / 1024**3	×
128	self._gigs_free = usage.free / 1024**3	×
129
130	def _on_get_size_result(self, result):	2✔
131	"""Handle the result of getting disk usage information and update the progress bar accordingly."""
132	self.setEnabled(True)	×
133	self.setValue(round(self._percent_full))	×
134	if self.critical:	×
135	p = self.palette()	×
136	p.setColor(QtGui.QPalette.Highlight, QtGui.QColor('red'))	×
137	self.setPalette(p)	×
138	self.setStatusTip(f'{self._directory}: {self._gigs_dir:.1f} GB • ' f'available space: {self._gigs_free:.1f} GB')	×
139
140
141	class Worker(QRunnable):	2✔
142	"""
143	A generic worker class for executing functions concurrently in a separate thread.
144
145	This class is designed to run functions concurrently in a separate thread and emit signals
146	to communicate the results or errors back to the main thread.
147
148	Adapted from: https://www.pythonguis.com/tutorials/multithreading-pyqt-applications-qthreadpool/
149
150	Attributes
151	----------
152	fn : Callable
153	The function to be executed concurrently.
154
155	args : tuple
156	Positional arguments for the function.
157
158	kwargs : dict
159	Keyword arguments for the function.
160
161	signals : WorkerSignals
162	An instance of WorkerSignals used to emit signals.
163
164	Methods
165	-------
166	run() -> None
167	The main entry point for running the worker. Executes the provided function and
168	emits signals accordingly.
169	"""
170
171	def __init__(self, fn: Callable[..., Any], args: Any, *kwargs: Any):	2✔
172	"""
173	Initialize the Worker instance.
174
175	Parameters
176	----------
177	fn : Callable
178	The function to be executed concurrently.
179
180	*args : tuple
181	Positional arguments for the function.
182
183	**kwargs : dict
184	Keyword arguments for the function.
185	"""
186	super().__init__()	×
187	self.fn = fn	×
188	self.args = args	×
189	self.kwargs = kwargs	×
190	self.signals: WorkerSignals = WorkerSignals()	×
191	if 'progress_callback' in signature(fn).parameters:	×
192	self.kwargs['progress_callback'] = self.signals.progress	×
193
194	def run(self) -> None:	2✔
195	"""
196	Execute the provided function and emit signals accordingly.
197
198	This method is the main entry point for running the worker. It executes the provided
199	function and emits signals to communicate the results or errors back to the main thread.
200
201	Returns
202	-------
203	None
204	"""
205	try:	×
206	result = self.fn(self.args, *self.kwargs)	×
207	except: # noqa: E722	×
208	# Handle exceptions and emit error signal with exception details
209	traceback.print_exc()	×
210	exctype, value = sys.exc_info()[:2]	×
211	self.signals.error.emit((exctype, value, traceback.format_exc()))	×
212	else:
213	# Emit result signal with the result of the task
214	self.signals.result.emit(result)	×
215	finally:
216	# Emit the finished signal to indicate completion
217	self.signals.finished.emit()	×
218
219
220	class DataFrameTableModel(QAbstractTableModel):	2✔
221	def __init__(self, args, df: pd.DataFrame, *kwargs):	2✔
222	super().__init__(args, *kwargs)	×
223	self._dataFrame = df	×
224
225	def dataFrame(self):	2✔
226	return self._dataFrame	×
227
228	def setDataFrame(self, data_frame: pd.DataFrame):	2✔
229	self.beginResetModel()	×
230	self._dataFrame = data_frame.copy()	×
231	self.endResetModel()	×
232
233	dataFrame = pyqtProperty(pd.DataFrame, fget=dataFrame, fset=setDataFrame)	2✔
234
235	def headerData(self, section, orientation, role=...):	2✔
236	"""
237	Get the header data for the specified section.
238
239	Parameters
240	----------
241	section : int
242	The section index.
243	orientation : Qt.Orientation
244	The orientation of the header.
245	role : int, optional
246	The role of the header data.
247
248	Returns
249	-------
250	QVariant
251	The header data.
252	"""
253	if role == Qt.DisplayRole:	×
254	if orientation == Qt.Horizontal:	×
255	return str(self._dataFrame.columns[section])	×
256	else:
257	return str(self._dataFrame.index[section])	×
258
259	def rowCount(self, parent=...):	2✔
260	"""
261	Get the number of rows in the model.
262
263	Parameters
264	----------
265	parent : QModelIndex, optional
266	The parent index.
267
268	Returns
269	-------
270	int
271	The number of rows.
272	"""
273	if isinstance(parent, QModelIndex) and parent.isValid():	×
274	return 0	×
275	return self.dataFrame.shape[0]	×
276
277	def columnCount(self, parent=...):	2✔
278	"""
279	Get the number of columns in the model.
280
281	Parameters
282	----------
283	parent : QModelIndex, optional
284	The parent index.
285
286	Returns
287	-------
288	int
289	The number of columns.
290	"""
291	if isinstance(parent, QModelIndex) and parent.isValid():	×
292	return 0	×
293	return self.dataFrame.shape[1]	×
294
295	def data(self, index, role=...):	2✔
296	"""
297	Get the data for the specified index.
298
299	Parameters
300	----------
301	index : QModelIndex
302	The index of the data.
303	role : int, optional
304	The role of the data.
305
306	Returns
307	-------
308	QVariant
309	The data for the specified index.
310	"""
311	if index.isValid():	×
312	row = self._dataFrame.index[index.row()]	×
313	col = self._dataFrame.columns[index.column()]	×
314	dat = self._dataFrame.iloc[row][col]	×
315	if role == Qt.DisplayRole:	×
316	if isinstance(dat, np.generic):	×
317	return dat.item()	×
318	return dat	×
319	return QVariant()	×
320
321	def sort(self, column, order=...):	2✔
322	"""
323	Sort the data based on the specified column and order.
324
325	Parameters
326	----------
327	column : int
328	The column index to sort by.
329	order : Qt.SortOrder, optional
330	The sort order.
331	"""
332	self.layoutAboutToBeChanged.emit()	×
333	col_name = self._dataFrame.columns.values[column]	×
334	self._dataFrame.sort_values(by=col_name, ascending=order == Qt.AscendingOrder, inplace=True)	×
335	self._dataFrame.reset_index(inplace=True, drop=True)	×
336	self.layoutChanged.emit()	×
337
338	def setData(self, index, value, role=Qt.DisplayRole):	2✔
339	"""
340	Set data at the specified index with the given value.
341
342	Parameters
343	----------
344	index : QModelIndex
345	The index where the data will be set.
346	value : Any
347	The new value to be set at the specified index.
348	role : int, optional
349	The role of the data. Default is Qt.DisplayRole.
350	"""
351	if index.isValid():	×
352	row = self._dataFrame.index[index.row()]	×
353	col = self._dataFrame.columns[index.column()]	×
354	self._dataFrame.at[row, col] = value	×
355	self.dataChanged.emit(index, index, [role])	×
356
357
358	class RemoteDevicesListView(QListView):	2✔
359	def __init__(self, args, *kwargs):	2✔
360	super().__init__(args, *kwargs)	×
361	self.setMouseTracking(True) # needed for status tips	×
362
363	def getDevices(self):	2✔
364	out = []	×
365	for idx in self.selectedIndexes():	×
366	out.append(self.model().itemData(idx)[Qt.UserRole])	×
367	return out	×
368
369
370	class RemoteDevicesItemModel(QStandardItemModel):	2✔
371	def __init__(self, args, iblrig_settings: RigSettings, *kwargs):	2✔
372	super().__init__(args, *kwargs)	×
373	self.remote_devices = get_remote_devices(iblrig_settings=iblrig_settings)	×
374	self.update()	×
375
376	@pyqtSlot()	2✔
377	def update(self):	2✔
378	self.clear()	×
379	for device_name, device_address in self.remote_devices.items():	×
380	item = QStandardItem(device_name)	×
381	item.setStatusTip(f'Remote Device "{device_name}" - {device_address}')	×
382	item.setData(device_name, Qt.UserRole)	×
383	self.appendRow(item)	×

int-brain-lab / iblrig / 10568073180

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous