環境配置:
? ? ? ? 電腦端:?ubuntu22.04實體機作為上位機
? ? ????ROS版本:ros2-humble
? ??????實體機器人: STM32 + 思嵐A1激光雷達
? ? ? ? 科大訊飛語音SDK?訊飛開放平臺-以語音交互為核心的人工智能開放平臺
實現步驟:
? ? ? ? 1. 下載和處理科大訊飛語音模塊
? ? ? ? (1)進入官網的控制臺
? ? ? ?
?(2)在左側導航欄中選擇 語音識別-> 語音聽寫
? ? ? ? (3)下載語音模塊
?
?2.科大訊飛SDK的處理
新建一個工作空間,里面新建兩個文件夾 src? ?voice_ros2
將SDK壓縮包解壓后的文件,放入voice_ros2中,進入sample目錄的iat_online_record_sample目錄下,運行下面的命令
source 64bit_make.sh
在bin目錄下執行對應的可執行文件了?
./iat_online_record_sample
?
?如果遇到下列問題:error while loading shared libraries: libmsc.so: cannot open shared object file: No such file or directory
就把在終端中進入下列目錄中
?執行命令:
sudo cp libmsc.so /usr/local/lib
sudo ldconfig
3.上位機實現
?
src 文件夾中放的是 兩個功能包,base 中是stm32的ROS2驅動包,teleop_twist_keyboard是github上下載的鍵盤控制節點功能包,地址如下:
GitHub - ros2/teleop_twist_keyboard at ardent
這個目錄下的文件是SDK解壓后的文件,其中 紅框中的voice.py是也單獨編寫的文件
import subprocess
import multiprocessing
import timedef run_iat_online_record_sample(queue):process = subprocess.Popen(["./bin/iat_online_record_sample"], stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.PIPE, )# Communicate with the processstdout, _ = process.communicate(input=b"0\n1\n")# Put the result into the queuequeue.put(stdout.decode('utf-8'))def main():while True:# Create a queue for communication between processesqueue = multiprocessing.Queue()# Start the processprocess = multiprocessing.Process(target=run_iat_online_record_sample, args=(queue,))process.start()# Wait for the process to finish and get the result from the queueprocess.join()result = queue.get()# Print the resultprint("Result:", result)# Save the result to a text file, clearing the file firstwith open("result.txt", "w") as f:f.write(result)# Ask user whether to continue recognitioncontinue_recognition = input("是否繼續識別? (0: 結束, 1: 繼續): ")if continue_recognition == "0":breakif __name__ == "__main__":main()
這個文件運行后會在當前目錄生成一個result.txt文件,如下圖,這個文件的內容每次識別之后都會更新,鍵盤節點就是通過獲取這個文件的數據來通過語音控制機器人移動的
4.修改teleop_twist_keyboard.py文件
在鍵盤控制的代碼前添加讀取文件數據的代碼
這里將剛剛識別到的語音過濾后存儲在voice_command[0]中,以供后續使用,下面會通過判斷voice_command[0]中的值來進行不同的操作
import sys
import threading
import time
import os
from std_msgs.msg import String
import geometry_msgs.msg
import rclpyif sys.platform == 'win32':import msvcrt
else:import termiosimport ttymsg = """
This node takes keypresses from the keyboard and publishes them
as Twist/TwistStamped messages. It works best with a US keyboard layout.
---------------------------
Moving around:u i oj k lm , .For Holonomic mode (strafing), hold down the shift key:
---------------------------U I OJ K LM < >t : up (+z)
b : down (-z)anything else : stopq/z : increase/decrease max speeds by 10%
w/x : increase/decrease only linear speed by 10%
e/c : increase/decrease only angular speed by 10%CTRL-C to quit
"""moveBindings = {'i': (1, 0, 0, 0),'o': (1, 0, 0, -1),'j': (0, 0, 0, 1),'l': (0, 0, 0, -1),'u': (1, 0, 0, 1),',': (-1, 0, 0, 0),'.': (-1, 0, 0, 1),'m': (-1, 0, 0, -1),'O': (1, -1, 0, 0),'I': (1, 0, 0, 0),'J': (0, 1, 0, 0),'L': (0, -1, 0, 0),'U': (1, 1, 0, 0),'<': (-1, 0, 0, 0),'>': (-1, -1, 0, 0),'M': (-1, 1, 0, 0),'t': (0, 0, 1, 0),'b': (0, 0, -1, 0),
}speedBindings = {'q': (1.1, 1.1),'z': (.9, .9),'w': (1.1, 1),'x': (.9, 1),'e': (1, 1.1),'c': (1, .9),
}def getKey(settings):if sys.platform == 'win32':# getwch() returns a string on Windowskey = msvcrt.getwch()else:tty.setraw(sys.stdin.fileno())# sys.stdin.read() returns a string on Linuxkey = sys.stdin.read(1)termios.tcsetattr(sys.stdin, termios.TCSADRAIN, settings)return keydef saveTerminalSettings():if sys.platform == 'win32':return Nonereturn termios.tcgetattr(sys.stdin)def restoreTerminalSettings(old_settings):if sys.platform == 'win32':returntermios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)def vels(speed, turn):return 'currently:\tspeed %s\tturn %s ' % (speed, turn)def main():settings = saveTerminalSettings()rclpy.init()node = rclpy.create_node('teleop_twist_keyboard')# parametersstamped = node.declare_parameter('stamped', False).valueframe_id = node.declare_parameter('frame_id', '').valueif not stamped and frame_id:raise Exception("'frame_id' can only be set when 'stamped' is True")if stamped:TwistMsg = geometry_msgs.msg.TwistStampedelse:TwistMsg = geometry_msgs.msg.Twistpub = node.create_publisher(TwistMsg, 'cmd_vel', 10)voice_command = [None] # Initializing as a listspinner = threading.Thread(target=rclpy.spin, args=(node,))spinner.start()speed = 0.5turn = 1.0x = 0.0y = 0.0z = 0.0th = 0.0status = 0.0twist_msg = TwistMsg()if stamped:twist = twist_msg.twisttwist_msg.header.stamp = node.get_clock().now().to_msg()twist_msg.header.frame_id = frame_idelse:twist = twist_msgtry:print(msg)print(vels(speed, turn))while True:print("當前工作路徑:", os.getcwd())with open('./voice_ros2/result.txt', 'r') as f:# with open('/home/lsg/xufen3_ws/voice_ros2/result.txt', 'r') as f:for line in f:if line.startswith('Result: ['):start = line.find('[')end = line.find(']')if start != -1 and end != -1:voice_command[0] = line[start + 1:end].strip()print("voice_command", voice_command[0])# Clearing the content of result.txtopen('./voice_ros2/result.txt', 'w').close()# open('/home/lsg/xufen3_ws/voice_ros2/result.txt', 'w').close()breakkey = getKey(settings)# print("鍵盤控制按鍵輸出", key)if key in moveBindings.keys():x = moveBindings[key][0]y = moveBindings[key][1]z = moveBindings[key][2]th = moveBindings[key][3]elif key in speedBindings.keys():speed = speed * speedBindings[key][0]turn = turn * speedBindings[key][1]print(vels(speed, turn))if (status == 14):print(msg)status = (status + 1) % 15elif voice_command[0] is not None:if voice_command[0] == "小車后退":print("語音控制小車前進", voice_command[0])x = moveBindings['i'][0]y = moveBindings['i'][1]z = moveBindings['i'][2]th = moveBindings['i'][3]elif voice_command[0] == "小車前進":print("語音控制小車后退", voice_command[0])x = moveBindings[','][0]y = moveBindings[','][1]z = moveBindings[','][2]th = moveBindings[','][3]elif voice_command[0] == "小車左轉":print("語音控制小車左轉", voice_command[0])x = moveBindings['j'][0]y = moveBindings['j'][1]z = moveBindings['j'][2]th = moveBindings['j'][3]elif voice_command[0] == "小車右轉":print("語音控制小車右轉", voice_command[0])x = moveBindings['l'][0]y = moveBindings['l'][1]z = moveBindings['l'][2]th = moveBindings['l'][3]elif voice_command[0] == "小車停":print("語音控制小車停", voice_command[0])x = moveBindings['k'][0]y = moveBindings['k'][1]z = moveBindings['k'][2]th = moveBindings['k'][3]voice_command[0] = Noneelse:x = 0.0y = 0.0z = 0.0th = 0.0if (key == '\x03'):breakif stamped:twist_msg.header.stamp = node.get_clock().now().to_msg()twist.linear.x = x * speedtwist.linear.y = y * speedtwist.linear.z = z * speedtwist.angular.x = 0.0twist.angular.y = 0.0twist.angular.z = th * turnpub.publish(twist_msg)# Print timestamp every secondtime.sleep(1)print("時間戳:", time.time())except Exception as e:print(e)finally:if stamped:twist_msg.header.stamp = node.get_clock().now().to_msg()twist.linear.x = 0.0twist.linear.y = 0.0twist.linear.z = 0.0twist.angular.x = 0.0twist.angular.y = 0.0twist.angular.z = 0.0pub.publish(twist_msg)rclpy.shutdown()spinner.join()restoreTerminalSettings(settings)if __name__ == '__main__':main()
5. 編譯運行
// xufen3_ws工作空間下
// 終端1:
colcon build. install/setup.bashros2 launch ros2_stm32_bridge driver.launch.py// 終端2:
. install/setup.bashros2 run teleop_twist_keyboard teleop_twist_keyboard// 終端3 ~/xufen3_ws/voice_ros2$ 目錄下 :python3 voice.py
然后就可以通過語音控制小車
在右側終端按1進行語音識別,此時將識別到小車前進的命令并打印,在左側終端按回車健獲取result中的命令,將輸出voice_command 小車前進,此時再按鍵ctrl+c,將輸出語音控制小車前進 小車前進并且小車開始移動。
目前的代碼需要按鍵才能加載進來語音的命令并控制小車移動,但好在實現了功能,后續還會繼續優化。
?
終端3中,輸入數字1? ? 然后 語音輸入指令 “小車前進” 或“? 小車后退”? 或 “小車左轉” 或“”小車右轉”?
等到終端3中,打印了語音指令后,鼠標移動到終端2,按下回車鍵即可小車移動。
需要按鍵控制,感覺發出語音指令后,要等好幾秒才能移動小車,還需要按鍵,不過還是初步實現了語音控制,后期優化,實現更實用的語音控制