[PyQt] Using Qt to emit a signal (or maybe with QTimer) - Printable Version +- Python Forum (https://python-forum.io) +-- Forum: Python Coding (https://python-forum.io/forum-7.html) +--- Forum: GUI (https://python-forum.io/forum-10.html) +--- Thread: [PyQt] Using Qt to emit a signal (or maybe with QTimer) (/thread-35170.html) |
Using Qt to emit a signal (or maybe with QTimer) - pyhill00 - Oct-06-2021 I am trying to send an .hdf5 file to a method in the main python file in the class here:class DesignerMainWindow(QtGui.QMainWindow, Ui_MainWindow): """Customization for Qt Designer created window""" signal_output_log = QtCore.Signal("QString") sig_clear_log = QtCore.Signal() def __init__(self, parent=None): super(DesignerMainWindow, self).__init__(parent) self.setupUi(self) self.image_analyzer = ImageAnalyzer(self) self.listener = watchdog_search.ObserverWrapper("/home/Test_Data/") self.on_finished_run(self.listener.wait_for_file()) def on_finished_run(self, tuple: ()): self.image_analyzer.load_image(str(tuple[0]), str(tuple[1]), from_remote=True)The .hdf5 file comes from this `watchdog_search.py:import time import traceback import os import h5py import queue from typing import Union from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler, DirCreatedEvent, FileCreatedEvent class NewFileHandler(FileSystemEventHandler): """h5 file creation handler for Watchdog""" def __init__(self): self.file_queue = queue.Queue() # callback for File/Directory created event, called by Observer. def on_created(self, event: Union[DirCreatedEvent, FileCreatedEvent]): if event.src_path[-4:] == "hdf5": # run callback with path string self.file_queue.put(event.src_path) class ObserverWrapper: """Encapsulated Observer boilerplate""" def __init__(self, path: str, recursive=True): self.path = path self.recursive = recursive self.observer = Observer() self.handler = NewFileHandler() self.observer.schedule(self.handler, path=path, recursive=recursive) self.start() def start(self): """ Starts observing for filesystem events. Runs self.routine() every 1 second. :param blocking: If true, blocks main thread until keyboard interrupt. """ self.observer.start() def stop(self): """ Stops the observer. When running self.start(blocking=True) then you don't need to call this. """ self.observer.stop() self.observer.join() def wait_for_file(self): """ Wait and Process newly created files """ max_retry_count = 3500 # for test purposes now but want to set an upper bound on verifying a file is finished. # will try h5 file for a max of 35 seconds (upper bound) to see if the file is finished. # Files are usually finished within 20-30 seconds # retry_interval_seconds = .01 # every hundreth it will try the file to see if it finished writing # wait for file to be added #print(self.handler.file_queue.get(block=True)) file_path = self.handler.file_queue.get(block=True) file_name = os.path.basename(file_path) # try to open the file retry_count = 0 while True: try: file = h5py.File(file_path, "r") file.close() return file_path, file_name except OSError: if retry_count < max_retry_count: retry_count += 1 print(f"h5 file <{file_path}> is locked, retrying {retry_count}/{max_retry_count}") time.sleep(retry_interval_seconds) else: print(f"h5 file <{file_path}> reached max retry count, skipping") except Exception as err: print(f"Got unexpected Error <{type(err).__name__}> while opening <{file_path}> ") traceback.print_exc()Currently I call this file by self.listener = watchdog_search.ObserverWrapper("/path/to/folder/of/interest")in the main.py file but this only sends one hdf5 file and doesn't send any more. Watchdog needs to stay open and send content to main.py every time there is a new hdf5 file available. Does anyone know how to use Qt to do this? I was also thinking about using QTimer but I am not sure how this would work exactly. This is not a question of watchdog but rather a question of asynchronous programming and Qt. I need at the end to send the file_path and file_name of the .hdf5 to on_finished_run() or perhaps have on_finished_run() be the subscriber for the signal but I am unsure how to do this when the try returns not the emitted signal but rather the file path and name. Any insight is greatly appreciated.
RE: Using Qt to emit a signal (or maybe with QTimer) - deanhystad - Oct-07-2021 I took this answer for a related question. https://stackoverflow.com/questions/68842730/changing-qlabel-text-using-watchdog-on-qt And maybe modified it to work with your code. I do not have the watchdog module installed and I don't know anything about your image analyzer. Really untested code here: # Many imports class Bridge(QtCore.QObject): '''Serves as a bridge between watchdog thread and Qt thread''' created = QtCore.Signal(FileSystemEvent) class Handler(FileSystemEventHandler): '''watchdog observer event handler''' def __init__(self): super().__init__() self.bridge = Bridge() def on_created(self, event): '''Emit signal when hdf5 file is created''' if event.src_path[-4:] == 'hdf5': self.bridge.created.emit(event) class MyMainWindow(QtGui.QMainWindow, Ui_MainWindow): """My designer created window""" def __init__(self, path): super().__init__() ... # Create observer to signal my new_file method when an # hdf5 file is created in path. handler = Handler() handler.bridge.created.connect(self.new_file) observer = Observer() observer.schedule(handler, path, recursive=True) observer.start() def new_file(self, event): '''Method called when hdf5 file is created. Passed a FileSystemEvent''' print(event.src_path) ... def main(): path = ... app = QtWidgets.QApplication() form = MyMainWindow(path) form.show() app.exec_() if __name__ == '__main__': main() |