Daily.co app for repeating what someone has said when they say "repeat".

daily-repeat.py

from google.cloud import speech
import argparse
import time
from daily import Daily, CallClient

SAMPLE_RATE = 44100
CHUNK_COUNT = 10
FRAME_COUNT = CHUNK_COUNT * SAMPLE_RATE // 100


class SendReceiveApp:
    def __init__(self):
        self.mic_device = Daily.create_microphone_device(
            "my-mic", sample_rate=SAMPLE_RATE, channels=1
        )
        self.speaker_device = Daily.create_speaker_device(
            "my-speaker", sample_rate=SAMPLE_RATE, channels=1
        )
        Daily.select_speaker_device("my-speaker")

        self.client = CallClient()
        self.client.update_inputs(
            {
                "camera": False,
                "microphone": {"isEnabled": True, "settings": {"deviceId": "my-mic"}},
            }
        )
        self.client.update_subscription_profiles(
            {"base": {"camera": "unsubscribed", "microphone": "subscribed"}}
        )

        self.is_recording = True
        self.recording = []

    def join(self, meeting_url):
        self.client.join(meeting_url)

    def leave(self):
        self.client.leave()

    def receive_audio(self):
        while True:
            buffer = self.speaker_device.read_frames(FRAME_COUNT)
            if len(buffer) > 0:
                if self.is_recording:
                    self.recording.append(buffer)
                yield buffer

    def send_recording(self):
        self.is_recording = False
        for frame in self.recording:
            self.mic_device.write_frames(frame)
        self.recording = []
        self.is_recording = True


def listen_print_loop(responses: object) -> str:
    for response in responses:
        if not response.results:
            continue
        result = response.results[0]
        if not result.alternatives:
            continue
        return result.alternatives[0].transcript


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("-m", "--meeting", required=True, help="Meeting URL")
    args = parser.parse_args()

    language_code = "en-US"  # a BCP-47 language tag

    client = speech.SpeechClient()
    config = speech.RecognitionConfig(
        encoding=speech.RecognitionConfig.AudioEncoding.LINEAR16,
        sample_rate_hertz=SAMPLE_RATE,
        language_code=language_code,
    )
    streaming_config = speech.StreamingRecognitionConfig(
        config=config, interim_results=True
    )

    Daily.init()

    app = SendReceiveApp()
    app.join(args.meeting)

    try:
        audio_generator = app.receive_audio()
        requests = (
            speech.StreamingRecognizeRequest(audio_content=content)
            for content in audio_generator
        )
        responses = client.streaming_recognize(streaming_config, requests)
        while True:
            transcript = listen_print_loop(responses)
            print(transcript)
            if "exit" in transcript.lower():
                print("exiting..")
                break
            if "repeat" in transcript.lower():
                app.send_recording()
    except KeyboardInterrupt:
        pass
    finally:
        app.leave()
        time.sleep(2)


if __name__ == "__main__":
    main()