h2o-py/h2o/demos.py
def _run_demo(body_fn, interactive, echo, testing):
    """
    Execute the demo, echoing commands and pausing for user input.

    :param body_fn: function that contains the sequence of demo's commands.
    :param interactive: If True, the user will be prompted to continue the demonstration after every segment.
    :param echo: If True, the python commands that are executed will be displayed.
    :param testing: Used for pyunit testing. h2o.init() will not be called if set to True.
    :type body_fn: function
    """
    import colorama
    from colorama import Style, Fore
    colorama.init()

    class StopExecution(Exception):
        """Helper class for cancelling the demo."""

    assert_is_type(body_fn, type(_run_demo))
    # Reformat description by removing extra spaces; then print it.
    if body_fn.__doc__:
        desc_lines = body_fn.__doc__.split("\n")
        while desc_lines[0].strip() == "":
            desc_lines = desc_lines[1:]
        while desc_lines[-1].strip() == "":
            desc_lines = desc_lines[:-1]
        strip_spaces = min(len(line) - len(line.lstrip(" ")) for line in desc_lines[1:] if line.strip() != "")
        maxlen = max(len(line) for line in desc_lines)
        print(Fore.CYAN)
        print("-" * maxlen)
        for line in desc_lines:
            print(line[strip_spaces:].rstrip())
        print("-" * maxlen)
        print(Style.RESET_ALL, end="")

    # Prepare the executor function
    def controller():
        """Print to console the next block of commands, and wait for keypress."""
        try:
            raise RuntimeError("Catch me!")
        except RuntimeError:
            print()
            # Extract and print lines that will be executed next
            if echo:
                tb = sys.exc_info()[2]
                fr = tb.tb_frame.f_back
                filename = fr.f_code.co_filename
                linecache.checkcache(filename)
                line = linecache.getline(filename, fr.f_lineno, fr.f_globals).rstrip()
                indent_len = len(line) - len(line.lstrip(" "))
                assert line[indent_len:] == "go()"
                i = fr.f_lineno
                output_lines = []
                n_blank_lines = 0
                while True:
                    i += 1
                    line = linecache.getline(filename, i, fr.f_globals).rstrip()
                    # Detect dedent
                    if line[:indent_len].strip() != "": break
                    line = line[indent_len:]
                    if line == "go()": break
                    style = Fore.LIGHTBLACK_EX if line.lstrip().startswith("#") else Style.BRIGHT
                    prompt = "... " if line.startswith(" ") else ">>> "
                    output_lines.append(Fore.CYAN + prompt + Fore.RESET + style + line + Style.RESET_ALL)
                    del style  # Otherwise exception print-outs may get messed-up...
                    if line.strip() == "":
                        n_blank_lines += 1
                        if n_blank_lines > 5: break  # Just in case we hit file end or something
                    else:
                        n_blank_lines = 0
                for line in output_lines[:-n_blank_lines]:
                    print(line)

            # Prompt for user input
            if interactive:
                print("\n" + Style.DIM + "(press any key)" + Style.RESET_ALL, end="")
                key = _wait_for_keypress()
                print("\r                     \r", end="")
                if key.lower() == "q":
                    raise StopExecution()

    # Replace h2o.init() with a stub when running in "test" mode
    _h2o_init = h2o.init
    if testing:
        h2o.init = lambda *args, **kwargs: None

    # Run the test
    try:
        body_fn(controller)
        print("\n" + Fore.CYAN + "---- End of Demo ----" + Style.RESET_ALL)
    except (StopExecution, KeyboardInterrupt):
        print("\n" + Fore.RED + "---- Demo aborted ----" + Style.RESET_ALL)

    # Clean-up
    if testing:
        h2o.init = _h2o_init
    print()
    colorama.deinit()
def _run_demo(body_fn, interactive, echo, testing):
    """
    Execute the demo, echoing commands and pausing for user input.

    :param body_fn: function that contains the sequence of demo's commands.
    :param interactive: If True, the user will be prompted to continue the demonstration after every segment.
    :param echo: If True, the python commands that are executed will be displayed.
    :param testing: Used for pyunit testing. h2o.init() will not be called if set to True.
    :type body_fn: function
    """
    import colorama
    from colorama import Style, Fore
    colorama.init()

    class StopExecution(Exception):
        """Helper class for cancelling the demo."""

    assert_is_type(body_fn, type(_run_demo))
    # Reformat description by removing extra spaces; then print it.
    if body_fn.__doc__:
        desc_lines = body_fn.__doc__.split("\n")
        while desc_lines[0].strip() == "":
            desc_lines = desc_lines[1:]
        while desc_lines[-1].strip() == "":
            desc_lines = desc_lines[:-1]
        strip_spaces = min(len(line) - len(line.lstrip(" ")) for line in desc_lines[1:] if line.strip() != "")
        maxlen = max(len(line) for line in desc_lines)
        print(Fore.CYAN)
        print("-" * maxlen)
        for line in desc_lines:
            print(line[strip_spaces:].rstrip())
        print("-" * maxlen)
        print(Style.RESET_ALL, end="")

    # Prepare the executor function
    def controller():
        """Print to console the next block of commands, and wait for keypress."""
        try:
            raise RuntimeError("Catch me!")
        except RuntimeError:
            print()
            # Extract and print lines that will be executed next
            if echo:
                tb = sys.exc_info()[2]
                fr = tb.tb_frame.f_back
                filename = fr.f_code.co_filename
                linecache.checkcache(filename)
                line = linecache.getline(filename, fr.f_lineno, fr.f_globals).rstrip()
                indent_len = len(line) - len(line.lstrip(" "))
                assert line[indent_len:] == "go()"
                i = fr.f_lineno
                output_lines = []
                n_blank_lines = 0
                while True:
                    i += 1
                    line = linecache.getline(filename, i, fr.f_globals).rstrip()
                    # Detect dedent
                    if line[:indent_len].strip() != "": break
                    line = line[indent_len:]
                    if line == "go()": break
                    style = Fore.LIGHTBLACK_EX if line.lstrip().startswith("#") else Style.BRIGHT
                    prompt = "... " if line.startswith(" ") else ">>> "
                    output_lines.append(Fore.CYAN + prompt + Fore.RESET + style + line + Style.RESET_ALL)
                    del style  # Otherwise exception print-outs may get messed-up...
                    if line.strip() == "":
                        n_blank_lines += 1
                        if n_blank_lines > 5: break  # Just in case we hit file end or something
                    else:
                        n_blank_lines = 0
                for line in output_lines[:-n_blank_lines]:
                    print(line)

            # Prompt for user input
            if interactive:
                print("\n" + Style.DIM + "(press any key)" + Style.RESET_ALL, end="")
                key = _wait_for_keypress()
                print("\r                     \r", end="")
                if key.lower() == "q":
                    raise StopExecution()

    # Replace h2o.init() with a stub when running in "test" mode
    _h2o_init = h2o.init
    if testing:
        h2o.init = lambda *args, **kwargs: None

    # Run the test
    try:
        body_fn(controller)
        print("\n" + Fore.CYAN + "---- End of Demo ----" + Style.RESET_ALL)
    except (StopExecution, KeyboardInterrupt):
        print("\n" + Fore.RED + "---- Demo aborted ----" + Style.RESET_ALL)

    # Clean-up
    if testing:
        h2o.init = _h2o_init
    print()
    colorama.deinit()
