Navigation

Examples

The source files are all in examples directory.

Overview

Callback Object Example

Example implementation for callbacks to the main thread. (Code)

Callback Service Example

Example implementation for callbacks to another service. (Code)

Local Service Example

How to use a service locally. (Code)

Logging Window Example

Show logging in a QTextEdit. (Code)

Matplotlib Example

Plot some random numbers using matplotlib. (Code)

PyQtGraph Example

Plot some random numbers using pyqtgraph. (Code)

PySide PyQt Example

Run PySide and PyQt widgets in parallel. (Code)

Queued Callback Example

Example implementation for callbacks with a QueuedService (Code)

Queued Service Example

Example implementation for a QueuedService (Code)

QtService Example

Demonstrates the need of using the signal/slot mechanism instead of normal function calls. (Code)

Warehouse Example

Examples used for the tutorial. See Warehouse example.

Code

Callback Object Example

callback_events/callback_objects.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
#!/usr/bin/python
# -*- coding: utf-8 -*-
'''
CALLBACK OBJECT EXAMPLE
-----------------------

Example implementation for callbacks to the main thread
using a CallbackObject and a normal Service


.. created on 17.09.2013

.. codeauthor:: Philipp Brimmers
                <P.Brimmers@yahoo.de>
'''

from __future__ import print_function, unicode_literals, division

# Built-In
import time

# External

# Internal
import PyroMP.log_server as log
from PyroMP import (callback,
                    CallbackObject,
                    CallbackServer,
                    Event,
                    NameServer,
                    Service)


class TestService(Service):

    def __init__(self, multiplex=False, async=True):
        super(TestService, self).__init__(multiplex, async)
        self.event = Event()

    def callback(self, callback):
        # call callback function directly
        callback.call("Callback test")

    def send_message(self, msg):
        # trigger event with argument
        self.event.trigger(msg)

    def trigger_event(self):
        self.event.trigger()


class LogObject(CallbackObject):
    """Object which receives the callbacks
    """

    def __init__(self, name):
        super(LogObject, self).__init__()
        self.name = name
        self.logger = log.create_logger(name)

    @callback
    def log(self, msg=None):
        self.logger.info("log(msg={!r})".format(msg))
        if msg is not None:
            self.logger.info(msg)


def main():
    with TestService(async=False) as service:
        # Start CallbackServer for CallbackObjects
        with CallbackServer():
            callback1 = LogObject("Callback1")
            callback2 = LogObject("Callback2")

            # normal callback
            service.callback(callback1.log)

            # normal function call
            callback1.log()
            callback2.log()

            # register for event
            service.event.register(callback1.log)
            service.event.register(callback2.log)

            # trigger event in different ways
            service.trigger_event()
            service.event.trigger("Direct triggered")
            service.send_message("Hallo Welt!")

            # callbacks are asynchronous
            # wait for them to be performed
            time.sleep(0.5)


if __name__ == '__main__':
    with NameServer():
        with log.LogServer():
            log.set_loglevel("INFO")
            main()

Callback Service Example

callback_events/callback_service.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
#!/usr/bin/python
# -*- coding: utf-8 -*-
'''
CALLBACK SERVICE EXAMPLE
------------------------

Example implementation for callbacks to another service
using a CallbackService and a normal Service


.. created on 17.09.2013

.. codeauthor:: Philipp Brimmers
                <P.Brimmers@yahoo.de>
'''

from __future__ import print_function, unicode_literals, division

# Built-In
import time

# External

# Internal
import PyroMP.log_server as log
from PyroMP import (callback,
                    CallbackService,
                    Event,
                    NameServer,
                    Service)


class TestService(Service):

    def __init__(self, multiplex=False, async=True):
        super(TestService, self).__init__(multiplex, async)
        self.event = Event()

    def send_message(self, msg):
        # trigger event with argument
        self.event.trigger(msg)

    def trigger_event(self):
        self.event.trigger()


