---
tags: [gradio-custom-component, Code]
title: gradio_livelog
short_description: A Live Log component for Gradio Interface
colorFrom: blue
colorTo: yellow
sdk: gradio
pinned: false
app_file: space.py
---
# `gradio_livelog`
<a href="https://pypi.org/project/gradio_livelog/" target="_blank"><img alt="PyPI - Version" src="https://img.shields.io/pypi/v/gradio_livelog"></a>
A Live Log Component for Gradio Interface
## Installation
```bash
pip install gradio_livelog
```
## Usage
```python
# demo/app.py
import gradio as gr
import torch
import time
import logging
import random
import numpy as np
from diffusers import StableDiffusionXLPipeline, EulerAncestralDiscreteScheduler
import threading
import queue
import asyncio
import spaces
# Import the component and ALL its utilities
from gradio_livelog import LiveLog
from gradio_livelog.utils import ProgressTracker, capture_logs
# --- 1. SETUP ---
MODEL_ID = "SG161222/RealVisXL_V5.0_Lightning"
MAX_SEED = np.iinfo(np.int32).max
# --- 2. LOGIC FOR THE "LIVELOG FEATURE DEMO" TAB ---
logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')
async def run_process(disable_console: bool, run_error_case: bool):
with capture_logs(disable_console=disable_console) as get_logs:
total_steps = 100
tracker = ProgressTracker(total=total_steps, description="Simulating a process...")
all_logs = []
last_log_content = None
initial_log = f"Starting simulated process with {total_steps} steps..."
logging.info(initial_log)
logs = [
{
"type": "log",
"level": "SUCCESS" if record.levelno == logging.INFO + 5 else record.levelname,
"content": record.getMessage()
}
for record in get_logs()
]
all_logs.extend(logs)
last_log_content = logs[-1]["content"] if logs else None
yield tracker.update(advance=0, status="running", logs=all_logs, log_content=None)
for i in range(total_steps):
await asyncio.sleep(0.03)
current_step = i + 1
if current_step == 10:
logging.warning(f"Low disk space warning at step {current_step}.")
elif current_step == 30:
logging.log(logging.INFO + 5, f"Asset pack loaded successfully at step {current_step}.")
elif current_step == 75:
logging.critical(f"Checksum mismatch! Data may be corrupt at step {current_step}.")
if run_error_case and current_step == 50:
logging.error("A fatal simulation error occurred! Aborting.")
logs = [
{
"type": "log",
"level": "SUCCESS" if record.levelno == logging.INFO + 5 else record.levelname,
"content": record.getMessage()
}
for record in get_logs()
]
all_logs.extend(logs)
last_log_content = logs[-1]["content"] if logs else last_log_content
yield tracker.update(advance=0, status="error", logs=all_logs, log_content=last_log_content)
return
logs = [
{
"type": "log",
"level": "SUCCESS" if record.levelno == logging.INFO + 5 else record.levelname,
"content": record.getMessage()
}
for record in get_logs()
]
all_logs.extend(logs)
if logs:
last_log_content = logs[-1]["content"]
yield tracker.update(advance=1, status="running", logs=all_logs, log_content=last_log_content)
final_log = "Process completed successfully!"
logging.log(logging.INFO + 5, final_log)
logs = [
{
"type": "log",
"level": "SUCCESS" if record.levelno == logging.INFO + 5 else record.levelname,
"content": record.getMessage()
}
for record in get_logs()
]
all_logs.extend(logs)
last_log_content = logs[-1]["content"] if logs else last_log_content
yield tracker.update(advance=0, status="success", logs=all_logs, log_content=last_log_content)
def update_livelog_properties(mode, color, lines, scroll):
return gr.update(display_mode=mode, background_color=color, line_numbers=lines, autoscroll=scroll)
def clear_output():
return None
async def run_success_case(disable_console: bool):
yield None
async for update in run_process(disable_console=disable_console, run_error_case=False):
yield update
async def run_error_case(disable_console: bool):
yield None
async for update in run_process(disable_console=disable_console, run_error_case=True):
yield update
# --- 3. LOGIC FOR THE "DIFFUSION PIPELINE INTEGRATION" TAB ---
diffusion_pipeline = None
def load_pipeline(on_load=True):
"""A function to load the model, ensuring it's only done once."""
global diffusion_pipeline
if diffusion_pipeline is None:
print("Loading Stable Diffusion model for the first time...")
pipe = StableDiffusionXLPipeline.from_pretrained(
MODEL_ID, torch_dtype=torch.float16, use_safetensors=True, add_watermarker=False
)
pipe.enable_vae_tiling()
pipe.enable_model_cpu_offload()
pipe.scheduler = EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config)
pipe.set_progress_bar_config(disable=True)
diffusion_pipeline = pipe
print("Model loaded successfully.")
if not on_load:
return diffusion_pipeline
@spaces.GPU(duration=60, enable_queue=True)
def run_diffusion_in_thread(prompt: str, update_queue: queue.Queue):
"""
This function now uses capture_logs to listen to internal diffusers logs
while retaining the superior data structure you designed.
"""
tracker = None
with capture_logs() as get_logs:
try:
pipe = load_pipeline(on_load=False)
seed = random.randint(0, MAX_SEED)
generator = torch.Generator(device="cuda").manual_seed(seed)
prompt_style = f"hyper-realistic 8K image of {prompt}. ultra-detailed, lifelike, high-resolution, sharp, vibrant colors, photorealistic"
negative_prompt_style = "cartoonish, low resolution, blurry, simplistic, abstract, deformed, ugly"
num_inference_steps = 10
all_logs = []
last_log_content = None
# Helper function to process and store new logs
def process_and_store_logs():
nonlocal all_logs, last_log_content
new_records = get_logs()
if new_records:
new_logs = [{"type": "log", "level": r.levelname, "content": r.getMessage()} for r in new_records]
all_logs.extend(new_logs)
last_log_content = all_logs[-1]["content"]
logging.info(f"Using seed: {seed}")
process_and_store_logs()
update_queue.put((None, {"type": "progress", "logs": all_logs, "current": 0, "total": num_inference_steps, "desc": "Diffusion Steps"}))
logging.info("Starting diffusion process...")
process_and_store_logs()
update_queue.put((None, {"type": "progress", "logs": all_logs, "current": 0, "total": num_inference_steps, "desc": "Diffusion Steps"}))
tracker = ProgressTracker(total=num_inference_steps, description="Diffusion Steps")
def progress_callback(pipe_instance, step, timestep, callback_kwargs):
process_and_store_logs() # Check for new logs from diffusers at each step
update_dict = tracker.update(logs=all_logs)
update_queue.put((None, update_dict))
return callback_kwargs
images = pipe(
prompt=prompt_style, negative_prompt=negative_prompt_style, width=1024, height=1024,
guidance_scale=3.0, num_inference_steps=num_inference_steps,
generator=generator, callback_on_step_end=progress_callback
).images
logging.log(logging.INFO + 5, "Image generated successfully!")
process_and_store_logs()
final_update = tracker.update(advance=0, status="success", logs=all_logs, log_content=last_log_content)
update_queue.put((images, final_update))
except Exception as e:
logging.error(f"Error in diffusion thread: {e}", exc_info=True)
process_and_store_logs() # Capture the final error log
if tracker:
error_update = tracker.update(advance=0, status="error", logs=all_logs, log_content=f"An error occurred: {e}")
update_queue.put((None, error_update))
finally:
update_queue.put(None)
@spaces.GPU(duration=60, enable_queue=True)
def generate(prompt):
"""This function starts the worker thread and yields updates from the queue."""
yield None, None
yield None, {"type": "log", "level": "INFO", "content": "Preparing generation..."}
update_queue = queue.Queue()
diffusion_thread = threading.Thread(target=run_diffusion_in_thread, args=(prompt, update_queue))
diffusion_thread.start()
while True:
update = update_queue.get()
if update is None: break
yield update
# --- 4. THE COMBINED GRADIO UI with TABS ---
with gr.Blocks(theme=gr.themes.Ocean()) as demo:
gr.HTML("<h1><center>LiveLog Component Showcase</center></h1>")
with gr.Tabs():
with gr.TabItem("LiveLog Feature Demo"):
gr.Markdown("### Test all features of the LiveLog component interactively.")
with gr.Row():
with gr.Column(scale=3):
feature_logger = LiveLog(
label="Process Output", line_numbers=True, height=550,
background_color="#000000", display_mode="full"
)
with gr.Column(scale=1):
with gr.Group():
gr.Markdown("### Component Properties")
display_mode_radio = gr.Radio(["full", "log", "progress"], label="Display Mode", value="full")
bg_color_picker = gr.ColorPicker(label="Background Color", value="#000000")
line_numbers_checkbox = gr.Checkbox(label="Show Line Numbers", value=True)
autoscroll_checkbox = gr.Checkbox(label="Autoscroll", value=True)
with gr.Group():
gr.Markdown("### Simulation Controls")
disable_console_checkbox = gr.Checkbox(label="Disable Python Console Output", value=False)
start_btn = gr.Button("Run Success Case", variant="primary")
error_btn = gr.Button("Run Error Case")
start_btn.click(fn=run_success_case, inputs=[disable_console_checkbox], outputs=feature_logger)
error_btn.click(fn=run_error_case, inputs=[disable_console_checkbox], outputs=feature_logger)
feature_logger.clear(fn=clear_output, inputs=None, outputs=feature_logger)
controls = [display_mode_radio, bg_color_picker, line_numbers_checkbox, autoscroll_checkbox]
for control in controls:
control.change(fn=update_livelog_properties, inputs=controls, outputs=feature_logger)
with gr.TabItem("Diffusion Pipeline Integration"):
gr.Markdown("### Use `LiveLog` to monitor a real image generation process.")
with gr.Row():
with gr.Column(scale=3):
with gr.Group():
prompt = gr.Textbox(
label="Enter your prompt", show_label=False,
placeholder="A cinematic photo of a robot in a floral garden...",
scale=8, container=False
)
run_button = gr.Button("Generate", scale=1, variant="primary")
livelog_viewer = LiveLog(
label="Process Monitor", height=250, display_mode="full", line_numbers=False
)
with gr.Column(scale=2):
result_gallery = gr.Gallery(
label="Result", columns=1, show_label=False, height=500, min_width=768, preview=True, allow_preview=True
)
run_button.click(fn=generate, inputs=[prompt], outputs=[result_gallery, livelog_viewer])
prompt.submit(fn=generate, inputs=[prompt], outputs=[result_gallery, livelog_viewer])
livelog_viewer.clear(fn=clear_output, inputs=None, outputs=[livelog_viewer])
# This ensures the model is downloaded/loaded once when the app starts.
demo.load(load_pipeline, None, None)
if __name__ == "__main__":
demo.queue(max_size=50).launch(debug=True)
```
## `LiveLog`
### Initialization
<table>
<thead>
<tr>
<th align="left">name</th>
<th align="left" style="width: 25%;">type</th>
<th align="left">default</th>
<th align="left">description</th>
</tr>
</thead>
<tbody>
<tr>
<td align="left"><code>value</code></td>
<td align="left" style="width: 25%;">
```python
typing.Union[
typing.List[typing.Dict[str, typing.Any]],
typing.Callable,
NoneType,
][
typing.List[typing.Dict[str, typing.Any]][
typing.Dict[str, typing.Any][str, Any]
],
Callable,
None,
]
```
</td>
<td align="left"><code>None</code></td>
<td align="left">The initial value, a list of log/progress dictionaries. Can be a callable.</td>
</tr>
<tr>
<td align="left"><code>label</code></td>
<td align="left" style="width: 25%;">
```python
str | None
```
</td>
<td align="left"><code>None</code></td>
<td align="left">The component label.</td>
</tr>
<tr>
<td align="left"><code>every</code></td>
<td align="left" style="width: 25%;">
```python
float | None
```
</td>
<td align="left"><code>None</code></td>
<td align="left">If `value` is a callable, run the function 'every' seconds.</td>
</tr>
<tr>
<td align="left"><code>height</code></td>
<td align="left" style="width: 25%;">
```python
int | str
```
</td>
<td align="left"><code>400</code></td>
<td align="left">The height of the log panel in pixels or CSS units.</td>
</tr>
<tr>
<td align="left"><code>autoscroll</code></td>
<td align="left" style="width: 25%;">
```python
bool
```
</td>
<td align="left"><code>True</code></td>
<td align="left">If True, the panel will automatically scroll to the bottom on new logs.</td>
</tr>
<tr>
<td align="left"><code>line_numbers</code></td>
<td align="left" style="width: 25%;">
```python
bool
```
</td>
<td align="left"><code>False</code></td>
<td align="left">If True, shows line numbers for logs.</td>
</tr>
<tr>
<td align="left"><code>background_color</code></td>
<td align="left" style="width: 25%;">
```python
str
```
</td>
<td align="left"><code>"#000000"</code></td>
<td align="left">The background color of the log panel as a CSS-valid string.</td>
</tr>
<tr>
<td align="left"><code>display_mode</code></td>
<td align="left" style="width: 25%;">
```python
"full" | "log" | "progress"
```
</td>
<td align="left"><code>"full"</code></td>
<td align="left">"full" (logs and progress), "log" (only logs), or "progress" (only progress bar).</td>
</tr>
<tr>
<td align="left"><code>disable_console</code></td>
<td align="left" style="width: 25%;">
```python
bool
```
</td>
<td align="left"><code>True</code></td>
<td align="left">If True, logs will not be propagated to the standard Python console.</td>
</tr>
<tr>
<td align="left"><code>show_download_button</code></td>
<td align="left" style="width: 25%;">
```python
bool
```
</td>
<td align="left"><code>True</code></td>
<td align="left">If True, shows the download button in the header.</td>
</tr>
<tr>
<td align="left"><code>show_copy_button</code></td>
<td align="left" style="width: 25%;">
```python
bool
```
</td>
<td align="left"><code>True</code></td>
<td align="left">If True, shows the copy button in the header.</td>
</tr>
<tr>
<td align="left"><code>show_clear_button</code></td>
<td align="left" style="width: 25%;">
```python
bool
```
</td>
<td align="left"><code>True</code></td>
<td align="left">If True, shows the clear button in the header.</td>
</tr>
<tr>
<td align="left"><code>show_label</code></td>
<td align="left" style="width: 25%;">
```python
bool
```
</td>
<td align="left"><code>True</code></td>
<td align="left">If True, will display label.</td>
</tr>
<tr>
<td align="left"><code>container</code></td>
<td align="left" style="width: 25%;">
```python
bool
```
</td>
<td align="left"><code>True</code></td>
<td align="left">If True, will place the component in a container.</td>
</tr>
<tr>
<td align="left"><code>scale</code></td>
<td align="left" style="width: 25%;">
```python
int | None
```
</td>
<td align="left"><code>None</code></td>
<td align="left">Relative size compared to adjacent Components.</td>
</tr>
<tr>
<td align="left"><code>min_width</code></td>
<td align="left" style="width: 25%;">
```python
int
```
</td>
<td align="left"><code>160</code></td>
<td align="left">Minimum pixel width, will wrap if not sufficient screen space.</td>
</tr>
<tr>
<td align="left"><code>visible</code></td>
<td align="left" style="width: 25%;">
```python
bool
```
</td>
<td align="left"><code>True</code></td>
<td align="left">If False, the component will be hidden.</td>
</tr>
<tr>
<td align="left"><code>elem_id</code></td>
<td align="left" style="width: 25%;">
```python
str | None
```
</td>
<td align="left"><code>None</code></td>
<td align="left">An optional string that is assigned as the id of this component in the HTML DOM.</td>
</tr>
<tr>
<td align="left"><code>elem_classes</code></td>
<td align="left" style="width: 25%;">
```python
list[str] | str | None
```
</td>
<td align="left"><code>None</code></td>
<td align="left">An optional string or list of strings assigned as the class of this component.</td>
</tr>
<tr>
<td align="left"><code>render</code></td>
<td align="left" style="width: 25%;">
```python
bool
```
</td>
<td align="left"><code>True</code></td>
<td align="left">If False, this component will not be rendered.</td>
</tr>
<tr>
<td align="left"><code>key</code></td>
<td align="left" style="width: 25%;">
```python
int | str | tuple[int | str, Ellipsis] | None
```
</td>
<td align="left"><code>None</code></td>
<td align="left">A unique key for the component.</td>
</tr>
</tbody></table>
### Events
| name | description |
|:-----|:------------|
| `change` | Triggered when the value of the LiveLog 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. |
| `clear` | This listener is triggered when the user clears the LiveLog using the clear button for the component. |
### User function
The impact on the users predict function varies depending on whether the component is used as an input or output for an event (or both).
- When used as an Input, the component only impacts the input signature of the user function.
- When used as an output, the component only impacts the return signature of the user function.
The code snippet below is accurate in cases where the component is used as both an input and an output.
- **As output:** Is passed, the preprocessed input data sent to the user's function in the backend.
- **As input:** Should return, the output data received by the component from the user's function in the backend.
```python
def predict(
value: typing.Optional[typing.List[typing.Dict[str, typing.Any]]][
typing.List[typing.Dict[str, typing.Any]][
typing.Dict[str, typing.Any][str, Any]
],
None,
]
) -> typing.Optional[typing.List[typing.Dict[str, typing.Any]]][
typing.List[typing.Dict[str, typing.Any]][
typing.Dict[str, typing.Any][str, Any]
],
None,
]:
return value
```
Raw data
{
"_id": null,
"home_page": null,
"name": "gradio-livelog",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": null,
"keywords": "gradio-custom-component, gradio-template-Code",
"author": null,
"author_email": "Eliseu Silva <elismasilva@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/f2/ab/3f5b5dd8715c3c41f18a15cb225b1eebf07c13315a3587b64abce5bcf7f3/gradio_livelog-0.0.2.tar.gz",
"platform": null,
"description": "---\ntags: [gradio-custom-component, Code]\ntitle: gradio_livelog\nshort_description: A Live Log component for Gradio Interface\ncolorFrom: blue\ncolorTo: yellow\nsdk: gradio\npinned: false\napp_file: space.py\n---\n\n# `gradio_livelog`\n<a href=\"https://pypi.org/project/gradio_livelog/\" target=\"_blank\"><img alt=\"PyPI - Version\" src=\"https://img.shields.io/pypi/v/gradio_livelog\"></a> \n\nA Live Log Component for Gradio Interface\n\n## Installation\n\n```bash\npip install gradio_livelog\n```\n\n## Usage\n\n```python\n# demo/app.py\n\nimport gradio as gr\nimport torch\nimport time\nimport logging\nimport random\nimport numpy as np\nfrom diffusers import StableDiffusionXLPipeline, EulerAncestralDiscreteScheduler\nimport threading\nimport queue\nimport asyncio\nimport spaces\n\n# Import the component and ALL its utilities\nfrom gradio_livelog import LiveLog\nfrom gradio_livelog.utils import ProgressTracker, capture_logs\n\n# --- 1. SETUP ---\nMODEL_ID = \"SG161222/RealVisXL_V5.0_Lightning\"\nMAX_SEED = np.iinfo(np.int32).max\n\n# --- 2. LOGIC FOR THE \"LIVELOG FEATURE DEMO\" TAB ---\nlogging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')\n\nasync def run_process(disable_console: bool, run_error_case: bool):\n with capture_logs(disable_console=disable_console) as get_logs:\n total_steps = 100\n tracker = ProgressTracker(total=total_steps, description=\"Simulating a process...\")\n all_logs = []\n last_log_content = None\n \n initial_log = f\"Starting simulated process with {total_steps} steps...\"\n logging.info(initial_log)\n logs = [\n {\n \"type\": \"log\",\n \"level\": \"SUCCESS\" if record.levelno == logging.INFO + 5 else record.levelname,\n \"content\": record.getMessage()\n }\n for record in get_logs()\n ]\n all_logs.extend(logs)\n last_log_content = logs[-1][\"content\"] if logs else None \n yield tracker.update(advance=0, status=\"running\", logs=all_logs, log_content=None)\n\n for i in range(total_steps):\n await asyncio.sleep(0.03)\n current_step = i + 1\n \n if current_step == 10:\n logging.warning(f\"Low disk space warning at step {current_step}.\")\n elif current_step == 30:\n logging.log(logging.INFO + 5, f\"Asset pack loaded successfully at step {current_step}.\")\n elif current_step == 75:\n logging.critical(f\"Checksum mismatch! Data may be corrupt at step {current_step}.\")\n \n if run_error_case and current_step == 50:\n logging.error(\"A fatal simulation error occurred! Aborting.\")\n logs = [\n {\n \"type\": \"log\",\n \"level\": \"SUCCESS\" if record.levelno == logging.INFO + 5 else record.levelname,\n \"content\": record.getMessage()\n }\n for record in get_logs()\n ]\n all_logs.extend(logs)\n last_log_content = logs[-1][\"content\"] if logs else last_log_content\n yield tracker.update(advance=0, status=\"error\", logs=all_logs, log_content=last_log_content)\n return\n \n logs = [\n {\n \"type\": \"log\",\n \"level\": \"SUCCESS\" if record.levelno == logging.INFO + 5 else record.levelname,\n \"content\": record.getMessage()\n }\n for record in get_logs()\n ]\n all_logs.extend(logs)\n if logs:\n last_log_content = logs[-1][\"content\"]\n yield tracker.update(advance=1, status=\"running\", logs=all_logs, log_content=last_log_content)\n \n final_log = \"Process completed successfully!\"\n logging.log(logging.INFO + 5, final_log)\n logs = [\n {\n \"type\": \"log\",\n \"level\": \"SUCCESS\" if record.levelno == logging.INFO + 5 else record.levelname,\n \"content\": record.getMessage()\n }\n for record in get_logs()\n ]\n all_logs.extend(logs)\n last_log_content = logs[-1][\"content\"] if logs else last_log_content\n yield tracker.update(advance=0, status=\"success\", logs=all_logs, log_content=last_log_content)\n \ndef update_livelog_properties(mode, color, lines, scroll):\n return gr.update(display_mode=mode, background_color=color, line_numbers=lines, autoscroll=scroll)\n\ndef clear_output():\n return None\n\nasync def run_success_case(disable_console: bool):\n yield None \n async for update in run_process(disable_console=disable_console, run_error_case=False):\n yield update\n\nasync def run_error_case(disable_console: bool):\n yield None\n async for update in run_process(disable_console=disable_console, run_error_case=True):\n yield update\n\n\n# --- 3. LOGIC FOR THE \"DIFFUSION PIPELINE INTEGRATION\" TAB ---\ndiffusion_pipeline = None\ndef load_pipeline(on_load=True):\n \"\"\"A function to load the model, ensuring it's only done once.\"\"\"\n global diffusion_pipeline\n if diffusion_pipeline is None:\n print(\"Loading Stable Diffusion model for the first time...\")\n pipe = StableDiffusionXLPipeline.from_pretrained(\n MODEL_ID, torch_dtype=torch.float16, use_safetensors=True, add_watermarker=False\n )\n pipe.enable_vae_tiling()\n pipe.enable_model_cpu_offload()\n pipe.scheduler = EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config)\n pipe.set_progress_bar_config(disable=True)\n diffusion_pipeline = pipe\n print(\"Model loaded successfully.\")\n \n if not on_load:\n return diffusion_pipeline\n\n@spaces.GPU(duration=60, enable_queue=True)\ndef run_diffusion_in_thread(prompt: str, update_queue: queue.Queue):\n \"\"\"\n This function now uses capture_logs to listen to internal diffusers logs\n while retaining the superior data structure you designed.\n \"\"\"\n tracker = None \n with capture_logs() as get_logs:\n try:\n pipe = load_pipeline(on_load=False)\n seed = random.randint(0, MAX_SEED)\n generator = torch.Generator(device=\"cuda\").manual_seed(seed)\n prompt_style = f\"hyper-realistic 8K image of {prompt}. ultra-detailed, lifelike, high-resolution, sharp, vibrant colors, photorealistic\"\n negative_prompt_style = \"cartoonish, low resolution, blurry, simplistic, abstract, deformed, ugly\"\n num_inference_steps = 10\n \n all_logs = []\n last_log_content = None\n\n # Helper function to process and store new logs\n def process_and_store_logs():\n nonlocal all_logs, last_log_content\n new_records = get_logs()\n if new_records:\n new_logs = [{\"type\": \"log\", \"level\": r.levelname, \"content\": r.getMessage()} for r in new_records]\n all_logs.extend(new_logs)\n last_log_content = all_logs[-1][\"content\"]\n \n logging.info(f\"Using seed: {seed}\")\n process_and_store_logs()\n update_queue.put((None, {\"type\": \"progress\", \"logs\": all_logs, \"current\": 0, \"total\": num_inference_steps, \"desc\": \"Diffusion Steps\"}))\n \n logging.info(\"Starting diffusion process...\")\n process_and_store_logs()\n update_queue.put((None, {\"type\": \"progress\", \"logs\": all_logs, \"current\": 0, \"total\": num_inference_steps, \"desc\": \"Diffusion Steps\"}))\n \n tracker = ProgressTracker(total=num_inference_steps, description=\"Diffusion Steps\")\n \n def progress_callback(pipe_instance, step, timestep, callback_kwargs):\n process_and_store_logs() # Check for new logs from diffusers at each step\n update_dict = tracker.update(logs=all_logs)\n update_queue.put((None, update_dict))\n return callback_kwargs\n \n images = pipe(\n prompt=prompt_style, negative_prompt=negative_prompt_style, width=1024, height=1024,\n guidance_scale=3.0, num_inference_steps=num_inference_steps,\n generator=generator, callback_on_step_end=progress_callback\n ).images\n \n logging.log(logging.INFO + 5, \"Image generated successfully!\")\n process_and_store_logs()\n \n final_update = tracker.update(advance=0, status=\"success\", logs=all_logs, log_content=last_log_content)\n update_queue.put((images, final_update))\n\n except Exception as e:\n logging.error(f\"Error in diffusion thread: {e}\", exc_info=True)\n process_and_store_logs() # Capture the final error log\n if tracker:\n error_update = tracker.update(advance=0, status=\"error\", logs=all_logs, log_content=f\"An error occurred: {e}\")\n update_queue.put((None, error_update))\n finally:\n update_queue.put(None)\n \n@spaces.GPU(duration=60, enable_queue=True)\ndef generate(prompt):\n \"\"\"This function starts the worker thread and yields updates from the queue.\"\"\" \n yield None, None\n yield None, {\"type\": \"log\", \"level\": \"INFO\", \"content\": \"Preparing generation...\"}\n update_queue = queue.Queue()\n diffusion_thread = threading.Thread(target=run_diffusion_in_thread, args=(prompt, update_queue))\n diffusion_thread.start()\n while True:\n update = update_queue.get()\n if update is None: break\n yield update\n\n\n# --- 4. THE COMBINED GRADIO UI with TABS ---\nwith gr.Blocks(theme=gr.themes.Ocean()) as demo:\n gr.HTML(\"<h1><center>LiveLog Component Showcase</center></h1>\")\n\n with gr.Tabs():\n with gr.TabItem(\"LiveLog Feature Demo\"): \n gr.Markdown(\"### Test all features of the LiveLog component interactively.\")\n with gr.Row():\n with gr.Column(scale=3):\n feature_logger = LiveLog(\n label=\"Process Output\", line_numbers=True, height=550,\n background_color=\"#000000\", display_mode=\"full\"\n )\n with gr.Column(scale=1):\n with gr.Group():\n gr.Markdown(\"### Component Properties\")\n display_mode_radio = gr.Radio([\"full\", \"log\", \"progress\"], label=\"Display Mode\", value=\"full\")\n bg_color_picker = gr.ColorPicker(label=\"Background Color\", value=\"#000000\")\n line_numbers_checkbox = gr.Checkbox(label=\"Show Line Numbers\", value=True)\n autoscroll_checkbox = gr.Checkbox(label=\"Autoscroll\", value=True)\n with gr.Group():\n gr.Markdown(\"### Simulation Controls\")\n disable_console_checkbox = gr.Checkbox(label=\"Disable Python Console Output\", value=False)\n start_btn = gr.Button(\"Run Success Case\", variant=\"primary\")\n error_btn = gr.Button(\"Run Error Case\")\n \n start_btn.click(fn=run_success_case, inputs=[disable_console_checkbox], outputs=feature_logger)\n error_btn.click(fn=run_error_case, inputs=[disable_console_checkbox], outputs=feature_logger)\n feature_logger.clear(fn=clear_output, inputs=None, outputs=feature_logger)\n controls = [display_mode_radio, bg_color_picker, line_numbers_checkbox, autoscroll_checkbox]\n for control in controls:\n control.change(fn=update_livelog_properties, inputs=controls, outputs=feature_logger)\n \n with gr.TabItem(\"Diffusion Pipeline Integration\"): \n gr.Markdown(\"### Use `LiveLog` to monitor a real image generation process.\")\n with gr.Row():\n with gr.Column(scale=3):\n with gr.Group():\n prompt = gr.Textbox(\n label=\"Enter your prompt\", show_label=False,\n placeholder=\"A cinematic photo of a robot in a floral garden...\",\n scale=8, container=False\n )\n run_button = gr.Button(\"Generate\", scale=1, variant=\"primary\")\n \n livelog_viewer = LiveLog(\n label=\"Process Monitor\", height=250, display_mode=\"full\", line_numbers=False\n )\n \n with gr.Column(scale=2):\n result_gallery = gr.Gallery(\n label=\"Result\", columns=1, show_label=False, height=500, min_width=768, preview=True, allow_preview=True\n )\n \n run_button.click(fn=generate, inputs=[prompt], outputs=[result_gallery, livelog_viewer])\n prompt.submit(fn=generate, inputs=[prompt], outputs=[result_gallery, livelog_viewer])\n livelog_viewer.clear(fn=clear_output, inputs=None, outputs=[livelog_viewer])\n \n # This ensures the model is downloaded/loaded once when the app starts.\n demo.load(load_pipeline, None, None)\n \nif __name__ == \"__main__\":\n demo.queue(max_size=50).launch(debug=True)\n```\n\n## `LiveLog`\n\n### Initialization\n\n<table>\n<thead>\n<tr>\n<th align=\"left\">name</th>\n<th align=\"left\" style=\"width: 25%;\">type</th>\n<th align=\"left\">default</th>\n<th align=\"left\">description</th>\n</tr>\n</thead>\n<tbody>\n<tr>\n<td align=\"left\"><code>value</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\ntyping.Union[\n typing.List[typing.Dict[str, typing.Any]],\n typing.Callable,\n NoneType,\n][\n typing.List[typing.Dict[str, typing.Any]][\n typing.Dict[str, typing.Any][str, Any]\n ],\n Callable,\n None,\n]\n```\n\n</td>\n<td align=\"left\"><code>None</code></td>\n<td align=\"left\">The initial value, a list of log/progress dictionaries. Can be a callable.</td>\n</tr>\n\n<tr>\n<td align=\"left\"><code>label</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\nstr | None\n```\n\n</td>\n<td align=\"left\"><code>None</code></td>\n<td align=\"left\">The component label.</td>\n</tr>\n\n<tr>\n<td align=\"left\"><code>every</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\nfloat | None\n```\n\n</td>\n<td align=\"left\"><code>None</code></td>\n<td align=\"left\">If `value` is a callable, run the function 'every' seconds.</td>\n</tr>\n\n<tr>\n<td align=\"left\"><code>height</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\nint | str\n```\n\n</td>\n<td align=\"left\"><code>400</code></td>\n<td align=\"left\">The height of the log panel in pixels or CSS units.</td>\n</tr>\n\n<tr>\n<td align=\"left\"><code>autoscroll</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\nbool\n```\n\n</td>\n<td align=\"left\"><code>True</code></td>\n<td align=\"left\">If True, the panel will automatically scroll to the bottom on new logs.</td>\n</tr>\n\n<tr>\n<td align=\"left\"><code>line_numbers</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\nbool\n```\n\n</td>\n<td align=\"left\"><code>False</code></td>\n<td align=\"left\">If True, shows line numbers for logs.</td>\n</tr>\n\n<tr>\n<td align=\"left\"><code>background_color</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\nstr\n```\n\n</td>\n<td align=\"left\"><code>\"#000000\"</code></td>\n<td align=\"left\">The background color of the log panel as a CSS-valid string.</td>\n</tr>\n\n<tr>\n<td align=\"left\"><code>display_mode</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\n\"full\" | \"log\" | \"progress\"\n```\n\n</td>\n<td align=\"left\"><code>\"full\"</code></td>\n<td align=\"left\">\"full\" (logs and progress), \"log\" (only logs), or \"progress\" (only progress bar).</td>\n</tr>\n\n<tr>\n<td align=\"left\"><code>disable_console</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\nbool\n```\n\n</td>\n<td align=\"left\"><code>True</code></td>\n<td align=\"left\">If True, logs will not be propagated to the standard Python console.</td>\n</tr>\n\n<tr>\n<td align=\"left\"><code>show_download_button</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\nbool\n```\n\n</td>\n<td align=\"left\"><code>True</code></td>\n<td align=\"left\">If True, shows the download button in the header.</td>\n</tr>\n\n<tr>\n<td align=\"left\"><code>show_copy_button</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\nbool\n```\n\n</td>\n<td align=\"left\"><code>True</code></td>\n<td align=\"left\">If True, shows the copy button in the header.</td>\n</tr>\n\n<tr>\n<td align=\"left\"><code>show_clear_button</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\nbool\n```\n\n</td>\n<td align=\"left\"><code>True</code></td>\n<td align=\"left\">If True, shows the clear button in the header.</td>\n</tr>\n\n<tr>\n<td align=\"left\"><code>show_label</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\nbool\n```\n\n</td>\n<td align=\"left\"><code>True</code></td>\n<td align=\"left\">If True, will display label.</td>\n</tr>\n\n<tr>\n<td align=\"left\"><code>container</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\nbool\n```\n\n</td>\n<td align=\"left\"><code>True</code></td>\n<td align=\"left\">If True, will place the component in a container.</td>\n</tr>\n\n<tr>\n<td align=\"left\"><code>scale</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\nint | None\n```\n\n</td>\n<td align=\"left\"><code>None</code></td>\n<td align=\"left\">Relative size compared to adjacent Components.</td>\n</tr>\n\n<tr>\n<td align=\"left\"><code>min_width</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\nint\n```\n\n</td>\n<td align=\"left\"><code>160</code></td>\n<td align=\"left\">Minimum pixel width, will wrap if not sufficient screen space.</td>\n</tr>\n\n<tr>\n<td align=\"left\"><code>visible</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\nbool\n```\n\n</td>\n<td align=\"left\"><code>True</code></td>\n<td align=\"left\">If False, the component will be hidden.</td>\n</tr>\n\n<tr>\n<td align=\"left\"><code>elem_id</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\nstr | None\n```\n\n</td>\n<td align=\"left\"><code>None</code></td>\n<td align=\"left\">An optional string that is assigned as the id of this component in the HTML DOM.</td>\n</tr>\n\n<tr>\n<td align=\"left\"><code>elem_classes</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\nlist[str] | str | None\n```\n\n</td>\n<td align=\"left\"><code>None</code></td>\n<td align=\"left\">An optional string or list of strings assigned as the class of this component.</td>\n</tr>\n\n<tr>\n<td align=\"left\"><code>render</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\nbool\n```\n\n</td>\n<td align=\"left\"><code>True</code></td>\n<td align=\"left\">If False, this component will not be rendered.</td>\n</tr>\n\n<tr>\n<td align=\"left\"><code>key</code></td>\n<td align=\"left\" style=\"width: 25%;\">\n\n```python\nint | str | tuple[int | str, Ellipsis] | None\n```\n\n</td>\n<td align=\"left\"><code>None</code></td>\n<td align=\"left\">A unique key for the component.</td>\n</tr>\n</tbody></table>\n\n\n### Events\n\n| name | description |\n|:-----|:------------|\n| `change` | Triggered when the value of the LiveLog 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. |\n| `clear` | This listener is triggered when the user clears the LiveLog using the clear button for the component. |\n\n\n\n### User function\n\nThe impact on the users predict function varies depending on whether the component is used as an input or output for an event (or both).\n\n- When used as an Input, the component only impacts the input signature of the user function.\n- When used as an output, the component only impacts the return signature of the user function.\n\nThe code snippet below is accurate in cases where the component is used as both an input and an output.\n\n- **As output:** Is passed, the preprocessed input data sent to the user's function in the backend.\n- **As input:** Should return, the output data received by the component from the user's function in the backend.\n\n ```python\n def predict(\n value: typing.Optional[typing.List[typing.Dict[str, typing.Any]]][\n typing.List[typing.Dict[str, typing.Any]][\n typing.Dict[str, typing.Any][str, Any]\n ],\n None,\n]\n ) -> typing.Optional[typing.List[typing.Dict[str, typing.Any]]][\n typing.List[typing.Dict[str, typing.Any]][\n typing.Dict[str, typing.Any][str, Any]\n ],\n None,\n]:\n return value\n ```\n \n",
"bugtrack_url": null,
"license": null,
"summary": "A Live Log Component for Gradio Interface",
"version": "0.0.2",
"project_urls": null,
"split_keywords": [
"gradio-custom-component",
" gradio-template-code"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "337fd62f7261a50b778347a8d9fbb85e3e617a432d32fc2cd5fcf9132a2fdff2",
"md5": "f0e4f5554df37d6a6b30c3aaab526a23",
"sha256": "35f4bae2ae395452d79da2552e784fa4d4b121ac7506df81178735c00aa350cc"
},
"downloads": -1,
"filename": "gradio_livelog-0.0.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "f0e4f5554df37d6a6b30c3aaab526a23",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 100751,
"upload_time": "2025-08-01T03:06:23",
"upload_time_iso_8601": "2025-08-01T03:06:23.994904Z",
"url": "https://files.pythonhosted.org/packages/33/7f/d62f7261a50b778347a8d9fbb85e3e617a432d32fc2cd5fcf9132a2fdff2/gradio_livelog-0.0.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "f2ab3f5b5dd8715c3c41f18a15cb225b1eebf07c13315a3587b64abce5bcf7f3",
"md5": "11b57d152589e36c70fd6169e8a399bf",
"sha256": "33466f9e84fafa4ebb0f958081e6ea945712927d769f007667c42e8df12945e8"
},
"downloads": -1,
"filename": "gradio_livelog-0.0.2.tar.gz",
"has_sig": false,
"md5_digest": "11b57d152589e36c70fd6169e8a399bf",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 180742,
"upload_time": "2025-08-01T03:06:25",
"upload_time_iso_8601": "2025-08-01T03:06:25.200728Z",
"url": "https://files.pythonhosted.org/packages/f2/ab/3f5b5dd8715c3c41f18a15cb225b1eebf07c13315a3587b64abce5bcf7f3/gradio_livelog-0.0.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-01 03:06:25",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "gradio-livelog"
}
JSON