1# Copyright (c) 2020 Ultimaker B.V.
2# Uranium is released under the terms of the LGPLv3 or higher.
3
4from typing import Any, Callable, Dict, Optional, cast
5
6from PyQt5.QtCore import QObject, QCoreApplication, QEvent, QTimer
7
8
9__all__ = ["TaskManager"]
10
11
12#
13# A custom event that's used to store a callback function and its parameters. When this event is handled, the handling
14# object should invoke the callFunction() method so the callback function will be invoked.
15#
16class _CallFunctionEvent(QEvent):
17
18    def __init__(self, task_manager: "TaskManager", func: Callable, args: Any, kwargs: Any,
19                 delay: Optional[float] = None) -> None:
20        super().__init__(task_manager.event_type)
21
22        self._task_manager = task_manager
23        self._function = func
24        self._args = args
25        self._kwargs = kwargs
26        self._delay = delay
27
28    @property
29    def delay(self) -> Optional[float]:
30        return self._delay
31
32    def callFunction(self) -> None:
33        self._function(*self._args, **self._kwargs)
34
35
36#
37#
38# This is not a singleton class. The TaskManager is intended to make it easier for certain task-management-ish classes
39# to handle tasks within the Qt event loop framework. It makes it easier to:
40#
41#  - Schedule a callback that will be picked up by the Qt event loop later.
42#  - Schedule a callback with a delay (given in seconds).
43#  - Remove all callbacks that has been scheduled but not yet invoked.
44#
45# This class uses QEvent, unique QEvent types, and QCoreApplication::postEvent() to achieve those functionality. A
46# unique QEvent type is assigned for each TaskManager instance, so each instance can cancel the QEvent posted by itself.
47# The unique QEvent type is retrieved via QEvent.registerEventType(), which will return a unique custom event type if
48# available. If no more custom event type is available, it will return -1. A custom/user event type is a number between
49# QEvent::User (1000) and QEvent::MaxUser (65535). See https://doc.qt.io/qt-5/qevent.html
50#
51# Here we use QCoreApplication.removePostedEvents() to remove posted but not yet dispatched events. Those are the events
52# that have been posted but not yet processed. You can consider this as cancelling a task that you have scheduled
53# earlier but it has not yet been executed. Because QCoreApplication.removePostedEvents() can use an eventType argument
54# to specify the event type you want to remove, here we use that unique custom event type for each TaskManager to
55# identify all events that are managed by the TaskManager itself. See https://doc.qt.io/qt-5/qcoreapplication.html
56#
57# According to my experience, QTimer doesn't seem to trigger events very accurately. I had for example, an expected
58# delay of 5.0 seconds, but I got an actual delay of 4.7 seconds. That's around 6% off. So, here we add a little
59# tolerance to all the specified delay.
60#
61class TaskManager(QObject):
62
63    TIME_TOLERANCE = 0.10  # Add 10% to the delayed events to compensate for timer inaccuracy.
64
65    # Acquires a new unique Qt event type integer.
66    @staticmethod
67    def acquireNewEventType() -> int:
68        # QCoreApplication.registerEventType() is thread-safe.
69        new_type = QEvent.registerEventType()
70        if new_type == -1:
71            raise RuntimeError("Failed to register new event type. All user event types are already taken.")
72        return new_type
73
74    def __init__(self, parent: Optional["QObject"]) -> None:
75        super().__init__(parent = parent)
76        self._event_type = TaskManager.acquireNewEventType()
77        # For storing all delayed events
78        self._delayed_events = dict()  # type: Dict[_CallFunctionEvent, Dict[str, Any]]
79
80    @property
81    def event_type(self) -> int:
82        return self._event_type
83
84    # Cleans up all the delayed events and remove all events that were posted by this TaskManager instance.
85    def cleanup(self) -> None:
86        for event in list(self._delayed_events.keys()):
87            self._cleanupDelayedCallEvent(event)
88        self._delayed_events.clear()
89
90        # Removes all events that have been posted to the QApplication.
91        QCoreApplication.instance().removePostedEvents(None, self._event_type)
92
93    # Schedules a callback function to be called later. If delay is given, the callback will be scheduled to call after
94    # the given amount of time. Otherwise, the callback will be scheduled to the QCoreApplication instance to be called
95    # the next time the event gets picked up.
96    def callLater(self, delay: float, callback: Callable, *args, **kwargs) -> None:
97        if delay < 0:
98            raise ValueError("delay must be a non-negative value, but got [%s] instead." % delay)
99
100        delay_to_use = None if delay <= 0 else delay
101        event = _CallFunctionEvent(self, callback, args, kwargs,
102                                   delay = delay_to_use)
103        if delay_to_use is None:
104            QCoreApplication.instance().postEvent(self, event)
105        else:
106            self._scheduleDelayedCallEvent(event)
107
108    def _scheduleDelayedCallEvent(self, event: "_CallFunctionEvent") -> None:
109        if event.delay is None:
110            return
111
112        timer = QTimer(self)
113        timer.setSingleShot(True)
114        timer.setInterval(event.delay * 1000 * (1 + self.TIME_TOLERANCE))
115        timer_callback = lambda e = event: self._onDelayReached(e)
116        timer.timeout.connect(timer_callback)
117        timer.start()
118        self._delayed_events[event] = {"event": event,
119                                       "timer": timer,
120                                       "timer_callback": timer_callback,
121                                       }
122
123    def _cleanupDelayedCallEvent(self, event: "_CallFunctionEvent") -> None:
124        info_dict = self._delayed_events.get(event)
125        if info_dict is None:
126            return
127
128        timer_callback = info_dict["timer_callback"]
129        timer = info_dict["timer"]
130        timer.stop()
131        timer.timeout.disconnect(timer_callback)
132
133        del self._delayed_events[event]
134
135    def _onDelayReached(self, event: "_CallFunctionEvent") -> None:
136        QCoreApplication.instance().postEvent(self, event)
137
138    # Handle Qt events
139    def event(self, event: "QEvent") -> bool:
140        # Call the function
141        if event.type() == self._event_type:
142            call_event = cast(_CallFunctionEvent, event)
143            call_event.callFunction()
144            self._cleanupDelayedCallEvent(call_event)
145            return True
146
147        return super().event(event)
148