<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 f_item(func, in_stream, keyword_arguments)
func
Step1:  Second step
Step2:  Third step
Step3:  ANOTHER EXAMPLE of f_item
Problem
If an item on stream x is greater than the maximum (positive) item seen so far on x then put the item on stream y, and 
iff an item on stream x is less than the minimum (negative) item seen so far on x then put the item on stream z
First step
Step4:  SLIDING WINDOWS OF STREAMS
f_item operates on the next item in a stream. f_window operates on the next sliding window of the input stream. 
A sliding window is specified by
Step5:  Synchronous Join
in_streams stands for a list of streams (whereas in_stream represents a single stream). A synchronous join applies a function func to the list in_streams[0][i], in_streams[1][i], in_streams[2][i],... for all i.
For example, suppose x is the stream of nonnegative integers, [0, 1, 2, 3, ...] and y is the stream of letters of the alphabet repeated forever, i.e. y is ['A', 'B', 'C', .....'Z', 'A', 'B', ..]. Then join_synch applies function func to the lists [0, 'A'], [1, 'B'], [2, 'C'], ....
The arguments of join_synch are func, in_streams, and additional keyword arguments. The arguments of func are a list of joined items from the streams of in_streams and the same additional keyword arguments as in join_synch.
Example of join_synch
Given streams w, x, y, z we want to set z[i] = sum([w[i], x[i], y[i]).
Step6:  Asynchronous Join
The function join_asynch has arguments func, in_streams - a list of streams, and additional keyword arguments.
When an item v is appended to the i-th stream of in_streams, func is applied to the tuple [i, v]. The first element of the tuple indicates the stream along which the item arrived, and the second element is the item itself.
Different runs of join_asynch may produce different results because items from the different streams arrive in different orders.
The arguments of func are index_item and additional keyword arguments. index_item is a tuple of the stream index and the item value, 
Example of asynchronous join
Print items as they arrive in streams x and y and put the items in stream z.
In this example func is h. Function h has an argument index_item and keyword argument out_stream which is also used in the call to join_asynch.
Step7:  Time-based Join
This function operates on streams in which items are timestamped. Elements in each stream consist of pairs (timestamp, item). The timestamps on each stream are monotone increasing, i.e. the timestamp of a later element on a stream must be greater than the timestamp of an earlier element on the same stream.
The following situation can arise. An element [t, v] of one input stream arrives before an element [t', v'] of a different input stream, where t > t'.
The arguments of join_timed are func and in_streams - a list of streams, and additional keyword arguments which appear in func.
The function func operates on a tuple of the form [timestamp, list_of_items] where list_of_items is a list of N values where N is the number of streams in in_streams.
Consider a tuple [T, L] where T is the timestamp and L is the list of items. If there is no element with timestamp T on stream k then L[k] is None. If there is an element [T, v] in the k-th stream then L[k] is v.
<ASSISTANT_TASK:>
Python Code:
from IoTPy.agent_types.simple import f_item
def f(item, M, multiples_stream, non_multiples_stream):
    if item%M:
        multiples_stream.append(item)
    else:
        non_multiples_stream.append(item)
Explanation: f_item(func, in_stream, keyword_arguments)
func: function
function that is called when a new element is appended to in_stream
func(item_of_stream, keyword_arguments)
The keyword arguments are the same for func and f_item
in_stream: a Stream or StreamArray
Example
Given a stream x of integers and empty streams y and z and an integer constant M. Put items that are multiples of M that appear in stream x into stream y and all other items into stream Z.
First step:  define func
func operates on an item of a stream and keyword arguments. In this example, func is f and the keyword arguments of f, in addition to item are:
M, multiples_stream, non_multiples_stream.
item is an item of the in_stream.
End of explanation
x = Stream(name='input stream')
y = Stream(name='even numbers in stream x')
z = Stream(name='odd numbers in stream x')
    
f_item(func=f, in_stream=x, M=2, multiples_stream=y, non_multiples_stream=z)
Explanation: Second step: define streams and call f_item( )
The keyword arguments of func are passed as keyword arguments of f_item.
The item argument of func is not passed as an argument to f_item; instead the input stream x is passed;
So, the call to f_item is:
f_item(func=f, in_stream=x, M=2, multiples_stream=y, non_multiples_stream=z)
with the keyword arguments and their values 
M=2, multiples_stream=y, non_multiples_stream=z
in f and also passed in the call to f_item.
When a new item is appended to stream x, function f is called.
End of explanation
from IoTPy.agent_types.sink import sink_element
from IoTPy.agent_types.simple import f_item
# STEP 1: Define function
def f(item, M, multiples_stream, non_multiples_stream):
    if item%M:
        multiples_stream.append(item)
    else:
        non_multiples_stream.append(item)
# STEP 2: Declare streams and call f_item
x = Stream(name='input stream')
y = Stream(name='even numbers in stream x')
z = Stream(name='odd numbers in stream x')
    
f_item(func=f, in_stream=x, M=2, multiples_stream=y, non_multiples_stream=z)
# STEP 3: test f_item()
# Put test values in the input streams.
x.extend(list(range(10)))
# Execute a step
run()
# Look at recent values of streams.
print ('recent values of stream y are')
print (recent_values(y))
print ('recent values of stream z are')
print (recent_values(z))
Explanation: Third step: Test the function
Put test values in the input streams.
x.extend(list(range(5)))
Execute a step
run()
Look at recent values of streams.
print ('recent values of stream y are')
print (recent_values(y))
Putting the steps together:
End of explanation
max_and_min = [0, 0]
# STEP 1: Define function
def g(item, max_and_min, max_stream, min_stream):
    if item > max_and_min[0]:
        max_and_min[0] = item
        max_stream.append(item)
    if item < max_and_min[1]:
        max_and_min[1] = item
        min_stream.append(item)
# STEP 2: Declare streams and call f_item
x = Stream(name='input stream')
y = Stream(name='new maxima')
z = Stream(name='new minima')
f_item(func=g, in_stream=x, max_and_min=max_and_min, max_stream=y, min_stream=z)
# STEP 3: test f_item()
# Put test values in the input streams.
x.extend([5, 4, 8, -3, -1, 10, -5, 6, 20, -12])
# Execute a step
run()
# Look at recent values of streams.
print ('recent values of stream y are')
print (recent_values(y))
print ('recent values of stream z are')
print (recent_values(z))
Explanation: ANOTHER EXAMPLE of f_item
Problem
If an item on stream x is greater than the maximum (positive) item seen so far on x then put the item on stream y, and 
iff an item on stream x is less than the minimum (negative) item seen so far on x then put the item on stream z
First step:  define func
func operates on an item of a stream and keyword arguments. In this example, func is g and the keyword arguments of g (in addition to item) are:
max_and_min, max_stream, min_stream.
Second step: define streams and call f_item( )
The keyword arguments of func are passed as keyword arguments of f_item.
Third step: Test the function
Put items in the input streams, run(), and look at the recent values of the output streams.
End of explanation
from IoTPy.agent_types.simple import f_window
x = Stream('x')
y = Stream('y')
def g(window, out_stream):
    out_stream.append(sum(window))
f_window(func=g, in_stream=x, window_size=2, step_size=1, out_stream=y)
# Put test values in the input streams.
x.extend(list(range(10)))
# Execute a step
run()
# Look at recent values of streams.
print ('recent values of stream y are')
print (recent_values(y))
Explanation: SLIDING WINDOWS OF STREAMS
f_item operates on the next item in a stream. f_window operates on the next sliding window of the input stream. 
A sliding window is specified by:
1. window_size and
2. step_size
both of which are positive integers.
If the input stream is x then the sequence of windows are:
x[0:window_size], x[step_size: step_size+window_size], ... , x[n * step_size : n * step_size + window_size]
For example, if window_size = 2 and step_size = 1 then the sequence of windows are:
x[0,1], x[1, 2], x[2, 3], x[3, 4], ..
Example of sliding windows
The call to f_window has arguments func, in_stream, window_size, and step_size, and other keyword arguments.
The arguments in func are window which is a list or array, and the same keyword arguments.
In this example the items of stream y are the sums of the sliding windows multiplied by multiplier. func is g which has a first argument, window which is a list, and keyword arguments out_stream and multiplier.
The call to f_window has the arguments func, in_stream, window_size, step_size and the same keyword arguments out_stream and multiplier as in g.
End of explanation
from IoTPy.agent_types.simple import join_synch
w = Stream(name='w')
x = Stream(name='x')
y = Stream(name='y')
z = Stream(name='z')
def h(alist, out_stream): out_stream.append(sum(alist))
join_synch(func=h, in_streams=[w, x, y], out_stream=z)
# Put test values in the input streams.
w.extend(list(range(100, 110)))
x.extend(list(range(0, 20, 2)))
y.extend(list(range(5)))
# Execute a step
run()
# Look at recent values of streams.
print ('recent values of stream w are')
print (recent_values(w))
print ('recent values of stream x are')
print (recent_values(x))
print ('recent values of stream y are')
print (recent_values(y))
print ('recent values of stream z are')
print (recent_values(z))
Explanation: Synchronous Join
in_streams stands for a list of streams (whereas in_stream represents a single stream). A synchronous join applies a function func to the list in_streams[0][i], in_streams[1][i], in_streams[2][i],... for all i.
For example, suppose x is the stream of nonnegative integers, [0, 1, 2, 3, ...] and y is the stream of letters of the alphabet repeated forever, i.e. y is ['A', 'B', 'C', .....'Z', 'A', 'B', ..]. Then join_synch applies function func to the lists [0, 'A'], [1, 'B'], [2, 'C'], ....
The arguments of join_synch are func, in_streams, and additional keyword arguments. The arguments of func are a list of joined items from the streams of in_streams and the same additional keyword arguments as in join_synch.
Example of join_synch
Given streams w, x, y, z we want to set z[i] = sum([w[i], x[i], y[i]).
End of explanation
from IoTPy.agent_types.simple import join_asynch
def h(index_item, out_stream):
    index, item = index_item
    print (index,': ', item)
    out_stream.append(item)
x = Stream('x')
y = Stream('y')
z = Stream('z')
join_asynch(func=h, in_streams=[x, y], out_stream=z)
# Put test values in the input streams.
x.extend([10, 11])
y.extend([0, 1, 2])
# Execute a step
run()
# Look at recent values of streams.
print ('recent values of stream z are')
print (recent_values(z))
# Put test values in the input streams.
x.extend([12, 13, 14])
y.extend([3])
# Execute a step
run()
# Look at recent values of streams.
print ('recent values of stream z are')
print (recent_values(z))
Explanation: Asynchronous Join
The function join_asynch has arguments func, in_streams - a list of streams, and additional keyword arguments.
When an item v is appended to the i-th stream of in_streams, func is applied to the tuple [i, v]. The first element of the tuple indicates the stream along which the item arrived, and the second element is the item itself.
Different runs of join_asynch may produce different results because items from the different streams arrive in different orders.
The arguments of func are index_item and additional keyword arguments. index_item is a tuple of the stream index and the item value, 
Example of asynchronous join
Print items as they arrive in streams x and y and put the items in stream z.
In this example func is h. Function h has an argument index_item and keyword argument out_stream which is also used in the call to join_asynch.
End of explanation
from IoTPy.agent_types.simple import join_timed
def f(timestamped_list):
    print (timestamped_list)
x = Stream(name='x')
y = Stream(name='y')
join_timed(func=f, in_streams=[x, y])
# Put test values in the input streams.
x.extend([[1, 'x[0]'], [3, 'x[1]']])
y.extend([[3, 'y[0]'], [5, 'y[2]'], [5, 'y[3]']])
# Execute a step
run()
# Put test values in the input streams.
x.append([4, 'x[2]'])
run()
x.append([5, 'x[3]'])
run()
Explanation: Time-based Join
This function operates on streams in which items are timestamped. Elements in each stream consist of pairs (timestamp, item). The timestamps on each stream are monotone increasing, i.e. the timestamp of a later element on a stream must be greater than the timestamp of an earlier element on the same stream.
The following situation can arise. An element [t, v] of one input stream arrives before an element [t', v'] of a different input stream, where t > t'.
The arguments of join_timed are func and in_streams - a list of streams, and additional keyword arguments which appear in func.
The function func operates on a tuple of the form [timestamp, list_of_items] where list_of_items is a list of N values where N is the number of streams in in_streams.
Consider a tuple [T, L] where T is the timestamp and L is the list of items. If there is no element with timestamp T on stream k then L[k] is None. If there is an element [T, v] in the k-th stream then L[k] is v.
End of explanation
<END_TASK>