CocoPi_All_Files/preset/app/36_pictureAnalysis.py

#!/usr/bin/env python
#version    :       2024.03.13
#language   :       en
#hardware   :       PI
#camera     :       CR5205

from maix import display
from maix import image
from maix import camera
import pyaudio
import wave
import os
from maix import mjpg
from maix import utils
import base64
import time
import requests
import json
import uuid
import threading
import socket
import uuid
import sys
sys.path.append('/root/')
from CocoPi import BUTTON

key_B = BUTTON(8)
key_C = BUTTON(13)
key_D = BUTTON(7)
camera.camera.config(size=(320,240))
image.load_freetype("/root/preset/fonts/simhei.ttf")
canvas = image.new(size = (320, 240))
# 播放
def voice_numberMap(value):
    valueScaled = float(value - 0) / float(100)
    return int(valueScaled * 31)

VOICESTATE = 0

VOICESTREAM = ""
VOICEPYAUDIO = pyaudio.PyAudio()
CHUNK = 1024

VOICEPLAYSTATE = True
VOICEWF = ""
VOICEPPATH = ""

VOICENUMP = str(voice_numberMap(100))
time.sleep(0.01)
SYSTEMVOICE = "amixer cset numid=8,iface=MIXER,name=\"LINEOUT volume\" "+ VOICENUMP+""
def playVoice():
    global VOICEPLAYSTATE
    VOICEPLAYSTATE = True
    VOICEPLAYSTATE = VoicePlayState()

def VoicePlayState():
    global VOICESTATE,VOICEDATA,CHUNK,VOICESTREAM,VOICEPYAUDIO,VOICEWF,VOICEPPATH
    if VOICESTATE == 0:
        VOICEWF = wave.open(VOICEPPATH, "rb")#(sys.argv[1], "rb"
        VOICEPYAUDIO = pyaudio.PyAudio()

        VOICESTREAM = VOICEPYAUDIO.open(format=VOICEPYAUDIO.get_format_from_width(VOICEWF.getsampwidth()),channels=VOICEWF.getnchannels(),rate=VOICEWF.getframerate(),output=True)

        VOICEDATA = VOICEWF.readframes(CHUNK)
        VOICESTATE = 1
        return True
    else:
        if len(VOICEDATA) > 0:
            try:
                VOICESTREAM.write(VOICEDATA)
                VOICEDATA = VOICEWF.readframes(CHUNK)
                return True
            except:
                VOICESTATE = 0
        else:
            VOICESTREAM.stop_stream()
            VOICESTREAM.close()
            VOICEPYAUDIO.terminate()
            VOICESTATE = 0
            return False

# 线程1
def thread_calsss_fun1():
    global isStateVoice
    while True:
        if isStateVoice==1:
            getImage()
        elif VOICEPPATH != "" and isStateVoice == 3: 
            playVoice()

# 关闭线程
# CocoPiThread1.join()

# 图片转base64
def encode_image(image_path):
    img_data = ""
    try:
        with open(image_path, "rb") as f:
            img_data = f.read()
    except:
        pass
    return str(base64.b64encode(img_data), "utf-8")

def getImage():
    global imgUrlStr,isStateVoice,cameraImgUrl,VOICEPPATH
    uid = str(uuid.uuid1().hex)
    strBase = encode_image(cameraImgUrl)
    VOICEPPATH = ""
    params = {
        "max_tokens": 4096,
        "messages":[
            {
                "role": "user",
                "content":[
                    {"text": "What does the film talk about?","type": "text"},
                    {"image_url": {"url": f"data:image/jpeg;base64,{strBase}"},"type": "image_url"}
                ]
            }
        ],
        "uid": uid,
        "stream": False
    }
    headers={
        "Content-Type":"application/json"
    }
    res = requests.post("https://gpt4.cocorobo.cn/imageAnalyse", headers=headers, data=json.dumps(params))
    try:
        if res.status_code == 200:
            isStateVoice = 3
            aa = json.loads(res.text)["FunctionResponse"]["choices"][0]["message"]["content"]
            result_num = len(aa)//45
            for i in range(result_num + 1):
                imgUrlStr.append("")
                imgUrlStr[i] = str(aa)[i*45:(i+1)*45]
            
            url = "https://gpt4.cocorobo.cn/getV831Audio"
            payload = json.dumps({"input": aa,"response_format":"mp3","voice": "shimmer","uid":uid})
            response=requests.request("POST", url, headers=headers, data=payload)
            if response.status_code == 200:
                # print(json.loads(response.text)["FunctionResponse"]["url"])
                a = "https://gpt4.cocorobo.cn" + json.loads(response.text)["FunctionResponse"]["url"]
                r = requests.get(a)
                with open("/root/preset/audio/" + uid+".wav","wb") as f:
                    f.write(r.content)
                VOICEPPATH = "/root/preset/audio/" + uid+".wav"
            
    except:
        isStateVoice = 3
        imgUrlStr = ["Fail"]
        pass
    
# 判断联网
def wifi_is_content():
    global getDateNum
    cmd = "ifconfig"

    res = os.popen(cmd).read()
    data = False
    if res.rfind("inet addr:")!=48:
        data = True

    return data


def getPrivateIp():
    st = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
    try:
        st.connect(("10.255.255.255",1))
        IP = st.getsockname()[0]
    except Exception:
        IP = "127.0.0.1"
    finally:
        st.close()
    return IP

isStateVoice = 0    # 变量为0开始录音，1 GPT 识别录音结果

ScreenOrientation = False
CAMERAROTATE = +90

ssidInfo = ""
passwordInfo = ""
checkConnectState = False
connectText = ""
startConnect=False
wifiConnectState = False
# PublicIp=""
PrivateIP=""
runConnectSig=True
connectSuccessSig=False

