528 lines
18 KiB
Python
528 lines
18 KiB
Python
|
"""
|
|||
|
|
2026/4/10
|
|||
|
|
TMY
|
|||
|
|
在connect_renode_19.py的基础上，添加单例模式，并添加总线类型映射
|
|||
|
|
"""
|
|||
|
|
# midware/drivers/renode_agent.py
|
|||
|
|
import threading
|
|||
|
|
import binascii
|
|||
|
|
import subprocess
|
|||
|
|
import time
|
|||
|
|
import threading
|
|||
|
|
import re
|
|||
|
|
import statistics
|
|||
|
|
from typing import Optional, Callable, List, Dict
|
|||
|
|
from loguru import logger
|
|||
|
|
|
|||
|
|
# ======================
|
|||
|
|
# 这里放你原来 connect_renode_19.py 的全部代码
|
|||
|
|
# 只加 2 个东西：单例 + 总线类型映射
|
|||
|
|
# ======================
|
|||
|
|
# from . import DRIVER_MAPPING
|
|||
|
|
# from midware.config.base_config import DEVICE_CONFIG_DICT
|
|||
|
|
|
|||
|
|
BUS_TYPE_MAP = {
|
|||
|
|
0x11: "UART",
|
|||
|
|
0x22: "CAN",
|
|||
|
|
0x33: "1553B",
|
|||
|
|
0x44: "NET",
|
|||
|
|
0x55: "AD",
|
|||
|
|
0x66: "OC",
|
|||
|
|
}
|
|||
|
|
# 合法的总线类型第一个字节（字符串形式，方便判断）
|
|||
|
|
LEGAL_BUS_PREFIX = {"11", "22", "33", "44", "55", "66"}
|
|||
|
|
|
|||
|
|
class RenodeAgent:
|
|||
|
|
_instance = None
|
|||
|
|
_lock = threading.Lock()
|
|||
|
|
|
|||
|
|
def __new__(cls, *args, **kwargs):
|
|||
|
|
with cls._lock:
|
|||
|
|
if cls._instance is None:
|
|||
|
|
cls._instance = super().__new__(cls)
|
|||
|
|
return cls._instance
|
|||
|
|
|
|||
|
|
def __init__(self,
|
|||
|
|
renode_path="renode",
|
|||
|
|
script_path=None,
|
|||
|
|
max_queue=50, # 指令队列最大长度（限流）
|
|||
|
|
reconnect_retries=5, # 自动重连次数
|
|||
|
|
socket_timeout=3):
|
|||
|
|
if hasattr(self, "inited"):
|
|||
|
|
return
|
|||
|
|
self.inited = True
|
|||
|
|
|
|||
|
|
# ========= 你的原有代码 =========
|
|||
|
|
# self.process = ...
|
|||
|
|
# self.async_queue = ...
|
|||
|
|
# self.start_renode()
|
|||
|
|
# self.start_async_listener()
|
|||
|
|
# 基础配置
|
|||
|
|
self.renode_path = renode_path
|
|||
|
|
self.script_path = script_path
|
|||
|
|
self.process: Optional[subprocess.Popen] = None
|
|||
|
|
self.lock = threading.Lock()
|
|||
|
|
|
|||
|
|
# 输出与异步
|
|||
|
|
self.output_buffer: List[str] = []
|
|||
|
|
self.async_running = False
|
|||
|
|
self.async_callback: Optional[Callable[[str], None]] = None
|
|||
|
|
|
|||
|
|
# 限流队列
|
|||
|
|
self.max_queue = max_queue
|
|||
|
|
self.cmd_queue: List[str] = []
|
|||
|
|
self.busy = False
|
|||
|
|
|
|||
|
|
# 自动重连
|
|||
|
|
self.reconnect_retries = reconnect_retries
|
|||
|
|
self.connected = False
|
|||
|
|
|
|||
|
|
# 时延统计
|
|||
|
|
self.latency_stats: Dict[str, List[float]] = {
|
|||
|
|
"send": [], "process": [], "total": []
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# ====================== 超强过滤规则 ======================
|
|||
|
|
self.filter_re = re.compile(
|
|||
|
|
r"\(.*?\)" # 过滤 (machine) (TestPlatform) 等
|
|||
|
|
r"|\d{2}:\d{2}:\d{2}\.\d+" # 过滤时间 20:15:26.1234
|
|||
|
|
r"|renode>|\$|->" # 过滤提示符
|
|||
|
|
r"|\x1b\[[0-9;]*m" # 过滤颜色码
|
|||
|
|
r"|^\s*$" # 过滤空行
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
self.last_send_t =0.0
|
|||
|
|
self.send_min_gap = 0.01 #控制下发间隔
|
|||
|
|
|
|||
|
|
'''
|
|||
|
|
# 你原来的函数，完全不变
|
|||
|
|
def send_sync(self, cmd: str, timeout=1.0):
|
|||
|
|
# 你的原有逻辑
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
def enqueue_cmd(self, cmd: str):
|
|||
|
|
# 你的原有逻辑
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
def start_async_listener(self, callback):
|
|||
|
|
# 你的原有异步监听
|
|||
|
|
pass
|
|||
|
|
'''
|
|||
|
|
def start(self) -> bool:
|
|||
|
|
try:
|
|||
|
|
args = [
|
|||
|
|
self.renode_path,
|
|||
|
|
"--console",
|
|||
|
|
#"--disable-ansi-colors",
|
|||
|
|
"-e", "set echo off; set line-editing off; set raw on"
|
|||
|
|
]
|
|||
|
|
if self.script_path:
|
|||
|
|
args.extend(["-e", f"include @{self.script_path}"])
|
|||
|
|
#args.extend(["-e", f"{self.script_path}"])
|
|||
|
|
|
|||
|
|
self.process = subprocess.Popen(
|
|||
|
|
args,
|
|||
|
|
stdin=subprocess.PIPE,
|
|||
|
|
stdout=subprocess.PIPE,
|
|||
|
|
stderr=subprocess.STDOUT,
|
|||
|
|
bufsize=0,
|
|||
|
|
universal_newlines=True,
|
|||
|
|
creationflags=subprocess.CREATE_NEW_CONSOLE #LWB
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
self.async_running = True
|
|||
|
|
self.connected = True
|
|||
|
|
threading.Thread(target=self._read_thread, daemon=True).start()
|
|||
|
|
threading.Thread(target=self._queue_worker, daemon=True).start()
|
|||
|
|
time.sleep(1)
|
|||
|
|
self.clear_buffers()
|
|||
|
|
print("✅ 仿真平台 启动成功")
|
|||
|
|
|
|||
|
|
sysbus_cmd = "logLevel 3" #3级-只显示error
|
|||
|
|
response = renode.send_sync(sysbus_cmd)
|
|||
|
|
|
|||
|
|
return True
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"❌ 启动失败: {e}")
|
|||
|
|
self.connected = False
|
|||
|
|
return self._try_reconnect()
|
|||
|
|
|
|||
|
|
# -------------------------------------------------------------------------
|
|||
|
|
# 后台读取线程（永远实时读取，永不堆积）
|
|||
|
|
# -------------------------------------------------------------------------
|
|||
|
|
def _read_thread(self):
|
|||
|
|
while self.async_running and self.process.poll() is None:
|
|||
|
|
try:
|
|||
|
|
line = self.process.stdout.readline()
|
|||
|
|
if not line:
|
|||
|
|
time.sleep(0.001)
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
cleaned = self.filter_line(line)
|
|||
|
|
if not cleaned:
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
with self.lock:
|
|||
|
|
self.output_buffer.append(cleaned)
|
|||
|
|
|
|||
|
|
if self.async_callback:
|
|||
|
|
self.async_callback(cleaned)
|
|||
|
|
|
|||
|
|
except Exception:
|
|||
|
|
break
|
|||
|
|
|
|||
|
|
self.connected = False
|
|||
|
|
print("⚠️ Renode 已断开")
|
|||
|
|
self._try_reconnect()
|
|||
|
|
|
|||
|
|
# -------------------------------------------------------------------------
|
|||
|
|
# 过滤垃圾输出
|
|||
|
|
# -------------------------------------------------------------------------
|
|||
|
|
def filter_line(self, line: str) -> Optional[str]:
|
|||
|
|
res = self.filter_re.sub("", line).strip()
|
|||
|
|
return res if res else None
|
|||
|
|
|
|||
|
|
|
|||
|
|
# -------------------------------------------------------------------------
|
|||
|
|
# 同步发送（带时延测量 + 限流）,
|
|||
|
|
# -------------------------------------------------------------------------
|
|||
|
|
def send_sync(
|
|||
|
|
self,
|
|||
|
|
cmd: str,
|
|||
|
|
timeout_ms: int = 20
|
|||
|
|
) -> tuple[Optional[str], float, float, float]:
|
|||
|
|
if not self.connected or not self.process:
|
|||
|
|
return None, 0, 0, 0
|
|||
|
|
|
|||
|
|
# 限流：队列满则等待
|
|||
|
|
while len(self.cmd_queue) >= self.max_queue:
|
|||
|
|
time.sleep(0.001)
|
|||
|
|
|
|||
|
|
# self.clear_buffers() #同步发送不再清空全局输出缓冲区，缓冲区只有一部读取线程消费 2026/6/3
|
|||
|
|
t0 = time.perf_counter()
|
|||
|
|
|
|||
|
|
# 发送
|
|||
|
|
try:
|
|||
|
|
self.process.stdin.write(f"{cmd}\n")
|
|||
|
|
self.process.stdin.flush()
|
|||
|
|
except:
|
|||
|
|
return None, 0, 0, 0
|
|||
|
|
|
|||
|
|
t1 = time.perf_counter()
|
|||
|
|
|
|||
|
|
# 等待结果
|
|||
|
|
timeout = time.time() + timeout_ms / 1000
|
|||
|
|
while time.time() < timeout:
|
|||
|
|
with self.lock:
|
|||
|
|
if self.output_buffer:
|
|||
|
|
res = "\n".join(self.output_buffer)
|
|||
|
|
self.output_buffer.clear()
|
|||
|
|
t2 = time.perf_counter()
|
|||
|
|
|
|||
|
|
send_lat = (t1 - t0) * 1000
|
|||
|
|
proc_lat = (t2 - t1) * 1000
|
|||
|
|
total_lat = (t2 - t0) * 1000
|
|||
|
|
|
|||
|
|
self.latency_stats["send"].append(send_lat)
|
|||
|
|
self.latency_stats["process"].append(proc_lat)
|
|||
|
|
self.latency_stats["total"].append(total_lat)
|
|||
|
|
|
|||
|
|
return res, send_lat, proc_lat, total_lat
|
|||
|
|
time.sleep(0.001)
|
|||
|
|
|
|||
|
|
return None, 0, 0, 0
|
|||
|
|
|
|||
|
|
'''
|
|||
|
|
# -------------------------------------------------------------------------
|
|||
|
|
# 同步发送（有时延测量 + 限流，不等待回复）,
|
|||
|
|
# 2026/6/4
|
|||
|
|
# -------------------------------------------------------------------------
|
|||
|
|
def send_sync(
|
|||
|
|
self,
|
|||
|
|
cmd: str,
|
|||
|
|
timeout_ms: int = 20
|
|||
|
|
) -> tuple[Optional[str], float, float, float]:
|
|||
|
|
if not self.connected or not self.process:
|
|||
|
|
return None,0,0,0
|
|||
|
|
#队列限流等待
|
|||
|
|
while len(self.cmd_queue) >=self.max_queue:
|
|||
|
|
time.sleep(0.001)
|
|||
|
|
t0 = time.perf_counter()
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
self.process.stdin.write(f"{cmd}\n")
|
|||
|
|
self.process.stdin.flush()
|
|||
|
|
|
|||
|
|
except:
|
|||
|
|
return None,0,0,0
|
|||
|
|
|
|||
|
|
t1 = time.perf_counter()
|
|||
|
|
|
|||
|
|
#【关键改动：不在阻塞轮询收应答，直接退出；所有应答给dispacher消费】
|
|||
|
|
send_lat = (t1 - t0)*1000
|
|||
|
|
|
|||
|
|
return None,send_lat, 0 , send_lat
|
|||
|
|
'''
|
|||
|
|
# -------------------------------------------------------------------------
|
|||
|
|
# 异步监听
|
|||
|
|
# -------------------------------------------------------------------------
|
|||
|
|
def start_async_listener(self, callback: Callable[[str], None]):
|
|||
|
|
self.async_callback = callback
|
|||
|
|
|
|||
|
|
def clear_buffers(self):
|
|||
|
|
with self.lock:
|
|||
|
|
self.output_buffer.clear()
|
|||
|
|
|
|||
|
|
# -------------------------------------------------------------------------
|
|||
|
|
# 指令队列限流（后台异步发送）
|
|||
|
|
# -------------------------------------------------------------------------
|
|||
|
|
def enqueue_cmd(self, cmd: str):
|
|||
|
|
if len(self.cmd_queue) < self.max_queue:
|
|||
|
|
self.cmd_queue.append(cmd)
|
|||
|
|
'''
|
|||
|
|
def _queue_worker(self): #老版本，停用 2026、6、4
|
|||
|
|
while self.async_running:
|
|||
|
|
if not self.connected or self.busy or not self.cmd_queue:
|
|||
|
|
time.sleep(0.001)
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
self.busy = True
|
|||
|
|
cmd = self.cmd_queue.pop(0)
|
|||
|
|
self.send_sync(cmd, timeout_ms=1) #150->1
|
|||
|
|
self.busy = False
|
|||
|
|
'''
|
|||
|
|
'''
|
|||
|
|
def _queue_worker(self): #新版本，2026、6、4
|
|||
|
|
while self.async_running:
|
|||
|
|
if not self.connected or not self.cmd_queue:
|
|||
|
|
time.sleep(0.001)
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
cmd = self.cmd_queue.pop(0)
|
|||
|
|
self.send_sync(cmd) #150->1
|
|||
|
|
logger.info(f" _queue_worker向管道发送数据：{cmd}")
|
|||
|
|
'''
|
|||
|
|
def _queue_worker(self): #新版本，2026、6、4
|
|||
|
|
while self.async_running:
|
|||
|
|
now = time.time()
|
|||
|
|
#时间没有到，让出CPU
|
|||
|
|
if now - self.last_send_t < self.send_min_gap:
|
|||
|
|
time.sleep(0.0005)
|
|||
|
|
continue
|
|||
|
|
if not self.connected or not self.cmd_queue:
|
|||
|
|
time.sleep(0.001)
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
cmd = self.cmd_queue.pop(0)
|
|||
|
|
self.send_sync(cmd) #150->1
|
|||
|
|
logger.info(f" _queue_worker向管道发送数据：{cmd}")
|
|||
|
|
# -------------------------------------------------------------------------
|
|||
|
|
# 自动重连
|
|||
|
|
# -------------------------------------------------------------------------
|
|||
|
|
def _try_reconnect(self) -> bool:
|
|||
|
|
print(f"🔌 尝试自动重连 ({self.reconnect_retries} 次)")
|
|||
|
|
for i in range(self.reconnect_retries):
|
|||
|
|
self.stop()
|
|||
|
|
time.sleep(1)
|
|||
|
|
if self.start():
|
|||
|
|
print(f"✅ 第 {i+1} 次重连成功")
|
|||
|
|
return True
|
|||
|
|
time.sleep(1)
|
|||
|
|
print("❌ 重连全部失败")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
# -------------------------------------------------------------------------
|
|||
|
|
# 时延统计
|
|||
|
|
# -------------------------------------------------------------------------
|
|||
|
|
def get_latency_report(self) -> Dict[str, float]:
|
|||
|
|
def stats(arr):
|
|||
|
|
if not arr: return 0,0,0
|
|||
|
|
return round(statistics.mean(arr),2), round(max(arr),2), round(min(arr),2)
|
|||
|
|
return {
|
|||
|
|
"send_avg,max,min": stats(self.latency_stats["send"]),
|
|||
|
|
"process_avg,max,min": stats(self.latency_stats["process"]),
|
|||
|
|
"total_avg,max,min": stats(self.latency_stats["total"]),
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
def clear_latency(self):
|
|||
|
|
for k in self.latency_stats:
|
|||
|
|
self.latency_stats[k].clear()
|
|||
|
|
|
|||
|
|
# -------------------------------------------------------------------------
|
|||
|
|
# 停止
|
|||
|
|
# -------------------------------------------------------------------------
|
|||
|
|
def stop(self):
|
|||
|
|
self.async_running = False
|
|||
|
|
self.connected = False
|
|||
|
|
try:
|
|||
|
|
if self.process:
|
|||
|
|
self.process.stdin.write("quit\n")
|
|||
|
|
self.process.stdin.flush()
|
|||
|
|
time.sleep(0.2)
|
|||
|
|
self.process.terminate()
|
|||
|
|
except:
|
|||
|
|
pass
|
|||
|
|
self.process = None
|
|||
|
|
print("🔌 Renode 已停止")
|
|||
|
|
# midware/drivers/renode_agent.py
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 第三步：异步数据分发（完全不影响你现有架构）
|
|||
|
|
#在 renode_agent.py 里加全局分发回调，直接把数据投递到你现有的驱动 recv
|
|||
|
|
# def renode_async_data_dispatcher(line: str):
|
|||
|
|
# try:
|
|||
|
|
# # 1. 转成字节
|
|||
|
|
|
|||
|
|
# byte_data = binascii.unhexlify(line.strip())
|
|||
|
|
# print(byte_data)
|
|||
|
|
|
|||
|
|
# # 2. 解析头部 3 字节
|
|||
|
|
# if len(byte_data) < 3:
|
|||
|
|
# return
|
|||
|
|
# bus_type_id = byte_data[0]
|
|||
|
|
# machine_type = byte_data[1]
|
|||
|
|
# machine_id = byte_data[2]
|
|||
|
|
# payload = byte_data[3:]
|
|||
|
|
|
|||
|
|
# # 3. 映射协议类型
|
|||
|
|
# protocol = BUS_TYPE_MAP.get(bus_type_id)
|
|||
|
|
# if not protocol:
|
|||
|
|
# return
|
|||
|
|
|
|||
|
|
# # 4. 找到对应设备（根据单机编号）
|
|||
|
|
# dev_name = None
|
|||
|
|
# for name, cfg in DEVICE_CONFIG_DICT.items():
|
|||
|
|
# if cfg.get("协议类型") == protocol and cfg.get("单机编号") == machine_id:
|
|||
|
|
# dev_name = name
|
|||
|
|
# break
|
|||
|
|
# if not dev_name:
|
|||
|
|
# return
|
|||
|
|
|
|||
|
|
# # 5. 把数据 **直接放入你现有的总线接收队列**
|
|||
|
|
# from midware.core.bus_udp_decoupler import bus_recv_queues
|
|||
|
|
# q = bus_recv_queues[protocol]
|
|||
|
|
# q.put({
|
|||
|
|
# "dev_name": dev_name,
|
|||
|
|
# "raw_data": payload,
|
|||
|
|
# "protocol": protocol,
|
|||
|
|
# })
|
|||
|
|
|
|||
|
|
# except Exception:
|
|||
|
|
# pass
|
|||
|
|
|
|||
|
|
|
|||
|
|
def renode_async_data_dispatcher(line: str):
|
|||
|
|
try:
|
|||
|
|
# -----------------------
|
|||
|
|
# 🔥 延迟导入（放在函数内部，不会循环引用！）
|
|||
|
|
# -----------------------
|
|||
|
|
from midware.core.bus_udp_decoupler import bus_recv_queues
|
|||
|
|
from midware.config.base_config import DEVICE_CONFIG_DICT
|
|||
|
|
|
|||
|
|
# print(f"异步监听到的数据{line}")
|
|||
|
|
logger.info(
|
|||
|
|
f"异步监听到的数据{line}"
|
|||
|
|
)
|
|||
|
|
# -----------------------
|
|||
|
|
# 0.1. 基础过滤
|
|||
|
|
# -----------------------
|
|||
|
|
line = line.strip()
|
|||
|
|
if len(line) < 6: # 至少3字节 = 6个十六进制字符
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
# -----------------------
|
|||
|
|
# 0.2. 【关键】判断前2个字符是否是合法总线类型
|
|||
|
|
# -----------------------
|
|||
|
|
first_byte_str = line[:2]
|
|||
|
|
if first_byte_str not in LEGAL_BUS_PREFIX:
|
|||
|
|
# 不是我们要的总线数据 → 直接丢弃
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
# -----------------------
|
|||
|
|
# 0.3. 判断整个字符串是否是合法十六进制（安全转换）
|
|||
|
|
# -----------------------
|
|||
|
|
def is_hex(s):
|
|||
|
|
try:
|
|||
|
|
int(s, 16)
|
|||
|
|
return True
|
|||
|
|
except:
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
if not is_hex(line):
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
# 1. 把 Renode 返回的字符串转成字节，首先判断首字符是否为
|
|||
|
|
byte_data = bytes.fromhex(line.strip())
|
|||
|
|
|
|||
|
|
if len(byte_data) < 3:
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
# 2. 解析 3 字节头部
|
|||
|
|
bus_type_id = byte_data[0]
|
|||
|
|
machine_type = byte_data[1]
|
|||
|
|
machine_id = byte_data[2]
|
|||
|
|
payload = byte_data[3:]
|
|||
|
|
|
|||
|
|
# 3. 映射协议名
|
|||
|
|
protocol = BUS_TYPE_MAP.get(bus_type_id)
|
|||
|
|
if not protocol:
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
# 4. 根据 协议类型 + 单机编号 → 找到设备名
|
|||
|
|
dev_name = None
|
|||
|
|
for name, cfg in DEVICE_CONFIG_DICT.items():
|
|||
|
|
if (cfg.get("protocol") == protocol and
|
|||
|
|
cfg.get("number") == machine_id):
|
|||
|
|
dev_name = name
|
|||
|
|
break
|
|||
|
|
|
|||
|
|
if not dev_name:
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
# 5. 放入你现有的接收队列（完全兼容你原来架构）
|
|||
|
|
q = bus_recv_queues[protocol]
|
|||
|
|
try:
|
|||
|
|
q.put_nowait({
|
|||
|
|
"dev_name": dev_name,
|
|||
|
|
"raw_data": payload,
|
|||
|
|
"protocol": protocol,
|
|||
|
|
})
|
|||
|
|
except:
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
except Exception:
|
|||
|
|
import traceback
|
|||
|
|
traceback.print_exc()
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 在 renode_agent.py 末尾加这个函数
|
|||
|
|
def start_renode_uart_redirect():
|
|||
|
|
from midware.config.base_config import DEVICE_CONFIG_DICT
|
|||
|
|
renode = RenodeAgent() # 单例
|
|||
|
|
|
|||
|
|
for dev_name, cfg in DEVICE_CONFIG_DICT.items():
|
|||
|
|
tcp_local = cfg.get("tcp_local_port", 1)
|
|||
|
|
tcp_bus = cfg.get("tcp_bus_port", 1)
|
|||
|
|
|
|||
|
|
if tcp_local != 1 and tcp_bus != 1:
|
|||
|
|
term_name = f"term_{dev_name}"
|
|||
|
|
cmd1 = f'emulation CreateServerSocketTerminal {tcp_bus} "{term_name}"'#临时注释2026/6/2
|
|||
|
|
cmd2 = f'connector Connect sysbus.{dev_name} {term_name}'
|
|||
|
|
|
|||
|
|
print(f"[Renode] 配置UART重定向: {cmd1}")
|
|||
|
|
renode.send_sync(cmd1, timeout_ms=2)
|
|||
|
|
time.sleep(0.1)
|
|||
|
|
|
|||
|
|
print(f"[Renode] 连接外设: {cmd2}")
|
|||
|
|
renode.send_sync(cmd2, timeout_ms=2)
|
|||
|
|
time.sleep(0.2)
|
|||
|
|
print("[Renode] UART重定向配置完成！")
|
|||
|
|
|
|||
|
|
# 全局单例，全局唯一
|
|||
|
|
renode = RenodeAgent(
|
|||
|
|
renode_path = r"E:\simulation\1_simulation\卫星仿真平台_后端.exe", #r"E:\codes\1_simulation\卫星仿真平台_后端.exe", #r"C:\Users\PingCe\Desktop\1_simulation\卫星仿真平台_后端.exe", # Windows可写绝对路径如 C:/Renode/renode.exe
|
|||
|
|
script_path = r'E:\simulation\1_simulation\generated.resc',#r'E:\codes\1_simulation\generated_771_ADC1.resc', #r'include @C:\Users\PingCe\Desktop\1_simulation\generated_771_ADC1.resc',
|
|||
|
|
max_queue=25, # 最多同时排队5条指令，防止压爆
|
|||
|
|
reconnect_retries=5
|
|||
|
|
)
|