caproto.ioc_examples.random_walk.RandomWalkIOC

class caproto.ioc_examples.random_walk.RandomWalkIOC(prefix: str, *, macros: Optional[Dict[str, str]] = None, parent: Optional[caproto.server.server.PVGroup] = None, name: Optional[str] = None)

This example contains a PV x that takes random steps at an update rate controlled by a second PV, dt.

RandomWalkIOC pvproperties

Attribute	Suffix	Docs	Type	Notes	Alarm Group
dt	dt	Delta time [sec]	float
x	x	The random float value	float	Startup

Methods

group_read(instance)	Generic read called for channels without get defined
group_write(instance, value)	Generic write called for channels without put defined

Attributes

default_values
type_map
type_map_read_only

pvproperty methods

x.startup(self, instance, async_lib)

This is a startup hook which periodically updates the value.

Source code: x.startup

    @x.startup
    async def x(self, instance, async_lib):
        """This is a startup hook which periodically updates the value."""
        while True:
            # Grab the current value from `self.x` and compute the next value:
            x = self.x.value + 2. * random.random() - 1.0

            # Update the ChannelData instance and notify any subscribers:
            await instance.write(value=x)

            # Let the async library wait for the next iteration
            await async_lib.sleep(self.dt.value)