File: multithreading_simple_.py

package info (click to toggle)
napari 0.6.6-3
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 12,036 kB
  • sloc: python: 112,264; xml: 72; makefile: 44; sh: 5
file content (54 lines) | stat: -rw-r--r-- 1,412 bytes parent folder | download
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
"""
Multithreading simple
=====================

.. tags:: interactivity
"""

import time

from qtpy.QtWidgets import QApplication, QHBoxLayout, QLabel, QWidget

from napari.qt import thread_worker


@thread_worker
def long_running_function():
    """Just a long running function, most like viewer.update."""
    time.sleep(2)  # long function
    return 'finished!'


def create_widget():
    widget = QWidget()
    layout = QHBoxLayout()
    widget.setLayout(layout)
    widget.status = QLabel('ready...')
    layout.addWidget(widget.status)
    widget.show()
    return widget


if __name__ == '__main__':
    app = QApplication([])
    wdg = create_widget()

    # call decorated function
    # By default, @thread_worker-decorated functions do not immediately start
    worker = long_running_function()
    # Signals are best connected *before* starting the worker.
    worker.started.connect(lambda: wdg.status.setText('worker is running...'))
    worker.returned.connect(lambda x: wdg.status.setText(f'returned {x}'))

    # # Connections may also be passed directly to the decorated function.
    # # The above syntax is equivalent to:
    # worker = long_running_function(
    #     _connect={
    #         'started': lambda: wdg.status.setText("worker is running..."),
    #         'returned': lambda x: wdg.status.setText(f"returned {x!r}"),
    #     }
    # )

    worker.start()

    app.exec_()