Advanced Usage

Expression Representation

Basic aspects of expression representation are covered in basic usages.

Contexts

In the Wolfram Language, symbols live in contexts, which serve as a namespace mechanism. Built-in functions belong to the System` context. User-defined symbols and functions are stored by default in the Global` context.

Wolfram Language expressions are represented in Python using the attributes of the factory wl:

>>> from wolframclient.language import wl
>>> wl.Range(3)
Range[3]

The attributes of the factory wl do not have a context attached:

>>> wl.myFunction(1)
myFunction[1]

The representation of a given expression using its string InputForm is readable but ambiguous since the context is resolved in the kernel during evaluation, and as such different kernels may return different results. See $ContextPath.

On the other hand, in the WXF format, expressions are represented with their full names. Context must be fully specified for all symbols, the exception being the System` context, which can be omitted. As a consequence, in WXF, a symbol with no context is always deserialized as a System` symbol.

The method evaluate() accepts two types of input:

• Python strings are treated as a string InputForm; as such, context is resolved during evaluation. User-defined functions are most of the time automatically created with the Global` context.

• Serializable Python objects are serialized to WXF before evaluation; as such, context must be explicitly specified, except for System` symbols.

System Context

In order to explicitly represent a system symbol in Python, the factory System can be used. First import it:

>>> from wolframclient.language import System

Create a Python object representing the built-in function Classify:

>>> System.Classify
System`Classify

Global Context

User-defined functions and variables are associated to the Global` context by default. The factory Global can be used to represent those symbols:

>>> from wolframclient.language import Global
>>> Global.f
Global`f

Arbitrary Contexts

The factory wl can be used to build symbols with arbitrary context, as well as subcontexts:

>>> wl.Developer.PackedArrayQ
Developer`PackedArrayQ

>>> wl.Global.f
Global`f

>>> wl.System.Predict
System`Predict

>>> wl.MyContext.MySubContext.myFunction
MyContext`MySubContext`myFunction

Use Cases

Create a new function max that takes a list of strings and returns the longer ones. The function definition is:

max[s:List[__String]] := MaximalBy[s, StringLength]

Set up a new local evaluator session:

>>> from wolframclient.evaluation import WolframLanguageSession
>>> from wolframclient.language import wl, Global, wlexpr
>>> session = WolframLanguageSession()

Most of the time, it is much more convenient to define functions using an InputForm string expression rather than wl. Define the function max for the current session:

>>> session.evaluate(wlexpr('max[s : List[__String]] := MaximalBy[s, StringLength]'))

Apply function Global.max to a list of strings:

>>> session.evaluate(Global.max(['hello', 'darkness', 'my', 'old', 'friend']))
['darkness']