class LogService(CallbackService):
    """Service which receives the callbacks
    """

    def init_resources(self):
        super(LogService, self).init_resources()
        # register for event
        with TestService(async=False) as service:
            service.event.register(self.log)

    def close_resources(self):
        super(LogService, self).close_resources()
        # unregister for event
        with TestService(async=False) as service:
            service.event.unregister(self.log)

    @callback
    def log(self, msg=None):
        logger = self.get_logger()
        logger.info("log(msg={!r})".format(msg))
        if msg is not None:
            logger.info(msg)


def main():
    with TestService(async=False) as service:
        with LogService():
            # trigger event in different ways
            service.trigger_event()
            service.event.trigger("Direct triggered")
            service.send_message("Hallo Welt!")

            # callbacks are asynchronous
            # wait for them to be performed
            time.sleep(0.5)


if __name__ == '__main__':
    with NameServer(), log.LogServer():
        log.set_loglevel(log.INFO)
        main()

Local Service Example

local_service/local_service.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
#!/usr/bin/python
# -*- coding: utf-8 -*-
'''
LOCAL SERVICE EXAMPLE
---------------------

Example implementation how to use a service as normal
python object instead of running in another process or thread


.. created on 19.02.2014

.. codeauthor:: Philipp Brimmers
                <P.Brimmers@yahoo.de>
'''

from __future__ import print_function, unicode_literals, division

# built-in
import os

# Intern
import PyroMP
import PyroMP.log_server as log


class FileService(PyroMP.Service):

    FILENAME = "test.txt"

    def init_resources(self):
        super(FileService, self).init_resources()
        # create empty file
        text = "LOGFILE:\n--------\n"
        with open(self.FILENAME, 'w') as f:
            f.write(text.encode('utf8'))

    def close_resources(self):
        super(FileService, self).close_resources()
        # print file using logging
        logger = self.get_logger()
        with open(self.FILENAME, 'r') as f:
            logger.info("\n" + f.read().decode('utf8'))
        # delete the file
        os.remove(self.FILENAME)

    def write_line(self, text):
        with open(self.FILENAME, 'a') as f:
            f.write((text + "\n").encode('utf8'))


def main():
    # disable the name server
    # to prevent from using an already running one
    PyroMP.NameServer.disable()
    # create service object
    service = FileService()
    # initialize the service and create log file
    service.init_resources()
    # write some lines
    service.write_line("first log entry")
    service.write_line("second log entry")
    service.write_line("third log entry")
    service.write_line("special characters: *°äöüàáì€$&%")
    # close the service object
    # print log file and delete it
    service.close_resources()

if __name__ == "__main__":
    log.set_loglevel("INFO")
    main()

Logging Window Example

qt_service/log_window.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
#!/usr/bin/python
# -*- coding: utf-8 -*-
'''
LOG WINDOW EXAMPLE
------------------

Additional requirements:

- PySide or PyQt4

Starts a LogServer and display all logs in a QTextEdit.


.. created on 30.12.2013
.. codeauthor:: Philipp Brimmers
                <P.Brimmers@yahoo.de>
'''

from __future__ import print_function, unicode_literals, division

# Built-In

# External
from PyroMP.Qt import QtCore, QtGui

# Internal
import PyroMP
import PyroMP.log_server as log


class LogStream(object):
    """File-like object that appends the text to the LogDialog.

    Used for the StreamHandler.
    """

    def write(self, text):
        dlg_conn = LogDialogService.get_connection(async=False)
        dlg_conn.connect()
        dlg_conn.append_log(text)


class LogDialog(QtGui.QDialog):

    append_log_signal = QtCore.Signal(object)

    def __init__(self):
        super(LogDialog, self).__init__()

        layout = QtGui.QVBoxLayout()

        self._text_edit = QtGui.QTextEdit()
        self._text_edit.setReadOnly(True)
        self._text_edit.setFont("Courier New")
        layout.addWidget(self._text_edit)
        self.setLayout(layout)

        self.append_log_signal.connect(self.append_log)
        self.setWindowTitle("Log Window")

    def append_log(self, text):
        text = self._text_edit.toPlainText() + text
        self._text_edit.setPlainText(text)


