pycroglia.core.compute package

Subpackages

Submodules

class pycroglia.core.compute.backend.Backend(*args, **kwargs)[source]

Bases: Protocol

Protocol for computation backends.

Defines a common interface for running tasks, regardless of whether they are executed via PyQt, multiprocessing, or another concurrency model.

cancel()[source]

Cancels execution of all submitted tasks.

Return type:

None

join()[source]

Block until all submitted tasks are finished.

Return type:

None

run()[source]

Start execution of all submitted tasks.

Return type:

None

submit(task, on_result, on_error=None, on_finish=None)[source]

Submit a task to the backend.

Parameters:

  • task (Task) – Task to be executed.

  • on_result (Callable) – Callback for successful results.

  • on_error (Callable | None) – Optional callback for errors.

  • on_finish (Callable | None) – Optional callback when task finishes.

Return type:

None

class pycroglia.core.compute.backend.Task(*args, **kwargs)[source]

Bases: Protocol

Protocol for executable tasks.

Any class implementing this protocol must define a run method that returns a dictionary with computation results.

run()[source]

Execute the task and return its result.

Returns:

Computation result as a dictionary.

Return type:

dict[str, Any]

class pycroglia.core.compute.computable.Computable[source]

Bases: ABC

Abstract base class for all computable objects.

Classes inheriting from Computable must implement the compute method. The compute method executes a specific computation and returns its results as a dictionary, allowing flexible storage of heterogeneous values.

abstract compute()[source]

Execute the computation and return its results.

Returns:

A dictionary containing the results of the computation. Keys should be descriptive strings (e.g., “cell_id”, “volume”, “centroid_distance”), and values can be of any type depending on the computation.

Return type:

dict[str, Any]

class pycroglia.core.compute.mp_pool.CancelFlag[source]

Bases: object

Lightweight cooperative cancellation flag for multiprocessing tasks.

This class provides a simple shared mechanism to signal cancellation across multiple worker processes using a multiprocessing.Manager.Value. Each Computable should periodically check the flag to terminate gracefully.

Example

```python flag = CancelFlag() if flag.is_set():

return # Stop early

```

_cancelled

Shared boolean indicating cancellation state.

Type:

mp.Value

is_set()[source]

Check whether cancellation has been requested.

Returns:

True if cancellation has been requested, False otherwise.

Return type:

bool

set()[source]

Signal all workers to cancel cooperatively.

Return type:

None

class pycroglia.core.compute.mp_pool.MPPool(processes=None)[source]

Bases: object

Multiprocessing pool manager for executing Computable tasks.

Provides submission, execution, and completion tracking for multiple concurrent tasks.

Parameters:

processes (int | None)

all_finished

Optional callback invoked when all submitted tasks have completed.

Type:

Callable[[], None] | None

cancel()[source]

Request cooperative cancellation for all running tasks.

Sets the shared cancellation flag. All tasks that periodically check it will terminate gracefully. Pending tasks are ignored.

Return type:

None

join()[source]

Block until all tasks are finished.

Return type:

None

run()[source]

API symmetry: in multiprocessing tasks start immediately.

Return type:

None

submit(computable, on_result, on_error=None, on_finish=None)[source]

Submit a Computable task to the pool.

Parameters:

  • computable (Computable) – The computation object to execute.

  • on_result (Callable[[dict[str, Any]], None]) – Callback for result data.

  • on_error (Callable[[str, Exception], None], optional) – Callback for errors.

  • on_finish (Callable[[str], None], optional) – Callback when task finishes.

Return type:

None

class pycroglia.core.compute.mp_pool.MPTask(computable, cancel_flag)[source]

Bases: object

A multiprocessing-compatible task that executes a Computable.

Parameters:
task_id

Unique identifier for this task.

Type:

str

computable

The computation object to run.

Type:

Computable

cancel_flag

Shared cancellation flag.

Type:

CancelFlag

run()[source]

Execute the computable and return its result.

The Computable is responsible for checking cancel_flag.is_set() periodically and exiting early if cancellation is requested.

Returns:

Computation result dictionary.

Return type:

dict[str, Any]

class pycroglia.core.compute.pool.Pool(backend)[source]

Bases: object

Unified Pool that delegates to a specific backend.

This class acts as a façade: it exposes a single API (submit, run, join) while delegating execution to the provided backend (e.g. QPool, MPPool).

Parameters:

backend (Backend)

backend

The backend used for task execution.

Type:

Backend

cancel()[source]

Cancels execution of all submitted tasks.

Return type:

None

join()[source]

Block until all tasks are finished.

Return type:

None

run()[source]

Start execution of all submitted tasks.

Return type:

None

submit(task, on_result, on_error=None, on_finish=None)[source]

Submit a task to the backend.

Parameters:

  • task (Task) – Task to execute.

  • on_result (Callable[[dict[str, Any]], None]) – Callback for results.

  • on_error (Callable[[str, Exception], None], optional) – Callback for errors.

  • on_finish (Callable[[str], None], optional) – Callback when the task finishes.