Trying to evaluate wl.max, which is the undefined symbol System`max, leads to an unevaluated expression:

>>> session.evaluate(wl.max(['hello', 'darkness', 'my', 'old', 'friend']))
max[['hello', 'darkness', 'my', 'old', 'friend']]

It is important to understand that wlexpr() applied to a string is equivalent to evaluating ToExpression on top of the string input, and as such some context inference is performed when evaluating. In the above example, the max function’s explicit name is Global`max. When the Python object wl.max is passed to evaluate(), it is serialized to WXF first, which has strict context specification rules; the only context that can be omitted is the System` context. As a consequence, any symbol without context is attached to the System` context; max is thus System`max, which is not defined.

Finally, terminate the session:

>>> session.stop()

Local Kernel Evaluation

The following sections provide executable demonstrations of the local evaluation features of the client library.

Note

all examples require a local Wolfram Engine installed in the default location.

Evaluation Methods

Synchronous

Initialize a session:

>>> from wolframclient.evaluation import WolframLanguageSession
>>> from wolframclient.language import wlexpr
>>> session=WolframLanguageSession()

Expressions involving scoped variables are usually more easily represented with wlexpr(). Compute an integral:

>>> session.evaluate(wlexpr('NIntegrate[Sqrt[x^2 + y^2 + z^2], {x, 0, 1}, {y, 0, 1}, {z, 0, 1}]'))
0.9605920064034617

Messages may be issued during evaluation. By default, the above evaluation methods log error messages with severity warning. It usually results in the message being printed out. It is also possible to retrieve both the evaluation result and the messages, wrapped in an instance of WolframKernelEvaluationResult, by using evaluate_wrap():

>>> eval = session.evaluate_wrap('1/0')
>>> eval.result
DirectedInfinity[]

Messages are stored as tuples of two elements, the message name and the formatted message:

>>> eval.messages
[('Power::infy', 'Infinite expression Power[0, -1] encountered.')]

Asynchronous

Some computations may take a significant time to finish. The library provides various form of control on evaluations.

Evaluate Future

Evaluation methods all have a future-based counterpart:

>>> from wolframclient.evaluation import WolframLanguageSession
>>> session = WolframLanguageSession()
>>> future = session.evaluate_future('Pause[3]; 1+1')

The future object is immediately returned; the computation is done in the background. Return the evaluated expression:

>>> future.result()
2

Sometimes a fine control over the maximum duration of an evaluation is required. TimeConstrained ensures that a given evaluation duration is not exceeding a timeout in seconds. When the timeout is reached, the symbol $Aborted is returned.

Wrap an artificially long evaluation to last at most one second:

>>> long_eval = wl.Pause(10)
>>> timeconstrained_eval = wl.TimeConstrained(long_eval, 1)

Evaluate the time-constrained expression:

>>> result = session.evaluate(timeconstrained_eval)

Check if the result is $Aborted:

>>> result.name == '$Aborted'
True

Terminate the session:

>>> session.terminate()

Concurrent Future

Sometimes, the result is not required immediately. Asynchronous evaluation is a way to start evaluations on a local kernel as a background task, without blocking the main Python execution. Asynchronous evaluation methods are evaluate_future(), evaluate_wrap_future() and evaluate_wxf_future(). They wrapped the evaluation result into a Future object. Result is the one that would be returned by the non-future method (e.g evaluate_future() returns the result of evaluate()).

Evaluate an artificially delayed code (using Pause), and print the time elapsed at each step:

from time import perf_counter
from wolframclient.evaluation import WolframLanguageSession

with WolframLanguageSession() as session:
    start = perf_counter()
    print('Starting an evaluation delayed by 2 seconds.')
    future = session.evaluate_future('Pause[2]; 1+1')
    print('After %.04fs, the code is running in the background, Python execution continues.' %
          (perf_counter()-start))
    # wait for up to 5 seconds.
    expr = future.result(timeout = 5)
    print('After %.02fs, result was available. Kernel evaluation returned: %s' 
      % (perf_counter()-start, expr))

The standard output should display:

Starting an evaluation delayed by 2 seconds.
After 0.0069s, the code is running in the background, Python execution continues.
After 2.02s, result was available. Kernel evaluation returned: 2

Coroutine and Asyncio APIs

asyncio provides high-level concurrent code and asynchronous evaluation using coroutines and the async/await keywords. Asynchronous evaluation based on asyncio requires an instance of WolframLanguageAsyncSession, whose methods are mostly coroutines.

Define a coroutine delayed_evaluation that artificially delays evaluation, using an asyncio sleep coroutine. Use this newly created coroutine to evaluate a first expression, wait for the coroutine to finish and evaluate the second:

import asyncio
import time
from wolframclient.evaluation import WolframLanguageAsyncSession
from wolframclient.language import wl

async def delayed_evaluation(delay, async_session, expr):
    await asyncio.sleep(delay)
    return await async_session.evaluate(expr)

async def main():
    async with WolframLanguageAsyncSession() as async_session:
        start = time.perf_counter()
        print('Starting two tasks sequentially.')
        result1 = await delayed_evaluation(1, async_session, wl.Range(3))
        # Compute the Total of the previous evaluation result:
        result2 = await delayed_evaluation(1, async_session, wl.Total(result1))
        print('After %.02fs, both evaluations finished returning: %s, %s'
            % (time.perf_counter()-start, result1, result2))

# python 3.5+
loop = asyncio.get_event_loop()
loop.run_until_complete(main())

# python 3.7+
# asyncio.run(main())

The timer printed in the standard output indicates that the total evaluation took roughly two seconds, which is the expected value for sequential evaluations:

Starting two tasks sequentially.
After 2.04s, both evaluations finished returning: [1, 2, 3], 6

When coroutines can be evaluated independently one from each other, it is convenient to run them in parallel. Start two independent coroutines in a concurrent fashion and wait for the result:

import asyncio
import time
from wolframclient.evaluation import WolframLanguageAsyncSession
from wolframclient.language import wl

async def delayed_evaluation(delay, async_session, expr):
    await asyncio.sleep(delay)
    return await async_session.evaluate(expr)

async def main():
    async with WolframLanguageAsyncSession() as async_session:
        start = time.perf_counter()
        print('Running two tasks concurrently.')
        task1 = asyncio.ensure_future(delayed_evaluation(1, async_session, '"hello"'))
        task2 = asyncio.ensure_future(delayed_evaluation(1, async_session, '"world!"'))
        # wait for the two tasks to finish
        result1 = await task1
        result2 = await task2
        print('After %.02fs, both evaluations finished returning: %s, %s'
              % (time.perf_counter()-start, result1, result2))

# python 3.5+
loop = asyncio.get_event_loop()
loop.run_until_complete(main())

# python 3.7+
# asyncio.run(main())

The total evaluation took roughly one second, indicating that both delayed coroutines ran in parallel:

Running two tasks concurrently.
After 1.03s, both evaluations finished returning: hello, world!

In the examples shown, only one Wolfram kernel was used, which is a single-threaded process. Evaluating two computation-heavy Wolfram Language expressions in parallel will have no impact on performance. This requires more than one kernel, which is exactly what kernel pool was designed for.

Kernel Pool

A WolframEvaluatorPool starts up a certain amount of evaluators and dispatches work load to them asynchronously. The pool is usable right after the first one has successfully started. Some may take more time to start and become available after a delay:

import asyncio
import time
from wolframclient.evaluation import WolframEvaluatorPool

async def main():
    async with WolframEvaluatorPool() as pool:
        start = time.perf_counter()
        tasks = [
            pool.evaluate('Pause[1]')
            for i in range(10)
            ]
        await asyncio.wait(tasks)
        print('Done after %.02fs, using up to %i kernels.' 
            % (time.perf_counter()-start, len(pool)))

# python 3.5+
loop = asyncio.get_event_loop()
loop.run_until_complete(main())

# python 3.7+
# asyncio.run(main())

Evaluation output shows that if more than one evaluator was started, the total time is less than ten seconds:

Done after 3.04s, using up to 4 kernels.

parallel_evaluate

It is possible to evaluate many expressions at once using parallel_evaluate(). This method starts a kernel pool and uses it to compute expressions yield from an iterable object. The pool is then terminated.

Import the function:

>>> from wolframclient.evaluation import parallel_evaluate

Build a list of ten delayed $ProcessID expressions, which returns the kernel process identified (pid) after one second:

>>> expressions = ['Pause[1]; $ProcessID' for _ in range(10)]

Evaluate in parallel and get back a list of ten pid values:

>>> parallel_evaluate(expressions)
[72094, 72098, 72095, 72096, 72099, 72097, 72094, 72098, 72095, 72096]

The result varies, but the pattern remains the same—namely, at least one process was started, and each process is eventually used more than once.

Logging

Logging is often an important part of an application. The library relies on the standard logging module and exposes various methods to control the level of information logged.

The first level of control is through the logging module itself. The Python library logs at various levels. Set up a basic configuration for the logging module to witness some messages in the standard output:

from wolframclient.evaluation import WolframLanguageSession
import logging

# set the root level to INFO
logging.basicConfig(level=logging.INFO)

try:
    session = WolframLanguageSession()
    # this will trigger some log messages with the process ID, the sockets
    # address and the startup timer.
    session.start()
    # Warning: Infinite expression Power[0, -1] encountered.
    res = session.evaluate('1/0')
finally:
    session.terminate()

The standard output should display:

INFO:wolframclient.evaluation.kernel.kernelcontroller:Kernel writes commands to socket: <Socket: uri=tcp://127.0.0.1:61335>
INFO:wolframclient.evaluation.kernel.kernelcontroller:Kernel receives evaluated expressions from socket: <Socket: uri=tcp://127.0.0.1:61336>
INFO:wolframclient.evaluation.kernel.kernelcontroller:Kernel process started with PID: 54259
INFO:wolframclient.evaluation.kernel.kernelcontroller:Kernel 54259 is ready. Startup took 1.74 seconds.

It is also possible to log from within the kernel. This feature is disabled by default. When initializing a WolframLanguageSession, the parameter kernel_loglevel can be specified with one of the following values to activate kernel logging: logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR.

Note

if a WolframLanguageSession is initialized with the default kernel_loglevel (i.e. logging.NOTSET), kernel logging is disable for the session, and it is not possible to activate it afterward.

From the Wolfram Language, it is possible to issue log messages using one of the following functions, given with its signature:

(* Sends a log message to Python with a given log level *)
ClientLibrary`debug[args__]
ClientLibrary`info[args__]
ClientLibrary`warn[args__]
ClientLibrary`error[args__]

The log level of the kernel is independent of the Python one. The following functions can be used to restrict the amount of log data sent by the kernel:

(* Sends only messages of a given level and above *)
ClientLibrary`SetDebugLogLevel[]
ClientLibrary`SetInfoLogLevel[]
ClientLibrary`SetWarnLogLevel[]
ClientLibrary`SetErrorLogLevel[]
(* Sends no message at all *)
ClientLibrary`DisableKernelLogging[]

