Hi @bouzidanas,
Awesome module. Thanks a lot for keeping it up. I am currently working on implementing this component to run code on streamlit using the subprocess module and so far it works really well. However, I wonder if there is a way to debounce automatically after each run without having to modify the code already written in the editor? I find that it affects reruns caused by buttons that modify session_state variables, however, once you debounce by modifying the code written in the editor the problem goes away
if "PYTHON_INTERPRETER_HISTORY" not in st.session_state:
st.session_state.PYTHON_INTERPRETER_HISTORY = {}
if "selected_output" not in st.session_state:
st.session_state.selected_output = None
if "selected_output_button" not in st.session_state:
st.session_state.selected_output_button = False
def update_chat_history(role, input, output):
index = len(st.session_state.PYTHON_INTERPRETER_HISTORY)
st.session_state.PYTHON_INTERPRETER_HISTORY[index] = {
"code_block": role,
"items": [{"type": "code_input", "content": input},
{"type": "code_output", "content": output}]
}
st.session_state.selected_output = index
def create_output_buttons():
for index, item in st.session_state.PYTHON_INTERPRETER_HISTORY.items():
if st.button(f"Output {index + 1}", key=f"button_{index}"):
st.session_state.selected_output = index
st.session_state.selected_output_button = True
# st.rerun()
def output_container(index, message):
with st.container(border=True):
st.markdown(f"**Output {index + 1}:**")
output_placeholder = st.empty()
with output_placeholder.container():
for item in message["items"]:
if item["type"] == "code_input":
st.code(item["content"], language="python")
elif item["type"] == "code_output":
if 'matplotlib_figure' in item["content"]:
fig_data = base64.b64decode(item['content']['matplotlib_figure'])
fig = pickle.loads(fig_data)
st.pyplot(fig)
if 'plotly_figure' in item["content"]:
fig_data = item['content']['plotly_figure']
if isinstance(fig_data, str):
fig_json = json.loads(fig_data)
else:
fig_json = fig_data
fig = go.Figure(fig_json)
st.plotly_chart(fig, use_container_width=True)
if "user_prints" in item["content"]:
st.write(item["content"]["user_prints"])
def display_selected_output():
logger.info(f"st.session_state.PYTHON_INTERPRETER_HISTORY: {st.session_state.PYTHON_INTERPRETER_HISTORY}")
logger.info(f"st.session_state.selected_output: {st.session_state.selected_output}")
if st.session_state.selected_output is not None and st.session_state.selected_output in st.session_state.PYTHON_INTERPRETER_HISTORY:
selected_index = st.session_state.selected_output
selected_message = st.session_state.PYTHON_INTERPRETER_HISTORY[selected_index]
output_container(selected_index, selected_message)
def display_chat_history():
logger.info(f"st.session_state.PYTHON_INTERPRETER_HISTORY: {st.session_state.PYTHON_INTERPRETER_HISTORY}")
logger.info(f"st.session_state.selected_output: {st.session_state.selected_output}")
if st.session_state.PYTHON_INTERPRETER_HISTORY:
latest_index = max(st.session_state.PYTHON_INTERPRETER_HISTORY.keys())
message = st.session_state.PYTHON_INTERPRETER_HISTORY[latest_index]
output_container(latest_index, message)
else:
st.write("No output history yet.")
def execute_in_subprocess(code):
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as tmp:
tmp.write(textwrap.dedent(code))
tmp.flush()
try:
python_code = textwrap.dedent('''
import json
import sys
import matplotlib.pyplot as plt
import plotly.graph_objects as go
import io
import base64
import pickle
def serialize(obj):
if isinstance(obj, (int, float, str, bool, type(None))):
return obj
return str(obj)
# Redirect stdout to capture print statements
import contextlib
user_output = io.StringIO()
with contextlib.redirect_stdout(user_output):
try:
with open("{0}", "r") as file:
exec(file.read())
output = {{k: serialize(v) for k, v in locals().items() if not k.startswith('_') and k != 'user_output'}}
# Capture matplotlib figure if it exists
mpl_fig = plt.gcf()
if mpl_fig.get_axes():
buf = io.BytesIO()
pickle.dump(mpl_fig, buf)
buf.seek(0)
output['matplotlib_figure'] = base64.b64encode(buf.getvalue()).decode('utf-8')
plt.close(mpl_fig)
# Capture Plotly figure if it exists
if 'fig' in locals() and isinstance(fig, go.Figure):
output['plotly_figure'] = json.loads(fig.to_json())
output['user_prints'] = user_output.getvalue()
except Exception as e:
print(json.dumps({{"error": str(e)}}), file=sys.stderr)
sys.exit(1)
print(json.dumps({{"output": output}}))
'''.format(tmp.name))
with st.expander("Subprocess Code", expanded=False):
st.code(python_code, language="python")
result = subprocess.run(
['python', '-c', python_code],
capture_output=True,
text=True,
check=True,
timeout=5
)
if result.returncode == 0:
try:
output = json.loads(result.stdout.strip())
return {
"status": "success",
"output": output["output"]
}
except json.JSONDecodeError:
return {
"status": "error",
"error": f"Invalid JSON output. Raw output: {result.stdout}"
}
else:
return {
"status": "error",
"error": f"Subprocess error: {result.stderr.strip()}"
}
except subprocess.TimeoutExpired:
return {'error': "Execution timed out (5 seconds)"}
finally:
os.unlink(tmp.name)
col1, col2 = st.columns(2)
with col1:
# @st.experimental_fragment
def streamlit_code_editor():
response_dict = code_editor(code="",
height=[19, 22],
lang="python",
theme="default",
shortcuts="vscode",
focus=True,
buttons=[{
"name": "Run",
"feather": "Play",
"primary": True,
"hasText": True,
"showWithIcon": True,
"commands": ["submit"],
"style": {"bottom": "0.44rem", "right": "0.4rem"}
}],
options={
"wrap": True,
"showLineNumbers": True
},
ghost_text="Type your code here",
response_mode=["blur", "debounce"],
allow_reset=True,
key="code_editor"
)
# Process code execution only when submitted
if response_dict['type'] == "submit":
code_text = response_dict['text']
result = execute_in_subprocess(code_text)
if result["status"] == "error":
update_chat_history("interpreter", code_text, result["error"])
else:
output = result["output"]
update_chat_history("interpreter", code_text, output)
st.session_state.selected_output_button = False
# st.rerun()
streamlit_code_editor()
# Move create_output_buttons outside of streamlit_code_editor
create_output_buttons()
with col2:
with st.container(border=False):
if not st.session_state.selected_output_button:
display_chat_history()
else:
display_selected_output()
logger.info(f"st.session_state.PYTHON_INTERPRETER_HISTORY: {st.session_state.PYTHON_INTERPRETER_HISTORY}")
logger.info(f"st.session_state.selected_output: {st.session_state.selected_output}")