Return type:

None

class pycroglia.core.compute.qt_pool.CancelFlag[source]

Bases: object

Lightweight cooperative cancellation flag for task management.

This class provides a simple mechanism for signalling cooperative cancellation between a task backend (e.g., QThreadPool) and the tasks it executes.

cancelled

Internal state indicating whether cancellation has been requested.

Type:

bool

is_set()[source]

Check whether cancellation has been requested.

Returns:

True if cancellation has been requested, False otherwise.

Return type:

bool

set()[source]

Request cooperative cancellation.

Sets the flag to True, signalling that any ongoing tasks should stop as soon as possible.

Return type:

None

class pycroglia.core.compute.qt_pool.QPool[source]

Bases: QObject

Thread pool manager for executing Computable tasks.

Provides submission, execution, and completion tracking for multiple concurrent tasks.

all_finished

Emitted when all submitted tasks have completed.

Type:

pyqtSignal

all_finished

str = …, revision: int = …, arguments: Sequence = …) -> PYQT_SIGNAL

types is normally a sequence of individual types. Each type is either a type object or a string that is the name of a C++ type. Alternatively each type could itself be a sequence of types each describing a different overloaded signal. name is the optional C++ name of the signal. If it is not specified then the name of the class attribute that is bound to the signal is used. revision is the optional revision of the signal that is exported to QML. If it is not specified then 0 is used. arguments is the optional sequence of the names of the signal’s arguments.

Type:

pyqtSignal(*types, name

cancel()[source]

Cancels execution of all submitted tasks.

join()[source]

Block until all tasks are finished.

Warning

This method blocks the main (GUI) thread if called inside a running Qt event loop. Prefer connecting to all_finished for a non-blocking alternative.

Return type:

None

run()[source]

Start all submitted tasks asynchronously.

Return type:

None

submit(computable, on_result, on_error=None, on_finish=None)[source]

Submit a Computable task to the pool.

Parameters:

  • computable (Computable) – The computation object to execute.

  • on_result (Callable[[dict[str, Any]], None]) – Callback for result data.

  • on_error (Callable[[str, Exception], None], optional) – Callback for errors.

  • on_finish (Callable[[str], None], optional) – Callback when task finishes.

class pycroglia.core.compute.qt_pool.QTask(computable, flag)[source]

Bases: QRunnable

A runnable task that executes a Computable object in a thread.

Parameters:
task_id

Unique identifier for this task.

Type:

str

computable

The computation object to run.

Type:

Computable

signals

Signal manager for result, error, and finished.

Type:

TaskSignals

run()[source]

Execute the task, unless the cancel flag is set.

Runs the compute method of the associated Computable. Emits:

  • result: when computation succeeds.

  • error: if an exception is raised.

  • finished: always, when the task ends.

class pycroglia.core.compute.qt_pool.QTaskSignal[source]

Bases: QObject

Signals available from a running Task.

result

Emitted with the computation result as a dict.

Type:

pyqtSignal

error

Emitted with (task_id, exception) if computation fails.

Type:

pyqtSignal

finished

Emitted with task_id when the task completes.

Type:

pyqtSignal

error

str = …, revision: int = …, arguments: Sequence = …) -> PYQT_SIGNAL

types is normally a sequence of individual types. Each type is either a type object or a string that is the name of a C++ type. Alternatively each type could itself be a sequence of types each describing a different overloaded signal. name is the optional C++ name of the signal. If it is not specified then the name of the class attribute that is bound to the signal is used. revision is the optional revision of the signal that is exported to QML. If it is not specified then 0 is used. arguments is the optional sequence of the names of the signal’s arguments.

Type:

pyqtSignal(*types, name

finished

str = …, revision: int = …, arguments: Sequence = …) -> PYQT_SIGNAL

types is normally a sequence of individual types. Each type is either a type object or a string that is the name of a C++ type. Alternatively each type could itself be a sequence of types each describing a different overloaded signal. name is the optional C++ name of the signal. If it is not specified then the name of the class attribute that is bound to the signal is used. revision is the optional revision of the signal that is exported to QML. If it is not specified then 0 is used. arguments is the optional sequence of the names of the signal’s arguments.

Type:

pyqtSignal(*types, name

result

str = …, revision: int = …, arguments: Sequence = …) -> PYQT_SIGNAL

types is normally a sequence of individual types. Each type is either a type object or a string that is the name of a C++ type. Alternatively each type could itself be a sequence of types each describing a different overloaded signal. name is the optional C++ name of the signal. If it is not specified then the name of the class attribute that is bound to the signal is used. revision is the optional revision of the signal that is exported to QML. If it is not specified then 0 is used. arguments is the optional sequence of the names of the signal’s arguments.

Type:

pyqtSignal(*types, name

Module contents