RealtimeSTT Client, AudioInput, and Boundary Detector

from RealtimeSTT import (
    AudioToTextRecorderClient,
    AudioInput,
    RealtimeSpeechBoundaryDetector,
    SpeechBoundaryEvent,
    SpeechBoundaryResult,
)

from RealtimeSTT import AudioToTextRecorderClient

client = AudioToTextRecorderClient(
    model="small.en",
    language="en",
    control_url="ws://127.0.0.1:8011",
    data_url="ws://127.0.0.1:8012",
    autostart_server=True,
)

client.text(on_transcription_finished=None) -> str

client.feed_audio(chunk, audio_meta_data, original_sample_rate=16000)

from RealtimeSTT import AudioToTextRecorderClient

if __name__ == "__main__":
    # Start stt-server separately or set autostart_server=True
    with AudioToTextRecorderClient(
        model="small.en",
        language="en",
        enable_realtime_transcription=True,
        on_realtime_transcription_update=lambda t: print(f"\r{t}   ", end=""),
    ) as client:
        while True:
            text = client.text()
            if text:
                print(f"\n{text}")

from RealtimeSTT import AudioInput

audio = AudioInput(
    input_device_index=None,   # None → PyAudio default device
    debug_mode=False,
    target_samplerate=16000,
    chunk_size=1024,
    audio_format=pyaudio.paInt16,
    channels=1,
    resample_to_target=True,
)

audio.setup()                       # opens the stream; returns True on success
audio.read_chunk() -> bytes         # reads one chunk from the open stream
audio.list_devices()                # prints all input devices with sample rates
audio.cleanup()                     # stops and closes the stream, terminates PyAudio

from RealtimeSTT import AudioInput

audio = AudioInput(target_samplerate=16000, chunk_size=1024)
if audio.setup():
    for _ in range(100):
        chunk = audio.read_chunk()
        # process chunk ...
    audio.cleanup()

from RealtimeSTT import RealtimeSpeechBoundaryDetector

detector = RealtimeSpeechBoundaryDetector(
    sample_rate=16000,
    sensitivity=0.6,         # 0 = conservative, 1 = eager
)

result = detector.process_bytes(pcm_chunk: bytes) -> SpeechBoundaryResult
result = detector.process_samples(samples)        -> SpeechBoundaryResult
detector.reset()                                  # clears all history

result.boundary_detected   # bool — True if at least one event was emitted
result.events              # List[SpeechBoundaryEvent]
result.latest_event        # SpeechBoundaryEvent | None — most recent event
result.current_energy_db   # float — energy of the last processed frame (dB)
result.current_rms         # float — RMS amplitude of the last frame
result.noise_floor_db      # float — rolling noise floor estimate (dB)
result.is_speech           # bool — whether the last frame was classified as speech
result.is_vowel_like       # bool — whether the last frame was vowel-like
result.voicing_score       # float — voicing confidence [0, 1] of the last frame
result.processed_frames    # int — number of frames processed in this call

event.boundary_sample        # int   — sample index of the boundary
event.boundary_time_seconds  # float — boundary time in seconds from stream start
event.score                  # float — detection confidence score
event.reason                 # str   — "vowel-ended" or "vowel-to-pause"
event.energy_db              # float — frame energy at the boundary (dB)
event.noise_floor_db         # float — noise floor at detection time (dB)
event.drop_db                # float — energy drop from peak to boundary (dB)
event.valley_depth_db        # float — valley depth relative to surrounding frames (dB)
event.latency_ms             # float — detection latency in milliseconds
event.created_at             # float — Unix timestamp when the event was created
event.as_dict()              # -> Dict[str, float]  — plain metadata dict

from RealtimeSTT import RealtimeSpeechBoundaryDetector

detector = RealtimeSpeechBoundaryDetector(sample_rate=16000, sensitivity=0.6)

CHUNK_BYTES = 3200  # 100 ms of 16 kHz int16 mono

with open("speech.pcm", "rb") as f:
    while True:
        chunk = f.read(CHUNK_BYTES)
        if not chunk:
            break

        result = detector.process_bytes(chunk)
        if result.boundary_detected:
            for event in result.events:
                print(
                    f"Boundary at {event.boundary_time_seconds:.3f}s "
                    f"(score={event.score:.2f}, reason={event.reason!r}, "
                    f"latency={event.latency_ms:.1f}ms)"
                )

# Start the WebSocket STT server (requires RealtimeSTT_server)
stt-server --model small.en --language en --control_port 8011 --data_port 8012

# Connect the interactive CLI client to a running server
stt

# Build and install the Kroko-ONNX extension
stt-install-kroko

# Example: start a server with a larger model on a specific port
stt-server --model large-v2 --language en --control_port 9011 --data_port 9012 --debug