import os import logging import rpyc from functools import lru_cache import typer from pathlib import Path app = typer.Typer() logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) ASR_HOST = os.environ.get("JASPER_ASR_RPYC_HOST", "localhost") ASR_PORT = int(os.environ.get("JASPER_ASR_RPYC_PORT", "8045")) @lru_cache() def transcribe_gen(asr_host=ASR_HOST, asr_port=ASR_PORT): logger.info(f"connecting to asr server at {asr_host}:{asr_port}") asr = rpyc.connect(asr_host, asr_port).root logger.info(f"connected to asr server successfully") return asr.transcribe @app.command() def transcribe_file(audio_file: Path): from pydub import AudioSegment transcriber = transcribe_gen() aud_seg = ( AudioSegment.from_file_using_temporary_files(audio_file) .set_channels(1) .set_sample_width(2) .set_frame_rate(24000) ) tscript_file_path = audio_file.with_suffix(".txt") transcription = transcriber(aud_seg.raw_data) with open(tscript_file_path, "w") as tf: tf.write(transcription) def main(): app() if __name__ == "__main__": main()