6825606791

Committed 10 Nov 2023 01:46PM UTC coverage: 75.745% (+0.3%) from 75.447%

Build # 6825606791

Build Type

Pull #1974

github

Committed by

web-flow

Commit Message

Merge f521db0a0 into 7e39d81a5

Pull Request Pull Request #1974: Liveview sub directory support

Run Details

1867 of 2924 branches covered (0.0%)

Branch coverage included in aggregate %.

36 of 44 new or added lines in 1 file covered. (81.82%)

2 existing lines in 1 file now uncovered.

9013 of 11440 relevant lines covered (78.78%)

0.79 hits per line

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

75.16

/mantidimaging/gui/windows/live_viewer/model.py

# Copyright (C) 2023 ISIS Rutherford Appleton Laboratory UKRI
# SPDX - License - Identifier: GPL-3.0-or-later
from __future__ import annotations

import time
from typing import TYPE_CHECKING
from pathlib import Path
from logging import getLogger
from PyQt5.QtCore import QFileSystemWatcher, QObject, pyqtSignal

if TYPE_CHECKING:
    from os import stat_result
    from mantidimaging.gui.windows.live_viewer.view import LiveViewerWindowPresenter

LOG = getLogger(__name__)


class Image_Data:
    """
    Image Data Class to store represent image data.

    ...

    Attributes
    ----------
    image_path : Path
        path to image file
    image_name : str
        name of image file
    image_size : int
        size of image file
    image_modified_time : float
        last modified time of image file
    """

    def __init__(self, image_path: Path):
        """
        Constructor for Image_Data class.

        Parameters
        ----------
        image_path : str
            path to image file
        """
        self.image_path = image_path
        self.image_name = image_path.name
        self._stat = image_path.stat()

    @property
    def stat(self) -> stat_result:
        return self._stat

    @property
    def image_modified_time(self) -> float:
        """Return the image modified time"""
        return self._stat.st_mtime


class SubDirectory:

    def __init__(self, path: Path) -> None:
        self.path = path
        self._stat = path.stat()
        self.mtime = self._stat.st_mtime

    @property
    def modification_time(self) -> float:
        return self.mtime


class LiveViewerWindowModel:
    """
    The model for the spectrum viewer window.

    ...

    Attributes
    ----------
    presenter : LiveViewerWindowPresenter
        presenter for the spectrum viewer window
    path : Path
        path to dataset
    images : list
        list of images in directory
    """

    def __init__(self, presenter: 'LiveViewerWindowPresenter'):
        """
        Constructor for LiveViewerWindowModel class.

        Parameters
        ----------
        presenter : LiveViewerWindowPresenter
            presenter for the spectrum viewer window
        """

        self.presenter = presenter
        self._dataset_path: Path | None = None
        self.image_watcher: ImageWatcher | None = None
        self.images: list[Image_Data] = []

    @property
    def path(self) -> Path | None:
        return self._dataset_path

    @path.setter
    def path(self, path: Path) -> None:
        self._dataset_path = path
        self.image_watcher = ImageWatcher(path)
        self.image_watcher.image_changed.connect(self._handle_image_changed_in_list)
        self.image_watcher.recent_image_changed.connect(self.handle_image_modified)
        self.image_watcher._handle_directory_change(str(path))

    def _handle_image_changed_in_list(self, image_files: list[Image_Data]) -> None:
        """
        Handle an image changed event. Update the image in the view.
        This method is called when the image_watcher detects a change
        which could be a new image, edited image or deleted image.

        :param image_files: list of image files
        """
        self.images = image_files
        self.presenter.update_image_list(image_files)

    def handle_image_modified(self, image_path: Path):
        self.presenter.update_image_modified(image_path)

    def close(self) -> None:
        """Close the model."""
        if self.image_watcher:
            self.image_watcher.remove_path()
            self.image_watcher = None
        self.presenter = None  # type: ignore # Model instance to be destroyed -type can be inconsistent