class LogDialogService(PyroMP.QtService):

    LOGGING = False

    def qt_main(self):
        """Create the gui elements
        """
        logger = self.get_logger()
        logger.debug("Create dialog")
        self._dlg = LogDialog()
        logger.debug("Show dialog")
        # QtService without any window will be stopped
        self._dlg.show()

        # Configure logging for log server loggers
        root_logger = log.get_server_root_logger()
        log_stream = LogStream()
        root_logger.add_streamhandler(log_stream)

    def append_log(self, text):
        # Signal is used instead of function call
        # normal calls are not possible because Qt has problems with threads
        self._dlg.append_log_signal.emit(text)


def main():
    with LogDialogService():
        LogDialogService.wait_for()


if __name__ == '__main__':
    with PyroMP.NameServer(), log.LogServer():
        main()

Matplotlib Example

plotting/matplotlib_example.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
#!/usr/bin/python
# -*- coding: utf-8 -*-
'''
MATPLOTLIB EXAMPLE
------------------

Additional requirements:

- PySide
- numpy
- matplotlib

Plots some random numbers using matplotlib.


.. created on 27.01.2014

.. codeauthor:: Philipp Brimmers
                <P.Brimmers@yahoo.de>
'''

from __future__ import print_function, unicode_literals, division

# Built-In
import time

# External
from numpy.random import normal
import PyroMP
import PyroMP.log_server as log

from PySide.QtCore import Signal
from PySide.QtGui import QDialog, QVBoxLayout

import matplotlib
# force matplotlib to use PySide
matplotlib.rcParams['backend.qt4'] = 'PySide'
if not matplotlib.get_backend() == 'Qt4Agg':
    matplotlib.use('Qt4Agg')
from matplotlib.figure import Figure
from matplotlib.backends.backend_qt4agg import FigureCanvasQTAgg as FigureCanvas
from matplotlib.backends.backend_qt4agg import NavigationToolbar2QTAgg as NavigationToolbar


class MatplotlibDialog(QDialog):

    add_value = Signal(object)

    def __init__(self):
        super(MatplotlibDialog, self).__init__()

        # create matplotlib gui elements
        # 8 inches wide and 6 inches high with a resolution of 80 dpi
        size = (8, 6)  # inch
        dpi = 80
        self._create_gui(size, dpi)

        # create the axes
        # use add_subplot instead of add_axis to enable subplot toolbox
        self._axes = self._fig.add_subplot(111)
        self._axes.set_xlabel("x-axis")
        self._axes.set_ylabel("gaussian distribution (mean=6, variance=0.04)")

        self.add_value.connect(self._add_value)
        self._x_value = 0

        self.setWindowTitle("matplotlib example")

    def _create_gui(self, size, dpi):
        # create figure
        self._fig = Figure(size, dpi=dpi)
        # create canvas widget
        self._canvas = FigureCanvas(self._fig)
        # create toolbar widget and connect it to canvas
        mpl_toolbar = NavigationToolbar(self._canvas, self)

        layout = QVBoxLayout()
        layout.addWidget(mpl_toolbar)
        layout.addWidget(self._canvas)
        self.setLayout(layout)

    def _add_value(self, value):
        self._axes.scatter([self._x_value], [value])
        self._canvas.draw()
        self._x_value += 1


class PlotService(PyroMP.QtService):

    def qt_main(self):
        """Create the gui
        """
        self._dlg = MatplotlibDialog()
        # show the empty window
        # QtService without any window will be stopped
        self._dlg.show()

    def add_value(self, value):
        """Adds an additional point to the plot
        """
        # Signal is used instead of function call
        # normal calls are not possible because Qt has problems with threads
        self._dlg.add_value.emit(value)


def main():
    # start NameServer and LogServer
    with PyroMP.NameServer(), log.LogServer():
        # start PlotService
        with PlotService(async=False) as service:

            # loop until window is closed
            while PlotService.is_running():
                # create Random value
                mean = 6
                std_dev = 0.2
                data_point = normal(mean, std_dev)

                # plot value
                service.add_value(data_point)

                # wait one second
                time.sleep(1)

if __name__ == "__main__":
    main()

PyQtGraph Example

