| import streamlit as st |
| from streamlit_js_eval import streamlit_js_eval |
| import uuid, io, sys, os |
|
|
| def get_user_id(): |
| if "user_id" not in st.session_state: |
| query_params = st.query_params |
| user_id = query_params.get("user_id", None) |
| if not user_id: |
| user_id = str(uuid.uuid4()) |
| st.query_params["user_id"] = user_id |
| st.session_state.user_id = user_id |
| return st.session_state.user_id |
|
|
| def get_credentials(use_own_creds: bool): |
| if use_own_creds: |
| HIVEMQ_USERNAME = st.text_input("HIVEMQ Username", type="password") |
| HIVEMQ_PASSWORD = st.text_input("HIVEMQ Password", type="password") |
| HIVEMQ_HOST = st.text_input("HIVEMQ Host", type="password") |
| DEVICE_ENDPOINT = st.text_input("Device ID", type="password") |
| PORT = st.number_input("Port", min_value=1, step=1, value=8883) |
| else: |
| HIVEMQ_USERNAME = os.environ.get("HIVEMQ_USERNAME") |
| HIVEMQ_PASSWORD = os.environ.get("HIVEMQ_PASSWORD") |
| HIVEMQ_HOST = os.environ.get("HIVEMQ_HOST") |
| DEVICE_ENDPOINT = os.environ.get("DEVICE_ID") |
| PORT = int(os.environ.get("PORT", 8883)) |
| return HIVEMQ_USERNAME, HIVEMQ_PASSWORD, HIVEMQ_HOST, DEVICE_ENDPOINT, PORT |
|
|
| def render_log_window(): |
| class StreamlitRedirector: |
| def write(self, message): |
| if message.strip(): |
| st.write(message) |
|
|
| sys.stdout = StreamlitRedirector() |
|
|
| def refresh_once(): |
| has_refreshed = st.query_params.get("has_refreshed", None) |
| if not has_refreshed: |
| st.query_params["has_refreshed"] = True |
| sys.stdout = sys.__stdout__ |
| streamlit_js_eval(js_expressions="parent.window.location.reload()") |
