GeezerVisonThe World Wide Web, restored to 800 pixels of glory. |
| Home|About|Gaming cabinet|Archives|How it works|Starter sites |
GeezerVison is showing github.com — Vapi-python/vapi_python/daily_call.py at main · Adam-s-tech/Vapi-python Expand file tree/ Copy pathdaily_call.pyMore file actions 175 lines (140 loc) · 5.29 KB / Copy pathdaily_call.pyFile metadata and controls175 lines (140 loc) · 5.29 KB 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 import daily import threading import pyaudio import json SAMPLE_RATE = 16000 NUM_CHANNELS = 1 CHUNK_SIZE = 640 def is_playable_speaker(participant): is_speaker = "userName" in participant["info"] and participant["info"]["userName"] == "Vapi Speaker" mic = participant["media"]["microphone"] is_subscribed = mic["subscribed"] == "subscribed" is_playable = mic["state"] == "playable" return is_speaker and is_subscribed and is_playable class DailyCall(daily.EventHandler): def __init__(self): daily.Daily.init() self.__audio_interface = pyaudio.PyAudio() self.__input_audio_stream = self.__audio_interface.open( format=pyaudio.paInt16, channels=NUM_CHANNELS, rate=SAMPLE_RATE, input=True, frames_per_buffer=CHUNK_SIZE, ) self.__output_audio_stream = self.__audio_interface.open( format=pyaudio.paInt16, channels=NUM_CHANNELS, rate=SAMPLE_RATE, output=True, frames_per_buffer=CHUNK_SIZE ) self.__mic_device = daily.Daily.create_microphone_device( "my-mic", sample_rate=SAMPLE_RATE, channels=NUM_CHANNELS ) self.__speaker_device = daily.Daily.create_speaker_device( "my-speaker", sample_rate=SAMPLE_RATE, channels=NUM_CHANNELS ) daily.Daily.select_speaker_device("my-speaker") self.__call_client = daily.CallClient(event_handler=self) self.__call_client.update_inputs({ "camera": False, "microphone": { "isEnabled": True, "settings": { "deviceId": "my-mic", "customConstraints": { "autoGainControl": {"exact": True}, "noiseSuppression": {"exact": True}, "echoCancellation": {"exact": True}, } } } }) self.__call_client.update_subscription_profiles({ "base": { "camera": "unsubscribed", "microphone": "subscribed" } }) self.__participants = dict(self.__call_client.participants()) del self.__participants["local"] self.__app_quit = False self.__app_error = None self.__app_joined = False self.__app_inputs_updated = False self.__start_event = threading.Event() self.__receive_bot_audio_thread = threading.Thread( target=self.receive_bot_audio) self.__send_user_audio_thread = threading.Thread( target=self.send_user_audio) self.__receive_bot_audio_thread.start() self.__send_user_audio_thread.start() def on_inputs_updated(self, inputs): self.__app_inputs_updated = True self.maybe_start() def on_joined(self, data, error): if error: print(f"Unable to join call: {error}") self.__app_error = error else: self.__app_joined = True print("Joined call!") self.maybe_start() def on_participant_joined(self, participant): self.__participants[participant["id"]] = participant def on_participant_left(self, participant, _): del self.__participants[participant["id"]] self.leave() def on_participant_updated(self, participant): self.__participants[participant["id"]] = participant if is_playable_speaker(participant): self.__call_client.send_app_message("playable") def join(self, meeting_url): self.__call_client.join(meeting_url, completion=self.on_joined) def leave(self): self.__app_quit = True self.__receive_bot_audio_thread.join() self.__send_user_audio_thread.join() self.__call_client.leave() def maybe_start(self): if self.__app_error: self.__start_event.set() if self.__app_inputs_updated and self.__app_joined: self.__start_event.set() def send_user_audio(self): self.__start_event.wait() if self.__app_error: print(f"Unable to receive mic audio!") return while not self.__app_quit: buffer = self.__input_audio_stream.read( CHUNK_SIZE, exception_on_overflow=False) if len(buffer) > 0: try: self.__mic_device.write_frames(buffer) except Exception as e: print(e) def receive_bot_audio(self): self.__start_event.wait() if self.__app_error: print(f"Unable to receive bot audio!") return while not self.__app_quit: buffer = self.__speaker_device.read_frames(CHUNK_SIZE) if len(buffer) > 0: self.__output_audio_stream.write(buffer, CHUNK_SIZE) def send_app_message(self, message): """ Send an application message to the assistant. :param message: The message to send (expects a dictionary). """ try: serialized_message = json.dumps(message) self.__call_client.send_app_message(serialized_message) except Exception as e: print(f"Failed to send app message: {e}") |
| You are visitor 01402520 — GeezerVison does not store the pages you look up. Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required. |