Dear all!
I have a detector which produces 2D images at video rates and higher. However, for live imaging rates in the range 10 Hz and more would be sufficient. Data comes as numpy arrays.
I face problems, when I want to display the data at the wanted rates.
The following code mocks up different ways to do this with streamlit. I am by no means a python - and especially no asyncio - expert, though.
- display, get the data, rerun
- get the data, display, rerun
- do displaying and data taking asynchronously
The third option does not need a st.experimental_rerun(). Part of the communication is done via streamlitās st.sessions_state which is not always mandatory, but keeps the examples as similar as possible.
Version 2 does not work well at all. I see only a grayed out image.
Versions 1&3 only work up to delays of about 0.25 secs, only
I am doing this out of my home office using a ssh tunnel to a server at work over 100 MBit/s connection.
I have no idea how to speed up this, but I guess that there is a way, as for example watching youtube videos about streamlit is not a problem at all. 
Perhaps streamlit is not the way to go for this type of application?
Thanks a lot for any thoughts on this!
Best wishes
Markus
import streamlit as st
import numpy as np
import time
import asyncio
# --------------------------------------------------------------
def ui():
if 'image' in st.session_state:
st.image(st.session_state['image'])
def work(delay):
time.sleep(delay)
st.session_state['image'] = np.random.random((512, 512))
# --------------------------------------------------------------
# --------------------------------------------------------------
async def async_ui():
with st.session_state['placeholder']:
ui()
r = await asyncio.sleep(0.01)
async def async_work(delay):
work(delay)
r = await asyncio.sleep(0.01)
# --------------------------------------------------------------
# --------------------------------------------------------------
def work_before_display(delay=1):
work(delay)
ui()
# --------------------------------------------------------------
def display_before_work(delay=1):
ui()
work(delay)
# --------------------------------------------------------------
async def do_all_asynchronously(delay=1):
if 'tasks' in st.session_state:
for t in st.session_state['tasks']:
t.cancel()
coroutines = [async_work(delay), async_ui()]
tasks = []
for s in coroutines:
tasks.append(asyncio.ensure_future(s))
if tasks:
st.session_state['tasks'] = tasks
res = await asyncio.gather(*tasks)
if __name__ == '__main__':
if 'placeholder' not in st.session_state:
st.session_state['placeholder'] = st.empty()
delay = 0.25
#display_before_work(delay=delay)
#work_before_display(delay=delay)
asyncio.run(do_all_asynchronously(delay=delay))
st.experimental_rerun()
see also
.
