root/Ayay
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加了通过网络主动上报图像的支持

Browse Source 
增加了指令维护通道的支持
This commit is contained in:
twice109
2026-01-07 10:59:03 +08:00
parent a697aab3e0
commit 3d47c8f009
47 changed files with 1613 additions and 1734 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
SHH.CameraDashboard/App.xaml.cs
View File

@@ -1,4 +1,5 @@
using SHH.CameraDashboard.Services;
using SHH.Contracts;
using SHH.ProcessLaunchers;
using System.Collections.ObjectModel;
using System.Windows;
@@ -31,6 +32,7 @@ namespace SHH.CameraDashboard
// 启动指令服务 (Port 6001)
CommandServer.Instance.Start(6001);
CommandServer.Instance.OnClientRegistered += SetupAutomaticConfiguration;
// 现在我们来配置启动
@@ -57,7 +59,9 @@ namespace SHH.CameraDashboard
string serviceArgs = $"" +
$"--pid {myPid} " +
$"--appid \"CameraApp_01\" " +
$"--uris \"127.0.0.1,6002&6001;\" " +
$"--uris \"127.0.0.1,6002,video,调试PC;\" " +
$"--uris \"127.0.0.1,6001,command,调试PC;\" " +
$"--uris \"192.168.1.100,6002,video,大屏展示;\" " +
$"--mode 1 " +
$"--ports \"5000,100\"";
@@ -86,6 +90,78 @@ namespace SHH.CameraDashboard
mainWin.Show();
}
/// <summary>
/// 在程序启动时订阅事件
/// </summary>
/// <param name="obj"></param>
private void SetupAutomaticConfiguration(ConnectedClient obj)
{
// 监听注册事件：每当有 Service (CommandClientWorker) 连上来注册成功
CommandServer.Instance.OnClientRegistered += (client) =>
{
Console.WriteLine($"[自动化] 检测到新服务上线: {client.ServiceId} ({client.Ip})");
// 放到线程池去执行，避免阻塞 UI 或网络接收线程
Task.Run(async () =>
{
// 1. 稍微延时一点点 (500ms)，给 Service 一点喘息时间准备接收指令
await Task.Delay(500);
// 2. 构造您指定的“206摄像头”配置
var cameraConfig = new CameraConfigDto
{
Id = 17798,
Name = "206摄像头",
Location = "404办公室",
IpAddress = "172.16.41.88",
Username = "admin",
Password = "abcd1234",
Port = 8000,
ChannelIndex = 1,
StreamType = 0,
Brand = DeviceBrand.HikVision.GetHashCode(), // 对应 DeviceBrand 枚举
RenderHandle = 0, // 初始化为0
MainboardIp = "", // 留空
MainboardPort = 0,
RtspPath = ""
};
// ★ 新增：一并带上订阅要求 ★
cameraConfig.AutoSubscriptions = new List<CameraConfigSubscribeDto>
{
// 第一条：显示帧，要求 8 帧
new CameraConfigSubscribeDto {
AppId = "UI_Display",
Type = 0,
TargetFps = 8,
Memo = "显示帧"
},
// 第二条：分析帧，要求 1 帧
new CameraConfigSubscribeDto {
AppId = "AI_Analysis",
Type = 0,
Memo = "分析帧",
TargetFps = 1
}
};
// 3. 封装协议包
var commandPacket = new
{
Action = "SyncCamera", // 告诉 Service 执行什么动作
Payload = cameraConfig, // 数据载荷
Time = DateTime.Now
};
// 4. 定向发送
// client.ServiceId 就是那个 "CameraApp_01"
CommandServer.Instance.SendCommand(client.ServiceId, commandPacket);
Console.WriteLine($"[自动化] 已向 {client.ServiceId} 下发配置: 206摄像头");
});
};
}
/// <summary>
/// 全局统一退出入口
/// </summary>

178
SHH.CameraDashboard/Invokes/CommandServer.cs
View File

@@ -1,40 +1,48 @@
using NetMQ;
using NetMQ.Sockets;
using Newtonsoft.Json;
using System;
using Newtonsoft.Json.Linq;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Text;
using System.Threading.Tasks;
namespace SHH.CameraDashboard.Services;
namespace SHH.CameraDashboard;
/// <summary>
/// [Dashboard端] 指令控制服务
/// 职责：双向通信通道。接收 Service 心跳/响应，向 Service 发送控制指令。
/// 核心模式：ROUTER (Dashboard) <--> DEALER (Service)
/// 职责：监听 6001 端口，接收 CameraService 的注册/心跳，并下发控制指令。
/// </summary>
public class CommandServer : IDisposable
{
// 单例模式
public static CommandServer Instance { get; } = new CommandServer();
// 事件：收到消息时触发 (ServiceId, MessageContent)
// =================================================================
// 事件定义
// =================================================================
// 当有新设备注册成功时触发 (UI 可以订阅这个来刷新列表)
public event Action<ConnectedClient>? OnClientRegistered;
// 当收到通用业务消息时触发
public event Action<string, string>? OnMessageReceived;
// =================================================================
// 内部成员
// =================================================================
private RouterSocket? _routerSocket;
private NetMQPoller? _poller;
// 【关键新增】发送队列：用于解决跨线程发送的安全问题
// UI线程 -> Enqueue -> Poller线程 -> Socket.Send
private NetMQQueue<CommandPacket>? _sendQueue;
// 在线设备表 (Key: Identity/AppId)
// 线程安全字典，存储客户端的详细信息（包括视频地址）
private readonly ConcurrentDictionary<string, ConnectedClient> _clients = new();
public int ListenPort { get; private set; }
public bool IsRunning => _poller != null && _poller.IsRunning;
// 在线设备表 (可选，用于记录谁在线)
// Key: ServiceId (Identity字符串)
private readonly ConcurrentDictionary<string, DateTime> _onlineClients = new();
// 获取当前所有在线客户端的副本
public List<ConnectedClient> GetClients() => _clients.Values.ToList();
private CommandServer() { }
@@ -47,17 +55,15 @@ public class CommandServer : IDisposable
{
// 1. 初始化 Router Socket
_routerSocket = new RouterSocket();
_routerSocket.Bind($"tcp://*:{ListenPort}");
_routerSocket.Bind($"tcp://*:{ListenPort}"); // 监听所有网卡
_routerSocket.ReceiveReady += OnSocketReady;
// 2. 初始化发送队列
// 2. 初始化发送队列 (确保 UI 线程可以安全发送)
_sendQueue = new NetMQQueue<CommandPacket>();
_sendQueue.ReceiveReady += OnQueueReady;
// 3. 启动 Poller (同时监听 Socket 接收 和 队列发送)
// 3. 启动 Poller
_poller = new NetMQPoller { _routerSocket, _sendQueue };
// RunAsync 会自动开启后台线程
_poller.RunAsync();
Console.WriteLine($"[Dashboard] 指令服务启动，监听: tcp://*:{ListenPort}");
@@ -65,33 +71,37 @@ public class CommandServer : IDisposable
catch (Exception ex)
{
Console.WriteLine($"[Dashboard] 指令端口绑定失败: {ex.Message}");
throw; // 必须抛出，让 App 感知
throw;
}
}
/// <summary>
/// 处理来自 Service 的网络消息 (运行在 Poller 线程)
/// [Poller线程] 处理网络接收
/// </summary>
private void OnSocketReady(object? sender, NetMQSocketEventArgs e)
{
try
{
// 1. 读取身份帧 (Identity)
// 只要 Service 端 DealerSocket 设置了 Identity，这里收到就是那个 ID
// Router 接收逻辑：
// Frame 1: 发送者的 Identity (NetMQ 自动处理)
// Frame 2: 真实数据
// 1. 读取身份 (Identity)
var identityBytes = e.Socket.ReceiveFrameBytes();
string serviceId = Encoding.UTF8.GetString(identityBytes);
string serviceId = Encoding.UTF8.GetString(identityBytes); // e.g., "CameraApp_01"
// 2. 读取内容帧 (假设 Dealer 直接发内容，中间无空帧)
// 如果你使用了 REQ/REP 模式，中间可能会有空帧，需注意兼容
// 2. 读取消息内容
// 兼容性处理：有些 Dealer 实现可能会发空帧，这里做个简单尝试
// 如果发现在 Identity 后紧跟的是空帧，则再读一帧
// 但在我们目前的 Dealer 实现中，是直接发的 JSON
string message = e.Socket.ReceiveFrameString();
if (string.IsNullOrWhiteSpace(message))
{
if (e.Socket.HasIn) message = e.Socket.ReceiveFrameString();
}
// 3. 简单的心跳保活逻辑
_onlineClients[serviceId] = DateTime.Now;
// 4. 触发业务事件
// 注意：这依然在 Poller 线程，UI 处理时需 Invoke
Console.WriteLine($"[指令] From {serviceId}: {message}");
OnMessageReceived?.Invoke(serviceId, message);
// 3. 协议解析与业务分发
ProcessMessage(serviceId, message, identityBytes);
}
catch (Exception ex)
{
@@ -100,37 +110,99 @@ public class CommandServer : IDisposable
}
/// <summary>
/// 处理发送队列 (运行在 Poller 线程)
/// 核心业务逻辑处理
/// </summary>
private void OnQueueReady(object? sender, NetMQQueueEventArgs<CommandPacket> e)
private void ProcessMessage(string serviceId, string json, byte[] identityBytes)
{
try
{
if (_routerSocket == null) return;
// 尝试解析基础结构
var jObj = JObject.Parse(json);
string action = jObj["Action"]?.ToString() ?? "Unknown";
// 从队列取出一个包
if (e.Queue.TryDequeue(out var packet, TimeSpan.Zero))
// 更新最后心跳时间 (如果已存在)
if (_clients.TryGetValue(serviceId, out var existingClient))
{
// Router 发送标准三步走：
// 1. 发送目标 Identity (More = true)
// 2. 发送空帧 (可选，取决于协议约定，Router-Dealer 直连通常不需要空帧)
// 3. 发送数据 (More = false)
existingClient.LastHeartbeat = DateTime.Now;
}
// 这里我们采用最简协议：[Identity][Data]
_routerSocket.SendMoreFrame(packet.TargetId)
.SendFrame(packet.JsonData);
Console.WriteLine($"[指令] To {packet.TargetId}: {packet.JsonData}");
// ★★★ 处理注册握手 ★★★
if (action == "Register")
{
HandleRegistration(serviceId, jObj, identityBytes);
}
else
{
// 其他业务消息，透传给上层
Console.WriteLine($"[指令] From {serviceId}: {json}");
OnMessageReceived?.Invoke(serviceId, json);
}
}
catch (Exception ex)
catch (JsonException)
{
Debug.WriteLine($"[Command Send Error] {ex.Message}");
Console.WriteLine($"[指令] 收到非 JSON 消息 From {serviceId}: {json}");
}
}
/// <summary>
/// 发送指令 (线程安全，可由 UI 线程调用)
/// 处理注册逻辑
/// </summary>
private void HandleRegistration(string serviceId, JObject jObj, byte[] identityBytes)
{
var payload = jObj["Payload"];
if (payload == null) return;
// 1. 提取客户端信息
var client = new ConnectedClient
{
ServiceId = serviceId,
Ip = payload["Ip"]?.ToString() ?? "Unknown",
// ★★★ 解析新字段 ★★★
WebPort = payload["WebPort"]?.Value<int>() ?? 5000,
Version = payload["Version"]?.ToString() ?? "Unknown",
Pid = payload["Pid"]?.Value<int>() ?? 0,
TargetVideoNodes = payload["TargetVideoNodes"]?.ToObject<List<string>>() ?? new List<string>(),
LastHeartbeat = DateTime.Now
};
// 2. 存入内存表 (Add or Update)
_clients.AddOrUpdate(serviceId, client, (key, old) => client);
Console.WriteLine($"[注册成功] {serviceId}");
// 3. 回复 ACK (握手确认)
// 告诉客户端：我收到你的注册了，连接建立成功
var ackPacket = new { Action = "ACK", Message = $"Registered {serviceId}", Time = DateTime.Now };
string ackJson = JsonConvert.SerializeObject(ackPacket);
// 直接在 Poller 线程发回，不需要走 Queue (因为拥有 Socket 所有权)
_routerSocket?.SendMoreFrame(identityBytes).SendFrame(ackJson);
// 4. 通知 UI 更新列表
OnClientRegistered?.Invoke(client);
}
/// <summary>
/// [Poller线程] 处理发送队列
/// </summary>
private void OnQueueReady(object? sender, NetMQQueueEventArgs<CommandPacket> e)
{
if (_routerSocket == null) return;
if (e.Queue.TryDequeue(out var packet, TimeSpan.Zero))
{
// Router 发送：[Identity] [Data]
_routerSocket.SendMoreFrame(packet.TargetId)
.SendFrame(packet.JsonData);
Console.WriteLine($"[发送] To {packet.TargetId}: {packet.JsonData}");
}
}
/// <summary>
/// [公共API] 向指定 Service 发送指令
/// </summary>
public void SendCommand(string targetServiceId, object commandData)
{
@@ -138,7 +210,6 @@ public class CommandServer : IDisposable
var json = JsonConvert.SerializeObject(commandData);
// ★★★ 核心修复：不直接操作 Socket，而是入队 ★★★
_sendQueue.Enqueue(new CommandPacket
{
TargetId = targetServiceId,
@@ -152,13 +223,12 @@ public class CommandServer : IDisposable
_poller?.Dispose();
_routerSocket?.Dispose();
_sendQueue?.Dispose();
_poller = null;
_routerSocket = null;
_sendQueue = null;
}
// 内部数据包结构
// =============================================================
// 数据模型
// =============================================================
private class CommandPacket
{
public string TargetId { get; set; } = "";

30
SHH.CameraDashboard/Invokes/ConnectedClient.cs Normal file
View File

@@ -0,0 +1,30 @@
namespace SHH.CameraDashboard;
/// <summary>
/// 在线客户端信息模型 (已更新)
/// </summary>
public class ConnectedClient
{
/// <summary> 唯一标识 (AppId) </summary>
public string ServiceId { get; set; } = string.Empty;
/// <summary> 版本号 </summary>
public string Version { get; set; } = "1.0.0";
/// <summary> 远程进程 ID </summary>
public int Pid { get; set; }
/// <summary> 客户端 IP </summary>
public string Ip { get; set; } = string.Empty;
/// <summary> WebAPI 端口 (Dashboard 调用 REST 接口用) </summary>
public int WebPort { get; set; }
/// <summary> 该客户端正在推流的目标地址 </summary>
public List<string> TargetVideoNodes { get; set; } = new List<string>();
public DateTime LastHeartbeat { get; set; }
// 辅助属性：拼接出完整的 API BaseUrl
public string WebApiUrl => $"http://{Ip}:{WebPort}";
}

109
SHH.CameraDashboard/Invokes/StreamReceiverService.cs
View File

@@ -1,67 +1,54 @@
using NetMQ;
using NetMQ.Sockets;
using System.Diagnostics; // 用于 Debug 输出
using Newtonsoft.Json;
using SHH.Contracts; // ★★★ 必须引用契约库 ★★★
using System.Diagnostics;
namespace SHH.CameraDashboard.Services;
namespace SHH.CameraDashboard;
public class StreamReceiverService : IDisposable
{
// 单例模式
public static StreamReceiverService Instance { get; } = new StreamReceiverService();
public event Action<string, byte[]>? OnFrameReceived;
// ★★★ 核心变更：使用强类型契约载体 ★★★
public event Action<VideoPayload>? OnPayloadReceived;
private SubscriberSocket? _subSocket;
private Task? _receiveTask;
// 【修复1】不要在这里初始化，改为在 Start 中初始化
private CancellationTokenSource? _cts;
public int ListenPort { get; private set; }
// 增加运行状态标记
// 运行状态检查
public bool IsRunning => _receiveTask != null && !_receiveTask.IsCompleted;
private StreamReceiverService() { }
public void Start(int port = 6000)
{
// 1. 防止重复启动
if (IsRunning) return;
ListenPort = port;
// 【修复1】每次启动时创建新的 TokenSource
_cts = new CancellationTokenSource();
try
{
// 2. 初始化 Socket
_subSocket = new SubscriberSocket();
// 【优化】设置高水位限制 (HWM)
// 如果 UI 处理不过来，积压超过 1000 帧直接丢弃，防止内存爆炸
// 设置高水位，防止 UI 卡顿时内存溢出
_subSocket.Options.ReceiveHighWatermark = 1000;
_subSocket.Bind($"tcp://*:{ListenPort}");
_subSocket.Subscribe(""); // 订阅所有内容（这是 Dealer-Router/Pub-Sub 的基础）
string bindAddr = $"tcp://*:{ListenPort}";
_subSocket.Bind(bindAddr);
_subSocket.Subscribe("");
Console.WriteLine($"[Dashboard] 视频流接收服务启动: {bindAddr}");
Console.WriteLine($"[Dashboard] 视频流接收服务启动: tcp://*:{ListenPort}");
}
catch (Exception ex)
{
Console.WriteLine($"[Dashboard] 致命错误 - 端口绑定失败: {ex.Message}");
// 清理资源
// 明确抛出异常，让 App.xaml.cs 知道启动失败了
_subSocket?.Dispose();
_subSocket = null;
// 【修复4】抛出异常让上层知道启动失败了
throw new Exception($"端口 {port} 绑定失败，可能被占用。", ex);
}
// 3. 启动任务
_receiveTask = Task.Run(ReceiveLoop, _cts.Token);
}
@@ -73,61 +60,55 @@ public class StreamReceiverService : IDisposable
{
try
{
// 【修复2】线程安全检查
if (_subSocket == null) break;
// 接收 Topic
if (!_subSocket.TryReceiveFrameString(TimeSpan.FromMilliseconds(500), out string cameraId))
// =========================================================
// 核心解析逻辑：适配 Service 端的 4 帧复合协议
// =========================================================
NetMQMessage msg = new NetMQMessage();
// 1. 非阻塞接收多帧消息
if (!_subSocket.TryReceiveMultipartMessage(TimeSpan.FromMilliseconds(500), ref msg))
continue;
// 接收 Payload
if (!_subSocket.TryReceiveFrameBytes(TimeSpan.FromMilliseconds(100), out byte[] jpgBytes))
continue;
// 2. 协议完整性检查
if (msg.FrameCount < 4) continue;
// 触发事件
OnFrameReceived?.Invoke(cameraId, jpgBytes);
}
catch (ObjectDisposedException)
{
// 【修复2】这是正常的退出流程（Socket被Dispose了），优雅退出循环
break;
// 3. 协议头校验 (Frame 0)
if (msg[0].ConvertToString() != "SHH_V1") continue;
// 4. 反序列化元数据 (Frame 1)
string json = msg[1].ConvertToString();
var payload = JsonConvert.DeserializeObject<VideoPayload>(json);
if (payload == null) continue;
// 5. 填充二进制图像数据 (Frame 2 & 3)
// 注意：NetMQ 的 msg 数据是非托管内存，转为 byte[] 实现了拷贝，安全供 UI 使用
if (payload.HasOriginalImage)
payload.OriginalImageBytes = msg[2].ToByteArray();
if (payload.HasTargetImage)
payload.TargetImageBytes = msg[3].ToByteArray();
// 6. 触发事件
OnPayloadReceived?.Invoke(payload);
}
catch (Exception ex)
{
// 记录日志，但不崩溃
Debug.WriteLine($"[ReceiverLoop Error] {ex.Message}");
Debug.WriteLine($"[Receiver Error] {ex.Message}");
}
}
Console.WriteLine("[Dashboard] 接收循环已停止");
}
/// <summary>
/// 停止服务（支持停止后重新 Start）
/// </summary>
public void Stop()
{
// 1. 发出取消信号
if (_cts != null && !_cts.IsCancellationRequested)
{
_cts.Cancel();
}
// 2. 销毁 Socket (这会触发 ReceiveLoop 中的 ObjectDisposedException 从而退出循环)
if (_subSocket != null)
{
try { _subSocket.Dispose(); } catch { }
_subSocket = null;
}
// 3. 清理 Token
_cts?.Dispose();
_cts = null;
_cts?.Cancel();
try { _subSocket?.Dispose(); } catch { }
_subSocket = null;
_receiveTask = null;
}
public void Dispose()
{
Stop();
}
public void Dispose() => Stop();
}

55
SHH.CameraDashboard/Pages/CameraWall/VideoTileViewModel.cs
View File

@@ -1,6 +1,6 @@
using System.Windows;
using System.Windows.Media.Imaging;
using SHH.CameraDashboard.Services; // 引用服务命名空间
using SHH.Contracts; // ★★★ 引用契约库 (VideoPayload) ★★★
namespace SHH.CameraDashboard;
@@ -8,7 +8,7 @@ public class VideoTileViewModel : ViewModelBase
{
private readonly string _boundCameraId;
// --- 属性定义 ---
// --- 属性定义 (保持不变) ---
private string _cameraName;
public string CameraName
{
@@ -37,38 +37,59 @@ public class VideoTileViewModel : ViewModelBase
CameraName = name;
StatusInfo = "等待信号...";
// 【修正 1】直接订阅单例服务
// 不需要判断 null，因为 Instance 是静态初始化的，永远存在
StreamReceiverService.Instance.OnFrameReceived += OnGlobalFrameReceived;
// ★★★ 变更 1: 订阅新的 OnPayloadReceived 事件 ★★★
// 旧的 OnFrameReceived(string, byte[]) 已经无法满足需求
StreamReceiverService.Instance.OnPayloadReceived += OnPayloadReceived;
}
// --- 事件回调 (后台线程) ---
private void OnGlobalFrameReceived(string cameraId, byte[] jpgData)
// ★★★ 变更 2: 参数变为 VideoPayload 实体对象 ★★★
private void OnPayloadReceived(VideoPayload payload)
{
// 1. 过滤：不是我的画面，直接忽略
if (cameraId != _boundCameraId) return;
// 1. 过滤：校验 Payload 中的 CameraId
if (payload.CameraId != _boundCameraId) return;
// 2. 解码：耗时操作在后台完成
var bitmap = BitmapHelper.ToBitmapImage(jpgData);
// 2. ★★★ 智能选图策略 ★★★
// 优先显示 AI 处理后的图 (TargetImageBytes)
// 如果没有处理图，则降级显示原始图 (OriginalImageBytes)
byte[] dataToShow = null;
if (payload.HasTargetImage && payload.TargetImageBytes != null)
{
dataToShow = payload.TargetImageBytes;
}
else if (payload.HasOriginalImage && payload.OriginalImageBytes != null)
{
dataToShow = payload.OriginalImageBytes;
}
// 如果两张图都没有，直接返回
if (dataToShow == null || dataToShow.Length == 0) return;
// 3. 解码图片 (耗时操作在后台完成)
var bitmap = BitmapHelper.ToBitmapImage(dataToShow);
if (bitmap == null) return;
// 3. 【修正 2】恢复 UI 更新逻辑
// 必须使用 Dispatcher，因为 VideoSource 绑定在界面上，只能在主线程修改
// 4. ★★★ 计算端到端延迟 ★★★
// 当前时间(接收端) - 采集时间(发送端) = 真实的网络+处理延迟
long latency = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() - payload.CaptureTimestamp;
// 5. UI 更新
Application.Current.Dispatcher.InvokeAsync(() =>
{
VideoSource = bitmap;
// 更新状态信息 (例如显示当前时间和数据大小)
StatusInfo = $"{DateTime.Now:HH:mm:ss} | {jpgData.Length / 1024} KB";
// 显示更丰富的信息：延迟毫秒数、数据量、当前时间
// 工业监控中，"延迟(ms)" 是比 "当前时间" 更重要的指标
StatusInfo = $"延迟: {latency}ms | {dataToShow.Length / 1024} KB | {DateTime.Now:HH:mm:ss}";
});
}
// --- 资源清理 ---
public void Unload()
{
// 【修正 3】从单例服务取消订阅
// 这一步至关重要，否则切换页面时会内存泄漏
StreamReceiverService.Instance.OnFrameReceived -= OnGlobalFrameReceived;
// ★★★ 变更 3: 取消订阅新的事件 ★★★
StreamReceiverService.Instance.OnPayloadReceived -= OnPayloadReceived;
// 清空图片引用，帮助 GC 回收内存
VideoSource = null;

30
SHH.CameraDashboard/Pages/CameraWall/VideoWallViewModel.cs
View File

@@ -1,14 +1,10 @@
using SHH.Contracts;
using System.Collections.ObjectModel;
using System.Collections.ObjectModel;
using System.Windows.Input;
namespace SHH.CameraDashboard
{
public class VideoWallViewModel : ViewModelBase
{
// 引用推流接收服务
private readonly VideoPushServer _pushServer;
// 视频列表
public ObservableCollection<VideoTileViewModel> VideoTiles { get; } = new ObservableCollection<VideoTileViewModel>();
@@ -27,35 +23,11 @@ namespace SHH.CameraDashboard
{
SetLayoutCommand = new RelayCommand<string>(ExecuteSetLayout);
// 1. 初始化并启动接收服务
_pushServer = new VideoPushServer();
_pushServer.OnFrameReceived += OnGlobalFrameReceived;
// 2. 启动监听端口 (比如 6000)
// 之后你的采集端 ForwarderClient 需要 Connect("tcp://你的IP:6000")
_pushServer.Start(6000);
// 3. 初始化格子 (不再需要传入 IP/Port 去主动连接了)
// 我们用 CameraId 或 Name 来作为匹配标识
InitVideoTiles();
}
/// <summary>
/// 全局接收回调：收到任何一路视频都会进这里
/// </summary>
private void OnGlobalFrameReceived(VideoPayload payload)
{
// 1. 在 VideoTiles 集合中找到对应的格子
// 假设 payload.CameraId 与我们 VideoTileViewModel 中的 ID 对应
//var targetTile = VideoTiles.FirstOrDefault(t => t.id == payload.CameraId);
//if (targetTile != null)
//{
// // 2. 将数据交给格子去渲染
// targetTile.UpdateFrame(payload);
//}
}
private void InitVideoTiles()
{
// 假设我们预设 4 个格子，分别对应不同的摄像头 ID

16
SHH.CameraSdk/Abstractions/Enums/SubscriptionType.cs
View File

@@ -8,38 +8,44 @@ namespace SHH.CameraSdk;
/// </summary>
public enum SubscriptionType
{
/// <summary>
/// 仅提供流
/// </summary>
[Description("仅提供流")]
Stream = 0,
/// <summary>
/// 本地窗口渲染
/// <para>直接在服务器端显示器绘制（如 OpenCV Window、WinForm 控件）</para>
/// </summary>
[Description("本地窗口显示")]
LocalWindow = 0,
LocalWindow = 1,
/// <summary>
/// 本地录像存储
/// <para>写入磁盘文件（如 MP4/AVI 格式，支持定时切割、循环覆盖）</para>
/// </summary>
[Description("本地录像存储")]
LocalRecord = 1,
LocalRecord = 2,
/// <summary>
/// 句柄绑定显示
/// <para>渲染到指定 HWND 窗口句柄（如 SDK 硬件解码渲染到客户端控件）</para>
/// </summary>
[Description("句柄绑定显示")]
HandleDisplay = 2,
HandleDisplay = 3,
/// <summary>
/// 自定义网络传输
/// <para>通过私有协议转发给第三方系统（如工控机、告警服务器）</para>
/// </summary>
[Description("自定义网络传输")]
NetworkTrans = 3,
NetworkTrans = 4,
/// <summary>
/// 网页端推流
/// <para>转码为 Web 标准协议（如 WebRTC、HLS、RTMP）供浏览器播放</para>
/// </summary>
[Description("网页端推流")]
WebPush = 4
WebPush = 5
}

224
SHH.CameraSdk/Configs/ServiceConfig.cs
View File

@@ -1,227 +1,141 @@
namespace SHH.CameraSdk;
/// <summary>
/// 全局服务配置模型 (V3 最终版)
/// <para>负责解析命令行参数，构建网络拓扑和身份标识</para>
/// </summary>
public class ServiceConfig
{
// ==========================================
// 1. 身份与进程属性
// 1. 基础属性
// ==========================================
/// <summary>
/// 父进程 PID (用于哨兵守护，--pid)
/// </summary>
public int ParentPid { get; private set; }
/// <summary>
/// 应用完整标识 (例如 "CameraApp_01", --appid)
/// </summary>
public string AppId { get; private set; } = "Unknown_01";
/// <summary>
/// 【核心】从 AppId 自动提取的数字编号
/// <para>规则：取最后一个下划线后的数字</para>
/// <para>示例："CameraApp_05" -> 5</para>
/// </summary>
public int NumericId { get; private set; } = 1;
// ==========================================
// 2. 网络连接属性 (分流)
// ==========================================
/// <summary>
/// 视频流目标地址列表 (对应 & 符号左侧)
/// <para>ZeroMQBridgeWorker 使用此列表</para>
/// </summary>
public List<string> VideoEndpoints { get; private set; } = new List<string>();
/// <summary>
/// 指令控制目标地址列表 (对应 & 符号右侧)
/// <para>CommandClientWorker 使用此列表</para>
/// </summary>
public List<string> CommandEndpoints { get; private set; } = new List<string>();
/// <summary>
/// WebAPI 基础端口 (--ports 的第一个值)
/// </summary>
public int BasePort { get; private set; } = 5000;
/// <summary>
/// 端口扫描范围 (--ports 的第二个值)
/// </summary>
public int MaxPortRange { get; private set; } = 100;
/// <summary>
/// 网络模式 (--mode)
/// </summary>
public NetworkMode Mode { get; private set; } = NetworkMode.Passive;
// ==========================================
// 3. 辅助属性
// ==========================================
/// <summary>
/// 是否需要执行 Connect 操作
/// </summary>
public bool ShouldConnect => Mode == NetworkMode.Active || Mode == NetworkMode.Hybrid;
// ==========================================
// 4. 解析入口 (Factory Method)
// 2. 目标地址列表 (类型变了！)
// ==========================================
// ★★★ 修改点：从 List<string> 变为 List<ServiceEndpoint> ★★★
public List<ServiceEndpoint> VideoEndpoints { get; private set; } = new List<ServiceEndpoint>();
public List<ServiceEndpoint> CommandEndpoints { get; private set; } = new List<ServiceEndpoint>();
// ==========================================
// 3. 工厂方法 (保持不变)
// ==========================================
public static ServiceConfig BuildFromArgs(string[] args)
{
var config = new ServiceConfig();
for (int i = 0; i < args.Length; i++)
{
// 1. 预处理 Key
var key = args[i].ToLower().Trim();
var value = (i + 1 < args.Length && !args[i + 1].StartsWith("--")) ? args[i + 1] : string.Empty;
bool consumed = !string.IsNullOrEmpty(value);
// 2. 预取 Value (如果存在且不是下一个 flag)
var value = (i + 1 < args.Length) ? args[i + 1] : string.Empty;
// 简单判断：如果 value 以 -- 开头，说明当前 key 是开关，或者参数值缺失
if (value.StartsWith("--")) value = string.Empty;
bool consumed = false; // 标记是否消耗了下一个参数
// 3. 匹配参数
switch (key)
{
case "--pid":
if (int.TryParse(value, out int pid)) config.ParentPid = pid;
consumed = true;
break;
case "--pid": if (int.TryParse(value, out int pid)) config.ParentPid = pid; break;
case "--appid":
if (!string.IsNullOrWhiteSpace(value))
{
config.AppId = value;
// ★★★ 立即解析数字编号 ★★★
config.NumericId = ParseIdFromAppId(value);
}
consumed = true;
break;
case "--uris":
if (!string.IsNullOrWhiteSpace(value))
{
// ★★★ 解析复杂 URI 字符串 ★★★
ParseUris(config, value);
}
consumed = true;
if (!string.IsNullOrWhiteSpace(value)) ParseSingleUriConfig(config, value);
break;
case "--mode":
if (int.TryParse(value, out int m) && Enum.IsDefined(typeof(NetworkMode), m))
{
config.Mode = (NetworkMode)m;
}
consumed = true;
break;
case "--mode": if (int.TryParse(value, out int m)) config.Mode = (NetworkMode)m; break;
case "--ports":
// 格式: "BasePort,Range" -> "6003,100"
if (!string.IsNullOrWhiteSpace(value) && value.Contains(","))
{
var parts = value.Split(',');
if (parts.Length >= 1)
{
if (int.TryParse(parts[0], out int baseP)) config.BasePort = baseP;
}
if (parts.Length >= 2)
{
if (int.TryParse(parts[1], out int range)) config.MaxPortRange = range;
}
if (parts.Length >= 1 && int.TryParse(parts[0], out int p)) config.BasePort = p;
if (parts.Length >= 2 && int.TryParse(parts[1], out int r)) config.MaxPortRange = r;
}
consumed = true;
break;
}
// 4. 如果消耗了 Value，跳过下一个索引
if (consumed) i++;
}
return config;
}
// ==========================================
// 5. 核心解析算法实现
// ==========================================
/// <summary>
/// 算法：提取下划线后的数字
/// </summary>
private static int ParseIdFromAppId(string appId)
{
if (string.IsNullOrWhiteSpace(appId)) return 1;
// 查找最后一个下划线
int lastIdx = appId.LastIndexOf('_');
// 确保下划线存在，且后面还有字符
if (lastIdx >= 0 && lastIdx < appId.Length - 1)
{
string numPart = appId.Substring(lastIdx + 1);
if (int.TryParse(numPart, out int id))
{
return id;
}
if (int.TryParse(appId.Substring(lastIdx + 1), out int id)) return id;
}
// 解析失败默认返回 1
return 1;
}
/// <summary>
/// 算法：解析 URI 列表并分流
/// <para>格式: IP,VideoPort&CommandPort</para>
/// <para>空缺处理: "&6001" (仅指令), "6002&" (仅视频)</para>
/// </summary>
private static void ParseUris(ServiceConfig config, string rawValue)
// ==========================================
// 4. 解析算法实现 (核心修改)
// ==========================================
private static void ParseSingleUriConfig(ServiceConfig config, string rawValue)
{
// 1. 按分号拆分不同主机配置
// "127.0.0.1,6002&6001; 192.168.1.5,&6001"
var groups = rawValue.Split(new[] { ';' }, StringSplitOptions.RemoveEmptyEntries);
var segments = rawValue.Split(new[] { ';' }, StringSplitOptions.RemoveEmptyEntries);
foreach (var group in groups)
foreach (var segment in segments)
{
// 2. 按逗号拆分 IP 和 端口段
var hostParts = group.Split(',');
if (hostParts.Length < 2) continue; // 格式非法
var parts = segment.Split(',');
if (parts.Length < 3) continue;
string ip = hostParts[0].Trim();
string portSection = hostParts[1].Trim(); // "6002&6001"
string ip = parts[0].Trim();
string portStr = parts[1].Trim();
string type = parts[2].Trim().ToLower();
// 3. 按 & 拆分端口 (注意：不要 RemoveEmptyEntries，位置很重要)
var ports = portSection.Split('&');
// ★★★ 提取第四个字段作为备注 ★★★
string desc = parts.Length >= 4 ? parts[3].Trim() : "未命名终端";
// --- 索引 0: 视频端口 ---
if (ports.Length > 0)
if (int.TryParse(portStr, out int port))
{
string p = ports[0].Trim();
if (!string.IsNullOrWhiteSpace(p) && int.TryParse(p, out int port))
string zmqUri = $"tcp://{ip}:{port}";
// 构建对象
var endpoint = new ServiceEndpoint
{
string uri = $"tcp://{ip}:{port}";
if (!config.VideoEndpoints.Contains(uri))
config.VideoEndpoints.Add(uri);
Uri = zmqUri,
Description = desc
};
// 添加前检查 Uri 是否重复 (备注不参与排重)
if (type == "video")
{
if (!config.VideoEndpoints.Any(e => e.Uri == zmqUri))
config.VideoEndpoints.Add(endpoint);
}
}
// --- 索引 1: 指令端口 ---
if (ports.Length > 1)
{
string p = ports[1].Trim();
if (!string.IsNullOrWhiteSpace(p) && int.TryParse(p, out int port))
else if (type == "command" || type == "text")
{
string uri = $"tcp://{ip}:{port}";
if (!config.CommandEndpoints.Contains(uri))
config.CommandEndpoints.Add(uri);
if (!config.CommandEndpoints.Any(e => e.Uri == zmqUri))
config.CommandEndpoints.Add(endpoint);
}
}
}
}
}
/// <summary>
/// [新增] 端点配置对象，包含地址和备注
/// </summary>
public class ServiceEndpoint
{
/// <summary>
/// ZeroMQ 连接地址 (e.g. "tcp://127.0.0.1:6001")
/// </summary>
public string Uri { get; set; } = string.Empty;
/// <summary>
/// 备注信息 (e.g. "调试机", "大屏")
/// </summary>
public string Description { get; set; } = string.Empty;
/// <summary>
/// ToString
/// </summary>
/// <returns></returns>
public override string ToString() => $"{Uri} ({Description})";
}

1
SHH.CameraSdk/Controllers/CamerasController.cs
View File

@@ -1,4 +1,5 @@
using Microsoft.AspNetCore.Mvc;
using SHH.Contracts;
namespace SHH.CameraSdk;

125
SHH.CameraSdk/Controllers/Dto/CameraConfigDto.cs
View File

@@ -1,125 +0,0 @@
using System.ComponentModel.DataAnnotations;
namespace SHH.CameraSdk;
// ==============================================================================
// 1. 物理与运行配置 DTO (对应 CRUD 操作)
// 用于设备新增/全量配置查询，包含基础身份、连接信息、运行参数等全量字段
// ==============================================================================
public class CameraConfigDto
{
// --- 基础身份 (Identity) ---
/// <summary>
/// 设备唯一标识
/// </summary>
[Required(ErrorMessage = "设备ID不能为空")]
[Range(1, long.MaxValue, ErrorMessage = "设备ID必须为正整数")]
public long Id { get; set; }
/// <summary>
/// 设备友好名称
/// </summary>
[MaxLength(64, ErrorMessage = "设备名称长度不能超过64个字符")]
public string Name { get; set; } = string.Empty;
/// <summary>
/// 摄像头品牌类型 (0:HikVision, 1:Dahua, 2:RTSP...)
/// </summary>
[Range(0, 10, ErrorMessage = "品牌类型值必须在0-10范围内")]
public int Brand { get; set; }
/// <summary>
/// 设备安装位置描述
/// </summary>
[MaxLength(128, ErrorMessage = "安装位置长度不能超过128个字符")]
public string Location { get; set; } = string.Empty;
// --- 核心连接 (Connectivity) - 修改此类参数触发冷重启 ---
/// <summary>
/// 摄像头IP地址
/// </summary>
[Required(ErrorMessage = "IP地址不能为空")]
[RegularExpression(@"^((25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}(25[0-5]|2[0-4]\d|[01]?\d\d?)$",
ErrorMessage = "请输入合法的IPv4地址")]
public string IpAddress { get; set; } = string.Empty;
/// <summary>
/// SDK端口 (如海康默认8000)
/// </summary>
[Range(1, 65535, ErrorMessage = "端口号必须在1-65535范围内")]
public ushort Port { get; set; } = 8000;
/// <summary>
/// 登录用户名
/// </summary>
[MaxLength(32, ErrorMessage = "用户名长度不能超过32个字符")]
public string Username { get; set; } = string.Empty;
/// <summary>
/// 登录密码
/// </summary>
[MaxLength(64, ErrorMessage = "密码长度不能超过64个字符")]
public string Password { get; set; } = string.Empty;
public long RenderHandle { get; set; }
/// <summary>
/// 通道号 (通常为1)
/// </summary>
[Range(1, 32, ErrorMessage = "通道号必须在1-32范围内")]
public int ChannelIndex { get; set; } = 1;
/// <summary>
/// RTSP流路径 (备用或非SDK模式使用)
/// </summary>
[MaxLength(256, ErrorMessage = "RTSP地址长度不能超过256个字符")]
public string RtspPath { get; set; } = string.Empty;
// --- 主板关联信息 (Metadata) ---
/// <summary>
/// 关联主板IP地址
/// </summary>
[RegularExpression(@"^((25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}(25[0-5]|2[0-4]\d|[01]?\d\d?)?$",
ErrorMessage = "请输入合法的IPv4地址")]
public string MainboardIp { get; set; } = string.Empty;
/// <summary>
/// 关联主板端口
/// </summary>
[Range(1, 65535, ErrorMessage = "主板端口号必须在1-65535范围内")]
public int MainboardPort { get; set; } = 80;
// --- 运行时参数 (Runtime Options) - 支持热更新 ---
/// <summary>
/// 码流类型 (0:主码流, 1:子码流)
/// </summary>
[Range(0, 1, ErrorMessage = "码流类型只能是0(主码流)或1(子码流)")]
public int StreamType { get; set; } = 0;
/// <summary>
/// 是否使用灰度图 (用于AI分析场景加速)
/// </summary>
public bool UseGrayscale { get; set; } = false;
/// <summary>
/// 是否启用图像增强 (去噪/锐化等)
/// </summary>
public bool EnhanceImage { get; set; } = true;
// --- 画面变换 (Transform) - 支持热更新 ---
/// <summary>
/// 是否允许图像压缩 (降低带宽占用)
/// </summary>
public bool AllowCompress { get; set; } = true;
/// <summary>
/// 是否允许图像放大 (提升渲染质量)
/// </summary>
public bool AllowExpand { get; set; } = false;
/// <summary>
/// 目标分辨率 (格式如 1920x1080，空则保持原图)
/// </summary>
[RegularExpression(@"^\d+x\d+$", ErrorMessage = "分辨率格式必须为 宽度x高度 (如 1920x1080)")]
public string TargetResolution { get; set; } = string.Empty;
}

2
SHH.CameraSdk/Controllers/Dto/SubscriptionDto.cs
View File

@@ -9,7 +9,7 @@ namespace SHH.CameraSdk;
public class SubscriptionDto
{
/// <summary>
/// 进程唯一标识 (如 "AI_Process_01"、"Main_Display_02")
/// 订阅标识 (如 "AI_Process_01"、"Main_Display_02")
/// </summary>
[Required(ErrorMessage = "进程标识 AppId 不能为空")]
[MaxLength(50, ErrorMessage = "AppId 长度不能超过 50 个字符")]

11
SHH.CameraSdk/Core/Manager/CameraManager.cs
View File

@@ -309,7 +309,7 @@ public class CameraManager : IDisposable, IAsyncDisposable
{
var options = new DynamicStreamOptions
{
StreamType = dto.StreamType,
StreamType = dto.StreamType ?? newConfig.StreamType,
RenderHandle = (IntPtr)dto.RenderHandle
};
device.ApplyOptions(options);
@@ -428,4 +428,13 @@ public class CameraManager : IDisposable, IAsyncDisposable
}
#endregion
/// <summary>
/// [新增] 获取当前管理的所有相机设备（兼容网络引擎接口）
/// </summary>
public IEnumerable<BaseVideoSource> GetAllCameras()
{
// 复用现有的 GetAllDevices 逻辑
return GetAllDevices();
}
}

6
SHH.CameraSdk/Core/Memory/SmartFrame.cs
View File

@@ -28,6 +28,12 @@ public class SmartFrame : IDisposable
/// <remarks> 内存由帧池预分配，全程复用，不触发 GC </remarks>
public Mat InternalMat { get; private set; }
/// <summary> [快捷属性] 原始图像宽度 (若 TargetMat 为空则返回 0) </summary>
public int InternalWidth => InternalMat?.Width ?? 0;
/// <summary> [快捷属性] 原始图像高度 (若 TargetMat 为空则返回 0) </summary>
public int InnernalHeight => InternalMat?.Height ?? 0;
/// <summary> 帧激活时间戳（记录帧被取出池的时刻） </summary>
public DateTime Timestamp { get; private set; }

134
SHH.CameraSdk/Core/Pipeline/GlobalStreamDispatcher.cs
View File

@@ -36,11 +36,6 @@ public static class GlobalStreamDispatcher
// =================================================================
public static event Action<long, SmartFrame> OnGlobalFrame;
// =================================================================
// 2. 原有：定向分发逻辑 (保留不动，给图像处理集群用)
// =================================================================
// private static ConcurrentDictionary<string, ...> _subscribers ...
/// <summary>
/// 统一入口：驱动层调用此方法分发图像
/// </summary>
@@ -71,6 +66,10 @@ public static class GlobalStreamDispatcher
/// </summary>
private static readonly ConcurrentDictionary<string, Action<long, SmartFrame>> _routingTable = new();
// [新增] 旁路订阅支持
// 用于 NetworkService 这种需要针对单个设备进行订阅/取消订阅的场景
private static readonly ConcurrentDictionary<string, ConcurrentDictionary<long, Action<SmartFrame>>> _deviceSpecificTable = new();
#endregion
#region --- 3. 订阅管理接口 (Subscription Management API) ---
@@ -98,27 +97,63 @@ public static class GlobalStreamDispatcher
);
}
///// <summary>
///// [新增] 精准订阅：仅监听指定设备的特定 AppId 帧
///// 优势：内部自动过滤 DeviceId，回调函数无需再写 if 判断
///// </summary>
///// <param name="appId">需求标识</param>
///// <param name="specificDeviceId">只接收此设备的帧</param>
///// <param name="handler">处理回调（注意：此处签名不含 deviceId，因为已隐式确定）</param>
//public static void Subscribe(string appId, long specificDeviceId, Action<SmartFrame> handler)
//{
// // 创建一个“过滤器”闭包
// Action<long, SmartFrame> wrapper = (id, frame) =>
// {
// // 只有当来源 ID 与订阅 ID 一致时，才触发用户的业务回调
// if (id == specificDeviceId)
// {
// handler(frame);
// }
// };
// // 将过滤器注册到基础路由表中
// Subscribe(appId, wrapper);
//}
/// <summary>
/// [新增] 精准订阅：仅监听指定设备的特定 AppId 帧
/// 优势：内部自动过滤 DeviceId，回调函数无需再写 if 判断
/// [重写] 精准订阅：仅监听指定设备的特定 AppId 帧
/// 修改说明：不再使用闭包 + 多播委托，而是存入二级字典，以便能精准取消
/// </summary>
/// <param name="appId">需求标识</param>
/// <param name="specificDeviceId">只接收此设备的帧</param>
/// <param name="handler">处理回调（注意：此处签名不含 deviceId，因为已隐式确定）</param>
public static void Subscribe(string appId, long specificDeviceId, Action<SmartFrame> handler)
{
// 创建一个“过滤器”闭包
Action<long, SmartFrame> wrapper = (id, frame) =>
{
// 只有当来源 ID 与订阅 ID 一致时，才触发用户的业务回调
if (id == specificDeviceId)
{
handler(frame);
}
};
if (string.IsNullOrWhiteSpace(appId) || handler == null) return;
// 将过滤器注册到基础路由表中
Subscribe(appId, wrapper);
// 1. 获取或创建该 AppId 的设备映射表
var deviceMap = _deviceSpecificTable.GetOrAdd(appId, _ => new ConcurrentDictionary<long, Action<SmartFrame>>());
// 2. 添加或更新该设备的订阅
// 注意：这里使用多播委托 (+)，支持同一个 App 同一个 Device 有多个处理逻辑（虽然很少见）
deviceMap.AddOrUpdate(specificDeviceId, handler, (_, existing) => existing + handler);
}
/// <summary>
/// [新增] 精准取消订阅：移除指定 AppId 下指定设备的订阅
/// NetworkService 必须调用此方法来防止内存泄漏
/// </summary>
public static void Unsubscribe(string appId, long specificDeviceId)
{
if (string.IsNullOrWhiteSpace(appId)) return;
// 1. 查找该 AppId 是否有记录
if (_deviceSpecificTable.TryGetValue(appId, out var deviceMap))
{
// 2. 移除该设备的订阅委托
if (deviceMap.TryRemove(specificDeviceId, out _))
{
// 可选：如果该 AppId 下没设备了，是否清理外层字典？(为了性能通常不清理，或者定期清理)
// Console.WriteLine($"[Dispatcher] {appId} 已停止订阅设备 {specificDeviceId}");
}
}
}
/// <summary>
@@ -192,6 +227,26 @@ public static class GlobalStreamDispatcher
}
}
// =========================================================
// B. [新增逻辑] 匹配设备级 AppId 订阅 (如 NetworkService)
// =========================================================
if (_deviceSpecificTable.TryGetValue(appId, out var deviceMap))
{
// 查找当前设备是否有订阅者
if (deviceMap.TryGetValue(deviceId, out var deviceHandler))
{
try
{
deviceHandler.Invoke(frame);
task.Context.AddLog($"帧任务 设备级 [Seq:{sequence}] 投递到 AppId:{appId}");
}
catch (Exception ex)
{
Console.WriteLine($"[DispatchError] DeviceSpecific AppId={appId}, Dev={deviceId}: {ex.Message}");
}
}
}
// 2. 匹配预设的全局通道（兼容旧版订阅逻辑）
switch (appId.ToUpperInvariant())
{
@@ -204,6 +259,43 @@ public static class GlobalStreamDispatcher
}
}
// =========================================================================
// 2. [旁路通道] 扫描设备级订阅表 (NetworkService, 录像服务 等)
// 这是外部服务“被动”监听的目标，不在 targetAppIds 白名单里也要发
// =========================================================================
if (!_deviceSpecificTable.IsEmpty)
{
// 遍历所有注册了旁路监听的 AppId (例如 "NetService")
foreach (var kvp in _deviceSpecificTable)
{
string sidecarAppId = kvp.Key;
var deviceMap = kvp.Value;
// 优化：如果这个 AppId 已经在上面的 targetAppIds 里处理过了，就跳过，防止重复发送
// (例如：如果设备未来真的把 NetService 加入了白名单，这里就不重复发了)
if (targetAppIds.Contains(sidecarAppId)) continue;
// 检查这个 AppId 下，是否有人订阅了当前这台设备
if (deviceMap.TryGetValue(deviceId, out var handler))
{
try
{
handler.Invoke(frame);
// task.Context.AddLog($"帧任务 [Seq:{sequence}] 旁路投递到: {sidecarAppId}");
}
catch (Exception ex)
{
Console.WriteLine($"[SidecarDispatchError] App={sidecarAppId}, Dev={deviceId}: {ex.Message}");
}
}
}
}
// =========================================================================
// 3. [上帝通道] 全局广播
// =========================================================================
OnGlobalFrame?.Invoke(deviceId, frame);
// 分发完成后记录遥测数据
GlobalTelemetry.RecordLog(sequence, task.Context);
}

12
SHH.CameraSdk/Core/Services/DisplayWindowManager.cs
View File

@@ -1,9 +1,4 @@
using OpenCvSharp;
using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
namespace SHH.CameraSdk
{
@@ -166,6 +161,13 @@ namespace SHH.CameraSdk
return;
}
// 1. 先检查队列容量 (虽然 BlockingCollection 没有完美的无锁 IsFull，但可以通过 Count 判断)
// 这是一个不需要 100% 精确的优化，只要能拦截掉大部分无用功即可
if (_uiActionQueue.Count >= 30)
{
return; // 直接丢弃，不进行克隆，节省 CPU
}
Mat frameClone = null;
try
{

1
SHH.CameraSdk/Core/Services/FileStorageService.cs
View File

@@ -77,6 +77,7 @@ public class FileStorageService : IStorageService
var list = JsonSerializer.Deserialize<List<VideoSourceConfig>>(json, _jsonOptions);
return list ?? new List<VideoSourceConfig>();
//return new List<VideoSourceConfig>();
}
catch (Exception ex)
{

5
SHH.CameraSdk/Drivers/HikVision/HikVideoSource.cs
View File

@@ -1,5 +1,7 @@
using OpenCvSharp;
using SHH.CameraSdk.HikFeatures;
using System.Runtime.ExceptionServices;
using System.Security;
namespace SHH.CameraSdk;
@@ -325,6 +327,9 @@ public class HikVideoSource : BaseVideoSource,
#region --- 解码与帧分发 (Decoding) ---
// 必须同时加上 SecurityCritical
[HandleProcessCorruptedStateExceptions]
[SecurityCritical]
private void SafeOnDecodingCallBack(int nPort, IntPtr pBuf, int nSize, ref HikPlayMethods.FRAME_INFO pFrameInfo, int nReserved1, int nReserved2)
{
// [优化] 维持心跳，防止被哨兵误杀

4
SHH.CameraSdk/SHH.CameraSdk.csproj
View File

@@ -21,6 +21,10 @@
<PackageReference Include="OpenCvSharp4.runtime.win" Version="4.11.0.20250507" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\SHH.Contracts\SHH.Contracts.csproj" />
</ItemGroup>
<ItemGroup>
<Using Include="System.Collections.Concurrent" />
<Using Include="System.Collections.ObjectModel" />

113
SHH.CameraService/CameraEngineWorker.cs
View File

@@ -1,113 +0,0 @@
using Microsoft.Extensions.Hosting;
using SHH.CameraSdk;
namespace SHH.CameraService;
/// <summary>
/// 相机服务核心引擎工作者（后台长驻服务）
/// <para>核心职责：</para>
/// <para>1. 管理 CameraManager 全生命周期（启动、配置、释放）</para>
/// <para>2. 初始化网络哨兵（ConnectivitySentinel），监控设备网络连通性</para>
/// <para>3. 无配置时自动添加默认测试设备，降低调试门槛</para>
/// <para>设计说明：</para>
/// <para>- 基于 BackgroundService，运行在独立后台线程，不阻塞 Web 主线程</para>
/// <para>- 与 CameraManager 强绑定，是所有相机设备的统一入口</para>
/// <para>- 包含容错机制，添加设备失败不影响整体服务启动</para>
public class CameraEngineWorker : BackgroundService
{
#region --- 依赖注入字段 ---
/// <summary>
/// 相机管理器实例（核心业务对象）
/// 功能：管理所有相机设备的生命周期、状态监控、配置更新
/// </summary>
private readonly CameraManager _manager;
/// <summary>
/// 网络连通性哨兵实例
/// 功能：周期性 Ping 设备、检测网络状态、触发断线重连
/// </summary>
private readonly ConnectivitySentinel _sentinel;
#endregion
#region --- 构造函数 ---
/// <summary>
/// 初始化相机引擎工作者实例
/// </summary>
/// <param name="manager">相机管理器（通过 DI 注入，已关联存储服务）</param>
/// <param name="sentinel">网络哨兵（通过 DI 注入，已预设监控周期）</param>
public CameraEngineWorker(CameraManager manager, ConnectivitySentinel sentinel)
{
_manager = manager ?? throw new ArgumentNullException(
nameof(manager), "相机管理器实例不能为空，核心引擎启动失败");
_sentinel = sentinel ?? throw new ArgumentNullException(
nameof(sentinel), "网络哨兵实例不能为空，设备监控功能失效");
}
#endregion
#region --- BackgroundService 核心实现 ---
/// <summary>
/// 启动引擎：初始化相机管理器并加载业务配置
/// <para>执行流程：</para>
/// <para>1. 启动 CameraManager（加载本地配置文件中的设备信息）</para>
/// <para>2. 加载默认业务逻辑（无设备时添加测试设备）</para>
/// <para>注意：网络哨兵的启动逻辑已内置在其构造函数中，此处无需额外调用</para>
/// </summary>
/// <param name="stoppingToken">服务停止令牌（响应应用关闭/重启信号）</param>
/// <returns>异步任务（引擎启动完成后结束）</returns>
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
Console.WriteLine("[Engine] 核心引擎启动中...");
Console.WriteLine("[Engine] 启动相机管理器（加载设备配置）");
try
{
// 启动相机管理器：加载 App_Data/Process_X 目录下的设备配置文件
await _manager.StartAsync();
Console.WriteLine("[Engine] 相机管理器启动成功，已加载配置文件中的设备");
}
catch (Exception ex)
{
Console.WriteLine($"[Engine] 相机管理器启动失败：{ex.Message}");
Console.WriteLine("[Engine] 警告：核心引擎将继续运行，但无法管理任何相机设备");
return;
}
Console.WriteLine("[Engine] 核心引擎启动完成，进入运行状态");
Console.WriteLine("[Engine] 提示：可通过 API 接口添加/编辑/删除设备，实时生效");
}
/// <summary>
/// 停止引擎：优雅释放资源
/// <para>执行流程：</para>
/// <para>1. 调用 CameraManager.DisposeAsync()，释放所有设备连接、句柄、线程资源</para>
/// <para>2. 调用基类 StopAsync()，标记服务停止状态</para>
/// <para>注意：必须先释放 CameraManager，避免设备连接泄露</para>
/// </summary>
/// <param name="cancellationToken">取消令牌（用于强制终止释放流程）</param>
/// <returns>异步任务（资源释放完成后结束）</returns>
public override async Task StopAsync(CancellationToken cancellationToken)
{
Console.WriteLine("[Engine] 核心引擎正在停止...");
try
{
// 释放相机管理器：停止所有设备取流、注销登录、释放非托管资源
await _manager.DisposeAsync();
Console.WriteLine("[Engine] 相机管理器资源已释放");
}
catch (Exception ex)
{
Console.WriteLine($"[Engine] 资源释放异常：{ex.Message}");
}
// 调用基类方法，完成服务停止
await base.StopAsync(cancellationToken);
Console.WriteLine("[Engine] 核心引擎已停止");
}
#endregion
}

123
SHH.CameraService/CommandBusProcessor.cs
View File

@@ -1,123 +0,0 @@
using Newtonsoft.Json;
using SHH.CameraSdk;
using SHH.Contracts;
namespace SHH.CameraService
{
/// <summary>
/// 指令业务逻辑分发器 (纯逻辑层)
/// <para>职责：解析业务参数 -> 调用 CameraManager -> 返回执行结果</para>
/// <para>注意：本类不处理网络协议，也不负责 RequestId 的回填，只关注业务本身</para>
/// </summary>
public static class CommandBusProcessor
{
/// <summary>
/// 核心业务入口
/// </summary>
/// <param name="manager">相机管理器实例</param>
/// <param name="payload">已解析的指令包</param>
/// <returns>执行结果 (不含 RequestId，由调用方补充)</returns>
public static CommandResult ProcessBusinessLogic(CameraManager manager, CommandPayload payload)
{
string cmd = payload.CmdCode.ToUpper();
// 忽略客户端发回的 ACK (如果是双向确认模式)
if (cmd == "REGISTER_ACK") return CommandResult.Ok();
// 解析 TargetId (CameraId)
long deviceId = 0;
// 只有非 SYSTEM 指令才需要解析设备ID
if (payload.TargetId != "SYSTEM" && !long.TryParse(payload.TargetId, out deviceId))
{
return CommandResult.Fail($"Invalid Device ID: {payload.TargetId}");
}
try
{
switch (cmd)
{
// ==========================================
// 1. PTZ 云台控制
// ==========================================
case "PTZ":
{
var device = manager.GetDevice(deviceId);
if (device == null) return CommandResult.Fail("Device Not Found");
if (!device.IsOnline) return CommandResult.Fail("Device Offline");
// 检查设备是否支持 PTZ 能力 (接口模式匹配)
if (device is IPtzFeature ptzFeature)
{
var ptzDto = JsonConvert.DeserializeObject<PtzControlDto>(payload.JsonParams);
if (ptzDto == null) return CommandResult.Fail("Invalid PTZ Params");
// 异步转同步执行 (Task.Wait 在后台线程是安全的)
if (ptzDto.Duration > 0)
{
// 点动模式 (例如：向左转 500ms)
ptzFeature.PtzStepAsync(ptzDto.Action, ptzDto.Duration, ptzDto.Speed).Wait();
}
else
{
// 持续模式 (开始转/停止转)
ptzFeature.PtzControlAsync(ptzDto.Action, ptzDto.Stop, ptzDto.Speed).Wait();
}
return CommandResult.Ok("PTZ Executed");
}
return CommandResult.Fail("Device does not support PTZ");
}
// ==========================================
// 2. 远程重启
// ==========================================
case "REBOOT":
{
var device = manager.GetDevice(deviceId);
if (device == null) return CommandResult.Fail("Device Not Found");
if (device is IRebootFeature rebootFeature)
{
rebootFeature.RebootAsync().Wait();
return CommandResult.Ok("Reboot command sent");
}
return CommandResult.Fail("Device does not support Reboot");
}
// ==========================================
// 3. 时间同步
// ==========================================
case "SYNC_TIME":
{
var device = manager.GetDevice(deviceId);
if (device == null) return CommandResult.Fail("Device Not Found");
if (device is ITimeSyncFeature timeFeature)
{
timeFeature.SetTimeAsync(DateTime.Now).Wait();
return CommandResult.Ok("Time synced");
}
return CommandResult.Fail("Device does not support TimeSync");
}
// ==========================================
// 4. 系统级指令 (心跳/诊断)
// ==========================================
case "PING":
return CommandResult.Ok("PONG");
default:
return CommandResult.Fail($"Unknown Command: {cmd}");
}
}
catch (AggregateException ae)
{
// 捕获异步任务内部的异常
return CommandResult.Fail($"Execution Error: {ae.InnerException?.Message}");
}
catch (Exception ex)
{
return CommandResult.Fail($"Execution Error: {ex.Message}");
}
}
}
}

346
SHH.CameraService/CommandBusService.cs
View File

@@ -1,346 +0,0 @@
using Microsoft.Extensions.Caching.Memory;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Configuration;
using NetMQ;
using NetMQ.Sockets;
using Newtonsoft.Json;
using SHH.CameraSdk;
using SHH.Contracts;
namespace SHH.CameraService
{
/// <summary>
/// 双模指令总线服务 (Enterprise V2)
/// <para>核心职责：建立 TCP 指令通道，接收客户端指令并分发给 CameraManager</para>
/// <para>增强特性：</para>
/// <para>1. 支持双模：被动监听 (Bind) 与 主动投递 (Connect)</para>
/// <para>2. 幂等性控制：利用 MemoryCache 防止客户端重试导致的重复执行</para>
/// <para>3. 顺序一致性：利用时间戳防止指令乱序</para>
/// </summary>
public class CommandBusService : BackgroundService
{
#region --- 1. 字段与依赖 ---
private readonly CameraManager _cameraManager;
private readonly IConfiguration _config;
private readonly IMemoryCache _cache; // 核心：用于请求去重
private readonly int _processId;
// 运行状态标志
private volatile bool _isRunning = false;
// 两种模式的 Socket (互斥存在)
private ResponseSocket? _repSocket; // 模式A: 被动监听 (Server-Listening)
private DealerSocket? _dealerSocket; // 模式B: 主动投递 (Server-Dialing)
// 顺序一致性锁：记录每个设备最后处理的指令时间戳
// Key: TargetId (设备ID), Value: Timestamp (最后执行时间)
private readonly Dictionary<string, DateTime> _deviceLastCmdTime = new();
#endregion
#region --- 2. 构造函数 ---
/// <summary>
/// 构造函数 (注意：必须在 Program.cs 注册 AddMemoryCache)
/// </summary>
public CommandBusService(CameraManager manager, IConfiguration config, IMemoryCache cache)
{
_cameraManager = manager;
_config = config;
_cache = cache;
// 获取当前进程 ID (默认为 1)
_processId = _config.GetValue<int>("ProcessId", 1);
}
#endregion
#region --- 3. 核心生命周期 ---
protected override Task ExecuteAsync(CancellationToken stoppingToken)
{
// 在后台线程启动，避免阻塞 Web 主线程
return Task.Run(() =>
{
_isRunning = true;
// 1. 读取网络策略
// 优先读取配置中的主动目标，如果没有则回退到被动监听
string? activeTargetIp = _config["Network:ActiveTargets:0:Ip"];
bool isActiveMode = !string.IsNullOrEmpty(activeTargetIp);
try
{
if (isActiveMode)
{
// === 模式 B: 主动投递 (Server Connects Client) ===
// 场景：服务器在内网，主动连接公网/固定IP的客户端
int cmdPort = _config.GetValue<int>("Network:ActiveTargets:0:CmdPort", 7000);
string addr = $"tcp://{activeTargetIp}:{cmdPort}";
RunActiveMode(addr, stoppingToken);
}
else
{
// === 模式 A: 被动监听 (Server Binds Port) ===
// 场景：服务器有固定IP，等待客户端连接
int basePort = _config.GetValue<int>("Network:Passive:CmdPortBase", 7000);
int listenPort = basePort + (_processId - 1);
string addr = $"tcp://*:{listenPort}";
RunPassiveMode(addr, stoppingToken);
}
}
catch (Exception ex)
{
Console.WriteLine($"[CmdBus] 致命错误停止: {ex.Message}");
}
finally
{
_isRunning = false;
CleanupSockets();
}
}, stoppingToken);
}
private void CleanupSockets()
{
try
{
_repSocket?.Dispose();
_dealerSocket?.Dispose();
}
catch { /* 忽略销毁时的异常 */ }
}
#endregion
#region --- 4. 模式实现：被动监听 (Passive) ---
private void RunPassiveMode(string address, CancellationToken token)
{
using (_repSocket = new ResponseSocket())
{
_repSocket.Bind(address);
Console.WriteLine($"[CmdBus] [被动模式] 指令监听已启动: {address}");
while (!token.IsCancellationRequested)
{
try
{
// 1. 阻塞等待请求 (超时1秒以便响应 Cancel 信号)
if (!_repSocket.TryReceiveFrameString(TimeSpan.FromSeconds(1), out string reqJson))
continue;
// 2. 处理业务 (带 去重 + ID回填 逻辑)
CommandResult result = this.ProcessRequest(reqJson);
// 3. 发送回执
// 注意：REP 模式必须发送应答，即使 result 为 null (Fire-and-Forget) 也建议发一个空 ACK 防止 Socket 状态错乱
// 但为了协议统一，建议 Passive 模式下总是返回结果
string respJson = result != null ? JsonConvert.SerializeObject(result) : "{}";
_repSocket.SendFrame(respJson);
}
catch (Exception ex)
{
Console.WriteLine($"[CmdBus-Passive] 异常: {ex.Message}");
}
}
}
}
#endregion
#region --- 5. 模式实现：主动投递 (Active) ---
private void RunActiveMode(string address, CancellationToken token)
{
// 外层循环：断线重连机制
while (!token.IsCancellationRequested)
{
try
{
using (_dealerSocket = new DealerSocket())
{
Console.WriteLine($"[CmdBus] [主动模式] 正在连接指令中心: {address}");
_dealerSocket.Connect(address);
// ★★★ 关键步骤：连接成功后，立即发送【身份注册包】 ★★★
// 客户端收到这个包后，才能在界面上显示"设备在线"
SendRegistration(_dealerSocket);
// 内层循环：消息收发
while (!token.IsCancellationRequested)
{
// 1. 接收指令
// DealerSocket 是异步全双工的，这里即使没收到消息也不会阻塞发送
if (!_dealerSocket.TryReceiveFrameString(TimeSpan.FromSeconds(1), out string reqJson))
{
// 空闲周期，可在此处添加心跳发送逻辑 (Ping)
continue;
}
// 2. 处理业务 (带 去重 + ID回填 逻辑)
CommandResult result = this.ProcessRequest(reqJson);
// 3. 发送结果 (QoS控制)
// 如果结果为 null，说明指令是 Fire-and-Forget (无需回执)，则不发送网络包节省带宽
if (result != null)
{
_dealerSocket.SendFrame(JsonConvert.SerializeObject(result));
}
}
}
}
catch (Exception ex)
{
Console.WriteLine($"[CmdBus-Active] 连接中断或异常: {ex.Message}");
// 避免死循环狂刷 CPU，等待 3 秒再重连
Thread.Sleep(3000);
}
}
}
/// <summary>
/// 发送身份注册包 (Active 模式专用)
/// </summary>
private void SendRegistration(DealerSocket socket)
{
try
{
// 计算实际端口信息
int portOffset = _processId - 1;
var regInfo = new ServerRegistrationDto
{
ProcessId = _processId,
InstanceId = $"Gateway_{_processId}",
ServerIp = GetLocalIpAddress(),
WebApiPort = 5000 + portOffset,
VideoPort = 5555 + portOffset,
CmdPort = 7000 + portOffset,
StartTime = DateTime.Now,
Description = "Active Mode Connection (V2)"
};
// 封装信封 (系统级指令)
var payload = new CommandPayload
{
CmdCode = "SERVER_REGISTER",
TargetId = "SYSTEM",
JsonParams = JsonConvert.SerializeObject(regInfo),
RequestId = Guid.NewGuid().ToString("N"),
RequireAck = false // 注册包通常不需要回执，只要连上就行
};
socket.SendFrame(JsonConvert.SerializeObject(payload));
Console.WriteLine($"[CmdBus] 身份注册包已发送 -> {regInfo.ServerIp}:{regInfo.WebApiPort}");
}
catch (Exception ex)
{
Console.WriteLine($"[CmdBus] 注册包发送失败: {ex.Message}");
}
}
private string GetLocalIpAddress()
{
try
{
var host = System.Net.Dns.GetHostEntry(System.Net.Dns.GetHostName());
foreach (var ip in host.AddressList)
{
if (ip.AddressFamily == System.Net.Sockets.AddressFamily.InterNetwork)
return ip.ToString();
}
}
catch { }
return "127.0.0.1";
}
#endregion
#region --- 6. 协议处理核心 (★★ V2 核心增强 ★★) ---
/// <summary>
/// 统一处理请求协议：去重 -> 排序 -> 执行 -> 回填 ID
/// </summary>
private CommandResult ProcessRequest(string json)
{
if (string.IsNullOrWhiteSpace(json)) return CommandResult.Fail("Empty Request");
CommandPayload? payload;
try { payload = JsonConvert.DeserializeObject<CommandPayload>(json); }
catch { return CommandResult.Fail("Invalid JSON Protocol"); }
if (payload == null) return CommandResult.Fail("Null Payload");
// =========================================================
// A. 【幂等性检查】(Idempotency Check)
// =========================================================
// 查缓存：如果这个 RequestId 10秒内处理过，直接返回上次的结果
// 这样即使客户端重试发了 10 次，业务逻辑也只跑 1 次
if (_cache.TryGetValue(payload.RequestId, out CommandResult cachedResult))
{
Console.WriteLine($"[Dedup] 拦截重复请求: {payload.RequestId} (Retry: {payload.RetryCount})");
return cachedResult;
}
// =========================================================
// B. 【顺序一致性检查】(Order Guarantee)
// =========================================================
// 防止乱序：比如先发的“停止”因为网络卡顿，比后发的“开始”晚到
if (payload.TargetId != "SYSTEM")
{
lock (_deviceLastCmdTime)
{
if (_deviceLastCmdTime.TryGetValue(payload.TargetId, out DateTime lastTime))
{
if (payload.Timestamp < lastTime)
{
Console.WriteLine($"[Order] 丢弃乱序指令: {payload.CmdCode}");
return CommandResult.Fail("Order Violation: Stale Command Dropped");
}
}
_deviceLastCmdTime[payload.TargetId] = payload.Timestamp;
}
}
// =========================================================
// C. 【业务执行】
// =========================================================
CommandResult result;
try
{
// 调用纯逻辑层
result = CommandBusProcessor.ProcessBusinessLogic(_cameraManager, payload);
}
catch (Exception ex)
{
result = CommandResult.Fail($"Internal Logic Error: {ex.Message}");
}
// =========================================================
// D. 【闭环回填】
// =========================================================
// 必须把身份证号贴回去，不然客户端不知道这是谁的回执
result.RequestId = payload.RequestId;
// =========================================================
// E. 【存入缓存】
// =========================================================
// 缓存 10 秒，覆盖客户端的重试窗口
_cache.Set(payload.RequestId, result, TimeSpan.FromSeconds(10));
// =========================================================
// F. 【QoS 过滤】
// =========================================================
// 如果客户端说不需要回信，返回 null
if (!payload.RequireAck)
{
return null;
}
return result;
}
#endregion
}
}

97
SHH.CameraService/CommandClientWorker.cs
View File

@@ -1,97 +0,0 @@
using Microsoft.Extensions.Hosting;
using NetMQ;
using NetMQ.Sockets;
using Newtonsoft.Json;
using SHH.CameraSdk;
using System.Text;
namespace SHH.CameraService;
public class CommandClientWorker : BackgroundService
{
private readonly ServiceConfig _config;
public CommandClientWorker(ServiceConfig config)
{
_config = config;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
// 1. 如果不是主动/混合模式，不需要连接
if (!_config.ShouldConnect) return;
// ★★★ 核心修正：直接读取解析好的指令地址列表 ★★★
// 这些地址来自参数 --uris "IP,VideoPort&CommandPort" 中的 CommandPort 部分
var cmdUris = _config.CommandEndpoints;
if (cmdUris.Count == 0)
{
Console.WriteLine("[指令] 未在参数中找到指令通道地址(位于&符号右侧)，跳过连接。");
return;
}
// 2. 初始化 Dealer Socket
using var dealer = new DealerSocket();
// 设置身份 (Identity)，让 Dashboard 知道我是 "CameraApp_01"
string myIdentity = _config.AppId;
dealer.Options.Identity = Encoding.UTF8.GetBytes(myIdentity);
// 3. 连接所有目标 (支持多点控制)
foreach (var uri in cmdUris)
{
Console.WriteLine($"[指令] 连接控制端: {uri}");
dealer.Connect(uri);
}
// 4. 发送登录包 (握手)
// 构造心跳包
var heartbeat = new
{
Type = "Login",
Id = myIdentity,
Time = DateTime.Now
};
string loginJson = JsonConvert.SerializeObject(heartbeat);
// 注意：Dealer Socket 发送是负载均衡的 (Round-Robin)。
// 如果连接了多个 Dashboard，SendFrame 一次只会发给其中一个。
// 为了确保所有 Dashboard 都能收到上线通知，我们根据连接数循环发送几次。
// (注：这只是初始化时的权宜之计，心跳包后续可以定时发送)
for (int i = 0; i < cmdUris.Count; i++)
{
dealer.SendFrame(loginJson);
await Task.Delay(10); // 稍微间隔，给 ZMQ 内部调度一点时间
}
Console.WriteLine($"[指令] 已发送登录包 (ID: {myIdentity})，进入监听循环...");
// 5. 监听循环
while (!stoppingToken.IsCancellationRequested)
{
try
{
// 非阻塞接收 (500ms 超时)，避免卡死线程
if (dealer.TryReceiveFrameString(TimeSpan.FromMilliseconds(500), out string msg))
{
Console.WriteLine($"[指令] 收到: {msg}");
// TODO: 在这里解析 JSON 并调用 CameraSDK 执行业务
// var cmd = JsonConvert.DeserializeObject<CommandModel>(msg);
// if (cmd.Action == "Reboot") ...
// 回复 ACK (确认收到)
// Dealer 会自动根据 Router 发来的 RoutingID 路由回去
dealer.SendFrame($"ACK: {msg} (From {myIdentity})");
}
}
catch (Exception ex)
{
Console.WriteLine($"[指令] 通信异常: {ex.Message}");
// 防止异常死循环刷屏
await Task.Delay(1000, stoppingToken);
}
}
}
}

139
SHH.CameraService/Core/CmdClients/CommandClientWorker.cs Normal file
View File

@@ -0,0 +1,139 @@
using Microsoft.Extensions.Hosting;
using NetMQ;
using NetMQ.Sockets;
using Newtonsoft.Json;
using SHH.CameraSdk;
using System.Text;
namespace SHH.CameraService;
public class CommandClientWorker : BackgroundService
{
private readonly ServiceConfig _config;
private readonly CommandDispatcher _dispatcher; // 注入分发器
public CommandClientWorker(ServiceConfig config, CommandDispatcher dispatcher)
{
_config = config;
_dispatcher = dispatcher;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
// =================================================================
// ★★★ 核心修复：强制让出主线程 ★★★
// 这行代码会让当前的 ExecuteAsync 立即返回一个未完成的 Task 给 Host，
// Host 就会认为 "这个服务启动好了"，然后继续去启动 WebAPI。
// 而剩下的代码会被调度到线程池里异步执行，互不干扰。
// =================================================================
await Task.Yield();
// 1. 如果不是主动/混合模式，不需要连接
if (!_config.ShouldConnect) return;
var cmdEndpoints = _config.CommandEndpoints;
if (cmdEndpoints.Count == 0)
{
Console.WriteLine("[指令] 未配置指令通道，跳过注册。");
return;
}
// 2. 初始化 Dealer Socket
using var dealer = new DealerSocket();
// ★★★ 关键：设置身份标识 (Identity) ★★★
// 服务端 (Router) 收到消息时，第一帧就是这个 ID
// 如果不设，ZMQ 会随机生成一个二进制 ID，服务端就不知道你是谁了
string myIdentity = _config.AppId;
dealer.Options.Identity = Encoding.UTF8.GetBytes(myIdentity);
// 3. 连接所有目标 (遍历 ServiceEndpoint 对象)
foreach (var ep in cmdEndpoints)
{
Console.WriteLine($"[指令] 连接控制端: {ep.Uri} [{ep.Description}]");
try
{
dealer.Connect(ep.Uri);
}
catch (Exception ex)
{
Console.WriteLine($"[指令] 连接失败 {ep.Uri}: {ex.Message}");
}
}
// 1. 获取本机 IP (简单的获取方式，用于上报给 Dashboard)
string localIp = "127.0.0.1";
try
{
// 简单获取首个非回环 IP，生产环境建议用更严谨的帮助类
var host = System.Net.Dns.GetHostEntry(System.Net.Dns.GetHostName());
localIp = host.AddressList.FirstOrDefault(ip =>
ip.AddressFamily == System.Net.Sockets.AddressFamily.InterNetwork)?.ToString() ?? "127.0.0.1";
}
catch { }
// 4. 构建注册/登录包
var registerPayload = new
{
Action = "Register",
Payload = new
{
// 1. AppId (身份)
Id = _config.AppId,
// 2. Version (程序集版本)
Version = System.Reflection.Assembly.GetEntryAssembly()?.GetName().Version?.ToString() ?? "1.0.0",
// 3. 进程 ID (用于远程监控)
Pid = Environment.ProcessId,
// 4. 关键端口信息
// 告诉 Dashboard：如果你想调我的 REST API，请访问这个端口
WebPort = _config.BasePort,
// 如果您有本地绑定的 ZMQ 端口也可以在这里上报
// VideoPort = _config.BasePort + 1,
// 基础网络信息
Ip = localIp,
// 附带信息：我是要把视频推给谁 (供 Dashboard 调试用)
TargetVideoNodes = _config.VideoEndpoints.Select(e => e.Uri).ToList()
},
Time = DateTime.Now
};
string json = JsonConvert.SerializeObject(registerPayload);
// 5. 发送注册包
// Dealer 连接建立是异步的，所以这里直接发，ZMQ 会在底层连接成功后自动把消息推出去
// 为了保险，对于多个 Endpoint，Dealer 默认是负载均衡发送的（轮询）。
// 如果想让每个 Endpoint 都收到注册包，这在 Dealer 模式下稍微有点特殊。
// 但通常我们只需要发一次，只要有一个 Dashboard 收到并建立会话即可。
// 或者简单粗暴：循环发送几次，确保覆盖。
Console.WriteLine($"[指令] 发送注册包: {json}");
dealer.SendFrame(json);
// 6. 进入监听循环 (等待 ACK 或 指令)
// 进入监听循环
while (!stoppingToken.IsCancellationRequested)
{
try
{
if (dealer.TryReceiveFrameString(TimeSpan.FromMilliseconds(500), out string msg))
{
Console.WriteLine($"[指令] 收到消息: {msg}");
// ★★★ 核心变化：直接扔给分发器 ★★★
// 无论未来加多少指令，这里都不用改代码
await _dispatcher.DispatchAsync(msg);
}
}
catch (Exception ex)
{
Console.WriteLine($"[指令] 异常: {ex.Message}");
}
}
}
}

46
SHH.CameraService/Core/CmdClients/CommandDispatcher.cs Normal file
View File

@@ -0,0 +1,46 @@
using Newtonsoft.Json.Linq;
namespace SHH.CameraService;
public class CommandDispatcher
{
// 路由表：Key = ActionName, Value = Handler
private readonly Dictionary<string, ICommandHandler> _handlers;
// 通过依赖注入拿到所有实现了 ICommandHandler 的类
public CommandDispatcher(IEnumerable<ICommandHandler> handlers)
{
_handlers = handlers.ToDictionary(h => h.ActionName, h => h);
}
public async Task DispatchAsync(string jsonMessage)
{
try
{
var jObj = JObject.Parse(jsonMessage);
string action = jObj["Action"]?.ToString();
var payload = jObj["Payload"];
if (string.IsNullOrEmpty(action)) return;
// 1. 查找是否有对应的处理器
if (_handlers.TryGetValue(action, out var handler))
{
await handler.ExecuteAsync(payload);
}
else if (action == "ACK")
{
// ACK 是特殊的，可以直接在这里处理或者忽略
Console.WriteLine($"[指令] 握手成功: {jObj["Message"]}");
}
else
{
Console.WriteLine($"[警告] 未知的指令: {action}");
}
}
catch (Exception ex)
{
Console.WriteLine($"[分发错误] {ex.Message}");
}
}
}

30
SHH.CameraService/Core/CmdClients/ConnectedClient.cs Normal file
View File

@@ -0,0 +1,30 @@
namespace SHH.CameraService;
/// <summary>
/// 在线客户端信息模型 (已更新)
/// </summary>
public class ConnectedClient
{
/// <summary> 唯一标识 (AppId) </summary>
public string ServiceId { get; set; } = string.Empty;
/// <summary> 版本号 </summary>
public string Version { get; set; } = "1.0.0";
/// <summary> 远程进程 ID </summary>
public int Pid { get; set; }
/// <summary> 客户端 IP </summary>
public string Ip { get; set; } = string.Empty;
/// <summary> WebAPI 端口 (Dashboard 调用 REST 接口用) </summary>
public int WebPort { get; set; }
/// <summary> 该客户端正在推流的目标地址 </summary>
public List<string> TargetVideoNodes { get; set; } = new List<string>();
public DateTime LastHeartbeat { get; set; }
// 辅助属性：拼接出完整的 API BaseUrl
public string WebApiUrl => $"http://{Ip}:{WebPort}";
}

18
SHH.CameraService/Core/CmdClients/ICommandHandler.cs Normal file
View File

@@ -0,0 +1,18 @@
namespace SHH.CameraService;
/// <summary>
/// 抽象指令处理器接口
/// </summary>
public interface ICommandHandler
{
/// <summary>
/// 该处理器支持的 Action 名称 (如 "AddCamera", "Reboot")
/// </summary>
string ActionName { get; }
/// <summary>
/// 执行指令逻辑
/// </summary>
/// <param name="payload">指令携带的数据 (JSON JToken)</param>
Task ExecuteAsync(Newtonsoft.Json.Linq.JToken payload);
}

103
SHH.CameraService/Core/CmdClients/SyncCameraHandler.cs Normal file
View File

@@ -0,0 +1,103 @@
using Newtonsoft.Json.Linq;
using SHH.CameraSdk; // 引用包含 FrameController 和 FrameRequirement 的命名空间
using SHH.Contracts;
namespace SHH.CameraService;
public class SyncCameraHandler : ICommandHandler
{
private readonly CameraManager _cameraManager;
public string ActionName => "SyncCamera";
public SyncCameraHandler(CameraManager cameraManager)
{
_cameraManager = cameraManager;
}
public async Task ExecuteAsync(JToken payload)
{
// 1. 解析配置
var dto = payload.ToObject<CameraConfigDto>();
if (dto == null) return;
// 2. 添加设备到管理器 (这一步是必须的，不然没有 Device 就没有 Controller)
var videoConfig = new VideoSourceConfig
{
Id = dto.Id,
Name = dto.Name,
IpAddress = dto.IpAddress,
Port = dto.Port,
Username = dto.Username,
Password = dto.Password,
ChannelIndex = dto.ChannelIndex,
StreamType = dto.StreamType,
Brand = (DeviceBrand)dto.Brand,
RenderHandle = (IntPtr)dto.RenderHandle,
MainboardIp = dto.MainboardIp,
MainboardPort = dto.MainboardPort,
// 必须给个默认值，防止空引用
VendorArguments = new Dictionary<string, string>(),
};
// 如果设备不存在才添加，如果已存在，后续逻辑会直接获取
if (_cameraManager.GetDevice(videoConfig.Id) == null)
{
_cameraManager.AddDevice(videoConfig);
}
// 3. 核心：直接获取设备实例
var device = _cameraManager.GetDevice(dto.Id);
if (device == null)
{
Console.WriteLine($"[SyncError] 设备 {dto.Id} 创建失败，无法执行自动订阅。");
return;
}
// 4. 拿到你的“宝贝”控制器 (FrameController)
var controller = device.Controller;
if (controller == null)
{
Console.WriteLine($"[SyncError] 设备 {dto.Id} 不支持流控调度 (Controller is null)。");
return;
}
// 5. 暴力注册订阅需求 (Loop AutoSubscriptions)
if (dto.AutoSubscriptions != null && dto.AutoSubscriptions.Count > 0)
{
foreach (var subItem in dto.AutoSubscriptions)
{
// 生成 AppId (照抄你给的逻辑)
string finalAppId = string.IsNullOrWhiteSpace(subItem.AppId)
? $"SUB_{Guid.NewGuid().ToString("N").Substring(0, 8).ToUpper()}"
: subItem.AppId;
Console.WriteLine($"[自动化] 正在注册流控: {finalAppId}, 目标: {subItem.TargetFps} FPS");
// 构造 FrameRequirement 对象 (完全匹配你 FrameController 的入参)
// 这里的属性赋值对应你代码里 req.Type, req.SavePath 等逻辑
var requirement = new FrameRequirement
{
AppId = finalAppId,
TargetFps = subItem.TargetFps, // 8帧 或 1帧
Type = (SubscriptionType)subItem.Type, // 业务类型 (LocalWindow, NetworkTrans...)
Memo = subItem.Memo ?? "Auto Sync",
// 其它字段给默认空值，防止 Controller 内部逻辑报错
Handle = "",
SavePath = ""
};
// ★★★ 见证奇迹的时刻：直接调用 Register ★★★
controller.Register(requirement);
}
}
//// 6. 启动设备
//// 你的积分算法会在 device 内部的推流循环中被 MakeDecision 调用
if (dto.ImmediateExecution)
await device.StartAsync();
Console.WriteLine($"[SyncSuccess] 设备 {dto.Id} 同步完成，策略已下发。");
}
}

24
SHH.CameraService/Core/Configs/PushTargetConfig.cs Normal file
View File

@@ -0,0 +1,24 @@
namespace SHH.CameraService;
/// <summary>
/// 定义发送的目标
/// </summary>
public class PushTargetConfig
{
/// <summary>
/// 目标名称
/// </summary>
public string Name { get; set; }
= string.Empty;
/// <summary>
/// NetMQ 地址 (如 "tcp://1.2.3.4:5555")
/// </summary>
public string Endpoint { get; set; }
= string.Empty;
/// <summary>
/// 独立队列容量 (隔离的关键)
/// </summary>
public int QueueCapacity { get; set; } = 10;
}

30
SHH.CameraService/Core/Configs/StreamTarget.cs Normal file
View File

@@ -0,0 +1,30 @@
namespace SHH.CameraService;
/// <summary>
/// 代表一个独立的推送目标
/// 包含：配置信息 + 专属于它的数据管道
/// </summary>
public class StreamTarget
{
/// <summary>
/// 配置
/// </summary>
public PushTargetConfig Config { get; }
/// <summary>
/// 管道
/// </summary>
public VideoDataChannel Channel { get; }
/// <summary>
/// 构造函数
/// </summary>
/// <param name="config"></param>
public StreamTarget(PushTargetConfig config)
{
Config = config;
// 为这个目标创建独立的管道，容量由配置决定
Channel = new VideoDataChannel(capacity: config.QueueCapacity);
}
}

52
SHH.CameraService/Core/NetSenders/CameraEngineWorker.cs Normal file
View File

@@ -0,0 +1,52 @@
using Microsoft.Extensions.Hosting;
using SHH.CameraSdk;
public class CameraEngineWorker : BackgroundService
{
private readonly CameraManager _manager;
public CameraEngineWorker(CameraManager manager)
{
// 理由：严谨性检查，防止因配置错误导致的空指针崩溃
_manager = manager ?? throw new ArgumentNullException(nameof(manager));
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
Console.WriteLine("[Engine] 正在启动核心引擎...");
try
{
// 1. 理由：启动 SDK 内部加载流程（从本地存储恢复设备）
await _manager.StartAsync();
Console.WriteLine("[Engine] 设备管理服务已启动。");
}
catch (Exception ex)
{
Console.WriteLine($"[Engine] 严重启动异常: {ex.Message}");
return; // 理由：核心组件失败，终止后续逻辑
}
// 2. 理由：Worker 必须保持活跃状态，以便作为宿主生命周期的一部分
while (!stoppingToken.IsCancellationRequested)
{
// 你可以在这里定期输出一些状态统计
// Console.WriteLine($"[Engine] 活跃设备数: {_manager.GetActiveCount()}");
await Task.Delay(TimeSpan.FromSeconds(30), stoppingToken);
}
}
public override async Task StopAsync(CancellationToken cancellationToken)
{
Console.WriteLine("[Engine] 正在执行优雅停机...");
try
{
// 理由：这是重构的核心。必须在 SDK 退出前释放所有非托管句柄
await _manager.DisposeAsync();
}
finally
{
await base.StopAsync(cancellationToken);
}
}
}

73
SHH.CameraService/Core/NetSenders/NetMQProtocolExtensions.cs Normal file
View File

@@ -0,0 +1,73 @@
using NetMQ;
using SHH.Contracts;
namespace SHH.CameraService
{
/// <summary>
/// 负责将业务契约转换为 ZeroMQ 传输协议
/// </summary>
public static class NetMQProtocolExtensions
{
private const string PROTOCOL_HEADER = "SHH_V1";
/// <summary>
/// 扩展方法：将 Payload 转为 NetMQMessage
/// 使用方法：var msg = payload.ToNetMqMessage();
/// </summary>
public static NetMQMessage ToNetMqMessage(this VideoPayload payload)
{
var msg = new NetMQMessage();
// Frame 0: 协议魔数
msg.Append(PROTOCOL_HEADER);
// Frame 1: 元数据 JSON
msg.Append(payload.GetMetadataJson());
// Frame 2: 原始图 (保持帧位对齐，无数据则发空帧)
if (payload.HasOriginalImage && payload.OriginalImageBytes != null)
msg.Append(payload.OriginalImageBytes);
else
msg.Append(Array.Empty<byte>());
// Frame 3: 处理图
if (payload.HasTargetImage && payload.TargetImageBytes != null)
msg.Append(payload.TargetImageBytes);
else
msg.Append(Array.Empty<byte>());
return msg;
}
/// <summary>
/// 扩展方法：从 NetMQMessage 还原 Payload
/// </summary>
public static VideoPayload ToVideoPayload(this NetMQMessage msg)
{
if (msg == null || msg.FrameCount < 4) return null;
// Frame 0 Check
if (msg[0].ConvertToString() != PROTOCOL_HEADER) return null;
// Frame 1: Metadata
string json = msg[1].ConvertToString();
var payload = VideoPayload.FromMetadataJson(json);
if (payload == null) return null;
// Frame 2: Raw Image
// 利用 BufferSize 避免不必要的内存拷贝，如果长度为0则跳过
if (payload.HasOriginalImage && msg[2].BufferSize > 0)
{
payload.OriginalImageBytes = msg[2].ToByteArray();
}
// Frame 3: Processed Image
if (payload.HasTargetImage && msg[3].BufferSize > 0)
{
payload.TargetImageBytes = msg[3].ToByteArray();
}
return payload;
}
}
}

62
SHH.CameraService/Core/NetSenders/NetMqSenderWorker.cs Normal file
View File

@@ -0,0 +1,62 @@
using Microsoft.Extensions.Hosting;
using NetMQ;
using NetMQ.Sockets;
namespace SHH.CameraService;
/// <summary>
/// NetMQ 发送工作者
/// 职责：从指定目标的 VideoDataChannel 读取 Payload，通过 ZeroMQ 发送出去
/// </summary>
public class NetMqSenderWorker : BackgroundService
{
private readonly StreamTarget _target;
// 构造函数注入特定的目标对象 (由 Program.cs 的工厂方法提供)
public NetMqSenderWorker(StreamTarget target)
{
_target = target;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
Console.WriteLine($"[NetMqSender] 正在连接至服务端: {_target.Config.Endpoint} ...");
// ★★★ 修正点：必须使用 PublisherSocket 来配合接收端的 SubscriberSocket ★★★
// 虽然是 Connect 模式，Publisher 依然可以 Connect
using var clientSocket = new PublisherSocket();
// 设置高水位 (HWM)
// 对于 Publisher，如果队列满了，默认行为就是丢弃旧数据，这非常符合视频流需求
clientSocket.Options.SendHighWatermark = 1000;
// 主动连接
clientSocket.Connect(_target.Config.Endpoint);
Console.WriteLine("[NetMqSender] 连接成功，开始从通道搬运数据...");
await foreach (var payload in _target.Channel.Reader.ReadAllAsync(stoppingToken))
{
try
{
var msg = payload.ToNetMqMessage();
// 发送消息
// PublisherSocket 的 TrySend 如果没人订阅或者队列满了，通常不会阻塞，而是直接丢弃或返回
// 注意：PUB 模式下，第一帧 ("SHH_V1") 会被当作订阅的主题 (Topic)。
// 你的接收端订阅了 "" (空字符串)，所以能收到以任何字符串开头的数据。
bool sent = clientSocket.TrySendMultipartMessage(msg);
if (!sent)
{
// 这种情况通常意味着网络断了且 HWM 队列也满了
Console.WriteLine($"[NetMqSender] 警告: 发送队列已满，正在丢帧...");
msg.Clear(); // 手动清理(可选)
}
}
catch (Exception ex)
{
Console.WriteLine($"[NetMqSender] 异常: {ex.Message}");
}
}
}
}

126
SHH.CameraService/Core/NetSenders/NetworkStreamingWorker.cs Normal file
View File

@@ -0,0 +1,126 @@
using Microsoft.Extensions.Hosting;
using OpenCvSharp;
using SHH.CameraSdk; // 引用 SDK 核心
using SHH.Contracts;
using System.Diagnostics;
namespace SHH.CameraService;
public class NetworkStreamingWorker : BackgroundService
{
// 注入所有注册的目标（云端、大屏等），实现动态分发
private readonly IEnumerable<StreamTarget> _targets;
// 编码参数：JPG 质量 75 (平衡画质与带宽)
// 工业经验：75 是甜点，体积只有 100 的 1/3，肉眼几无区别。
// 如果您确实需要 100，请注意带宽压力。此处我保留您要求的 100，但建议未来调优。
private readonly int[] _encodeParams = { (int)ImwriteFlags.JpegQuality, 100 };
public NetworkStreamingWorker(IEnumerable<StreamTarget> targets)
{
_targets = targets;
}
protected override Task ExecuteAsync(CancellationToken stoppingToken)
{
Console.WriteLine("[StreamWorker] 启动流媒体采集引擎...");
// =========================================================
// 订阅逻辑：接入 "上帝模式" (God Mode)
// =========================================================
// 理由：NetMQ 网关需要无差别地获取所有设备的图像。
GlobalStreamDispatcher.OnGlobalFrame += ProcessFrame;
//Console.WriteLine($"[StreamWorker] 已挂载至全局广播总线，正在监听 {GlobalStreamDispatcher.OnGlobalFrame?.GetInvocationList().Length ?? 0} 个订阅者...");
var tcs = new TaskCompletionSource();
stoppingToken.Register(() =>
{
// 停止时反注册，防止静态事件内存泄漏
GlobalStreamDispatcher.OnGlobalFrame -= ProcessFrame;
Console.WriteLine("[StreamWorker] 已断开全局广播连接");
tcs.SetResult();
});
return tcs.Task;
}
/// <summary>
/// [回调函数] 处理每一帧图像
/// 注意：此方法运行在 SDK 的采集线程池中，必须极速处理，严禁阻塞！
/// </summary>
private void ProcessFrame(long deviceId, SmartFrame frame)
{
try
{
// 1. 基础校验 (合法性检查)
if (frame == null || frame.InternalMat.Empty()) return;
long startTick = Stopwatch.GetTimestamp();
// =========================================================
// 2. 一次编码 (One Encode) - CPU 消耗点
// =========================================================
// 理由：在这里同步编码是最安全的，因为出了这个函数 frame 内存就会失效。
// 且只编一次，后续分发给 10 个目标也只用这一份数据。
byte[] jpgBytes = null;
// 如果有更小的图片, 原始图片不压缩, 除非有特殊需求
if (frame.TargetMat == null)
{
jpgBytes = EncodeImage(frame.InternalMat);
}
// 双流支持：如果存在处理后的 AI 图，也一并编码
byte[] targetBytes = null;
if (frame.TargetMat != null && !frame.TargetMat.Empty())
{
targetBytes = EncodeImage(frame.TargetMat);
}
// =========================================================
// 3. 构建 Payload (数据载荷)
// =========================================================
var payload = new VideoPayload
{
CameraId = deviceId.ToString(),
CaptureTimestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
OriginalImageBytes = jpgBytes, // 引用赋值
TargetImageBytes = targetBytes, // 引用赋值
OriginalWidth = frame.TargetWidth,
OriginalHeight = frame.TargetHeight,
DispatchTimestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()
};
// 计算转码耗时(ms)
double processMs = (Stopwatch.GetTimestamp() - startTick) * 1000.0 / Stopwatch.Frequency;
payload.Diagnostics["encode_ms"] = Math.Round(processMs, 2);
// =========================================================
// 4. 动态扇出 (Dynamic Fan-Out) - 内存消耗极低
// =========================================================
// 遍历所有目标，往各自独立的管道里写数据。
// 实现了"物理隔离"：一个管道满了(云端卡顿)，不影响另一个管道(大屏流畅)。
foreach (var target in _targets)
{
// WriteLog 是非阻塞的。满了就丢弃，返回 false。
target.Channel.WriteLog(payload);
}
}
catch (Exception ex)
{
// 极少发生的内存错误，打印日志但不抛出，避免崩溃 SDK 线程
Console.WriteLine($"[StreamWorker] 采集处理异常: {ex.Message}");
}
}
/// <summary>
/// 辅助：OpenCV 内存编码
/// </summary>
private byte[] EncodeImage(Mat mat)
{
// ImEncode 将 Mat 编码为一维字节数组 (托管内存)
Cv2.ImEncode(".jpg", mat, out byte[] buf, _encodeParams);
return buf;
}
}

0
SHH.CameraService/PipelineConfigurator.cs → SHH.CameraService/Core/NetSenders/PipelineConfigurator.cs
View File

40
SHH.CameraService/Core/NetSenders/VideoDataChannel.cs Normal file
View File

@@ -0,0 +1,40 @@
using System.Threading.Channels;
using SHH.Contracts;
namespace SHH.CameraService
{
/// <summary>
/// 视频数据内部总线 (线程安全的生产者-消费者通道)
/// <para>作用：解耦 [采集编码线程] 与 [网络发送线程]</para>
/// </summary>
public class VideoDataChannel
{
// 限制容量为 100 帧。如果积压超过 100 帧，说明发送端彻底堵死了，必须丢帧。
private readonly Channel<VideoPayload> _channel;
public VideoDataChannel(int capacity = 10)
{
var options = new BoundedChannelOptions(capacity)
{
FullMode = BoundedChannelFullMode.DropOldest, // 核心策略：满了就丢弃最旧的帧
SingleReader = false, // 允许多个发送 Worker (如 CloudWorker, ScreenWorker) 同时读取
SingleWriter = true // 只有一个采集线程在写
};
_channel = Channel.CreateBounded<VideoPayload>(options);
}
/// <summary>
/// [生产者] 写入一个封装好的数据包 (非阻塞)
/// </summary>
public void WriteLog(VideoPayload payload)
{
// TryWrite 永远不会等待，满了就丢旧的写入新的，返回 true
_channel.Writer.TryWrite(payload);
}
/// <summary>
/// [消费者] 读取器
/// </summary>
public ChannelReader<VideoPayload> Reader => _channel.Reader;
}
}

0
SHH.CameraService/ParentProcessSentinel.cs → SHH.CameraService/Core/ParentProcessSentinel.cs
View File

91
SHH.CameraService/NetworkStreamManager.cs
View File

@@ -1,91 +0,0 @@
using SHH.CameraSdk;
using System.Collections.Concurrent;
namespace SHH.CameraService;
/// <summary>
/// 网络推流管理器
/// 职责：管理 ZeroMQ 推流任务的生命周期
/// 类似于 DisplayWindowManager，它负责订阅数据并将其桥接到传输层
/// </summary>
public class NetworkStreamManager
{
private readonly VideoDataChannel _channel;
// 记录当前活跃的推流任务，防止重复订阅
private readonly ConcurrentDictionary<string, bool> _activeStreams = new();
public NetworkStreamManager(VideoDataChannel channel)
{
_channel = channel;
}
/// <summary>
/// 启动推流任务
/// </summary>
public void StartStream(string appId, long deviceId)
{
// 1. 防止重复启动
if (_activeStreams.ContainsKey(appId)) return;
// 2. 向全局分发器订阅精准数据
// 这里实现了业务逻辑的闭环：只有被 Manager 管理的任务才会消耗 CPU 去转码
GlobalStreamDispatcher.Subscribe(appId, deviceId, (frame) =>
{
// --- 这里的代码运行在分发线程中 ---
// A. 转码 (耗时操作封装在这里，不污染 Controller)
byte[] jpgBytes = EncodeFrameToJpg(frame);
if (jpgBytes != null && jpgBytes.Length > 0)
{
var payload = new VideoPayload
{
CameraId = appId, // 使用 AppId 作为 Topic (给 Dashboard 订阅用)
OriginalImageBytes = jpgBytes,
CaptureTime = DateTime.Now,
OriginalWidth = frame.TargetWidth,
OriginalHeight = frame.TargetHeight
};
// B. 写入传输通道
_ = _channel.WriteAsync(payload);
}
});
_activeStreams.TryAdd(appId, true);
Console.WriteLine($"[Network] 推流任务已启动: {appId} -> Device {deviceId}");
}
/// <summary>
/// 停止推流任务
/// </summary>
public void StopStream(string appId)
{
if (_activeStreams.TryRemove(appId, out _))
{
// 1. 从全局分发器注销
GlobalStreamDispatcher.Unsubscribe(appId);
Console.WriteLine($"[Network] 推流任务已停止: {appId}");
}
}
// --- 辅助方法 ---
private byte[] EncodeFrameToJpg(SmartFrame frame)
{
try
{
// 优先使用处理后的 TargetMat，如果没有则用原始的 InternalMat
var mat = frame.TargetMat ?? frame.InternalMat;
if (mat != null && !mat.Empty())
{
// 80 质量平衡体积与画质
return mat.ImEncode(".jpg", new int[] { 1, 80 });
}
}
catch (Exception ex)
{
Console.WriteLine($"[Network] 转码失败: {ex.Message}");
}
return Array.Empty<byte>();
}
}

239
SHH.CameraService/Program.cs
View File

@@ -1,7 +1,7 @@
using Microsoft.AspNetCore.Builder;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.OpenApi.Models;
using SHH.CameraSdk; // 引用你的业务核心
using SHH.CameraSdk;
namespace SHH.CameraService;
@@ -9,163 +9,152 @@ public class Program
{
public static async Task Main(string[] args)
{
// 缓冲时间 (您之前写了20000ms即20秒，可能是为了附加调试器。如果觉得太慢可以改回 2000)
for(var i=1; i<10; i++)
// 1. 理由：缓冲时间 10 秒, 供附加调试工具使用
for (var i = 1; i < 10; i++)
Thread.Sleep(1000);
// 1. 解析配置
// =============================================================
// 2. 基础环境与配置 (理由：明确身份 ID 和 监听端口)
// =============================================================
var config = ServiceConfig.BuildFromArgs(args);
// ---【补全变量定义】---
// A. 补全 webPort (统一使用 config.BasePort)
int webPort = config.BasePort;
// B. 补全 processIdInt (用于 FileStorage 和 CameraSdk)
// 逻辑：尝试将 AppId 解析为数字；如果 AppId 是字符串(如"CameraApp_01")，则默认给 1，或者根据 BasePort 推算
int processIdInt = config.NumericId;
Console.Title = $"SHH Gateway - {config.AppId} (Web: {webPort})";
#region --- 2. 硬件环境预热 ---
InitHardwareEnv();
#endregion
#region --- 3. 构建 WebHost ---
// 硬件预热 (理由：确保底层驱动库在 Web 容器启动前完全就绪)
HikNativeMethods.NET_DVR_Init();
HikSdkManager.ForceWarmUp();
var builder = WebApplication.CreateBuilder(args);
// ★★★ 核心：注入全局配置 ★★★
// =============================================================
// 3. 依赖注入注册 (DI)
// =============================================================
builder.Services.AddSingleton(config);
// -------------------------------------------------------------
// A. 注册新架构组件
// -------------------------------------------------------------
builder.Services.AddSingleton<VideoDataChannel>();
// 推流服务 (连接 config.TargetClients 里的 :6002)
builder.Services.AddHostedService<ZeroMQBridgeWorker>();
// 指令客户端 (连接 config.TargetClients 里的 :6001)
builder.Services.AddHostedService<CommandClientWorker>();
// 进程守护
builder.Services.AddHostedService<ParentProcessSentinel>();
// -------------------------------------------------------------
// B. 注册 SDK 业务服务
// -------------------------------------------------------------
// 使用刚刚补全的 processIdInt
builder.Services.AddSingleton<IStorageService>(new FileStorageService(processIdInt));
builder.Services.AddSingleton<CameraManager>();
// 注册缩放与增亮业务（不注册则不实现）
builder.Services.AddSingleton<ProcessingConfigManager>();
builder.Services.AddSingleton<DisplayWindowManager>();
builder.Services.AddSingleton<NetworkStreamManager>();
builder.Services.AddSingleton<ImageScaleCluster>(sp => new ImageScaleCluster(4, sp.GetRequiredService<ProcessingConfigManager>()));
builder.Services.AddSingleton<ImageEnhanceCluster>(sp => new ImageEnhanceCluster(4, sp.GetRequiredService<ProcessingConfigManager>()));
builder.Services.AddSingleton(sp => new ImageScaleCluster(4, sp.GetRequiredService<ProcessingConfigManager>()));
builder.Services.AddSingleton(sp => new ImageEnhanceCluster(4, sp.GetRequiredService<ProcessingConfigManager>()));
builder.Services.AddHostedService<PipelineConfigurator>();
// 使用补全的 processIdInt
builder.Services.AddCameraSdk(processIdInt);
// 接入 SDK 核心逻辑
builder.Services.AddCameraSdk(config.NumericId);
// 注册后台引擎 (理由：托管长周期的硬件状态监控)
builder.Services.AddHostedService<CameraEngineWorker>();
builder.Services.AddSingleton<ConnectivitySentinel>();
builder.Services.AddControllers().AddApplicationPart(typeof(CamerasController).Assembly);
builder.Services.AddControllers().AddApplicationPart(typeof(MonitorController).Assembly);
// 配置 Web 相关的服务
ConfigureWebServices(builder, config);
// -------------------------------------------------------------
// C. Web API 基础
// -------------------------------------------------------------
builder.Services.AddControllers().AddControllersAsServices();
builder.Services.AddEndpointsApiExplorer();
builder.Services.AddSwaggerGen(c =>
// 配置进程守护
builder.Services.AddHostedService<ParentProcessSentinel>();
// =============================================================
// 4. 接受启动传参, 并支持将视频进行网络广播
// =============================================================
// 1. 读取配置创建 targets (可以是 1 个，也可以是 10 个)
var netTargets = new List<StreamTarget>();
if (config.VideoEndpoints != null)
{
// 【修正】使用 config.AppId
c.SwaggerDoc("v1", new OpenApiInfo { Title = $"Gateway {config.AppId}", Version = "v1" });
});
foreach(var cfgVideo in config.VideoEndpoints)
{
netTargets.Add(new StreamTarget(new PushTargetConfig
{
Name = cfgVideo.Description, Endpoint = cfgVideo.Uri, QueueCapacity = 10,
}));
}
}
// 2. 注册 Targets (供采集者用)
builder.Services.AddSingleton<IEnumerable<StreamTarget>>(netTargets);
builder.Services.AddCors(o => o.AddPolicy("AllowAll", p => p.AllowAnyOrigin().AllowAnyHeader().AllowAnyMethod()));
// 3. 注册采集者 (它会注入上面的 targets，进行编码和分发)
builder.Services.AddHostedService<NetworkStreamingWorker>();
// 4. 为每个 Target 注册一个独立的发送者
foreach (var target in netTargets)
{
builder.Services.AddHostedService(sp => new NetMqSenderWorker(target));
}
// =============================================================
// 5. 命令管道配置
// =============================================================
// 负责连接 Dashboard，注册身份，接收重启/控制指令
builder.Services.AddHostedService<CommandClientWorker>();
// 1. 注册分发器
builder.Services.AddSingleton<CommandDispatcher>();
// 2. 注册具体的指令处理器 (每写一个新的 Handler，就在这里注册一下，或者用反射批量注册)
builder.Services.AddSingleton<ICommandHandler, SyncCameraHandler>();
// =============================================================
// 6. 构建与管道配置
// =============================================================
var app = builder.Build();
//// =======================================================================
//// ★★★ 核心接入点：连接 [现有分发器] 与 [新推流通道] ★★★
//// =======================================================================
//// 1. 获取刚刚注册的数据通道
//var videoChannel = app.Services.GetRequiredService<VideoDataChannel>();
////var config = app.Services.GetRequiredService<ServiceConfig>();
//// 2. 订阅你现有的全局事件 (这里就是“取货”的地方)
//// 每当 HikVideoSource 采集到一帧并调用 Dispatch 时，这里就会触发
//GlobalStreamDispatcher.OnGlobalFrame += (deviceId, smartFrame) =>
//{
// // 3. 数据处理：将 OpenCvSharp Mat 转为 JPG 字节流 (网络传输必须压缩)
// byte[] jpgData = EncodeToJpg(smartFrame);
// if (jpgData != null && jpgData.Length > 0)
// {
// // 4. 封装载荷
// var payload = new VideoPayload
// {
// // 使用 AppId 或 DeviceId 作为标识
// CameraId = config.AppId,
// OriginalImageBytes = jpgData,
// CaptureTime = DateTime.Now,
// OriginalWidth = smartFrame.TargetWidth,
// OriginalHeight = smartFrame.TargetHeight
// };
// // 5. 扔进通道 (Fire-and-Forget，不阻塞你原来的显示逻辑)
// // WriteAsync 是 ValueTask，这里忽略等待，追求最高吞吐
// _ = videoChannel.WriteAsync(payload);
// }
//};
//Console.WriteLine("[System] 全局流已桥接到 ZeroMQ 推流通道");
// 核心修复：同步点火逻辑 (理由：在 Web 开启前完成设备池的初步构建)
await StartBusinessLogic(app);
app.UseSwagger();
app.UseSwaggerUI();
app.UseSwaggerUI(c =>
{
c.SwaggerEndpoint("/swagger/v1/swagger.json", $"SHH Gateway #{config.AppId}");
});
app.MapGet("/", () => $"SHH Gateway {config.AppId} is running.");
app.UseCors("AllowAll");
// 理由：正式映射控制器路由
app.MapControllers();
// 【修正】使用 webPort
Console.WriteLine($"[System] Web API 已启动: http://0.0.0.0:{webPort}");
await app.RunAsync($"http://0.0.0.0:{webPort}");
#endregion
}
static void InitHardwareEnv()
{
Console.WriteLine("=== 工业级视频接入服务启动 ===");
// =============================================================
// 5. 正式启动
// =============================================================
await app.RunAsync($"http://0.0.0.0:{config.BasePort}");
}
/// <summary>
/// 内存转码：Mat -> Jpg Bytes
/// 对齐业务启动：激活单例并启动相机管理器
/// </summary>
static byte[] EncodeToJpg(SmartFrame frame)
static async Task StartBusinessLogic(WebApplication app)
{
try
var manager = app.Services.GetRequiredService<CameraManager>();
// 激活哨兵逻辑 (理由：显式 Get 触发单例构造，否则不工作)
_ = app.Services.GetRequiredService<ConnectivitySentinel>();
// 启动相机任务加载
await manager.StartAsync();
Console.WriteLine("[System] 核心业务逻辑已激活。");
}
/// <summary>
/// 注册 Web API 支持
/// </summary>
static void ConfigureWebServices(WebApplicationBuilder builder, ServiceConfig cfg)
{
builder.Services.AddCors(options =>
{
// 假设 SmartFrame 内部持有 OpenCvSharp.Mat 类型的 InternalMat
if (frame != null && frame.InternalMat != null && !frame.InternalMat.Empty())
{
// 80 是 JPG 质量参数，平衡画质与带宽
return frame.InternalMat.ImEncode(".jpg", new int[] { 1, 80 });
}
}
catch
options.AddPolicy("AllowAll", p => p.AllowAnyOrigin().AllowAnyHeader().AllowAnyMethod());
});
// ★★★★★ 补全点：跨项目控制器加载 ★★★★★
// 理由：Controller 定义在 SDK 项目中，必须通过 AddApplicationPart 显式挂载
builder.Services.AddControllers(options =>
{
// 容错处理，防止一帧损坏导致程序崩溃
}
return Array.Empty<byte>();
options.Filters.Add<UserActionFilter>();
})
.AddApplicationPart(typeof(CamerasController).Assembly) // 必备：加载相机控制接口
.AddApplicationPart(typeof(MonitorController).Assembly); // 必备：加载监控接口
builder.Services.AddEndpointsApiExplorer();
builder.Services.AddSwaggerGen(c =>
{
c.SwaggerDoc("v1", new OpenApiInfo { Title = $"SHH Gateway #{cfg.AppId}", Version = "v1" });
});
}
}

6
SHH.CameraService/SHH.CameraService.csproj
View File

@@ -7,6 +7,12 @@
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<Compile Remove="Dels\**" />
<EmbeddedResource Remove="Dels\**" />
<None Remove="Dels\**" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="NetMQ" Version="4.0.2.2" />
</ItemGroup>

63
SHH.CameraService/VideoDataChannel.cs
View File

@@ -1,63 +0,0 @@
using System.Threading.Channels;
namespace SHH.CameraService;
/// <summary>
/// 视频数据高速通道
/// <para>作用：解耦 采集线程(Producer) 和 发送线程(Consumer)</para>
/// <para>特性：使用 BoundedChannel，当网络发送慢时，自动丢弃旧帧(DropOldest)，防止内存溢出。</para>
/// </summary>
public class VideoDataChannel
{
// 创建一个有限容量的通道 (容量 5)
// 如果发送端太慢，这就满了，DropOldest 会丢弃最旧的帧，保证实时性
private readonly Channel<VideoPayload> _channel = Channel.CreateBounded<VideoPayload>(
new BoundedChannelOptions(5)
{
FullMode = BoundedChannelFullMode.DropOldest, // 核心策略：丢弃旧帧
SingleReader = true, // 只有一个 ZeroMQWorker 在读
SingleWriter = false //可能有多个相机线程在写
});
// ★★★ 新增：公开 Reader 属性，让外部可以直接调用 ReadAsync ★★★
public ChannelReader<VideoPayload> Reader => _channel.Reader;
/// <summary>
/// 写入数据 (生产者调用)
/// </summary>
public ValueTask WriteAsync(VideoPayload payload)
{
return _channel.Writer.WriteAsync(payload);
}
/// <summary>
/// 读取数据流 (消费者调用)
/// </summary>
public IAsyncEnumerable<VideoPayload> ReadAllAsync(CancellationToken ct)
{
return _channel.Reader.ReadAllAsync(ct);
}
}
// 附带：如果您的项目中还没有定义 VideoPayload，这里是一个最小实现
// 如果 SHH.Contracts 中已有，请忽略此类
public class VideoPayload
{
/// <summary> 相机唯一标识 </summary>
public string CameraId { get; set; } = string.Empty;
/// <summary> 采集时间 </summary>
public DateTime CaptureTime { get; set; }
/// <summary> 发送时间 </summary>
public DateTime DispatchTime { get; set; }
/// <summary> 原始宽 </summary>
public int OriginalWidth { get; set; }
/// <summary> 原始高 </summary>
public int OriginalHeight { get; set; }
/// <summary> 已编码的图片数据 (JPG) </summary>
public byte[] OriginalImageBytes { get; set; } = Array.Empty<byte>();
}

87
SHH.CameraService/ZeroMQBridgeWorker.cs
View File

@@ -1,87 +0,0 @@
using Microsoft.Extensions.Hosting;
using NetMQ;
using NetMQ.Sockets;
using SHH.CameraSdk;
namespace SHH.CameraService;
public class ZeroMQBridgeWorker : BackgroundService
{
private readonly ServiceConfig _config;
private readonly VideoDataChannel _channel; // 数据源
public ZeroMQBridgeWorker(ServiceConfig config, VideoDataChannel channel)
{
_config = config;
_channel = channel;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
// 1. 如果不是主动/混合模式，不需要连接
if (!_config.ShouldConnect) return;
// ★★★ 核心修正：直接读取解析好的视频地址列表 ★★★
// 这些地址来自参数 --uris "IP,VideoPort&CommandPort" 中的 VideoPort 部分 (符号左边)
var streamUris = _config.VideoEndpoints;
if (streamUris.Count == 0)
{
Console.WriteLine("[推流] 未在参数中找到视频通道地址(位于&符号左侧)，跳过连接。");
return;
}
// 2. 初始化 Publisher Socket
// 特点：只需 Send 一次，底层会自动分发给所有 Connect 的 Dashboard
using var pubSocket = new PublisherSocket();
// 设置发送高水位 (HWM)
// 防止网络拥塞或接收端处理慢时，内存无限增长。超过50帧积压就开始丢弃旧帧。
pubSocket.Options.SendHighWatermark = 50;
// 3. 连接所有视频目标
foreach (var uri in streamUris)
{
Console.WriteLine($"[推流] 连接视频接收端: {uri}");
pubSocket.Connect(uri);
}
Console.WriteLine($"[推流] 服务就绪 (AppId: {_config.AppId})，等待视频帧...");
// 4. 推流循环
while (!stoppingToken.IsCancellationRequested)
{
try
{
// 从通道读取最新帧 (支持异步等待)
// 注意：这里使用了之前 VideoDataChannel 暴露出来的 Reader 属性
var payload = await _channel.Reader.ReadAsync(stoppingToken);
// 简单校验
if (payload == null || payload.OriginalImageBytes == null || payload.OriginalImageBytes.Length == 0)
continue;
// 构造 Topic (通常用 AppId 作为 Topic，这样 Dashboard 可以按需订阅)
string topic = _config.AppId;
// 发送两帧：[Topic] [ImageBytes]
// 这样 Dashboard 的 Subscriber 可以通过 Subscribe(topic) 来过滤
pubSocket.SendMoreFrame(topic)
.SendFrame(payload.OriginalImageBytes);
// 调试日志 (生产环境建议注释掉，否则刷屏)
// Console.WriteLine($"[推流] Sent {payload.OriginalImageBytes.Length} bytes");
}
catch (OperationCanceledException)
{
break; // 正常退出
}
catch (Exception ex)
{
Console.WriteLine($"[推流] 发送异常: {ex.Message}");
// 发生错误稍微停顿，防止死循环占用 CPU
await Task.Delay(1000, stoppingToken);
}
}
}
}

140
SHH.CameraService/ZeroMqBridgeService.cs
View File

@@ -1,140 +0,0 @@
using Microsoft.Extensions.Hosting;
using OpenCvSharp;
using SHH.Contracts;
using SHH.NetMQ;
namespace SHH.CameraSdk
{
/// <summary>
/// ZeroMQ 消息桥接服务（后台服务）。
/// 核心职责：订阅系统全局视频帧广播，将帧数据编码为标准协议格式后，通过 ZeroMQ 分发至外部系统。
/// 设计特性：
/// <para>1. 无侵入集成：通过订阅 GlobalStreamDispatcher 事件，无需修改原有帧处理流程。</para>
/// <para>2. 自动适配：支持动态增删设备，无需手动注册设备监听。</para>
/// <para>3. 安全隔离：帧数据深拷贝，避免跨线程内存访问冲突。</para>
/// </summary>
public class ZeroMqBridgeService : BackgroundService
{
#region --- 依赖注入字段 ---
/// <summary>
/// ZeroMQ 分发服务器（用于广播帧数据至多个订阅端）
/// </summary>
private readonly DistributorServer _distributor;
/// <summary>
/// ZeroMQ 转发客户端（用于定向推送帧数据至指定目标）
/// </summary>
private readonly ForwarderClient _forwarder;
#endregion
#region --- 构造函数 ---
/// <summary>
/// 初始化 <see cref="ZeroMqBridgeService"/> 实例。
/// </summary>
/// <param name="distributor">ZeroMQ 分发服务器实例（通过 DI 注入）</param>
/// <param name="forwarder">ZeroMQ 转发客户端实例（通过 DI 注入）</param>
public ZeroMqBridgeService(DistributorServer distributor, ForwarderClient forwarder)
{
_distributor = distributor ?? throw new ArgumentNullException(nameof(distributor));
_forwarder = forwarder ?? throw new ArgumentNullException(nameof(forwarder));
}
#endregion
#region --- 后台服务核心逻辑 ---
/// <summary>
/// 启动后台服务，订阅全局视频帧广播。
/// </summary>
/// <param name="stoppingToken">服务停止令牌（用于优雅关闭）</param>
protected override Task ExecuteAsync(CancellationToken stoppingToken)
{
Console.WriteLine("[ZeroMQ Bridge] 正在连接全局视频帧广播总线...");
// 订阅全局帧广播事件：所有设备的帧数据都会触发该事件
// 无需手动绑定设备，动态增删的设备自动适配
GlobalStreamDispatcher.OnGlobalFrame += OnGlobalFrameReceived;
Console.WriteLine("[ZeroMQ Bridge] 全局总线连接成功！已开始监听所有设备帧数据。");
Console.WriteLine("[ZeroMQ Bridge] 说明：动态增删的设备会自动转发，无需重启服务。");
// 返回空任务：服务通过事件驱动，无需阻塞主线程
return Task.CompletedTask;
}
/// <summary>
/// 停止后台服务，取消事件订阅以避免内存泄漏。
/// </summary>
/// <param name="cancellationToken">取消令牌</param>
public override Task StopAsync(CancellationToken cancellationToken)
{
Console.WriteLine("[ZeroMQ Bridge] 正在停止服务，取消全局总线订阅...");
// 取消事件订阅：必须执行，否则会导致内存泄漏
GlobalStreamDispatcher.OnGlobalFrame -= OnGlobalFrameReceived;
Console.WriteLine("[ZeroMQ Bridge] 服务已安全停止。");
return base.StopAsync(cancellationToken);
}
#endregion
#region --- 帧数据处理核心逻辑 ---
/// <summary>
/// 全局帧数据接收回调（事件处理函数）。
/// 处理流程：安全检查 → 帧数据深拷贝 → JPG 编码 → 封装为标准协议 → ZeroMQ 分发。
/// </summary>
/// <param name="deviceId">产生该帧的设备唯一标识</param>
/// <param name="frame">智能帧对象（包含原始/处理后图像数据）</param>
private void OnGlobalFrameReceived(long deviceId, SmartFrame frame)
{
try
{
// 1. 安全校验：跳过空帧或已释放的帧
var sourceMat = frame.TargetMat ?? frame.InternalMat;
if (sourceMat == null || sourceMat.Empty() || sourceMat.IsDisposed)
return;
// 2. 深拷贝图像数据：避免跨线程访问冲突（原帧可能被其他模块异步释放）
using var safeMat = sourceMat.Clone();
// 3. 图像编码：将 OpenCV Mat 转换为 JPG 字节数组（质量70，平衡画质与性能）
var jpgEncodeParams = new int[] { (int)ImwriteFlags.JpegQuality, 70 };
byte[] jpgBytes = safeMat.ImEncode(".jpg", jpgEncodeParams);
// 4. 封装为标准传输协议：使用 SHH.Contracts 中的 VideoPayload 统一格式
var videoPayload = new VideoPayload
{
CameraId = deviceId.ToString(), // 设备ID（转为字符串，兼容协议标准）
CaptureTime = DateTime.Now, // 帧采集时间（当前时间）
DispatchTime = DateTime.Now, // 帧分发时间（当前时间）
OriginalWidth = safeMat.Width, // 图像原始宽度（编码后宽度）
OriginalHeight = safeMat.Height, // 图像原始高度（编码后高度）
OriginalImageBytes = jpgBytes // JPG 编码后的二进制数据
};
// 5. 传递订阅者ID：保持与原帧的订阅者关联
if (frame.SubscriberIds.Any())
videoPayload.SubscriberIds.AddRange(frame.SubscriberIds);
// 6. ZeroMQ 分发：同时执行广播和定向推送（根据业务需求选择，可按需注释）
_distributor.Broadcast(videoPayload); // 广播给所有订阅端
_forwarder.Push(videoPayload); // 定向推送给指定目标
// 调试日志（生产环境建议注释，避免性能损耗）
// Console.WriteLine($"[ZeroMQ Bridge] 转发设备 {deviceId} 帧数据，大小：{jpgBytes.Length / 1024}KB");
}
catch (Exception ex)
{
// 异常隔离：单个帧处理失败不影响整体服务运行
Console.WriteLine($"[ZeroMQ Bridge] 帧转发失败（设备ID：{deviceId}）：{ex.Message}");
}
}
#endregion
}
}

174
SHH.Contracts/Commands/CameraConfigDto.cs Normal file
View File

@@ -0,0 +1,174 @@
using System.Collections.Generic;
using System.ComponentModel.DataAnnotations;
namespace SHH.Contracts
{
// ==============================================================================
// 1. 物理与运行配置 DTO (对应 CRUD 操作)
// 用于设备新增/全量配置查询，包含基础身份、连接信息、运行参数等全量字段
// ==============================================================================
public class CameraConfigDto
{
// --- 基础身份 (Identity) ---
/// <summary>
/// 设备唯一标识
/// </summary>
[Required(ErrorMessage = "设备ID不能为空")]
[Range(1, long.MaxValue, ErrorMessage = "设备ID必须为正整数")]
public long Id { get; set; }
/// <summary>
/// 设备友好名称
/// </summary>
[MaxLength(64, ErrorMessage = "设备名称长度不能超过64个字符")]
public string Name { get; set; }
/// <summary>
/// 摄像头品牌类型 (0:HikVision, 1:Dahua, 2:RTSP...)
/// </summary>
[Range(0, 10, ErrorMessage = "品牌类型值必须在0-10范围内")]
public int Brand { get; set; }
/// <summary>
/// 设备安装位置描述
/// </summary>
[MaxLength(128, ErrorMessage = "安装位置长度不能超过128个字符")]
public string Location { get; set; }
// --- 主板关联信息 (Metadata) ---
/// <summary>
/// 关联主板IP地址
/// </summary>
[RegularExpression(@"^((25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}(25[0-5]|2[0-4]\d|[01]?\d\d?)?$",
ErrorMessage = "请输入合法的IPv4地址")]
public string MainboardIp { get; set; } = string.Empty;
/// <summary>
/// 关联主板端口
/// </summary>
[Range(0, 65535, ErrorMessage = "主板端口号必须在1-65535范围内")]
public int MainboardPort { get; set; } = 0;
// --- 核心连接 (Connectivity) - 修改此类参数触发冷重启 ---
/// <summary>
/// 摄像头IP地址
/// </summary>
[Required(ErrorMessage = "IP地址不能为空")]
[RegularExpression(@"^((25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}(25[0-5]|2[0-4]\d|[01]?\d\d?)$",
ErrorMessage = "请输入合法的IPv4地址")]
public string IpAddress { get; set; }
/// <summary>
/// 登录用户名
/// </summary>
[MaxLength(32, ErrorMessage = "用户名长度不能超过32个字符")]
public string Username { get; set; }
/// <summary>
/// 登录密码
/// </summary>
[MaxLength(64, ErrorMessage = "密码长度不能超过64个字符")]
public string Password { get; set; }
/// <summary>
/// SDK端口 (如海康默认8000)
/// </summary>
[Range(1, 65535, ErrorMessage = "端口号必须在1-65535范围内")]
public ushort Port { get; set; }
/// <summary>
/// 通道号 (通常为1)
/// </summary>
[Range(0, 256, ErrorMessage = "通道号必须在0-256范围内")]
public int ChannelIndex { get; set; }
/// <summary>
/// 码流类型 (0:主码流, 1:子码流)
/// </summary>
[Range(0, 1, ErrorMessage = "码流类型只能是0(主码流)或1(子码流)")]
public int StreamType { get; set; }
// 渲染句柄 (通常下发时为0，由本地窗口绑定时再指定，或者此处仅作占位)
public long RenderHandle { get; set; }
/// <summary>
/// RTSP流路径 (备用或非SDK模式使用)
/// </summary>
[MaxLength(256, ErrorMessage = "RTSP地址长度不能超过256个字符")]
public string RtspPath { get; set; }
// --- 运行时参数 (Runtime Options) - 支持热更新 ---
/// <summary>
/// 是否使用灰度图 (用于AI分析场景加速)
/// </summary>
public bool UseGrayscale { get; set; } = false;
/// <summary>
/// 是否启用图像增强 (去噪/锐化等)
/// </summary>
public bool EnhanceImage { get; set; } = true;
// --- 画面变换 (Transform) - 支持热更新 ---
/// <summary>
/// 是否允许图像压缩 (降低带宽占用)
/// </summary>
public bool AllowCompress { get; set; } = true;
/// <summary>
/// 是否允许图像放大 (提升渲染质量)
/// </summary>
public bool AllowExpand { get; set; } = false;
/// <summary>
/// 目标分辨率 (格式如 1920x1080，空则保持原图)
/// </summary>
[RegularExpression(@"^\d+x\d+$", ErrorMessage = "分辨率格式必须为 宽度x高度 (如 1920x1080)")]
public string TargetResolution { get; set; } = string.Empty;
/// <summary>
/// 随配置一并下发的自动订阅请求
/// </summary>
public List<CameraConfigSubscribeDto> AutoSubscriptions { get; set; }
= new List<CameraConfigSubscribeDto>();
/// <summary>
/// 是否立即执行
/// </summary>
public bool ImmediateExecution { get; set; }
}
/// <summary>
/// 订阅项
/// </summary>
public class CameraConfigSubscribeDto
{
/// <summary>
/// 订阅标识
/// </summary>
public string AppId { get; set; }
/// <summary>
/// 订阅业务类型 SubscriptionType
/// </summary>
public int Type { get; set; }
/// <summary>
/// 要求的帧率：8帧或1帧
/// </summary>
public int TargetFps { get; set; }
/// <summary>
/// 备注
/// </summary>
public string Memo { get; set; }
/// <summary>
/// 是否需要高清晰度
/// </summary>
public bool NeedHighDefinition { get; set; }
= false;
}
}

1
SHH.Contracts/SHH.Contracts.csproj
View File

@@ -6,6 +6,7 @@
<ItemGroup>
<PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
<PackageReference Include="System.ComponentModel.Annotations" Version="5.0.0" />
</ItemGroup>
</Project>

97
SHH.Contracts/VideoPayload.cs
View File

@@ -1,120 +1,81 @@
using System;
using System.Collections.Generic;
using Newtonsoft.Json;
// 注意：如果不想依赖 Newtonsoft，也可以用 System.Text.Json，但 Newtonsoft 在 Std 2.0 中兼容性更好
namespace SHH.Contracts
{
/// <summary>
/// 视频数据传输契约（增强版）
/// 视频数据传输契约（纯净版 POCO）
/// </summary>
public class VideoPayload
{
/// <summary>
/// 初始化 <see cref="VideoPayload"/> 类的新实例。
/// </summary>
public VideoPayload()
{
// 预分配一个容量为 16 的列表，以减少内存分配和垃圾回收的压力。
SubscriberIds = new List<string>(16);
Diagnostics = new Dictionary<string, object>(4);
}
#region --- 元数据 (Metadata) ---
#region --- 1. 元数据 (Metadata) ---
/// <summary>
/// 获取订阅了此帧数据的客户端ID列表。
/// </summary>
public List<string> SubscriberIds { get; }
/// <summary>
/// 获取或设置摄像头的唯一标识符。
/// </summary>
public string CameraId { get; set; }
/// <summary>
/// 获取或设置图像的采集时间，即从SDK获取到图像数据的时间。
/// 采集时间戳 (Unix 毫秒)
/// </summary>
public DateTime CaptureTime { get; set; }
public long CaptureTimestamp { get; set; }
/// <summary>
/// 获取或设置图像的分发时间，即服务器准备将此帧数据发送给客户端的时间。
/// 分发时间戳 (Unix 毫秒)
/// </summary>
public DateTime DispatchTime { get; set; }
public long DispatchTimestamp { get; set; }
/// <summary>
/// 获取或设置图像的原始宽度。
/// </summary>
public int OriginalWidth { get; set; }
/// <summary>
/// 获取或设置图像的原始高度。
/// </summary>
public int OriginalHeight { get; set; }
/// <summary>
/// 获取或设置经过处理后的目标图像宽度。
/// </summary>
public int TargetWidth { get; set; }
public int TargetHeight { get; set; }
public List<string> SubscriberIds { get; }
public Dictionary<string, object> Diagnostics { get; }
/// <summary>
/// 获取或设置经过处理后的目标图像高度。
/// 指示标志：是否存在原始图
/// </summary>
public int TargetHeight { get; set; }
public bool HasOriginalImage { get; set; }
/// <summary>
/// 指示标志：是否存在处理图
/// </summary>
public bool HasTargetImage { get; set; }
#endregion
#region --- 核心二进制数据 ---
#region --- 2. 二进制数据 (Binary) ---
/// <summary>
/// 获取或设置原始图像的二进制数据（例如，从SDK直接获取的JPG或YUV数据）。
/// 此属性被标记为 <see cref="JsonIgnore"/>，以防止在序列化元数据时将其包含在内，从而避免严重的性能问题。
/// </summary>
// 标记 JsonIgnore，防止被错误序列化
[JsonIgnore]
public byte[] OriginalImageBytes { get; set; }
/// <summary>
/// 获取或设置经过处理后的目标图像的二进制数据（例如，经过缩放、画框或其他AI处理后的图像）。
/// 此属性可为空，表示此帧可能只包含原始图像或没有图像数据。
/// 同样，此属性也被标记为 <see cref="JsonIgnore"/>。
/// </summary>
[JsonIgnore]
public byte[] TargetImageBytes { get; set; }
#endregion
#region --- 序列化与反序列化辅助方法 ---
#region --- 3. 辅助方法 (仅保留 JSON 逻辑) ---
/// <summary>
/// 将当前对象的元数据序列化为一个纯净的 JSON 字符串。
/// 此方法会自动忽略所有二进制数据（<see cref="OriginalImageBytes"/> 和 <see cref="TargetImageBytes"/>）。
/// 获取纯元数据的 JSON 字符串
/// </summary>
/// <returns>包含元数据的 JSON 字符串。</returns>
public string GetMetadataJson()
{
// 创建一个匿名对象，该对象仅包含需要被序列化的元数据字段。
// 这比直接序列化整个对象更安全、更高效。
var metadata = new
{
CameraId,
CaptureTime,
DispatchTime,
OriginalWidth,
OriginalHeight,
TargetWidth,
TargetHeight,
SubscriberIds,
// 附加一个标志，指示此载荷中是否包含目标图像数据，以便接收端进行判断。
HasTargetImage = (TargetImageBytes != null && TargetImageBytes.Length > 0)
};
return JsonConvert.SerializeObject(metadata);
// 在序列化前自动更新标志位，防止逻辑不同步
this.HasOriginalImage = (OriginalImageBytes != null && OriginalImageBytes.Length > 0);
this.HasTargetImage = (TargetImageBytes != null && TargetImageBytes.Length > 0);
return JsonConvert.SerializeObject(this);
}
/// <summary>
/// 从一个 JSON 字符串反序列化，创建一个新的 <see cref="VideoPayload"/> 对象。
/// 注意：反序列化后，对象中的二进制图像数据（<see cref="OriginalImageBytes"/> 和 <see cref="TargetImageBytes"/>）将为 null，
/// 需要在后续步骤中手动填充。
/// </summary>
/// <param name="json">包含元数据的 JSON 字符串。</param>
/// <returns>一个新的 <see cref="VideoPayload"/> 对象，其元数据已填充。</returns>
public static VideoPayload FromMetadataJson(string json)
{
return JsonConvert.DeserializeObject<VideoPayload>(json);

2
SHH.NetMQ/DistributorServer.cs
View File

@@ -35,7 +35,7 @@ namespace SHH.NetMQ
if (payload == null) return;
// 补充发送时间
payload.DispatchTime = DateTime.Now;
payload.DispatchTimestamp = DateTime.Now.Ticks;
// 准备数据帧
string jsonMeta = payload.GetMetadataJson();

4
SHH.NetMQ/ForwarderClient.cs
View File

@@ -83,8 +83,8 @@ namespace SHH.NetMQ
var metaJson = JsonConvert.SerializeObject(new
{
payload.CameraId,
payload.CaptureTime,
payload.DispatchTime,
payload.CaptureTimestamp,
payload.DispatchTimestamp,
payload.OriginalWidth,
payload.OriginalHeight,
// 如果有订阅者ID列表也带上