行空板之智能互动艺术展示大屏
本帖最后由 云天 于 2024-10-4 12:43 编辑在当今数字化时代，人工智能技术的发展为艺术创作和信息展示带来了革命性的变化。特别是在公共艺术和教育领域，互动式体验的需求日益增长。在这样的背景下，我们提出了一个创新的人工智能艺术互动项目——“智能互动艺术展示大屏”。
【项目背景】
随着人工智能技术的不断进步，特别是自然语言处理和图像识别技术的发展，机器已经能够理解和创造人类艺术。为了将这些先进的技术带入公众视野，同时提供一种新颖的互动体验，我们设计了“智能互动艺术展示大屏”项目。该项目结合了LattePanda 拿铁熊猫和行空板两种硬件平台，通过物联网技术实现跨设备的通信和协作，打造一个能够响应观众语音指令并实时展示生成图像的艺术展览系统。
【项目目标】
增强艺术互动性：通过行空板接收观众的语音指令，转化为文本信息，再由人工智能模型生成相应的图像，实现艺术作品与观众之间的互动。
展示AI创造力：利用讯飞星火认知大模型的文本到图像生成能力，展现人工智能在艺术创作领域的潜力。
教育与娱乐结合：为观众提供一个教育性质的娱乐体验，让他们了解人工智能的工作原理，同时享受艺术创作的乐趣。
创新展示方式：使用LattePanda 拿铁熊猫和大屏幕作为图像展示平台，为观众带来震撼的视觉体验。
【项目实现】
行空板：作为用户交互的前端设备，通过麦克风捕捉用户的语音指令，通过按键触发语音识别和图像生成流程。生成的图像通过Easy Iot物联网平台发送。
讯飞星火认知大模型：接收行空板发送的文本信息，利用其强大的文本理解能力，生成与文本内容匹配的图像。
Easy Iot物联网平台：作为信息传输的桥梁，将行空板生成的文本和图像信息传输给LattePanda 拿铁熊猫。
LattePanda 拿铁熊猫：连接大屏幕，接收来自物联网平台的图像信息，并在大屏幕上实时展示生成的艺术作品。
【项目创新点】
多模态交互：结合语音和图像两种模态，提供丰富的用户体验。
实时图像生成：快速响应用户指令，实时生成并展示图像。
物联网集成：利用物联网技术实现设备间的无缝通信。
【项目展示图】
【演示视频】https://www.bilibili.com/video/BV1vbxfeEEBJ/?share_source=copy_web
【发送程序】
1.Mind+添加扩展，行空板、讯飞语音、代码生成器、MQTT、OpenCV、讯飞星火。
1.语音识别
[*]程序使用了df_xfyun_speech库中的XfIat类来进行语音识别。
[*]XfIat是讯飞语音识别的接口，用于将录制的音频转换为文本。
[*]用户通过按钮触发录音，录音结束后，音频文件通过iat.recognition("record.wav")进行识别，识别结果存储在“语音识别结果”变量中。
2.图像生成
[*]程序中使用了Spark库的tti方法进行图像生成。
[*]根据搜索结果，Spark库的tti方法用于构建图片生成LLM（Large Language Model）实例，该实例可以向大模型发送文本请求，大模型根据请求生成相应的图片。
[*]在代码中，当语音识别的结果包含“绘”和“画”时，会触发图像生成的逻辑，调用spark.tti(语音识别结果)生成图片。
3.物联网发送
[*]程序使用了siot库来进行物联网通信。
[*]siot.init初始化物联网客户端，配置了客户端ID、服务器地址、端口、用户名和密码。
[*]siot.connect和siot.loop用于连接物联网服务器并保持连接。
[*]当用户按下按钮B时，程序会通过siot.publish方法将数据发布到指定的物联网主题，这里的topic="z4ksqL6Ig"是物联网平台的主题，data=spark.imageBase是要发送的数据，是图像的Base64编码字符串。
这个脚本整合了AI图像生成、MQTT通信和OpenCV图像显示的功能。它通过讯飞星火大语言模型生成图像，然后通过MQTT协议将图像传输到物联网平台。接收到图像后，使用OpenCV库在窗口中显示图像。【修改扩展】 以上代码，要实现需修改扩展库存“libraries”——“Spark.py”。文件所在路径：C:\Users\lenovo\AppData\Local\DFScratch\extensions\ac000108-spark-thirdex\python\libraries 修改“imageBase”，成为类变量。是从讯飞返回的图片base64编码字符串。下面是修改后的“Spark.py”文件内容。
import _thread as thread
import os
import base64
import hashlib
import hmac
import json
from urllib.parse import urlparse , urlencode
import ssl
from datetime import datetime
from time import mktime
from wsgiref.handlers import format_date_time
import websocket
import openpyxl
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
from PIL import Image
class Spark:
def __init__(self,appid,api_secret,api_key,Personality):
self.appid = appid
self.api_secret = api_secret
self.api_key = api_key
self.gpt_url = "wss://spark-api.xf-yun.com/v3.5/chat"
self.tti_url = "https://spark-api.cn-huabei-1.xf-yun.com/v2.1/tti"
self.domain = "generalv3.5"
self.Personality = Personality
self.msgs = ""
self.history = ""
self.flag = True
self.imageBase=""
# 收到websocket错误的处理
def on_error(self,ws, error):
#print("### error:", error)
pass
# 收到websocket关闭的处理
def on_close(self,ws):
#print("### closed ###")
pass
# 收到websocket连接建立的处理
def on_open(self,ws):
#print("### open ###")
thread.start_new_thread(self.run, (ws,))
def run(self,ws, *args):
#print("### run ###")
data = json.dumps(self.gen_params(appid=ws.appid, query=ws.query, domain=ws.domain))
ws.send(data)
# 收到websocket消息的处理
def on_message(self,ws, message):
data = json.loads(message)
code = data['header']['code']
if code != 0:
#print(f'请求错误: {code}, {data}')
ws.close()
else:
choices = data["payload"]["choices"]
status = choices["status"]
content = choices["text"]["content"]
self.msgs = self.msgs + content
#print(content,end='')
if status == 2:
self.flag = False
ws.close()
ws.keep_running = False
def gen_params(self,appid, query, domain):
text = []
text.extend(self.Personality)
text.extend(self.history)
text.append({"role": "user", "content": query})
data = {
"header": {
"app_id": appid,
"uid": "1234"
},
"parameter": {
"chat": {
"domain": domain,
"temperature": 0.5,
"max_tokens": 4096,
"auditing": "default"
}
},
"payload": {
"message": {
"text":text
}
}
}
return data
def ask(self,prompt,history):
self.flag = True
self.msgs = ""
self.history = history
try:
host = urlparse(self.gpt_url).netloc
path = urlparse(self.gpt_url).path
websocket.enableTrace(False)
now = datetime.now()
date = format_date_time(mktime(now.timetuple()))
signature_origin = "host: " + host + "\n"
signature_origin += "date: " + date + "\n"
signature_origin += "GET " + path + " HTTP/1.1"
signature_sha = hmac.new(self.api_secret.encode('utf-8'), signature_origin.encode('utf-8'),digestmod=hashlib.sha256).digest()
signature_sha_base64 = base64.b64encode(signature_sha).decode(encoding='utf-8')
authorization_origin = f'api_key="{self.api_key}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature_sha_base64}"'
authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
v = {
"authorization": authorization,
"date": date,
"host": host
}
wsUrl = self.gpt_url + '?' + urlencode(v)
ws = websocket.WebSocketApp(wsUrl, on_message=self.on_message,on_open=self.on_open, on_error=self.on_error, on_close=self.on_close )
ws.appid = self.appid
ws.query = prompt
ws.domain = self.domain
ws.history = history
ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
while(self.flag):
pass
return self.msgs
except:
pass
def tti(self,prompt):
stidx = self.tti_url.index("://")
host = self.tti_url
schema = self.tti_url[:stidx + 3]
edidx = host.index("/")
if edidx <= 0:
raise AssembleHeaderException("invalid request url:" + self.tti_url)
path = host
host = host[:edidx]
now = datetime.now()
date = format_date_time(mktime(now.timetuple()))
signature_origin = "host: {}\ndate: {}\n{} {} HTTP/1.1".format(host, date, 'POST', path)
signature_sha = hmac.new(self.api_secret.encode('utf-8'), signature_origin.encode('utf-8'),
digestmod=hashlib.sha256).digest()
signature_sha = base64.b64encode(signature_sha).decode(encoding='utf-8')
authorization_origin = "api_key=\"%s\", algorithm=\"%s\", headers=\"%s\", signature=\"%s\"" % (
self.api_key, "hmac-sha256", "host date request-line", signature_sha)
authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
values = {
"host": host,
"date": date,
"authorization": authorization
}
url = self.tti_url + "?" + urlencode(values)
print(url)
body= {
"header": {
"app_id": self.appid,
"uid":"123456789"
},
"parameter": {
"chat": {
"domain": "general",
"width": 1280,
"height": 720
}
},
"payload": {
"message":{
"text":[
{
"role":"user",
"content":prompt
}
]
}
}
}
response = requests.post(url,json=body,headers={'content-type': "application/json"}).text
data = json.loads(response)
code = data['header']['code']
if code != 0:
print(f'请求错误: {code}, {data}')
else:
text = data["payload"]["choices"]["text"]
imageContent = text
self.imageBase = imageContent["content"]
imgdata = base64.b64decode(self.imageBase)
with open("1.jpg",'wb') as f:
f.write(imgdata)
img = Image.open("1.jpg")
# 旋转方式二
img2 = img.rotate(90,expand=True) # 自定义旋转度数
img2 = img2.resize((240, 320)) # 改变图片尺寸
img2.save("1.png")
Mind+连接行空板，在终端中使用命令，cd mindplus/.lib/thirdExtension/进入第三方扩展文件夹，使用ls命令查看是否存在ac000108-spark-thirdex文件夹。如果存在，使用rm -r ac000108-spark-thirdex删除。再次运行图形化程序，Mind+会将修改后的扩展，下载到行空板中。【接收程序】
1.物联网接收图片
[*]程序初始化了pygame模块，并导入了numpy和base64库。
[*]使用MQTT协议来初始化物联网通信，通过siot.init配置了物联网客户端，包括客户端ID、服务器地址、端口、用户名和密码。
[*]通过siot.connect发起连接，siot.loop保持连接，这样程序可以持续地监听物联网服务器的消息。
[*]通过siot.subscribe订阅了特定的MQTT主题"z4ksqL6Ig"，以便接收来自该主题的消息。
[*]当MQTT从主题"z4ksqL6Ig"接收到消息时，会触发全局的img和BiaoShi变量。img变量存储Base64编码的图片数据，BiaoShi变量用作标识，表示有新的图片接收。
2.显示图片
[*]程序加载了"ding.wav"音乐文件，并在接收到新图片时播放，作为提示音。
[*]使用pygame创建了一个名为"Mind+'s Windows"的窗口，并设置为全屏显示。
[*]读取了"logo.jpg"图片文件，并在窗口中显示。
[*]在主循环中，如果BiaoShi标识为1，表示有新的图片接收，程序会重置BiaoShi为0，并播放音乐提示音。
[*]接收到的图片数据（Base64编码）被解码并转换为numpy数组，然后使用cv2.imdecode函数解码为图像格式。
[*]如果BiaoShi标识不为1，程序会继续在窗口中显示当前的图片，通过cv2.imshow函数将图片image展示在"Mind+'s Windows"窗口中。
这个程序通过MQTT协议接收物联网平台发送的Base64编码图片，解码后使用OpenCV在pygame窗口中展示图片，并在接收到新图片时播放声音提示。
【图片展示】
通过这个项目，我们不仅能够展示人工智能技术的最新成果，还能够探索艺术与科技融合的新可能性，为公众带来前所未有的互动艺术体验。
页:
[1]