pCanBridge 扩展多通道支持：每通道独立TCP链路 + 上下行对称通道标注(m_sSrcAux)
pCanBridge： - 配置改为 channel = <通道名>,<IP>,<工作端口> 重复行（兼容旧 can_host/can_port 单通道回退）， 每通道一条独立 CanEndpoint(TCP Client) + 接收线程，对应 CANET 的 CAN0=4001..CAN7=4008 - 上行 CAN_0x%08X 不变，m_sSrcAux=通道名区分来源通道 - 下行按 m_sSrcAux 路由到对应通道；回退规则：default_channel > 单通道 > 多通道告警丢弃 - buildReport 逐通道统计（连接/收发帧数/错误/重连） pCCU： - 新增 bcu_can_channel 配置（默认 CAN0），BCU 断路器下行 Post 前 SetSourceAux 指定目标通道， key 保持 CAN_TX_0x%08X（与上行通道标注对称，key 格式不变） 配置/文档：pCanBridge.moos、CanBridge_Info、pCCU.moos、missions/h100.moos、 test/pccu/pccu_it.moos 同步新格式。 验证：伪造 CANET Server 冒烟通过（多通道建链/上行解析发布/SQLite channel 列/ 下行通道路由与回退丢弃/srcAux 经 MOOSDB 往返）； pccu 集成测试 PASSED（BCU 断路器下行 3 帧经新路由全部正确）； pccuTest 183/183、PowerMangerTests 64/64 通过。
This commit is contained in:
+167
-53
@@ -14,11 +14,14 @@ namespace canbridge {
|
||||
CanBridge::CanBridge() {}
|
||||
|
||||
CanBridge::~CanBridge() {
|
||||
if (m_endpoint) {
|
||||
m_endpoint->stop();
|
||||
delete m_endpoint;
|
||||
m_endpoint = nullptr;
|
||||
for (size_t i = 0; i < m_channels.size(); ++i) {
|
||||
if (m_channels[i].endpoint) {
|
||||
m_channels[i].endpoint->stop();
|
||||
delete m_channels[i].endpoint;
|
||||
m_channels[i].endpoint = nullptr;
|
||||
}
|
||||
}
|
||||
m_channels.clear();
|
||||
if (m_db) {
|
||||
m_db->close();
|
||||
delete m_db;
|
||||
@@ -27,7 +30,11 @@ CanBridge::~CanBridge() {
|
||||
}
|
||||
|
||||
//---------------------------------------------------------
|
||||
// OnStartUp：读取配置并启动 CAN 链路
|
||||
// OnStartUp：读取配置并启动各通道 CAN 链路
|
||||
//
|
||||
// 通道配置（可重复多行）：
|
||||
// channel = <通道名>,<转换器IP>,<工作端口>
|
||||
// 兼容旧单通道配置：can_host / can_port / can_channel_name
|
||||
|
||||
bool CanBridge::OnStartUp() {
|
||||
AppCastingMOOSApp::OnStartUp();
|
||||
@@ -37,6 +44,11 @@ bool CanBridge::OnStartUp() {
|
||||
if (!m_MissionReader.GetConfiguration(GetAppName(), sParams))
|
||||
reportConfigWarning("No config block found for " + GetAppName());
|
||||
|
||||
// 旧单通道配置缓存（存在 can_host 时回退使用）
|
||||
bool hasLegacy = false;
|
||||
string legacyHost, legacyName = "CAN0";
|
||||
long legacyPort = 0;
|
||||
|
||||
STRING_LIST::iterator p;
|
||||
for (p = sParams.begin(); p != sParams.end(); p++) {
|
||||
string orig = *p;
|
||||
@@ -45,19 +57,55 @@ bool CanBridge::OnStartUp() {
|
||||
string value = stripBlankEnds(line);
|
||||
|
||||
bool handled = true;
|
||||
if (param == "can_host") m_canHost = value;
|
||||
else if (param == "can_port") m_canPort = atol(value.c_str());
|
||||
else if (param == "can_channel_name") m_channelName = value;
|
||||
else if (param == "dbpath") m_dbPath = value;
|
||||
if (param == "channel") {
|
||||
// channel = <通道名>,<IP>,<端口>
|
||||
string name = stripBlankEnds(biteStringX(value, ','));
|
||||
string host = stripBlankEnds(biteStringX(value, ','));
|
||||
string port = stripBlankEnds(value);
|
||||
ChannelCfg cfg;
|
||||
cfg.name = name;
|
||||
cfg.host = host;
|
||||
cfg.port = atol(port.c_str());
|
||||
if (cfg.name.empty() || cfg.host.empty() || cfg.port <= 0) {
|
||||
reportConfigWarning("bad channel line (want name,host,port): " + orig);
|
||||
} else {
|
||||
Channel ch;
|
||||
ch.cfg = cfg;
|
||||
m_channels.push_back(ch);
|
||||
}
|
||||
}
|
||||
else if (param == "default_channel") m_defaultChannel = value;
|
||||
else if (param == "dbpath") m_dbPath = value;
|
||||
// 旧单通道配置（向后兼容）
|
||||
else if (param == "can_host") { legacyHost = value; hasLegacy = true; }
|
||||
else if (param == "can_port") { legacyPort = atol(value.c_str()); hasLegacy = true; }
|
||||
else if (param == "can_channel_name") { legacyName = value; }
|
||||
else handled = false;
|
||||
|
||||
if (!handled)
|
||||
reportUnhandledConfigWarning(orig);
|
||||
}
|
||||
|
||||
// 无 channel 行但有旧配置 -> 构造单通道
|
||||
if (m_channels.empty() && hasLegacy &&
|
||||
!legacyHost.empty() && legacyPort > 0) {
|
||||
ChannelCfg cfg;
|
||||
cfg.name = legacyName;
|
||||
cfg.host = legacyHost;
|
||||
cfg.port = legacyPort;
|
||||
Channel ch;
|
||||
ch.cfg = cfg;
|
||||
m_channels.push_back(ch);
|
||||
reportConfigWarning("legacy can_host/can_port config used; "
|
||||
"prefer 'channel = name,host,port' lines");
|
||||
}
|
||||
|
||||
registerVariables();
|
||||
|
||||
// 数据库（CAN 帧落库）
|
||||
if (m_channels.empty())
|
||||
reportRunWarning("no CAN channel configured");
|
||||
|
||||
// 数据库（CAN 帧落库，channel 列区分通道）
|
||||
m_db = new CanDbStore(m_dbPath);
|
||||
if (!m_db->open()) {
|
||||
reportRunWarning("CAN db open failed: " + m_dbPath + " (" + m_db->lastError() + ")");
|
||||
@@ -65,19 +113,26 @@ bool CanBridge::OnStartUp() {
|
||||
m_db = nullptr;
|
||||
}
|
||||
|
||||
// CAN 链路（TCP Client，收帧回调在接收线程中触发）
|
||||
m_endpoint = new CanEndpoint();
|
||||
m_endpoint->configure(m_canHost, m_canPort);
|
||||
m_endpoint->setFrameCallback(
|
||||
[this](uint8_t fi, uint32_t id, const uint8_t* data, int dlc) {
|
||||
handleFrame(fi, id, data, dlc);
|
||||
});
|
||||
if (!m_endpoint->start()) {
|
||||
reportRunWarning("CAN endpoint start failed: " + m_canHost);
|
||||
// 各通道 CAN 链路（TCP Client，收帧回调在各自接收线程中触发）
|
||||
for (size_t i = 0; i < m_channels.size(); ++i) {
|
||||
Channel& ch = m_channels[i];
|
||||
ch.endpoint = new CanEndpoint();
|
||||
ch.endpoint->configure(ch.cfg.host, ch.cfg.port);
|
||||
const string name = ch.cfg.name; // 按通道捕获
|
||||
ch.endpoint->setFrameCallback(
|
||||
[this, name](uint8_t fi, uint32_t id, const uint8_t* data, int dlc) {
|
||||
handleFrame(name, fi, id, data, dlc);
|
||||
});
|
||||
if (!ch.endpoint->start())
|
||||
reportRunWarning("CAN endpoint start failed: " + ch.cfg.name +
|
||||
" (" + ch.cfg.host + ")");
|
||||
}
|
||||
|
||||
cout << "pCanBridge started: " << m_canHost << ":" << m_canPort
|
||||
<< " (" << m_channelName << "), db: " << m_dbPath << endl;
|
||||
cout << "pCanBridge started with " << m_channels.size() << " channel(s):";
|
||||
for (size_t i = 0; i < m_channels.size(); ++i)
|
||||
cout << " " << m_channels[i].cfg.name << "="
|
||||
<< m_channels[i].cfg.host << ":" << m_channels[i].cfg.port;
|
||||
cout << ", db: " << m_dbPath << endl;
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -91,7 +146,7 @@ bool CanBridge::OnConnectToServer() {
|
||||
|
||||
//---------------------------------------------------------
|
||||
// registerVariables：订阅下行指令（CAN_TX_0x*，pCCU 发布的
|
||||
// BCU 断路器控制帧等）。
|
||||
// BCU 断路器控制帧等，通道经 m_sSrcAux 指定）。
|
||||
|
||||
void CanBridge::registerVariables() {
|
||||
AppCastingMOOSApp::RegisterVariables();
|
||||
@@ -114,43 +169,93 @@ bool CanBridge::OnNewMail(MOOSMSG_LIST &NewMail) {
|
||||
}
|
||||
|
||||
//---------------------------------------------------------
|
||||
// handleTxMessage：MOOS 下行消息 -> CANET 13 字节帧 -> USBCAN
|
||||
// findChannel：按通道名查找运行通道
|
||||
|
||||
CanBridge::Channel* CanBridge::findChannel(const std::string& name) {
|
||||
for (size_t i = 0; i < m_channels.size(); ++i) {
|
||||
if (m_channels[i].cfg.name == name && m_channels[i].endpoint)
|
||||
return &m_channels[i];
|
||||
}
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
//---------------------------------------------------------
|
||||
// resolveTxChannel：下行目标通道解析（回退规则）
|
||||
// 1) srcAux 非空：命中通道则用之，否则告警丢弃
|
||||
// 2) srcAux 为空：default_channel 命中则用之
|
||||
// 3) 仍无：仅单通道时用该通道，多通道告警丢弃
|
||||
|
||||
CanBridge::Channel* CanBridge::resolveTxChannel(const std::string& srcAux) {
|
||||
if (!srcAux.empty()) {
|
||||
Channel* ch = findChannel(srcAux);
|
||||
if (!ch)
|
||||
reportRunWarning("CAN_TX unknown channel: '" + srcAux + "'");
|
||||
return ch;
|
||||
}
|
||||
|
||||
if (!m_defaultChannel.empty()) {
|
||||
Channel* ch = findChannel(m_defaultChannel);
|
||||
if (ch) return ch;
|
||||
}
|
||||
|
||||
if (m_channels.size() == 1)
|
||||
return m_channels[0].endpoint ? &m_channels[0] : nullptr;
|
||||
|
||||
reportRunWarning("CAN_TX no channel (empty srcAux) with " +
|
||||
to_string(m_channels.size()) + " channels, dropped");
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
//---------------------------------------------------------
|
||||
// handleTxMessage：MOOS 下行消息 -> CANET 13 字节帧 -> 对应通道
|
||||
//
|
||||
// m_sKey = "CAN_TX_0x%08X"（CAN ID）
|
||||
// m_sVal = 二进制数据域（dlc<=8）
|
||||
// m_sKey = "CAN_TX_0x%08X"（CAN ID）
|
||||
// m_sVal = 二进制数据域（dlc<=8）
|
||||
// m_sSrcAux = 目标通道名（空时按回退规则）
|
||||
// 帧信息：id>0x7FF 视为扩展帧(bit7=1)，DLC=数据字节数。
|
||||
|
||||
void CanBridge::handleTxMessage(CMOOSMsg& msg) {
|
||||
if (!m_endpoint) return;
|
||||
++m_txCount;
|
||||
|
||||
uint32_t id = 0;
|
||||
// "CAN_TX_0x" 前缀长 9 字符（注意不是 8）
|
||||
int sr = std::sscanf(msg.m_sKey.c_str() + 9, "%x", &id);
|
||||
if (sr != 1) {
|
||||
++m_txDropCount;
|
||||
reportRunWarning("CAN_TX bad key: " + msg.m_sKey);
|
||||
return;
|
||||
}
|
||||
if (!msg.IsBinary() || msg.GetBinaryDataSize() <= 0 ||
|
||||
msg.GetBinaryDataSize() > 8) {
|
||||
++m_txDropCount;
|
||||
reportRunWarning("CAN_TX bad data size: " + msg.m_sKey);
|
||||
return;
|
||||
}
|
||||
|
||||
Channel* ch = resolveTxChannel(msg.GetSourceAux());
|
||||
if (!ch) {
|
||||
++m_txDropCount;
|
||||
return;
|
||||
}
|
||||
|
||||
const unsigned char* d = msg.GetBinaryData();
|
||||
if (!d) return;
|
||||
if (!d) {
|
||||
++m_txDropCount;
|
||||
return;
|
||||
}
|
||||
int dlc = static_cast<int>(msg.GetBinaryDataSize());
|
||||
uint8_t frameInfo = static_cast<uint8_t>(((id > 0x7FF) ? 0x80 : 0x00) | dlc);
|
||||
|
||||
if (m_endpoint->sendFrame(frameInfo, id, d, dlc)) {
|
||||
++m_txCount;
|
||||
if (ch->endpoint->sendFrame(frameInfo, id, d, dlc)) {
|
||||
char hex[32] = {0};
|
||||
for (int i = 0; i < dlc && i < 8; ++i)
|
||||
std::snprintf(hex + i * 2, 3, "%02X", d[i]);
|
||||
std::cout << "[CAN-TX] 0x" << std::hex << id << std::dec
|
||||
<< " dlc=" << dlc << " data=" << hex << std::endl;
|
||||
std::cout << "[CAN-TX " << ch->cfg.name << "] 0x" << std::hex << id
|
||||
<< std::dec << " dlc=" << dlc << " data=" << hex << std::endl;
|
||||
} else {
|
||||
reportRunWarning("CAN_TX send failed: " + msg.m_sKey +
|
||||
(m_endpoint->isConnected() ? "" : " (未连接)"));
|
||||
reportRunWarning("CAN_TX send failed on " + ch->cfg.name + ": " +
|
||||
msg.m_sKey +
|
||||
(ch->endpoint->isConnected() ? "" : " (未连接)"));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -166,12 +271,12 @@ bool CanBridge::Iterate() {
|
||||
//---------------------------------------------------------
|
||||
// handleFrame：CAN 帧 -> CMOOSMsg（二进制） -> MOOSDB
|
||||
//
|
||||
// 在 CanEndpoint 接收线程中调用；m_Comms::Post 线程安全。
|
||||
// 在各通道 CanEndpoint 接收线程中调用；m_Comms::Post 线程安全。
|
||||
|
||||
void CanBridge::handleFrame(uint8_t frameInfo, uint32_t id,
|
||||
const uint8_t* data, int dlc) {
|
||||
void CanBridge::handleFrame(const std::string& channel, uint8_t frameInfo,
|
||||
uint32_t id, const uint8_t* data, int dlc) {
|
||||
// 落库（WAL + 预处理语句，~60帧/秒无压力）
|
||||
if (m_db) m_db->onFrame(m_channelName, frameInfo, id, data, dlc);
|
||||
if (m_db) m_db->onFrame(channel, frameInfo, id, data, dlc);
|
||||
|
||||
char key[32];
|
||||
std::snprintf(key, sizeof(key), "CAN_0x%08X", id);
|
||||
@@ -180,32 +285,41 @@ void CanBridge::handleFrame(uint8_t frameInfo, uint32_t id,
|
||||
CMOOSMsg msg(MOOS_NOTIFY, key,
|
||||
static_cast<unsigned int>(dlc),
|
||||
const_cast<uint8_t*>(data));
|
||||
msg.SetSourceAux(m_channelName); // 通道 -> m_sSrcAux
|
||||
msg.SetSourceAux(channel); // 通道 -> m_sSrcAux
|
||||
msg.SetDoubleAux(static_cast<double>(frameInfo)); // FF/RTR/DLC -> m_dfVal2
|
||||
m_Comms.Post(msg);
|
||||
++m_notifyCount;
|
||||
}
|
||||
|
||||
//---------------------------------------------------------
|
||||
// buildReport：AppCasting 报告
|
||||
// buildReport：AppCasting 报告（逐通道）
|
||||
|
||||
bool CanBridge::buildReport() {
|
||||
m_msgs << "============================================" << "\n";
|
||||
m_msgs << "pCanBridge CAN<->MOOSDB 双向透传" << "\n";
|
||||
m_msgs << "pCanBridge CAN<->MOOSDB 多通道双向透传" << "\n";
|
||||
m_msgs << "============================================" << "\n";
|
||||
if (m_endpoint) {
|
||||
m_msgs << "目标: " << m_endpoint->host() << ":"
|
||||
<< m_endpoint->port() << " (" << m_channelName << ")" << "\n";
|
||||
m_msgs << "连接状态: " << (m_endpoint->isConnected() ? "已连接" : "断开") << "\n";
|
||||
m_msgs << "收帧数: " << m_endpoint->frameCount() << "\n";
|
||||
m_msgs << "发布数: " << m_notifyCount << "\n";
|
||||
m_msgs << "下行帧数: " << m_endpoint->txCount() << " (错误 "
|
||||
<< m_endpoint->txErrorCount() << ")" << "\n";
|
||||
m_msgs << "MOOS下发数: " << m_txCount << "\n";
|
||||
m_msgs << "错误数: " << m_endpoint->errorCount() << "\n";
|
||||
m_msgs << "重连次数: " << m_endpoint->reconnectCount() << "\n";
|
||||
m_msgs << "最近收帧: " << m_endpoint->lastFrameTime() << "\n";
|
||||
m_msgs << "通道数: " << m_channels.size() << "\n";
|
||||
if (!m_defaultChannel.empty())
|
||||
m_msgs << "默认通道: " << m_defaultChannel << "\n";
|
||||
m_msgs << "MOOS下发数: " << m_txCount << " (丢弃 " << m_txDropCount << ")\n";
|
||||
|
||||
for (size_t i = 0; i < m_channels.size(); ++i) {
|
||||
const Channel& ch = m_channels[i];
|
||||
m_msgs << "--------------------------------------------" << "\n";
|
||||
m_msgs << "[" << ch.cfg.name << "] " << ch.cfg.host << ":"
|
||||
<< ch.cfg.port << "\n";
|
||||
if (ch.endpoint) {
|
||||
m_msgs << " 连接状态: "
|
||||
<< (ch.endpoint->isConnected() ? "已连接" : "断开") << "\n";
|
||||
m_msgs << " 收帧数: " << ch.endpoint->frameCount() << "\n";
|
||||
m_msgs << " 下行帧数: " << ch.endpoint->txCount()
|
||||
<< " (错误 " << ch.endpoint->txErrorCount() << ")\n";
|
||||
m_msgs << " 错误数: " << ch.endpoint->errorCount() << "\n";
|
||||
m_msgs << " 重连次数: " << ch.endpoint->reconnectCount() << "\n";
|
||||
m_msgs << " 最近收帧: " << ch.endpoint->lastFrameTime() << "\n";
|
||||
}
|
||||
}
|
||||
|
||||
m_msgs << "============================================" << "\n";
|
||||
if (m_db) {
|
||||
m_msgs << "数据库: " << m_dbPath
|
||||
<< (m_db->isOpen() ? " (已打开)" : " (未打开)") << "\n";
|
||||
|
||||
Reference in New Issue
Block a user