cengal.parallel_execution.coroutines.integrations.wxpython.versions.v_0.wxpython
1#!/usr/bin/env python 2# coding=utf-8 3 4# Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space> 5# 6# Licensed under the Apache License, Version 2.0 (the "License"); 7# you may not use this file except in compliance with the License. 8# You may obtain a copy of the License at 9# 10# http://www.apache.org/licenses/LICENSE-2.0 11# 12# Unless required by applicable law or agreed to in writing, software 13# distributed under the License is distributed on an "AS IS" BASIS, 14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 15# See the License for the specific language governing permissions and 16# limitations under the License. 17 18 19__all__ = [ 20 'CoroApp', 21 'bind_coro_explicit', 22 'bind_coro_implicit', 23 'bind_coro', 24 'abind_coro_explicit', 25 'abind_coro_implicit', 26 'abind_coro', 27 'asyncio_bind_coro_explicit', 28 'asyncio_bind_coro_implicit', 29 'asyncio_bind_coro', 30 'bind_asyncio_coro', 31 'bind_running_coro', 32 'bind_running_asyncio_coro', 33 'wx_exec_in_coro', 34 'bind_to', 35 'bind', 36 'event_handler_implicit', 37 'event_handler', 38 'event_handler_explicit', 39 'asyncio_event_handler', 40 'aevent_handler', 41 'blocking_event_handler_implicit', 42 'blocking_event_handler_explicit', 43] 44 45 46""" 47Module Docstring 48Docstrings: http://www.python.org/dev/peps/pep-0257/ 49""" 50 51__author__ = "ButenkoMS <gtalk@butenkoms.space>" 52__copyright__ = "Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>" 53__credits__ = ["ButenkoMS <gtalk@butenkoms.space>", ] 54__license__ = "Apache License, Version 2.0" 55__version__ = "4.4.1" 56__maintainer__ = "ButenkoMS <gtalk@butenkoms.space>" 57__email__ = "gtalk@butenkoms.space" 58# __status__ = "Prototype" 59__status__ = "Development" 60# __status__ = "Production" 61 62 63from cengal.parallel_execution.coroutines.coro_scheduler import CoroScheduler, CoroSchedulerType, Interface, CoroID, Coro, AnyWorker, current_interface, current_coro_scheduler, cs_coro, cs_acoro 64from cengal.parallel_execution.coroutines.coro_standard_services.loop_yield import gly, CoroPriority, gly_patched, agly_patched 65from cengal.parallel_execution.coroutines.coro_standard_services.run_coro import RunCoro 66from cengal.parallel_execution.coroutines.coro_standard_services.put_coro import PutCoro 67from cengal.parallel_execution.coroutines.coro_standard_services.asyncio_loop import AsyncioLoopRequest 68from cengal.parallel_execution.coroutines.coro_standard_services.async_event_bus import AsyncEventBusRequest, try_send_async_event 69from cengal.parallel_execution.coroutines.coro_standard_services.simple_yield import Yield 70from cengal.parallel_execution.coroutines.coro_standard_services.shutdown_loop import ShutdownLoop 71from cengal.parallel_execution.coroutines.coro_standard_services.instance import InstanceRequest 72from cengal.parallel_execution.coroutines.coro_tools.await_coro import await_task_prim 73from cengal.parallel_execution.coroutines.coro_tools.run_in_loop import run_in_loop 74from cengal.parallel_execution.coroutines.coro_tools.coro_flow_control import graceful_coro_destroyer 75from cengal.code_flow_control.smart_values import ValueHolder 76from cengal.data_generation.id_generator import IDGenerator 77from cengal.time_management.sleep_tools import try_sleep, get_usable_min_sleep_interval 78from cengal.math.numbers import RationalNumber 79 80from math import ceil 81from inspect import signature, Signature, ismethod 82from contextlib import contextmanager, asynccontextmanager 83from functools import wraps, update_wrapper, partial 84from asyncio import Future, Task, create_task 85from typing import Optional, Any, Callable, Hashable, Set, Union, Awaitable, Coroutine 86 87 88from wx import App, Window, Frame, Dialog, Timer, GetTopLevelParent, EVT_TIMER, EVT_CLOSE, ID_ANY 89 90 91class BindedEntityIsAboutToBeDestroyed(Exception): 92 pass 93 94 95def sec_to_ms(sec: RationalNumber) -> int: 96 return int(ceil(sec * 1000)) 97 98 99class CoroApp(App): 100 def __init__(self, default_priority: CoroPriority = CoroPriority.normal): 101 self._cs: CoroSchedulerType = current_coro_scheduler() 102 self._cs.on_idle_handlers.add(self._on_system_loop_idle) 103 self._cs.idle_managers_num += 1 104 self._default_priority: CoroPriority = default_priority 105 self._ly = gly(self._default_priority) 106 self._binded_coros: Set = set() 107 self._idle_for: Optional[int] = None # in milliseconds 108 super(CoroApp, self).__init__() 109 110 def OnInit(self): 111 super().OnInit() 112 self._system_loop_timer = Timer(self) 113 self.Bind(EVT_TIMER, self._on_system_loop_timer, self._system_loop_timer) 114 self._system_loop_timer.StartOnce(1) # run every 1 millisecond 115 return True 116 117 def OnExit(self): 118 super().OnExit() 119 i: Interface = current_interface() 120 i.log.info('Application is exiting.') 121 i.log.info('Destroying all binded coros...') 122 for furure_or_coro_id in self._binded_coros: 123 if isinstance(furure_or_coro_id, Future): 124 furure_or_coro_id.cancel() 125 else: 126 graceful_coro_destroyer(i, 0.1, furure_or_coro_id, BindedEntityIsAboutToBeDestroyed) 127 128 i.log.info('All binded coroutines destroyed.') 129 self._cs.idle_managers_num -= 1 130 self._cs.on_idle_handlers.discard(self._on_system_loop_idle) 131 i.log.info('Application is exiting... Done.') 132 return super().OnExit() 133 134 def _on_system_loop_timer(self, event): 135 # from datetime import datetime 136 # print(f'Cengal Timer >> {datetime.now()}>>') 137 self._ly() 138 if self._idle_for is None: 139 self._system_loop_timer.StartOnce(1) # run every 1 millisecond 140 else: 141 # print(f'Idle for {self._idle_for} milliseconds.') 142 idle_for = self._idle_for 143 self._idle_for = None 144 self._system_loop_timer.StartOnce(idle_for) 145 146 def _on_system_loop_idle(self, next_event_after: Optional[RationalNumber]): 147 if next_event_after is None: 148 self._idle_for = max(1, sec_to_ms(get_usable_min_sleep_interval())) 149 else: 150 self._idle_for = max(1, sec_to_ms(next_event_after)) 151 152 153def bind_coro_explicit(wx_entity: Union[App, Frame, Dialog], coro: AnyWorker, *args, **kwargs): 154 i: Interface = current_interface() 155 coro_id: CoroID = i(PutCoro, coro, *args, **kwargs) 156 bind_running_coro(wx_entity, coro_id) 157 158 159def bind_coro_implicit(wx_entity: Union[App, Frame, Dialog], coro: AnyWorker, *args, **kwargs): 160 i: Interface = current_interface() 161 coro_id: CoroID = i(PutCoro, cs_coro(coro), *args, **kwargs) 162 bind_running_coro(wx_entity, coro_id) 163 164 165bind_coro = bind_coro_implicit 166 167 168async def abind_coro_explicit(wx_entity: Union[App, Frame, Dialog], coro: AnyWorker, *args, **kwargs): 169 i: Interface = current_interface() 170 coro_id: CoroID = await i(PutCoro, coro, *args, **kwargs) 171 bind_running_coro(wx_entity, coro_id) 172 173 174async def abind_coro_implicit(wx_entity: Union[App, Frame, Dialog], coro: AnyWorker, *args, **kwargs): 175 i: Interface = current_interface() 176 coro_id: CoroID = await i(PutCoro, cs_coro(coro), *args, **kwargs) 177 bind_running_coro(wx_entity, coro_id) 178 179 180abind_coro = abind_coro_implicit 181 182 183async def asyncio_bind_coro_explicit(wx_entity: Union[App, Frame, Dialog], coro: AnyWorker, *args, **kwargs): 184 coro_id: CoroID = await await_task_prim(PutCoro, coro, *args, **kwargs) 185 bind_running_coro(wx_entity, coro_id) 186 187 188async def asyncio_bind_coro_implicit(wx_entity: Union[App, Frame, Dialog], coro: AnyWorker, *args, **kwargs): 189 coro_id: CoroID = await await_task_prim(PutCoro, cs_coro(coro), *args, **kwargs) 190 bind_running_coro(wx_entity, coro_id) 191 192 193asyncio_bind_coro = asyncio_bind_coro_implicit 194 195 196def bind_asyncio_coro(wx_entity: Union[App, Frame, Dialog], awaitable: Awaitable): 197 coro_task: Task = create_task(awaitable) 198 bind_running_asyncio_coro(wx_entity, coro_task) 199 200 201def bind_running_coro(wx_entity: Union[App, Frame, Dialog], coro_id: CoroID): 202 if isinstance(wx_entity, Window): 203 wx_entity = GetTopLevelParent(wx_entity) 204 205 if isinstance(wx_entity, App): 206 wx_entity._binded_coros.add(coro_id) 207 elif isinstance(wx_entity, (Frame, Dialog)): 208 def close_binded_coro(event): 209 i: Interface = current_interface() 210 graceful_coro_destroyer(i, 0.1, coro_id, BindedEntityIsAboutToBeDestroyed) 211 event.Skip() 212 213 wx_entity.Bind(EVT_CLOSE, close_binded_coro) 214 else: 215 raise RuntimeError(f'wx_entity must be App, Frame, Dialog or a widget (Window) on a top level window, not {type(wx_entity)}') 216 217 218def bind_running_asyncio_coro(wx_entity: Union[App, Frame, Dialog], coro_future: Future): 219 if isinstance(wx_entity, App): 220 wx_entity._binded_coros.add(coro_future) 221 elif isinstance(wx_entity, (Frame, Dialog)): 222 def close_binded_asyncio_coro(event): 223 coro_future.cancel() 224 event.Skip() 225 226 wx_entity.Bind(EVT_CLOSE, close_binded_asyncio_coro) 227 228 229def wx_exec_in_coro(func: Callable) -> Callable: 230 @wraps(func) 231 def wrapper(*args, **kwargs): 232 cs: CoroSchedulerType = current_coro_scheduler() 233 cs.high_cpu_utilisation_mode = False 234 cs.use_internal_sleep = False 235 i: Interface = current_interface() 236 i(AsyncioLoopRequest().use_higher_level_sleep_manager()) 237 i(AsyncioLoopRequest().ensure_loop(interrupt_when_no_requests=True)) 238 i(AsyncioLoopRequest().turn_on_loops_intercommunication()) 239 240 app_or_tuple = func(*args, **kwargs) 241 on_exit = lambda ret: ret 242 if isinstance(app_or_tuple, App): 243 app = app_or_tuple 244 elif isinstance(app_or_tuple, tuple): 245 app, on_exit = app_or_tuple 246 else: 247 raise RuntimeError("wx_exec_in_coro must return either wx.App or (wx.App, on_exit) tuple") 248 249 # Run the event loop 250 ret = app.MainLoop() 251 ret = i(RunCoro, cs_coro(on_exit), ret) 252 i(ShutdownLoop) 253 return ret 254 255 return cs_coro(wrapper) 256 257 258def bind_to(wx_entity: Union[App, Frame, Dialog], wrapped_coro: Callable): 259 if not (hasattr(wrapped_coro, '__wrapped__') and hasattr(wrapped_coro, '_decorator_func')): 260 return wrapped_coro 261 262 coro: Union[AnyWorker, Coroutine] = wrapped_coro.__wrapped__ 263 decorator_func = wrapped_coro._decorator_func 264 new_wrapped_coro = decorator_func(coro) 265 # if hasattr(wrapped_coro, '__self__'): 266 # new_wrapped_coro = new_wrapped_coro.__get__(wrapped_coro.__self__, wrapped_coro.__self__.__class__) 267 268 new_wrapped_coro._binded_entity = wx_entity 269 if hasattr(wrapped_coro, '__self__'): 270 new_wrapped_coro.__self__ = wrapped_coro.__self__ 271 272 return new_wrapped_coro 273 274 275def bind(wrapped_coro: Callable): 276 if hasattr(wrapped_coro, '__self__'): 277 return bind_to(wrapped_coro.__self__, wrapped_coro) 278 else: 279 return wrapped_coro 280 281 282def event_handler_implicit(coro: AnyWorker): 283 def wrapper(*args, **kwargs): 284 coro_sign: Signature = signature(coro) 285 binded_entity = None 286 if 2 == len(coro_sign.parameters): 287 # The first parameter is self since event handler must have single parameter: event 288 # binded_entity = wrapper.__self__ 289 binded_entity = args[0] 290 291 i: Interface = current_interface() 292 if hasattr(wrapper, '_binded_entity') and hasattr(wrapper, '__self__'): 293 args = (wrapper.__self__,) + args 294 295 coro_id: CoroID = i(PutCoro, cs_coro(coro), *args, **kwargs) 296 if hasattr(wrapper, '_binded_entity'): 297 binded_entity = getattr(wrapper, '_binded_entity') 298 else: 299 if binded_entity is None: 300 binded_entity = App.Get() 301 302 bind_running_coro(binded_entity, coro_id) 303 304 wrapper.__wrapped__ = coro 305 wrapper._decorator_func = event_handler_implicit 306 return wrapper 307 308 309event_handler = event_handler_implicit 310aevent_handler = event_handler_implicit 311 312 313def event_handler_explicit(coro: AnyWorker): 314 @wraps(coro) 315 def wrapper(*args, **kwargs): 316 coro_sign: Signature = signature(coro) 317 binded_entity = None 318 if 3 == len(coro_sign.parameters): 319 # The first parameter is self since event handler must have single parameter: event 320 # binded_entity = wrapper.__self__ 321 binded_entity = args[1] 322 323 i: Interface = current_interface() 324 if hasattr(wrapper, '_binded_entity') and hasattr(wrapper, '__self__'): 325 args = (wrapper.__self__,) + args 326 327 coro_id: CoroID = i(PutCoro, coro, *args, **kwargs) 328 if hasattr(wrapper, '_binded_entity'): 329 binded_entity = getattr(wrapper, '_binded_entity') 330 else: 331 if binded_entity is None: 332 binded_entity = App.Get() 333 334 bind_running_coro(binded_entity, coro_id) 335 336 wrapper.__wrapped__ = coro 337 wrapper._decorator_func = event_handler_explicit 338 return wrapper 339 340 341def asyncio_event_handler(coro: Coroutine): 342 @wraps(coro) 343 def wrapper(*args, **kwargs): 344 coro_sign: Signature = signature(coro) 345 binded_entity = None 346 if 2 == len(coro_sign.parameters): 347 # The first parameter is self since event handler must have single parameter: event 348 # binded_entity = wrapper.__self__ 349 binded_entity = args[0] 350 351 if hasattr(wrapper, '_binded_entity') and hasattr(wrapper, '__self__'): 352 args = (wrapper.__self__,) + args 353 354 coro_task: Task = create_task(coro(*args, **kwargs)) 355 if hasattr(wrapper, '_binded_entity'): 356 binded_entity = getattr(wrapper, '_binded_entity') 357 else: 358 if binded_entity is None: 359 binded_entity = App.Get() 360 361 bind_running_asyncio_coro(binded_entity, coro_task) 362 363 wrapper.__wrapped__ = coro 364 wrapper._decorator_func = asyncio_event_handler 365 return wrapper 366 367 368def blocking_event_handler_implicit(coro: AnyWorker): 369 @wraps(coro) 370 def wrapper(*args, **kwargs): 371 i: Interface = current_interface() 372 if hasattr(wrapper, '_binded_entity') and hasattr(wrapper, '__self__'): 373 args = (wrapper.__self__,) + args 374 375 i(RunCoro, cs_coro(coro), *args, **kwargs) 376 377 wrapper.__wrapped__ = coro 378 wrapper._decorator_func = blocking_event_handler_implicit 379 return wrapper 380 381 382def blocking_event_handler_explicit(coro: AnyWorker): 383 @wraps(coro) 384 def wrapper(*args, **kwargs): 385 i: Interface = current_interface() 386 if hasattr(wrapper, '_binded_entity') and hasattr(wrapper, '__self__'): 387 args = (wrapper.__self__,) + args 388 389 i(RunCoro, coro, *args, **kwargs) 390 391 wrapper.__wrapped__ = coro 392 wrapper._decorator_func = blocking_event_handler_explicit 393 return wrapper
class
CoroApp(wx.core.App):
100class CoroApp(App): 101 def __init__(self, default_priority: CoroPriority = CoroPriority.normal): 102 self._cs: CoroSchedulerType = current_coro_scheduler() 103 self._cs.on_idle_handlers.add(self._on_system_loop_idle) 104 self._cs.idle_managers_num += 1 105 self._default_priority: CoroPriority = default_priority 106 self._ly = gly(self._default_priority) 107 self._binded_coros: Set = set() 108 self._idle_for: Optional[int] = None # in milliseconds 109 super(CoroApp, self).__init__() 110 111 def OnInit(self): 112 super().OnInit() 113 self._system_loop_timer = Timer(self) 114 self.Bind(EVT_TIMER, self._on_system_loop_timer, self._system_loop_timer) 115 self._system_loop_timer.StartOnce(1) # run every 1 millisecond 116 return True 117 118 def OnExit(self): 119 super().OnExit() 120 i: Interface = current_interface() 121 i.log.info('Application is exiting.') 122 i.log.info('Destroying all binded coros...') 123 for furure_or_coro_id in self._binded_coros: 124 if isinstance(furure_or_coro_id, Future): 125 furure_or_coro_id.cancel() 126 else: 127 graceful_coro_destroyer(i, 0.1, furure_or_coro_id, BindedEntityIsAboutToBeDestroyed) 128 129 i.log.info('All binded coroutines destroyed.') 130 self._cs.idle_managers_num -= 1 131 self._cs.on_idle_handlers.discard(self._on_system_loop_idle) 132 i.log.info('Application is exiting... Done.') 133 return super().OnExit() 134 135 def _on_system_loop_timer(self, event): 136 # from datetime import datetime 137 # print(f'Cengal Timer >> {datetime.now()}>>') 138 self._ly() 139 if self._idle_for is None: 140 self._system_loop_timer.StartOnce(1) # run every 1 millisecond 141 else: 142 # print(f'Idle for {self._idle_for} milliseconds.') 143 idle_for = self._idle_for 144 self._idle_for = None 145 self._system_loop_timer.StartOnce(idle_for) 146 147 def _on_system_loop_idle(self, next_event_after: Optional[RationalNumber]): 148 if next_event_after is None: 149 self._idle_for = max(1, sec_to_ms(get_usable_min_sleep_interval())) 150 else: 151 self._idle_for = max(1, sec_to_ms(next_event_after))
The wx.App class represents the application and is used to:
- bootstrap the wxPython system and initialize the underlying gui toolkit
- set and get application-wide properties
- implement the native windowing system main message or event loop, and to dispatch events to window instances
- etc.
Every wx application must have a single wx.App instance, and all
creation of UI objects should be delayed until after the wx.App object
has been created in order to ensure that the gui platform and wxWidgets
have been fully initialized.
Normally you would derive from this class and implement an OnInit
method that creates a frame and then calls self.SetTopWindow(frame),
however wx.App is also usable on its own without derivation.
:note: In Python the wrapper for the C++ class wxApp has been renamed tp
:class:wx.PyApp. This wx.App class derives from wx.PyApp, and is
responsible for handling the Python-specific needs for bootstrapping the
wxWidgets library and other Python integration related requirements.
CoroApp( default_priority: cengal.parallel_execution.coroutines.coro_standard_services.loop_yield.versions.v_0.loop_yield.CoroPriority = <CoroPriority.normal: 1>)
101 def __init__(self, default_priority: CoroPriority = CoroPriority.normal): 102 self._cs: CoroSchedulerType = current_coro_scheduler() 103 self._cs.on_idle_handlers.add(self._on_system_loop_idle) 104 self._cs.idle_managers_num += 1 105 self._default_priority: CoroPriority = default_priority 106 self._ly = gly(self._default_priority) 107 self._binded_coros: Set = set() 108 self._idle_for: Optional[int] = None # in milliseconds 109 super(CoroApp, self).__init__()
Construct a wx.App object.
:param redirect: Should sys.stdout and sys.stderr be
redirected? Defaults to False. If filename is None
then output will be redirected to a window that pops up
as needed. (You can control what kind of window is created
for the output by resetting the class variable
outputWindowClass to a class of your choosing.)
:param filename: The name of a file to redirect output to, if redirect is True.
:param useBestVisual: Should the app try to use the best
available visual provided by the system (only relevant on
systems that have more than one visual.) This parameter
must be used instead of calling SetUseBestVisual later
on because it must be set before the underlying GUI
toolkit is initialized.
:param clearSigInt: Should SIGINT be cleared? This allows the app to terminate upon a Ctrl-C in the console like other GUI apps will.
:note: You should override OnInit to do application initialization to ensure that the system, toolkit and wxWidgets are fully initialized.
def
OnInit(self):
111 def OnInit(self): 112 super().OnInit() 113 self._system_loop_timer = Timer(self) 114 self.Bind(EVT_TIMER, self._on_system_loop_timer, self._system_loop_timer) 115 self._system_loop_timer.StartOnce(1) # run every 1 millisecond 116 return True
OnInit(self) -> bool
def
OnExit(self):
118 def OnExit(self): 119 super().OnExit() 120 i: Interface = current_interface() 121 i.log.info('Application is exiting.') 122 i.log.info('Destroying all binded coros...') 123 for furure_or_coro_id in self._binded_coros: 124 if isinstance(furure_or_coro_id, Future): 125 furure_or_coro_id.cancel() 126 else: 127 graceful_coro_destroyer(i, 0.1, furure_or_coro_id, BindedEntityIsAboutToBeDestroyed) 128 129 i.log.info('All binded coroutines destroyed.') 130 self._cs.idle_managers_num -= 1 131 self._cs.on_idle_handlers.discard(self._on_system_loop_idle) 132 i.log.info('Application is exiting... Done.') 133 return super().OnExit()
OnExit(self) -> int
Inherited Members
- wx.core.App
- outputWindowClass
- stdioWin
- saveStdio
- OnPreInit
- SetTopWindow
- MainLoop
- RedirectStdio
- RestoreStdio
- SetOutputWindowAttributes
- InitLocale
- ResetLocale
- Get
- wx._core.PyApp
- GTKAllowDiagnosticsControl
- GTKSuppressDiagnostics
- GetAssertMode
- GetComCtl32Version
- GetDisplayMode
- GetExitOnFrameDelete
- GetLayoutDirection
- GetMacAboutMenuItemId
- GetMacExitMenuItemId
- GetMacHelpMenuTitleName
- GetMacPreferencesMenuItemId
- GetMainTopWindow
- GetTopWindow
- GetUseBestVisual
- IsActive
- IsDisplayAvailable
- MacHideApp
- MacNewFile
- MacOpenFile
- MacOpenFiles
- MacOpenURL
- MacPrintFile
- MacReopenApp
- OSXEnableAutomaticTabbing
- OSXIsGUIApplication
- OnInitGui
- OnRun
- SafeYield
- SafeYieldFor
- SetAssertMode
- SetDisplayMode
- SetExitOnFrameDelete
- SetMacAboutMenuItemId
- SetMacExitMenuItemId
- SetMacHelpMenuTitleName
- SetMacPreferencesMenuItemId
- SetNativeTheme
- SetUseBestVisual
- TryAfter
- TryBefore
- TopWindow
- UseBestVisual
- LayoutDirection
- ExitOnFrameDelete
- DisplayMode
- AssertMode
- wx._core.AppConsole
- DeletePendingEvents
- ExitMainLoop
- FilterEvent
- GetAppDisplayName
- GetAppName
- GetClassName
- GetInstance
- GetMainLoop
- GetTraits
- GetVendorDisplayName
- GetVendorName
- HasPendingEvents
- IsMainLoopRunning
- IsScheduledForDestruction
- OnEventLoopEnter
- OnEventLoopExit
- ProcessPendingEvents
- ResumeProcessingOfPendingEvents
- ScheduleForDestruction
- SetAppDisplayName
- SetAppName
- SetCLocale
- SetClassName
- SetInstance
- SetVendorDisplayName
- SetVendorName
- SuspendProcessingOfPendingEvents
- UsesEventLoop
- Yield
- Traits
- VendorName
- VendorDisplayName
- ClassName
- AppName
- AppDisplayName
- wx._core.EvtHandler
- AddFilter
- AddPendingEvent
- Connect
- Disconnect
- GetEvtHandlerEnabled
- GetNextHandler
- GetPreviousHandler
- IsUnlinked
- ProcessEvent
- ProcessEventLocally
- QueueEvent
- RemoveFilter
- SafelyProcessEvent
- SetEvtHandlerEnabled
- SetNextHandler
- SetPreviousHandler
- Unlink
- PreviousHandler
- NextHandler
- EvtHandlerEnabled
- Bind
- Unbind
- wx._core.Object
- Destroy
- GetClassInfo
- GetRefData
- IsSameAs
- Ref
- SetRefData
- UnRef
- RefData
- ClassInfo
- wx._core.EventFilter
- Event_Ignore
- Event_Processed
- Event_Skip
def
bind_coro_explicit( wx_entity: Union[wx.core.App, wx._core.Frame, wx._core.Dialog], coro: Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Awaitable[Any]]], *args, **kwargs):
def
bind_coro_implicit( wx_entity: Union[wx.core.App, wx._core.Frame, wx._core.Dialog], coro: Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Awaitable[Any]]], *args, **kwargs):
def
bind_coro( wx_entity: Union[wx.core.App, wx._core.Frame, wx._core.Dialog], coro: Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Awaitable[Any]]], *args, **kwargs):
async def
abind_coro_explicit( wx_entity: Union[wx.core.App, wx._core.Frame, wx._core.Dialog], coro: Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Awaitable[Any]]], *args, **kwargs):
async def
abind_coro_implicit( wx_entity: Union[wx.core.App, wx._core.Frame, wx._core.Dialog], coro: Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Awaitable[Any]]], *args, **kwargs):
async def
abind_coro( wx_entity: Union[wx.core.App, wx._core.Frame, wx._core.Dialog], coro: Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Awaitable[Any]]], *args, **kwargs):
async def
asyncio_bind_coro_explicit( wx_entity: Union[wx.core.App, wx._core.Frame, wx._core.Dialog], coro: Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Awaitable[Any]]], *args, **kwargs):
async def
asyncio_bind_coro_implicit( wx_entity: Union[wx.core.App, wx._core.Frame, wx._core.Dialog], coro: Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Awaitable[Any]]], *args, **kwargs):
async def
asyncio_bind_coro( wx_entity: Union[wx.core.App, wx._core.Frame, wx._core.Dialog], coro: Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Awaitable[Any]]], *args, **kwargs):
def
bind_asyncio_coro( wx_entity: Union[wx.core.App, wx._core.Frame, wx._core.Dialog], awaitable: Awaitable):
def
bind_running_coro( wx_entity: Union[wx.core.App, wx._core.Frame, wx._core.Dialog], coro_id: int):
202def bind_running_coro(wx_entity: Union[App, Frame, Dialog], coro_id: CoroID): 203 if isinstance(wx_entity, Window): 204 wx_entity = GetTopLevelParent(wx_entity) 205 206 if isinstance(wx_entity, App): 207 wx_entity._binded_coros.add(coro_id) 208 elif isinstance(wx_entity, (Frame, Dialog)): 209 def close_binded_coro(event): 210 i: Interface = current_interface() 211 graceful_coro_destroyer(i, 0.1, coro_id, BindedEntityIsAboutToBeDestroyed) 212 event.Skip() 213 214 wx_entity.Bind(EVT_CLOSE, close_binded_coro) 215 else: 216 raise RuntimeError(f'wx_entity must be App, Frame, Dialog or a widget (Window) on a top level window, not {type(wx_entity)}')
def
bind_running_asyncio_coro( wx_entity: Union[wx.core.App, wx._core.Frame, wx._core.Dialog], coro_future: _asyncio.Future):
219def bind_running_asyncio_coro(wx_entity: Union[App, Frame, Dialog], coro_future: Future): 220 if isinstance(wx_entity, App): 221 wx_entity._binded_coros.add(coro_future) 222 elif isinstance(wx_entity, (Frame, Dialog)): 223 def close_binded_asyncio_coro(event): 224 coro_future.cancel() 225 event.Skip() 226 227 wx_entity.Bind(EVT_CLOSE, close_binded_asyncio_coro)
def
wx_exec_in_coro(func: Callable) -> Callable:
230def wx_exec_in_coro(func: Callable) -> Callable: 231 @wraps(func) 232 def wrapper(*args, **kwargs): 233 cs: CoroSchedulerType = current_coro_scheduler() 234 cs.high_cpu_utilisation_mode = False 235 cs.use_internal_sleep = False 236 i: Interface = current_interface() 237 i(AsyncioLoopRequest().use_higher_level_sleep_manager()) 238 i(AsyncioLoopRequest().ensure_loop(interrupt_when_no_requests=True)) 239 i(AsyncioLoopRequest().turn_on_loops_intercommunication()) 240 241 app_or_tuple = func(*args, **kwargs) 242 on_exit = lambda ret: ret 243 if isinstance(app_or_tuple, App): 244 app = app_or_tuple 245 elif isinstance(app_or_tuple, tuple): 246 app, on_exit = app_or_tuple 247 else: 248 raise RuntimeError("wx_exec_in_coro must return either wx.App or (wx.App, on_exit) tuple") 249 250 # Run the event loop 251 ret = app.MainLoop() 252 ret = i(RunCoro, cs_coro(on_exit), ret) 253 i(ShutdownLoop) 254 return ret 255 256 return cs_coro(wrapper)
def
bind_to( wx_entity: Union[wx.core.App, wx._core.Frame, wx._core.Dialog], wrapped_coro: Callable):
259def bind_to(wx_entity: Union[App, Frame, Dialog], wrapped_coro: Callable): 260 if not (hasattr(wrapped_coro, '__wrapped__') and hasattr(wrapped_coro, '_decorator_func')): 261 return wrapped_coro 262 263 coro: Union[AnyWorker, Coroutine] = wrapped_coro.__wrapped__ 264 decorator_func = wrapped_coro._decorator_func 265 new_wrapped_coro = decorator_func(coro) 266 # if hasattr(wrapped_coro, '__self__'): 267 # new_wrapped_coro = new_wrapped_coro.__get__(wrapped_coro.__self__, wrapped_coro.__self__.__class__) 268 269 new_wrapped_coro._binded_entity = wx_entity 270 if hasattr(wrapped_coro, '__self__'): 271 new_wrapped_coro.__self__ = wrapped_coro.__self__ 272 273 return new_wrapped_coro
def
bind(wrapped_coro: Callable):
def
event_handler_implicit( coro: Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Awaitable[Any]]]):
283def event_handler_implicit(coro: AnyWorker): 284 def wrapper(*args, **kwargs): 285 coro_sign: Signature = signature(coro) 286 binded_entity = None 287 if 2 == len(coro_sign.parameters): 288 # The first parameter is self since event handler must have single parameter: event 289 # binded_entity = wrapper.__self__ 290 binded_entity = args[0] 291 292 i: Interface = current_interface() 293 if hasattr(wrapper, '_binded_entity') and hasattr(wrapper, '__self__'): 294 args = (wrapper.__self__,) + args 295 296 coro_id: CoroID = i(PutCoro, cs_coro(coro), *args, **kwargs) 297 if hasattr(wrapper, '_binded_entity'): 298 binded_entity = getattr(wrapper, '_binded_entity') 299 else: 300 if binded_entity is None: 301 binded_entity = App.Get() 302 303 bind_running_coro(binded_entity, coro_id) 304 305 wrapper.__wrapped__ = coro 306 wrapper._decorator_func = event_handler_implicit 307 return wrapper
def
event_handler( coro: Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Awaitable[Any]]]):
283def event_handler_implicit(coro: AnyWorker): 284 def wrapper(*args, **kwargs): 285 coro_sign: Signature = signature(coro) 286 binded_entity = None 287 if 2 == len(coro_sign.parameters): 288 # The first parameter is self since event handler must have single parameter: event 289 # binded_entity = wrapper.__self__ 290 binded_entity = args[0] 291 292 i: Interface = current_interface() 293 if hasattr(wrapper, '_binded_entity') and hasattr(wrapper, '__self__'): 294 args = (wrapper.__self__,) + args 295 296 coro_id: CoroID = i(PutCoro, cs_coro(coro), *args, **kwargs) 297 if hasattr(wrapper, '_binded_entity'): 298 binded_entity = getattr(wrapper, '_binded_entity') 299 else: 300 if binded_entity is None: 301 binded_entity = App.Get() 302 303 bind_running_coro(binded_entity, coro_id) 304 305 wrapper.__wrapped__ = coro 306 wrapper._decorator_func = event_handler_implicit 307 return wrapper
def
event_handler_explicit( coro: Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Awaitable[Any]]]):
314def event_handler_explicit(coro: AnyWorker): 315 @wraps(coro) 316 def wrapper(*args, **kwargs): 317 coro_sign: Signature = signature(coro) 318 binded_entity = None 319 if 3 == len(coro_sign.parameters): 320 # The first parameter is self since event handler must have single parameter: event 321 # binded_entity = wrapper.__self__ 322 binded_entity = args[1] 323 324 i: Interface = current_interface() 325 if hasattr(wrapper, '_binded_entity') and hasattr(wrapper, '__self__'): 326 args = (wrapper.__self__,) + args 327 328 coro_id: CoroID = i(PutCoro, coro, *args, **kwargs) 329 if hasattr(wrapper, '_binded_entity'): 330 binded_entity = getattr(wrapper, '_binded_entity') 331 else: 332 if binded_entity is None: 333 binded_entity = App.Get() 334 335 bind_running_coro(binded_entity, coro_id) 336 337 wrapper.__wrapped__ = coro 338 wrapper._decorator_func = event_handler_explicit 339 return wrapper
def
asyncio_event_handler(coro: Coroutine):
342def asyncio_event_handler(coro: Coroutine): 343 @wraps(coro) 344 def wrapper(*args, **kwargs): 345 coro_sign: Signature = signature(coro) 346 binded_entity = None 347 if 2 == len(coro_sign.parameters): 348 # The first parameter is self since event handler must have single parameter: event 349 # binded_entity = wrapper.__self__ 350 binded_entity = args[0] 351 352 if hasattr(wrapper, '_binded_entity') and hasattr(wrapper, '__self__'): 353 args = (wrapper.__self__,) + args 354 355 coro_task: Task = create_task(coro(*args, **kwargs)) 356 if hasattr(wrapper, '_binded_entity'): 357 binded_entity = getattr(wrapper, '_binded_entity') 358 else: 359 if binded_entity is None: 360 binded_entity = App.Get() 361 362 bind_running_asyncio_coro(binded_entity, coro_task) 363 364 wrapper.__wrapped__ = coro 365 wrapper._decorator_func = asyncio_event_handler 366 return wrapper
def
aevent_handler( coro: Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Awaitable[Any]]]):
283def event_handler_implicit(coro: AnyWorker): 284 def wrapper(*args, **kwargs): 285 coro_sign: Signature = signature(coro) 286 binded_entity = None 287 if 2 == len(coro_sign.parameters): 288 # The first parameter is self since event handler must have single parameter: event 289 # binded_entity = wrapper.__self__ 290 binded_entity = args[0] 291 292 i: Interface = current_interface() 293 if hasattr(wrapper, '_binded_entity') and hasattr(wrapper, '__self__'): 294 args = (wrapper.__self__,) + args 295 296 coro_id: CoroID = i(PutCoro, cs_coro(coro), *args, **kwargs) 297 if hasattr(wrapper, '_binded_entity'): 298 binded_entity = getattr(wrapper, '_binded_entity') 299 else: 300 if binded_entity is None: 301 binded_entity = App.Get() 302 303 bind_running_coro(binded_entity, coro_id) 304 305 wrapper.__wrapped__ = coro 306 wrapper._decorator_func = event_handler_implicit 307 return wrapper
def
blocking_event_handler_implicit( coro: Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Awaitable[Any]]]):
369def blocking_event_handler_implicit(coro: AnyWorker): 370 @wraps(coro) 371 def wrapper(*args, **kwargs): 372 i: Interface = current_interface() 373 if hasattr(wrapper, '_binded_entity') and hasattr(wrapper, '__self__'): 374 args = (wrapper.__self__,) + args 375 376 i(RunCoro, cs_coro(coro), *args, **kwargs) 377 378 wrapper.__wrapped__ = coro 379 wrapper._decorator_func = blocking_event_handler_implicit 380 return wrapper
def
blocking_event_handler_explicit( coro: Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, Awaitable[Any]]]):
383def blocking_event_handler_explicit(coro: AnyWorker): 384 @wraps(coro) 385 def wrapper(*args, **kwargs): 386 i: Interface = current_interface() 387 if hasattr(wrapper, '_binded_entity') and hasattr(wrapper, '__self__'): 388 args = (wrapper.__self__,) + args 389 390 i(RunCoro, coro, *args, **kwargs) 391 392 wrapper.__wrapped__ = coro 393 wrapper._decorator_func = blocking_event_handler_explicit 394 return wrapper