while True:
    if wifi_is_content():
        break
    else:
        canvas = camera.capture()
        if ssidInfo!="" and passwordInfo!="":
            startConnect=True
            #connectText = "Waitting for Connection..."
            #canvas.draw_string(10,40, ssidInfo+" "+str(len(ssidInfo))+" "+str(type(ssidInfo)), scale = 1.5, color = (0,0,0), thickness = 1)
            canvas.draw_string(10,50, "Connecting to wifi...", scale = 1.5, color = (0,0,0), thickness = 1)
            display.show(canvas)
        if startConnect==True:
            canvas_1 = image.new(size = (320, 320), color = (255,255,255), mode = "RGB")
            canvas.draw_image(canvas_1,0,0, alpha=0.4)
            if wifiConnectState == True:
                if key_B.is_pressed():
                    while (key_B.is_pressed() == True):
                        time.sleep(0.001)
                    startConnect= False
                    connectText = ""
                    checkConnectState = False
                    connectSuccessSig = False
                    wifiConnectState = False
                    runConnectSig== True
                    ssidInfo=""
                    passwordInfo=""
                if checkConnectState == True:
                    if connectSuccessSig == False:
                        # PublicIp=getPublicIp()
                        connectSuccessSig = True
                    connectText = "Wifi connection successfully！"
                    # canvas.draw_string(10,80, "The WLAN PUBLIC IP:" + PublicIp, scale = 1.5, color = (0,0,0), thickness = 1)
                    canvas.draw_string(10,110, "THE WLAN PRIVATE IP:" + PrivateIP, scale = 1.5, color = (0,0,0), thickness = 1)
                    # canvas.draw_image((image.open("/root/preset/img/restart_ff0000_24x24.png")).rotate(0, adjust=0),8,216,alpha=1)
                    ssidInfo=""
                    passwordInfo=""
                    canvas.clear()
                    break
                else:
                    pass
            else:
                if checkConnectState == False:
                    os.system("wifi_disconnect_ap_test")
                    os.system('wifi_connect_chinese_ap_test '+ssidInfo+' '+passwordInfo+'')
                    checkConnectState = True
                if checkConnectState == True:
                    if key_B.is_pressed():
                        while (key_B.is_pressed() == True):
                            time.sleep(0.001)
                        startConnect= False
                        connectText = ""
                        checkConnectState = False
                        connectSuccessSig = False
                        wifiConnectState = False
                        runConnectSig== True
                        passwordInfo = ""
                        ssidInfo = ""
                    else:
                        if runConnectSig== True:
                            connectText = "Connecting to wifi,please wait..."
                            runConnectSig= False
                        else:
                            passwordInfo = ""
                            ssidInfo = ""
                            connectText = "Wifi connection failed！"
                            canvas.draw_image((image.open("/root/preset/img/restart_ff0000_24x24.png")).rotate(0, adjust=0),8,216,alpha=1)
                    
            
            canvas.draw_string(10,50, connectText, scale = 1.5, color = (0,0,0) , thickness = 1)
        else:
            mks = canvas.find_qrcodes()
            for mk in mks:
                #外框数据
                X = mk['x']
                Y = mk['y']
                W = mk['w']
                H = mk['h']            
                #二维码信息
                string = mk['payload']
                codeData = string.split(";")
                ssidInfo = codeData[0].split(":")[1]
                passwordInfo = codeData[1].split(":")[1]
                #画外框
                canvas.draw_rectangle(X, Y, X + W, Y + H, color=(0, 0, 255), thickness = 2) 
                #打印信息
                canvas.draw_string(int(X) , int(Y - 45) , "wifi name:"+ssidInfo, scale = 1, color = (255, 0, 0), thickness = 2)  #内框ID
                canvas.draw_string(int(X) , int(Y - 25) , "password:"+passwordInfo, scale = 1, color = (255, 0, 0), thickness = 2)  #内框ID
        canvas.draw_image((image.open("/root/preset/img/exit_ff0000_24x24.png")).rotate(0, adjust=0),288,216,alpha=1)
        display.show(canvas)
imgUrlStr = []
# 线程1
CocoPiThread1 = threading.Thread(target=thread_calsss_fun1)
# 开启线程
CocoPiThread1.start()
cameraImgUrl = ""
# getImage()
image_num = 0
while True:
    if key_D.is_pressed():
        isStateVoice = 0
        imgUrlStr = []
        VOICEPPATH = ""
    if len(imgUrlStr) > 0 and isStateVoice ==3:
        canvas.clear()
        canvas.draw_image((image.open("/root/preset/img/clockwise_bfbfbf_24x24.png")).rotate(0, adjust=0),2,2,alpha=1)
        if len(imgUrlStr) > 0:
            for i in range(len(imgUrlStr)):
                canvas.draw_string(0,15*(i + 2), str(imgUrlStr[i]), scale = 1, color = (255,255,255) , thickness = 1)
    elif isStateVoice != 2 and isStateVoice != 1:
        canvas.clear()
        canvas = camera.capture()
        if key_C.is_pressed():
            while not (key_C.is_pressed() == False):
                time.sleep(0.1)
            image_num = image_num + 1
            cameraImgUrl = "/root/user/img/image"+str(image_num)+ ".jpg"
            save_path = cameraImgUrl
            canvas.save(save_path)
            display.show(canvas)
            time.sleep(1)
            imgUrlStr = []
            isStateVoice = 1
        canvas.draw_image((image.open("/root/preset/img/camera_bfbfbf_24x24.png")).rotate(0, adjust=0),292,2,alpha=1)
    display.show(canvas)