-
Notifications
You must be signed in to change notification settings - Fork 1
/
CombineThreadWorker.py
138 lines (122 loc) · 7.35 KB
/
CombineThreadWorker.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
#
# Object running the combination routines when running in GUI mode.
# This object will be run as a sub-thread to leave the UI responsive, both so that
# Mouse and window responds and so that the user can click a Cancel button up there
# to stop a long-running process. There is no good "thread cancel" signal in Python
# so the "cancel" is implemented by setting a flag which is periodically polled in this thread.
#
from PyQt5.QtCore import QObject, pyqtSignal
import MasterMakerExceptions
from ConsoleCallback import ConsoleCallback
from DataModel import DataModel
from FileCombiner import FileCombiner
from FileDescriptor import FileDescriptor
from SessionController import SessionController
class CombineThreadWorker(QObject):
# Signals emitted from the thread
finished = pyqtSignal() # Tell interested parties that we are finished
console_line = pyqtSignal(str) # Add a line to the console object in the UI
remove_from_ui = pyqtSignal(str) # Remove given file (full path) from the UI table
def __init__(self, data_model: DataModel,
descriptors: [FileDescriptor],
output_path: str,
session_controller: SessionController):
"""
Initiaize the combine-thread-worker object
:param data_model: Data Model giving all the relevant options for this job
:param descriptors: List of files to be combined
:param output_path: Absolute path name where combined result is to go
:param session_controller: Session controller for this subtask
"""
QObject.__init__(self)
self._data_model = data_model
self._descriptors = descriptors
self._output_path = output_path
self._session_controller = session_controller
def run_combination_session(self):
"""
Run the file-combination session represented by this object
"""
# Create a console output object. This is passed in to the various math routines
# to allow them to output progress. We use this indirect method of getting progress reports
# so that it can go to the console window in this case, but the same worker code can send
# progress lines to the standard system output when being run from the command line
console = ConsoleCallback(self.console_callback)
console.message("Starting session", 0)
file_combiner = FileCombiner(self._session_controller, self.file_moved_callback)
# Do actual work
try:
# Are we using grouped processing?
if self._data_model.get_group_by_size() \
or self._data_model.get_group_by_filter() \
or self._data_model.get_group_by_temperature():
file_combiner.process_groups(self._data_model, self._descriptors,
self._output_path,
console)
else:
# Not grouped, producing a single output file. Get output file location
file_combiner.original_non_grouped_processing(self._descriptors, self._data_model,
self._output_path,
console)
except FileNotFoundError as exception:
self.error_dialog("File not found", f"File \"{exception.filename}\" not found or not readable")
except MasterMakerExceptions.NoGroupOutputDirectory as exception:
self.error_dialog("Group Directory Missing",
f"The specified output directory \"{exception.get_directory_name()}\""
f" does not exist and could not be created.")
except MasterMakerExceptions.NotAllFlatFrames:
self.error_dialog("The selected files are not all Flat Frames",
"If you know the files are flat frames, they may not have proper FITS data "
"internally. Check the \"Ignore FITS file type\" box to proceed anyway.")
except MasterMakerExceptions.IncompatibleSizes:
self.error_dialog("The selected files can't be combined",
"To be combined into a master file, the files must have identical X and Y "
"dimensions, and identical Binning values.")
except MasterMakerExceptions.NoAutoCalibrationDirectory as exception:
self.error_dialog("Auto Calibration Directory Missing",
f"The specified directory for auto-calibration files, "
f"\"{exception.get_directory_name()}\","
f" does not exist or could not be read.")
except MasterMakerExceptions.AutoCalibrationDirectoryEmpty as exception:
self.error_dialog("Auto Calibration Directory Empty",
f"The specified directory for auto-calibration files, "
f"\"{exception.get_directory_name()}\","
f" does not contain any calibration files (or cannot be read).")
except MasterMakerExceptions.NoSuitableAutoBias:
self.error_dialog("No matching calibration file",
"No bias or dark file of appropriate size could be found in the provided "
"calibration file directory.")
except PermissionError as exception:
self.error_dialog("Unable to write file",
f"The specified output file, "
f"\"{exception.filename}\","
f" cannot be written or replaced: \"permission error\"")
except MasterMakerExceptions.AutoCalibrationNoBiasFiles:
self.error_dialog("No Bias Files",
f"The auto-directory does not contain any Bias files")
except MasterMakerExceptions.SessionCancelled:
self.console_callback("*** Session cancelled ***")
self.finished.emit()
def console_callback(self, message: str):
"""
The console object has produced a line it would like displayed. We'll emit it as a signal
from this sub-thread, so it can be picked up by the main thread and displayed in the console
frame in the user interface.
:param message: Message to be written to the command line or console window
"""
self.console_line.emit(message)
def error_dialog(self, short_message: str, long_message: str):
"""
Display an error message from an exceptionon the console, with suitable formatting
:param short_message: Brief description of the problem
:param long_message: Longer explanatory text when it is available
"""
self.console_callback("*** ERROR *** " + short_message + ": " + long_message)
def file_moved_callback(self, file_moved_from_path: str):
"""
Method that is called back when a file is moved after being processed
Send this information back to the main task by emitting a signal.
This allows us to remove it from the user interface, since the path will no longer be valid
:param file_moved_from_path: Where *was* the file that we just moved?
"""
self.remove_from_ui.emit(file_moved_from_path)