yay

app.py

@@ -3,6 +3,7 @@ import random
 import time
 
 import gradio as gr
+
 from gradio_logsview import LogsView
 
 
@@ -82,7 +83,7 @@ from gradio_logsview import LogsView
 def fn_process():
     # Run a process and capture all logs from the process
     yield from LogsView.run_process(
-        cmd=[mergekit-yaml", "config.yaml", "merge", "--copy-", "--cuda", "--low-cpu-memory"]
+        cmd=["mergekit-yaml", "config.yaml", "merge", "--copy-", "--cuda", "--low-cpu-memory"]
     )
 
 with gr.Blocks() as demo:

space.py

@@ -3,7 +3,8 @@ import random
 import time
 
 import gradio as gr
-from gradio_logsview import LogsView
+
+from gradio_logsview import LogsView, LogsViewRunner
 
 
 def random_values(failing: bool = False):
@@ -26,19 +27,27 @@ def random_values(failing: bool = False):
 
 
 def fn_process_success():
-    yield from LogsView.run_process(["python", "-u", "script.py"])
+    runner = LogsViewRunner()
+    yield from runner.run_process(["python", "-u", "demo/script.py"])
+    yield runner.log(f"Runner: {runner}")
 
 
 def fn_process_failing():
-    yield from LogsView.run_process(["python", "-u", "script.py", "--failing"])
+    runner = LogsViewRunner()
+    yield from runner.run_process(["python", "-u", "demo/script.py", "--failing"])
+    yield runner.log(f"Runner: {runner}")
 
 
 def fn_thread_success():
-    yield from LogsView.run_thread(random_values, log_level=logging.INFO, failing=False)
+    runner = LogsViewRunner()
+    yield from runner.run_thread(random_values, log_level=logging.INFO, failing=False)
+    yield runner.log(f"Runner: {runner}")
 
 
 def fn_thread_failing():
-    yield from LogsView.run_thread(random_values, log_level=logging.INFO, failing=True)
+    runner = LogsViewRunner()
+    yield from runner.run_thread(random_values, log_level=logging.INFO, failing=True)
+    yield runner.log(f"Runner: {runner}")
 
 
 markdown_top = """
@@ -56,16 +65,15 @@ markdown_bottom = """
 ## Installation
 
 ```
-pip install https://huggingface.co/spaces/Wauplin/gradio_logsview/resolve/main/gradio_logsview-0.0.1-py3-none-any.whl
+pip install https://huggingface.co/spaces/Wauplin/gradio_logsview/resolve/main/gradio_logsview-0.0.3-py3-none-any.whl
 ```
 
 or add this line to your `requirements.txt`:
 
 ```
-gradio_logsview@https://huggingface.co/spaces/Wauplin/gradio_logsview/resolve/main/gradio_logsview-0.0.1-py3-none-any.whl
+gradio_logsview@https://huggingface.co/spaces/Wauplin/gradio_logsview/resolve/main/gradio_logsview-0.0.3-py3-none-any.whl
 ```
 
-
 ## How to run in a thread?
 
 With `LogsView.run_thread`, you can run a function in a separate thread and capture logs in real-time.
@@ -77,7 +85,9 @@ from gradio_logsview import LogsView
 def fn_thread():
     # Run `my_function` in a separate thread
     # All logs above `INFO` level will be captured and displayed in real-time.
-    yield from LogsView.run_thread(my_function, log_level=logging.INFO, arg1="value1")
+    runner = LogsViewRunner() # Initialize the runner
+    yield from runner.run_thread(my_function, log_level=logging.INFO, arg1="value1")
+    yield runner.log(f"Runner: {runner}") # Log any message
 
 with gr.Blocks() as demo:
     logs = LogsView()
@@ -94,9 +104,11 @@ from gradio_logsview import LogsView
 
 def fn_process():
     # Run a process and capture all logs from the process
-    yield from LogsView.run_process(
-        [mergekit-yaml", "config.yaml", "merge", "--copy-", "--cuda", "--low-cpu-memory"]
+    runner = LogsViewRunner() # Initialize the runner
+    yield from runner.run_process(
+        cmd=["mergekit-yaml", "config.yaml", "merge", "--copy-", "--cuda", "--low-cpu-memory"]
     )
+    yield runner.log(f"Runner: {runner}") # Log any message
 
 with gr.Blocks() as demo:
     logs = LogsView()
@@ -110,8 +122,9 @@ with gr.Blocks() as demo:
 - [ ] format logs client-side (front-end)
 - [ ] scrollable logs if more than N lines (front-end)
 - [ ] format each log only once (front-end)
-- [ ] stop process if `run_process` gets cancelled (back-end)
-- [ ] correctly pass error stacktrace in `run_thread` (back-end)
+- [x] stop process if `run_process` gets cancelled (back-end)
+- [x] correctly pass error stacktrace in `run_thread` (back-end)
+- [ ] correctly catch print statements in `run_thread` (back-end)
 - [ ] disable interactivity + remove all code editing logic (both?)
 - [ ] how to handle progress bars? (i.e when logs are overwritten in terminal)
 """

src/README.md

@@ -97,7 +97,7 @@ from gradio_logsview import LogsView
 def fn_process():
     # Run a process and capture all logs from the process
     yield from LogsView.run_process(
-        cmd=[mergekit-yaml", "config.yaml", "merge", "--copy-", "--cuda", "--low-cpu-memory"]
+        cmd=["mergekit-yaml", "config.yaml", "merge", "--copy-", "--cuda", "--low-cpu-memory"]
     )
 
 with gr.Blocks() as demo:

src/backend/gradio_logsview/__init__.py

@@ -1,3 +1,3 @@
-from .logsview import LogsView
+from .logsview import Log, LogsView, LogsViewRunner
 
-__all__ = ["LogsView"]
+__all__ = ["Log", "LogsView", "LogsViewRunner"]

src/backend/gradio_logsview/logsview.py

@@ -1,12 +1,10 @@
 """LogsView() custom component"""
 
-from __future__ import annotations
-
 import logging
 import queue
 import subprocess
-import threading
 import time
+import traceback
 from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import datetime
@@ -14,19 +12,163 @@ from functools import wraps
 from logging.handlers import QueueHandler
 from queue import Queue
 from threading import Thread
-from typing import Any, Callable, Generator, Iterable, List, Literal
+from typing import Any, Callable, Generator, List, Literal, NoReturn
 
 from gradio.components.base import Component
 from gradio.events import Events
 
+LOGGING_LEVEL_T = Literal["INFO", "DEBUG", "WARNING", "ERROR", "CRITICAL"]
+
 
 @dataclass
 class Log:
-    level: Literal["INFO", "DEBUG", "WARNING", "ERROR", "CRITICAL"]
+    level: LOGGING_LEVEL_T
     message: str
     timestamp: str
 
 
+class LogsViewRunner:
+    def __init__(self, date_format: str = "%Y-%m-%d %H:%M:%S"):
+        self.date_format = date_format
+        self.logs: List[Log] = []
+
+        # Runner state
+        self.completed = False
+        self.error = False
+        self.process: subprocess.Popen | None = None
+        self.thread: Thread | None = None
+
+    def log(
+        self,
+        message: str,
+        level: LOGGING_LEVEL_T = "INFO",
+        timestamp: datetime | float | int | str | None = None,
+    ) -> List[Log]:
+        if timestamp is None:
+            timestamp = time.time()
+        if isinstance(timestamp, (float, int)):
+            timestamp = datetime.fromtimestamp(timestamp)
+        if isinstance(timestamp, datetime):
+            timestamp = timestamp.strftime(self.date_format)
+        self.logs.append(Log(level=level, message=message, timestamp=timestamp))
+        return self.logs
+
+    def run_process(
+        self, command: List[str], **kwargs
+    ) -> Generator[List[Log], None, None]:
+        """Run a command in a subprocess and yield logs in real-time."""
+        self.process = subprocess.Popen(
+            command,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            text=True,
+            **kwargs,
+        )
+
+        if self.process.stdout is None:
+            raise ValueError("stdout is None")
+
+        yield self.log(f"Running {' '.join(command)}")
+        for line in self.process.stdout:
+            yield self.log(line.strip())
+
+        self.process.stdout.close()
+        return_code = self.process.wait()
+        if return_code:
+            yield self.log(f"Process exited with code {return_code}", level="ERROR")
+            self.completed = True
+            self.error = True
+        else:
+            yield self.log("Process exited successfully", level="INFO")
+            self.completed = True
+            self.error = False
+
+    def run_thread(
+        self,
+        fn: Callable,
+        log_level: int = logging.INFO,
+        logger_name: str | None = None,
+        **kwargs,
+    ) -> Generator[List[Log], None, None]:
+        """Run a function in a thread and capture logs in real-time to yield them."""
+        yield self.log(
+            f"Running {fn.__name__}({', '.join(f'{k}={v}' for k, v in kwargs.items())})"
+        )
+
+        error_queue, wrapped_fn = non_failing_fn(fn)
+        self.thread = Thread(target=wrapped_fn, kwargs=kwargs)
+
+        def _log(record: logging.LogRecord) -> bool:
+            """Handle log record and return True if log should be yielded."""
+            if record.thread != self.thread.ident:
+                return False  # Skip if not from the thread
+            if logger_name and not record.name.startswith(logger_name):
+                return False  # Skip if not from the logger
+            if record.levelno < log_level:
+                return False  # Skip if too verbose
+            self.log(
+                message=record.getMessage(),
+                level=record.levelname,
+                timestamp=record.created,
+            )
+            return True
+
+        with capture_logging(log_level) as log_queue:
+            # Start thread and loop to capture and yield logs from the thread
+            self.thread.start()
+            while self.thread.is_alive():
+                while True:
+                    try:
+                        if _log(log_queue.get_nowait()):
+                            yield self.logs
+                    except queue.Empty:
+                        break
+                self.thread.join(timeout=0.1)  # adjust the timeout as needed
+
+        # After the thread completes, yield any remaining logs
+        while True:
+            try:
+                if _log(log_queue.get_nowait()):
+                    yield self.logs
+            except queue.Empty:
+                break
+
+        try:
+            error = error_queue.get_nowait()
+        except queue.Empty:
+            error = None
+        if error is not None:
+            msg = (
+                f"Error in '{fn.__name__}':"
+                + "\n"
+                + "\n".join(
+                    line.strip("\n")
+                    for line in traceback.format_tb(error.__traceback__)
+                    if line.strip()
+                )
+                + "\n\n"
+                + str(error)
+            )
+            yield self.log(msg, level="ERROR")
+            self.completed = True
+            self.error = True
+        else:
+            yield self.log("Thread completed successfully")
+            self.completed = True
+            self.error = False
+
+    def __repr__(self) -> str:
+        return f"<LogsViewRunner nb_logs={len(self.logs)} completed={self.completed} error={self.error}>"
+
+    def __del__(self):
+        if self.process and self.process.poll() is None:
+            print(f"Killing process: {self.process}")
+            self.process.kill()
+        if self.thread and self.thread.is_alive():
+            print(f"Joining thread: {self.thread}")
+            self.thread.join()
+
+
 class LogsView(Component):
     """
     Creates a component to visualize logs from a subprocess in real-time.
@@ -86,7 +228,7 @@ class LogsView(Component):
             value=value,
         )
 
-    def preprocess(self, payload: str | None) -> "LogsView":
+    def preprocess(self, payload: str | None) -> NoReturn:
         """
         Parameters:
             payload: string corresponding to the code
@@ -117,111 +259,6 @@ class LogsView(Component):
     def example_value(self) -> Any:
         return [Log("INFO", "Hello World", datetime.now().isoformat())]
 
-    @classmethod
-    def run_process(
-        cls, command: List[str], date_format: str = "%Y-%m-%d %H:%M:%S"
-    ) -> Generator[List[Log], None, None]:
-        """Run a command in a subprocess and yield logs in real-time."""
-        process = subprocess.Popen(
-            command,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.STDOUT,
-            text=True,
-        )
-
-        if process.stdout is None:
-            raise ValueError("stdout is None")
-
-        logs = []
-
-        def _log(level: str, message: str):
-            log = Log(
-                level=level,
-                message=message,
-                timestamp=datetime.now().strftime(date_format),
-            )
-            logs.append(log)
-            return logs
-
-        _log("INFO", f"Running {' '.join(command)}")
-        for line in process.stdout:
-            yield _log("INFO", line.strip())
-
-        # TODO: what if task is cancelled but process is still running?
-
-        process.stdout.close()
-        return_code = process.wait()
-        if return_code:
-            yield _log("ERROR", f"Process exited with code {return_code}")
-        else:
-            yield _log("INFO", "Process exited successfully")
-
-    @classmethod
-    def run_thread(
-        cls,
-        fn: Callable,
-        log_level: int = logging.INFO,
-        logger_name: str | None = None,
-        date_format: str = "%Y-%m-%d %H:%M:%S",
-        **kwargs,
-    ) -> Generator[List[Log], None, None]:
-        """Run a function in a thread and capture logs in real-time to yield them."""
-        logs = [
-            Log(
-                level="INFO",
-                message=f"Running {fn.__name__}({', '.join(f'{k}={v}' for k, v in kwargs.items())})",
-                timestamp=datetime.now().strftime(date_format),
-            )
-        ]
-        yield logs
-
-        thread = Thread(target=non_failing_fn(fn), kwargs=kwargs)
-
-        def _log(record: logging.LogRecord) -> bool:
-            """Handle log record and return True if log should be yielded."""
-            if record.thread != thread.ident:
-                return False  # Skip if not from the thread
-            if logger_name and not record.name.startswith(logger_name):
-                return False  # Skip if not from the logger
-            if record.levelno < log_level:
-                return False  # Skip if too verbose
-            log = Log(
-                level=record.levelname,
-                message=record.getMessage(),
-                timestamp=datetime.fromtimestamp(record.created).strftime(date_format),
-            )
-            logs.append(log)
-            return True
-
-        with capture_logging(log_level) as log_queue:
-            thread.start()
-
-            # Loop to capture and yield logs from the thread
-            while thread.is_alive():
-                while True:
-                    try:
-                        if _log(log_queue.get_nowait()):
-                            yield logs
-                    except queue.Empty:
-                        break
-                thread.join(timeout=0.1)  # adjust the timeout as needed
-
-        # After the thread completes, yield any remaining logs
-        while True:
-            try:
-                if _log(log_queue.get_nowait()):
-                    yield logs
-            except queue.Empty:
-                break
-
-        logs.append(
-            Log(
-                level="INFO",
-                message="Thread completed successfully",
-                timestamp=datetime.now().strftime(date_format),
-            )
-        )
-
 
 @contextmanager
 def capture_logging(log_level: int) -> Generator[Queue, None, None]:
@@ -240,11 +277,13 @@ def capture_logging(log_level: int) -> Generator[Queue, None, None]:
 
 
 def non_failing_fn(fn: Callable, *args, **kwargs) -> Callable:
+    error_queue = queue.Queue()
+
     @wraps(fn)
     def _inner(*args, **kwargs):
         try:
-            return fn(*args, **kwargs)
+            fn(*args, **kwargs)
         except Exception as e:
-            logging.error(f"Error in {fn.__name__}: {e}", stack_info=True)
+            error_queue.put(e)
 
-    return _inner
+    return error_queue, _inner

src/demo/app.py

@@ -3,7 +3,7 @@ import random
 import time
 
 import gradio as gr
-from gradio_logsview import LogsView
+from gradio_logsview import LogsView, LogsViewRunner
 
 
 def random_values(failing: bool = False):
@@ -26,19 +26,27 @@ def random_values(failing: bool = False):
 
 
 def fn_process_success():
-    yield from LogsView.run_process(["python", "-u", "demo/script.py"])
+    runner = LogsViewRunner()
+    yield from runner.run_process(["python", "-u", "demo/script.py"])
+    yield runner.log(f"Runner: {runner}")
 
 
 def fn_process_failing():
-    yield from LogsView.run_process(["python", "-u", "demo/script.py", "--failing"])
+    runner = LogsViewRunner()
+    yield from runner.run_process(["python", "-u", "demo/script.py", "--failing"])
+    yield runner.log(f"Runner: {runner}")
 
 
 def fn_thread_success():
-    yield from LogsView.run_thread(random_values, log_level=logging.INFO, failing=False)
+    runner = LogsViewRunner()
+    yield from runner.run_thread(random_values, log_level=logging.INFO, failing=False)
+    yield runner.log(f"Runner: {runner}")
 
 
 def fn_thread_failing():
-    yield from LogsView.run_thread(random_values, log_level=logging.INFO, failing=True)
+    runner = LogsViewRunner()
+    yield from runner.run_thread(random_values, log_level=logging.INFO, failing=True)
+    yield runner.log(f"Runner: {runner}")
 
 
 markdown_top = """
@@ -53,6 +61,18 @@ In the process example, logs are generated by a Python script but any command ca
 
 
 markdown_bottom = """
+## Installation
+
+```
+pip install https://huggingface.co/spaces/Wauplin/gradio_logsview/resolve/main/gradio_logsview-0.0.3-py3-none-any.whl
+```
+
+or add this line to your `requirements.txt`:
+
+```
+gradio_logsview@https://huggingface.co/spaces/Wauplin/gradio_logsview/resolve/main/gradio_logsview-0.0.3-py3-none-any.whl
+```
+
 ## How to run in a thread?
 
 With `LogsView.run_thread`, you can run a function in a separate thread and capture logs in real-time.
@@ -64,7 +84,9 @@ from gradio_logsview import LogsView
 def fn_thread():
     # Run `my_function` in a separate thread
     # All logs above `INFO` level will be captured and displayed in real-time.
-    yield from LogsView.run_thread(my_function, log_level=logging.INFO, arg1="value1")
+    runner = LogsViewRunner() # Initialize the runner
+    yield from runner.run_thread(my_function, log_level=logging.INFO, arg1="value1")
+    yield runner.log(f"Runner: {runner}") # Log any message
 
 with gr.Blocks() as demo:
     logs = LogsView()
@@ -81,15 +103,29 @@ from gradio_logsview import LogsView
 
 def fn_process():
     # Run a process and capture all logs from the process
-    yield from LogsView.run_process(
-        cmd=[mergekit-yaml", "config.yaml", "merge", "--copy-", "--cuda", "--low-cpu-memory"]
+    runner = LogsViewRunner() # Initialize the runner
+    yield from runner.run_process(
+        cmd=["mergekit-yaml", "config.yaml", "merge", "--copy-", "--cuda", "--low-cpu-memory"]
     )
+    yield runner.log(f"Runner: {runner}") # Log any message
 
 with gr.Blocks() as demo:
     logs = LogsView()
     btn = gr.Button("Run process")
     btn.click(fn_process, outputs=logs)
 ```
+
+## TODO
+
+- [ ] display logs with colors (front-end)
+- [ ] format logs client-side (front-end)
+- [ ] scrollable logs if more than N lines (front-end)
+- [ ] format each log only once (front-end)
+- [x] stop process if `run_process` gets cancelled (back-end)
+- [x] correctly pass error stacktrace in `run_thread` (back-end)
+- [ ] correctly catch print statements in `run_thread` (back-end)
+- [ ] disable interactivity + remove all code editing logic (both?)
+- [ ] how to handle progress bars? (i.e when logs are overwritten in terminal)
 """
 
 with gr.Blocks() as demo:

src/demo/space.py

@@ -1,7 +1,9 @@
 
+import os
+
 import gradio as gr
+
 from app import demo as app
-import os
 
 _docs = {'LogsView': {'description': 'Creates a component to visualize logs from a subprocess in real-time.', 'members': {'__init__': {'value': {'type': 'str | Callable | tuple[str] | None', 'default': 'None', 'description': 'Default value to show in the code editor. If callable, the function will be called whenever the app loads to set the initial value of the component.'}, 'every': {'type': 'float | None', 'default': 'None', 'description': "If `value` is a callable, run the function 'every' number of seconds while the client connection is open. Has no effect otherwise. The event can be accessed (e.g. to cancel it) via this component's .load_event attribute."}, 'lines': {'type': 'int', 'default': '5', 'description': None}, 'label': {'type': 'str | None', 'default': 'None', 'description': 'The label for this component. Appears above the component and is also used as the header if there are a table of examples for this component. If None and used in a `gr.Interface`, the label will be the name of the parameter this component is assigned to.'}, 'show_label': {'type': 'bool | None', 'default': 'None', 'description': 'if True, will display label.'}, 'container': {'type': 'bool', 'default': 'True', 'description': 'If True, will place the component in a container - providing some extra padding around the border.'}, 'scale': {'type': 'int | None', 'default': 'None', 'description': 'relative size compared to adjacent Components. For example if Components A and B are in a Row, and A has scale=2, and B has scale=1, A will be twice as wide as B. Should be an integer. scale applies in Rows, and to top-level Components in Blocks where fill_height=True.'}, 'min_width': {'type': 'int', 'default': '160', 'description': 'minimum pixel width, will wrap if not sufficient screen space to satisfy this value. If a certain scale value results in this Component being narrower than min_width, the min_width parameter will be respected first.'}, 'visible': {'type': 'bool', 'default': 'True', 'description': 'If False, component will be hidden.'}, 'elem_id': {'type': 'str | None', 'default': 'None', 'description': 'An optional string that is assigned as the id of this component in the HTML DOM. Can be used for targeting CSS styles.'}, 'elem_classes': {'type': 'list[str] | str | None', 'default': 'None', 'description': 'An optional list of strings that are assigned as the classes of this component in the HTML DOM. Can be used for targeting CSS styles.'}, 'render': {'type': 'bool', 'default': 'True', 'description': 'If False, component will not render be rendered in the Blocks context. Should be used if the intention is to assign event listeners now but render the component later.'}}, 'postprocess': {'value': {'type': 'list[Log]', 'description': 'Expects a list of `Log` logs.'}}, 'preprocess': {'return': {'type': 'LogsView', 'description': 'Passes the code entered as a `str`.'}, 'value': None}}, 'events': {'change': {'type': None, 'default': None, 'description': 'Triggered when the value of the LogsView changes either because of user input (e.g. a user types in a textbox) OR because of a function update (e.g. an image receives a value from the output of an event trigger). See `.input()` for a listener that is only triggered by user input.'}, 'input': {'type': None, 'default': None, 'description': 'This listener is triggered when the user changes the value of the LogsView.'}, 'focus': {'type': None, 'default': None, 'description': 'This listener is triggered when the LogsView is focused.'}, 'blur': {'type': None, 'default': None, 'description': 'This listener is triggered when the LogsView is unfocused/blurred.'}}}, '__meta__': {'additional_interfaces': {'Log': {'source': '@dataclass\nclass Log:\n    level: Literal[\n        "INFO", "DEBUG", "WARNING", "ERROR", "CRITICAL"\n    ]\n    message: str\n    timestamp: str'}, 'LogsView': {'source': 'class LogsView(Component):\n    EVENTS = [\n        Events.change,\n        Events.input,\n        Events.focus,\n        Events.blur,\n    ]\n\n    def __init__(\n        self,\n        value: str | Callable | tuple[str] | None = None,\n        *,\n        every: float | None = None,\n        lines: int = 5,\n        label: str | None = None,\n        show_label: bool | None = None,\n        container: bool = True,\n        scale: int | None = None,\n        min_width: int = 160,\n        visible: bool = True,\n        elem_id: str | None = None,\n        elem_classes: list[str] | str | None = None,\n        render: bool = True,\n    ):\n        self.language = "shell"\n        self.lines = lines\n        self.interactive = False\n        super().__init__(\n            label=label,\n            every=every,\n            show_label=show_label,\n            container=container,\n            scale=scale,\n            min_width=min_width,\n            visible=visible,\n            elem_id=elem_id,\n            elem_classes=elem_classes,\n            render=render,\n            value=value,\n        )\n\n    def preprocess(self, payload: str | None) -> "LogsView":\n        raise NotImplementedError(\n            "LogsView cannot be used as an input component."\n        )\n\n    def postprocess(self, value: List[Log]) -> List[Log]:\n        return value\n\n    def api_info(self) -> dict[str, Any]:\n        return {\n            "items": {\n                "level": "string",\n                "message": "string",\n                "timestamp": "number",\n            },\n            "title": "Logs",\n            "type": "array",\n        }\n\n    def example_payload(self) -> Any:\n        return [\n            Log(\n                "INFO",\n                "Hello World",\n                datetime.now().isoformat(),\n            )\n        ]\n\n    def example_value(self) -> Any:\n        return [\n            Log(\n                "INFO",\n                "Hello World",\n                datetime.now().isoformat(),\n            )\n        ]\n\n    @classmethod\n    def run_process(\n        cls,\n        command: List[str],\n        date_format: str = "%Y-%m-%d %H:%M:%S",\n    ) -> Generator[List[Log], None, None]:\n        process = subprocess.Popen(\n            command,\n            stdout=subprocess.PIPE,\n            stderr=subprocess.STDOUT,\n            text=True,\n        )\n\n        if process.stdout is None:\n            raise ValueError("stdout is None")\n\n        logs = []\n\n        def _log(level: str, message: str):\n            log = Log(\n                level=level,\n                message=message,\n                timestamp=datetime.now().strftime(\n                    date_format\n                ),\n            )\n            logs.append(log)\n            return logs\n\n        _log("INFO", f"Running {\' \'.join(command)}")\n        for line in process.stdout:\n            yield _log("INFO", line.strip())\n\n        # TODO: what if task is cancelled but process is still running?\n\n        process.stdout.close()\n        return_code = process.wait()\n        if return_code:\n            yield _log(\n                "ERROR",\n                f"Process exited with code {return_code}",\n            )\n        else:\n            yield _log(\n                "INFO", "Process exited successfully"\n            )\n\n    @classmethod\n    def run_thread(\n        cls,\n        fn: Callable,\n        log_level: int = logging.INFO,\n        logger_name: str | None = None,\n        date_format: str = "%Y-%m-%d %H:%M:%S",\n        **kwargs,\n    ) -> Generator[List[Log], None, None]:\n        logs = [\n            Log(\n                level="INFO",\n                message=f"Running {fn.__name__}({\', \'.join(f\'{k}={v}\' for k, v in kwargs.items())})",\n                timestamp=datetime.now().strftime(\n                    date_format\n                ),\n            )\n        ]\n        yield logs\n\n        thread = Thread(\n            target=non_failing_fn(fn), kwargs=kwargs\n        )\n\n        def _log(record: logging.LogRecord) -> bool:\n            if record.thread != thread.ident:\n                return False  # Skip if not from the thread\n            if logger_name and not record.name.startswith(\n                logger_name\n            ):\n                return False  # Skip if not from the logger\n            if record.levelno < log_level:\n                return False  # Skip if too verbose\n            log = Log(\n                level=record.levelname,\n                message=record.getMessage(),\n                timestamp=datetime.fromtimestamp(\n                    record.created\n                ).strftime(date_format),\n            )\n            logs.append(log)\n            return True\n\n        with capture_logging(log_level) as log_queue:\n            thread.start()\n\n            # Loop to capture and yield logs from the thread\n            while thread.is_alive():\n                while True:\n                    try:\n                        if _log(log_queue.get_nowait()):\n                            yield logs\n                    except queue.Empty:\n                        break\n                thread.join(\n                    timeout=0.1\n                )  # adjust the timeout as needed\n\n        # After the thread completes, yield any remaining logs\n        while True:\n            try:\n                if _log(log_queue.get_nowait()):\n                    yield logs\n            except queue.Empty:\n                break\n\n        logs.append(\n            Log(\n                level="INFO",\n                message="Thread completed successfully",\n                timestamp=datetime.now().strftime(\n                    date_format\n                ),\n            )\n        )'}}, 'user_fn_refs': {'LogsView': ['Log', 'LogsView']}}}
 
@@ -122,7 +124,7 @@ from gradio_logsview import LogsView
 def fn_process():
     # Run a process and capture all logs from the process
     yield from LogsView.run_process(
-        cmd=[mergekit-yaml", "config.yaml", "merge", "--copy-", "--cuda", "--low-cpu-memory"]
+        cmd=["mergekit-yaml", "config.yaml", "merge", "--copy-", "--cuda", "--low-cpu-memory"]
     )
 
 with gr.Blocks() as demo:

src/pyproject.toml

@@ -8,7 +8,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "gradio_logsview"
-version = "0.0.1"
+version = "0.0.3"
 description = "Visualize logs in your Gradio app"
 readme = "README.md"
 license = "MIT"