Summarizer app

main.py

import pydub
import queue
import whisperx
import json

from langchain_ollama import ChatOllama


TMP_DIR = Path('temp')
if not TMP_DIR.exists():
    TMP_DIR.mkdir(exist_ok=True, parents=True)

MEDIA_STREAM_CONSTRAINTS = { 
    "video": False,
    "audio": {
        "echoCancellation": False,
        "noiseSuppression": True,
        "autoGainControl": True,
    },  
}

class Counter:

    def __init__(self, duration, display):
        self.duration=duration
        self.display=display

    @st.fragment(run_every=1)
    def decrease(self):
        self.show()
        self.duration -= 1

    def show(self):
        mm, ss = self.duration//60, self.duration%60
        self.display.metric("Countdown", f"{mm:02d}:{ss:02d}")


def summarize_text(text):
    llm = ChatOllama(model="llama3.1")
    messages = [ 
        (   
            "system",
            "You are a helpful assistant that provides a summarize of the following text in a concise single paragraph. The output only contains the translated TLDR summary to Spanish in Markdown format",
        ),  
        ("human", text),
    ]   
    ai_msg = llm.invoke(messages)

    return ai_msg.content
atch_size = 4
compute_type = "int8"

st.write("""
# Summarizer
""")

if "wavpath" not in st.session_state:
    cur_time = time.strftime("%Y-%m-%d_%H:%M:%S", time.localtime())
    tmp_wavpath = TMP_DIR / f'{cur_time}.wav'
    st.session_state["wavpath"] = str(tmp_wavpath)

wavpath = st.session_state["wavpath"]

counter = Counter(60, st.empty())
counter.show()

@st.fragment(run_every=1)
def start():
    counter.decrease()


def on_change():
    ctx = st.session_state["sendonly-audio"]
    if ctx.state.playing:
        print("***************")
        start()

webrtc_ctx = webrtc_streamer(
    key="sendonly-audio",
    mode=WebRtcMode.SENDONLY,
    media_stream_constraints=MEDIA_STREAM_CONSTRAINTS,
    on_change=on_change,
)

if "audio_buffer" not in st.session_state:
    st.session_state["audio_buffer"] = pydub.AudioSegment.empty()

status_indicator = st.empty()
transcript = st.empty()
lottie = False
while True:
    if webrtc_ctx.audio_receiver:
        try:
            audio_frames = webrtc_ctx.audio_receiver.get_frames(timeout=1)
        except queue.Empty:
            status_indicator.info("No frame arrived.")
            continue

        for i, audio_frame in enumerate(audio_frames):
            sound = pydub.AudioSegment(
                data=audio_frame.to_ndarray().tobytes(),
                sample_width=audio_frame.format.bytes,
                frame_rate=audio_frame.sample_rate,
                channels=len(audio_frame.layout.channels),
            )
            st.session_state["audio_buffer"] += sound

pyproject.toml

[project]
name = "ai-scrum-master"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [ 
    "langchain-ollama>=0.3.0",
    "pydub>=0.25.1",
    "streamlit>=1.44.1",
    "streamlit-webrtc>=0.57.0",
    "whisperx>=3.3.1",
]