<commit_before>"""
This module consists of functions that operate on a single input stream
to produce a single output stream. These stream-functions encapsulate
functions that operate on standard data types such as integers. 

Agents in the module:
   1. map_element
   2. signal_element
   3. filter_element
   4. map_list
   5. map_window
   6. map_window_list
   7. timed_window

In addition functions that return streams are:
   map_element_f (version of map_element)
   filter_element_f (version of filter_element)
   map_window_f (version of map_window)
   map_list_f (version of map_list)
   map_array_f (stream array version of map_list)
   timed_window_f (version of timed_window)

"""

import sys
import os
sys.path.append(os.path.abspath("../core"))
sys.path.append(os.path.abspath("../helper_functions"))

from agent import Agent
from stream import Stream, _multivalue
from check_agent_parameter_types import *

#------------------------------------------------------------------------------------
def map_element(
        func, in_stream, out_stream,
        state=None, call_streams=None, name=None,
        *args, **kwargs):
    """
    This agent maps the function func from its single input stream to its
    single output stream.

    Parameters
    ----------
        func: function
           function from an element of the in_stream to an element of
           the out_stream. func may also have state and kwargs.
        in_stream: Stream or StreamArray
           The single input stream of this agent
        out_stream: Stream or StreamArray
           The single output stream of the agent
        state: object
           The state of the agent
        call_streams: list of Stream
           The list of call_streams. A new value in any stream in this
           list causes a state transition of this agent.
        name: Str
           Name of the agent created by this function.
        *args, **kwargs:
           Positional and keyword parameters, if any, for func.
    Returns
    -------
        Agent.
         The agent created by this function.
    Uses
    ----
       * Agent
       * check_map_agent_arguments
       * check_num_args_in_func
    Used by
    -------
        map_element_f
    Examples
    --------
    With
    map_element(func=lambda x:2*x, in_stream=u, out_stream=v)
    then:
        v[i] = 2*u[i], for all i in streams u, v

    With
    def f(x, state): return 2*x+state, state+10
    map_element(func=f, in_stream=u, out_stream=v, state=0)
    then:
       state[i] = 10*i
       v[i] = 2*u[i] + state[i]


    """
    check_map_agent_arguments(func, in_stream, out_stream, call_streams, name)
    check_num_args_in_func(state, name, func, args, kwargs)

    # The transition function for this agent.
    def transition(in_lists, state):
        num_in_streams = 1
        check_in_lists_type(name, in_lists, num_in_streams)
        in_list = in_lists[0]
        input_list = in_list.list[in_list.start:in_list.stop]
        # If the new input data is empty then return an empty list for
        # the single output stream, and leave the state and the starting
        # point for the single input stream unchanged.
        if input_list is None or len(input_list) == 0:
            return ([[]], state, [in_list.start])

        if state is None:
            output_list = [func(v, *args, **kwargs) for v in input_list]
        else:
            output_list = [[]]*len(input_list)
            for i in range(len(input_list)):
                output_list[i], state = func(input_list[i], state, *args, **kwargs)
                
        return ([output_list], state, [in_list.start+len(input_list)])
    # Finished transition

    # Create agent
    return Agent([in_stream], [out_stream], transition, state, call_streams, name)


#------------------------------------------------------------------------------------
def signal_element(
        func, in_stream, out_stream,
        state=None, name=None,
        *args, **kwargs):
    """
    This agent executes func when it reads a new value on
    in_stream. The function func operates on kwargs and possibly
    a state. Note that func does not operate on elements of in_stream.
    This is because in_stream serves merely as a signaling mechanism
    that informs this agent that it needs to execute a step, i.e.,
    call func.

    Parameters
    ----------
        func: function
           function that operates on kwargs and possibly the state.
        in_stream: Stream or StreamArray
           The single input stream of this agent
        out_stream: Stream or StreamArray
           The single output stream of the agent
        state: object
           The state of the agent
        name: Str
           Name of the agent created by this function.
        *args, **kwargs:
           Positional and keyword parameters, if any, for func.
    Returns
    -------
        Agent.
         The agent created by this function.
    Uses
    ----
       * Agent
       * check_map_agent_arguments

    """
    call_streams=None
    check_map_agent_arguments(func, in_stream, out_stream, call_streams, name)

    # The transition function for this agent.
    def transition(in_lists, state):
        num_in_streams = 1
        check_in_lists_type(name, in_lists, num_in_streams)
        in_list = in_lists[0]
        input_list = in_list.list[in_list.start:in_list.stop]
        # If the new input data is empty then return an empty list for
        # the single output stream, and leave the state and the starting
        # point for the single input stream unchanged.
        if input_list is None or len(input_list) == 0:
            return ([[]], state, [in_list.start])
        # In the following output is (typically) _no_value if the
        # variables listed in kwargs do not change, and is any other
        # value if any variable does change.
        else:
            if state is None:
                output = func(*args, **kwargs)
            else:
                output, state = func(state, *args, **kwargs)

        return ([[output]], state, [in_list.start+len(input_list)])
    # Finished transition

    # Create agent
    return Agent([in_stream], [out_stream], transition, state, call_streams, name)