def _run_demo ( body_fn , interactive , echo , testing ) : import colorama from colorama import Style , Fore colorama . init ( ) class StopExecution ( Exception ) : """Helper class for cancelling the demo.""" assert_is_type ( body_fn , type ( _run_demo ) ) # Reformat description by removing extra spaces; then print it. if body_fn . __doc__ : desc_lines = body_fn . __doc__ . split ( "\n" ) while desc_lines [ 0 ] . strip ( ) == "" : desc_lines = desc_lines [ 1 : ] while desc_lines [ - 1 ] . strip ( ) == "" : desc_lines = desc_lines [ : - 1 ] strip_spaces = min ( len ( line ) - len ( line . lstrip ( " " ) ) for line in desc_lines [ 1 : ] if line . strip ( ) != "" ) maxlen = max ( len ( line ) for line in desc_lines ) print ( Fore . CYAN ) print ( "-" * maxlen ) for line in desc_lines : print ( line [ strip_spaces : ] . rstrip ( ) ) print ( "-" * maxlen ) print ( Style . RESET_ALL , end = "" ) # Prepare the executor function def controller ( ) : """Print to console the next block of commands, and wait for keypress.""" try : raise RuntimeError ( "Catch me!" ) except RuntimeError : print ( ) # Extract and print lines that will be executed next if echo : tb = sys . exc_info ( ) [ 2 ] fr = tb . tb_frame . f_back filename = fr . f_code . co_filename linecache . checkcache ( filename ) line = linecache . getline ( filename , fr . f_lineno , fr . f_globals ) . rstrip ( ) indent_len = len ( line ) - len ( line . lstrip ( " " ) ) assert line [ indent_len : ] == "go()" i = fr . f_lineno output_lines = [ ] n_blank_lines = 0 while True : i += 1 line = linecache . getline ( filename , i , fr . f_globals ) . rstrip ( ) # Detect dedent if line [ : indent_len ] . strip ( ) != "" : break line = line [ indent_len : ] if line == "go()" : break style = Fore . LIGHTBLACK_EX if line . lstrip ( ) . startswith ( "#" ) else Style . BRIGHT prompt = "... " if line . startswith ( " " ) else ">>> " output_lines . append ( Fore . CYAN + prompt + Fore . RESET + style + line + Style . RESET_ALL ) del style # Otherwise exception print-outs may get messed-up... if line . strip ( ) == "" : n_blank_lines += 1 if n_blank_lines > 5 : break # Just in case we hit file end or something else : n_blank_lines = 0 for line in output_lines [ : - n_blank_lines ] : print ( line ) # Prompt for user input if interactive : print ( "\n" + Style . DIM + "(press any key)" + Style . RESET_ALL , end = "" ) key = _wait_for_keypress ( ) print ( "\r                     \r" , end = "" ) if key . lower ( ) == "q" : raise StopExecution ( ) # Replace h2o.init() with a stub when running in "test" mode _h2o_init = h2o . init if testing : h2o . init = lambda * args , * * kwargs : None # Run the test try : body_fn ( controller ) print ( "\n" + Fore . CYAN + "---- End of Demo ----" + Style . RESET_ALL ) except ( StopExecution , KeyboardInterrupt ) : print ( "\n" + Fore . RED + "---- Demo aborted ----" + Style . RESET_ALL ) # Clean-up if testing : h2o . init = _h2o_init print ( ) colorama . deinit ( )
Execute the demo, echoing commands and pausing for user input.

    :param body_fn: function that contains the sequence of demo's commands.
    :param interactive: If True, the user will be prompted to continue the demonstration after every segment.
    :param echo: If True, the python commands that are executed will be displayed.
    :param testing: Used for pyunit testing. h2o.init() will not be called if set to True.
    :type body_fn: function
Execute the demo echoing commands and pausing for user input .