CocoPi_upgrade/preset/app/30_training_data_collect.py

#!/usr/bin/env python
#version    :       2023.12.31
#language   :       ch

from maix import camera, display, zbar, image
import socket
import os
import sys
sys.path.append('/root/')
import http.client
from CocoPi import BUTTON
import time
import json
import requests

key_A = BUTTON(14)
key_B = BUTTON(8)
key_C = BUTTON(13)
key_D = BUTTON(7)
ScreenOrientation = False
image.load_freetype("/root/preset/fonts/CascadiaCodePL-Italic.ttf")
camera.camera.config(size=(320,240))

if ScreenOrientation:
    CAMERAROTATE = +180
else:
    CAMERAROTATE = +90

def getWifiConnectState():
    cmd = "wifi_get_connection_info_test 1"
    res = os.popen(cmd).read()
    wifiInfo = {}
    if res.find('get connection infomation successfully!') != -1:
        wifiInfo["state"]=True
        wifiInfo["AP"] = res[res.find("Connected AP: ")+13:res.find("IP address: ")-1]
        wifiInfo["IP"] = res[res.find("IP address: ")+12:res.find("frequency")-1]
        wifiInfo["frequency"] = res[res.find("frequency: ")+10:res.find("RSSI")-1]
        wifiInfo["RSSI"] = res[res.find("RSSI: ")+6:res.find("link_speed")-1]
        wifiInfo["link_speed"] = res[res.find("link_speed: ")+10:res.find("IP address: ")-1]
        wifiInfo["noise"] = res[res.find("noise: ")+6:res.find("noise: ")+11]
    else:
        wifiInfo["state"]=False
        wifiInfo["AP"] = "N/A"
        wifiInfo["IP"] = "N/A"
        wifiInfo["frequency"] = "N/A"
        wifiInfo["RSSI"] = "N/A"
        wifiInfo["link_speed"] = "N/A"
        wifiInfo["noise"] = "N/A"
    return wifiInfo

name = ''
host = ''
isCollect = False
image_num = 0
save_path = ""
startConnect=False

def getWifiConnectState():
    cmd = "wifi_get_connection_info_test 1"
    res = os.popen(cmd).read()
    wifiInfo = {}
    if res.find('get connection infomation successfully!') != -1:
        wifiInfo["state"]=True
        wifiInfo["AP"] = res[res.find("Connected AP: ")+13:res.find("IP address: ")-1]
        wifiInfo["IP"] = res[res.find("IP address: ")+12:res.find("frequency")-1]
        wifiInfo["frequency"] = res[res.find("frequency: ")+10:res.find("RSSI")-1]
        wifiInfo["RSSI"] = res[res.find("RSSI: ")+6:res.find("link_speed")-1]
        wifiInfo["link_speed"] = res[res.find("link_speed: ")+10:res.find("IP address: ")-1]
        wifiInfo["noise"] = res[res.find("noise: ")+6:res.find("noise: ")+11]
    else:
        wifiInfo["state"]=False
        wifiInfo["AP"] = "N/A"
        wifiInfo["IP"] = "N/A"
        wifiInfo["frequency"] = "N/A"
        wifiInfo["RSSI"] = "N/A"
        wifiInfo["link_speed"] = "N/A"
        wifiInfo["noise"] = "N/A"
    return wifiInfo

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