Control the log level at both the Python and the kernel levels:

from wolframclient.evaluation import WolframLanguageSession
import logging

# set the Python root logger level to INFO
logging.basicConfig(level=logging.INFO)

# Start a new session, with kernel logging activated and log level set to INFO.
with WolframLanguageSession(kernel_loglevel=logging.INFO) as session:
    # This message is printed
    session.evaluate('ClientLibrary`info["ON -- Example message printed from the kernel \
with log level INFO --"]')
    # This one is not because its level is debug. 
    session.evaluate('ClientLibrary`debug["OFF -- Debug message."]')
    
    # Disable logging from the kernel side
    session.evaluate('ClientLibrary`DisableKernelLogging[]')
    # These messages will not be sent to Python.
    session.evaluate('ClientLibrary`fatal["OFF -- Fatal message. Not printed"]')
    session.evaluate('ClientLibrary`info["OFF -- End of kernel evaluation. Not printed"]')

Extending Serialization: Writing an Encoder

Serialization of a Python object involves encoders, which convert an input object into a stream of bytes. The library defines encoders for most built-in Python types and for some core libraries. It stores a mapping between types and encoder implementation. In order to serialize more classes, new encoders must be registered.

An encoder is a function of two arguments, the serializer and an object, associated with a type. The object is guaranteed to be an instance of the associated type.