#------------------------------------------------------------------------------------
def filter_element(
        func, in_stream, out_stream,
        state=None, call_streams=None, name=None,
        *args, **kwargs):
    """
    This agent uses the boolean function func to filter its single input stream
    to produce a single output stream.

    Parameters
    ----------
        func: function
           function from an element of the in_stream to Boolean.
        in_stream: Stream
           The single input stream of this agent
        out_stream: Stream
           The single output stream of the agent
        state: object
           The state of the agent
        call_streams: list of Stream
           The list of call_streams. A new value in any stream in this
           list causes a state transition of this agent.
        name: Str
           Name of the agent created by this function.
    Returns
    -------
        Agent.
         The agent created by this function.
    Uses
    ----
       * Agent
       * check_map_agent_arguments
       * check_num_args_in_func
    
    Used by
    -------
        filter_element_f
    Example
    -------
    With
    filter_element(func=lambda x: x % 2, in_stream=u, out_stream=v)
    then:
        stream v consists of the elements of stream u that are odd.
        The even elements are filtered out.

    """
    check_map_agent_arguments(func, in_stream, out_stream, call_streams, name)
    check_num_args_in_func(state, name, func, args, kwargs)

    # The transition function for this agent.
    def transition(in_lists, state):
        num_in_streams = 1
        check_in_lists_type(name, in_lists, num_in_streams)
        in_list = in_lists[0]
        input_list = in_list.list[in_list.start:in_list.stop]
        # If the new input data is empty then return an empty list for
        # the single output stream, and leave the state and the starting
        # point for the single input stream unchanged.
        if input_list is None or len(input_list) == 0:
            return ([[]], state, [in_list.start])

        if state is None:
            output_list = [v for v in input_list if not func(v, *args, **kwargs) ]
        else:
            output_list = []
            for i in range(len(input_list)):
                boole, state = func(input_list[i], state, *args, **kwargs)
                if not boole: output_list.append(input_list[i])
                
        return ([output_list], state, [in_list.start+len(input_list)])
    # Finished transition

    # Create agent
    return Agent([in_stream], [out_stream], transition, state, call_streams, name)


#------------------------------------------------------------------------------------
def map_element_f(func, in_stream, state=None, *args, **kwargs):
    """
    map_element_f returns out_stream, a stream obtained by applying function to each
    element of the input stream, in_stream.

    Parameters<commit_msg>Version 0.0<commit_after>"""
This module consists of functions that operate on a single input stream
to produce a single output stream. These stream-functions encapsulate
functions that operate on standard data types such as integers. 

Agents in the module:
   1. map_element
   2. signal_element
   3. filter_element
   4. map_list
   5. map_window
   6. map_window_list
   7. timed_window

In addition functions that return streams are:
   map_element_f (version of map_element)
   filter_element_f (version of filter_element)
   map_window_f (version of map_window)
   map_list_f (version of map_list)
   map_array_f (stream array version of map_list)
   timed_window_f (version of timed_window)