ssidInfo = ""
passwordInfo = ""
checkConnectState = False
connectText = ""
startConnectWifi=False
wifiConnectState = False
# PublicIp=""
PrivateIP=""
runConnectSig=True
connectSuccessSig=False
while True:
    canvas = camera.capture()
    IP = getPrivateIp()
    if ssidInfo!="" and passwordInfo!="":
        startConnectWifi=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, "正在连接WIFI，请等待...", scale = 1.5, color = (0,0,0), thickness = 1)
        display.show(canvas)
    if startConnectWifi==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)
                startConnectWifi= False
                connectText = ""
                checkConnectState = False
                connectSuccessSig = False
                wifiConnectState = False
                runConnectSig== True
                ssidInfo=""
                passwordInfo=""
            if checkConnectState == True:
                if connectSuccessSig == False:
                    # PublicIp=getPublicIp()
                    PrivateIP=getPrivateIp()
                    connectSuccessSig = True
                connectText = "WiFi连接成功！"
                canvas.draw_string(10,50, "WiFi连接成功！", scale = 1.5, color = (0,0,0) , thickness = 1)
                canvas.draw_string(10,80, "局域网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=""
                display.show(canvas)
                time.sleep(3)
                canvas.clear()
                
                break
            else:
                pass
        else:
            if checkConnectState == False:
                os.system("wifi_disconnect_ap_test")
                os.system('wifi_connect_chinese_ap_test '+ssidInfo+' '+passwordInfo+'')
                wifiConnectState=getWifiConnectState()["state"]
                checkConnectState = True
            if checkConnectState == True:
                if key_B.is_pressed():
                    while (key_B.is_pressed() == True):
                        time.sleep(0.001)
                    startConnectWifi= False
                    connectText = ""
                    checkConnectState = False
                    connectSuccessSig = False
                    wifiConnectState = False
                    runConnectSig== True
                    passwordInfo = ""
                    ssidInfo = ""
                else:
                    if runConnectSig== True:
                        connectText = "正在连接WIFI，请等待..."
                        runConnectSig= False
                    else:
                        passwordInfo = ""
                        ssidInfo = ""
                        connectText = "WIFI连接失败！"
                        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名："+ssidInfo, scale = 1, color = (255, 0, 0), thickness = 2)  #内框ID
            canvas.draw_string(int(X) , int(Y - 25) , "密码："+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)
    #canvas.draw_image((image.open("/root/preset/img/camera_bfbfbf_24x24.png")).rotate(0, adjust=0),292,2,alpha=1)
    display.show(canvas)
  
while True:
    canvas.clear()
    canvas = camera.capture()
    # IP = extract_ip()
    if str(name) != '' and str(host) != '':
        startConnect=True
    if startConnect==True:
        _COCOCLOUD_SEND_ENDPOINT = "https://" + host.split("//")[1] + "v831_update_code_connect_device"
        _COCOCLOUD_SEND_DATA = {"DatasetName":"" + str(name) + ""}
        try:
            _COCOCLOUD_SEND_REQUEST = requests.post(_COCOCLOUD_SEND_ENDPOINT, json = _COCOCLOUD_SEND_DATA , headers = { "Content-type": "application/json" }, timeout = 60)
            print(str(_COCOCLOUD_SEND_REQUEST.status_code)+", "+str(_COCOCLOUD_SEND_REQUEST.content))
        except BaseException as e:
            print(e)
        pass
        isCollect = True
        break
    else:
        mks = canvas.find_qrcodes()
        for mk in mks:
            #外框数据
            X = mk['x']
            Y = mk['y']
            W = mk['w']
            H = mk['h']            
            #二维码信息
            string = mk['payload']
            
            try:
                codeData = string.split(";")
                name = codeData[0].split("=")[1]
                host = codeData[1].split("=")[1]
            except:
                pass
            #画外框
            canvas.draw_rectangle(X, Y, X + W, Y + H, color=(0, 0, 255), thickness = 2) 
            #打印信息
            canvas.draw_string(int(X) , int(Y - 35) , str(string), scale = 1, color = (255, 0, 0), thickness = 2)  #内框ID

    canvas.draw_string(0, 0 , "扫描二维码来访问图片标注服务器", scale = 1, color = (255, 0, 0), thickness = 2)  #内框ID
    canvas.draw_string(0, 20 , "", scale = 1, color = (255, 0, 0), thickness = 2)  #内框ID
    # canvas.draw_image((image.open("/root/preset/img/camera_ff0000_24x24.png")).rotate(0, adjust=0),290,2,alpha=1)
    canvas.draw_image((image.open("/root/preset/img/exit_ff0000_24x24.png")).rotate(0, adjust=0),290,208,alpha=1)
        
    display.show(canvas)

isAlTake = False
while True:
    canvas1 = camera.capture()
    if isCollect:
        if not isAlTake:
            if key_C.is_pressed() :
                while not (key_C.is_pressed() == False):
                    time.sleep(0.01)
                image_num = image_num + 1
                save_path = ''.join([str(x) for x in ["/root/user/img/image", str(image_num), ".jpg"]])
                canvas1.save(save_path)
                isAlTake = True
            canvas1.draw_string(0, 0 , "请拍照来采集图片数据", scale = 1, color = (255, 0, 0), thickness = 2)  #内框ID
            canvas1.draw_image((image.open("/root/preset/img/camera_ff0000_24x24.png")).rotate(0, adjust=0),290,2,alpha=1)
            canvas1.draw_image((image.open("/root/preset/img/exit_ff0000_24x24.png")).rotate(0, adjust=0),290,208,alpha=1)
            display.show(canvas1)

        if isAlTake:
            canvas1.draw_image((image.open(save_path)),0,0)
            # if key_D.is_pressed() :
            #     while not (key_D.is_pressed() == False):
            #         time.sleep(0.01)
            #     img = open(save_path,'rb')
            #     data = {"file": img}
            #     param = {"name": str(name)}
            #     res = requests.post("https://" + host.split("//")[1] + "v831_upload_image", files=data, data=param)
            #     print(res.status_code)
            #     if str(res.status_code) == "200":
            #         isAlTake = False
            # if key_C.is_pressed() :
            #     while not (key_C.is_pressed() == False):
            #         time.sleep(0.01)
            #     isAlTake = False
            img = open(save_path,'rb')
            data = {"file": img}
            param = {"name": str(name)}
            res = requests.post("https://" + host.split("//")[1] + "v831_upload_image", files=data, data=param)
            print(res.status_code)
            if str(res.status_code) == "200":
                isAlTake = False
            # canvas1.draw_image((image.open("/root/preset/img/upload_ff0000_24x24.png")).rotate(0, adjust=0),5,2,alpha=1)
            # canvas1.draw_image((image.open("/root/preset/img/cancel_ff0000_24x24.png")).rotate(0, adjust=0),280,2,alpha=1)
            canvas1.draw_image((image.open("/root/preset/img/exit_ff0000_24x24.png")).rotate(0, adjust=0),290,208,alpha=1)
            display.show(canvas1)

    display.show(canvas1)