|
- from collections import deque
- import paddlehub as hub
-
- import os
- import asyncio
- import re
- import paddle
-
- import PIL.Image as Image
- import cv2
- # import paddlex as pdx
- import asyncio
- import json
- from garbage import predict
-
-
- os.system("SET CUDA_VISIBLE_DEVICES=0")
-
- model = hub.Module(name='plato-mini', version='1.0.0')
- model._interactive_mode = True # 开启交互模式
- model.max_turn = 20 # 对话轮次配置
- model.context = deque(maxlen=model.max_turn) # 对话上下文的存储队列
-
-
- from wechaty_puppet import FileBox, ScanStatus,MessageType # type: ignore
- from wechaty import Wechaty, Contact
- from wechaty.user import Message
- from typing import Optional, Union
- from wechaty.user import Message, Room
- os.environ['WECHATY_PUPPET_SERVICE_ENDPOINT'] = 'ip'
- os.environ['WECHATY_PUPPET_SERVICE_TOKEN'] = 'your token'
-
-
- from wechaty import (
- Contact,
- FileBox,
- Message,
- Wechaty,
- ScanStatus,
- )
-
- robot_state = 0
-
- def mural_transform(image_old_path,img_name):
-
-
- model = hub.Module(name='msgnet')
- model = hub.Module(directory='D:\\project\\python\\dia_bot\\Mural_Gan\\msgnet')
- result = model.predict(origin=[image_old_path], style='D:\\project\\python\\dia_bot\\Mural_Gan\\style_image\\style_2.jpg', visualization=True, save_path ='D:\\project\\python\\dia_bot\\Mural_Gan\\style_tranfer')
- new_path = 'D:\\project\\python\\dia_bot\\Mural_Gan\\style_tranfer\\' + img_name
- cv2.imwrite(new_path, result[0])
- return new_path
-
- async def on_message(msg: Message):
- """
- Message Handler for the Bot
- """
- talker = msg.talker()
- # await msg.text()
-
- if msg.text() == 'ding':
- await msg.say('dong')
-
- if msg.text() == 'hi' or msg.text() =='你好':
- global robot_state
- robot_state = 1000
- await talker.say("Hello, My name is thomas! Now I will introduce myself to you!")
- await talker.say("I'm your Personal Living Assistant")
- await talker.say("You konw? Murals can be seen not only in the Mogao Grottoes, and if you say me '壁画' or 'Mural', you also can enjoy murals's beauty.")
- await talker.say('Otherwise, I will be your Private Medical assistance! You can send me a message according to the format, for example "医疗:你的问题" or "Medicine:Your question" and I will try my best to answer your question about medicine!')
- await talker.say('Besides, I can help you sort garbage and you can say me "垃圾分类" or "Garbage Classification". ')
- await talker.say('Moreover,if you feel bored, say me "聊天" or "chat" and I will be your Siri.')
- await talker.say("I have rich functions and it's up you to dig...")
- await msg.say('Now begin your exploration!')
- robot_state = 0
-
- if msg.text() == 'Mural' or msg.text() == "壁画":
- robot_state = 2
- await talker.say('I will take you to Mogao Caves and please take me one picture!')
-
- if msg.text() == '聊会天吧' or msg.text() == 'chat':
- robot_state = 3
- # model = hub.Module(name='plato-mini', version='1.0.0')
- # model._interactive_mode = True
- # model.max_turn = 100
- # model.context = deque(maxlen=model.max_turn)
- await talker.say("Let's begin! And if you want to end the chat, you can say me '再见' or 'Bye'.")
-
- if robot_state == 3 and isinstance(msg.text(), str) and len(msg.text()) > 0 \
- and msg._payload.type == MessageType.MESSAGE_TYPE_TEXT:
- # and msg.text().startswith('[Chat]'): # Use a special token '[Test]' to select messages to respond.
- if msg.text() == '再见' or msg.text() == 'Bye':
- robot_state = 0
-
- bot_response = model.predict(data=msg.text())[0]
- await msg.say(bot_response) # Return the text generated by PaddleHub chatbot
-
- if msg.text() == "垃圾分类" or msg.text() == 'Garbage Classification':
- robot_state = 4
- await talker.say("Please send the garbage's picture to me!")
-
- if robot_state == 2 and msg.type() == Message.Type.MESSAGE_TYPE_IMAGE:
- await talker.say('Image received! Please wait a minute!')
- # 将Message转换为FileBox
-
- file_box_1 = await msg.to_file_box()
- # 获取图片名
- img_name = file_box_1.name
-
- # 图片保存的路径
- img_path = 'D:\\project\\python\\dia_bot\\Mural_Gan\\test_image\\' + img_name
-
- # 将图片保存为本地文件
- await file_box_1.to_file(file_path=img_path)
-
- # 调用图片风格转换的函数
- img_new_path = mural_transform(img_path,img_name)
-
- # 从新的路径获取图片
- file_box_2 = FileBox.from_file(img_new_path)
- robot_state = 0
-
- await msg.say(file_box_2)
-
- if "医疗" in msg.text() or "Medicine" in msg.text():
- module = hub.Module(name="Medicine-dialogue")
- if ':' in msg.text() and msg.text().split(':')[0] == '医疗' :
- print(msg.text().split(':')[1])
- question = msg.text().split(':')[1]
- robot_state = 3
-
- if ":" in msg.text() and msg.text().split(':')[0] == 'Medicine':
- print(msg.text().split(':')[1])
- robot_state =3
- question = msg.text().split(':')[1]
- print(question)
- await talker.say('Question received! Please wait a minute!')
-
- test_texts = []
- test_texts.append(question)
- # generate包含3个参数,texts为输入文本列表,use_gpu指定是否使用gpu,beam_width指定beam search宽度。
- results = module.generate(texts=test_texts, use_gpu=False, beam_width=1)
- robot_state = 0
- for result in results:
- print(result)
- await msg.say(result[0])
-
-
- if robot_state == 4 and msg.type() == Message.Type.MESSAGE_TYPE_IMAGE:
-
- await talker.say('Image received! Please wait a moment!')
- # 将Message转换为FileBox
- file_box_2 = await msg.to_file_box()
- # 获取图片名
- img_name = file_box_2.name
-
- # 图片保存的路径
- img_path = 'D:\\project\\python\\dia_bot\\garbage\\picture\\' + img_name
- await file_box_2.to_file(file_path=img_path)
- result = predict.main(img_path)
- f_obj=open('D:\\project\\python\\dia_bot\\garbage\\garbage_classification.json','r',encoding='utf8')
-
-
- # print(res[i])
- category_id = result[0][0]
- score = result[1][0]
- score="%.2f%%"%(score*100)
- content=json.load(f_obj)[str(category_id)]
- content = "This garbage is %s and it's score is %s."%(content,str(score))
-
- robot_state = 0
-
- await talker.say(content)
-
- print(content)
-
- # number=result[0]['category']
-
- # score=result[0]['score']
-
- # score="%.2f%%"%(score*100)
-
- # content=json.load(f_obj)[number]
-
- # content = "This garbage is %s and it's score is %s."%(content,str(score))
- # robot_state = 0
-
- # await talker.say(content)
-
-
- # elif isinstance(msg.text(), str) and len(msg.text()) > 0 \
- # and msg._payload.type == MessageType.MESSAGE_TYPE_TEXT : # Use a special token '[Test]' to select messages to respond.
- # bot_response = model.predict(data=msg.text())[0]
- # await msg.say(bot_response) # Return the text generated by PaddleHub chatbot
-
- async def on_scan(
- qrcode: str,
- status: ScanStatus,
- _data,
- ):
- """
- Scan Handler for the Bot
- """
- print('Status: ' + str(status))
- print('View QR Code Online: https://wechaty.js.org/qrcode/' + qrcode)
-
-
- async def on_login(user: Contact):
- """
- Login Handler for the Bot
- """
- print(user)
- # TODO: To be written
-
-
- async def main():
- """
- Async Main Entry
- """
- #
- # Make sure we have set WECHATY_PUPPET_SERVICE_TOKEN in the environment variables.
- # Learn more about services (and TOKEN) from https://wechaty.js.org/docs/puppet-services/
- #
- if 'WECHATY_PUPPET_SERVICE_TOKEN' not in os.environ:
- print('''
- Error: WECHATY_PUPPET_SERVICE_TOKEN is not found in the environment variables
- You need a TOKEN to run the Python Wechaty. Please goto our README for details
- https://github.com/wechaty/python-wechaty-getting-started/#wechaty_puppet_service_token
- ''')
-
- bot = Wechaty()
-
- bot.on('scan', on_scan)
- bot.on('login', on_login)
- bot.on('message', on_message)
-
- await bot.start()
-
- print('[Python Wechaty] Ding Dong Bot started.')
-
- asyncio.run(main())
-
|