行空板M10扩展板校园监控车 简单

import _thread as thread
import base64
import hashlib
import hmac
import json
from urllib.parse import urlparse
import ssl
from datetime import datetime
from time import mktime
from urllib.parse import urlencode
from wsgiref.handlers import format_date_time
import websocket
from websocket import WebSocketApp
import sys
import os

appid = "XXX"  
api_secret = "XXX"
api_key = "XXX"

imageunderstanding_url = "wss://spark-api.cn-huabei-1.xf-yun.com/v2.1/image"

class Ws_Param:
    def __init__(self, APPID, APIKey, APISecret, imageunderstanding_url):
        self.APPID = APPID
        self.APIKey = APIKey
        self.APISecret = APISecret
        self.host = urlparse(imageunderstanding_url).netloc
        self.path = urlparse(imageunderstanding_url).path
        self.ImageUnderstanding_url = imageunderstanding_url

    def create_url(self):
        now = datetime.now()
        date = format_date_time(mktime(now.timetuple()))
        signature_origin = "host: " + self.host + "\n"
        signature_origin += "date: " + date + "\n"
        signature_origin += "GET " + self.path + " HTTP/1.1"
        signature_sha = hmac.new(self.APISecret.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.APIKey}", 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": self.host}
        return self.ImageUnderstanding_url + '?' + urlencode(v)

def on_error(ws, error):
    print("### error:", error)
    ws.is_closed = True

def on_close(ws, *args):
    print("Connection closed")
    ws.is_closed = True

def on_open(ws):
    def run(*args):
        # 使用当前问题构建消息
        messages = [
            {
                "role": "user",
                "content": base64.b64encode(ws.imagedata).decode('utf-8'),
                "content_type": "image",
                "image_param": {"image_type": "PNG"}
            },
            {
                "role": "user",
                "content": ws.current_question,
                "content_type": "text"
            }
        ]
        data = json.dumps(gen_params(appid=ws.appid, messages=messages))
        ws.send(data)
    thread.start_new_thread(run, ())

def on_message(ws, message):
    data = json.loads(message)
    code = data['header']['code']
    if code != 0:
        print(f'Error: {code}, {data}')
        ws.close()
    else:
        choices = data["payload"]["choices"]
        status = choices["status"]
        content = choices["text"][0]["content"]
        print(content, end="")
        ws.answer += content
        if status == 2:
            print()  
            ws.close()

def gen_params(appid, messages):
    return {
        "header": {"app_id": appid},
        "parameter": {
            "chat": {
                "domain": "imagev3",
                "temperature": 0.5,
                "top_k": 4,
                "max_tokens": 2028,
                "auditing": "default"
            }
        },
        "payload": {"message": {"text": messages}}
    }

def main(appid, api_key, api_secret, imageunderstanding_url, question, image_path):
    # 读取图像数据
    try:
        if not os.path.exists(image_path):
            print(f"错误：图片文件 {image_path} 不存在")
            return f"识别失败：图片文件不存在 - {image_path}"
            
        with open(image_path, 'rb') as f:
            imagedata = f.read()
    except Exception as e:
        print(f"读取图片文件失败: {str(e)}")
        return f"识别失败：无法读取图片 - {str(e)}"
    
    wsParam = Ws_Param(appid, api_key, api_secret, imageunderstanding_url)
    wsUrl = wsParam.create_url()
    
    ws = websocket.WebSocketApp(
        wsUrl,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close,
        on_open=on_open
    )
    
    # 添加自定义属性
    ws.appid = appid
    ws.current_question = question
    ws.imagedata = imagedata  # 传递图像数据
    ws.answer = ""
    ws.is_closed = False  # 跟踪连接状态
    
    print("Sending request...")
    ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
    
    # 确保连接关闭
    if not ws.is_closed:
        ws.close()
    
    return ws.answer

if __name__ == '__main__':
    # 示例调用
    image_path = "Mind+.png"
    question = "请说出这张照片中有哪些内容？"
    
    print(f"Question: {question}")
    print("Answer:", end="")
    
    answer = main(
        appid=appid,
        api_key=api_key,
        api_secret=api_secret,
        imageunderstanding_url=imageunderstanding_url,
        question=question,
        image_path=image_path
    )
    
    print("\nAPI调用完成，程序结束")