Skip to content

IOStream

autogen.io.IOStream #

Bases: InputStream, OutputStream, Protocol

A protocol for input/output streams.

print #

print(*objects, sep=' ', end='\n', flush=False)

Print data to the output stream.

    Args:
        objects (any): The data to print.
        sep (str, optional): The separator between objects. Defaults to " ".
        end (str, optional): The end of the output. Defaults to "

". flush (bool, optional): Whether to flush the output. Defaults to False.

Source code in autogen/io/base.py
def print(self, *objects: Any, sep: str = " ", end: str = "\n", flush: bool = False) -> None:
    """Print data to the output stream.

    Args:
        objects (any): The data to print.
        sep (str, optional): The separator between objects. Defaults to " ".
        end (str, optional): The end of the output. Defaults to "\n".
        flush (bool, optional): Whether to flush the output. Defaults to False.
    """
    ...  # pragma: no cover

send #

send(message)

Send data to the output stream.

PARAMETER DESCRIPTION
message

BaseMessage from autogen.messages.base_message

TYPE: BaseMessage

Source code in autogen/io/base.py
def send(self, message: BaseMessage) -> None:
    """Send data to the output stream.

    Args:
        message (BaseMessage): BaseMessage from autogen.messages.base_message
    """
    ...

input #

input(prompt='', *, password=False)

Read a line from the input stream.

PARAMETER DESCRIPTION
prompt

The prompt to display. Defaults to "".

TYPE: str DEFAULT: ''

password

Whether to read a password. Defaults to False.

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
str

The line read from the input stream.

TYPE: str

Source code in autogen/io/base.py
def input(self, prompt: str = "", *, password: bool = False) -> str:
    """Read a line from the input stream.

    Args:
        prompt (str, optional): The prompt to display. Defaults to "".
        password (bool, optional): Whether to read a password. Defaults to False.

    Returns:
        str: The line read from the input stream.

    """
    ...  # pragma: no cover

set_global_default staticmethod #

set_global_default(stream)

Set the default input/output stream.

PARAMETER DESCRIPTION
stream

The input/output stream to set as the default.

TYPE: IOStream

Source code in autogen/io/base.py
@staticmethod
def set_global_default(stream: "IOStream") -> None:
    """Set the default input/output stream.

    Args:
        stream (IOStream): The input/output stream to set as the default.
    """
    IOStream._global_default = stream

get_global_default staticmethod #

get_global_default()

Get the default input/output stream.

RETURNS DESCRIPTION
IOStream

The default input/output stream.

TYPE: IOStream

Source code in autogen/io/base.py
@staticmethod
def get_global_default() -> "IOStream":
    """Get the default input/output stream.

    Returns:
        IOStream: The default input/output stream.
    """
    if IOStream._global_default is None:
        raise RuntimeError("No global default IOStream has been set")
    return IOStream._global_default

get_default staticmethod #

get_default()

Get the default input/output stream.

RETURNS DESCRIPTION
IOStream

The default input/output stream.

TYPE: IOStream

Source code in autogen/io/base.py
@staticmethod
def get_default() -> "IOStream":
    """Get the default input/output stream.

    Returns:
        IOStream: The default input/output stream.
    """
    iostream = IOStream._default_io_stream.get()
    if iostream is None:
        iostream = IOStream.get_global_default()
        # Set the default IOStream of the current context (thread/cooroutine)
        IOStream.set_default(iostream)
    return iostream

set_default staticmethod #

set_default(stream)

Set the default input/output stream.

PARAMETER DESCRIPTION
stream

The input/output stream to set as the default.

TYPE: IOStream

Source code in autogen/io/base.py
@staticmethod
@contextmanager
def set_default(stream: Optional["IOStream"]) -> Iterator[None]:
    """Set the default input/output stream.

    Args:
        stream (IOStream): The input/output stream to set as the default.
    """
    global _default_io_stream
    try:
        token = IOStream._default_io_stream.set(stream)
        yield
    finally:
        IOStream._default_io_stream.reset(token)

    return