Utility functions

edifice.utilities.alert(message, choices=None)

Displays a message in an alert box.

If choices is specified, the alert box contain a list of buttons showing each of the choices, and this function will return the user’s choice.

Parameters:

• message (str) – message to display

• choices (Optional[Sequence[str]]) – optional list of choice texts, which will be displayed as buttons.

Return type:

int | None

Returns:

Index of chosen option.

edifice.utilities.file_dialog(caption='', directory='', file_filter=None)

Displays a file choice dialog.

Parameters:

• caption (str) – the file dialog’s caption

• directory (str) – starting directory for the file dialog

• file_filter (Optional[Sequence[str]]) –

  Sequence of allowed file extensions. For example:

  "*.cpp *.cc *.C *.c++"
  "C++ files (*.cpp *.cc *.C *.c++)"
  

  are both valid ways of specifying a file filter.

Return type:

str | None

Returns:

Path of chosen file

edifice.utilities.palette_dump(palette, palette_compare=None)

Dump the palette to the console.

Parameters:

• palette (QPalette) – Dump the palette to the console.

• palette_compare (Optional[QPalette]) – Compare the palette with another palette if palette_compare is not None.

Return type:

None

edifice.utilities.palette_edifice_dark()

Edifice dark theme palette. This is the Qt Linux default dark theme palette, with some adjustments.

Return type:

QPalette

edifice.utilities.palette_edifice_light()

Edifice light theme palette. This is the Qt Linux default light theme palette, with some adjustments.

Return type:

QPalette

async edifice.utilities.run_subprocess_with_callback(subprocess, callback)

Run an async subprocess in a ProcessPoolExecutor and return the result.

The advantage of run_subprocess_with_callback() is that it behaves well and cleans up properly in the event of exceptions and cancellation. This function is useful for a long-running parallel worker subprocess for which we want to report progress back to the main GUI event loop. Like pytorch stuff.

Parameters:

• subprocess (Callable[[Callable[[ParamSpec(_P_callback)], None]], Awaitable[TypeVar(_T_subprocess)]]) –

  The async function to run in a ProcessPoolExecutor. The subprocess function will run in a sub-process in a new event loop. This subprocess function takes a single argument: a function with the same type as the callback function. The subprocess function must be picklable.

• callback (Callable[[ParamSpec(_P_callback)], None]) – The callback function to pass to the subprocess when it starts. This function will run in the main process event loop. All of the arguments to the callback function must be picklable.

Return type:

TypeVar(_T_subprocess)

The subprocess will be started when run_subprocess_with_callback() is entered, and the subprocess is guaranteed to be terminated when run_subprocess_with_callback() completes.

The subprocess will be started with the ‘spawn’ start method, so it will not inherit any file handles from the calling process.

While the subprocess is running, it may call the supplied callback function. The callback function will run in the main event loop of the calling process.

If this async run_subprocess_with_callback() function is cancelled, the subprocess will be terminated by calling Process.terminate(). Termination of the subprocess will occur even if the subprocess is blocked. Note that CancelledError will not be raised in the subprocess, instead the subprocess will be terminated immediately. If you want to perform sudden cleanup and halt of the subprocess then send it a message as in the below “Example of Queue messaging.”

Exceptions raised in the subprocess will be re-raised from run_subprocess_with_callback().

Exceptions raised in the callback will be suppressed.

Example

async def my_subprocess(callback: Callable[[int], None]) -> str:
    # This function will run in a subprocess in a new event loop.
    callback(1)
    await asyncio.sleep(1)
    callback(2)
    await asyncio.sleep(1)
    callback(3)
    return "done"

def my_callback(x:int) -> None:
    # This function will run in the main process event loop.
    print(f"callback {x}")

async def main() -> None:
    y = await run_subprocess_with_callback(my_subprocess, my_callback)

    # If this main() function is cancelled while awaiting the
    # subprocess then the subprocess will be terminated.

    print(f"my_subprocess returned {y}")

Note

Because “only picklable objects can be executed” by a ProcessPoolExecutor, we cannot pass a local function as the subprocess. The best workaround is to define at the module top-level a subprocess function which takes all its parameters as arguments, and then use functools.partial to bind local values to the subprocess parameters.

The callback does not have this problem; we can pass a local function as the callback.

The run_subprocess_with_callback() function provides a callback function for messaging back up to the main process, but it does not provide a built-in way to message down to the subprocess. To accomplish this we can create and pass a messaging object to the subprocess, for example a multiprocessing.managers.SyncManager.Queue.

Example of Queue messaging from the main process to the subprocess

async def my_subprocess(
    # This function will run in a subprocess in a new event loop.
    queue: queue.Queue[str],
    callback: typing.Callable[[int], None],
) -> str:
    while (msg := queue.get()) != "finish":
        callback(len(msg))
    return "done"

async def main() -> None:
    with multiprocessing.Manager() as manager:
        msg_queue: queue.Queue[str] = manager.Queue()

        def local_callback(x:int) -> None:
            # This function will run in the main process event loop.
            print(f"callback {x}")

        async def send_messages() -> None:
            msg_queue.put("one")
            msg_queue.put("finish")

        y, _ = await asyncio.gather(
            run_subprocess_with_callback(
                functools.partial(my_subprocess, msg_queue),
                local_callback,
            ),
            send_messages())
        )

        print(f"my_subprocess returned {y}")

edifice.utilities.set_trace()

Set a tracepoint in the Python debugger that works with PyQt.

PDB does not work well with PyQt applications. edifice.set_trace() is equivalent to pdb.set_trace(), but it can properly pause the PyQt event loop to enable use of the debugger (users of PySide need not worry about this).

edifice.utilities.theme_is_light()

Detect the operating environment theme.

True if light theme, false if dark theme.

Example:

def initializer():
    palette = palette_edifice_light() if theme_is_light() else palette_edifice_dark()
    cast(QApplication, QApplication.instance()).setPalette(palette)
    return palette

palette, _ = ed.use_state(initializer)

with Window():

Return type:

bool