class ImageWatcher(QObject):
    """
    A class to watch a directory for new images.

    ...

    Attributes
    ----------
    directory : Path
        path to directory to watch
    watcher : QFileSystemWatcher
        file system watcher to watch directory
    image_changed : pyqtSignal
        signal emitted when an image is added or removed

    Methods
    -------
    find_images()
        Find all the images in the directory
    sort_images_by_modified_time(images)
        Sort the images by modified time.
    """
    image_changed = pyqtSignal(list)  # Signal emitted when an image is added or removed
    recent_image_changed = pyqtSignal(Path)

    def __init__(self, directory: Path):
        """
        Constructor for ImageWatcher class which inherits from QObject.

        Parameters
        ----------
        directory : Path
            path to directory to watch
        """

        super().__init__()
        self.directory = directory
        self.watcher = QFileSystemWatcher()
        self.watcher.directoryChanged.connect(self._handle_directory_change)

        self.recent_file_watcher = QFileSystemWatcher()
        self.recent_file_watcher.fileChanged.connect(self.handle_image_modified)

        self.sub_directories: dict[Path, SubDirectory] = {}
        self.add_sub_directory(SubDirectory(self.directory))

    def find_images(self, directory: Path) -> list[Image_Data]:
        """
        Find all the images in the directory.
        """
        image_files = []
        for file_path in directory.iterdir():
            if self._is_image_file(file_path.name):
                try:
                    image_obj = Image_Data(file_path)
                    image_files.append(image_obj)
                except FileNotFoundError:
                    continue

        return image_files

    def find_sub_directories(self, directory: Path) -> None:
        # COMPAT python < 3.12 - Can replace with Path.walk()
        for filename in directory.glob("**/*"):
            if filename.is_dir():
                self.add_sub_directory(SubDirectory(filename))

    def sort_sub_directory_by_modified_time(self) -> None:
        self.sub_directories = dict(
            sorted(self.sub_directories.items(), key=lambda p: p[1].modification_time, reverse=True))

    @staticmethod
    def sort_images_by_modified_time(images: list[Image_Data]) -> list[Image_Data]:
        """
        Sort the images by modified time.

        :param images: list of image objects to sort by modified time
        :return: sorted list of images
        """
        return sorted(images, key=lambda x: x.image_modified_time)

    def _handle_directory_change(self, directory: str) -> None:
        """
        Handle a directory change event. Update the list of images
        to reflect directory changes and emit the image_changed signal
        with the sorted image list.

        :param directory: directory that has changed
        """
        directory_path = Path(directory)

        # Force the modification time of signal directory, because file changes may not update
        # parent dir mtime
        if directory_path.exists():
            this_dir = SubDirectory(directory_path)
            this_dir.mtime = time.time()
            self.add_sub_directory(this_dir)
        else:
            self.remove_sub_directory(directory_path)

        self.find_sub_directories(directory_path)

        self.sort_sub_directory_by_modified_time()

        for newest_directory in self.sub_directories.values():
            try:
                images = self.find_images(newest_directory.path)
            except FileNotFoundError:
                images = []

            if len(images) > 0:
                break

        images = self.sort_images_by_modified_time(images)
        self.update_recent_watcher(images[-1:])
        self.image_changed.emit(images)

    @staticmethod
    def _is_image_file(file_name: str) -> bool:
        """
        Check if a file is an tiff or tif image file.

        :param file_name: name of file
        :return: True if file is an image file
        """
        image_extensions = ['.tif', '.tiff']
        file_names = any(file_name.lower().endswith(ext) for ext in image_extensions)
        return file_names

    def remove_path(self):
        """
        Remove the currently set path
        """
        self.watcher.removePaths([str(path) for path in self.sub_directories.keys()])
        self.recent_file_watcher.removePaths(self.recent_file_watcher.files())
        assert len(self.watcher.files()) == 0
        assert len(self.watcher.directories()) == 0
        assert len(self.recent_file_watcher.files()) == 0
        assert len(self.recent_file_watcher.directories()) == 0

    def update_recent_watcher(self, images: list[Image_Data]) -> None:
        self.recent_file_watcher.removePaths(self.recent_file_watcher.files())
        self.recent_file_watcher.addPaths([str(image.image_path) for image in images])

    def handle_image_modified(self, file_path):
        self.recent_image_changed.emit(Path(file_path))

    def add_sub_directory(self, sub_dir: SubDirectory):
        if sub_dir.path not in self.sub_directories:
            self.watcher.addPath(str(sub_dir.path))

        self.sub_directories[sub_dir.path] = sub_dir

    def remove_sub_directory(self, sub_dir: Path):
        if sub_dir in self.sub_directories:
            self.watcher.removePath(str(sub_dir))

        del self.sub_directories[sub_dir]

