野火【鲁班猫】卡片电脑创意氛围赛（第二期）基于LubanCat...

ChenChenRui

大规模预训练语言模型（Large Language Model, LLM）的出现推动了机器人领域的发展。这些模型通过在大规模文本数据上进行预训练，可以学习到丰富的语言知识和语义表示。然后，这些模型可以通过微调来适应特定的任务或领域。自然语言是一种大众掌握的技能。通过使用自然语言与计算机交互，降低了新手的使用难度，直观有效，降低了学习成本。本项目以机械臂为大语言模型的身体，实现具身智能。

1. 什么是具身智能？

具身智能通过在物理世界和数字世界的学习和进化，达到理解世界、互动交互并完成任务的目标。

具身智能是由“本体”和“智能体”耦合而成且能够在复杂环境中执行任务的智能系统。一般认为，具身智能具有如下的几个核心要素：

第一是本体，作为实际的执行者，是在物理或者虚拟世界进行感知和任务执行的机构。

本体通常是具有物理实体的机器人，可以有多种形态。本体的能力边界会限制智能体的能力发挥，本体具备环境感知能力、运动能力和操作执行能力，是连接数字世界和物理世界的载体，本项目采用三轴步进机械臂。

具身智能的第二个要素是智能体（EmbodiedAgents），是具身于本体之上的智能核心，负责感知、理解、决策、控制等的核心工作。

智能体可以感知复杂环境，理解环境所包含的语义信息，能够和环境进行交互；可以理解具体任务，并且根据环境的变化和目标状态做出决策，进而控制本体完成任务。

本项目的方案

硬件方面：

鲁班猫4、usb摄像头、usb音频模块、机械臂及其控制器

机械臂为三轴步进机械臂。控制器为stm32，接收串口指令完成运动学正逆解以及圆弧直线插补等算法。

上位机为鲁班猫4通过串口与stm32连接，完成对机械臂的控制以及连接usb摄像头和usb音频获取环境。

软件方面：

基于ChatGPT的强大自然语言理解和推理能力，生成控制机器人的相关代码并通过串口向机械臂发送指令，利用usb摄像头获取环境和物体信息，以及通过usb音频模块获取人类指令并向人类反馈状态。

物体定位：采用opencv

OpenCV是一个开源的计算机视觉和机器学习软件库，由英特尔公司发起并开发。它提供了丰富的图像处理和计算机视觉的功能，涵盖了从基本的图像操作、图像处理到目标检测、人脸识别等高级功能。OpenCV广泛应用于各种领域，包括医学图像处理、安防监控、自动驾驶、机器人技术等。

      本项目采用将采集到的图片先进行仿射变换完成投影到xy平面后再转换到hsv颜色空间后寻找物体轮廓和定位。

1. <blockquote>import cv2

复制代码

语言转文字为调用讯飞api

1. import base64
   
2. import hashlib
   
3. import hmac
   
4. import json
   
5. import os
   
6. import time
   
7. 
   
8. import requests
   
9. 
   
10. lfasr_host = 'http://raasr.xfyun.cn/api'
    
11. 
    
12. # 请求的接口名
    
13. api_prepare = '/prepare'
    
14. api_upload = '/upload'
    
15. api_merge = '/merge'
    
16. api_get_progress = '/getProgress'
    
17. api_get_result = '/getResult'
    
18. # 文件分片大小10M
    
19. file_piece_sice = 10485760
    
20. 
    
21. # ——————————————————转写可配置参数————————————————
    
22. # 参数可在官网界面（https://doc.xfyun.cn/rest_api/%E8%AF%AD%E9%9F%B3%E8%BD%AC%E5%86%99.html）查看，根据需求可自行在gene_params方法里添加修改
    
23. # 转写类型
    
24. lfasr_type = 0
    
25. # 是否开启分词
    
26. has_participle = 'false'
    
27. has_seperate = 'true'
    
28. # 多候选词个数
    
29. max_alternatives = 0
    
30. # 子用户标识
    
31. suid = ''
    
32. 
    
33. 
    
34. class SliceIdGenerator:
    
35.     """slice id生成器"""
    
36. 
    
37.     def __init__(self):
    
38.         self.__ch = 'aaaaaaaaa`'
    
39. 
    
40.     def getNextSliceId(self):
    
41.         ch = self.__ch
    
42.         j = len(ch) - 1
    
