fixed_reply.py

from typing import Dict, Any

from aiflows.base_flows import AtomicFlow


class FixedReplyFlow(AtomicFlow):
    """ This class implements a FixedReplyFlow. It's used to reply with a fixed reply.
    
    *Configuration Parameters*:
    
    - `name` (str): The name of the flow.
    
    - `description` (str): A description of the flow. This description is used to generate the help message of the flow.
    
    - `fixed_reply` (str): The fixed reply to reply with.
    
    - The other configuration parameters are inherited from the default configuration of AtomicFlow (see AtomicFlow)
    
    *Input Interface*:
        
    - None
            
    Output Interface:
    
    - `fixed_reply` (str): The fixed reply.
            
    :param \**kwargs: The keyword arguments passed to the AtomicFlow constructor. Among these is the flow_config which should also contain the "fixed_reply" key.
    :type \**kwargs: Dict[str, Any]
    """
    REQUIRED_KEYS_CONFIG = ["fixed_reply"]

    __default_flow_config = {
        "input_interface": [],
        "output_interface": ["fixed_reply"],
    }

    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    def run(self,
            input_data: Dict[str, Any]) -> Dict[str, Any]:
        """ Runs the FixedReplyFlow. It's used to reply with a fixed reply.
        
        :param input_data: The input data dictionary
        :type input_data: Dict[str, Any]
        :return: The fixed reply
        :rtype: Dict[str, Any]
        """

        return {"fixed_reply": self.flow_config["fixed_reply"]}