plotting/pyqtgraph_example.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
#!/usr/bin/python
# -*- coding: utf-8 -*-
'''
PYQTGRAPH EXAMPLE
-----------------

Additional requirements:

- PySide
- numpy
- pyqtgraph

Plots some random numbers using pyqtgraph.


.. created on 27.01.2014

.. codeauthor:: Philipp Brimmers
                <P.Brimmers@yahoo.de>
'''

from __future__ import print_function, unicode_literals, division

# Built-In
import time

# External
from numpy.random import normal
import PyroMP
import PyroMP.log_server as log

from PySide.QtCore import QObject, Signal
import pyqtgraph


class PlotWindowManager(QObject):
    """Manages the plot window
    """

    # Signal is used instead of function call
    # normal calls are not possible because Qt has problems with threads
    add_value = Signal(object)

    def __init__(self):
        super(PlotWindowManager, self).__init__()
        # show an empty window
        # QtService without any window will be stopped
        labels = {"left": ("gaussian distribution (mean=6, variance=4)", ),
                 "bottom": ("x-axis", )}
        self._plot_window = pyqtgraph.plot(title="pyqtgraph example",
                                           labels=labels)

        self.add_value.connect(self._add_value)
        self._x_pos = 0
        self._old_value = None

    def _add_value(self, value):
        """Adds an additional point to the plot
        """
        if self._old_value is not None:
            self._plot_window.plot([self._x_pos - 1, self._x_pos],
                                   [self._old_value, value])
        self._x_pos += 1
        self._old_value = value


class PlotService(PyroMP.QtService):

    def qt_main(self):
        """Create the gui
        """
        self._window_manager = PlotWindowManager()

    def add_value(self, value):
        """Adds an additional point to the plot
        """
        # Signal is used instead of function call
        # normal calls are not possible because Qt has problems with threads
        self._window_manager.add_value.emit(value)


def main():
    # start NameServer and LogServer
    with PyroMP.NameServer(), log.LogServer():
        # start PlotService
        with PlotService(async=False) as service:

            # loop until window is closed
            while PlotService.is_running():
                # create Random value
                mean = 6
                std_dev = 2
                data_point = normal(mean, std_dev)

                # plot value
                service.add_value(data_point)

                # wait one second
                time.sleep(1)

if __name__ == "__main__":
    main()

PySide PyQt Example

pyside_pyqt_combination/pyqt_service.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
#!/usr/bin/python
# -*- coding: utf-8 -*-
'''
PYSIDE PYQT EXAMPLE
-------------------

Additional requirements:

- PySide AND PyQt4

Send text from PySide to PyQt widget.

Start pyqt_service.py first and then pyside_service.py.
You can write text into the PySide widget and
it will be forwarded to the PyQt4 one.


.. created on 30.12.2013

.. codeauthor:: Philipp Brimmers
                <P.Brimmers@yahoo.de>
'''

from __future__ import print_function, unicode_literals, division

# Built-In

# External
from PyQt4.QtGui import (QVBoxLayout,
                          QDialog,
                          QTextEdit)
from PyroMP.Qt import QtCore

# Internal
import PyroMP
import PyroMP.log_server as log


class PySideDialogService(PyroMP.QtService):
    pass


class OutputDialog(QDialog):

    append_log_signal = QtCore.Signal(object)

    def __init__(self):
        super(OutputDialog, self).__init__()

        layout = QVBoxLayout()

        self._text_edit = QTextEdit()
        self._text_edit.setReadOnly(True)
        layout.addWidget(self._text_edit)
        self.setLayout(layout)

        self.append_log_signal.connect(self.print_)
        self.setWindowTitle("PyQt Dialog")

    def print_(self, text):
        self._text_edit.setPlainText(text)


class PyQtDialogService(PyroMP.QtService):

    def qt_main(self):
        logger = self.get_logger()
        logger.debug("Create dialog")
        self._dlg = OutputDialog()
        logger.debug("Show dialog")
        # QtService without any window will be stopped
        self._dlg.show()

    def print_(self, text):
        # Signal is used instead of function call
        # normal calls are not possible because Qt has problems with threads
        self._dlg.append_log_signal.emit(text)


def main():
    with PyQtDialogService():
        # blocks until window is closed
        PyQtDialogService.wait_for()


