Skip to content

Instantly share code, notes, and snippets.

@Jackiexiao
Last active February 8, 2024 09:29
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save Jackiexiao/57aa4abbff675a3ac6a57e55a9a0fde7 to your computer and use it in GitHub Desktop.
Save Jackiexiao/57aa4abbff675a3ac6a57e55a9a0fde7 to your computer and use it in GitHub Desktop.
Streamlit subprocess streaming output to the web page / redirect std out
"""
Run subprocess command and stream output to the web page
streamlit run web_ui.py --server.port 10006 --browser.gatherUsageStats False
streamlit >=1.30
# print_numbers.py
import time
for i in range(1, 200):
print(i)
time.sleep(0.01)
"""
import streamlit as st
import subprocess
from typing import List
def run_command_and_stream_output(
command: List[str],
expander_name: str = "Log",
expanded: bool = False,
height: int = 200,
max_log_lines: int = 100,
) -> None:
"""run subprocess command and stream output to the web page
Attention:
- for python script, you need to add `-u` argument force the stdout and stderr streams to be unbuffered
- for example: `command = ["python", '-u', 'print_numbers.py']`
Args:
command (List[str]): command to run
expander_name (str): name of the expander
expanded (bool): whether the expander is expanded
height (int): height of the expander
max_log_lines (int): maximum number of lines to display in the log
Raises:
Exception: if the command returns non-zero exit status
Examples:
>>> run_command_and_stream_output(["python", "-u", "print_numbers.py"])
"""
process = subprocess.Popen(
command,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
universal_newlines=True,
)
log = []
with st.expander(expander_name, expanded=expanded):
with st.container(height=height):
log_placeholder = st.empty()
while process.poll() is None:
line = process.stdout.readline()
if not line:
continue
log.append(line.strip())
if len(log) > max_log_lines:
log.pop(0)
log_placeholder.code("\n".join(log), line_numbers=True)
if process.returncode != 0:
st.error(f"Error: {process.returncode}")
raise Exception(
f"Command '{' '.join(command)}' returned non-zero exit status {process.returncode}"
)
if __name__ == "__main__":
start_button = st.button("Run External Script")
if start_button:
command = ["python", "-u", "print_numbers.py"]
run_command_and_stream_output(command)
st.success("Script has been executed successfully")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment