using System;
|
using System.Collections.Generic;
|
using System.Diagnostics;
|
using System.Linq;
|
using System.Text.RegularExpressions;
|
using System.Threading;
|
using System.Threading.Tasks;
|
using Aging.Infrastructure.Contracts;
|
using CanSimulator;
|
using log4net;
|
|
namespace Aging.Infrastructure.Drivers
|
{
|
/// <summary>
|
/// 全系统唯一的 CAN 设备驱动。实现 IDeviceDriver，命令语义（SEND/QUERY）与原适配器一致。
|
/// Local 与 Broker 的差异下沉到构造时选择的 <see cref="ICanTransport"/> 后端：
|
/// Local → <see cref="NodeChannelTransport"/>（TCP 直连 CanSimulatorServer）
|
/// Broker → <see cref="BrokerRpcTransport"/>（命名管道委托 DeviceBroker 宿主）
|
/// 接收帧由后端 FramePushed 事件驱动，进入本地缓冲并触发 <see cref="DriverFrameReceived"/>，
|
/// 使调试面板与自动化程序对两种执行模式完全无感知。
|
/// </summary>
|
public sealed class CanDeviceDriver : IDeviceDriver
|
{
|
private static readonly ILog Log = LogManager.GetLogger(typeof(CanDeviceDriver));
|
|
// 信息型查询：响应去掉传输层的 OK: 前缀，还原为遗留 "VERB:payload" 形态（面板按此解析）
|
private static readonly Regex InfoVerb =
|
new(@"^OK:([A-Z][A-Z_]*):", RegexOptions.Compiled);
|
|
private readonly ICanTransport _transport;
|
private readonly string _busName;
|
private readonly int _nodeIndex;
|
private readonly Queue<CanFrame> _recvBuffer = new();
|
private readonly object _bufferLock = new();
|
|
/// <summary>宿主侧待捎带推送行（EV:RECV 原文），供 BrokerServer 随响应追加给客户端。</summary>
|
private readonly Queue<string> _pendingPushLines = new();
|
private const int MaxPendingPushLines = 1000;
|
private readonly Stopwatch _latency = new();
|
|
private bool _disposed;
|
private string _lastError = string.Empty;
|
private DateTime? _lastErrorTime;
|
private long _txCount;
|
private int _reconnectCount;
|
|
public string DeviceId { get; }
|
public bool IsConnected => !_disposed && _transport.IsConnected;
|
|
public event EventHandler<DeviceConnectionEventArgs>? ConnectionChanged;
|
|
/// <summary>接收帧即时事件（Local/Broker 统一由后端推送驱动）</summary>
|
public event EventHandler<CanFrame>? DriverFrameReceived;
|
|
/// <summary>Local 设备：从端点解析 bus:node，TCP 直连模拟器（host/port 取配置或覆盖值）。</summary>
|
public CanDeviceDriver(string deviceId, string endpoint,
|
string hostOverride = null, int? portOverride = null)
|
{
|
DeviceId = deviceId ?? throw new ArgumentNullException(nameof(deviceId));
|
(_busName, _nodeIndex) = ParseEndpoint(endpoint);
|
_transport = new NodeChannelTransport(hostOverride, portOverride, deviceId);
|
WireTransport();
|
}
|
|
/// <summary>Broker 设备：注入代理传输后端（由 DriverFactory 依 ExecutionMode 构造）。</summary>
|
internal CanDeviceDriver(string deviceId, string endpoint, ICanTransport transport)
|
{
|
DeviceId = deviceId ?? throw new ArgumentNullException(nameof(deviceId));
|
(_busName, _nodeIndex) = ParseEndpoint(endpoint);
|
_transport = transport ?? throw new ArgumentNullException(nameof(transport));
|
WireTransport();
|
}
|
|
/// <summary>Broker 设备代理工厂：客户端/Bootstrapper 据此构造委托宿主的统一驱动。</summary>
|
public static CanDeviceDriver CreateBrokerProxy(string deviceId, string endpoint, string pipeName, string configHash)
|
=> new CanDeviceDriver(deviceId, endpoint, new BrokerRpcTransport(deviceId, pipeName, configHash));
|
|
private void WireTransport()
|
{
|
_transport.FramePushed += OnFramePushed;
|
_transport.TransportBroken += () => OnConnection(false, "通道断开");
|
}
|
|
private static (string, int) ParseEndpoint(string endpoint)
|
{
|
if (string.IsNullOrWhiteSpace(endpoint))
|
throw new ArgumentException("CAN 端点不能为空", nameof(endpoint));
|
var parts = endpoint.Split(':');
|
if (parts.Length < 2 || !int.TryParse(parts[1], out int node) || node < 0)
|
throw new ArgumentException($"CAN 端点格式应为 总线名称:节点编号，实际: {endpoint}");
|
return (parts[0], node);
|
}
|
|
public async Task ConnectAsync(CancellationToken ct)
|
{
|
ThrowIfDisposed();
|
_latency.Restart();
|
try
|
{
|
await _transport.ConnectAsync(_busName, _nodeIndex, ct).ConfigureAwait(false);
|
_lastError = string.Empty;
|
OnConnection(true, "连接成功");
|
}
|
catch (Exception ex)
|
{
|
_lastError = ex.Message;
|
_lastErrorTime = DateTime.Now;
|
Log.Error($"[CAN] 设备 {DeviceId} 连接失败: {ex.Message}", ex);
|
throw;
|
}
|
finally
|
{
|
if (_latency.IsRunning) _latency.Stop();
|
}
|
}
|
|
public async Task DisconnectAsync()
|
{
|
ThrowIfDisposed();
|
try { await _transport.DisconnectAsync().ConfigureAwait(false); }
|
finally { OnConnection(false, "主动断开"); }
|
}
|
|
public async Task<string> SendCommandAsync(string command, CancellationToken ct)
|
{
|
ThrowIfDisposed();
|
await EnsureConnectedAsync(ct).ConfigureAwait(false);
|
_latency.Restart();
|
try
|
{
|
// 命令串原样透传（SEND:frameId:data）；响应去 OK: 传输前缀交由 QUERY 语义，
|
// 发送响应保留遗留 "OK:TX:…" 形态（上层与测试按 OK:TX 断言）
|
string resp = await _transport.TransceiveAsync(command, ct).ConfigureAwait(false);
|
Interlocked.Increment(ref _txCount);
|
return resp;
|
}
|
catch (Exception ex)
|
{
|
RecordError(ex);
|
throw;
|
}
|
finally
|
{
|
if (_latency.IsRunning) _latency.Stop();
|
}
|
}
|
|
public async Task<string> QueryAsync(string query, CancellationToken ct)
|
{
|
ThrowIfDisposed();
|
await EnsureConnectedAsync(ct).ConfigureAwait(false);
|
_latency.Restart();
|
try
|
{
|
string op = (query ?? "").Split(':')[0].ToUpperInvariant();
|
|
// 接收类查询由本地缓冲应答（推送已实时入队），无需往返模拟器
|
if (op == "RECV") return HandleRecv();
|
if (op == "RECV_ALL") return HandleRecvAll();
|
|
string resp = await _transport.TransceiveAsync("QUERY:" + query, ct).ConfigureAwait(false);
|
return NormalizeResponse(resp);
|
}
|
catch (Exception ex)
|
{
|
RecordError(ex);
|
throw;
|
}
|
finally
|
{
|
if (_latency.IsRunning) _latency.Stop();
|
}
|
}
|
|
public DeviceDriverStatus GetStatus() => new DeviceDriverStatus
|
{
|
IsConnected = IsConnected,
|
IsHealthy = IsConnected && string.IsNullOrEmpty(_lastError),
|
LastError = _lastError,
|
LastErrorTime = _lastErrorTime,
|
LatencyMs = _latency.IsRunning ? _latency.Elapsed.TotalMilliseconds : 0,
|
ReconnectCount = _reconnectCount
|
};
|
|
// ==================== 推送与缓冲 ====================
|
|
/// <summary>后端 EV:RECV 行 → 解析为帧 → 入本地缓冲 + 触发即时事件。</summary>
|
private void OnFramePushed(string line)
|
{
|
if (_disposed) return;
|
try
|
{
|
var frame = ParsePush(line);
|
if (frame == null) return;
|
|
lock (_bufferLock)
|
{
|
while (_recvBuffer.Count >= 1000) _recvBuffer.Dequeue();
|
_recvBuffer.Enqueue(frame);
|
|
// 累积原始推送行供宿主捎带给远端客户端（满则丢最老）
|
while (_pendingPushLines.Count >= MaxPendingPushLines) _pendingPushLines.Dequeue();
|
_pendingPushLines.Enqueue(line);
|
}
|
DriverFrameReceived?.Invoke(this, frame);
|
}
|
catch (Exception ex)
|
{
|
Log.Warn($"[CAN] {DeviceId} 推送帧解析失败: {ex.Message} [{line}]");
|
}
|
}
|
|
private static CanFrame ParsePush(string line)
|
{
|
// EV:RECV:{frameIdHex}:{dataHex}:{ts}:{srcNodeIndex}
|
if (!line.StartsWith("EV:RECV:", StringComparison.Ordinal)) return null;
|
var body = line.Substring("EV:RECV:".Length);
|
var parts = body.Split(':');
|
if (parts.Length < 2) return null;
|
|
uint id = CanFrame.ParseFrameId(parts[0]);
|
byte[] data = parts.Length >= 2 ? CanFrame.ParseData(parts[1]) : Array.Empty<byte>();
|
return new CanFrame
|
{
|
FrameId = id,
|
Format = id > 0x7FF ? CanFrameFormat.Extended : CanFrameFormat.Standard,
|
Type = CanFrameType.Data,
|
Data = data,
|
SourceNodeId = parts.Length >= 4 ? parts[3] : "",
|
Timestamp = DateTime.Now
|
};
|
}
|
|
private string HandleRecv()
|
{
|
lock (_bufferLock)
|
{
|
if (_recvBuffer.Count == 0) return "RECV:EMPTY";
|
var f = _recvBuffer.Dequeue();
|
return $"RECV:{f.FrameIdHex}:{f.DataHex}:{f.Timestamp:HH:mm:ss.fff}";
|
}
|
}
|
|
private string HandleRecvAll()
|
{
|
List<CanFrame> batch = new();
|
lock (_bufferLock)
|
{
|
while (_recvBuffer.Count > 0) batch.Add(_recvBuffer.Dequeue());
|
}
|
if (batch.Count == 0) return "RECV_ALL:0";
|
var items = batch.Select(f => $"{f.FrameIdHex}:{f.DataHex}");
|
return $"RECV_ALL:{batch.Count}:{string.Join("|", items)}";
|
}
|
|
/// <summary>
|
/// 信息型查询（VERB:…）去掉传输层 OK: 前缀还原遗留格式；
|
/// 动作确认（中文提示等）保留 OK:。
|
/// </summary>
|
private static string NormalizeResponse(string resp)
|
{
|
if (string.IsNullOrEmpty(resp)) return resp;
|
var m = InfoVerb.Match(resp);
|
if (m.Success)
|
return resp.Substring(3); // 去 "OK:"
|
return resp;
|
}
|
|
// ==================== 杂项 ====================
|
|
/// <summary>
|
/// 宿主专用：取出并清空自上次以来累积的推送行，供 BrokerServer 拼接到响应 readback 尾部。
|
/// </summary>
|
public List<string> DrainPendingPushLines()
|
{
|
lock (_bufferLock)
|
{
|
var list = _pendingPushLines.ToList();
|
_pendingPushLines.Clear();
|
return list;
|
}
|
}
|
|
private void EnsureConnected()
|
{
|
if (!IsConnected) throw new InvalidOperationException($"[CAN] 设备 {DeviceId} 未连接");
|
}
|
|
/// <summary>按需连接：首次 I/O 时若未连接则自动建立通道（契合“按需连接”设计）。</summary>
|
private async Task EnsureConnectedAsync(CancellationToken ct)
|
{
|
if (IsConnected) return;
|
await ConnectAsync(ct).ConfigureAwait(false);
|
}
|
|
private void RecordError(Exception ex)
|
{
|
_lastError = ex.Message;
|
_lastErrorTime = DateTime.Now;
|
Log.Error($"[CAN] 设备 {DeviceId} 操作失败: {ex.Message}", ex);
|
}
|
|
private void OnConnection(bool connected, string reason)
|
{
|
if (!connected) _reconnectCount++;
|
ConnectionChanged?.Invoke(this, new DeviceConnectionEventArgs
|
{
|
DeviceId = DeviceId,
|
IsConnected = connected,
|
Reason = reason
|
});
|
}
|
|
public void Dispose()
|
{
|
if (_disposed) return;
|
_disposed = true;
|
try
|
{
|
_transport.FramePushed -= OnFramePushed;
|
_transport.Dispose();
|
}
|
catch (Exception ex)
|
{
|
Log.Error($"[CAN] 设备 {DeviceId} 释放异常: {ex.Message}", ex);
|
}
|
}
|
|
private void ThrowIfDisposed()
|
{
|
if (_disposed) throw new ObjectDisposedException(nameof(CanDeviceDriver));
|
}
|
}
|
}
|
