我一直在尝试让 STT 代码使用 HTTP 将其转录的文本发送到我的 Raspberry Pi 4 B 中。我在测试时遇到了一些代码错误，但后来，在修复所有错误后，我收到了消息 错误：500 序列化请求异常！运行代码时在 Pycharm 控制台中。树莓派没有报任何错误。
如果有帮助的话，我会在控制台中发布代码和文本，但我在网上找不到太多关于这个特定错误的信息
import queue
import re
import sys
import os
from google.cloud import speech
import pyaudio
import requests
# Set your Google Cloud Speech API credentials
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "C://Users//danim//OneDrive//Desktop//programming//workpls-401909-66721bf5ced8.json"
# Audio recording parameters
RATE = 48000
CHUNK = int(RATE / 10) # 100ms
class MicrophoneStream:
"""Opens a recording stream as a generator yielding the audio chunks."""
def __init__(self: object, rate: int = RATE, chunk: int = CHUNK) -> None:
"""The audio -- and generator -- is guaranteed to be on the main thread."""
self._rate = rate
self._chunk = chunk
# Create a thread-safe buffer of audio data
self._buff = queue.Queue()
self.closed = True
def __enter__(self: object) -> object:
self._audio_interface = pyaudio.PyAudio()
self._audio_stream = self._audio_interface.open(
format=pyaudio.paInt16,
channels=1,
rate=self._rate,
input=True,
frames_per_buffer=self._chunk,
stream_callback=self._fill_buffer,
)
self.closed = False
return self
def __exit__(
self: object,
type: object,
value: object,
traceback: object,
) -> None:
self._audio_stream.stop_stream()
self._audio_stream.close()
self.closed = True
self._buff.put(None)
self._audio_interface.terminate()
def _fill_buffer(
self: object,
in_data: object,
frame_count: int,
time_info: object,
status_flags: object,
) -> object:
self._buff.put(in_data)
return None, pyaudio.paContinue
def generator(self: object) -> object:
while not self.closed:
chunk = self._buff.get()
if chunk is None:
return
data = [chunk]
while True:
try:
chunk = self._buff.get(block=False)
if chunk is None:
return
data.append(chunk)
except queue.Empty:
break
yield b"".join(data)
def send_location_to_pi(location):
pi_url = 'http://<192.168.1.128>:8080/location' # Replace with your Raspberry Pi's IP
data = {'location': location}
response = requests.post(pi_url, json=data)
if response.status_code == 200:
print("Location sent to Pi successfully.")
else:
print("Failed to send location to Pi.")
def listen_print_loop(responses: object) -> str:
num_chars_printed = 0
for response in responses:
if not response.results:
continue
result = response.results[0]
if not result.alternatives:
continue
transcript = result.alternatives[0].transcript
overwrite_chars = " " * (num_chars_printed - len(transcript))
if not result.is_final:
sys.stdout.write(transcript + overwrite_chars + "\r")
sys.stdout.flush()
num_chars_printed = len(transcript)
else:
print(transcript + overwrite_chars)
if re.search(r"\b(exit|quit)\b", transcript, re.I):
print("Exiting..")
break
num_chars_printed = 0
# Send the recognized location to the Raspberry Pi
send_location_to_pi(transcript)
if __name__ == "__main__":
try:
audio_client = speech.SpeechClient()
config = speech.RecognitionConfig(
encoding=speech.RecognitionConfig.AudioEncoding.LINEAR16,
sample_rate_hertz=RATE,
language_code='en-US',
)
with MicrophoneStream(RATE, CHUNK) as stream:
audio_generator = stream.generator()
requests = (
stream.StreamingRecognizeRequest(audio_content=content, location=transcript)
for content in audio_generator
)
responses = audio_client.streaming_recognize(config=config, requests=requests)
listen_print_loop(responses)
except Exception as e:
print(f"Error: {e}")
回应： C:\Users\danim\PycharmProjects\EXAM env\Scripts\python.exe C:\Users\danim\OneDrive\Desktop\programming\STT.py 错误：500 序列化请求异常！
进程已完成，退出代码为 0
对于 Pi
from flask import Flask, request, jsonify
import requests
import time
app = Flask(__name__)
# Function to store the received location
current_location = 'Tarragona' # Set an initial location
@app.route('/location', methods=['POST'])
def receive_location():
data = request.get_json()
location = data.get('location')
if location:
# Store the received location for use in the weather retrieval
global current_location
current_location = location
return 'Location received and stored successfully', 200
else:
return 'Invalid location data', 400
# Function to get weather info for a location
def get_weather(location):
api_key = '<api key>' # Replace with your OpenWeatherMap API key
base_url = 'https://api.openweathermap.org/data/2.5/weather'
params = {'q': location, 'appid': api_key, 'units': 'metric'}
response = requests.get(base_url, params=params)
if response.status_code == 200:
data = response.json()
temperature = data['main']['temp']
description = data['weather'][0]['description']
response_text = f'The weather in {location} is {description} with a temperature of {temperature} degrees Celsius'
return response_text
else:
return 'Weather data not available'
if __name__ == '__main__':
app.run(host='0.0.0.0', port=8080)
while True:
location = current_location # Use the stored location
weather_info = get_weather(location)
print(weather_info)
time.sleep(10) # Adjust the interval as needed
回应： 服务 Flask 应用程序“”（延迟加载） 环境：生产 警告：这是一个开发服务器。不要在生产部署中使用它 使用生产 WSGI 服务器代替 调试模式：关闭 运行于 http://0.0.0.0:8080/ （按 CTRL+C 退出）
我在网上找不到任何有关此问题的信息，请提供任何帮助
我尝试使用 SSH，但是，1. 我不知道它是如何工作的，2. 无法让它工作
我的计划是让我能够大声说出一个城市的声音，让 STT 代码拾取我所说的内容，将其转录，而不是将其打印在 Pycharm 控制台中，而是将其发送到 Pi，文本将在其中存储替换为变量“位置”，然后 Pi 将运行天气代码并告诉我该城市的天气信息。后来，Pi 的响应告诉我天气详细信息将被发送回我的笔记本电脑，并被放入 TTS 代码中，该代码也可以工作。