"""

import sys
import os
sys.path.append(os.path.abspath("../core"))
sys.path.append(os.path.abspath("../helper_functions"))

from agent import Agent
from stream import Stream, _multivalue
from check_agent_parameter_types import *

#------------------------------------------------------------------------------------
def map_element(
        func, in_stream, out_stream,
        state=None, call_streams=None, name=None,
        *args, **kwargs):
    """
    This agent maps the function func from its single input stream to its
    single output stream.

    Parameters
    ----------
        func: function
           function from an element of the in_stream to an element of
           the out_stream. func may also have state and kwargs.
        in_stream: Stream or StreamArray
           The single input stream of this agent
        out_stream: Stream or StreamArray
           The single output stream of the agent
        state: object
           The state of the agent
        call_streams: list of Stream
           The list of call_streams. A new value in any stream in this
           list causes a state transition of this agent.
        name: Str
           Name of the agent created by this function.
        *args, **kwargs:
           Positional and keyword parameters, if any, for func.
    Returns
    -------
        Agent.
         The agent created by this function.
    Uses
    ----
       * Agent
       * check_map_agent_arguments
       * check_num_args_in_func
    Used by
    -------
        map_element_f
    Examples
    --------
    With
    map_element(func=lambda x:2*x, in_stream=u, out_stream=v)
    then:
        v[i] = 2*u[i], for all i in streams u, v

    With
    def f(x, state): return 2*x+state, state+10
    map_element(func=f, in_stream=u, out_stream=v, state=0)
    then:
       state[i] = 10*i
       v[i] = 2*u[i] + state[i]


    """
    check_map_agent_arguments(func, in_stream, out_stream, call_streams, name)
    check_num_args_in_func(state, name, func, args, kwargs)

    # The transition function for this agent.
    def transition(in_lists, state):
        num_in_streams = 1
        check_in_lists_type(name, in_lists, num_in_streams)
        in_list = in_lists[0]
        input_list = in_list.list[in_list.start:in_list.stop]
        # If the new input data is empty then return an empty list for
        # the single output stream, and leave the state and the starting
        # point for the single input stream unchanged.
        if input_list is None or len(input_list) == 0:
            return ([[]], state, [in_list.start])

        if state is None:
            output_list = [func(v, *args, **kwargs) for v in input_list]
        else:
            output_list = [[]]*len(input_list)
            for i in range(len(input_list)):
                output_list[i], state = func(input_list[i], state, *args, **kwargs)
                
        return ([output_list], state, [in_list.start+len(input_list)])
    # Finished transition

    # Create agent
    return Agent([in_stream], [out_stream], transition, state, call_streams, name)


#------------------------------------------------------------------------------------
def signal_element(
        func, in_stream, out_stream,
        state=None, name=None,
        *args, **kwargs):
    """
    This agent executes func when it reads a new value on
    in_stream. The function func operates on kwargs and possibly
    a state. Note that func does not operate on elements of in_stream.
    This is because in_stream serves merely as a signaling mechanism
    that informs this agent that it needs to execute a step, i.e.,
    call func.

    Parameters
    ----------
        func: function
           function that operates on kwargs and possibly the state.
        in_stream: Stream or StreamArray
           The single input stream of this agent
        out_stream: Stream or StreamArray
           The single output stream of the agent
        state: object
           The state of the agent
        name: Str
           Name of the agent created by this function.
        *args, **kwargs:
           Positional and keyword parameters, if any, for func.
    Returns
    -------
        Agent.
         The agent created by this function.
    Uses
    ----
       * Agent
       * check_map_agent_arguments

    """
    call_streams=None
    check_map_agent_arguments(func, in_stream, out_stream, call_streams, name)

    # The transition function for this agent.
    def transition(in_lists, state):
        num_in_streams = 1
        check_in_lists_type(name, in_lists, num_in_streams)
        in_list = in_lists[0]
        input_list = in_list.list[in_list.start:in_list.stop]
        # If the new input data is empty then return an empty list for
        # the single output stream, and leave the state and the starting
        # point for the single input stream unchanged.
        if input_list is None or len(input_list) == 0:
            return ([[]], state, [in_list.start])
        # In the following output is (typically) _no_value if the
        # variables listed in kwargs do not change, and is any other
        # value if any variable does change.
        else:
            if state is None:
                output = func(*args, **kwargs)
            else:
                output, state = func(state, *args, **kwargs)

        return ([[output]], state, [in_list.start+len(input_list)])
    # Finished transition

    # Create agent
    return Agent([in_stream], [out_stream], transition, state, call_streams, name)


#------------------------------------------------------------------------------------
def filter_element(
        func, in_stream, out_stream,
        state=None, call_streams=None, name=None,
        *args, **kwargs):
    """
    This agent uses the boolean function func to filter its single input stream
    to produce a single output stream.

    Parameters
    ----------
        func: function
           function from an element of the in_stream to Boolean.
        in_stream: Stream
           The single input stream of this agent
        out_stream: Stream
           The single output stream of the agent
        state: object
           The state of the agent
        call_streams: list of Stream
           The list of call_streams. A new value in any stream in this
           list causes a state transition of this agent.
        name: Str
           Name of the agent created by this function.
    Returns
    -------
        Agent.
         The agent created by this function.
    Uses
    ----
       * Agent
       * check_map_agent_arguments
       * check_num_args_in_func
    
    Used by
    -------
        filter_element_f
    Example
    -------
    With
    filter_element(func=lambda x: x % 2, in_stream=u, out_stream=v)
    then:
        stream v consists of the elements of stream u that are odd.
        The even elements are filtered out.

    """
    check_map_agent_arguments(func, in_stream, out_stream, call_streams, name)
    check_num_args_in_func(state, name, func, args, kwargs)

    # The transition function for this agent.
    def transition(in_lists, state):
        num_in_streams = 1
        check_in_lists_type(name, in_lists, num_in_streams)
        in_list = in_lists[0]
        input_list = in_list.list[in_list.start:in_list.stop]
        # If the new input data is empty then return an empty list for
        # the single output stream, and leave the state and the starting
        # point for the single input stream unchanged.
        if input_list is None or len(input_list) == 0:
            return ([[]], state, [in_list.start])

        if state is None:
            output_list = [v for v in input_list if func(v, *args, **kwargs) ]
        else:
            output_list = []
            for i in range(len(input_list)):
                boole, state = func(input_list[i], state, *args, **kwargs)
                if boole: output_list.append(input_list[i])
                
        return ([output_list], state, [in_list.start+len(input_list)])
    # Finished transition

    # Create agent
    return Agent([in_stream], [out_stream], transition, state, call_streams, name)


#------------------------------------------------------------------------------------
def map_element_f(func, in_stream, state=None, *args, **kwargs):
    """
    map_element_f returns out_stream, a stream obtained by applying function to each
    element of the input stream, in_stream.

    Parameters
{'fraction_non_alphanumeric': 0.1131101206652897, 'fraction_numerical': 0.00293510164148277, 'mean_word_length': 2.0226712666338096, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5734934', 'n_tokens_mistral': 5095, 'n_tokens_neox': 4993, 'n_words': 1834}