Register a new encoder for a user-defined class:

from __future__ import absolute_import, print_function, unicode_literals

from wolframclient.language import wl
from wolframclient.serializers import export, wolfram_encoder

# define a new class.
class Animal(object):
    pass

# register a new encoder for instances of the Animal class.
@wolfram_encoder.dispatch(Animal)
def encode_animal(serializer, animal):
    # encode the class as a symbol called Animal
    return serializer.encode(wl.Animal)

# create a new instance
animal = Animal()
# serialize it
result = export(animal)
print(result) # b'Animal'

During export, for each object to serialize, the proper encoder is found by inspecting the type hierarchy (field __mro__). First, check that an encoder is associated with the object type; if not, repeat with the first parent type until one is found. The default encoder, associated with object, is used as the last resort.

Register some encoders for a hierarchy of classes:

from wolframclient.language import wl
from wolframclient.serializers import export, wolfram_encoder

# define a hierarchy of classes.
class Animal(object):
    pass

class Fish(Animal):
    pass

class Tuna(Fish):
    pass

# will not have its own encoder.
class Salmon(Fish):
    pass

# register a new encoder for Animal.
@wolfram_encoder.dispatch(Animal)
def encode_animal(serializer, animal):
    return serializer.encode(wl.Animal)

# register a new encoder for Fish.
@wolfram_encoder.dispatch(Fish)
def encode_fish(serializer, animal):
    return serializer.encode(wl.Fish)

