options = ['1', '2', '3', '']

if array_of_options:

return options

eprint("Recording. Press option and ENTER (or just ENTER for default):")

eprint("1. transcript with openai API (default)")

eprint("2. transcript locally with whisper.cpp (with `--speed-up`)")

eprint("3. transcript locally with whisper.cpp")

max_rms = 0

min_rms = 1e5

pct = 0

threshold = 0.15

def __init__(self):

if sf is None:

eprint("The soundfile module is not available. Please install it using 'pip install soundfile'.", file=sys.stderr)

raise SoundDeviceError("The soundfile module is required but not installed.")

try:

import sounddevice as sd

if not args.silent:

eprint("Initializing sound device...")

self.sd = sd

except Exception as e:

eprint(f"An error occurred while initializing the sound device: {e}")

raise SoundDeviceError("Failed to initialize the sound device.")

def callback(self, indata, frames, time, status):

"""This is called (from a separate thread) for each audio block."""

rms = np.sqrt(np.mean(indata**2))

self.max_rms = max(self.max_rms, rms)

self.min_rms = min(self.min_rms, rms)

rng = self.max_rms - self.min_rms

if rng > 0.001:

self.pct = (rms - self.min_rms) / rng

else:

self.pct = 0.5

self.q.put(indata.copy())

def get_prompt(self):

num = 10

if np.isnan(self.pct) or self.pct < self.threshold:

cnt = 0

else:

cnt = int(self.pct * 10)

#bar = "░" * cnt + "█" * (num - cnt)

bar = "#" * cnt + "_" * (num - cnt)

bar = bar[:num]

dur = time.time() - self.start_time

return f"Recording. Press option and ENTER (or just ENTER for default)... {dur:.1f}sec {bar}"

# This block is responsible for handling the KeyboardInterrupt exception

# which is triggered by pressing [Ctrl]+[C]. When caught, it simply returns

# from the function without proceeding to the transcription step.

def raw_record_only(self):

self.q = queue.Queue()

filename = tempfile.mktemp(suffix=".wav")

sample_rate = 16000 # 16kHz

self.start_time = time.time()

with self.sd.InputStream(samplerate=sample_rate, channels=1, callback=self.callback):

if not args.silent:

choice = session.prompt(self.get_prompt, refresh_interval=0.1)

else:

choice = input("") # silently wait for ENTER in silent mode.

# Only proceed with transcription if there are audio frames in the queue

if not self.q.empty():

with sf.SoundFile(filename, mode="x", samplerate=sample_rate, channels=1) as file:

while not self.q.empty():

file.write(self.q.get())

else:

eprint("No audio recorded.")

return choice, filename

def transcribe_with_openai_api(self, filename):

with open(filename, "rb") as fh:

transcript = client.audio.transcriptions.create(model="whisper-1", file=fh)

transcript_text = transcript.text

return transcript_text

def run_whisper_cpp_in_temp_dir(self, filename, extra_flags=[]):

try:

with tempfile.TemporaryDirectory() as temp_dir:

base = os.path.splitext(os.path.basename(filename))[0]

temp_filename = os.path.join(temp_dir, f"{base}.wav")

shutil.copyfile(filename, temp_filename)

eprint(f"Processing {temp_filename} with whisper.cpp in {temp_dir}")

eprint(f"Base filename: {base}")

command = ["/usr/bin/time", f"--output={base}.wav.time", "whisper.cpp", "--model", "/usr/share/whisper.cpp-model-large/large.bin"] + extra_flags + ["-otxt", "-ovtt", "-osrt", "-ocsv", temp_filename]

eprint(f"Running command: {' '.join(command)}")

result = subprocess.run(command, cwd=temp_dir)

eprint(f"Command result: {result}")

with open(f"{temp_dir}/{base}.wav.time", 'r') as f:

print(f.read())

with open(f"{temp_dir}/{base}.wav.txt", 'r') as f:

transcript_text = f.read()

finally:

shutil.rmtree(temp_dir, ignore_errors=True)

return transcript_text

def transcribe_with_whisper_cpp(self, filename, extra_flags=[]):

if arg_should_not_wait:

return

transcript_trimmed = transcript.strip()

eprint(f"As local transcripts take long, let me wait for you confirming with 'Enter' before exiting. Transcript:\n{transcript_trimmed}")