Demonstration of manipulating graph position w/a slider and buttons

slider_demo.py

import altair as alt
import streamlit as st
import pandas as pd
import numpy as np
import time


window_size = 10000


def generate(
    min_value: float, max_value: float, normal_dy: float, num_timesteps, file_name: str = "gendata.csv", initial_x: int = 0, initial_y: float = 0.0
) -> pd.DataFrame:
    """Generate some synthetic data"""
    sample = np.random.randn()
    total_range = max_value - min_value
    difference = sample * total_range
    last_value = min_value + difference

    # xy = []

    with open(str(file_name), "w", encoding="utf-8") as out_fp:
        # out_fp.write(f"x,y\n")
        # out_fp.write(f"0,{last_value}\n")

        bootstrap = False
        xvals, yvals = [], []

        for i in range(num_timesteps):
            if not bootstrap and initial_y > 0:
                bootstrap = True
                last_value = initial_y
            else:
                sample = np.random.randn()
                difference = sample * normal_dy

                spikeprob = np.random.randn()
                is_spike = spikeprob < 0.01
                if is_spike:
                    # let's amp up the signal
                    difference *= 2

                last_value += difference
            # out_fp.write(f"{initial_x+i+1},{last_value}\n")
            # xy.append((initial_x+i+1, last_value))
            xvals.append(initial_x+i+1)
            yvals.append(last_value)

    df = pd.DataFrame({'x': xvals, 'y': yvals}, dtype=float)
    return df


@st.cache_data
def load_data() -> pd.DataFrame:
    """Load the datasource into a DataFrame and return it; caching necessitates live
    data updates to place dataframe in st.session_state and concatenate new data"""
    data_load_state = st.text("generating synthetic data")
    df = generate(0, 1000, 10, 100000)
    data_load_state = st.text("synthetic data loaded!")
    st.session_state["df"] = df
    return df

def load_data_update(initial_x: int = 0, initial_y: int = 0) -> pd.DataFrame:
    """Simulate loading new data points"""
    data_load_state = st.text("generating live-synthetic data")
    df = generate(0, 1000, 10, 100, "upd.csv", initial_x=initial_x, initial_y=initial_y)
    st.session_state["df_upd"] = df
    data_load_state = st.text("synthetic live-data loaded!")
    return df

KEY_SLIDER = "Timestep Scrubber"
KEY_LOD = "Sample Size"
KEY_ANIM = "Animate Checkbox"

STATE_LOD = "LOD"
STATE_TS = "position"

lod_range = [100, 4600]
lod_step_sz = 500
lod_default = 2100

cb_animate = True


def _pos() -> int:
    return st.session_state[STATE_TS]


def _set_pos(value: int) -> None:
    st.session_state[STATE_TS] = value


def _lod() -> int:
    return st.session_state[STATE_LOD]


def _set_lod(value: int) -> None:
    st.session_state[STATE_LOD] = value


def _df() -> pd.DataFrame:
    df = st.session_state["df"]
    return df

def _set_df(df: pd.DataFrame) -> None:
    st.session_state["df"] = df


def _df_delta() -> pd.DataFrame:
    df = st.session_state["df_delta"]
    return df


def _set_df_delta(df: pd.DataFrame) -> None:
    st.session_state["df_delta"] = df


if STATE_TS not in st.session_state:
    _set_pos(0)

if STATE_LOD not in st.session_state:
    _set_lod(lod_default)

if "df" not in st.session_state:
    # Only put the historical data into session state if we haven't already loaded
    df = load_data()
    _set_df(df)
else:
    df = _df()

    if KEY_ANIM in st.session_state and st.session_state[KEY_ANIM]:
        df_delta = load_data_update(initial_x=df.shape[0], initial_y=df.iloc[-1].y)

        # grab new data...
        _set_df_delta(df_delta)
        df = pd.concat((df, df_delta), axis=0)
        _set_df(df)


y_min, y_max = df.y.min(), df.y.max()
y_range = y_min - 100, y_max + 100

st.title("Synthetic Data")

xmin, xmax = 0, df.shape[0] - window_size

st.text(f"xmin: {xmin}, xmax: {xmax}")
st.text(f"ymin: {y_range[0]}, ymax: {y_range[1]}")


def on_slider():
    """Event handler for propagating changes in the timeline slider"""
    _set_pos(st.session_state[KEY_SLIDER])


def on_lod():
    """Event handler for propagating changes in the selected resampling rate"""
    # Grab updated value from the LOD control and propagate out to app
    _set_lod(st.session_state[KEY_LOD])


def set_slider_value(value: int) -> None:
    """Utility function to update the value of the timeline slider"""
    # st.session_state[KEY_SLIDER] = value
    _set_pos(value)


def on_reverse():
    """Button click handler for user initiated navigation backwards through timeline"""
    new_x = max(0, x0 - window_size)
    _set_pos(new_x)


def on_forward():
    """Button click handler for user initiated navigation forwards through timeline"""
    new_x = min(xmax, x0 + window_size)
    set_slider_value(new_x)


col1, col2, col3 = st.columns([0.15, 0.8, 0.15])

with col1:
    l_button = st.button("Back", on_click=on_reverse)

with col2:
    # set_slider_value(xmax)
    st.slider(
        "Timeline Adjustment",
        min_value=0,
        max_value=xmax,
        key=KEY_SLIDER,
        step=window_size,
        value=_pos(),
        on_change=on_slider,
    )

with col3:
    r_button = st.button("Forward", on_click=on_forward)

col1, col2, col3, col4 = st.columns([0.15, 0.25, 0.25, 0.35])

with col2:
    cb_animate = st.checkbox("Animate", value=True, key=KEY_ANIM)

with col3:
    st.slider(
        "Resample Detail",
        min_value=lod_range[0],
        max_value=lod_range[1],
        key=KEY_LOD,
        step=lod_step_sz,
        value=_lod(),
        on_change=on_lod,
    )

st.subheader("Synth Data")

x0 = _pos()
lod = _lod()

chart = (
    alt.Chart(df.iloc[x0 : x0 + window_size])
    .transform_sample(lod)
    .mark_line()
    .encode(x=alt.X("x"), y=alt.Y("y", scale=alt.Scale(domain=y_range)))
    .interactive()
)

st.altair_chart(chart, theme="streamlit", use_container_width=True)
st.text(f"position: {st.session_state['position']}")

if cb_animate:
    # animated playback of viz data
    if x0 + window_size < df.shape[0]:
        ts = _pos()
        _set_pos(ts + 1000)

    # avoid never ending refreshes w/sleep
    time.sleep(0.1)
    # when animation is enabled, auto-run the next data update
    st.rerun()