# register a new encoder for Tuna.
@wolfram_encoder.dispatch(Tuna)
def encode_tuna(serializer, animal):
    # encode the class as a function using class name
    return serializer.encode(wl.Tuna)


expr = {'fish' : Fish(), 'tuna': Tuna(), 'salmon': Salmon()}
result = export(expr)
print(result) # b'<|"fish" -> Fish, "tuna" -> Tuna, "salmon" -> Fish|>'

Note: the encoder for Animal is never used, not even for the instance of Salmon, because Fish has a dedicated encoder, and type Fish appears first in the method resolution order of type Salmon:

>>> Salmon.__mro__
(<class '__main__.Salmon'>, <class '__main__.Fish'>, <class '__main__.Animal'>, <class 'object'>)

Extending WXF Parsing: Writing a WXFConsumer

Integer Eigenvalues

Use the Wolfram Client Library to access the Wolfram Language algebra functions. Compute the integer Eigenvalues on a Python matrix of integers:

from wolframclient.evaluation import WolframLanguageSession
from wolframclient.language import wl

with WolframLanguageSession() as session:
    # define a matrix of integers
    array = [
        [-1, 1, 1], 
        [1, -1, 1], 
        [1, 1, -1]]
    # expression to evaluate
    expr = wl.Eigenvalues(array)
    # send expression to the kernel for evaluation.
    res = session.evaluate(expr)
    print(res)  # [-2, -2, 1]

Complex Eigenvalues

Python has the built-in class complex. By default, the function binary_deserialize() deserializes Wolfram Language functions using a generic class WLFunction but conveniently provides a way to extend the mapping. Define ComplexFunctionConsumer, a subclass of WXFConsumer that overrides the method build_function(). The subclassed method maps Complex to the built-in Python class complex.

from wolframclient.evaluation import WolframLanguageSession
from wolframclient.language import wl
from wolframclient.deserializers import WXFConsumer, binary_deserialize

class ComplexFunctionConsumer(WXFConsumer):
    """Implement a consumer that maps Complex to python complex types."""
    
    # represent the symbol Complex as a Python class
    Complex = wl.Complex

    def build_function(self, head, args, **kwargs):
        # return a built in complex if head is Complex and argument length is 2.
        if head == self.Complex and len(args) == 2:
            return complex(*args)
        # otherwise delegate to the super method (default case).
        else:
            return super().build_function(head, args, **kwargs)

with WolframLanguageSession() as session:
    array = [
        [0, -2, 0], 
        [1, 0, -1], 
        [0, 2, 0]]
    # expression to evaluate
    expr = wl.Eigenvalues(array)
    # send expression to the kernel for evaluation.
    wxf = session.evaluate_wxf(expr)
    # get the WXF bytes and parse them using the complex consumer:
    complex_result = binary_deserialize(
        wxf, 
        consumer=ComplexFunctionConsumer())
    print(complex_result)  # [2j, -2j, 0]

Symbolic Eigenvalues

A Python-Heavy Approach

Sometimes the resulting expression of an evaluation is a symbolic exact value, which nonetheless could be approximated to a numerical result. The eigenvalues of are , and .

It is possible to build a subclass of WXFConsumer that can convert a subset of all Wolfram Language symbols into pure built-in Python objects. It has to deal with Plus and Times and converts Pi to math.pi, Rational to fractions.Fraction and Complex to complex. It results in a significant code inflation but provides a detailed review of the extension mechanism. However, as will be shown, this is not really necessary:

from wolframclient.evaluation import WolframLanguageSession
from wolframclient.language import wl
from wolframclient.deserializers import WXFConsumer, binary_deserialize

import math
import fractions

# define Complex symbol once and for all
Complex = wl.Complex

