space.py

import logging
import random
import time

import gradio as gr

from gradio_logsview import LogsView, LogsViewRunner


def random_values(failing: bool = False):
    for i in range(10):
        logging.log(
            random.choice(
                [  # Random levels
                    logging.INFO,
                    logging.DEBUG,
                    logging.WARNING,
                    logging.ERROR,
                    logging.CRITICAL,
                ]
            ),
            f"Value {i+1}",  # Random values
        )
        time.sleep(random.uniform(0, 1))
        if failing and i == 5:
            raise ValueError("Failing!!")


def fn_process_success():
    runner = LogsViewRunner()
    yield from runner.run_process(["python", "-u", "demo/script.py"])
    yield runner.log(f"Runner: {runner}")


def fn_process_failing():
    runner = LogsViewRunner()
    yield from runner.run_process(["python", "-u", "demo/script.py", "--failing"])
    yield runner.log(f"Runner: {runner}")


def fn_thread_success():
    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():
    runner = LogsViewRunner()
    yield from runner.run_thread(random_values, log_level=logging.INFO, failing=True)
    yield runner.log(f"Runner: {runner}")


markdown_top = """
# LogsView Demo

This demo shows how to use the `LogsView` component to display logs from a process or a thread in real-time.

Click on any button to launch a process or a thread and see the logs displayed in real-time.
In the thread example, logs are generated randomly with different log levels.
In the process example, logs are generated by a Python script but any command can be executed.
"""


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.
You can configure which logs to capture (log level and logger name).

```py
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.
    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()
    btn = gr.Button("Run thread")
    btn.click(fn_thread, outputs=logs)
```

## How to run in a process?

With `LogsView.run_process`, you can run a command in a separate process and capture logs from the process in real-time.
    
```py
from gradio_logsview import LogsView

def fn_process():
    # Run a process and capture all logs from the process
    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:
    gr.Markdown(markdown_top)

    with gr.Row():
        btn_thread_success = gr.Button("Run thread (success)")
        btn_thread_failing = gr.Button("Run thread (failing)")
    with gr.Row():
        btn_process_success = gr.Button("Run process (success)")
        btn_process_failing = gr.Button("Run process (failing)")
    logs = LogsView()

    gr.Markdown(markdown_bottom)

    btn_thread_failing.click(fn_thread_failing, outputs=logs)
    btn_thread_success.click(fn_thread_success, outputs=logs)
    btn_process_failing.click(fn_process_failing, outputs=logs)
    btn_process_success.click(fn_process_success, outputs=logs)


if __name__ == "__main__":
    demo.launch()