if __name__ == '__main__':
    with PyroMP.NameServer(), log.LogServer():
        main()

pyside_pyqt_combination/pyside_service.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
#!/usr/bin/python
# -*- coding: utf-8 -*-
'''
PYSIDE PYQT EXAMPLE
-------------------

Additional requirements:

- PySide AND PyQt4

Send text from PySide to PyQt widget.

Start pyqt_service.py first and then pyside_service.py.
You can write text into the PySide widget and
it will be forwarded to the PyQt4 one.


.. created on 30.12.2013

.. codeauthor:: Philipp Brimmers
                <P.Brimmers@yahoo.de>
'''

from __future__ import print_function, unicode_literals, division

# Built-In

# External
from PySide.QtGui import (QVBoxLayout,
                          QDialog,
                          QTextEdit)

# Internal
import PyroMP
import PyroMP.log_server as log


class PyQtDialogService(PyroMP.QtService):
    pass


class OutputDialog(QDialog):

    def __init__(self):
        super(OutputDialog, self).__init__()

        layout = QVBoxLayout()

        self._proxy = PyQtDialogService.get_connection()
        self._proxy.connect()

        self._text_edit = QTextEdit()
        self._text_edit.textChanged.connect(self.forward_text)
        layout.addWidget(self._text_edit)
        self.setLayout(layout)

        self.setWindowTitle("PySide Dialog")

    def forward_text(self):
        text = self._text_edit.toPlainText()
        self._proxy.print_(text)


class PySideDialogService(PyroMP.QtService):

    def qt_main(self):
        logger = self.get_logger()
        logger.debug("Create dialog")
        self._dlg = OutputDialog()
        logger.debug("Show dialog")
        # QtService without any window will be stopped
        self._dlg.show()


def main():
    with PySideDialogService():
        # blocks until window is closed
        PySideDialogService.wait_for()


if __name__ == '__main__':
    with PyroMP.NameServer(), log.LogServer():
        main()

Queued Callbacks Example

callback_events/queued_callback.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
#!/usr/bin/python
# -*- coding: utf-8 -*-
'''
QUEUED CALLBACK EXAMPLE
-----------------------

Example implementation for callbacks with a QueuedService
to a CallbackObject

.. created on 17.09.2013

.. codeauthor:: Philipp Brimmers
                <P.Brimmers@yahoo.de>
'''

from __future__ import print_function, unicode_literals, division

# Built-In
import time

# External

# Internal
import PyroMP.log_server as log
from PyroMP import (callback,
                    CallbackObject,
                    CallbackServer,
                    Event,
                    NameServer,
                    QueuedService,
                    Service)
from PyroMP.errors import ForbiddenAttributeAccessError


class FakeEvent(object):

    def register(self, obj):
        pass


class TestService(Service):

    def __init__(self, multiplex=False, async=True):
        super(TestService, self).__init__(multiplex, async)
        self.event = Event()

    def callback(self, callback):
        # call callback function directly
        callback.call("Callback test")

    def send_message(self, msg):
        # trigger event with argument
        self.event.trigger(msg)

    def trigger_event(self):
        self.event.trigger()


class QueuedTestService(QueuedService):

    CLASS = TestService


class LogObject(CallbackObject):
    """Object which receives the callbacks
    """

    def __init__(self, name):
        super(LogObject, self).__init__()
        self.name = name
        self.logger = log.create_logger(name)

    @callback
    def log(self, msg=None):
        self.logger.info("log(msg={!r})".format(msg))
        if msg is not None:
            self.logger.info(msg)