43.         while j >= 0:
    
44.             cj = ch[j]
    
45.             if cj != 'z':
    
46.                 ch = ch[:j] + chr(ord(cj) + 1) + ch[j + 1:]
    
47.                 break
    
48.             else:
    
49.                 ch = ch[:j] + 'a' + ch[j + 1:]
    
50.                 j = j - 1
    
51.         self.__ch = ch
    
52.         return self.__ch
    
53. 
    
54. 
    
55. class RequestApi(object):
    
56.     def __init__(self, appid, secret_key, upload_file_path):
    
57.         self.appid = appid
    
58.         self.secret_key = secret_key
    
59.         self.upload_file_path = upload_file_path
    
60. 
    
61.     # 根据不同的apiname生成不同的参数,本示例中未使用全部参数您可在官网(https://doc.xfyun.cn/rest_api/%E8%AF%AD%E9%9F%B3%E8%BD%AC%E5%86%99.html)查看后选择适合业务场景的进行更换
    
62.     def gene_params(self, apiname, taskid=None, slice_id=None):
    
63.         appid = self.appid
    
64.         secret_key = self.secret_key
    
65.         upload_file_path = self.upload_file_path
    
66.         ts = str(int(time.time()))
    
67.         m2 = hashlib.md5()
    
68.         m2.update((appid + ts).encode('utf-8'))
    
69.         md5 = m2.hexdigest()
    
70.         md5 = bytes(md5, encoding='utf-8')
    
71.         # 以secret_key为key, 上面的md5为msg， 使用hashlib.sha1加密结果为signa
    
72.         signa = hmac.new(secret_key.encode('utf-8'), md5, hashlib.sha1).digest()
    
73.         signa = base64.b64encode(signa)
    
74.         signa = str(signa, 'utf-8')
    
75.         file_len = os.path.getsize(upload_file_path)
    
76.         file_name = os.path.basename(upload_file_path)
    
77.         param_dict = {}
    
78. 
    
79.         if apiname == api_prepare:
    
80.             # slice_num是指分片数量，如果您使用的音频都是较短音频也可以不分片，直接将slice_num指定为1即可
    
81.             slice_num = int(file_len / file_piece_sice) + (0 if (file_len % file_piece_sice == 0) else 1)
    
82.             param_dict['app_id'] = appid
    
83.             param_dict['signa'] = signa
    
84.             param_dict['ts'] = ts
    
85.             param_dict['file_len'] = str(file_len)
    
86.             param_dict['file_name'] = file_name
    
87.             param_dict['slice_num'] = str(slice_num)
    
88.         elif apiname == api_upload:
    
89.             param_dict['app_id'] = appid
    
90.             param_dict['signa'] = signa
    
91.             param_dict['ts'] = ts
    
92.             param_dict['task_id'] = taskid
    
93.             param_dict['slice_id'] = slice_id
    
94.         elif apiname == api_merge:
    
95.             param_dict['app_id'] = appid
    
96.             param_dict['signa'] = signa
    
97.             param_dict['ts'] = ts
    
98.             param_dict['task_id'] = taskid
    
99.             param_dict['file_name'] = file_name
    
100.         elif apiname == api_get_progress or apiname == api_get_result:
     
101.             param_dict['app_id'] = appid
     
102.             param_dict['signa'] = signa
     
103.             param_dict['ts'] = ts
     
104.             param_dict['task_id'] = taskid
     
105.         return param_dict
     
106. 
     
107.     # 请求和结果解析，结果中各个字段的含义可参考：https://doc.xfyun.cn/rest_api/%E8%AF%AD%E9%9F%B3%E8%BD%AC%E5%86%99.html
     
108.     def gene_request(self, apiname, data, files=None, headers=None):
     
109.         response = requests.post(lfasr_host + apiname, data=data, files=files, headers=headers)
     
110.         result = json.loads(response.text)
     
111.         if result["ok"] == 0:
     
112.             print("{} success:".format(apiname) + str(result))
     
113.             return result
     
114.         else:
     
115.             print("{} error:".format(apiname) + str(result))
     
116.             exit(0)
     
117.             return result
     
118. 
     
119.     # 预处理
     
120.     def prepare_request(self):
     
121.         return self.gene_request(apiname=api_prepare,
     
122.                                  data=self.gene_params(api_prepare))
     
123. 
     
