Agent module¶

This module contains the Agent class. The Agent and Stream classes are the building blocks of PythonStreams.

class Agent.Agent(in_streams, out_streams, transition, state=None, call_streams=None, stream_manager=None, name=None)[source]¶

Bases: object

An agent is an automaton: a state-transition machine. An agent is initialized in __init__ and a state transition is executed by next().

An agent has lists of: (1) input streams, (2) output streams and (3) call streams. Streams are described in Stream.py.

During a state transition an agent: (1) May read values from its input streams. (Note that

reading values in a stream does not change the stream.)

2. Append values to the tails of its output streams.
3. Change the agent’s own state.

When a call stream is modified the agent’s next() method is called which causes the agent to execute a state transition.

The default is that every input stream is also a call stream, i.e., the agent executes a state transition when any of its input streams is modified. For performance reasons, we may not want the agent to execute state transitions each time any of its input streams is modified; we may want the agent to execute state transitions periodically — for example, every second. In this case, the call streams will be different from the input streams. A call stream that has a value appended to it every second will cause the agent to execute a state transition every second.

Parameters:

in_streams : list of streams out_streams : list of streams call_streams : list of streams When a new value is added to a stream in this list a state transition is invoked. This the usual way (but not the only way) in which state transitions occur. A state transiton for an agent ag can also be executed by calling ag.next() state: object The state of the agent. The state is updated after a transition. transition: function This function is called by next() which is the state-transition operation for this agent. An agent’s state transition is specified by its transition function. stream_manager : function Each stream has management variables, such as whether the stream is open or closed. After a state-transition the agent executes the stream_manager function to modify the management variables of the agent’s output and call streams. name : str, optional name of this agent

Attributes

_in_lists: list of InList

InList defines the slice of a list. The j-th element of _in_lists is an InList that defines the slice of the j-th input stream that may be read by this agent in a state transition. For example, if listj = _in_lists[j].lists startj = _in_lists[j].start stopj = _in_lists[j].stop Then this agent can read the slice: listj[startj:stopj] of the jth input stream. This slice is a slice of the most recent values of the stream.

_out_lists: list

The j-th element of _out_lists is the list of values to be appended to the j-th output stream after the state transition.

Methods

next(stream_name=None)

Execute a state transition. The method has 3 parts: (i) set up the data structures to execute a state transition, (ii) call transition to: (a) get the values to be appended to output streams, (b) get the next state, and (c) update ‘start’ indices for each input stream. The ‘start’ pointers are indices where the agent asserts that it will no longer access elements of its input streams with indices earlier (i.e. smaller) that ‘start’. (iii) update data structures after the transition.

next(stream_name=None)[source]¶

Execute the next state transition.

This function does the following: Part 1: set up data structures for the state transition. Part 2: execute the state transition by calling self.transition Part 3: update data structures after the transition.

This method can be called by any agent and is called whenever a value is appended to any stream in call_streams

Parameters:	stream_name : str, optional A new value was appended to the stream with name stream_name as a result of which this agent executes a state transition.

Agent.EPSILON = 1e-12¶

class Agent.InList¶

Bases: tuple

Attributes

list	Alias for field number 0
start	Alias for field number 1
stop	Alias for field number 2

Methods

