Generators

As described in the Dynamic Parameters user guide, Param offers a variety of programming models. While Dynamic parameters allow you to build applications in a “pull” based paradigm, dependencies and watchers allow setting up a “push” based model where data flows through the application when an event is triggered. How such events are triggered is largely dependent on the use case; e.g. often such events will emanate from a UI that updates parameters. However, generators can be another source of event generation, and offer an easy and flexible approach for generating data that arrives from an external source.

Let’s start with a simple but concrete example. Let’s say we are observing an external sensor measuring the output of a tidal gauge:

import random

def tidal_gauge(last=[0]):
    last[0] += random.gauss(0, 1)/2.
    return last[0]

tidal_gauge()

-0.38245112833319356

Instead of scheduling periodic events on an event loop we can simply write a generator function to poll the sensor at some defined interval:

import time

def poll_sensor(timeout=0.5):
    while True:
        yield tidal_gauge()
        time.sleep(timeout)

Now let us declare a TidalGauge class that will reflect the value of our sensor.

import param

class TidalGauge(param.Parameterized):

    height = param.Number(allow_refs=True)

Note how we declare that the height parameter to allow_refs, i.e. we enable the ability for it to resolve references, which include generators.

gauge = TidalGauge(height=poll_sensor)

Now let’s look at the gauge and watch the height value update every 0.5 seconds (try evaluating the cell below a few times):

gauge

TidalGauge(height=0.44767356849198175, name='TidalGauge00002')

Expressions

To watch the tidal gauge update live we can obtain the gauge value as a reactive expression:

gauge.param.height.rx()

0.44767356849198175

Note that rx can also evaluate a generator directly:

gauge_rx = param.rx(poll_sensor)

gauge_rx

None

Just like other expressions, generator expressions allow operator chaining and applying methods. So let’s say we know that our sensor has a fixed error, we can correct this directly by adding a value to our expression:

corrected = gauge_rx + 8

corrected

8.122590300918748

The precise value is interesting but maybe we want to build a system that warns us if the value exceeds some threshold. Using a where expression this can be expressed very simply:

(corrected > 10).rx.where('Risk of flooding', 'Everything is normal')

'Everything is normal'

The same thing can also easily be achieved by using parameter binding:

param.bind(lambda value: 'Risk of flooding' if value > 10 else 'Everything is normal', corrected).rx()

'Everything is normal'

Classes

In the context of a class we can depend on a parameter that is driven by a generator as normal:

class TidalWarning(TidalGauge):

    message = param.String()
    
    @param.depends('height', watch=True)
    def _update_message(self):
        self.message = 'Risk of flooding' if self.height > 10 else 'Everything is normal'

warn = TidalWarning(height=poll_sensor)

warn.param['message'].rx()

''

Asynchronous Generators

Generators are powerful and internally Param will execute the generators on a separate thread to ensure that it does not block the main event loop. However, often an asynchronous generator is more appropriate since many operations that rely on generators are waiting on some external event, i.e. by polling a file or network resource, which can be done asynchronously.

Regular generators and asynchronous generators can be used interchangeably:

import asyncio

async def poll_sensor_async(timeout=0.5):
    while True:
        yield tidal_gauge()
        await asyncio.sleep(timeout)

async_gauge = TidalGauge(height=poll_sensor_async)

async_gauge.param.height.rx()

0.0

Dependencies

Now let us say that our generator itself has some dependency, e.g. we want to be able to control the sampling frequency of our gauge. To achieve this we can bind a parameter to our generator:

class VariableRateTidalGauge(TidalGauge):

    frequency = param.Integer(default=2, doc="""
      Frequency in Hz to sample the tidal gauge sensor at.""")

variable_gauge = VariableRateTidalGauge(frequency=5)

variable_gauge.height = param.bind(poll_sensor_async, timeout=1/variable_gauge.param.frequency.rx())

variable_gauge.param.height.rx()

0.0

Now we can vary the frequency at which our generator samples the sensor:

variable_gauge.frequency = 1