SSL: 使用Watson流式语音转文本时出现CERTIFICATE_VERIFY_FAILED错误

我在YouTube上观看了一个教程,讲解如何进行实时语音转文本,教程中建议克隆这个仓库。我对代码稍作修改,将APIKEYurl直接嵌入代码中,不再需要.cfg文件。但我遇到了以下错误:[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1123)

这是我的代码:

## Copyright 2016 IBM## Licensed under the Apache License, Version 2.0 (the "License"); you may# not use this file except in compliance with the License. You may obtain# a copy of the License at##    http://www.apache.org/licenses/LICENSE-2.0## Unless required by applicable law or agreed to in writing, software# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the# License for the specific language governing permissions and limitations# under the License.import argparseimport base64import jsonimport threadingimport timeimport pyaudioimport websocketfrom websocket._abnf import ABNFCHUNK = 1024FORMAT = pyaudio.paInt16# Even if your default input is multi channel (like a webcam mic),# it's really important to only record 1 channel, as the STT service# does not do anything useful with stereo. You get a lot of "hmmm"# back.CHANNELS = 1# Rate is important, nothing works without it. This is a pretty# standard default. If you have an audio device that requires# something different, change this.RATE = 44100RECORD_SECONDS = 5FINALS = []LAST = NoneREGION_MAP = {    'us-east': 'gateway-wdc.watsonplatform.net',    'us-south': 'stream.watsonplatform.net',    'eu-gb': 'stream.watsonplatform.net',    'eu-de': 'stream-fra.watsonplatform.net',    'au-syd': 'gateway-syd.watsonplatform.net',    'jp-tok': 'gateway-syd.watsonplatform.net',}def read_audio(ws, timeout):    """Read audio and sent it to the websocket port.    This uses pyaudio to read from a device in chunks and send these    over the websocket wire.    """    global RATE    p = pyaudio.PyAudio()    # NOTE(sdague): if you don't seem to be getting anything off of    # this you might need to specify:    #    #    input_device_index=N,    #    # Where N is an int. You'll need to do a dump of your input    # devices to figure out which one you want.    RATE = int(p.get_default_input_device_info()['defaultSampleRate'])    stream = p.open(format=FORMAT,                    channels=CHANNELS,                    rate=RATE,                    input=True,                    frames_per_buffer=CHUNK)    print("* recording")    rec = timeout or RECORD_SECONDS    for _ in range(0, int(RATE / CHUNK * rec)):        data = stream.read(CHUNK)        # print("Sending packet... %d" % i)        # NOTE(sdague): we're sending raw binary in the stream, we        # need to indicate that otherwise the stream service        # interprets this as text control messages.        ws.send(data, ABNF.OPCODE_BINARY)    # Disconnect the audio stream    stream.stop_stream()    stream.close()    print("* done recording")    # In order to get a final response from STT we send a stop, this    # will force a final=True return message.    data = {"action": "stop"}    ws.send(json.dumps(data).encode('utf8'))    # ... which we need to wait for before we shutdown the websocket    time.sleep(1)    ws.close()    # ... and kill the audio device    p.terminate()def on_message(self, msg):    """Print whatever messages come in.    While we are processing any non trivial stream of speech Watson    will start chunking results into bits of transcripts that it    considers "final", and start on a new stretch. It's not always    clear why it does this. However, it means that as we are    processing text, any time we see a final chunk, we need to save it    off for later.    """    global LAST    data = json.loads(msg)    if "results" in data:        if data["results"][0]["final"]:            FINALS.append(data)            LAST = None        else:            LAST = data        # This prints out the current fragment that we are working on        print(data['results'][0]['alternatives'][0]['transcript'])def on_error(self, error):    """Print any errors."""    print(error)def on_close(ws):    """Upon close, print the complete and final transcript."""    global LAST    if LAST:        FINALS.append(LAST)    transcript = "".join([x['results'][0]['alternatives'][0]['transcript']                          for x in FINALS])    print(transcript)def on_open(ws):    """Triggered as soon a we have an active connection."""    args = ws.args    data = {        "action": "start",        # this means we get to send it straight raw sampling        "content-type": "audio/l16;rate=%d" % RATE,        "continuous": True,        "interim_results": True,        # "inactivity_timeout": 5, # in order to use this effectively        # you need other tests to handle what happens if the socket is        # closed by the server.        "word_confidence": True,        "timestamps": True,        "max_alternatives": 3    }    # Send the initial control message which sets expectations for the    # binary stream that follows:    ws.send(json.dumps(data).encode('utf8'))    # Spin off a dedicated thread where we are going to read and    # stream out audio.    threading.Thread(target=read_audio,                     args=(ws, args.timeout)).start()def get_url():    # See    # https://console.bluemix.net/docs/services/speech-to-text/websockets.html#websockets    # for details on which endpoints are for each region.    region = 'eu-gb'    host = REGION_MAP[region]    return ("wss://{}/speech-to-text/api/v1/recognize"           "?model=en-US_BroadbandModel").format(host)def get_auth():    return ("apikey", 'XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX')def parse_args():    parser = argparse.ArgumentParser(        description='Transcribe Watson text in real time')    parser.add_argument('-t', '--timeout', type=int, default=5)    # parser.add_argument('-d', '--device')    # parser.add_argument('-v', '--verbose', action='store_true')    args = parser.parse_args()    return argsdef main():    # Connect to websocket interfaces    headers = {}    userpass = ":".join(get_auth())    headers["Authorization"] = "Basic " + base64.b64encode(        userpass.encode()).decode()    url = get_url()    # If you really want to see everything going across the wire,    # uncomment this. However realize the trace is going to also do    # things like dump the binary sound packets in text in the    # console.    #    # websocket.enableTrace(True)    ws = websocket.WebSocketApp(url,                                header=headers,                                on_message=on_message,                                on_error=on_error,                                on_close=on_close)    ws.on_open = on_open    ws.args = parse_args()    # This gives control over the WebSocketApp. This is a blocking    # call, so it won't return until the ws.close() gets called (after    # 6 seconds in the dedicated thread).    ws.run_forever()if __name__ == "__main__":    main()

顺便说一下,我使用的是Mac电脑。


回答:

我找到了解决方法。我只需要运行

/Applications/Python\ 3.9/Install\ Certificates.command

在Mac上,然后更改URL,因为代码已经过时了。

原来的URL是:

stream.watsonplatform.net/speech-to-text/api/v1/recognize

我改成了:

api.eu-gb.text-to-speech.watson.cloud.ibm.com/instances/xxxxxxxxxx/v1/synthesize 

新的URL可以在您的IBM Cloud仪表板中找到。

Related Posts

L1-L2正则化的不同系数

我想对网络的权重同时应用L1和L2正则化。然而,我找不…

使用scikit-learn的无监督方法将列表分类成不同组别,有没有办法?

我有一系列实例,每个实例都有一份列表,代表它所遵循的不…

f1_score metric in lightgbm

我想使用自定义指标f1_score来训练一个lgb模型…

通过相关系数矩阵进行特征选择

我在测试不同的算法时,如逻辑回归、高斯朴素贝叶斯、随机…

可以将机器学习库用于流式输入和输出吗?

已关闭。此问题需要更加聚焦。目前不接受回答。 想要改进…

在TensorFlow中,queue.dequeue_up_to()方法的用途是什么?

我对这个方法感到非常困惑,特别是当我发现这个令人费解的…

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注