class MathConsumer(WXFConsumer):
    """Implement a consumer with basic arithmetic operation."""
    
    # Specific convertion for Pi, other symbols use the default method.
    def consume_symbol(self, current_token, tokens, **kwargs):
        # Convert symbol Pi to its numeric value as defined in Python
        if current_token.data == 'Pi':
            return math.pi
        else:
            return super().consume_symbol(current_token, tokens, **kwargs)

    # Associate heads with the method to convert them to Python types.
    DISPATCH = {
        Complex: 'build_complex',
        wl.Rational: 'build_rational',
        wl.Plus: 'build_plus',
        wl.Times: 'build_times'
    }
    # Overload the method that builds functions. 
    def build_function(self, head, args, **kwargs):
        # check if there is a specific function associated to the function head
        builder_func = self.DISPATCH.get(head, None)
        if builder_func is not None:
            try:
                # get the class method and apply it to the arguments.
                return getattr(self, builder_func)(*args)
            except Exception:
                # instead of failing, fallback to default case.
                return super().build_function(head, args, **kwargs)
        # heads not listed in DISPATCH are delegated to parent's method
        else:
            return super().build_function(head, args, **kwargs)

    def build_plus(self, *args):
        total = 0
        for arg in args:
            total = total + arg
        return total

    def build_times(self, *args):
        total = 1
        for arg in args:
            total = total * arg
        return total

    def build_rational(self, *args):
        if len(args) != 2:
            raise ValueError('Rational format not supported.')
        return fractions.Fraction(args[0], args[1])
    
    def build_complex(self, *args):
        if len(args) != 2:
            raise ValueError('Complex format not supported.')
        return complex(args[0], args[1])

with WolframLanguageSession() as session:
    array = [
        [wl.Pi, -2, 0], 
        [1, wl.Pi, -1],
        [0, 2, wl.Pi]]

    # expression to evaluate: Eigenvalues[array]
    expr = wl.Eigenvalues(array)

    # Eigenvalues are exact, but the result is a symbolic expression:
    # [Times[Rational[1, 2], Plus[Complex[0, 4], Times[2, Pi]]], 
    # Times[Rational[1, 2], Plus[Complex[0, -4], Times[2, Pi]]], Pi]
    print(session.evaluate(expr))
    
    # Use evaluate_wxf to evaluate without deserializing the result.
    wxf = session.evaluate_wxf(expr)
    # deserialize  using the math consumer:
    complex_result = binary_deserialize(wxf, consumer=MathConsumer())
    # get a numerical result, only made of built-in Python types.
    # [(3.141592653589793+2j), (3.141592653589793-2j), 3.141592653589793]
    print(complex_result)

A Wolfram Language Alternative

It is recommended to delegate as much as possible to the Wolfram Language. Instead of implementing a (fragile) counterpart of core functions such as Plus or Times, it is best to compute a numerical result within the kernel. This can be achieved with the function N. Once applied to the eigenvalues, the result becomes a mixture of complex values and reals, which was already dealt with in the previous section:

from wolframclient.evaluation import WolframLanguageSession
from wolframclient.language import wl
from wolframclient.deserializers import WXFConsumer, binary_deserialize

# represent the symbol Complex as a Python class
Complex = wl.Complex

class ComplexFunctionConsumer(WXFConsumer):
    """Implement a consumer that maps Complex to python complex types."""
    def build_function(self, head, args, **kwargs):
        if head == Complex and len(args) == 2:
            return complex(*args)
        else:
            return super().build_function(head, args, **kwargs)

with WolframLanguageSession() as session:
    array = [
        [wl.Pi, -2, 0], 
        [1, wl.Pi, -1],
        [0, 2, wl.Pi]]

    # expression to evaluate: N[EigenValues[array]]
    expr = wl.N(wl.Eigenvalues(array))
    
    # evaluate without deserializing
    wxf = session.evaluate_wxf(expr)
    # deserialize using the math consumer:
    complex_result = binary_deserialize(wxf, consumer=ComplexFunctionConsumer())
    # [(3.141592653589793+2j), (3.141592653589793-2j), 3.141592653589793]
    print(complex_result)