plesty.server.view.workers

Run presenter calls off the GUI thread and hand the result back as a signal.

A Bench call may block for seconds — start waits for the probe, stop for the escalation — and Qt paints nothing while its thread is busy. Worker puts the call on a thread pool and emits done(result) or failed(message) on the GUI thread, where the window connects its slots.

Classes

`_Signals`
`_Task`
`Worker`	A small pool running one presenter call per task.

Module Contents

class plesty.server.view.workers._Signals

Bases: PySide6.QtCore.QObject

done

failed

class plesty.server.view.workers._Task(action: collections.abc.Callable[[], Any], signals: _Signals)

Bases: PySide6.QtCore.QRunnable

Parameters:

action (collections.abc.Callable[[], Any])
signals (_Signals)

_action

_signals

run() → None

Return type:: None

class plesty.server.view.workers.Worker(parent: PySide6.QtCore.QObject | None = None, threads: int = 4)

Bases: PySide6.QtCore.QObject

A small pool running one presenter call per task.

Usage:

worker.run(lambda: bench.start(name), on_done=self._show_status, on_error=self._alert)

Create the pool.

Parameters:

parent (PySide6.QtCore.QObject | None) – Qt parent.
threads (int) – Concurrent tasks — more than one so a long stop does not block a status refresh.

_pool

_live: list[_Signals] = []

run(action: collections.abc.Callable[[], Any], on_done: collections.abc.Callable[[Any], None] | None = None, on_error: collections.abc.Callable[[str], None] | None = None) → None

Run action on the pool; call on_done or on_error on the GUI thread.

Parameters:

action (collections.abc.Callable[[], Any])
on_done (collections.abc.Callable[[Any], None] | None)
on_error (collections.abc.Callable[[str], None] | None)

Return type:

None

wait(timeout_ms: int = 30000) → bool

Block until every task finished (tests); True when none is left.

Parameters:: timeout_ms (int)
Return type:: bool

property busy: bool

Whether any task is still running.

Return type:: bool