@PyWoody The proof of the pudding is in the eating (meshing or not)
NB I still muse about the use case, but the responsiveness hasn’t suffered (CallAfter & the like)
PS well, a rough ride gets the Layout into a tangle and, I’m afraid, must be queued! (that’s where the convenience of CallAfter is unbeatable)
and, naturally, there is an almost C & P version for the PPE
(but it’s always good to keep separate concerns in different modules before the project grows)
PPE.py (5.1 KB) task.py (634 Bytes)
import threading as thd
from concurrent.futures import ThreadPoolExecutor
import time
import random
import wx
def task(abort, gauge):
sleep = random.randint(1, 10) / 10
rate = int(random.randint(10, 100) / 10)
completed = 0
while completed < 100:
time.sleep(sleep)
completed = completed + rate
gauge.SetValue(completed)
if abort.is_set():
break
return completed
def sizer_layout(sizer):
sizer.Layout()
class Gui(wx.Frame):
def __init__(self):
super().__init__(None, wx.ID_ANY, size=(600, 400),
title='Thread Executor Testing')
self.sb = self.CreateStatusBar(style=wx.SB_FLAT)
self.sb_timer = None
self.pnl = wx.Panel(self)
vbox = wx.BoxSizer(wx.VERTICAL)
self.info = wx.InfoBar(self.pnl)
vbox.Add(self.info, 0, wx.EXPAND)
hbox = wx.BoxSizer(wx.HORIZONTAL)
self.btn_add_task = wx.Button(self.pnl, wx.ID_ANY, 'Add Task')
self.btn_add_task.Bind(wx.EVT_BUTTON, self.add_task)
hbox.Add(self.btn_add_task)
btn = wx.Button(self.pnl, wx.ID_ANY, 'Thread Info')
btn.Bind(wx.EVT_BUTTON, self.thread_info)
hbox.Add(btn)
self.btn_abort = wx.Button(self.pnl, wx.ID_ANY, 'Abort?')
self.btn_abort.Bind(wx.EVT_BUTTON, self.abort)
hbox.Add(self.btn_abort, 0, wx.LEFT, 50)
vbox.Add(hbox)
self.results_vbox = wx.BoxSizer(wx.VERTICAL)
vbox.Add(self.results_vbox, 0, wx.EXPAND|wx.LEFT, 15)
self.pnl.SetSizer(vbox)
self.Bind(wx.EVT_CLOSE, self.evt_close)
self.task_num = 0
self.executor = ThreadPoolExecutor(max_workers=3)
self.futures = {} # future: sizer
self.evt_abort = thd.Event()
self.Show()
self.btn_abort.Hide()
def add_task(self, _):
hbox = wx.BoxSizer(wx.HORIZONTAL)
gauge = wx.Gauge(self.pnl)
hbox.Add(gauge, 1, wx.EXPAND|wx.RIGHT, 35)
self.task_num += 1
hbox.Add(wx.StaticText(
self.pnl, label=f'task: {self.task_num}'),
0, wx.EXPAND|wx.RIGHT, 15)
self.results_vbox.Add(hbox, 0, wx.EXPAND)
self.pnl.Layout()
f = self.executor.submit(task, self.evt_abort, gauge)
f.add_done_callback(self.done)
self.futures[f] = hbox
def done(self, future):
if sizer := self.futures.get(future):
if future.cancelled():
val = 0
txt = 'cancelled'
else:
res = future.result()
if res < 100:
val = res
txt = 'aborted'
else:
val = 100
txt = 'COMPLETED'
ctrl = sizer.GetChildren()
ctrl[0].GetWindow().SetValue(val)
ctrl[1].GetWindow().SetLabel(txt)
wx.CallAfter(sizer_layout, sizer)
if val == 100 and not self.still_running():
self.btn_abort.Hide()
def thread_info(self, _):
for entry in thd.enumerate():
print(entry)
def abort(self, _):
self.btn_abort.Disable()
self.evt_abort.set()
if self.sb_timer:
self.sb_timer.cancel()
self.btn_abort.SetLabel('ABORTED !!!')
self.sb.SetBackgroundColour('yellow')
self.sb.SetStatusText("that's it!")
def evt_close(self, _):
if self.btn_add_task.IsEnabled():
self.btn_add_task.Disable()
for entry in self.futures:
entry.cancel()
if self.still_running():
if not self.sb_timer:
self.btn_abort.Show()
self.sb.SetBackgroundColour('red')
self.sb.SetStatusText('tasks are still running..')
self.sb_timer = thd.Timer(1.5, self.auto_reset_sb)
self.sb_timer.start()
else:
if self.sb_timer:
self.sb_timer.cancel()
self.thread_info(None)
self.executor.shutdown(cancel_futures=True)
self.info.SetShowHideEffects(
wx.SHOW_EFFECT_ROLL_TO_BOTTOM,
wx.SHOW_EFFECT_NONE)
self.info.SetEffectDuration(2000)
self.info.ShowMessage(
'Executor shut down..', wx.ICON_INFORMATION)
self.thread_info(None)
self.Destroy()
def still_running(self):
for entry in self.futures:
if not entry.done():
return True
def auto_reset_sb(self):
self.sb.SetBackgroundColour(None)
self.sb.SetStatusText('')
self.sb_timer = None
app = wx.App()
Gui()
app.MainLoop()