66 lines
2.3 KiB
Python
66 lines
2.3 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
# -*- coding: utf-8 -*-
|
|||
|
|
import json
|
|||
|
|
import wave
|
|||
|
|
import io
|
|||
|
|
import os
|
|||
|
|
import logging
|
|||
|
|
from ..common_utils import decode_str2bytes
|
|||
|
|
|
|||
|
|
class STTBase:
|
|||
|
|
def __init__(self, RATE=16000, cfg_path=None, debug=False):
|
|||
|
|
self.RATE = RATE
|
|||
|
|
self.debug = debug
|
|||
|
|
self.asr_cfg = self.parse_json(cfg_path)
|
|||
|
|
|
|||
|
|
def parse_json(self, cfg_path):
|
|||
|
|
cfg = None
|
|||
|
|
self.hotwords = None
|
|||
|
|
if cfg_path is not None:
|
|||
|
|
with open(cfg_path, 'r', encoding='utf-8') as f:
|
|||
|
|
cfg = json.load(f)
|
|||
|
|
self.hotwords = cfg.get('hot_words', None)
|
|||
|
|
logging.info(f"load STT config file: {cfg_path}")
|
|||
|
|
logging.info(f"Hot words: {self.hotwords}")
|
|||
|
|
else:
|
|||
|
|
logging.warning("No STT config file provided, using default config.")
|
|||
|
|
return cfg
|
|||
|
|
|
|||
|
|
def add_hotword(self, hotword):
|
|||
|
|
"""add hotword to list"""
|
|||
|
|
if self.hotwords is None:
|
|||
|
|
self.hotwords = ""
|
|||
|
|
if isinstance(hotword, str):
|
|||
|
|
self.hotwords = self.hotwords + " " + "hotword"
|
|||
|
|
elif isinstance(hotword, (list, tuple)):
|
|||
|
|
# 将hotwords转换为str，并用空格隔开
|
|||
|
|
self.hotwords = self.hotwords + " " + " ".join(hotword)
|
|||
|
|
else:
|
|||
|
|
raise TypeError("hotword must be str or list")
|
|||
|
|
|
|||
|
|
def check_audio_type(self, audio_data):
|
|||
|
|
"""check audio data type and convert it to bytes if necessary."""
|
|||
|
|
if isinstance(audio_data, bytes):
|
|||
|
|
pass
|
|||
|
|
elif isinstance(audio_data, list):
|
|||
|
|
audio_data = b''.join(audio_data)
|
|||
|
|
elif isinstance(audio_data, str):
|
|||
|
|
audio_data = decode_str2bytes(audio_data)
|
|||
|
|
elif isinstance(audio_data, io.BytesIO):
|
|||
|
|
wf = wave.open(audio_data, 'rb')
|
|||
|
|
audio_data = wf.readframes(wf.getnframes())
|
|||
|
|
else:
|
|||
|
|
raise TypeError(f"audio_data must be bytes, str or io.BytesIO, but got {type(audio_data)}")
|
|||
|
|
return audio_data
|
|||
|
|
|
|||
|
|
def text_postprecess(self, result, data_id='text'):
|
|||
|
|
"""postprecess recognized result."""
|
|||
|
|
text = result[data_id]
|
|||
|
|
if isinstance(text, list):
|
|||
|
|
text = ''.join(text)
|
|||
|
|
return text.replace(' ', '')
|
|||
|
|
|
|||
|
|
def recognize(self, audio_data, queue=None):
|
|||
|
|
"""recognize audio data to text"""
|
|||
|
|
pass
|