"""Streamlit presentation layer for the first Meeting Assistant MVP.""" from __future__ import annotations from datetime import date from typing import Any from uuid import uuid4 import streamlit as st from mka.application.config import AppSettings, ConfigurationError from mka.application.meeting_service import ( STAGES, AppProgressEvent, MeetingDetails, MeetingProcessingService, ParticipantInput, ProcessingOptions, stable_id, ) from mka.application.people_yaml import ( PeopleYamlError, export_people_yaml, import_people_yaml, ) from mka.integrations.meeting_lab import ( MeetingLabGateway, MeetingLabUnavailableError, ) STAGE_LABELS = { "preparing": "Preparation", "transcription": "Transcription", "diarization": "Diarization", "protocol_generation": "Protocol generation", } def _new_participant() -> dict[str, str]: return { "row_id": uuid4().hex, "participant_id": "", "display_name": "", "role": "", "organization": "", "attendance_status": "present", } def _initialize_state() -> None: st.session_state.setdefault("participants", [_new_participant()]) st.session_state.setdefault("outcome", None) st.session_state.setdefault("edited_protocol", "") def _people_to_rows(people: list[ParticipantInput]) -> list[dict[str, str]]: """Create fresh widget rows while preserving reusable person IDs.""" return [ { "row_id": uuid4().hex, "participant_id": person.participant_id, "display_name": person.display_name, "role": person.role, "organization": person.organization, "attendance_status": person.attendance_status, } for person in people ] def _render_participants() -> list[ParticipantInput]: st.subheader("People") st.caption("Record whether each relevant person attended or was only mentioned.") imported_file = st.file_uploader( "Import people", type=["yaml", "yml"], help="Replace the current People list from a versioned YAML export.", ) if st.button("Import people list", disabled=imported_file is None): try: imported_people = import_people_yaml(imported_file.getvalue()) except PeopleYamlError as exc: st.error(str(exc)) else: st.session_state.participants = _people_to_rows(imported_people) st.session_state.people_import_message = f"Imported {len(imported_people)} people." st.rerun() if message := st.session_state.pop("people_import_message", None): st.success(message) rows = st.session_state.participants remove_index: int | None = None for index, row in enumerate(rows): row_id = row["row_id"] columns = st.columns([2, 2, 2, 2, 2, 0.6]) row["display_name"] = columns[0].text_input( "Name", value=row["display_name"], key=f"name_{row_id}" ) suggested_id = row["participant_id"] or stable_id(row["display_name"]) row["participant_id"] = columns[1].text_input( "Participant ID", value=suggested_id, key=f"id_{row_id}" ) row["role"] = columns[2].text_input("Role", value=row["role"], key=f"role_{row_id}") row["organization"] = columns[3].text_input( "Organization / department", value=row["organization"], key=f"organization_{row_id}", ) row["attendance_status"] = columns[4].selectbox( "Attendance", options=["present", "mentioned_only"], format_func=lambda value: { "present": "Present / participated", "mentioned_only": "Mentioned, but not present", }[value], index=0 if row["attendance_status"] == "present" else 1, key=f"attendance_{row_id}", ) if columns[5].button("Remove", key=f"remove_{row_id}"): remove_index = index if remove_index is not None: rows.pop(remove_index) st.rerun() action_columns = st.columns(2) if action_columns[0].button("Add person"): rows.append(_new_participant()) st.rerun() people = [ ParticipantInput( participant_id=row["participant_id"], display_name=row["display_name"], role=row["role"], organization=row["organization"], attendance_status=row["attendance_status"], ) for row in rows ] exported_yaml = export_people_yaml(people).encode("utf-8") action_columns[1].download_button( "Export people", data=exported_yaml, file_name="people.yaml", mime="application/yaml", ) return people def _progress_callback( status_box: Any, stage_table: Any, progress_slot: Any, states: dict[str, str], ) -> Any: progress_bar = None def update(event: AppProgressEvent) -> None: nonlocal progress_bar if event.stage in states: states[event.stage] = "running" if event.status == "started" else event.status if event.stage == "failed": running = next( (stage for stage, status in states.items() if status == "running"), None, ) if running: states[running] = "failed" elapsed = f"{event.elapsed_seconds:.1f} s" message = event.message or STAGE_LABELS.get(event.stage, event.stage) status_box.info(f"{message} — elapsed {elapsed}") stage_table.table( [{"Stage": STAGE_LABELS[stage], "Status": states[stage]} for stage in STAGES] ) if event.progress is not None: if progress_bar is None: progress_bar = progress_slot.progress(0.0) progress_bar.progress( min(max(event.progress, 0.0), 1.0), text=f"{STAGE_LABELS.get(event.stage, event.stage)}: {event.progress:.0%}", ) return update def _render_result() -> None: outcome = st.session_state.outcome if outcome is None: return st.divider() st.header("Protocol result") if not outcome.succeeded: st.error(f"Processing failed during {outcome.failed_stage}: {outcome.error_message}") if outcome.run_dir: st.code(str(outcome.run_dir)) st.caption("Intermediate artifacts and run metadata were preserved here.") return st.success("Processing completed. Review the generated protocol before use.") st.caption(f"Run artifacts: {outcome.run_dir}") with st.expander("Original generated protocol", expanded=False): st.markdown(outcome.original_protocol or "") edited = st.text_area( "Editable protocol", key="edited_protocol", height=500, help="The original protocol.md remains unchanged.", ) if st.button("Save edited protocol", type="primary"): path = MeetingProcessingService.save_edited_protocol(outcome.run_dir, edited) st.success(f"Saved edited protocol to {path}") def main() -> None: st.set_page_config(page_title="Meeting Assistant", layout="wide") _initialize_state() st.title("Meeting Assistant") st.caption("Create meeting context, run Meeting Lab, and review the protocol.") st.header("Meeting and audio") audio = st.file_uploader("Audio recording", type=["wav", "flac", "m4a"]) title = st.text_input("Meeting title") description = st.text_area("Description / context", height=100) metadata_columns = st.columns(3) language = metadata_columns[0].selectbox("Meeting language", options=["de", "en"]) has_date = metadata_columns[1].checkbox("Meeting date is known", value=True) selected_date: date | None = ( metadata_columns[2].date_input("Meeting date") if has_date else None ) participants = _render_participants() st.header("Processing options") audio_normalization = st.checkbox( "Audio normalization", value=True, help=( "Normalize speech loudness during preparation. WAV, FLAC, and M4A " "are always converted to the canonical Meeting Lab audio format, " "regardless of this setting." ), ) diarization_enabled = st.checkbox( "Enable speaker diarization", help="Speaker labels remain anonymous; identities are never inferred.", ) if st.button("Start processing", type="primary", disabled=audio is None): if not title.strip(): st.error("Meeting title is required.") elif any( not item.display_name.strip() or not item.participant_id.strip() for item in participants ): st.error("Every person row needs a name and participant ID.") else: settings = AppSettings.from_environment() try: service = MeetingProcessingService(settings, MeetingLabGateway()) meeting = MeetingDetails( title=title, language=language, meeting_date=selected_date, description=description, ) meeting_id = stable_id(title) audio_path = service.preserve_upload(meeting_id, audio.name, audio) st.header("Processing status") status_box = st.empty() stage_table = st.empty() progress_slot = st.empty() states = {stage: "pending" for stage in STAGES} if not diarization_enabled: states["diarization"] = "skipped" callback = _progress_callback(status_box, stage_table, progress_slot, states) outcome = service.process( audio_path, meeting, participants, ProcessingOptions( diarization_enabled=diarization_enabled, audio_normalization=audio_normalization, ), progress_sink=callback, ) st.session_state.outcome = outcome st.session_state.edited_protocol = outcome.original_protocol or "" if outcome.succeeded: status_box.success("Processing completed.") else: status_box.error( f"Processing failed during {outcome.failed_stage}: " f"{outcome.error_message} Artifacts were preserved." ) except (ConfigurationError, MeetingLabUnavailableError, ValueError) as exc: st.error(str(exc)) except Exception as exc: st.error(f"Unable to process meeting: {exc}") _render_result() if __name__ == "__main__": main()