| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125 |
- from conifg_info import return_false_on_error
- # from mcu.McuDebug import McuDebug
- from utils.SingletonType import SingletonType
- # from module.base_mode.dynamic_parameter import *
- import json
- import ast
- import re
- import os
- class OneDataSet(metaclass=SingletonType):
- def __init__(self):
- super().__init__()
- self.mcu = None
- # self.mcu_debug = McuDebug(windows, is_debug=False, is_deviation=True)
- # self.init()
- # def init(self):
- # self.create_command_list()
- # 发送json给到mcu
- @return_false_on_error
- def deal_send_json_to_mcu(self, command):
- _text = command.strip()
- if _text.startswith("55"):
- flag = "cmd命令"
- elif _text.startswith("{"):
- flag = "json"
- else:
- flag = "cmd命令(无校验)"
- if flag == "json":
- try:
- print("json_text:{}".format(_text))
- # json_text 可能是文本格式的 JSON(如 {"code":1}),
- # 先解析成对象再重新序列化,规范化后再下发
- json_text = self.safe_load_json(_text)
- json_text = json.dumps(json_text)
- except Exception as e:
- print("错误:", str(e))
- return False
- print("json_text:{}".format(json_text))
- self.mcu.send_json_to(data=json_text)
- if flag == "cmd命令":
- self.mcu.send_hex_cmd_to(data=_text, need_del_cr=True)
- if flag == "cmd命令(无校验)":
- # 需要增加55 55 校验位
- self.mcu.send_hex_cmd_to(data=_text, need_del_cr=False)
- return True
- def safe_load_json(self, text: str):
- """
- 兼容各种非标准JSON字符串的解析函数
- 优先级: 标准JSON → 单引号替换 → Python字面量 → 宽松JSON
- """
- if not isinstance(text, str) or not text.strip():
- raise ValueError("输入必须是非空字符串")
- text = text.strip()
- # 1. 尝试标准JSON解析
- try:
- return json.loads(text)
- except json.JSONDecodeError:
- pass
- # 2. 单引号 → 双引号(仅当没有双引号冲突时)
- if "'" in text and '"' not in text:
- try:
- fixed = text.replace("'", '"')
- return json.loads(fixed)
- except json.JSONDecodeError:
- pass
- # 3. 尝试Python字面量解析(处理True/False/None/单引号等)
- try:
- result = ast.literal_eval(text)
- if isinstance(result, (dict, list)):
- return result
- except (ValueError, SyntaxError):
- pass
- # 4. 宽松JSON:处理尾部逗号、未转义单引号等
- try:
- # 移除尾部逗号
- cleaned = re.sub(r',\s*([}\]])', r'\1', text)
- # 将未转义的单引号替换为双引号(简单场景)
- cleaned = re.sub(r"(?<!\\)'", '"', cleaned)
- return json.loads(cleaned)
- except json.JSONDecodeError:
- pass
- raise ValueError(f"无法解析的JSON字符串: {text[:100]}")
- # 读取本地常用命令-JSON.txt,解析出名称和JSON内容
- def read_command_file(self):
- path = os.path.join(os.getcwd(), "command.txt")
- commands = []
- if not os.path.exists(path):
- return commands
- with open(path, "r", encoding="utf-8") as f:
- lines = f.readlines()
- name = None
- for line in lines:
- line = line.strip()
- if not line:
- continue
- if line.startswith("#"):
- name = line.lstrip("#").strip()
- continue
- if name is not None:
- commands.append((name, line))
- name = None
- return commands
- # 左侧渲染常用命令按钮列表
- def create_command_list(self):
- commands = self.read_command_file()
- json_dicts = []
- for name, json_text in commands:
- # print(name, json_text)
- json_dicts.append({"name": name, "command": json_text})
- return json.dumps(json_dicts)
|