1	# Copyright (C) 2023 ISIS Rutherford Appleton Laboratory UKRI
2	# SPDX - License - Identifier: GPL-3.0-or-later
3	from __future__ import annotations	1✔
4
5	import time	1✔
6	from typing import TYPE_CHECKING	1✔
7	from pathlib import Path	1✔
8	from logging import getLogger	1✔
9	from PyQt5.QtCore import QFileSystemWatcher, QObject, pyqtSignal	1✔
10
11	if TYPE_CHECKING:
12	from os import stat_result
13	from mantidimaging.gui.windows.live_viewer.view import LiveViewerWindowPresenter
14
15	LOG = getLogger(__name__)	1✔
16
17
18	class Image_Data:	1✔
19	"""
20	Image Data Class to store represent image data.
21
22	...
23
24	Attributes
25	----------
26	image_path : Path
27	path to image file
28	image_name : str
29	name of image file
30	image_size : int
31	size of image file
32	image_modified_time : float
33	last modified time of image file
34	"""
35
36	def __init__(self, image_path: Path):	1✔
37	"""
38	Constructor for Image_Data class.
39
40	Parameters
41	----------
42	image_path : str
43	path to image file
44	"""
45	self.image_path = image_path	1✔
46	self.image_name = image_path.name	1✔
47	self._stat = image_path.stat()	1✔
48
49	@property	1✔
50	def stat(self) -> stat_result:	1✔
51	return self._stat	×
52
53	@property	1✔
54	def image_modified_time(self) -> float:	1✔
55	"""Return the image modified time"""
56	return self._stat.st_mtime	1✔
57
58
59	class SubDirectory:	1✔
60
61	def __init__(self, path: Path) -> None:	1✔
62	self.path = path	1✔
63	self._stat = path.stat()	1✔
64	self.mtime = self._stat.st_mtime	1✔
65
66	@property	1✔
67	def modification_time(self) -> float:	1✔
68	return self.mtime	1✔
69
70
71	class LiveViewerWindowModel:	1✔
72	"""
73	The model for the spectrum viewer window.
74
75	...
76
77	Attributes
78	----------
79	presenter : LiveViewerWindowPresenter
80	presenter for the spectrum viewer window
81	path : Path
82	path to dataset
83	images : list
84	list of images in directory
85	"""
86
87	def __init__(self, presenter: 'LiveViewerWindowPresenter'):	1✔
88	"""
89	Constructor for LiveViewerWindowModel class.
90
91	Parameters
92	----------
93	presenter : LiveViewerWindowPresenter
94	presenter for the spectrum viewer window
95	"""
96
97	self.presenter = presenter	×
98	self._dataset_path: Path \| None = None	×
99	self.image_watcher: ImageWatcher \| None = None	×
100	self.images: list[Image_Data] = []	×
101
102	@property	1✔
103	def path(self) -> Path \| None:	1✔
104	return self._dataset_path	×
105
106	@path.setter	1✔
107	def path(self, path: Path) -> None:	1✔
108	self._dataset_path = path	×
109	self.image_watcher = ImageWatcher(path)	×
110	self.image_watcher.image_changed.connect(self._handle_image_changed_in_list)	×
111	self.image_watcher.recent_image_changed.connect(self.handle_image_modified)	×
NEW 112	self.image_watcher._handle_directory_change(str(path))	×
113
114	def _handle_image_changed_in_list(self, image_files: list[Image_Data]) -> None:	1✔
115	"""
116	Handle an image changed event. Update the image in the view.
117	This method is called when the image_watcher detects a change
118	which could be a new image, edited image or deleted image.
119
120	:param image_files: list of image files
121	"""
122	self.images = image_files	×
123	self.presenter.update_image_list(image_files)	×
124
125	def handle_image_modified(self, image_path: Path):	1✔
126	self.presenter.update_image_modified(image_path)	×
127
128	def close(self) -> None:	1✔
129	"""Close the model."""
130	if self.image_watcher:	×
131	self.image_watcher.remove_path()	×
132	self.image_watcher = None	×
133	self.presenter = None # type: ignore # Model instance to be destroyed -type can be inconsistent	×
134
135
136	class ImageWatcher(QObject):	1✔
137	"""
138	A class to watch a directory for new images.
139
140	...
141
142	Attributes
143	----------
144	directory : Path
145	path to directory to watch
146	watcher : QFileSystemWatcher
147	file system watcher to watch directory
148	image_changed : pyqtSignal
149	signal emitted when an image is added or removed
150
151	Methods
152	-------
153	find_images()
154	Find all the images in the directory
155	sort_images_by_modified_time(images)
156	Sort the images by modified time.
157	"""
158	image_changed = pyqtSignal(list) # Signal emitted when an image is added or removed	1✔
159	recent_image_changed = pyqtSignal(Path)	1✔
160
161	def __init__(self, directory: Path):	1✔
162	"""
163	Constructor for ImageWatcher class which inherits from QObject.
164
165	Parameters
166	----------
167	directory : Path
168	path to directory to watch
169	"""
170
171	super().__init__()	1✔
172	self.directory = directory	1✔
173	self.watcher = QFileSystemWatcher()	1✔
174	self.watcher.directoryChanged.connect(self._handle_directory_change)	1✔
175
176	self.recent_file_watcher = QFileSystemWatcher()	1✔
177	self.recent_file_watcher.fileChanged.connect(self.handle_image_modified)	1✔
178
179	self.sub_directories: dict[Path, SubDirectory] = {}	1✔
180	self.add_sub_directory(SubDirectory(self.directory))	1✔
181
182	def find_images(self, directory: Path) -> list[Image_Data]:	1✔
183	"""
184	Find all the images in the directory.
185	"""
186	image_files = []	1✔
187	for file_path in directory.iterdir():	1✔
188	if self._is_image_file(file_path.name):	1✔
189	try:	1✔
190	image_obj = Image_Data(file_path)	1✔
191	image_files.append(image_obj)	1✔
192	except FileNotFoundError:	1✔
193	continue	1✔
194
195	return image_files	1✔
196
197	def find_sub_directories(self, directory: Path) -> None:	1✔
198	# COMPAT python < 3.12 - Can replace with Path.walk()
199	for filename in directory.glob("*/"):	1✔
200	if filename.is_dir():	1✔
201	self.add_sub_directory(SubDirectory(filename))	1✔
202
203	def sort_sub_directory_by_modified_time(self) -> None:	1✔
204	self.sub_directories = dict(	1✔
205	sorted(self.sub_directories.items(), key=lambda p: p[1].modification_time, reverse=True))
206
207	@staticmethod	1✔
208	def sort_images_by_modified_time(images: list[Image_Data]) -> list[Image_Data]:	1✔
209	"""
210	Sort the images by modified time.
211
212	:param images: list of image objects to sort by modified time
213	:return: sorted list of images
214	"""
215	return sorted(images, key=lambda x: x.image_modified_time)	1✔
216
217	def _handle_directory_change(self, directory: str) -> None:	1✔
218	"""
219	Handle a directory change event. Update the list of images
220	to reflect directory changes and emit the image_changed signal
221	with the sorted image list.
222
223	:param directory: directory that has changed
224	"""
225	directory_path = Path(directory)	1✔
226
227	# Force the modification time of signal directory, because file changes may not update
228	# parent dir mtime
229	if directory_path.exists():	1!
230	this_dir = SubDirectory(directory_path)	1✔
231	this_dir.mtime = time.time()	1✔
232	self.add_sub_directory(this_dir)	1✔
233	else:
NEW 234	self.remove_sub_directory(directory_path)	×
235
236	self.find_sub_directories(directory_path)	1✔
237
238	self.sort_sub_directory_by_modified_time()	1✔
239
240	for newest_directory in self.sub_directories.values():	1!
241	try:	1✔
242	images = self.find_images(newest_directory.path)	1✔
NEW 243	except FileNotFoundError:	×
NEW 244	images = []	×
245
246	if len(images) > 0:	1✔
247	break	1✔
248
249	images = self.sort_images_by_modified_time(images)	1✔
250	self.update_recent_watcher(images[-1:])	1✔
251	self.image_changed.emit(images)	1✔
252
253	@staticmethod	1✔
254	def _is_image_file(file_name: str) -> bool:	1✔
255	"""
256	Check if a file is an tiff or tif image file.
257
258	:param file_name: name of file
259	:return: True if file is an image file
260	"""
261	image_extensions = ['.tif', '.tiff']	1✔
262	file_names = any(file_name.lower().endswith(ext) for ext in image_extensions)	1✔
263	return file_names	1✔
264
265	def remove_path(self):	1✔
266	"""
267	Remove the currently set path
268	"""
NEW UNCOV 269	self.watcher.removePaths([str(path) for path in self.sub_directories.keys()])	×
270	self.recent_file_watcher.removePaths(self.recent_file_watcher.files())	×
271	assert len(self.watcher.files()) == 0	×
272	assert len(self.watcher.directories()) == 0	×
273	assert len(self.recent_file_watcher.files()) == 0	×
274	assert len(self.recent_file_watcher.directories()) == 0	×
275
276	def update_recent_watcher(self, images: list[Image_Data]) -> None:	1✔
277	self.recent_file_watcher.removePaths(self.recent_file_watcher.files())	1✔
278	self.recent_file_watcher.addPaths([str(image.image_path) for image in images])	1✔
279
280	def handle_image_modified(self, file_path):	1✔
UNCOV 281	self.recent_image_changed.emit(Path(file_path))	×
282
283	def add_sub_directory(self, sub_dir: SubDirectory):	1✔
284	if sub_dir.path not in self.sub_directories:	1✔
285	self.watcher.addPath(str(sub_dir.path))	1✔
286
287	self.sub_directories[sub_dir.path] = sub_dir	1✔
288
289	def remove_sub_directory(self, sub_dir: Path):	1✔
NEW 290	if sub_dir in self.sub_directories:	×
NEW 291	self.watcher.removePath(str(sub_dir))	×
292
NEW 293	del self.sub_directories[sub_dir]	×

mantidproject / mantidimaging / 6825606791

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