basic streaming SSE w/ FastHTML and Plotly

hello_fasthtml_plotly_stream.py

import json
import time
import asyncio
import random
from fasthtml.common import *
import plotly.express as px
import pandas as pd

hdrs = (
    Script(src='static/js/htmx.min.js'),
    Script(src='static/js/plotly-2.32.0.min.js'),
    Link(rel='stylesheet', href='static/css/pico.min.css', type='text/css'),
    Link(rel='stylesheet', href='static/css/goat.css', type='text/css'),
)

app,rt = fast_app(
    hdrs=hdrs,
    htmx=False,
    pico=False,
    surreal=False
)

bg_color = '#13171f'
@app.get("/static/{fname:path}.{ext:static}")
def static(fname: str, ext: str):
  return FileResponse(f'static/{fname}.{ext}')

@rt('/')
def get():
    chart_div = Div(id='chartDiv')
    plot_script = Script(
        f"""
            var data = [{{
                x: [0,1],
                y: [1,2],
                mode: 'lines',
                line: {{color: '#FFF'}}
            }}]
            
            var layout = {{
                template:"plotly_dark",
                plot_bgcolor:"{bg_color}",
                paper_bgcolor:"{bg_color}",
            }}
            
            Plotly.newPlot('chartDiv', data, layout);
            const source = new EventSource('/stream');
            source.onmessage = function(event) {{
              const msg = JSON.parse(event.data);
              console.log('raw_msg', msg);
              var update = {{
                x: [msg.x],
                y: [msg.y]
              }}
              Plotly.extendTraces('chartDiv', update, [0]);
            }};
            
            source.onerror = function(err) {{
              console.error('SSE error: ', err);
            }};
        """
    )

    return Div(
    (
        P('GOAT RACER ONE', cls='goat-logo', hx_get="/change"),
        P('2025', cls='sevenseg')),
        Div(chart_div, plot_script),

        cls='container'
    )

@rt('/change')
def get(): return P('Nice to be here!')


count_x = 10

@app.get("/stream")
async def stream():
    """
    Async generator that yields a new data point every second.
    The payload format is:
        data: <JSON>\n\n
    which is the required Server‑Sent Events syntax.
    """
    global count_x
    async def event_generator():
        global count_x
        while True:
            await asyncio.sleep(0.5)

            # Example payload – **array** for each field!
            payload = {
                "x": [count_x],            # <-- note the outer []!
                "y": [random.uniform(5, 15)]   # <-- same here
            }
            count_x += 1

            # SSE requires the `data:` prefix and a double newline
            msg = f"data: {json.dumps(payload)}\n\n"
            yield msg.encode("utf-8")   # StreamingResponse expects bytes

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
    )
serve()