diff --git a/.gitignore b/.gitignore index a159afc..8f27803 100644 --- a/.gitignore +++ b/.gitignore @@ -161,4 +161,13 @@ cython_debug/ .idea/ .vscode/ -nohup.out \ No newline at end of file +nohup.out + +# Local clinic runtime data +clinic_appointments.db +clinic_appointments.json +clinic_appointments.csv + +# Locally downloaded Whisper models; tiny.en.pt is the small tracked fixture. +whisperflow/models/base.pt +whisperflow/models/tiny.pt diff --git a/TWILIO_SETUP.md b/TWILIO_SETUP.md new file mode 100644 index 0000000..806211f --- /dev/null +++ b/TWILIO_SETUP.md @@ -0,0 +1,80 @@ +# Twilio Clinic Assistant + +This project includes a Czech phone assistant for clinic appointment requests. + +## Run locally + +```bash +cd /Users/bobbysixkiller/vioce/whisper-flow +.venv/bin/uvicorn twilio_clinic_assistant:APP --host 127.0.0.1 --port 8000 +``` + +## Expose to Twilio + +Twilio must reach your local webhook over HTTPS. The simplest local tunnel is ngrok: + +```bash +ngrok http 8000 +``` + +Use the HTTPS URL from ngrok, for example: + +```text +https://example.ngrok-free.app/voice +``` + +## Configure Twilio + +In Twilio Console, open your phone number and set: + +- Voice configuration: `A call comes in` +- Webhook method: `POST` +- Webhook URL: `https://example.ngrok-free.app/voice` +- Enable call recording for this number or Voice configuration. +- Recording status callback URL: `https://example.ngrok-free.app/recording-status` +- Recording status callback method: `POST` + +The webhook returns TwiML with ``. +Twilio sends recognized Czech speech back as `SpeechResult`. + +## Appointment output + +Confirmed requests are saved to: + +```text +clinic_appointments.db +``` + +The webhook automatically stores Twilio's `From` phone number in the +`caller_phone` column so the doctor can call the patient back. VS Code-friendly +exports are refreshed after every saved request: + +```text +clinic_appointments.json +clinic_appointments.csv +``` + +For speech quality, the webhook uses Czech speech hints and stores confidence +scores for each collected field. If Twilio reports low confidence, the assistant +asks the patient to repeat the answer once before moving on. The exported +columns `name_confidence`, `reason_confidence`, `requested_time_confidence`, and +`raw_transcript_json` help the office spot answers that may need a callback. + +Whole-call recordings are stored by Twilio. The app stores recording metadata in +the appointment export when Twilio calls `/recording-status`: `recording_sid`, +`recording_url`, `recording_status`, `recording_duration`, `recording_channels`, +`recording_track`, and `recording_start_time`. The `recording_url` points to the +audio file in Twilio; access to the file requires the Twilio account credentials. + +You can inspect saved requests while the server is running: + +```text +http://127.0.0.1:8000/appointments +``` + +## Current scope + +This is an appointment intake assistant. It does not yet check a real calendar, +reserve a specific slot, send SMS confirmations, or integrate with medical +software. The next production step is to connect the confirmation step to a +calendar or scheduling backend. diff --git a/run.sh b/run.sh index cc65753..2e0dd49 100755 --- a/run.sh +++ b/run.sh @@ -20,13 +20,16 @@ elif [ $1 = "-local" ]; then echo "Running format, linter and tests" rm -rf .venv - # Use python3.12 if available as it's more stable for currently pinned dependencies - PYTHON_CMD="python3" + # Use Python >=3.10 for currently pinned dependencies. if command -v python3.12 >/dev/null 2>&1; then - PYTHON_CMD="python3.12" + python3.12 -m venv .venv + elif command -v python3.11 >/dev/null 2>&1; then + python3.11 -m venv .venv + elif command -v uv >/dev/null 2>&1; then + uv venv .venv --python 3.11 --seed + else + python3 -m venv .venv fi - - $PYTHON_CMD -m venv .venv source .venv/bin/activate pip install --upgrade pip wheel # Pin setuptools < 70 for openai-whisper compatibility @@ -63,7 +66,11 @@ elif [ $1 = "-benchmark" ]; then kill $(lsof -t -i:8181) elif [ $1 = "-run-server" ]; then echo "Running WhisperFlow server" - kill $(lsof -t -i:8181) + source .venv/bin/activate + SERVER_PID=$(lsof -t -i:8181) + if [ -n "$SERVER_PID" ]; then + kill $SERVER_PID + fi uvicorn whisperflow.fast_server:app --host 0.0.0.0 --port 8181 elif [ $1 = "-test-package" ]; then echo "Running WhisperFlow package setup" @@ -88,4 +95,4 @@ else fi trap : 0 -echo >&2 '*** DONE ***' \ No newline at end of file +echo >&2 '*** DONE ***' diff --git a/tests/audio/test_audio.py b/tests/audio/test_audio.py index 20f857a..9d26297 100644 --- a/tests/audio/test_audio.py +++ b/tests/audio/test_audio.py @@ -4,12 +4,28 @@ import asyncio import pytest import numpy as np +import pyaudio import whisperflow.audio.microphone as mic +def has_input_device(): + """Return whether PortAudio can see any input device.""" + audio = pyaudio.PyAudio() + try: + return any( + audio.get_device_info_by_index(index).get("maxInputChannels", 0) > 0 + for index in range(audio.get_device_count()) + ) + finally: + audio.terminate() + + @pytest.mark.asyncio async def test_capture_mic(): """test capturing microphone""" + if not has_input_device(): + pytest.skip("PortAudio has no available input device") + stop_event = asyncio.Event() audio_chunks = queue.Queue() diff --git a/twilio_clinic_assistant.py b/twilio_clinic_assistant.py new file mode 100644 index 0000000..e2f6fcf --- /dev/null +++ b/twilio_clinic_assistant.py @@ -0,0 +1,554 @@ +#!/usr/bin/env python3 +"""Twilio Voice webhook for a Czech clinic appointment assistant.""" + +from __future__ import annotations + +import csv +import json +import sqlite3 +from datetime import datetime, timezone +from html import escape +from pathlib import Path +from typing import Any + +from fastapi import FastAPI, Form, Query +from fastapi.responses import PlainTextResponse + + +APP = FastAPI(title="Clinic Phone Assistant") +BASE_DIR = Path(__file__).resolve().parent +APPOINTMENTS_DB_PATH = BASE_DIR / "clinic_appointments.db" +LOW_CONFIDENCE_THRESHOLD = 0.65 +COMMON_SPEECH_HINTS = ( + "praktický lékař,kontrola,preventivní prohlídka,odběry krve,recept," + "neschopenka,bolest,teplota,kašel,rýma,moč,pálení při močení,tlak," + "cukrovka,cholesterol,pondělí,úterý,středa,čtvrtek,pátek,dopoledne," + "odpoledne,ráno,zítra,příští týden" +) +STEP_SPEECH_HINTS = { + "name": "jméno,příjmení,Novák,Nováková,Svoboda,Svobodová,Dvořák,Dvořáková", + "reason": COMMON_SPEECH_HINTS, + "time": ( + "dnes,zítra,pozítří,pondělí,úterý,středa,čtvrtek,pátek," + "ráno,dopoledne,poledne,odpoledne,večer,příští týden" + ), + "confirm": "ano,jo,správně,souhlas,ne,nesouhlas,špatně,jedna,dva", +} +APPOINTMENT_COLUMNS = ( + "id", + "created_at", + "call_sid", + "name", + "caller_phone", + "name_confidence", + "reason", + "reason_confidence", + "requested_time", + "requested_time_confidence", + "status", + "raw_transcript_json", + "recording_sid", + "recording_url", + "recording_status", + "recording_duration", + "recording_channels", + "recording_track", + "recording_start_time", +) +CALL_STATE: dict[str, dict[str, str]] = {} + + +def xml_response(body: str) -> PlainTextResponse: + return PlainTextResponse( + '\n' + body, + media_type="application/xml", + ) + + +def say(text: str) -> str: + return f'{escape(text)}' + + +def gather(prompt: str, action: str, step: str) -> str: + hints = STEP_SPEECH_HINTS.get(step, COMMON_SPEECH_HINTS) + return ( + f'' + f"{say(prompt)}" + "" + ) + + +def redirect(path: str) -> str: + return f'{escape(path)}' + + +def normalize_input(speech_result: str | None, digits: str | None) -> str: + if speech_result and speech_result.strip(): + return speech_result.strip() + if digits and digits.strip(): + return digits.strip() + return "" + + +def normalize_phone(phone: str) -> str: + """Normalize phone numbers from form-encoded Twilio callbacks.""" + clean = phone.strip() + if phone.startswith(" ") and clean.startswith("420"): + return f"+{clean}" + return clean + + +def parse_confidence(value: str | None) -> float | None: + """Parse Twilio's optional Confidence value.""" + if value in (None, ""): + return None + try: + return float(value) + except ValueError: + return None + + +def should_repeat( + step: str, confidence: float | None, call_state: dict[str, str] +) -> bool: + """Repeat once when Twilio reports a low-confidence speech result.""" + if confidence is None or confidence >= LOW_CONFIDENCE_THRESHOLD: + return False + retry_key = f"{step}_retry" + if call_state.get(retry_key) == "1": + return False + call_state[retry_key] = "1" + return True + + +def store_answer( + call_state: dict[str, str], + field_name: str, + answer: str, + confidence: float | None, +) -> None: + """Store the recognized text plus confidence for later review.""" + call_state[field_name] = answer + if confidence is not None: + call_state[f"{field_name}_confidence"] = f"{confidence:.3f}" + + +def raw_transcript(call_state: dict[str, str]) -> str: + """Return per-field transcript details as JSON.""" + payload = {} + for field_name in ("name", "reason", "requested_time"): + payload[field_name] = { + "text": call_state.get(field_name, ""), + "confidence": call_state.get(f"{field_name}_confidence", ""), + } + return json.dumps(payload, ensure_ascii=False) + + +def init_database(db_path: Path | None = None) -> None: + """Create or migrate the shared appointment database.""" + db_path = db_path or APPOINTMENTS_DB_PATH + with sqlite3.connect(db_path) as conn: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS appointments ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + created_at TEXT NOT NULL, + call_sid TEXT NOT NULL DEFAULT '', + name TEXT NOT NULL DEFAULT '', + caller_phone TEXT NOT NULL DEFAULT '', + name_confidence TEXT NOT NULL DEFAULT '', + reason TEXT NOT NULL, + reason_confidence TEXT NOT NULL DEFAULT '', + requested_time TEXT NOT NULL, + requested_time_confidence TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL, + raw_transcript_json TEXT NOT NULL DEFAULT '', + recording_sid TEXT NOT NULL DEFAULT '', + recording_url TEXT NOT NULL DEFAULT '', + recording_status TEXT NOT NULL DEFAULT '', + recording_duration TEXT NOT NULL DEFAULT '', + recording_channels TEXT NOT NULL DEFAULT '', + recording_track TEXT NOT NULL DEFAULT '', + recording_start_time TEXT NOT NULL DEFAULT '' + ) + """ + ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS call_recordings ( + call_sid TEXT PRIMARY KEY, + recording_sid TEXT NOT NULL DEFAULT '', + recording_url TEXT NOT NULL DEFAULT '', + recording_status TEXT NOT NULL DEFAULT '', + recording_duration TEXT NOT NULL DEFAULT '', + recording_channels TEXT NOT NULL DEFAULT '', + recording_track TEXT NOT NULL DEFAULT '', + recording_start_time TEXT NOT NULL DEFAULT '', + updated_at TEXT NOT NULL + ) + """ + ) + columns = { + row[1] for row in conn.execute("PRAGMA table_info(appointments)").fetchall() + } + for column in ( + "call_sid", + "name", + "caller_phone", + "name_confidence", + "reason_confidence", + "requested_time_confidence", + "raw_transcript_json", + "recording_sid", + "recording_url", + "recording_status", + "recording_duration", + "recording_channels", + "recording_track", + "recording_start_time", + ): + if column not in columns: + conn.execute( + f"ALTER TABLE appointments ADD COLUMN {column} TEXT NOT NULL DEFAULT ''" + ) + + +def appointment_export_paths(db_path: Path | None = None) -> tuple[Path, Path]: + """Return human-readable export paths next to the SQLite database.""" + db_path = db_path or APPOINTMENTS_DB_PATH + return db_path.with_suffix(".json"), db_path.with_suffix(".csv") + + +def load_appointments() -> list[dict[str, Any]]: + init_database() + with sqlite3.connect(APPOINTMENTS_DB_PATH) as conn: + conn.row_factory = sqlite3.Row + rows = conn.execute( + """ + SELECT + id, created_at, call_sid, name, caller_phone, name_confidence, + reason, reason_confidence, requested_time, + requested_time_confidence, status, raw_transcript_json, + recording_sid, recording_url, recording_status, recording_duration, + recording_channels, recording_track, recording_start_time + FROM appointments + ORDER BY id + """ + ).fetchall() + return [{column: row[column] for column in APPOINTMENT_COLUMNS} for row in rows] + + +def export_appointments() -> None: + """Export saved appointments to files that VS Code can display directly.""" + appointments = load_appointments() + json_path, csv_path = appointment_export_paths() + + json_path.write_text( + json.dumps(appointments, ensure_ascii=False, indent=2) + "\n", + encoding="utf-8", + ) + with csv_path.open("w", encoding="utf-8", newline="") as csv_file: + writer = csv.DictWriter(csv_file, fieldnames=APPOINTMENT_COLUMNS) + writer.writeheader() + writer.writerows(appointments) + + +def save_appointment(appointment: dict[str, Any]) -> None: + init_database() + with sqlite3.connect(APPOINTMENTS_DB_PATH) as conn: + conn.execute( + """ + INSERT INTO appointments ( + created_at, call_sid, name, caller_phone, name_confidence, + reason, reason_confidence, requested_time, + requested_time_confidence, status, raw_transcript_json, + recording_sid, recording_url, recording_status, recording_duration, + recording_channels, recording_track, recording_start_time + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + appointment["created_at"], + appointment.get("call_sid", ""), + appointment.get("name", ""), + appointment.get("caller_phone", ""), + appointment.get("name_confidence", ""), + appointment.get("reason", ""), + appointment.get("reason_confidence", ""), + appointment.get("requested_time", ""), + appointment.get("requested_time_confidence", ""), + appointment.get("status", "requested"), + appointment.get("raw_transcript_json", ""), + appointment.get("recording_sid", ""), + appointment.get("recording_url", ""), + appointment.get("recording_status", ""), + appointment.get("recording_duration", ""), + appointment.get("recording_channels", ""), + appointment.get("recording_track", ""), + appointment.get("recording_start_time", ""), + ), + ) + export_appointments() + + +def save_recording_metadata(recording: dict[str, str]) -> None: + """Store Twilio call recording metadata and attach it to the appointment.""" + init_database() + updated_at = datetime.now(timezone.utc).isoformat() + with sqlite3.connect(APPOINTMENTS_DB_PATH) as conn: + conn.execute( + """ + INSERT INTO call_recordings ( + call_sid, recording_sid, recording_url, recording_status, + recording_duration, recording_channels, recording_track, + recording_start_time, updated_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(call_sid) DO UPDATE SET + recording_sid = excluded.recording_sid, + recording_url = excluded.recording_url, + recording_status = excluded.recording_status, + recording_duration = excluded.recording_duration, + recording_channels = excluded.recording_channels, + recording_track = excluded.recording_track, + recording_start_time = excluded.recording_start_time, + updated_at = excluded.updated_at + """, + ( + recording.get("call_sid", ""), + recording.get("recording_sid", ""), + recording.get("recording_url", ""), + recording.get("recording_status", ""), + recording.get("recording_duration", ""), + recording.get("recording_channels", ""), + recording.get("recording_track", ""), + recording.get("recording_start_time", ""), + updated_at, + ), + ) + conn.execute( + """ + UPDATE appointments + SET recording_sid = ?, + recording_url = ?, + recording_status = ?, + recording_duration = ?, + recording_channels = ?, + recording_track = ?, + recording_start_time = ? + WHERE call_sid = ? + """, + ( + recording.get("recording_sid", ""), + recording.get("recording_url", ""), + recording.get("recording_status", ""), + recording.get("recording_duration", ""), + recording.get("recording_channels", ""), + recording.get("recording_track", ""), + recording.get("recording_start_time", ""), + recording.get("call_sid", ""), + ), + ) + export_appointments() + + +def state_for(call_sid: str) -> dict[str, str]: + return CALL_STATE.setdefault(call_sid, {}) + + +@APP.get("/health") +async def health() -> dict[str, str]: + return {"status": "ok"} + + +@APP.api_route("/voice", methods=["GET", "POST"]) +async def voice() -> PlainTextResponse: + response = ( + "" + + say( + "Upozornění: pro přesné vyřízení bude tento hovor nahráván. " + "Pokračováním s tím souhlasíte." + ) + + gather( + "Dobrý den, dovolali jste se do ordinace praktického lékaře. " + "U telefonu je automatická sestra ordinace. " + "Pomohu vám předat žádost o objednání. " + "Mluvte prosím pomalu a krátce. " + "Nejprve prosím řekněte své celé jméno.", + "/voice/collect?step=name", + "name", + ) + + say("Neslyšel jsem odpověď. Zkusíme to ještě jednou.") + + redirect("/voice") + + "" + ) + return xml_response(response) + + +@APP.post("/voice/collect") +async def collect( + step: str = Query(...), + CallSid: str = Form("unknown"), + From: str = Form(""), + SpeechResult: str | None = Form(None), + Confidence: str | None = Form(None), + Digits: str | None = Form(None), +) -> PlainTextResponse: + answer = normalize_input(SpeechResult, Digits) + confidence = parse_confidence(Confidence) + call_state = state_for(CallSid) + + if not answer: + return xml_response( + "" + + gather( + "Neslyšel jsem vás dobře. Zopakujte to prosím.", + f"/voice/collect?step={step}", + step, + ) + + say( + "Omlouvám se, odpověď se nepodařilo rozpoznat. Zavolejte prosím později." + ) + + "" + + "" + ) + + if should_repeat(step, confidence, call_state): + return xml_response( + "" + + gather( + "Nejsem si jistý, že jsem správně rozuměl. " + "Řekněte to prosím ještě jednou, pomalu a krátce.", + f"/voice/collect?step={step}", + step, + ) + + say("Omlouvám se, odpověď se nepodařilo rozpoznat.") + + "" + + "" + ) + + if step == "name": + store_answer(call_state, "name", answer, confidence) + next_prompt = ( + f"Děkuji. Mám jméno {answer}. " + "Teď mi prosím jednou krátkou větou řekněte důvod návštěvy." + ) + return xml_response( + "" + + gather(next_prompt, "/voice/collect?step=reason", "reason") + + say("Neslyšel jsem důvod návštěvy.") + + redirect("/voice/collect?step=reason") + + "" + ) + + if step == "reason": + store_answer(call_state, "reason", answer, confidence) + return xml_response( + "" + + gather( + "Rozumím. Teď prosím řekněte, kdy by se vám termín hodil. " + "Například zítra dopoledne, nebo příští úterý v deset.", + "/voice/collect?step=time", + "time", + ) + + say("Neslyšel jsem požadovaný termín.") + + redirect("/voice/collect?step=time") + + "" + ) + + if step == "time": + store_answer(call_state, "requested_time", answer, confidence) + summary = ( + f"Rekapitulace. Jméno: {call_state.get('name', 'neuvedeno')}. " + f"Důvod návštěvy: {call_state.get('reason', 'neuvedeno')}. " + f"Požadovaný termín: {answer}. " + "Pokud je to správně, řekněte ano nebo stiskněte jedničku. " + "Pokud ne, řekněte ne nebo stiskněte dvojku." + ) + return xml_response( + "" + + gather(summary, "/voice/collect?step=confirm", "confirm") + + say("Potvrzení jsem neslyšel.") + + redirect("/voice/collect?step=confirm") + + "" + ) + + if step == "confirm": + lowered = answer.lower() + if any(word in lowered for word in ("ano", "jo", "správně", "souhlas", "1")): + appointment = { + "created_at": datetime.now(timezone.utc).isoformat(), + "call_sid": CallSid, + "caller_phone": normalize_phone(From), + "name": call_state.get("name", ""), + "name_confidence": call_state.get("name_confidence", ""), + "reason": call_state.get("reason", ""), + "reason_confidence": call_state.get("reason_confidence", ""), + "requested_time": call_state.get("requested_time", ""), + "requested_time_confidence": call_state.get( + "requested_time_confidence", "" + ), + "status": "requested", + "raw_transcript_json": raw_transcript(call_state), + } + save_appointment(appointment) + CALL_STATE.pop(CallSid, None) + return xml_response( + "" + + say( + "Děkuji, žádost o objednání jsem uložil. " + "Pan doktor nebo sestřička vás bude kontaktovat s potvrzením " + "přesného termínu. Na shledanou." + ) + + "" + + "" + ) + + CALL_STATE.pop(CallSid, None) + return xml_response( + "" + + say("Dobře, začneme znovu.") + + redirect("/voice") + + "" + ) + + return xml_response( + "" + + say("Omlouvám se, nastala chyba v hovoru. Zavolejte prosím znovu.") + + "" + + "" + ) + + +@APP.get("/appointments") +async def appointments() -> list[dict[str, Any]]: + return load_appointments() + + +@APP.post("/recording-status") +async def recording_status( + CallSid: str = Form(""), + RecordingSid: str = Form(""), + RecordingUrl: str = Form(""), + RecordingStatus: str = Form(""), + RecordingDuration: str = Form(""), + RecordingChannels: str = Form(""), + RecordingTrack: str = Form(""), + RecordingStartTime: str = Form(""), +) -> dict[str, str]: + """Receive Twilio recording status callbacks for whole-call recordings.""" + save_recording_metadata( + { + "call_sid": CallSid, + "recording_sid": RecordingSid, + "recording_url": RecordingUrl, + "recording_status": RecordingStatus, + "recording_duration": RecordingDuration, + "recording_channels": RecordingChannels, + "recording_track": RecordingTrack, + "recording_start_time": RecordingStartTime, + } + ) + return {"status": "ok"} diff --git a/voice_assistant.py b/voice_assistant.py new file mode 100644 index 0000000..4249799 --- /dev/null +++ b/voice_assistant.py @@ -0,0 +1,676 @@ +#!/usr/bin/env python3 +"""Local real-time Czech voice assistant client for WhisperFlow.""" + +from __future__ import annotations + +import argparse +import asyncio +import csv +import datetime as dt +import json +import queue +import re +import signal +import sqlite3 +import threading +import time +from dataclasses import dataclass, field +from pathlib import Path + +import numpy as np +import pyaudio +import torch +import whisper +import websockets +from omnivoice import OmniVoice + + +SAMPLE_RATE = 16000 +CHUNK_SIZE = 1024 +FORMAT = pyaudio.paInt16 +CHANNELS = 1 +SILENCE_THRESHOLD = 500 +MIN_SPEECH_CHUNKS = 8 +SILENCE_CHUNKS_TO_CLOSE = 12 +TTS_SAMPLE_RATE = 24000 +APPOINTMENT_COLUMNS = ( + "id", + "created_at", + "call_sid", + "name", + "caller_phone", + "name_confidence", + "reason", + "reason_confidence", + "requested_time", + "requested_time_confidence", + "status", + "raw_transcript_json", + "recording_sid", + "recording_url", + "recording_status", + "recording_duration", + "recording_channels", + "recording_track", + "recording_start_time", +) + + +@dataclass +class ClinicDialog: + """Small deterministic appointment intake dialog.""" + + db_path: Path + step: str = "name" + name: str = "" + caller_phone: str = "" + reason: str = "" + requested_time: str = "" + appointments: list[dict[str, str]] = field(default_factory=list) + + def greeting(self) -> str: + """Return the first prompt for the caller.""" + return ( + "Dobrý den, dovolali jste se do ordinace praktického lékaře. " + "Jsem hlasový asistent pro objednávání termínů. " + "Nejprve mi prosím řekněte své celé jméno." + ) + + def handle(self, text: str) -> str: + """Advance the intake dialog and return the next spoken response.""" + clean_text = clean_transcript(text) + lower = clean_text.lower() + + if any(word in lower for word in ("konec", "ukončit", "nashledanou")): + self.step = "done" + return "Dobře, končím hovor. Na shledanou." + + if self.step == "name": + self.name = clean_text + self.step = "reason" + return f"Děkuji. S čím k nám prosím jdete, {self.name}?" + + if self.step == "reason": + self.reason = clean_text + self.step = "time" + return ( + "Rozumím. Kdy by se vám termín hodil? " + "Řekněte prosím den a přibližný čas, například zítra dopoledne, " + "nebo příští úterý v deset." + ) + + if self.step == "time": + self.requested_time = clean_text + appointment_id = save_appointment( + self.db_path, + self.name, + self.reason, + self.requested_time, + self.caller_phone, + ) + self.appointments.append( + { + "id": str(appointment_id), + "name": self.name, + "caller_phone": self.caller_phone, + "reason": self.reason, + "requested_time": self.requested_time, + } + ) + self.step = "done" + preparation = preparation_instructions(self.reason) + return ( + "Děkuji, žádost o objednání jsem uložil. " + f"Jméno: {self.name}. " + f"Termín: {self.requested_time}. " + f"Důvod návštěvy: {self.reason}. " + f"{preparation} " + "Ordinace vám termín ještě potvrdí. Děkujeme za zavolání. Na shledanou." + ) + + return "Hovor už je ukončený. Pro nový termín spusťte asistenta znovu." + + +def clean_transcript(text: str, max_len: int = 160) -> str: + """Normalize and limit STT output before it reaches dialog/TTS.""" + clean = re.sub(r"\s+", " ", text).strip(" .,!?:;") + parts = [part.strip() for part in re.split(r"[,.;]", clean) if part.strip()] + deduped = [] + for part in parts: + if not deduped or part.lower() != deduped[-1].lower(): + deduped.append(part) + clean = ", ".join(deduped) if deduped else clean + if len(clean) > max_len: + clean = clean[:max_len].rsplit(" ", 1)[0] + return clean or "neuvedeno" + + +def init_database(db_path: Path) -> None: + """Create the appointment table if needed.""" + with sqlite3.connect(db_path) as conn: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS appointments ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + created_at TEXT NOT NULL, + call_sid TEXT NOT NULL DEFAULT '', + name TEXT NOT NULL DEFAULT '', + caller_phone TEXT NOT NULL DEFAULT '', + name_confidence TEXT NOT NULL DEFAULT '', + reason TEXT NOT NULL, + reason_confidence TEXT NOT NULL DEFAULT '', + requested_time TEXT NOT NULL, + requested_time_confidence TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL, + raw_transcript_json TEXT NOT NULL DEFAULT '', + recording_sid TEXT NOT NULL DEFAULT '', + recording_url TEXT NOT NULL DEFAULT '', + recording_status TEXT NOT NULL DEFAULT '', + recording_duration TEXT NOT NULL DEFAULT '', + recording_channels TEXT NOT NULL DEFAULT '', + recording_track TEXT NOT NULL DEFAULT '', + recording_start_time TEXT NOT NULL DEFAULT '' + ) + """ + ) + columns = { + row[1] for row in conn.execute("PRAGMA table_info(appointments)").fetchall() + } + if "name" not in columns: + conn.execute( + "ALTER TABLE appointments ADD COLUMN name TEXT NOT NULL DEFAULT ''" + ) + if "caller_phone" not in columns: + conn.execute( + "ALTER TABLE appointments ADD COLUMN caller_phone TEXT NOT NULL DEFAULT ''" + ) + for column in ( + "call_sid", + "name_confidence", + "reason_confidence", + "requested_time_confidence", + "raw_transcript_json", + "recording_sid", + "recording_url", + "recording_status", + "recording_duration", + "recording_channels", + "recording_track", + "recording_start_time", + ): + if column not in columns: + conn.execute( + f"ALTER TABLE appointments ADD COLUMN {column} TEXT NOT NULL DEFAULT ''" + ) + export_appointments(db_path) + + +def appointment_export_paths(db_path: Path) -> tuple[Path, Path]: + """Return human-readable export paths next to the SQLite database.""" + return db_path.with_suffix(".json"), db_path.with_suffix(".csv") + + +def load_saved_appointments(db_path: Path) -> list[dict[str, str]]: + """Read saved appointments from SQLite in a stable column order.""" + if not db_path.exists(): + return [] + + with sqlite3.connect(db_path) as conn: + conn.row_factory = sqlite3.Row + rows = conn.execute( + """ + SELECT + id, created_at, call_sid, name, caller_phone, name_confidence, + reason, reason_confidence, requested_time, + requested_time_confidence, status, raw_transcript_json, + recording_sid, recording_url, recording_status, recording_duration, + recording_channels, recording_track, recording_start_time + FROM appointments + ORDER BY id + """ + ).fetchall() + + return [ + {column: str(row[column]) for column in APPOINTMENT_COLUMNS} for row in rows + ] + + +def export_appointments(db_path: Path) -> None: + """Export SQLite appointments to JSON and CSV files readable in VS Code.""" + appointments = load_saved_appointments(db_path) + json_path, csv_path = appointment_export_paths(db_path) + + json_path.write_text( + json.dumps(appointments, ensure_ascii=False, indent=2) + "\n", + encoding="utf-8", + ) + with csv_path.open("w", encoding="utf-8", newline="") as csv_file: + writer = csv.DictWriter(csv_file, fieldnames=APPOINTMENT_COLUMNS) + writer.writeheader() + writer.writerows(appointments) + + +def save_appointment( + db_path: Path, + name: str, + reason: str, + requested_time: str, + caller_phone: str = "", +) -> int: + """Persist an appointment request and return its id.""" + init_database(db_path) + with sqlite3.connect(db_path) as conn: + cursor = conn.execute( + """ + INSERT INTO appointments ( + created_at, call_sid, name, caller_phone, name_confidence, + reason, reason_confidence, requested_time, + requested_time_confidence, status, raw_transcript_json, + recording_sid, recording_url, recording_status, recording_duration, + recording_channels, recording_track, recording_start_time + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + dt.datetime.now().isoformat(timespec="seconds"), + "", + name, + caller_phone, + "", + reason, + "", + requested_time, + "", + "requested", + "", + "", + "", + "", + "", + "", + "", + "", + ), + ) + appointment_id = int(cursor.lastrowid) + export_appointments(db_path) + return appointment_id + + +def preparation_instructions(reason: str) -> str: + """Return short pre-visit instructions based on the appointment reason.""" + lower = reason.lower() + base = "Vezměte si prosím kartičku pojišťovny, občanský průkaz a seznam léků." + urine_keywords = ("moč", "močí", "močový", "pálení", "ledvin", "urolog", "cukr") + blood_keywords = ("krev", "odběr", "preventiv", "kontrola", "cukrov") + + if any(keyword in lower for keyword in urine_keywords): + return ( + base + " Pokud jdete kvůli moči nebo pálení při močení, " + "přineste prosím ranní vzorek moči v čisté nádobce." + ) + if any(keyword in lower for keyword in blood_keywords): + return base + " Pokud půjdete na odběry krve, přijďte prosím nalačno." + return base + + +def choose_input_device(audio: pyaudio.PyAudio, preferred: str | None) -> int | None: + """Choose an input device by substring, otherwise first available input.""" + fallback = None + for index in range(audio.get_device_count()): + info = audio.get_device_info_by_index(index) + if info.get("maxInputChannels", 0) <= 0: + continue + if fallback is None: + fallback = index + if preferred and preferred.lower() in str(info.get("name", "")).lower(): + return index + return fallback + + +def choose_output_device(audio: pyaudio.PyAudio, preferred: str | None) -> int | None: + """Choose an output device by substring, otherwise first available output.""" + fallback = None + for index in range(audio.get_device_count()): + info = audio.get_device_info_by_index(index) + if info.get("maxOutputChannels", 0) <= 0: + continue + if fallback is None: + fallback = index + if preferred and preferred.lower() in str(info.get("name", "")).lower(): + return index + return fallback + + +def pick_omnivoice_device() -> str: + """Select the best local device supported by OmniVoice/PyTorch.""" + if torch.cuda.is_available(): + return "cuda:0" + if hasattr(torch.backends, "mps") and torch.backends.mps.is_available(): + return "mps" + return "cpu" + + +def pick_omnivoice_dtype(device: str) -> torch.dtype: + """Use float16 only where the backend reliably supports it.""" + return torch.float16 if device.startswith("cuda") else torch.float32 + + +class OmniVoiceSpeaker: + """Generate assistant responses with OmniVoice and play them locally.""" + + def __init__(self, instruct: str | None, output_device: str | None): + self.instruct = instruct + self.audio = pyaudio.PyAudio() + self.output_device_index = choose_output_device(self.audio, output_device) + if self.output_device_index is None: + self.audio.terminate() + raise RuntimeError("PortAudio nevidí žádné výstupní audio zařízení.") + + device = self.audio.get_device_info_by_index(self.output_device_index) + print(f"REPRODUKTOR: {device.get('name')}") + + ov_device = pick_omnivoice_device() + print(f"Načítám OmniVoice na {ov_device}...") + self.model = OmniVoice.from_pretrained( + "k2-fsa/OmniVoice", + device_map=ov_device, + dtype=pick_omnivoice_dtype(ov_device), + load_asr=False, + ) + + def speak(self, text: str) -> None: + """Generate Czech speech with OmniVoice and play it via PyAudio.""" + print(f"ASISTENT: {text}") + kwargs = {"text": text, "language": "Czech"} + if self.instruct: + kwargs["instruct"] = self.instruct + audio = self.model.generate(**kwargs) + samples = np.asarray(audio[0], dtype=np.float32).reshape(-1) + pcm = (np.clip(samples, -1.0, 1.0) * 32767).astype(np.int16).tobytes() + + stream = self.audio.open( + format=pyaudio.paInt16, + channels=1, + rate=TTS_SAMPLE_RATE, + output=True, + output_device_index=self.output_device_index, + ) + try: + stream.write(pcm) + finally: + stream.stop_stream() + stream.close() + + def close(self) -> None: + """Release PortAudio resources.""" + self.audio.terminate() + + +def start_microphone( + chunks: queue.Queue[bytes], + stop_event: threading.Event, + capture_enabled: threading.Event, + preferred_device: str | None, +) -> tuple[pyaudio.PyAudio, pyaudio.Stream]: + """Start a blocking PyAudio input stream in the current process.""" + audio = pyaudio.PyAudio() + device_index = choose_input_device(audio, preferred_device) + if device_index is None: + audio.terminate() + raise RuntimeError("PortAudio nevidí žádné vstupní audio zařízení.") + + device = audio.get_device_info_by_index(device_index) + print(f"MIKROFON: {device.get('name')}") + stream = audio.open( + format=FORMAT, + channels=CHANNELS, + rate=SAMPLE_RATE, + input=True, + input_device_index=device_index, + frames_per_buffer=CHUNK_SIZE, + ) + + def read_loop() -> None: + while not stop_event.is_set(): + try: + data = stream.read(CHUNK_SIZE, exception_on_overflow=False) + if capture_enabled.is_set(): + chunks.put(data) + except OSError as error: + print(f"Chyba mikrofonu: {error}") + stop_event.set() + + threading.Thread(target=read_loop, daemon=True).start() + return audio, stream + + +def drain_queue(chunks: queue.Queue[bytes]) -> None: + """Drop queued microphone audio.""" + while True: + try: + chunks.get_nowait() + except queue.Empty: + return + + +def speak_without_feedback( + speaker: OmniVoiceSpeaker, + text: str, + chunks: queue.Queue[bytes], + capture_enabled: threading.Event, +) -> None: + """Speak while microphone capture is paused, then clear echo residue.""" + capture_enabled.clear() + drain_queue(chunks) + speaker.speak(text) + time.sleep(0.25) + drain_queue(chunks) + capture_enabled.set() + + +async def send_audio( + websocket, chunks: queue.Queue[bytes], stop_event: threading.Event +): + """Send queued microphone chunks to WhisperFlow.""" + while not stop_event.is_set(): + try: + chunk = chunks.get(timeout=0.1) + except queue.Empty: + await asyncio.sleep(0.01) + continue + await websocket.send(chunk) + + +async def receive_transcripts( + websocket, dialog: ClinicDialog, speaker: OmniVoiceSpeaker +): + """Handle final transcript segments and speak dialog responses.""" + async for message in websocket: + event = json.loads(message) + text = event.get("data", {}).get("text", "").strip() + if not text: + continue + marker = "..." if event.get("is_partial") else "" + print(f"VY{marker}: {text}") + if not event.get("is_partial"): + response = dialog.handle(text) + speaker.speak(response) + if dialog.step == "done": + return + + +async def run(args: argparse.Namespace) -> None: + """Run the local voice assistant.""" + chunks: queue.Queue[bytes] = queue.Queue() + stop_event = threading.Event() + capture_enabled = threading.Event() + audio = None + stream = None + speaker = None + dialog = ClinicDialog(Path(args.database), caller_phone=args.caller_phone or "") + + def handle_signal(_signum, _frame): + stop_event.set() + + signal.signal(signal.SIGINT, handle_signal) + signal.signal(signal.SIGTERM, handle_signal) + + try: + speaker = OmniVoiceSpeaker(args.tts_instruct, args.output_device) + audio, stream = start_microphone( + chunks, stop_event, capture_enabled, args.device + ) + speak_without_feedback(speaker, dialog.greeting(), chunks, capture_enabled) + + async with websockets.connect(args.ws_url) as websocket: + sender = asyncio.create_task(send_audio(websocket, chunks, stop_event)) + receiver = asyncio.create_task( + receive_transcripts(websocket, dialog, speaker) + ) + done, pending = await asyncio.wait( + {sender, receiver}, return_when=asyncio.FIRST_COMPLETED + ) + for task in pending: + task.cancel() + for task in done: + task.result() + finally: + stop_event.set() + if stream: + stream.stop_stream() + stream.close() + if audio: + audio.terminate() + if speaker: + speaker.close() + + +def transcribe_audio(model, chunks: list[bytes]) -> str: + """Transcribe int16 PCM chunks using local Whisper.""" + samples = ( + np.frombuffer(b"".join(chunks), dtype=np.int16).astype(np.float32) / 32768.0 + ) + result = model.transcribe(samples, language="cs", fp16=False, temperature=0.0) + return result.get("text", "").strip() + + +# pylint: disable-next=too-many-locals,too-many-statements +def run_direct( + args: argparse.Namespace, +) -> None: + """Run without WebSocket: mic -> local Whisper -> local TTS.""" + chunks: queue.Queue[bytes] = queue.Queue() + stop_event = threading.Event() + capture_enabled = threading.Event() + audio = None + stream = None + speaker = None + dialog = ClinicDialog(Path(args.database), caller_phone=args.caller_phone or "") + speech_chunks: list[bytes] = [] + silent_chunks = 0 + has_speech = False + + def handle_signal(_signum, _frame): + stop_event.set() + + signal.signal(signal.SIGINT, handle_signal) + signal.signal(signal.SIGTERM, handle_signal) + + print(f"Načítám Whisper model {args.stt_model}...") + model = whisper.load_model(f"whisperflow/models/{args.stt_model}") + + try: + speaker = OmniVoiceSpeaker(args.tts_instruct, args.output_device) + audio, stream = start_microphone( + chunks, stop_event, capture_enabled, args.device + ) + speak_without_feedback(speaker, dialog.greeting(), chunks, capture_enabled) + print("Poslouchám. Mluvte česky, po větě udělejte krátkou pauzu.") + + while not stop_event.is_set() and dialog.step != "done": + try: + chunk = chunks.get(timeout=0.1) + except queue.Empty: + continue + + level = int(np.max(np.abs(np.frombuffer(chunk, dtype=np.int16)))) + if level > args.silence_threshold: + has_speech = True + silent_chunks = 0 + speech_chunks.append(chunk) + continue + + if has_speech: + silent_chunks += 1 + speech_chunks.append(chunk) + + if ( + has_speech + and silent_chunks >= args.silence_chunks + and len(speech_chunks) >= args.min_speech_chunks + ): + utterance = speech_chunks[:] + speech_chunks = [] + silent_chunks = 0 + has_speech = False + + text = transcribe_audio(model, utterance) + if not text: + continue + print(f"VY: {text}") + response = dialog.handle(text) + speak_without_feedback(speaker, response, chunks, capture_enabled) + + if dialog.appointments: + print(f"ULOŽENO: {dialog.appointments[-1]}") + finally: + stop_event.set() + if stream: + stream.stop_stream() + stream.close() + if audio: + audio.terminate() + if speaker: + speaker.close() + + +def parse_args() -> argparse.Namespace: + """Parse command-line arguments for the clinic assistant.""" + parser = argparse.ArgumentParser(description="Talk to the clinic assistant.") + parser.add_argument("--ws-url", default="ws://127.0.0.1:8181/ws") + parser.add_argument("--device", help="Substring of input device name") + parser.add_argument("--output-device", help="Substring of output device name") + parser.add_argument( + "--stt-model", + default="base.pt", + help="Whisper model file from whisperflow/models, e.g. tiny.pt or base.pt.", + ) + parser.add_argument( + "--tts-instruct", + default="female, moderate pitch", + help="OmniVoice instruction. Use supported items only.", + ) + parser.add_argument( + "--database", + default="clinic_appointments.db", + help="SQLite database for saved appointment requests.", + ) + parser.add_argument( + "--caller-phone", + default="", + help="Phone number to save with the appointment when known.", + ) + parser.add_argument("--websocket", action="store_true", help="Use WhisperFlow /ws") + parser.add_argument("--silence-threshold", type=int, default=SILENCE_THRESHOLD) + parser.add_argument("--min-speech-chunks", type=int, default=MIN_SPEECH_CHUNKS) + parser.add_argument("--silence-chunks", type=int, default=SILENCE_CHUNKS_TO_CLOSE) + return parser.parse_args() + + +if __name__ == "__main__": + parsed_args = parse_args() + if parsed_args.websocket: + asyncio.run(run(parsed_args)) + else: + run_direct(parsed_args) diff --git a/whisperflow/config.py b/whisperflow/config.py index 7cead39..f694692 100644 --- a/whisperflow/config.py +++ b/whisperflow/config.py @@ -26,6 +26,7 @@ def get_float(name: str, default: float) -> float: # model / transcription DEFAULT_MODEL = os.environ.get("WF_MODEL", "tiny.en.pt") +DEFAULT_LANGUAGE = os.environ.get("WF_LANGUAGE", "en") TRANSCRIBE_TIMEOUT = get_float("WF_TRANSCRIBE_TIMEOUT", 30.0) MAX_WINDOW_CHUNKS = get_int("WF_MAX_WINDOW_CHUNKS", 1000) diff --git a/whisperflow/fast_server.py b/whisperflow/fast_server.py index 2a18a1a..c39377c 100644 --- a/whisperflow/fast_server.py +++ b/whisperflow/fast_server.py @@ -1,8 +1,9 @@ """ fast api declaration """ +import asyncio import logging from typing import List, Optional -from contextlib import asynccontextmanager +from contextlib import asynccontextmanager, suppress from fastapi import ( FastAPI, @@ -27,9 +28,10 @@ async def stop_all_sessions(): """stop and drop every active session (used on shutdown)""" - for session in list(sessions.values()): - await session.stop() - sessions.clear() + for session_id, session in list(sessions.items()): + with suppress(asyncio.CancelledError): + await session.stop() + sessions.pop(session_id, None) @asynccontextmanager @@ -97,7 +99,9 @@ async def websocket_endpoint(websocket: WebSocket): session = None async def transcribe_async(chunks: list): - return await ts.transcribe_pcm_chunks_async(model, chunks) + return await ts.transcribe_pcm_chunks_async( + model, chunks, lang=config.DEFAULT_LANGUAGE + ) async def send_back_async(data: dict): await websocket.send_json(data) @@ -118,5 +122,8 @@ async def send_back_async(data: dict): await websocket.close() finally: if session: - await session.stop() - sessions.pop(session.id, None) + try: + with suppress(asyncio.CancelledError): + await session.stop() + finally: + sessions.pop(session.id, None)