CocoPi_upgrade/preset/app/35_CocoTerminal.py

#!/usr/bin/env python
#version    :       2023.09.07
#language   :       en
from evdev import InputDevice, categorize, ecodes
import asyncio
import pty
import os
import sys
from maix import display
from maix import image
from maix import camera
from maix import mjpg
from maix import utils
import base64
import time
import requests
import json
import threading

camera.camera.config(size=(320,240))
image.load_freetype("/root/preset/fonts/simhei.ttf")
canvas = image.new(size = (320, 240))
try:
    # 获取键盘设备
    keyboard = InputDevice('/dev/input/event1')  # 替换成实际的键盘设备路径
except:
    keyboard = ""

if keyboard=="":
    canvas.draw_string(0,12, "Keyboard not detected", scale = 1.3, color = (255,255,255) , thickness = 1)
    display.show(canvas)
    time.sleep(3)
    sys.exit()
def v831_display_show_canvas(displayShow):
    global _canvas_y,_canvas_x
    CANVASSHOWIMGAGE = ""
    if ScreenOrientation:
        displayShowCanvas = image.new(size = (240, 320))
        displayShowCanvas.draw_rectangle(0,0,240,320, color=(0,0,0), thickness=-1)
        displayShowCanvas.draw_image(displayShow,_canvas_x,_canvas_y,alpha=1)
        displayShowVER = displayShowCanvas.crop(0,0,240,320)
        displayShowVER = displayShowVER.rotate(-90, adjust=1)
        CANVASSHOWIMGAGE = displayShowVER
        display.show(displayShowVER)
    else:
        displayShowCanvas = image.new(size = (320, 240))
        displayShowCanvas.draw_rectangle(0,0,320,240, color=(0,0,0), thickness=-1)
        displayShowCanvas.draw_image(displayShow,_canvas_x,_canvas_y,alpha=1)
        CANVASSHOWIMGAGE = displayShowCanvas
        display.show(displayShowCanvas)

    # jpg = utils.rgb2jpg(CANVASSHOWIMGAGE.convert("RGB").tobytes(), CANVASSHOWIMGAGE.width, CANVASSHOWIMGAGE.height)
    # img_base64 = base64.b64encode(jpg)
    # img_base64_str = "data:image/jpg;base64," + str(img_base64)[2:-1]
    # time.sleep(0.1)
    # try:
    #     if img_base64_str.rfind("data:image/jpg;base64,")>-1:
    #         print(img_base64_str)
    #     pass
    # except Exception as e:
    #     #print(str(e))
    #     pass
def get_post_chatgpt(datas):
    # 目标URL
    url = "https://gpt.cocorobo.cn/v831AskForTopic"
    payload = json.dumps({"topic":datas})
    headers={
        'Content-Type':'application/json'
    }
    print("payload"+str(payload))
    response=requests.request("POST",url,headers=headers,data=payload)
    print("response.text",response.text)
    if response.status_code == 200:
        result_num = len(json.loads(response.text)["FunctionResponse"]["result"])
        # keyboardStr[list_num] = 
    else:
        keyboardStr[list_num] = "fail"

ScreenOrientation = False
_canvas_x = 0
_canvas_y = 0
keyboardStr = [""]
list_num = 0
line = b""
lines = []
process =  None

keyboard = InputDevice("/dev/input/event1")
master, slave = pty.openpty()

#等待控制台的输出
def read_lines(master):
    global lines, line,keyboardStr,list_num,isShow
    char = os.read(master, 1)
    if char == b"\n":
        # lines.append(line.decode().strip())
        lines.append(line.decode().rstrip())
        line = b""
    else:
        line += char