124.     # 上传
     
125.     def upload_request(self, taskid, upload_file_path):
     
126.         file_object = open(upload_file_path, 'rb')
     
127.         try:
     
128.             index = 1
     
129.             sig = SliceIdGenerator()
     
130.             while True:
     
131.                 content = file_object.read(file_piece_sice)
     
132.                 if not content or len(content) == 0:
     
133.                     break
     
134.                 files = {
     
135.                     "filename": self.gene_params(api_upload).get("slice_id"),
     
136.                     "content": content
     
137.                 }
     
138.                 response = self.gene_request(api_upload,
     
139.                                              data=self.gene_params(api_upload, taskid=taskid,
     
140.                                                                    slice_id=sig.getNextSliceId()),
     
141.                                              files=files)
     
142.                 if response.get('ok') != 0:
     
143.                     # 上传分片失败
     
144.                     print('upload slice fail, response: ' + str(response))
     
145.                     return False
     
146.                 print('upload slice ' + str(index) + ' success')
     
147.                 index += 1
     
148.         finally:
     
149.             'file index:' + str(file_object.tell())
     
150.             file_object.close()
     
151.         return True
     
152. 
     
153.     # 合并
     
154.     def merge_request(self, taskid):
     
155.         return self.gene_request(api_merge, data=self.gene_params(api_merge, taskid=taskid))
     
156. 
     
157.     # 获取进度
     
158.     def get_progress_request(self, taskid):
     
159.         return self.gene_request(api_get_progress, data=self.gene_params(api_get_progress, taskid=taskid))
     
160. 
     
161.     # 获取结果
     
162.     def get_result_request(self, taskid):
     
163.         return self.gene_request(api_get_result, data=self.gene_params(api_get_result, taskid=taskid))
     
164. 
     
165.     def all_api_request(self):
     
166.         # 1. 预处理
     
167.         pre_result = self.prepare_request()
     
168.         taskid = pre_result["data"]
     
169.         # 2 . 分片上传
     
170.         self.upload_request(taskid=taskid, upload_file_path=self.upload_file_path)
     
171.         # 3 . 文件合并
     
172.         self.merge_request(taskid=taskid)
     
173.         # 4 . 获取任务进度
     
174.         while True:
     
175.             # 每隔20秒获取一次任务进度
     
176.             progress = self.get_progress_request(taskid)
     
177.             progress_dic = progress
     
178.             if progress_dic['err_no'] != 0 and progress_dic['err_no'] != 26605:
     
179.                 print('task error: ' + progress_dic['failed'])
     
180.                 return
     
181.             else:
     
182.                 data = progress_dic['data']
     
183.                 task_status = json.loads(data)
     
184.                 if task_status['status'] == 9:
     
185.                     print('task ' + taskid + ' finished')
     
186.                     break
     
187.                 print('The task ' + taskid + ' is in processing, task status: ' + str(data))
     
188. 
     
189.             # 每次获取进度间隔20S
     
190.             time.sleep(20)
     
191.         # 5 . 获取结果
     
192.         res = self.get_result_request(taskid=taskid)
     
193.         # 保存到txt
     
194.         data = res['data'].strip('[').strip(']')
     
195.         for d in eval(data):
     
196.             a = d['onebest']
     
197.             fw.write(a + '\n') # 逐行保存
     
198. 
     
199. 
     
200. # 注意：如果出现requests模块报错："NoneType" object has no attribute 'read', 请尝试将requests模块更新到2.20.0或以上版本(本demo测试版本为2.20.0)
     
201. # 输入讯飞开放平台的appid，secret_key和待转写的文件路径
     
202. if __name__ == '__main__':
     
203.     # 单个语音文件解析并保存
     
204.     root = "D:/xxxx/"
     
205.     filename = "xxxx" # mp3文件名，需修改
     
206.     speechpath = root +filename
     
207.     save_path = speechpath + '.txt' # txt与mp3同名，并保存在同一路径下
     
208.     fw = open(save_path, 'w')
     
209. 
     
210.     # 解析语音并保存到txt
     
211.     api = RequestApi(appid="xxx", secret_key="xxx", upload_file_path=speechpath+".mp3")
     
212.     api.all_api_request()

复制代码

因为此项目还在参加比赛，所以未公开整个工程和大语言模型部分。等后续比赛结束整理资料后再开源。

おおおじ。

真不错