one_data_set_control.py 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125
  1. from conifg_info import return_false_on_error
  2. # from mcu.McuDebug import McuDebug
  3. from utils.SingletonType import SingletonType
  4. # from module.base_mode.dynamic_parameter import *
  5. import json
  6. import ast
  7. import re
  8. import os
  9. class OneDataSet(metaclass=SingletonType):
  10. def __init__(self):
  11. super().__init__()
  12. self.mcu = None
  13. # self.mcu_debug = McuDebug(windows, is_debug=False, is_deviation=True)
  14. # self.init()
  15. # def init(self):
  16. # self.create_command_list()
  17. # 发送json给到mcu
  18. @return_false_on_error
  19. def deal_send_json_to_mcu(self, command):
  20. _text = command.strip()
  21. if _text.startswith("55"):
  22. flag = "cmd命令"
  23. elif _text.startswith("{"):
  24. flag = "json"
  25. else:
  26. flag = "cmd命令(无校验)"
  27. if flag == "json":
  28. try:
  29. print("json_text:{}".format(_text))
  30. # json_text 可能是文本格式的 JSON(如 {"code":1}),
  31. # 先解析成对象再重新序列化,规范化后再下发
  32. json_text = self.safe_load_json(_text)
  33. json_text = json.dumps(json_text)
  34. except Exception as e:
  35. print("错误:", str(e))
  36. return False
  37. print("json_text:{}".format(json_text))
  38. self.mcu.send_json_to(data=json_text)
  39. if flag == "cmd命令":
  40. self.mcu.send_hex_cmd_to(data=_text, need_del_cr=True)
  41. if flag == "cmd命令(无校验)":
  42. # 需要增加55 55 校验位
  43. self.mcu.send_hex_cmd_to(data=_text, need_del_cr=False)
  44. return True
  45. def safe_load_json(self, text: str):
  46. """
  47. 兼容各种非标准JSON字符串的解析函数
  48. 优先级: 标准JSON → 单引号替换 → Python字面量 → 宽松JSON
  49. """
  50. if not isinstance(text, str) or not text.strip():
  51. raise ValueError("输入必须是非空字符串")
  52. text = text.strip()
  53. # 1. 尝试标准JSON解析
  54. try:
  55. return json.loads(text)
  56. except json.JSONDecodeError:
  57. pass
  58. # 2. 单引号 → 双引号(仅当没有双引号冲突时)
  59. if "'" in text and '"' not in text:
  60. try:
  61. fixed = text.replace("'", '"')
  62. return json.loads(fixed)
  63. except json.JSONDecodeError:
  64. pass
  65. # 3. 尝试Python字面量解析(处理True/False/None/单引号等)
  66. try:
  67. result = ast.literal_eval(text)
  68. if isinstance(result, (dict, list)):
  69. return result
  70. except (ValueError, SyntaxError):
  71. pass
  72. # 4. 宽松JSON:处理尾部逗号、未转义单引号等
  73. try:
  74. # 移除尾部逗号
  75. cleaned = re.sub(r',\s*([}\]])', r'\1', text)
  76. # 将未转义的单引号替换为双引号(简单场景)
  77. cleaned = re.sub(r"(?<!\\)'", '"', cleaned)
  78. return json.loads(cleaned)
  79. except json.JSONDecodeError:
  80. pass
  81. raise ValueError(f"无法解析的JSON字符串: {text[:100]}")
  82. # 读取本地常用命令-JSON.txt,解析出名称和JSON内容
  83. def read_command_file(self):
  84. path = os.path.join(os.getcwd(), "command.txt")
  85. commands = []
  86. if not os.path.exists(path):
  87. return commands
  88. with open(path, "r", encoding="utf-8") as f:
  89. lines = f.readlines()
  90. name = None
  91. for line in lines:
  92. line = line.strip()
  93. if not line:
  94. continue
  95. if line.startswith("#"):
  96. name = line.lstrip("#").strip()
  97. continue
  98. if name is not None:
  99. commands.append((name, line))
  100. name = None
  101. return commands
  102. # 左侧渲染常用命令按钮列表
  103. def create_command_list(self):
  104. commands = self.read_command_file()
  105. json_dicts = []
  106. for name, json_text in commands:
  107. # print(name, json_text)
  108. json_dicts.append({"name": name, "command": json_text})
  109. return json.dumps(json_dicts)