feat: sync all micropython board configurations and scripts

mixly/boards/default/micropython_esp32c2/build/lib/mini_xunfei.py

@@ -0,0 +1,284 @@
+"""
+MINI_XUNFEI
+
+Micropython library for the MINI_XUNFEI(ASR, LLM)
+=======================================================
+@dahanzimin From the Mixly Team
+
+"""
+import time
+import hmac
+import json
+import hashlib
+import rtctime
+import websocket
+from mixgo_mini import onboard_bot
+from base64 import b64decode, b64encode
+from urllib import urlencode, urlparse
+
+class Ws_Param:
+    def __init__(self, APPID, APIKey, APISecret, Spark_url):
+        self.APPID = APPID
+        self.APIKey = APIKey
+        self.APISecret = APISecret
+        self.url = Spark_url
+        self.urlparse = urlparse(Spark_url)
+
+    def create_url(self):
+        date = rtctime.rfc1123_time()
+        signature_origin = "host: " + self.urlparse.netloc + "\n"
+        signature_origin += "date: " + date + "\n"
+        signature_origin += "GET " + self.urlparse.path + " HTTP/1.1"
+        signature_sha = hmac.new(self.APISecret.encode('utf-8'), signature_origin.encode('utf-8'), digestmod=hashlib.sha256).digest()
+        signature_base64 = b64encode(signature_sha).decode('utf-8')
+        authorization_origin = ('api_key="{}", algorithm="hmac-sha256", headers="host date request-line", signature="{}"'.format(self.APIKey, signature_base64))
+        authorization = b64encode(authorization_origin.encode('utf-8')).decode('utf-8')
+        headers = {"authorization": authorization, "date": date, "host": self.urlparse.netloc}
+        return self.url + '?' + urlencode(headers)
+
+#语音听写
+class ASR_WebSocket(Ws_Param):
+    def __init__(self, APPID, APIKey, APISecret, url='ws://iat-api.xfyun.cn/v2/iat'):
+        super().__init__(APPID, APIKey, APISecret, url)
+        self.ws = None
+        self.business = {
+                        "domain": "iat",
+                        "language": "zh_cn",
+                        "accent": "mandarin",
+                        "vad_eos": 1000,
+                        "nbest": 1,
+                        "wbest": 1,
+                    }
+
+    def connect(self):
+        self.ws = websocket.connect(self.create_url())
+        self.ws.settimeout(1000)
+
+    def _frame(self, status, buf):
+        if status == 0:
+            return {"common": {"app_id": self.APPID}, "business": self.business, "data": {"status": status, "format": "audio/L16;rate=8000", "audio": str(b64encode(buf), 'utf-8'), "encoding": "raw"}}
+        else:
+            return {"data": {"status": status, "format": "audio/L16;rate=8000", "audio": str(b64encode(buf), 'utf-8'), "encoding": "raw"}}
+
+    def on_message(self, message):
+        result = ""
+        msg = json.loads(message)
+        code = msg["code"]
+        if code != 0:
+            raise AttributeError("%s Code:%s" % (msg["message"], code))
+        else:
+            data = msg["data"]["result"]["ws"]
+            for i in data:
+                for w in i["cw"]:
+                    result += w["w"]
+            if msg["data"]["status"]== 2:
+                return result, False
+        return result, True 
+
+    def receive_messages(self):
+        msg = ""
+        while True:
+            t = self.on_message(self.ws.recv())
+            msg += t[0]
+            if not t[1]:
+                break
+        return msg
+
+    def run(self, seconds=3, pace=True, ibuf=1600, timeout=2000):
+        try:
+            _state = 0 
+            self.connect()
+            _star = time.ticks_ms()
+            _size = int(ibuf * seconds * 10) #100ms/次
+            onboard_bot.pcm_en(True) #PCM开启
+            if pace: print('[',end ="")
+            while _size > 0:
+                if onboard_bot.pcm_any():
+                    _size -= ibuf
+                    _star = time.ticks_ms()
+                    buf = onboard_bot.pcm_read(ibuf)
+                    if pace: print('=',end ="")
+                    # 第一帧处理
+                    if _state == 0:
+                        d = self._frame(_state, buf)
+                        _state = 1
+                    # 中间帧处理 
+                    else:
+                        d = self._frame(_state, buf)
+                    self.ws.send(json.dumps(d))
+                if time.ticks_diff(time.ticks_ms(), _star) > timeout:
+                    raise OSError("Timeout pcm read error")
+            # 最后一帧处理
+            d = self._frame(2, b'\x00')
+            self.ws.send(json.dumps(d))
+            onboard_bot.pcm_en(False)  #PCM关闭
+            if pace: print(']')
+            msg = self.receive_messages()
+            self.ws.close()
+            return msg
+        except Exception as e:
+            onboard_bot.pcm_en(False)  #PCM关闭
+            if "403 Forbidden" in str(e):
+                raise OSError("Access denied, Please try updating clock time")
+            else:
+                print("Run error: %s" % (e))
+
+#中英识别大模型
+class IAT_WebSocket(ASR_WebSocket):
+    def __init__(self, APPID, APIKey, APISecret, url='ws://iat.xf-yun.com/v1', accent="mandarin", res_id=None):
+        super().__init__(APPID, APIKey, APISecret, url)
+        self.res_id = res_id
+        self.business = {
+                        "domain": "slm",
+                        "language": "zh_cn",
+                        "accent": accent,
+                        "result": {
+                            "encoding": "utf8",
+                            "compress": "raw",
+                            "format": "plain"
+                        }
+                    }
+
+    def _frame(self, status, buf):
+        if status == 0:
+            return {"header": {"status": status, "app_id": self.APPID, "res_id": self.res_id}, "parameter": {"iat": self.business}, "payload": {"audio": { "audio": str(b64encode(buf), 'utf-8'), "sample_rate": 8000, "encoding": "raw"}}}
+        else:
+            return {"header": {"status": status, "app_id": self.APPID, "res_id": self.res_id}, "payload": {"audio": { "audio": str(b64encode(buf), 'utf-8'), "sample_rate": 8000, "encoding": "raw"}}}
+
+    def on_message(self, message):
+        result = ""
+        msg = json.loads(message)
+        code = msg['header']["code"]
+        if code != 0:
+            raise AttributeError("%s Code:%s" % (msg['header']["message"], code))
+        else:
+            if "payload" in msg:
+                text = msg["payload"]["result"]["text"]
+                data =  json.loads(b64decode(text).decode())['ws']
+                for i in data:
+                    for w in i["cw"]:
+                        result += w["w"]
+            if msg["header"]["status"]== 2:
+                return result, False
+        return result, True 
+
+#大模型
+class LLM_WebSocket(Ws_Param):
+    Model_url = {
+            "Spark Ultra-32K": ("ws://spark-api.xf-yun.com/v4.0/chat", "4.0Ultra"),
+            "Spark Max-32K": ("ws://spark-api.xf-yun.com/chat/max-32k", "max-32k"),
+            "Spark Max": ("ws://spark-api.xf-yun.com/v3.5/chat", "generalv3.5"),
+            "Spark Pro-128K": (" ws://spark-api.xf-yun.com/chat/pro-128k", "pro-128k"),
+            "Spark Pro": ("ws://spark-api.xf-yun.com/v3.1/chat", "generalv3"),
+            "Spark Lite": ("ws://spark-api.xf-yun.com/v1.1/chat", "lite"),
+            "Spark kjwx": ("ws://spark-openapi-n.cn-huabei-1.xf-yun.com/v1.1/chat_kjwx", "kjwx"),
+            "Spark X1-32K": ("ws://spark-api.xf-yun.com/v1/x1", "x1"),
+            "Spark Customize": ("ws://sparkcube-api.xf-yun.com/v1/customize", "max"),
+        }
+
+    def __init__(self, APPID, APIKey, APISecret, model='Spark Ultra-32K', system="你是知识渊博的助理，习惯简短表达", answers=50):
+        self.ws = None
+        self.answers = answers
+        self._url = self.Model_url[model] if model in self.Model_url else model
+        super().__init__(APPID, APIKey, APISecret, self._url[0])
+        self._function = [{}, []] #[回调函数, 功能描述]
+        self._messages = [{"role": "system", "content": system}]
+
+    def connect(self):
+        self.ws = websocket.connect(self.create_url())
+        self.ws.settimeout(1000)
+
+    def _params(self):
+        d = {
+            "header": {"app_id": self.APPID},
+            "parameter": {
+                "chat": {
+                    "domain": self._url[1],
+                    "random_threshold": 0.5,
+                    "max_tokens": 2048,
+                    "auditing": "default"
+                }
+            },
+            "payload": {
+                "message": {
+                    "text": self._messages
+                }
+            }
+        }
+        if self._function[1]:
+            d["payload"]["functions"] = {"text": self._function[1]}
+        self.ws.send(json.dumps(d))
+
+    def function_call(self, callback, name, description, params):
+        """功能回调名称, 描述, ((参数名, 类型, 描述), ...)"""
+        properties = {"type": "object", "properties":{}, "required":[]}
+        for arg in params:
+            if len(arg) >= 3:
+                properties["properties"][arg[0]] = {"type": arg[1], "description": arg[2]}
+                if arg[0] not in properties["required"]:
+                    properties["required"].append(arg[0])
+            else:
+                raise AttributeError('Invalid Input , format is (name, type, description)')
+        self._function[0][name] = callback
+        self._function[1].append({"name": name, "description": description, "parameters": properties})
+
+    def empty_history(self):
+        self._messages = []
+
+    def add_history(self, role, content):
+        self._messages.append({
+            "role": role,
+            "content": content
+        })
+
+    def on_message(self, message, reas):
+        result = ""
+        msg = json.loads(message)
+        code = msg['header']['code']
+        if code != 0:
+            raise AttributeError("%s Code:%s" % (msg["header"]["message"], code))
+        else:
+            choices = msg["payload"]["choices"]
+            text = choices["text"][0]
+            #推理
+            if "reasoning_content" in text and reas:
+                print("reasoning: ", text["reasoning_content"])
+            #回调
+            if "tool_calls" in text:
+                function = text['tool_calls'][0]['function']
+                if str(function['name']) in self._function[0] and function['arguments']:
+                    self._function[0][function['name']](json.loads(function['arguments']))
+            if "function_call" in text:
+                if str(text['function_call']['name']) in self._function[0] and text['function_call']['arguments']:
+                    self._function[0][text['function_call']['name']](json.loads(text['function_call']['arguments']))
+            #答复
+            if "content" in text:
+                result += text["content"]
+                if choices["status"] == 2:
+                    return result, False
+        return result, True 
+
+    def receive_messages(self, reas):
+        msg = ""
+        while True:
+            t = self.on_message(self.ws.recv(), reas)
+            msg += t[0]
+            if not t[1]:
+                break
+        return msg
+
+    def run(self, question, reas=True):
+        try:
+            self.connect()
+            self.add_history("user", question)
+            self._params()
+            while self.answers < len(self._messages):
+                del self._messages[0]
+            msg = self.receive_messages(reas)
+            self.ws.close()
+            return msg
+        except Exception as e:
+            if "403 Forbidden" in str(e):
+                raise OSError("Access denied, Please try updating clock time")
+            else:
+                print("Run error: %s" % (e))