I am developing a solution that is a voice chatbot using a Raspberry pi 4 and aws lex due to having many if condition and many stuff to check It takes a little bit of time to reach the recording part that will capture the user's voice
which is sometimes not convenient since the user will speak but it didn't reach the record part
if loadingVid.poll() is None:
os.killpg(os.getpgid(loadingVid.pid), signal.SIGTERM)
this is the part that I think needs to be improved on that I need to play videos in the background to know that his voice has been processed and I have to check if the video is on or not to kill it when the answer has arrived from lex
Also, The video is played by omxplayer which is a player that is specially designed for The Raspberry pi to encode the video on the GPU directly I am using Subprocess to open the video but I can't use it to terminate I don't know the reason but this method work
I think that I need to make the checking in a separate thread or something but this will complicate the whole code
Also, I am currently using 2 threads for hot word detection and motion detection
This is how I am recording
def record():
"""Record audio from the microphone"""
os.system( "sox -d -t wavpcm -c 1 -b 16 -r 16000 -e signed-integer --endian little - silence 1 0 1% 5 0.8t 5% -highpass 300> request.wav" )
This the main function to get better idea about the scirpit and how things work and if you would have other suggestion to improve on
I am not looking for an answer I am looking to guidance how to approach this problem and what can I improve on and what to look out for
If there is somthing not claer Please feel free to ask me I will answer
def main():
"""
Main function:
1. Load environment variables and start idle video.
2. Initialize Amazon Lex runtime client.
3. Set max waiting time, current session id, and last response to initial values.
4. Enter an infinite loop:
5. If the last response's session state's intent state is "Fulfilled" or "Failed":
a. Wait for a hot word and return the time elapsed waiting for it.
b. If the idle time duration is greater than the max waiting time, start a new session and play a greeting video. Otherwise, play a confirmation video.
6. If the last response is None:
a. Wait for a hot word and return the time elapsed waiting for it.
b. Start a new session and play a greeting video.
7. Stop the listening video and start a loading video.
8. Record audio and send it to the Amazon Lex runtime to get a response.
9. Handle the response by playing an audio file, displaying an image, or playing a video.
10. Stop the loading video and start the idle video again.
"""
dotenv.load_dotenv()
hologram.minimze()
hologram.hide_cursor()
# hologram.bench_idle()
idle = hologram.play_idle("/home/alexa/Videos/menu.mp4", 5)
lexruntimev2 = boto3.client(
"lexv2-runtime",
aws_access_key_id=os.environ.get("aws_access_key_id"),
aws_secret_access_key=os.environ.get("aws_secret_access_key"),
region_name="us-east-1",
)
maxWaitingTime = 30.0
currentSessionId = None
last_response = None
current_process = None
listeningVid = None
intermediate_vid = False
while True:
if last_response and (
last_response["sessionState"]["intent"]["state"] == "Fulfilled"
or last_response["sessionState"]["intent"]["state"] == "Failed"
):
if idle.poll() is not None:
idle = hologram.play_idle("/home/alexa/Videos/menu.mp4", 5)
# print(last_response)
# wait for hot word and return time elased waiting for it
idleTimeDuration = triggers.wait_for_triggers()
# print(idleTimeDuration)
if idleTimeDuration > maxWaitingTime:
currentSessionId = lex.newSession()
greeting_vid = lex.say_greeting()
else:
listeningVid = lex.say_confirm_listening()
elif last_response == None:
triggers.wait_for_triggers()
currentSessionId = lex.newSession()
lex.say_greeting()
# if this is a follow-up question (slot elicitation)
else:
# listeningVid = lex.say_confirm_listening()
listeningVid= hologram.play_idle("/home/alexa/project/video/speaking.mp4",8)
lex.record()
# say one moment please
loadingVid = lex.say_one_moment()
# if the idle vid is still running (return value is still none)
if idle.poll() is None:
# then kill it
os.killpg(os.getpgid(idle.pid), signal.SIGTERM)
if listeningVid is not None:
# then kill the listening
if listeningVid.poll() is None:
os.killpg(os.getpgid(listeningVid.pid), signal.SIGTERM)
# listeningVid.kill()
response, responseAssetURL = lex.recognize_audio(lexruntimev2, currentSessionId)
last_response = response
video_case_handling = True
Image_case_handling = True
if responseAssetURL != None:
if ".png" not in responseAssetURL.lower():
audio.play_audio("audio/lex_response.mpeg")
video_case_handling = False
if ".png" in responseAssetURL.lower():
Image_case_handling = False
feh = hologram.display_image(
responseAssetURL, "/home/alexa/project/images/image.png"
)
# hologram.play_with_mpv(responseAssetURL)
if Image_case_handling:
hologram.play_with_omx(responseAssetURL, 9)
# hologram.play_idle()
# time.sleep(5)
# loadingVid.terminate()
# if the loading vid is still running (return value is still none)
if loadingVid.poll() is None:
# then kill it
os.killpg(os.getpgid(loadingVid.pid), signal.SIGTERM)
if responseAssetURL == None:
intermediate_vid = True
intermediate = hologram.play_idle(
"/home/alexa/project/video/speaking.mp4", 8
)
if video_case_handling:
audio.play_audio("audio/lex_response.mpeg")
if intermediate_vid:
os.killpg(os.getpgid(intermediate.pid), signal.SIGTERM)
intermediate_vid = False
if Image_case_handling == False:
idle = hologram.play_idle("/home/alexa/Videos/menu.mp4", 5)
os.system("pkill feh")
I tried doing multiprocessing but I think it won't help tried also reducing the if the condition that I used to have and simplify what is needed
CodePudding user response：
After my search and looking into I found couple things that will helped me speed up
Wrap the code as a class and have states and each state will have lock using the thread module
Change some of the operators instead of == use is which can lead to better performance specially that I have multiple of them