# 用来判断是不是终端有新的输出
isShow = ""   
# 输入字符串
inputStr = "" 
# 跟踪Ctrl键的状态
ctrl_pressed = False
#读取键盘事件写入
def write_io():
    global keyboardStr,list_num,isShow,inputStr,ctrl_pressed,lines
    for event in keyboard.read_loop():
        if event.type == ecodes.EV_KEY:
            key_event = categorize(event)
            if key_event.keystate == key_event.key_down:
                try:
                    input_command = ""
                    if key_event.keycode == "KEY_LEFTCTRL" or key_event.keycode == "KEY_RIGHTCTRL":
                        ctrl_pressed = True
                    elif ctrl_pressed and key_event.keycode == "KEY_C":
                        # print("Ctrl+C 组合键被按下")
                        input_command = "\x1A\x03\x02\x03\x1D\x1A"
                        # pass
                    else:
                        ctrl_pressed = False

                    if ctrl_pressed == False:
                        if key_event.keycode[4:] == "SPACE":
                            inputStr = inputStr + " "
                            input_command =  " "
                            lines[len(lines)-1] = "root@CocoPi:/# "+inputStr
                        elif key_event.keycode[4:] == "SLASH":
                            inputStr = inputStr + "/"
                            input_command =  "/"
                            lines[len(lines)-1] = "root@CocoPi:/# "+inputStr
                        elif key_event.keycode[4:] == "DOT":
                            inputStr = inputStr + "."
                            input_command =  "."
                            lines[len(lines)-1] = "root@CocoPi:/# "+inputStr
                        elif key_event.keycode[4:] == "MINUS":
                            inputStr = inputStr + "_"
                            input_command =  "_"
                            lines[len(lines)-1] = "root@CocoPi:/# "+inputStr
                        elif key_event.keycode[4:] == "BACKSPACE":
                            inputStr = inputStr[:-1]
                            input_command = ""
                            lines[len(lines)-1] = "root@CocoPi:/# "+inputStr
                            # if len(keyboardStr[list_num]) ==0 and list_num>0:
                            #     list_num = list_num - 1
                        elif key_event.keycode[4:] == "ENTER":
                            inputStr = ""
                            input_command = "\n"
                            # time.sleep(1)
                            # lines.append("root@CocoPi:/# ")
                        else:
                            input_command = key_event.keycode[4:].lower()
                            inputStr = inputStr + key_event.keycode[4:].lower()
                            # if list_num == len(lines):
                            lines[len(lines)-1] = "root@CocoPi:/# "+inputStr
                    print("\x1A\x03\x02\x03\x1D\x1A")
                    print(input_command)
                    os.write(master, input_command.encode())
                except EOFError:
                    break

async def console_io():
    global master, slave, process,list_num,keyboardStr,isShow
    process = await asyncio.create_subprocess_exec('adb_shell', stdin=slave, stdout=slave, stderr=slave)

    while True:
        #读取所有的行数，放在lines
        read_lines(master)
        #屏幕循环的显示所有的数据
        # print(lines)
        #屏幕读取
        if isShow.lower() != " ".join(lines).lower():
            isShow = " ".join(lines)
            showTreminal(1)
            list_num = len(lines)
        else:
            pass

    os.close(master)
    os.close(slave)

def showTreminal(val):
    global master, slave, process,list_num,keyboardStr,lines,inputStr
    for i,box in enumerate(lines):
        if len(box)>0:
            keyboardStr.append(box)   
            # if box.rfind("root@CocoPi")>-1 and lines[i-1] == box:
            #     pass
            # else:
            #     keyboardStr.append(box)   
        elif i == len(lines)-1:
            # lines.append("root@CocoPi:/# ")
            keyboardStr.append("root@CocoPi:/# ")
    
    list_num = len(keyboardStr)-1
    canvas.clear()
    if len(keyboardStr)>20:
        for i in range(len(keyboardStr)-1,len(keyboardStr)-20,-1):
            canvas.draw_string(0,12*(i-(len(keyboardStr)-20)), str(keyboardStr[i]), scale = 0.785, color = (255,255,255) , thickness = 1)
    else:
        for i in range(len(keyboardStr)):
            canvas.draw_string(0,12*i, str(keyboardStr[i]), scale = 0.785, color = (255,255,255) , thickness = 1)
    v831_display_show_canvas(canvas)
    # display.show(canvas)
    
def main():
    thread = threading.Thread(target=write_io)
    thread.start()
    asyncio.run(console_io())
    thread.join()


main()