def main():
    logger = log.create_logger("MAIN")
    # Start CallbackServer for CallbackObjects
    with CallbackServer():
        with QueuedTestService() as service:
            callback1 = LogObject("Callback1")
            callback2 = LogObject("Callback2")

            # normal callback
            service.callback(callback1.log)

            # register for event
            service.event.register(callback1.log)
            service.event.register(callback2.log)

            # trigger event in different ways
            service.trigger_event()
            service.sync.send_message("Hallo Welt!")

            try:
                # events may not be triggered directly
                # in queued services
                service.sync.event.trigger()
                logger.error("ForbiddenAttributeAccessError was "
                             "NOT raised")
            except ForbiddenAttributeAccessError:
                logger.info("ForbiddenAttributeAccessError was "
                            "raised (as expected)")

            try:
                # only PyroMP.Event objects can be accessed
                # to register and unregister
                service.sync.fake_event.register(callback1.log)
                logger.error("ForbiddenAttributeAccessError was "
                             "NOT raised")
            except ForbiddenAttributeAccessError:
                logger.info("ForbiddenAttributeAccessError was "
                            "raised (as expected)")

            # callbacks are asynchronous
            # wait for them to be performed
            time.sleep(0.5)


if __name__ == '__main__':
    with NameServer():
        with log.LogServer():
            log.set_loglevel(log.INFO)
            main()

Queued Service Example

queued_service/queued_service.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
#!/usr/bin/python
# -*- coding: utf-8 -*-
'''
QUEUED SERVICE EXAMPLE
----------------------

How to implement an QueuedService and
how is the execution order influenced by the priorities?

The function are called in alternating order, but we expect:

- Slow operation
- Slow operation
- Slow operation
- Fast operation
- Fast operation
- Fast operation


.. created on 17.09.2013
.. codeauthor:: Philipp Brimmers
                <P.Brimmers@yahoo.de>
'''

from __future__ import print_function, unicode_literals, division

# Built-in
import time

# Internal
import PyroMP.log_server as log
from PyroMP import (NameServer,
                    priority,
                    QueuedService,
                    Service)


class TestService(Service):
    """"Service is wrapped by the QueuedService,
    that manages the order of execution"""

    @priority(5)
    def fast_operation(self):
        logger = self.get_logger()
        logger.info("Fast operation")
        time.sleep(1)
        return 1

    @priority(10)
    def slow_operation(self):
        logger = self.get_logger()
        logger.info("Slow operation")
        time.sleep(5)
        return 5


class QueuedTestService(QueuedService):

    CLASS = TestService


def main():
    with QueuedTestService() as service:
        logger = log.create_logger("MAIN")
        service.slow_operation()
        service.fast_operation()
        service.slow_operation()
        service.fast_operation()
        result1 = service.slow_operation()
        result2 = service.fast_operation()
        logger.info("All operations started")

        logger.info("Wait to finish all operations")
        service.sync.stop_listener()
        logger.info("Listener has stopped")

        logger.info("Result1 = {} (expect 5)".format(result1.value))
        logger.info("Result2 = {} (expect 1)".format(result2.value))


if __name__ == '__main__':
    with NameServer():
        with log.LogServer():
            log.set_loglevel(log.INFO)
            main()

QtService Example

qt_service/qt_service.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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
#!/usr/bin/python
# -*- coding: utf-8 -*-
'''
QT SERVICE EXAMPLE
------------------

Additional requirements:

- PySide or PyQt4

Demonstrates the need of using the signal/slot
mechanism instead of normal function calls.

The control dialog can start a output window either using a threaded or
a multiplexed server.
The service provides to function to print text.
One uses normal function calls, the other forwards the
text to the function using the signal/slot mechanism.

Try it and see what happens.

.. Created on 30.12.2013

.. codeauthor:: Philipp Brimmers
                <P.Brimmers@yahoo.de>
'''

from __future__ import print_function, unicode_literals, division

# Built-In

# External

# Internal
import PyroMP
import PyroMP.log_server as log
from PyroMP.Qt import QtCore, QtGui


class OutputDialog(QtGui.QDialog):

    print_signal = QtCore.Signal(object)

    def __init__(self):
        super(OutputDialog, self).__init__()
        self._text_edit = QtGui.QTextEdit()
        self._text_edit.setReadOnly(True)

        layout = QtGui.QHBoxLayout()
        layout.addWidget(self._text_edit)
        self.setLayout(layout)

        self.print_signal.connect(self.print_)
        self.setWindowTitle("Output")

    def print_(self, text):
        """Add a new line, that contains ``text`` to the QTextEdit"""
        text = self._text_edit.toPlainText() + text + "\n"
        self._text_edit.setPlainText(text)