count(...)
index((value, [start, ...)	Raises ValueError if the value is not present.

list¶

Alias for field number 0

start¶

Alias for field number 1

stop¶

Alias for field number 2

ML package¶

MakeProcess module¶

Operators module¶

This module has functions that convert operations on standard Python data structures to operations on streams.

The module has three collections of functions: (1) functions that convert operations on standard Python data structures to operations on streams. These functions operate on a list of input streams to generate a list of output streams. The functions deal with the following data structures:

1. lists,
2. individual elements of lists,
3. sliding windows, and
4. timed windows.

(2) functions that map the general case of multiple input streams and multiple output streams described above to the following special cases:

1. merge: an arbitrary number of input streams and a single output stream.
2. split: a single input stream and an arbitrary number of output streams.
3. op: a single input stream and a single output stream.
4. source: no input and an arbitrary number of output streams.

(e) sink: no ouput and an arbitrary number of input streams. These special cases simplify functions that need to be written for standard Python data structures. You can always use the multiple inputs and outputs case even if there is only one or no input or output; however, the functions for merge, split, op, source, and sink are simpler than the multiple input and output case.

(3) a function that provides a single common signature for converting operations on Python structures to operations on streams regardless of whether the function has no inputs, a single input stream, a list of input streams, or no outputs, a single output stream or a list of output streams. (12 October 2015. Mani. Changed initialization of output_lists.)

Operators.adjustable_window_agent(f, inputs, outputs, state, call_streams=None, window_size=None, step_size=None)[source]¶

Operators.adjustable_window_func(f, inputs, num_outputs, state, call_streams, window_size, step_size)[source]¶

Operators.assert_is_list(x)[source]¶

Operators.assert_is_list_of_lists(x, list_size=None)[source]¶

Operators.assert_is_list_of_streams(x)[source]¶

Operators.assert_is_list_of_streams_or_None(x)[source]¶

Operators.assert_is_list_or_None(x)[source]¶

Operators.asynch_element_agent(f, inputs, outputs, state, call_streams, window_size, step_size)[source]¶

Operators.asynch_element_func(f, inputs, num_outputs, state, call_streams=None, window_size=None, step_size=None)[source]¶

Operators.awf(inputs, outputs, func, window_size, step_size, state=None, call_streams=None, **kwargs)[source]¶

Operators.dynamic_window_agent(f, input_stream, output_stream, state, min_window_size, max_window_size, step_size)[source]¶

Operators.dynamic_window_func(f, inputs, state, min_window_size, max_window_size, step_size)[source]¶

Operators.ef(inputs, outputs, func, state=None, call_streams=None, **kwargs)[source]¶

Operators.element_agent(f, inputs, outputs, state, call_streams, window_size, step_size)[source]¶

Operators.element_func(f, inputs, num_outputs, state, call_streams, window_size, step_size)[source]¶

Operators.h(f_type, *args)[source]¶

Calls the appropriate wrapper function given the name of the wrapper. The wrapper functions are list_func, element_func, window_func, ... for wrapper names ‘list’, ‘element’, ‘window’,..

Operators.h_agent(f_type, *args)[source]¶

Calls the appropriate wrapper function given the name of the wrapper. The wrapper functions are list_agent, element_agent, window_agent, ... for wrapper names ‘list’, ‘element’, ‘window’,..

Operators.lf(inputs, outputs, func, state=None, call_streams=None, **kwargs)[source]¶

Operators.list_agent(f, inputs, outputs, state, call_streams, window_size, step_size)[source]¶

Operators.list_func(f, inputs, num_outputs, state, call_streams, window_size, step_size)[source]¶

Operators.list_index_for_timestamp(in_list, start_index, timestamp)[source]¶

A helper function for timed operators. The basic idea is to return the earliest index in in_list.list[start_index:in_list.stop] with a time field that is greater than or equal to timestamp. If no such index exists then return a negative number.

Parameters:	in_list: InList InList = namedtuple(‘InList’, [‘list’, ‘start’, ‘stop’]) A slice into a stream. start_index: nonnegative integer A pointer into in_list.list timestamp: number
Returns:	Returns positive integer i where: either: ‘FOUND TIME WINDOW IN IN_LIST’ i >= start_index and i <= in_list.stop and (in_list[start_index] >= timestamp or in_list.list[i-2][0] < timestamp <= in_list.list[i-1][0] ) ) or: ‘NO TIME WINDOW IN IN_LIST’ i < 0 (negative i indicates no time window) and (in_list.list[in_list.stop-1] <= timestamp or the list is empty, i.e. (in_list.start = in_list.stop)

Operators.main()[source]¶

Operators.many_outputs_source(f_type, f, num_outputs, state, call_streams, window_size, step_size)[source]¶

Operators.many_to_many(f_type, f, in_streams, num_outputs, state, call_streams, window_size, step_size)[source]¶

Operators.many_to_many_agent(f_type, f, in_streams, out_streams, state, call_streams, window_size, step_size)[source]¶

Operators.merge(f_type, f, in_streams, state, call_streams, window_size, step_size)[source]¶

Operators.merge_agent(f_type, f, in_streams, out_stream, state, call_streams, window_size, step_size)[source]¶

Operators.op(f_type, f, in_stream, state, call_streams, window_size, step_size)[source]¶

Operators.op_agent(f_type, f, in_stream, out_stream, state, call_streams, window_size, step_size)[source]¶

Operators.remove_novalue_and_open_multivalue(l)[source]¶

This function returns a list which is the same as the input parameter l except that (1) _no_value elements in l are deleted and (2) each _multivalue element in l is opened

i.e., for an object _multivalue(list_x) each element of list_x appears in the returned list.

Operators.single_output_source(f_type, f, num_outputs, state, call_streams, window_size, step_size)[source]¶

Operators.single_output_source_agent(f_type, f, out_stream, state, call_streams, window_size, step_size)[source]¶

Operators.sink(f_type, f, in_stream, state, call_streams, window_size, step_size)[source]¶

Operators.sink_merge(f_type, f, in_streams, state, call_streams, window_size, step_size)[source]¶

Operators.split(f_type, f, in_stream, num_outputs, state, call_streams, window_size, step_size)[source]¶

Operators.split_agent(f_type, f, in_stream, out_streams, state, call_streams, window_size, step_size)[source]¶

Operators.stream_agent(inputs, outputs, f_type, f, state=None, call_streams=None, window_size=None, step_size=None)[source]¶

Provides a common signature for converting functions f on standard Python data structures to streams.

Parameters:

f_type : {‘element’, ‘list’, ‘window’, ‘timed’, ‘asynch_element’} f_type identifies the type of function f where f is the next parameter. f : function inputs : {Stream, list of Streams} When stream_func has: no input streams, inputs is None a single input Stream, inputs is a single Stream multiple input Streams, inputs is a list of Streams. outputs : list of Streams state : object state is None or is an arbitrary object. The state captures all the information necessary to continue processing the input streams. call_streams : None or list of Stream If call_streams is None then the program sets it to inputs (converting inputs to a list of Stream if necessary). This function is called when, and only when any stream in call_streams is modified. window_size : None or int window_size must be a positive integer if f_type is ‘window’ or ‘timed’. window_size is the size of the moving window on which the function operates. step_size : None or int step_size must be a positive integer if f_type is ‘window’ or ‘timed’. step_size is the number of steps by which the moving window moves on each execution of the function.

Returns:

None

Operators.stream_func(inputs, f_type, f, num_outputs, state=None, call_streams=None, window_size=None, step_size=None)[source]¶

Provides a common signature for converting functions f on standard Python data structures to streams.

Parameters:

f_type : {‘element’, ‘list’, ‘window’, ‘timed’, ‘asynch_element’} f_type identifies the type of function f where f is the next parameter. f : function inputs : {Stream, list of Streams} When stream_func has: no input streams, inputs is None a single input Stream, inputs is a single Stream multiple input Streams, inputs is a list of Streams. num_outputs : int A nonnegative integer which is the number of output streams of this function. state : object state is None or is an arbitrary object. The state captures all the information necessary to continue processing the input streams. call_streams : None or list of Stream If call_streams is None then the program sets it to inputs (converting inputs to a list of Stream if necessary). This function is called when, and only when any stream in call_streams is modified. window_size : None or int window_size must be a positive integer if f_type is ‘window’ or ‘timed’. window_size is the size of the moving window on which the function operates. step_size : None or int step_size must be a positive integer if f_type is ‘window’ or ‘timed’. step_size is the number of steps by which the moving window moves on each execution of the function.

Returns:

list of Streams Function f is applied to the appropriate data structure in the input streams to put values in the output streams. stream_func returns the output streams.

Operators.tf(inputs, outputs, func, window_size, step_size, state=None, call_streams=None, **kwargs)[source]¶

Operators.timed_agent(f, inputs, outputs, state, call_streams, window_size, step_size)[source]¶

Operators.timed_func(f, inputs, num_outputs, state, call_streams, window_size, step_size)[source]¶

Operators.wf(inputs, outputs, func, window_size, step_size, state=None, call_streams=None, **kwargs)[source]¶

Operators.window_agent(f, inputs, outputs, state, call_streams, window_size, step_size)[source]¶

Operators.window_func(f, inputs, num_outputs, state, call_streams, window_size, step_size)[source]¶

RemoteQueue module¶

Stream module¶

This module contains the Stream class. The Stream and Agent classes are the building blocks of PythonStreams. (12 October 2015. Mani. Fixed bug. Made _no_value and _close classes rather than object.)

class Stream.Stream(name='NoName', proc_name='UnknownProcess', initial_value=[], stream_size=4096, buffer_size=1024)[source]¶

Bases: object

A stream is a sequence of values. Agents can: (1) Append values to the tail of stream and close a stream. (2) Read a stream. (3) Subscribe to be notified when a value is added to a stream. (See Agent.py for details of agents.)

The ONLY way in which a stream can be modified is that values can be appended to its tail. The length of a stream (number of elements in its sequence) can stay the same or increase, but never decreases. If at some point, the length of a stream is k, then from that point onwards, the first k elements of the stream remain unchanged.

A stream is written by only one agent. Any number of agents can read a stream, and any number of agents can subscribe to a stream. An agent can be a reader and a subscriber and a writer of the same stream. An agent may subscribe to a stream without reading the stream’s values; for example the agent may subscribe to a clock stream and the agent executes a state transition when the the clock stream has a new value, regardless of the value.

Parameters:

name : str, optional name of the stream. Though the name is optional a named stream helps with debugging. default : ‘NoName’ proc_name : str, optional The name of the process in which this agent executes. default: ‘UnknownProcess’ initial_value : list or array, optional The list (or array) of initial values in the stream. default : [] stream_size: int, optional stream_size must be a positive integer. It is the largest number of the most recent elements in the stream that are in main memory. default : DEFAULT_STREAM_SIZE where DEFAULT_STREAM_SIZE is specified in SystemParameters.py buffer_size : int, optional buffer_size must be a positive integer. An exception may be thrown if an agent reads an element with index i in the stream where i is less than the length of the stream - buffer_size. default : DEFAULT_BUFFER_SIZE_FOR_STREAM specified in SystemParameters.py

Notes

1. AGENTS SUBSCRIBING TO A STREAM

An agent is a state-transition automaton and the only action that an agent executes is a state transition. If agent x is a subscriber to a stream s then x.next() — a state transition of x — is invoked whenever messages are appended to s.

The only point at which an agent executes a state transition is when a stream to which the agent subscribes is modified.

An agent x subscribes to a stream s by executing

s.call(x).

An agent x unsubscribes from a stream s by executing:

s.delete_caller(x)

2. AGENTS READING A STREAM

2.1 Agent registers for reading

An agent can read a stream only after it registers with the stream as a reader. An agents r registers with a stream s by executing:

s.reader(r)

An agent r deletes its registration for reading s by executing:

s.delete_reader(r)

2.2 Slice of a stream that can be read by an agent

At any given point, an agent r that has registered to read a stream s can only read some of the most recent values in the stream. The number of values that an agent can read may vary from agent to agent. A reader r can only read a slice:

s[s.start[r]+s.offset: s.stop+s.offset]

of stream s where start[r], stop and offset are defined later.

3. WRITING A STREAM

3.1 Extending a stream

When an agent is created it is passed a list of streams that it can write.

An agent adds a single element v to a stream s by executing:

s.append(v)

An agent adds the sequence of values in a list l to a stream s by executing:

s.extend(l)

The operations append and extend of streams are analogous to operations with the same names on lists.

3.2 Closing a Stream

A stream is either closed or open. Initially a stream is open. An agent that writes a stream s can close s by executing:

s.close()

A closed stream cannot be modified.

4. MEMORY

4.1 The most recent values of a stream

The most recent elements of a stream are stored in main memory. In addition, the user can specify whether all or part of the stream is saved to a file.

Associated with each stream s is a list (or array) s.recent that includes the most recent elements of s. If the value of s is a sequence:

s[0], ..., s[n-1],

at a point in a computation then at that point, s.recent is a list

s[m], .., s[n-1]

for some m, followed by some padding (usually a sequence of zeroes, as described later).

The system ensures that all readers of stream s only read elements of s that are in s.recent.

4.2 Slice of a stream that can be read

Associated with a reader r of stream s is an integer s.start[r]. Reader r can only read the slice:

s.recent[s.start[r] : ]

of s.recent.

For readers r1 and r2 of a stream s the values s.start[r1] and s.start[r2] may be different.

4.3 When a reader finishes reading part of a stream

Reader r informs stream s that it will only read values with indexes greater than or equal to j in the list, recent, by executing

s.set_start(r, j)

which causes s.start[r] to be set to j.

5. OPERATION

5.1 Memory structure

Associated with a stream is: (1) a list, recent. (2) a nonnegative integer stop where:

1. recent[ : stop] contains the most recent values of the stream,
2. the slice recent[stop:] is padded with padding values (either 0 or 0.0).

3. a nonnegative integer s.offset where

   recent[i] = stream[i + offset]

   for 0 <= i < s.stop

Example: if the sequence of values in a stream is:

0, 1, .., 949

and s.offset is 900, then

s.recent[i] = s[900+i] for i in 0, 1, ..., 49.

Invariant:

len(s) = s.offset + s.stop

where len(s) is the number of values in stream s.

The size of s.recent is the parameter stream_size of s. Recommendations for the value of stream_size are given after a few paragraphs.

The maximum size of the list that an agent can read is the parameter, buffer_size. Set buffer_size large enough so that the size of the slice that any agent wants to read is less than buffer_size. If an agent is slow compared to the rate at which the stream grows then the buffer_size should be large. For example, if an agent is reading the element in the stream at location i, and the stream has grown to l elements then buffer_size must be greater than l - i.

(In later implementations, if an agent reads a part of stream s that is not in s.recent, then the value read is obtained from values saved to a file.)

The entire stream, or the stream up to offset, can be saved in a file for later processing. You can also specify that no part of the stream is saved to a file. (Note, if the stream s is not saved, and any agent reads an element of the stream s that is not in main memory, then an exception is raised.)

In the current implementation old values of the stream are not saved.

5.2 Memory Management

We illustrate memory management with the following example with stream_size=4 and buffer_size=1

Assume that a point in time, for a stream s, the list of values in the stream is [1, 2, 3, 10, 20]; stream_size=4; s.offset=3; s.stop=2; and s.recent = [10, 20, 0, 0]. The size s.recent is stream_size (i.e. 4). The s.stop (i.e. 2) most recent values in the stream are 10 followed by a later value, 20. s[3] == 10 == s.recent[0] s[4] == 20 == s.recent[1] The values in s.recent[s.stop:] are padded values (zeroes).

A reader r of stream s has access to the list:

s.recent[s.start[r] : s.stop]

So, if for a reader r, s.start[r] is 0, then r has access to the two most recent values in the stream, i.e., the list [10, 20]. If for another reader q, s.start[q]=1, then q has access to the list [20]. And for another reader u, s.start[q]=2, then u has access to the empty list [].

When a value v is appended to stream s, then v is inserted in s.recent, replacing a padded value, and s.stop is incremented. If the empty space (i.e., the number of padded values) in s.recent decreases below buffer_size then a new version of s.recent is created containing only the buffer_size most recent values of the stream.

Example: Start with the same example as the previous example with buffer_size = 2

Then a new value, 30 is appended to the stream, making the list of values in s: [1, 2, 3, 10, 20, 30] s.stop = 3; s.offset is unchanged (i.e. 3) and s.recent = [10, 20, 30, 0]. Now the size of the empty space in s.recent is 1, which is less than buffer_size. So, the program sets s.recent to [20, 30, 0, 0], keeping the buffer_size (i.e. 2) most recent values in s.recent and removing older values from main memory, and it sets s.stop to buffer_size, and increments offset by s.stop - buffer_size. Now

s.stop = 2 s.offset = 4

Attributes

recent	(list) A list of the most recent values of the stream. recent is a NumPy array if specified.
stop	(int) index into the list recent. s.recent[:s.stop] contains the s.stop most recent values of stream s. s.recent[s.stop:] contains padded values.
offset: int	index into the stream used to map the location of an element in the entire stream with the location of the same element in s.recent, which only contains the most recent elements of the stream. For a stream s: s.recent[i] = s[i + s.offset] for i in range(s.stop)
start	(dict of readers.) key = reader value = start index of the reader Reader r can read the slice: s.recent[s.start[r] : s.stop ] in s.recent which is equivalent to the following slice in the entire stream: s[s.start[r]+s.offset: s.stop+s.offset]
subscribers_set: set	the set of subscribers for this stream. Subscribers are agents to be notified when an element is added to the stream.
closed: boolean	True if and only if the stream is closed. An exception is thrown if a value is appended to a closed stream.
_buffer_size: int	Invariant: For every reader r of stream s: s.stop - s.start[r] < s._buffer_size A reader can only access _buffer_size number of consecutive, most recent, elements in the stream.
_begin	(int) index into the list, recent recent[:_begin] is not being accessed by any reader; therefore recent[:_begin] can be deleted from main memory. Invariant: for all readers r: _begin <= min(start[r])

Methods

append(value)	Append a single value to the end of the stream.
call(agent)	Register a subscriber for this stream.
close()	Close this stream.”
delete_caller(agent)	Delete a subscriber for this stream.
delete_reader(reader)	Delete this reader from this stream.
extend(value_list)	Extend the stream by the list of values, value_list.
print_recent()
reader(reader[, start_index])	Register a reader.
set_name(name)
set_start(reader, start)	The reader tells the stream that it is only accessing elements of the list recent with index start or higher.

append(value)[source]¶

Append a single value to the end of the stream.

call(agent)[source]¶

Register a subscriber for this stream.

close()[source]¶

Close this stream.”

delete_caller(agent)[source]¶

Delete a subscriber for this stream.

delete_reader(reader)[source]¶

Delete this reader from this stream.

extend(value_list)[source]¶

Extend the stream by the list of values, value_list.

Parameters:	value_list: list

print_recent()[source]¶

reader(reader, start_index=0)[source]¶

Register a reader.

The newly registered reader starts reading list recent from index start, i.e., reads the slice recent[start_index:s.stop] If reader has already been registered with this stream its start value is updated to the parameter in the call.

set_name(name)[source]¶

set_start(reader, start)[source]¶

The reader tells the stream that it is only accessing elements of the list recent with index start or higher.

class Stream.StreamArray(name=None)[source]¶

Bases: Stream.Stream

Methods

append(value)	Append a single value to the end of the stream.
call(agent)	Register a subscriber for this stream.
close()	Close this stream.”
delete_caller(agent)	Delete a subscriber for this stream.
delete_reader(reader)	Delete this reader from this stream.
extend(a)	Extend the stream by an numpy ndarray.
print_recent()
reader(reader[, start_index])	Register a reader.
set_name(name)
set_start(reader, start)	The reader tells the stream that it is only accessing elements of the list recent with index start or higher.

extend(a)[source]¶

Extend the stream by an numpy ndarray.

Parameters:	a: np.ndarray or list

class Stream.StreamSeries(name=None)[source]¶

Bases: Stream.Stream

Methods

append(value)	Append a single value to the end of the stream.
call(agent)	Register a subscriber for this stream.
close()	Close this stream.”
delete_caller(agent)	Delete a subscriber for this stream.
delete_reader(reader)	Delete this reader from this stream.
extend(value_list)	Extend the stream by the list of values, value_list.
print_recent()
reader(reader[, start_index])	Register a reader.
set_name(name)
set_start(reader, start)	The reader tells the stream that it is only accessing elements of the list recent with index start or higher.

class Stream.StreamTimed(name=None)[source]¶

Bases: Stream.Stream

Methods

append(value)	Append a single value to the end of the stream.
call(agent)	Register a subscriber for this stream.
close()	Close this stream.”
delete_caller(agent)	Delete a subscriber for this stream.
delete_reader(reader)	Delete this reader from this stream.
extend(value_list)	Extend the stream by the list of values, value_list.
print_recent()
reader(reader[, start_index])	Register a reader.
set_name(name)
set_start(reader, start)	The reader tells the stream that it is only accessing elements of the list recent with index start or higher.

class Stream.TimeAndValue¶

Bases: tuple

Attributes

time	Alias for field number 0
value	Alias for field number 1

Methods

count(...)
index((value, [start, ...)	Raises ValueError if the value is not present.

time¶

Alias for field number 0

value¶

Alias for field number 1

SystemParameters module¶

SYSTEM_PARAMETERS

examples_element_wrapper module¶

This module contains examples of the ‘element’ wrapper. A function that takes a single element of an input stream and generates a single element of an output stream is an example of a function that is wrapped by the ‘element’ wrapper to create a function on streams.

The module has the following parts: (1) op or simple operator:

single input, single output

2. sink:

   single input, no outputs

3. source:

   no input, multiple outputs

4. split:

   single input, multiple output

5. merge:

   multiple input, single output

6. general case:

   multiple input, multiple output

All of the first 5 cases are specializations of the general case; however, the syntactic sugar they provide can be helpful.

For each of the above cases we first consider agents that are stateless and then consider agents with state.

examples_element_wrapper.average(v, state)[source]¶

Parameters:	v : number The next element of the input stream of the agent. state: (n, cumulative) The state of the agent where n : number The value of the next element in the agent’s input stream. cumulative : number The sum of the values that the agent has received on its input stream.
Returns:	(mean, state) mean : floating point number The average of the values received so far by the agent state : (n, cumulative) The new state of the agent.

examples_element_wrapper.average_stream(stream)[source]¶

examples_element_wrapper.boolean_of_values_greater_than_threshold(stream, threshold)[source]¶

examples_element_wrapper.cumulative_stream(stream)[source]¶

examples_element_wrapper.cumulative_sum(v, cumulative)[source]¶

Parameters:	v : number The next element of the input stream of the agent. cumulative: number The state of the agent. The state is the sum of all the values received on the agent’s input stream.
Returns:	(cumulative, cumulative) cumulative : number The state after the transition, i.e., the sum of values received on the agent’s input stream including the value received in this transition.

examples_element_wrapper.discard_odds(stream)[source]¶

examples_element_wrapper.discard_odds_1(stream)[source]¶

examples_element_wrapper.double(v)[source]¶

examples_element_wrapper.double_stream(stream)[source]¶

examples_element_wrapper.even(v)[source]¶

examples_element_wrapper.even2(v)[source]¶

examples_element_wrapper.even3(v)[source]¶

examples_element_wrapper.even_1(v)[source]¶

examples_element_wrapper.even_odd(m)[source]¶

examples_element_wrapper.even_odd_stream(stream)[source]¶

examples_element_wrapper.evens_and_halves(stream)[source]¶

examples_element_wrapper.evens_and_halves_3(stream)[source]¶

examples_element_wrapper.exp_mult_div_stream(stream, exponent, multiplier, divisor)[source]¶

examples_element_wrapper.file_to_stream(filename, output_stream, time_period=0)[source]¶

Parameters:	filename: str output_stream: Stream time_period: int or float, nonnegative

examples_element_wrapper.inrange_and_outlier_streams(x_and_y_streams, A, B, DELTA)[source]¶

examples_element_wrapper.join_streams(list_of_streams)[source]¶

examples_element_wrapper.main()[source]¶

examples_element_wrapper.max_and_min_seen_across_all_streams(list_of_streams)[source]¶

examples_element_wrapper.max_and_min_stream(list_of_streams)[source]¶

examples_element_wrapper.max_and_min_with_names(list_of_numbers, state)[source]¶

examples_element_wrapper.max_of_all_inputs(list_of_elements, previous_max)[source]¶

examples_element_wrapper.max_seen_across_all_streams(list_of_streams)[source]¶

examples_element_wrapper.max_stream(list_of_streams)[source]¶

examples_element_wrapper.max_with_index(list_of_numbers, state)[source]¶

examples_element_wrapper.mean_stream(list_of_streams)[source]¶

examples_element_wrapper.multiply_elements_in_stream(stream, multiplier)[source]¶

examples_element_wrapper.print0(stream)[source]¶

examples_element_wrapper.print_stream(stream)[source]¶

examples_element_wrapper.print_streams(list_of_streams)[source]¶

examples_element_wrapper.print_sums(list_of_streams)[source]¶

examples_element_wrapper.rand(output_stream, num_outputs, time_period=0)[source]¶

Parameters:	output_stream: Stream num_outputs: int, positive time_period: int or float, positive

examples_element_wrapper.single_stream_of_random_numbers(timer_stream)[source]¶

examples_element_wrapper.square(v)[source]¶

examples_element_wrapper.square_and_double(m)[source]¶

examples_element_wrapper.square_and_double_stream(stream)[source]¶

examples_element_wrapper.square_stream(stream)[source]¶

examples_element_wrapper.stream_to_file(stream, filename)[source]¶

examples_element_wrapper.temperature_and_humidity_streams(stream, DELTA)[source]¶

examples_element_wrapper.timer(output_stream, num_outputs, time_period)[source]¶

Parameters:	stream: Stream num_outputs: int, positive time_period: int or float, positive

examples_timed_window_wrapper module¶

examples_timed_window_wrapper.avg_of_max_and_min_in_timed_list(timed_list, state)[source]¶

examples_timed_window_wrapper.avg_of_max_and_min_values_in_all_timed_lists(list_of_timed_lists, state)[source]¶

examples_timed_window_wrapper.diff_of_counts_in_lists(list_of_two_lists)[source]¶

examples_timed_window_wrapper.exponential_smoothed_timed_windows(input_stream, func, alpha, window_size, step_size, initial_state)[source]¶

Parameters

input_stream: Stream

A previously defined stream This is the only input stream of the agent.

func: function

func operates on a list of TimeAndValue objects and returns an object that can be smoothed exponentially.

alpha: number

The exponential smoothing parameter.

window_size, step_size, initial_state:

Already defined.

examples_timed_window_wrapper.max_and_min_of_values_in_timed_list(timed_list)[source]¶

examples_timed_window_wrapper.max_and_min_values_in_all_timed_lists(list_of_timed_lists)[source]¶

examples_timed_window_wrapper.max_sums_timed_windows(list_of_timed_lists, state)[source]¶

examples_timed_window_wrapper.sum_values_in_all_timed_lists(list_of_timed_lists)[source]¶

examples_timed_window_wrapper.sum_values_in_timed_list(timed_list)[source]¶

examples_window_wrapper module¶

This module contains examples of the ‘window’ wrapper. A window wrapper wraps a function that has a parameter which is a list or a list of lists and that returns a value or a list of values. The wrapped function operates on a sliding window of a stream or a list of sliding windows of a list of streams, and returns a stream or a list of streams.

A sliding window is defined by a window size and a step size. An operation on a sliding window carries out its first operation only when the size of the stream is at least the window size. A window is a list of the specified size. An operation is carried out on the window; then the window is moved forward in the stream by the step size.

For example, if the operation on the window is sum, and the window size is 3 and the step size is 2, and the stream is [5, 7] at a point in time t0, then no window operation is carried at t0. If at a later time, t1, the stream is [5, 7, 8] then the sum operation is carried out on this window of size 3 to return 20 at t1. Then the window operation waits until the window steps forward by 2. If at a later time, t2, the stream is [5, 7, 8, 2], no operation is carried out at t2. At a later time t3, if the stream is [5, 7, 8, 2, 5] then an operation is carried out on the window [8, 2, 5] to give 15.

A window operation on multiple input streams waits until sliding windows are available on all the input streams. The window sizes and step sizes for all windows are identical.

The examples below deal with stateless and stateful cases of single and multiple input streams and single and multiple output streams. We don’t use windows for sources or for sinks because simple elements are adequate for this purpose. Likewise, we don’t use windows for asynchronous merges.

examples_window_wrapper.avg_of_difference_of_two_windows(list_of_two_windows, state)[source]¶

examples_window_wrapper.combine_windows(input_stream, weights)[source]¶

examples_window_wrapper.diff_of_means_of_two_windows(list_of_two_windows)[source]¶

examples_window_wrapper.exp_smoothing_mean_and_std_of_stream(stream, alpha, window_size, step_size)[source]¶

examples_window_wrapper.extent(lst)[source]¶

examples_window_wrapper.inrange_and_outlier_streams(list_of_two_streams, window_size, step_size, alpha, threshold)[source]¶

examples_window_wrapper.max_and_min(lst)[source]¶

examples_window_wrapper.mean_inc(lst, state)[source]¶

examples_window_wrapper.mean_of_window(stream)[source]¶

source_stream module¶

source_stream.main()[source]¶

source_stream.source_stream(output_stream, number_of_values, time_period, func, **kwargs)[source]¶

Periodically appends a value to output_stream. The values appended are obtained by calling the function func and passing it keyword arguments.

If number_of_values is non-negative, then it is the maximum number of values inserted into output_stream. If number_of_values is negative then values are appended to output_stream forever.

If time_period is 0 then number_of_values must be non-negative; in this case all the values are appended to output_stream when source_stream is called.

Parameters:	output_stream: Stream (Could also be a list.) The stream to which values are appended. (Note: Appending messages to a list forever will cause memory overflow.) number_of_values: int time_period: int or float, nonnegative The time between successive appends to output_stream. func: function The return value of this function is appended to output_stream.

tutorial_average_of_stream module¶

tutorial_average_of_stream.update_avg(v, state)[source]¶

tutorial_element_wrapper_stream_operator module¶

tutorial_exponential_smoothing module¶

tutorial_exponential_smoothing.update_smooth(lst, state, h, alpha)[source]¶

tutorial_filter_1 module¶

tutorial_filter_1.ft(v, threshold)[source]¶

tutorial_many_to_many module¶

tutorial_merge module¶

tutorial_merge.dif(lst)[source]¶

tutorial_sine module¶

tutorial_split module¶

tutorial_split.execute(v, g, h)[source]¶

weave_streams module¶