Toolkits and running headless

Each TraitsExecutor requires a running event loop to communicate results from background tasks to the foreground. In a typical GUI application, that event loop will be the event loop from the current GUI toolkit - for example Qt or Wx. However, one can also use the main thread asyncio event loop from Python’s standard library. This is potentially useful when writing automated tests, or when using Traits Futures in a headless setting (for example within a compute job).

Specifying a toolkit

To explicitly specify which toolkit to use, you need to provide the event_loop parameter when instantiating the TraitsExecutor. The library currently provides four different event loops: AsyncioEventLoop, QtEventLoop, WxEventLoop and ETSEventLoop.

By default, if no event loop is explicitly specified, an instance of ETSEventLoop is used. This follows the usual ETS rules to determine which toolkit to use based on the value of the ETS_TOOLKIT environment variable, on whether any other part of the ETS machinery has already “fixed” the toolkit, and on which toolkits are available in the current Python environment.

Running Traits Futures in a headless setting

In general, if you’re writing code that’s not GUI-oriented, you probably don’t want to be using Traits Futures at all: the library is explicitly designed for working in a GUI-based setting, in situations where you don’t want your computational or other tasks to block the GUI event loop and generate the impression of an unresponsive GUI. Instead, you might execute your tasks directly in the main thread or, if you need to take advantage of thread-based parallelism, use the concurrent.futures framework directly.

However, you may find yourself in a situation where you already have Traits Futures-based code that was written for a GUI setting, but that you want to be able to execute in a environment that doesn’t have the Qt or Wx toolkits available. In that case, Traits Futures can use the AsyncioEventLoop to deliver results to the main thread’s asyncio event loop instead of to a GUI framework’s event loop.

Here’s an example script that uses the AsyncioEventLoop in order to execute Traits Futures tasks within the context of an asyncio event loop.

# (C) Copyright 2018-2021 Enthought, Inc., Austin, TX
# All rights reserved.
#
# This software is provided without warranty under the terms of the BSD
# license included in LICENSE.txt and may be redistributed only under
# the conditions described in the aforementioned license. The license
# is also available online at http://www.enthought.com/licenses/BSD.txt
#
# Thanks for using Enthought open source!

"""
Running Traits Futures without a GUI, using the asyncio event loop.
"""

import asyncio
import random

from traits_futures.api import (
    AsyncioEventLoop,
    submit_iteration,
    TraitsExecutor,
)


def approximate_pi(sample_count=10 ** 8, report_interval=10 ** 6):
    """
    Yield successive approximations to π via Monte Carlo methods.
    """
    # approximate pi/4 by throwing points at a unit square and
    # counting the proportion that land in the quarter circle.
    inside = total = 0
    for i in range(sample_count):
        if i > 0 and i % report_interval == 0:
            yield 4 * inside / total  # <- partial result
        x, y = random.random(), random.random()
        inside += x * x + y * y < 1
        total += 1
    return 4 * inside / total


async def future_wrapper(traits_future):
    """
    Wrap a Traits Futures future as a schedulable coroutine.
    """

    def set_result(event):
        traits_future = event.object
        asyncio_future.set_result(traits_future.result)

    # Once we can assume a minimum Python version of 3.7, this should
    # be changed to use get_running_event_loop instead of get_event_loop.
    asyncio_future = asyncio.get_event_loop().create_future()

    traits_future.observe(set_result, "done")

    return await asyncio_future


def print_progress(event):
    """
    Progress reporter for the π calculation.
    """
    print(f"π is approximately {event.new:.6f}")


if __name__ == "__main__":
    traits_executor = TraitsExecutor(event_loop=AsyncioEventLoop())
    traits_future = submit_iteration(traits_executor, approximate_pi)
    traits_future.observe(print_progress, "result_event")

    # For Python 3.7 and later, just use asyncio.run.
    asyncio.get_event_loop().run_until_complete(future_wrapper(traits_future))