class OutputDialogService(PyroMP.QtService):

    def qt_main(self):
        logger = self.get_logger()
        logger.debug("Create dialog")
        self._dlg = OutputDialog()
        logger.debug("Show dialog")
        # QtService without any window will be stopped
        self._dlg.show()

    def print_with_signal(self, text):
        """emits a signal, that calls the print_ function"""
        self._dlg.print_signal.emit(text)

    def print_with_function_call(self, text):
        """calls print_ function directly"""
        self._dlg.print_(text)


class ControlDialog(QtGui.QDialog):

    def __init__(self):
        super(ControlDialog, self).__init__()

        layout = QtGui.QVBoxLayout()

        control_layout = QtGui.QHBoxLayout()

        self._start_button = QtGui.QPushButton("Start")
        self._start_button.clicked.connect(self.start_output_dialog_service)
        control_layout.addWidget(self._start_button)

        self._server_mode_box = QtGui.QComboBox()
        self._server_mode_box.addItems(["threaded",
                                        "multiplexed"])
        control_layout.addWidget(self._server_mode_box)

        self._stop_button = QtGui.QPushButton("Stop")
        self._stop_button.setDisabled(True)
        self._stop_button.clicked.connect(self.stop_output_dialog_service)
        control_layout.addWidget(self._stop_button)

        layout.addLayout(control_layout)

        self._text_edit = QtGui.QLineEdit()
        layout.addWidget(self._text_edit)
        self.setLayout(layout)

        send_layout = QtGui.QHBoxLayout()

        self._send_button = QtGui.QPushButton("Send (function call)")
        self._send_button.setDisabled(True)
        self._send_button.clicked.connect(self.print_with_function_call)
        send_layout.addWidget(self._send_button)

        self._send_signal_button = QtGui.QPushButton("Send (signal/slot)")
        self._send_signal_button.setDisabled(True)
        self._send_signal_button.clicked.connect(self.print_with_signal)
        send_layout.addWidget(self._send_signal_button)

        layout.addLayout(send_layout)

        self.setWindowTitle("Control Dialog")

    def print_with_signal(self):
        text = self._text_edit.text()
        dlg_conn = OutputDialogService.get_connection(async=False)
        dlg_conn.connect()
        dlg_conn.print_with_signal(text)
        self._text_edit.clear()

    def print_with_function_call(self):
        text = self._text_edit.text()
        dlg_conn = OutputDialogService.get_connection(async=False)
        dlg_conn.connect()
        dlg_conn.print_with_function_call(text)
        self._text_edit.clear()

    def start_output_dialog_service(self):
        # Disable buttons
        self._start_button.setDisabled(True)
        self._server_mode_box.setDisabled(True)

        # Stop service
        QtGui.QApplication.setOverrideCursor(QtGui.QCursor(QtCore.Qt.WaitCursor))
        multiplex = self._server_mode_box.currentIndex() == 1
        try:
            OutputDialogService.start(multiplex)
        finally:
            QtGui.QApplication.restoreOverrideCursor()

        # Enable buttons
        self._stop_button.setEnabled(True)
        self._send_button.setEnabled(True)
        self._send_signal_button.setEnabled(True)

    def stop_output_dialog_service(self):
        # Disable buttons
        self._stop_button.setDisabled(True)
        self._send_button.setDisabled(True)
        self._send_signal_button.setDisabled(True)

        # Stop service
        QtGui.QApplication.setOverrideCursor(QtGui.QCursor(QtCore.Qt.WaitCursor))
        try:
            OutputDialogService.stop()
        finally:
            QtGui.QApplication.restoreOverrideCursor()

        # Enable buttons
        self._start_button.setEnabled(True)
        self._server_mode_box.setEnabled(True)


def main():
    app = QtGui.QApplication([])
    dialog = ControlDialog()
    dialog.show()
    app.exec_()


if __name__ == '__main__':
    with PyroMP.NameServer(), log.LogServer():
        main()

Table Of Contents

Previous topic

QtService

Next topic

Exceptions

This Page

Navigation

© Copyright 2013, Philipp Brimmers. Created using Sphinx 1.1.3.