131 the display problem for package downloading progress

.github/workflows/cli_ci.yaml

@@ -11,7 +11,8 @@ jobs:
   build:
     strategy:
       matrix:
-        os: [ubuntu-latest, windows-latest, macos-latest]
+        #os: [ubuntu-latest, windows-latest, macos-latest]
+        os: [ubuntu-latest, windows-latest]
         python-version: ["3.10", "3.11"]
     runs-on: ${{ matrix.os }}
     steps:

chat/src/og_terminal/terminal_chat.py

@@ -19,6 +19,7 @@
 from rich.console import Group
 from rich.panel import Panel
 from rich.table import Table
+from rich.text import Text
 from rich.progress import Progress
 from rich.rule import Rule
 from rich.live import Live
@@ -184,10 +185,12 @@ def handle_action_output(segments, respond, values):
         new_stderr += respond.console_stderr
         new_stderr = process_char_stream(new_stderr)
     values.append(("text", (new_stdout, new_stderr), []))
+    total_output = new_stdout
     if new_stderr:
-        new_stdout += "\n" + new_stderr
+        total_output = new_stdout + "\n" + new_stderr
+    text = Text.from_ansi(total_output)
     syntax = Syntax(
-        f"{new_stdout}",
+        f"{text.plain}",
         "text",
         line_numbers=True,
     )

sdk/src/og_sdk/utils.py

@@ -42,17 +42,20 @@ async def generate_async_chunk(
 def process_char_stream(stream):
     buffer = []
     i = 0
+
+    def carriage_return(buf):
+        if "\n" in buf:
+            pop_buf = []
+            for _ in range(buf[::-1].index("\n")):
+                pop_buf.append(buf.pop())
+
     while i < len(stream):
         c = stream[i]
         if c in ["\b", "\r"]:
             # Handle escape characters
             escape_dict = {
                 "\b": lambda buf: buf.pop() if buf else None,  # backspace
-                "\r": lambda buf: [
-                    buf.pop() for _ in range(len(buf) - buf[::-1].index("\n") - 1)
-                ]
-                if "\n" in buf
-                else buf.clear(),  # carriage return
+                "\r": carriage_return,
             }
             escape_dict[c](buffer)
             i += 1

sdk/tests/utils_test.py

@@ -0,0 +1,54 @@
+#! /usr/bin/env python3
+# vim:fenc=utf-8
+#
+# Copyright © 2023 imotai <imotai@imotai-ub>
+#
+# Distributed under terms of the MIT license.
+
+""" """
+import pytest
+import json
+from og_sdk.utils import process_char_stream
+
+
+def test_process_char_stream():
+    stream0 = "  Downloading pyfiglet-1.0.2-py3-none-any.whl (1.1 MB)\r\n\x1b[?25l     \x1b[38;5;237m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\x1b[0m \x1b[32m0.0/1.1 MB\x1b[0m \x1b[31m?\x1b[0m eta \x1b[36m-:--:--\x1b[0m"
+    stream1 = "\r\x1b[2K     \x1b[38;5;237m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\x1b[0m \x1b[32m0.0/1.1 MB\x1b[0m \x1b[31m?\x1b[0m eta \x1b[36m-:--:--\x1b[0m"
+    output1 = process_char_stream(stream0 + stream1)
+    output2 = process_char_stream(output1 + stream1)
+    assert output1 == output2
+
+def test_empty_string():
+    assert process_char_stream("") == ""
+
+def test_single_character():
+    assert process_char_stream("a") == "a"
+
+def test_multiple_characters():
+    assert process_char_stream("abc") == "abc"
+
+def test_backspace():
+    assert process_char_stream("ab\b") == "a"
+
+def test_carriage_return():
+    assert process_char_stream("ab\r") == "ab"
+
+def test_carriage_return_with_newline():
+    assert process_char_stream("ab\r\n") == "ab\n"
+
+def test_backspace_and_carriage_return():
+    assert process_char_stream("ab\b\r") == "a"
+
+
+def test_carriage_return_and_backspace():
+    assert process_char_stream("ab\r\b") == "a"
+
+
+def test_mixed_escape_characters_and_regular_characters():
+    assert process_char_stream("ab\b\r\ncde") == "a\ncde"
+
+
+def test_special_characters():
+    assert process_char_stream("ab!@#$%^&*()_+{}|:\";'<>,.?/`~") == "ab!@#$%^&*()_+{}|:\";'<>,.?/`~"
+
+

up/src/og_up/up.py

@@ -65,8 +65,9 @@ def run_install_cli(live, segments):
     result_code = 0
     refresh(live, segments)
     outputs = ""
-    for code, output in run_with_realtime_print(command=["pip", "install", "-U", "og_proto",
-        "og_sdk", "og_chat"]):
+    for code, output in run_with_realtime_print(
+        command=["pip", "install", "-U", "og_proto", "og_sdk", "og_chat"]
+    ):
         outputs += output
         result_code = code
     if result_code == 0:

up/tests/up_tests.py

@@ -154,6 +154,7 @@ def test_install_cli():
     with Live(Group(*segments), console=console) as live:
         assert run_install_cli(live, segments)
 
+
 @pytest.mark.skipif(sys.platform.startswith("win"), reason="skip on windows")
 def test_start_azure_openai_smoketest():
     console = Console()