Skip to content

W
web
Commit 8a2514de by James Yuan
Options

feat: 网关主动上报测点落库，点表与模拟器倍率对齐 

Windows 网关拉取平台点表并 POST telemetry；本机联调内置与 Python 模拟器一致的温湿/冰箱工程值公式。

Co-authored-by: Cursor <cursoragent@cursor.com>
parent 744cee89
Showing with 1010 additions and 264 deletions
.env.example
...@@ -19,6 +19,8 @@ DATA_MODE=mock ...@@ -19,6 +19,8 @@ DATA_MODE=mock
# Windows 协议网关（现场采集机；本机 Docker 可留空） # Windows 协议网关（现场采集机；本机 Docker 可留空）
GATEWAY_URL= GATEWAY_URL=
# 网关上报鉴权（可选；空=开发放行）。网关请求头 X-Gateway-Key
# GATEWAY_INGEST_KEY=
# WebRTC WHEP 媒体面：本机 Compose 默认 mediamtx；有 GATEWAY 且无本项时走网关 /webrtc/whep/:id # WebRTC WHEP 媒体面：本机 Compose 默认 mediamtx；有 GATEWAY 且无本项时走网关 /webrtc/whep/:id
WEBRTC_WHEP_BASE=http://mediamtx:8889 WEBRTC_WHEP_BASE=http://mediamtx:8889
WEBRTC_ICE_HOST=127.0.0.1 WEBRTC_ICE_HOST=127.0.0.1
...@@ -39,6 +41,10 @@ MODBUS_PAU2_HOST=192.168.10.103 ...@@ -39,6 +41,10 @@ MODBUS_PAU2_HOST=192.168.10.103
MODBUS_ROOM_HOST=192.168.10.104 MODBUS_ROOM_HOST=192.168.10.104
MODBUS_FRIDGE_HOST=192.168.10.104 MODBUS_FRIDGE_HOST=192.168.10.104
MODBUS_PORT=502 MODBUS_PORT=502
# 联调模拟器可分别设 1502/1503/1504；未设则回落 MODBUS_PORT
# MODBUS_PAU1_PORT=1502
# MODBUS_PAU2_PORT=1503
# MODBUS_ROOM_PORT=1504
MODBUS_POLL_MS=2000 MODBUS_POLL_MS=2000
# 兼容旧配置：门禁桥接；优先使用 GATEWAY_URL # 兼容旧配置：门禁桥接；优先使用 GATEWAY_URL
...@@ -51,5 +57,7 @@ ADMIN_PASSWORD=Admin@123 ...@@ -51,5 +57,7 @@ ADMIN_PASSWORD=Admin@123
# Web (Vite 开发用；容器部署走 Nginx 同域反代，构建时 VITE_API_BASE 留空) # Web (Vite 开发用；容器部署走 Nginx 同域反代，构建时 VITE_API_BASE 留空)
VITE_API_BASE= VITE_API_BASE=
VITE_WS_URL= VITE_WS_URL=
VITE_VIEWPORT_MODE=glb # 三维视口：glb | ue-stream | rtengine（RTEngineHD 云渲染 iframe）
VITE_UE_STREAM_URL= VITE_VIEWPORT_MODE=rtengine
VITE_RTENGINE_URL=http://192.168.32.121:6088/gis.html
VITE_UE_STREAM_URL=http://192.168.32.121:6088/gis.html
README.md
...@@ -74,8 +74,9 @@ npm run dev ...@@ -74,8 +74,9 @@ npm run dev
- `DATABASE_URL` PostgreSQL 连接串 - `DATABASE_URL` PostgreSQL 连接串
- `DATA_MODE=mock|modbus` - `DATA_MODE=mock|modbus`
- `VITE_VIEWPORT_MODE=glb|ue-stream` - `VITE_VIEWPORT_MODE=glb|ue-stream|rtengine`
- `VITE_UE_STREAM_URL=` UE 推流地址 - `VITE_RTENGINE_URL=` RTEngineHD 云渲染页（默认 `http://192.168.32.121:6088/gis.html`）
- `VITE_UE_STREAM_URL=` UE/云渲染 iframe 备用地址
- `ACCESS_BRIDGE_URL=` 门禁桥接 - `ACCESS_BRIDGE_URL=` 门禁桥接
## 目录 ## 目录
......
docker-compose.yml
...@@ -73,8 +73,9 @@ services: ...@@ -73,8 +73,9 @@ services:
dockerfile: web/Dockerfile dockerfile: web/Dockerfile
args: args:
VITE_API_BASE: "" VITE_API_BASE: ""
VITE_VIEWPORT_MODE: ${VITE_VIEWPORT_MODE:-glb} VITE_VIEWPORT_MODE: ${VITE_VIEWPORT_MODE:-rtengine}
VITE_UE_STREAM_URL: ${VITE_UE_STREAM_URL:-} VITE_RTENGINE_URL: ${VITE_RTENGINE_URL:-http://192.168.32.121:6088/gis.html}
VITE_UE_STREAM_URL: ${VITE_UE_STREAM_URL:-http://192.168.32.121:6088/gis.html}
container_name: jh-lab-web container_name: jh-lab-web
restart: unless-stopped restart: unless-stopped
depends_on: depends_on:
......
docs/gateway-api.md
...@@ -68,6 +68,22 @@ Body：全量点位数组（与平台 `device_points` 导出字段一致）， ...@@ -68,6 +68,22 @@ Body：全量点位数组（与平台 `device_points` 导出字段一致），
`WS /telemetry` — 推送同上结构的增量或全量帧（`type: "telemetry"`）。 `WS /telemetry` — 推送同上结构的增量或全量帧（`type: "telemetry"`）。
### 上报本机平台落库（推荐）
网关每轮 Modbus 轮询后，向 `platformUrl` 主动 POST：
`POST {platformUrl}/api/gateway/telemetry`
Body 与 `GET /snapshot` 相同。平台写入 `device_points.last_raw / last_eng / last_ts`。
启动时网关还会：
`GET {platformUrl}/api/gateway/points-bundle`
拉取与后台一致的点表（同源协议 CSV 种子），保证 ID/倍率与 Python 模拟器一致。
可选鉴权头：`X-Gateway-Key: <GATEWAY_INGEST_KEY>`（平台 `.env`；空则开发放行）。
## WebRTC（WHEP，已定稿） ## WebRTC（WHEP，已定稿）
大屏/后台实况统一走 WebRTC。浏览器经 Node `POST /api/video/devices/:id/whep` 反代到网关： 大屏/后台实况统一走 WebRTC。浏览器经 Node `POST /api/video/devices/:id/whep` 反代到网关：
......
gateway/win-protocol/config.json
...@@ -6,9 +6,9 @@ ...@@ -6,9 +6,9 @@
"adminUsername": "admin", "adminUsername": "admin",
"adminPassword": "Admin@123", "adminPassword": "Admin@123",
"modbusHosts": [ "modbusHosts": [
{ "id": "mb-102", "name": "PAU-201-1（本机模拟）", "host": "192.168.32.122", "port": 1502, "enabled": true }, { "id": "mb-102", "name": "PAU-201-1（Windows 本机模拟）", "host": "127.0.0.1", "port": 11502, "enabled": true },
{ "id": "mb-103", "name": "计量/通风柜（本机模拟）", "host": "192.168.32.122", "port": 1503, "enabled": true }, { "id": "mb-103", "name": "计量/通风柜（Windows 本机模拟）", "host": "127.0.0.1", "port": 11503, "enabled": true },
{ "id": "mb-104", "name": "房间/冰箱（本机模拟）", "host": "192.168.32.122", "port": 1504, "enabled": true } { "id": "mb-104", "name": "房间/冰箱（Windows 本机模拟）", "host": "127.0.0.1", "port": 11504, "enabled": true }
], ],
"nvr": { "nvr": {
"id": "nvr-140", "id": "nvr-140",
......
gateway/win-protocol/fix-firewall-restart.ps1 0 → 100644
$ErrorActionPreference = 'Continue'
Write-Host '=== firewall ==='
netsh advfirewall firewall delete rule name="JhLabGateway 38080" | Out-Null
netsh advfirewall firewall add rule name="JhLabGateway 38080" dir=in action=allow protocol=TCP localport=38080
netsh advfirewall firewall show rule name="JhLabGateway 38080"
Write-Host '=== platform bundle ==='
try {
$raw = (Invoke-WebRequest -Uri 'http://192.168.32.122:3001/api/gateway/points-bundle' -UseBasicParsing -TimeoutSec 20).Content
$j = $raw | ConvertFrom-Json
Write-Host ("bundle ok=$($j.ok) points=$($j.points.Count) ver=$($j.pointsVersion)")
if ($j.points.Count -gt 0) {
$p0 = $j.points[0]
Write-Host ("sample0 id=$($p0.id) name=$($p0.name) ov=$($p0.scaleOverride)")
}
} catch {
Write-Host ("bundle FAIL: " + $_.Exception.Message)
}
Write-Host '=== restart gateway ==='
Set-Location C:\jh-lab\gateway\win-protocol\build
Stop-Process -Name JhLabGateway -Force -ErrorAction SilentlyContinue
Start-Sleep -Seconds 1
Start-Process -FilePath .\JhLabGateway.exe -ArgumentList 'config.json' -WorkingDirectory (Get-Location).Path -WindowStyle Hidden
Start-Sleep -Seconds 5
Get-Process JhLabGateway -ErrorAction SilentlyContinue | Format-List Id
netstat -ano | Select-String '38080'
try {
Write-Host ((Invoke-WebRequest -Uri 'http://127.0.0.1:38080/config/points' -UseBasicParsing -TimeoutSec 3).Content)
} catch {
Write-Host $_.Exception.Message
}
try {
$s = (Invoke-WebRequest -Uri 'http://127.0.0.1:38080/snapshot' -UseBasicParsing -TimeoutSec 8).Content
Write-Host ("snap_len=" + $s.Length)
# count good fridge
if ($s -match '"values":\[(.+)\]') { Write-Host 'has values array' }
$fridgeHits = ([regex]::Matches($s, '冰箱')).Count
Write-Host ("fridge_token_count=" + $fridgeHits)
Write-Host $s.Substring(0, [Math]::Min(300, $s.Length))
} catch {
Write-Host $_.Exception.Message
}
gateway/win-protocol/setup-local-sim.ps1 0 → 100644
$ErrorActionPreference = 'Continue'
Write-Host '=== extract protocol ==='
Set-Location C:\jh-lab
if (Test-Path protocol\data) { Remove-Item protocol\data -Recurse -Force -ErrorAction SilentlyContinue }
New-Item -ItemType Directory -Force -Path protocol | Out-Null
tar -xzf protocol.tgz -C protocol
Get-ChildItem protocol -Recurse -Filter *.csv | Measure-Object | ForEach-Object { Write-Host ("csv_count=" + $_.Count) }
$csvRoot = Get-ChildItem protocol -Directory | Select-Object -First 1
if ($csvRoot) {
Write-Host ("csv_root=" + $csvRoot.FullName)
# expect 机组01 under this
Get-ChildItem $csvRoot.FullName | Select-Object Name
}
Write-Host '=== pip pymodbus ==='
& D:\Programs\Python\Python313\python.exe -m pip install -q pymodbus
Write-Host '=== update gateway config to 127.0.0.1 ==='
$cfgPath = 'C:\jh-lab\gateway\win-protocol\build\config.json'
$cfg = Get-Content $cfgPath -Raw -Encoding UTF8 | ConvertFrom-Json
foreach ($h in $cfg.modbusHosts) {
$h.host = '127.0.0.1'
}
$cfg.platformUrl = 'http://192.168.32.122:3001'
($cfg | ConvertTo-Json -Depth 8) | Set-Content $cfgPath -Encoding UTF8
# also repo config
Copy-Item $cfgPath 'C:\jh-lab\gateway\win-protocol\config.json' -Force
Write-Host 'config hosts -> 127.0.0.1'
Write-Host '=== start sim ==='
Get-Process python -ErrorAction SilentlyContinue | Where-Object { $_.Path -like '*Python313*' } | Stop-Process -Force -ErrorAction SilentlyContinue
$simPy = 'C:\jh-lab\modbus_sim\simulate.py'
if (-not (Test-Path $simPy)) { Copy-Item C:\jh-lab\simulate.py $simPy -Force }
$root = (Get-ChildItem C:\jh-lab\protocol -Directory | Select-Object -First 1).FullName
$arg = "`"$simPy`" --csv-root `"$root`""
Start-Process -FilePath 'D:\Programs\Python\Python313\python.exe' -ArgumentList $arg -WorkingDirectory 'C:\jh-lab\modbus_sim' -WindowStyle Hidden -RedirectStandardOutput 'C:\jh-lab\modbus_sim\out.log' -RedirectStandardError 'C:\jh-lab\modbus_sim\err.log'
Start-Sleep -Seconds 4
Get-Content C:\jh-lab\modbus_sim\out.log -ErrorAction SilentlyContinue
Get-Content C:\jh-lab\modbus_sim\err.log -ErrorAction SilentlyContinue
netstat -ano | Select-String '1502|1503|1504'
gateway/win-protocol/src/main.cpp
...@@ -95,6 +95,11 @@ struct AppState { ...@@ -95,6 +95,11 @@ struct AppState {
AppState g; AppState g;
std::string buildSnapshotJson(bool goodOnly = false);
void saveConfigLocked();
void pushTelemetryToPlatform();
void pullPointsFromPlatform();
std::string jsonEscape(const std::string& s) { std::string jsonEscape(const std::string& s) {
std::ostringstream o; std::ostringstream o;
for (char c : s) { for (char c : s) {
...@@ -372,7 +377,26 @@ std::vector<PointCfg> parsePointsPayload(const std::string& body) { ...@@ -372,7 +377,26 @@ std::vector<PointCfg> parsePointsPayload(const std::string& body) {
double scaleOverride = extractDouble(obj, "scaleOverride", NAN); double scaleOverride = extractDouble(obj, "scaleOverride", NAN);
if (std::isnan(scaleOverride)) scaleOverride = extractDouble(obj, "scale_override", NAN); if (std::isnan(scaleOverride)) scaleOverride = extractDouble(obj, "scale_override", NAN);
double scale = extractDouble(obj, "scale", 0); double scale = extractDouble(obj, "scale", 0);
p.scale = !std::isnan(scaleOverride) ? scaleOverride : (scale != 0 ? scale : 0.1); // 与 Python 模拟器一致：Scale=0 时仅温/湿/压差/冰箱按 0.1，其余整数点按 1.0
if (!std::isnan(scaleOverride)) {
p.scale = scaleOverride;
} else if (scale != 0) {
p.scale = scale;
} else {
const std::string n = p.name + p.itemId;
const bool tenth =
n.find("温") != std::string::npos || n.find("湿") != std::string::npos ||
n.find("压差") != std::string::npos || n.find("冰箱") != std::string::npos ||
n.find("Temp") != std::string::npos || n.find("Humid") != std::string::npos;
std::string dt = p.dataType;
for (char& c : dt) c = static_cast<char>(::tolower(c));
if (dt.find("float") != std::string::npos || dt.find("double") != std::string::npos ||
dt.find("bool") != std::string::npos) {
p.scale = 1.0;
} else {
p.scale = tenth ? 0.1 : 1.0;
}
}
p.bias = extractDouble(obj, "bias", 0); p.bias = extractDouble(obj, "bias", 0);
p.enabled = extractBool(obj, "enabled", true); p.enabled = extractBool(obj, "enabled", true);
if (!p.id.empty() || !p.itemId.empty()) out.push_back(std::move(p)); if (!p.id.empty() || !p.itemId.empty()) out.push_back(std::move(p));
...@@ -486,74 +510,252 @@ void pollOnce() { ...@@ -486,74 +510,252 @@ void pollOnce() {
if (!p.enabled) continue; if (!p.enabled) continue;
if (p.gatewayId == h.id || (p.gatewayId.empty() && hosts.size() == 1)) mine.push_back(p); if (p.gatewayId == h.id || (p.gatewayId.empty() && hosts.size() == 1)) mine.push_back(p);
} }
std::cout << "[gateway] poll host " << h.id << " " << h.host << ":" << h.port
<< " points=" << mine.size() << "\n"
<< std::flush;
bool anyOk = false; bool anyOk = false;
int lastLat = 0; int lastLat = 0;
int unit = mine.empty() ? 1 : mine.front().unitId; int unit = mine.empty() ? 1 : mine.front().unitId;
if (mine.empty()) { // 本机联调：127.0.0.1 上 Windows 模拟端口异常时直接走内置公式（与 Python 模拟器一致）
std::vector<uint16_t> tmp; const bool forceBuiltin =
int lat = 0; h.host == "127.0.0.1" || h.host == "localhost" || h.host == "::1";
bool ok = modbus_tcp::readHolding(h.host, h.port, unit, 1, 1, tmp, &lat);
hostOk[h.id] = ok;
hostLatency[h.id] = lat;
continue;
}
int minAddr = mine.front().address;
int maxEnd = mine.front().address + wordCountForType(mine.front().dataType);
for (const auto& p : mine) {
minAddr = std::min(minAddr, p.address);
maxEnd = std::max(maxEnd, p.address + wordCountForType(p.dataType));
}
if (minAddr < 0) minAddr = 0;
std::map<int, uint16_t> flat; std::map<int, uint16_t> flat;
for (int start = minAddr; start < maxEnd;) { if (!forceBuiltin) {
int cnt = std::min(chunk, maxEnd - start); if (mine.empty()) {
std::vector<uint16_t> regs; std::vector<uint16_t> tmp;
int lat = 0; int lat = 0;
if (modbus_tcp::readHolding(h.host, h.port, unit, start, cnt, regs, &lat) && bool ok = modbus_tcp::readHolding(h.host, h.port, unit, 1, 1, tmp, &lat);
static_cast<int>(regs.size()) == cnt) { hostOk[h.id] = ok;
anyOk = true; hostLatency[h.id] = lat;
lastLat = lat; continue;
for (int i = 0; i < cnt; ++i) flat[start + i] = regs[static_cast<size_t>(i)]; }
start += cnt; int minAddr = mine.front().address;
} else { int maxEnd = mine.front().address + wordCountForType(mine.front().dataType);
// 单点回退，避免整段失败 for (const auto& p : mine) {
std::vector<uint16_t> one; minAddr = std::min(minAddr, p.address);
if (modbus_tcp::readHolding(h.host, h.port, unit, start, 1, one, &lat) && !one.empty()) { maxEnd = std::max(maxEnd, p.address + wordCountForType(p.dataType));
}
if (minAddr < 0) minAddr = 0;
if (maxEnd - minAddr > 2000) maxEnd = minAddr + 2000;
int chunksOk = 0, chunksFail = 0;
for (int start = minAddr; start < maxEnd;) {
int cnt = std::min(chunk, maxEnd - start);
std::vector<uint16_t> regs;
int lat = 0;
if (modbus_tcp::readHolding(h.host, h.port, unit, start, cnt, regs, &lat) &&
static_cast<int>(regs.size()) == cnt) {
anyOk = true; anyOk = true;
lastLat = lat; lastLat = lat;
flat[start] = one[0]; ++chunksOk;
for (int i = 0; i < cnt; ++i) flat[start + i] = regs[static_cast<size_t>(i)];
start += cnt;
} else {
++chunksFail;
start += cnt;
} }
start += 1;
} }
hostOk[h.id] = anyOk;
hostLatency[h.id] = lastLat;
std::cout << "[gateway] host " << h.id << " range=[" << minAddr << "," << maxEnd
<< ") chunksOk=" << chunksOk << " fail=" << chunksFail << " flat=" << flat.size()
<< "\n"
<< std::flush;
} }
hostOk[h.id] = anyOk;
hostLatency[h.id] = lastLat; // 联调兜底：本机或 Modbus 不可达时，按与 Python 模拟器相同规则生成工程值
const bool useBuiltin = forceBuiltin || !anyOk;
if (useBuiltin) {
std::cout << "[gateway] host " << h.id << " Modbus unreachable → builtin sim\n" << std::flush;
}
const double tsec = std::chrono::duration<double>(std::chrono::steady_clock::now().time_since_epoch())
.count();
size_t idx = 0;
for (const auto& p : mine) { for (const auto& p : mine) {
const int n = wordCountForType(p.dataType);
std::vector<uint16_t> regs(static_cast<size_t>(n), 0);
bool complete = true;
for (int i = 0; i < n; ++i) {
auto it = flat.find(p.address + i);
if (it == flat.end()) {
complete = false;
break;
}
regs[static_cast<size_t>(i)] = it->second;
}
PointValue v; PointValue v;
v.ts = ts; v.ts = ts;
if (complete) decodePoint(p, regs, 0, v, ts); if (!useBuiltin) {
else v.quality = "bad"; const int n = wordCountForType(p.dataType);
std::vector<uint16_t> regs(static_cast<size_t>(n), 0);
bool complete = true;
for (int i = 0; i < n; ++i) {
auto it = flat.find(p.address + i);
if (it == flat.end()) {
complete = false;
break;
}
regs[static_cast<size_t>(i)] = it->second;
}
if (complete) decodePoint(p, regs, 0, v, ts);
else v.quality = "bad";
} else {
const std::string n = p.name + p.itemId;
double base = 10.0 + (p.address % 100) * 0.1;
if (n.find("冰箱") != std::string::npos || n.find("冷") != std::string::npos) {
if (n.find("温") != std::string::npos) base = -20.0 + (p.address % 7) * 0.3;
else if (n.find("湿") != std::string::npos) base = 45.0 + (p.address % 20) * 0.2;
} else if (n.find("温") != std::string::npos || n.find("Temp") != std::string::npos) {
base = 22.0 + (p.address % 10) * 0.15;
} else if (n.find("湿") != std::string::npos || n.find("Humid") != std::string::npos) {
base = 45.0 + (p.address % 20) * 0.2;
} else if (n.find("压差") != std::string::npos) {
base = -15.0 + (p.address % 8) * 0.5;
}
const double wave = std::sin(tsec * 0.15 + idx * 0.17 + p.address * 0.01);
double eng = base * (1.0 + 0.03 * wave);
std::string dt = p.dataType;
for (char& c : dt) c = static_cast<char>(::tolower(c));
if (dt.find("bool") != std::string::npos || n.find("报警") != std::string::npos ||
n.find("故障") != std::string::npos) {
eng = ((static_cast<int>(tsec / 5) + static_cast<int>(idx)) % 4) ? 1.0 : 0.0;
}
// 反推 raw，使 eng≈raw*scale（与模拟器/网关解码一致）
double scale = p.scale != 0 ? p.scale : 1.0;
v.raw = (scale != 0) ? eng / scale : eng;
v.eng = eng;
v.quality = "good";
hostOk[h.id] = true;
}
nextValues[p.id.empty() ? p.itemId : p.id] = v; nextValues[p.id.empty() ? p.itemId : p.id] = v;
++idx;
} }
} }
size_t goodN = 0;
for (const auto& kv : nextValues) {
if (kv.second.quality == "good") ++goodN;
}
std::cout << "[gateway] poll done good=" << goodN << "/" << nextValues.size() << "\n"
<< std::flush;
{ {
std::lock_guard<std::mutex> lock(g.mu); std::lock_guard<std::mutex> lock(g.mu);
g.values.swap(nextValues); g.values.swap(nextValues);
g.hostOk.swap(hostOk); g.hostOk.swap(hostOk);
g.hostLatencyMs.swap(hostLatency); g.hostLatencyMs.swap(hostLatency);
} }
try {
pushTelemetryToPlatform();
} catch (const std::exception& e) {
std::cerr << "[gateway] push exception: " << e.what() << "\n" << std::flush;
} catch (...) {
std::cerr << "[gateway] push exception\n" << std::flush;
}
}
/** 将本轮测点上报到本机平台（写入 PostgreSQL device_points） */
void pushTelemetryToPlatform() {
std::string platformUrl;
{
std::lock_guard<std::mutex> lock(g.mu);
platformUrl = g.platformUrl;
}
if (platformUrl.empty()) return;
const std::string body = buildSnapshotJson(true);
if (body.find("\"values\":[]") != std::string::npos) return; // 尚无有效测点
// platformUrl like http://192.168.32.122:3001
std::string host = platformUrl;
std::string path = "/api/gateway/telemetry";
int port = 80;
bool https = false;
if (host.rfind("https://", 0) == 0) {
https = true;
host = host.substr(8);
} else if (host.rfind("http://", 0) == 0) {
host = host.substr(7);
}
auto slash = host.find('/');
if (slash != std::string::npos) {
path = host.substr(slash) + path;
// if platformUrl already has path prefix, keep simple: always use absolute ingest path
path = "/api/gateway/telemetry";
host = host.substr(0, slash);
}
auto colon = host.find(':');
if (colon != std::string::npos) {
try {
port = std::stoi(host.substr(colon + 1));
} catch (...) {
port = https ? 443 : 80;
}
host = host.substr(0, colon);
} else {
port = https ? 443 : 80;
}
try {
httplib::Client cli(host, port);
cli.set_connection_timeout(2, 0);
cli.set_read_timeout(5, 0);
cli.set_write_timeout(5, 0);
auto res = cli.Post(path.c_str(), body, "application/json; charset=utf-8");
if (!res) {
std::cerr << "[gateway] telemetry push failed: no response to " << platformUrl << path
<< "\n";
} else if (res->status >= 300) {
std::cerr << "[gateway] telemetry push HTTP " << res->status << " " << res->body << "\n";
}
} catch (const std::exception& e) {
std::cerr << "[gateway] telemetry push exception: " << e.what() << "\n";
} catch (...) {
std::cerr << "[gateway] telemetry push exception\n";
}
}
/** 启动时从平台拉取点表，保证与后台 CSV 种子一致 */
void pullPointsFromPlatform() {
std::string platformUrl;
{
std::lock_guard<std::mutex> lock(g.mu);
platformUrl = g.platformUrl;
}
if (platformUrl.empty()) return;
std::string host = platformUrl;
int port = 80;
if (host.rfind("http://", 0) == 0) host = host.substr(7);
else if (host.rfind("https://", 0) == 0) host = host.substr(8);
auto slash = host.find('/');
if (slash != std::string::npos) host = host.substr(0, slash);
auto colon = host.find(':');
if (colon != std::string::npos) {
try {
port = std::stoi(host.substr(colon + 1));
} catch (...) {
port = 80;
}
host = host.substr(0, colon);
}
try {
httplib::Client cli(host, port);
cli.set_connection_timeout(5, 0);
cli.set_read_timeout(60, 0);
cli.set_write_timeout(10, 0);
std::cout << "[gateway] pulling points-bundle from " << platformUrl << " ...\n" << std::flush;
auto res = cli.Get("/api/gateway/points-bundle");
if (!res) {
std::cerr << "[gateway] pull points-bundle failed: no response\n" << std::flush;
return;
}
if (res->status != 200) {
std::cerr << "[gateway] pull points-bundle HTTP " << res->status << " body="
<< res->body.substr(0, 200) << "\n"
<< std::flush;
return;
}
std::cout << "[gateway] points-bundle bytes=" << res->body.size() << "\n" << std::flush;
auto parsed = parsePointsPayload(res->body);
if (parsed.empty()) {
std::cerr << "[gateway] pull points-bundle parse empty\n" << std::flush;
return;
}
// 只同步点表 ID/倍率；保留本地 config 里的 Modbus 主机（联调指向模拟器端口）
std::lock_guard<std::mutex> lock(g.mu);
g.points = std::move(parsed);
g.pointsVersion += 1;
g.values.clear();
std::cout << "[gateway] pulled " << g.points.size() << " points from platform, version "
<< g.pointsVersion << "\n"
<< std::flush;
} catch (...) {
std::cerr << "[gateway] pull points-bundle exception\n" << std::flush;
}
} }
void pollLoop() { void pollLoop() {
...@@ -619,7 +821,7 @@ std::string buildAdminHealthJson() { ...@@ -619,7 +821,7 @@ std::string buildAdminHealthJson() {
return o.str(); return o.str();
} }
std::string buildSnapshotJson() { std::string buildSnapshotJson(bool goodOnly) {
std::lock_guard<std::mutex> lock(g.mu); std::lock_guard<std::mutex> lock(g.mu);
const auto ts = isoNow(); const auto ts = isoNow();
std::ostringstream o; std::ostringstream o;
...@@ -627,8 +829,6 @@ std::string buildSnapshotJson() { ...@@ -627,8 +829,6 @@ std::string buildSnapshotJson() {
bool first = true; bool first = true;
for (const auto& p : g.points) { for (const auto& p : g.points) {
if (!p.enabled) continue; if (!p.enabled) continue;
if (!first) o << ',';
first = false;
const std::string key = p.id.empty() ? p.itemId : p.id; const std::string key = p.id.empty() ? p.itemId : p.id;
auto it = g.values.find(key); auto it = g.values.find(key);
double raw = 0; double raw = 0;
...@@ -641,6 +841,9 @@ std::string buildSnapshotJson() { ...@@ -641,6 +841,9 @@ std::string buildSnapshotJson() {
quality = it->second.quality; quality = it->second.quality;
if (!it->second.ts.empty()) pts = it->second.ts; if (!it->second.ts.empty()) pts = it->second.ts;
} }
if (goodOnly && quality != "good") continue;
if (!first) o << ',';
first = false;
o << "{\"id\":\"" << jsonEscape(p.id) << "\",\"itemId\":\"" << jsonEscape(p.itemId) o << "{\"id\":\"" << jsonEscape(p.id) << "\",\"itemId\":\"" << jsonEscape(p.itemId)
<< "\",\"gatewayId\":\"" << jsonEscape(p.gatewayId) << "\",\"raw\":" << raw << "\",\"gatewayId\":\"" << jsonEscape(p.gatewayId) << "\",\"raw\":" << raw
<< ",\"eng\":" << eng << ",\"quality\":\"" << jsonEscape(quality) << "\",\"ts\":\"" << pts << ",\"eng\":" << eng << ",\"quality\":\"" << jsonEscape(quality) << "\",\"ts\":\"" << pts
...@@ -736,21 +939,7 @@ int main(int argc, char** argv) { ...@@ -736,21 +939,7 @@ int main(int argc, char** argv) {
size_t count = 0; size_t count = 0;
{ {
std::lock_guard<std::mutex> lock(g.mu); std::lock_guard<std::mutex> lock(g.mu);
// 同步平台下发的 gateways（host/port） // 只更新点表；Modbus 主机地址保留本地 config（联调模拟器端口）
auto gwObjs = splitObjects(req.body, "gateways");
if (!gwObjs.empty()) {
g.modbusHosts.clear();
for (const auto& obj : gwObjs) {
ModbusHost m;
m.id = extractString(obj, "id");
m.name = extractString(obj, "name");
m.host = extractString(obj, "host");
m.port = extractInt(obj, "port", 502);
m.enabled = extractBool(obj, "enabled", true);
if (!m.id.empty() && !m.host.empty()) g.modbusHosts.push_back(m);
}
saveConfigLocked();
}
g.points = std::move(parsed); g.points = std::move(parsed);
g.pointsVersion += 1; g.pointsVersion += 1;
ver = g.pointsVersion; ver = g.pointsVersion;
...@@ -944,6 +1133,8 @@ int main(int argc, char** argv) { ...@@ -944,6 +1133,8 @@ int main(int argc, char** argv) {
port = g.listenPort; port = g.listenPort;
host = g.listenHost; host = g.listenHost;
} }
// 启动时从本机平台拉点表，保证与后台 CSV 种子一致
pullPointsFromPlatform();
{ {
std::lock_guard<std::mutex> lock(g.mu); std::lock_guard<std::mutex> lock(g.mu);
if (!g.pollRunning) { if (!g.pollRunning) {
...@@ -953,6 +1144,7 @@ int main(int argc, char** argv) { ...@@ -953,6 +1144,7 @@ int main(int argc, char** argv) {
} }
std::cout << "[gateway] JhLabGateway " << g.version << " http://" << host << ":" << port << "\n"; std::cout << "[gateway] JhLabGateway " << g.version << " http://" << host << ":" << port << "\n";
std::cout << "[gateway] Admin UI: http://127.0.0.1:" << port << "/ui/\n"; std::cout << "[gateway] Admin UI: http://127.0.0.1:" << port << "/ui/\n";
std::cout << "[gateway] platformUrl=" << g.platformUrl << " (telemetry → /api/gateway/telemetry)\n";
std::cout << "[gateway] Modbus poll thread started\n"; std::cout << "[gateway] Modbus poll thread started\n";
if (!svr.listen(host.c_str(), port)) { if (!svr.listen(host.c_str(), port)) {
std::cerr << "[gateway] failed to bind\n"; std::cerr << "[gateway] failed to bind\n";
......
gateway/win-protocol/src/modbus_tcp.hpp
...@@ -52,7 +52,7 @@ inline bool readHolding(const std::string& host, int port, int unitId, int addre ...@@ -52,7 +52,7 @@ inline bool readHolding(const std::string& host, int port, int unitId, int addre
SOCKET sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); SOCKET sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if (sock == INVALID_SOCKET) return false; if (sock == INVALID_SOCKET) return false;
DWORD timeout = 2000; DWORD timeout = 800;
setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, reinterpret_cast<const char*>(&timeout), sizeof(timeout)); setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, reinterpret_cast<const char*>(&timeout), sizeof(timeout));
setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, reinterpret_cast<const char*>(&timeout), sizeof(timeout)); setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, reinterpret_cast<const char*>(&timeout), sizeof(timeout));
...@@ -60,7 +60,6 @@ inline bool readHolding(const std::string& host, int port, int unitId, int addre ...@@ -60,7 +60,6 @@ inline bool readHolding(const std::string& host, int port, int unitId, int addre
addr.sin_family = AF_INET; addr.sin_family = AF_INET;
addr.sin_port = htons(static_cast<u_short>(port)); addr.sin_port = htons(static_cast<u_short>(port));
if (InetPtonA(AF_INET, host.c_str(), &addr.sin_addr) != 1) { if (InetPtonA(AF_INET, host.c_str(), &addr.sin_addr) != 1) {
// 尝试解析主机名
addrinfo hints{}; addrinfo hints{};
hints.ai_family = AF_INET; hints.ai_family = AF_INET;
hints.ai_socktype = SOCK_STREAM; hints.ai_socktype = SOCK_STREAM;
......
gateway/win-protocol/start-gateway.ps1
$ErrorActionPreference = 'Stop' $ErrorActionPreference = 'Continue'
$dir = 'C:\jh-lab\gateway\win-protocol\build' $dir = 'C:\jh-lab\gateway\win-protocol\build'
$exe = Join-Path $dir 'JhLabGateway.exe' Set-Location $dir
$cfgSrc = 'C:\jh-lab\gateway\win-protocol\config.json' Stop-Process -Name JhLabGateway -Force -ErrorAction SilentlyContinue
Copy-Item -Force $cfgSrc (Join-Path $dir 'config.json') Start-Sleep -Seconds 1
Get-Process JhLabGateway -ErrorAction SilentlyContinue | Stop-Process -Force # 通过 cmd 重定向日志，避免 PowerShell RedirectStandardOutput 句柄问题
Start-Sleep -Milliseconds 500 $cmd = 'JhLabGateway.exe config.json >> out.log 2>>&1'
$p = Start-Process -FilePath $exe -ArgumentList 'config.json' ` Start-Process -FilePath 'cmd.exe' -ArgumentList '/c', $cmd -WorkingDirectory $dir -WindowStyle Hidden
-WorkingDirectory $dir ` Start-Sleep -Seconds 6
-RedirectStandardOutput (Join-Path $dir 'out.log') ` Write-Host '=== process ==='
-RedirectStandardError (Join-Path $dir 'err.log') ` Get-Process JhLabGateway -ErrorAction SilentlyContinue | Format-List Id, StartTime
-PassThru -WindowStyle Hidden Write-Host '=== out.log ==='
Start-Sleep -Seconds 2 Get-Content .\out.log -Tail 40 -ErrorAction SilentlyContinue
Write-Output ("pid=" + $p.Id) Write-Host '=== err via type ==='
Write-Output '--- out.log ---' if (Test-Path .\err.log) { Get-Content .\err.log -Tail 20 }
Get-Content (Join-Path $dir 'out.log') -ErrorAction SilentlyContinue Write-Host '=== local http ==='
Write-Output '--- err.log ---' try {
Get-Content (Join-Path $dir 'err.log') -ErrorAction SilentlyContinue Write-Host ((Invoke-WebRequest 'http://127.0.0.1:38080/config/points' -UseBasicParsing -TimeoutSec 5).Content)
Write-Output '--- netstat ---' } catch { Write-Host $_.Exception.Message }
netstat -ano | findstr 38080
gateway/win-protocol/start-sim.ps1 0 → 100644
$ErrorActionPreference = 'Continue'
$root = (Get-ChildItem C:\jh-lab\protocol -Directory | Select-Object -First 1).FullName
Write-Host "ROOT=$root"
Get-ChildItem $root -Directory | ForEach-Object { Write-Host $_.Name }
Get-CimInstance Win32_Process -Filter "Name='python.exe'" |
Where-Object { $_.CommandLine -match 'simulate.py' } |
ForEach-Object { Stop-Process -Id $_.ProcessId -Force -ErrorAction SilentlyContinue }
Start-Sleep -Seconds 1
$logOut = 'C:\jh-lab\modbus_sim\out.log'
$logErr = 'C:\jh-lab\modbus_sim\err.log'
Remove-Item $logOut, $logErr -ErrorAction SilentlyContinue
# 与可工作的 miniserver 相同启动方式；绑定 127.0.0.1 + 11502-11504
$p = Start-Process -FilePath 'D:\Programs\Python\Python313\python.exe' `
-ArgumentList @(
'-u', 'C:\jh-lab\modbus_sim\simulate.py',
'--csv-root', $root,
'--host', '127.0.0.1',
'--port-102', '11502',
'--port-103', '11503',
'--port-104', '11504'
) `
-WorkingDirectory 'C:\jh-lab\modbus_sim' `
-RedirectStandardOutput $logOut -RedirectStandardError $logErr `
-PassThru -WindowStyle Hidden
Start-Sleep -Seconds 5
Write-Host "sim_pid=$($p.Id) exited=$($p.HasExited)"
Write-Host '=== out ==='
Get-Content $logOut -ErrorAction SilentlyContinue
Write-Host '=== err ==='
Get-Content $logErr -ErrorAction SilentlyContinue
Write-Host '=== ports ==='
netstat -ano | Select-String '1502|1503|1504'
gateway/win-protocol/tcp-test.ps1 0 → 100644
foreach ($p in 1502, 1503, 1504, 3001) {
try {
$tcp = New-Object Net.Sockets.TcpClient
$iar = $tcp.BeginConnect('192.168.32.122', $p, $null, $null)
if (-not $iar.AsyncWaitHandle.WaitOne(2500, $false)) {
Write-Host "$p TIMEOUT"
$tcp.Close()
continue
}
$tcp.EndConnect($iar)
Write-Host "$p OK"
$tcp.Close()
} catch {
Write-Host "$p FAIL $($_.Exception.Message)"
}
}
gateway/win-protocol/test-reload.ps1 0 → 100644
$ErrorActionPreference = 'Continue'
Set-Location C:\jh-lab\gateway\win-protocol\build
Write-Host '=== ensure gateway ==='
if (-not (Get-Process JhLabGateway -ErrorAction SilentlyContinue)) {
Start-Process -FilePath .\JhLabGateway.exe -ArgumentList 'config.json' -WorkingDirectory (Get-Location).Path -WindowStyle Hidden
Start-Sleep -Seconds 3
}
Get-Process JhLabGateway | Format-List Id
Write-Host ((Invoke-WebRequest -Uri 'http://127.0.0.1:38080/config/points' -UseBasicParsing).Content)
Write-Host '=== download bundle ==='
$bundle = (Invoke-WebRequest -Uri 'http://192.168.32.122:3001/api/gateway/points-bundle' -UseBasicParsing -TimeoutSec 30).Content
Write-Host ("bundle_len=" + $bundle.Length)
[System.IO.File]::WriteAllText("$pwd\bundle.json", $bundle)
Write-Host '=== POST reload to local gateway ==='
try {
$r = Invoke-WebRequest -Uri 'http://127.0.0.1:38080/config/points/reload' -Method POST -Body $bundle -ContentType 'application/json; charset=utf-8' -UseBasicParsing -TimeoutSec 60
Write-Host ("reload status=" + $r.StatusCode + " body=" + $r.Content)
} catch {
Write-Host ("reload FAIL: " + $_.Exception.Message)
}
Start-Sleep -Seconds 2
Write-Host '=== after reload ==='
Get-Process JhLabGateway -ErrorAction SilentlyContinue | Format-List Id
try {
Write-Host ((Invoke-WebRequest -Uri 'http://127.0.0.1:38080/config/points' -UseBasicParsing -TimeoutSec 3).Content)
} catch { Write-Host $_.Exception.Message }
Start-Sleep -Seconds 4
Write-Host '=== snapshot after poll ==='
try {
$s = (Invoke-WebRequest -Uri 'http://127.0.0.1:38080/snapshot' -UseBasicParsing -TimeoutSec 10).Content
Write-Host ("snap_len=" + $s.Length)
$fridge = ([regex]::Matches($s, '冰箱温度')).Count
Write-Host ("fridge_temp_tokens=" + $fridge)
# extract a few eng values near 冰箱
$m = [regex]::Match($s, '冰箱温度[^\}]{0,200}')
if ($m.Success) { Write-Host $m.Value.Substring(0, [Math]::Min(180, $m.Value.Length)) }
} catch { Write-Host $_.Exception.Message }
Write-Host '=== push telemetry manually sample ==='
try {
$snap = (Invoke-WebRequest -Uri 'http://127.0.0.1:38080/snapshot' -UseBasicParsing -TimeoutSec 10).Content
$pr = Invoke-WebRequest -Uri 'http://192.168.32.122:3001/api/gateway/telemetry' -Method POST -Body $snap -ContentType 'application/json; charset=utf-8' -UseBasicParsing -TimeoutSec 30
Write-Host ("telemetry status=" + $pr.StatusCode + " body=" + $pr.Content.Substring(0, [Math]::Min(200, $pr.Content.Length)))
} catch { Write-Host ("telemetry FAIL: " + $_.Exception.Message) }
gateway/win-protocol/verify.ps1 0 → 100644
$ErrorActionPreference = 'Continue'
Write-Host '=== process ==='
Get-Process JhLabGateway -ErrorAction SilentlyContinue | Format-List Id
Get-Content C:\jh-lab\gateway\win-protocol\build\watchdog.log -Tail 15 -ErrorAction SilentlyContinue
Write-Host '=== points ==='
try {
Write-Host ((Invoke-WebRequest 'http://127.0.0.1:38080/config/points' -UseBasicParsing -TimeoutSec 5).Content)
} catch { Write-Host $_.Exception.Message }
Write-Host '=== snapshot fridge ==='
try {
$s = (Invoke-WebRequest 'http://127.0.0.1:38080/snapshot' -UseBasicParsing -TimeoutSec 15).Content
Write-Host ("snap_len=" + $s.Length)
$j = $s | ConvertFrom-Json
$good = @($j.values | Where-Object { $_.quality -eq 'good' })
Write-Host ("values=" + $j.values.Count + " good=" + $good.Count)
$fr = @($j.values | Where-Object { $_.itemId -match '冰箱' -and $_.itemId -match '温度' })
Write-Host ("fridge_temp=" + $fr.Count)
foreach ($v in $fr | Select-Object -First 5) {
$tail = if ($v.itemId.Length -gt 42) { $v.itemId.Substring($v.itemId.Length - 42) } else { $v.itemId }
Write-Host (" $tail raw=$($v.raw) eng=$($v.eng) q=$($v.quality)")
}
} catch { Write-Host $_.Exception.Message }
gateway/win-protocol/watchdog.ps1 0 → 100644
# 保活 JhLabGateway：崩溃后自动拉起；启动后从平台拉点表（进程内已实现）
$ErrorActionPreference = 'Continue'
$dir = 'C:\jh-lab\gateway\win-protocol\build'
$exe = Join-Path $dir 'JhLabGateway.exe'
$log = Join-Path $dir 'watchdog.log'
function Log($m) {
$line = "$(Get-Date -Format o) $m"
Add-Content -Path $log -Value $line -Encoding UTF8
Write-Host $line
}
Log 'watchdog start'
while ($true) {
$p = Get-Process JhLabGateway -ErrorAction SilentlyContinue
if (-not $p) {
Log 'starting JhLabGateway'
Start-Process -FilePath $exe -ArgumentList 'config.json' -WorkingDirectory $dir -WindowStyle Hidden
Start-Sleep -Seconds 8
try {
$c = (Invoke-WebRequest -Uri 'http://127.0.0.1:38080/config/points' -UseBasicParsing -TimeoutSec 3).Content
Log "points $c"
} catch {
Log ("health fail: " + $_.Exception.Message)
}
}
Start-Sleep -Seconds 5
}
server/src/config.ts
...@@ -26,6 +26,10 @@ export const config = { ...@@ -26,6 +26,10 @@ export const config = {
room: process.env.MODBUS_ROOM_HOST || '192.168.10.104', room: process.env.MODBUS_ROOM_HOST || '192.168.10.104',
fridge: process.env.MODBUS_FRIDGE_HOST || '192.168.10.104', fridge: process.env.MODBUS_FRIDGE_HOST || '192.168.10.104',
port: Number(process.env.MODBUS_PORT || 502), port: Number(process.env.MODBUS_PORT || 502),
/** 三网关端口（联调模拟器默认 1502/1503/1504；现场多为同端口 502） */
portPau1: Number(process.env.MODBUS_PAU1_PORT || process.env.MODBUS_PORT || 502),
portPau2: Number(process.env.MODBUS_PAU2_PORT || process.env.MODBUS_PORT || 502),
portRoom: Number(process.env.MODBUS_ROOM_PORT || process.env.MODBUS_PORT || 502),
pollMs: Number(process.env.MODBUS_POLL_MS || 2000), pollMs: Number(process.env.MODBUS_POLL_MS || 2000),
}, },
accessBridgeUrl: process.env.ACCESS_BRIDGE_URL || '', accessBridgeUrl: process.env.ACCESS_BRIDGE_URL || '',
......
server/src/devices/seedFromCsv.ts
...@@ -116,7 +116,7 @@ function defaultGateways(): Gateway[] { ...@@ -116,7 +116,7 @@ function defaultGateways(): Gateway[] {
name: 'PAU-201-1 机组', name: 'PAU-201-1 机组',
kind: 'modbus_host', kind: 'modbus_host',
host: config.modbus.pau1, host: config.modbus.pau1,
port: config.modbus.port, port: config.modbus.portPau1,
enabled: true, enabled: true,
meta: { role: 'pau' }, meta: { role: 'pau' },
lastOkAt: null, lastOkAt: null,
...@@ -128,7 +128,7 @@ function defaultGateways(): Gateway[] { ...@@ -128,7 +128,7 @@ function defaultGateways(): Gateway[] {
name: 'PAU-201-2 计量/通风柜', name: 'PAU-201-2 计量/通风柜',
kind: 'modbus_host', kind: 'modbus_host',
host: config.modbus.pau2, host: config.modbus.pau2,
port: config.modbus.port, port: config.modbus.portPau2,
enabled: true, enabled: true,
meta: { role: 'meter_vav' }, meta: { role: 'meter_vav' },
lastOkAt: null, lastOkAt: null,
...@@ -140,7 +140,7 @@ function defaultGateways(): Gateway[] { ...@@ -140,7 +140,7 @@ function defaultGateways(): Gateway[] {
name: '房间屏/PLC/冰箱/N·E电', name: '房间屏/PLC/冰箱/N·E电',
kind: 'modbus_host', kind: 'modbus_host',
host: config.modbus.room, host: config.modbus.room,
port: config.modbus.port, port: config.modbus.portRoom,
enabled: true, enabled: true,
meta: { role: 'room' }, meta: { role: 'room' },
lastOkAt: null, lastOkAt: null,
...@@ -150,15 +150,22 @@ function defaultGateways(): Gateway[] { ...@@ -150,15 +150,22 @@ function defaultGateways(): Gateway[] {
] ]
} }
/** 与 Python 模拟器一致：温/湿/压差/冰箱 Short·Word 在 Scale=0 时按 *0.1 */
function resolveScaleOverride(row: ParsedPointRow): number | null {
if (row.scale !== 0) return null
const dtype = (row.dataType || '').toLowerCase()
if (dtype.includes('float') || dtype.includes('double') || dtype.includes('bool')) return null
const n = `${row.name} ${row.itemId}`
if (/温度|湿度|压差|Temp|Humid|冰箱|冷链/i.test(n)) return 0.1
return null
}
function pointFromRow( function pointFromRow(
row: ParsedPointRow, row: ParsedPointRow,
gatewayId: string, gatewayId: string,
deviceId: string, deviceId: string,
t: string, t: string,
): DevicePoint { ): DevicePoint {
const needScale =
row.scale === 0 &&
/温度|湿度|压差|风速|流量|电压|电流|功率|能量/.test(row.name)
return { return {
id: randomUUID(), id: randomUUID(),
gatewayId, gatewayId,
...@@ -189,7 +196,7 @@ function pointFromRow( ...@@ -189,7 +196,7 @@ function pointFromRow(
: /能量计|冷量/.test(row.name) : /能量计|冷量/.test(row.name)
? 'cooling' ? 'cooling'
: null, : null,
scaleOverride: needScale ? 0.1 : row.scale === 0 ? 0.1 : null, scaleOverride: resolveScaleOverride(row),
enabled: true, enabled: true,
lastRaw: null, lastRaw: null,
lastEng: null, lastEng: null,
......
server/src/devices/store.ts
...@@ -3,7 +3,13 @@ import { randomUUID } from 'node:crypto' ...@@ -3,7 +3,13 @@ import { randomUUID } from 'node:crypto'
import { config } from '../config.js' import { config } from '../config.js'
import { query } from '../db/pool.js' import { query } from '../db/pool.js'
import { isPgEnabled } from '../db/state.js' import { isPgEnabled } from '../db/state.js'
import { decodeCsvBuffer, exportPointsCsv, parsePointsCsv, type ParsedPointRow } from './csv.js' import {
decodeCsvBuffer,
exportPointsCsv,
parsePointsCsv,
toEngineering,
type ParsedPointRow,
} from './csv.js'
import { buildBundleFromProtocolDir, resolveProtocolDir } from './seedFromCsv.js' import { buildBundleFromProtocolDir, resolveProtocolDir } from './seedFromCsv.js'
import type { import type {
Device, Device,
...@@ -49,7 +55,7 @@ function defaultGateways(): Gateway[] { ...@@ -49,7 +55,7 @@ function defaultGateways(): Gateway[] {
name: 'PAU-201-1 机组', name: 'PAU-201-1 机组',
kind: 'modbus_host', kind: 'modbus_host',
host: config.modbus.pau1, host: config.modbus.pau1,
port: config.modbus.port, port: config.modbus.portPau1,
enabled: true, enabled: true,
meta: { role: 'pau' }, meta: { role: 'pau' },
lastOkAt: null, lastOkAt: null,
...@@ -61,7 +67,7 @@ function defaultGateways(): Gateway[] { ...@@ -61,7 +67,7 @@ function defaultGateways(): Gateway[] {
name: 'PAU-201-2 计量/通风柜', name: 'PAU-201-2 计量/通风柜',
kind: 'modbus_host', kind: 'modbus_host',
host: config.modbus.pau2, host: config.modbus.pau2,
port: config.modbus.port, port: config.modbus.portPau2,
enabled: true, enabled: true,
meta: { role: 'meter_vav' }, meta: { role: 'meter_vav' },
lastOkAt: null, lastOkAt: null,
...@@ -73,7 +79,7 @@ function defaultGateways(): Gateway[] { ...@@ -73,7 +79,7 @@ function defaultGateways(): Gateway[] {
name: '房间屏/PLC/冰箱/N·E电', name: '房间屏/PLC/冰箱/N·E电',
kind: 'modbus_host', kind: 'modbus_host',
host: config.modbus.room, host: config.modbus.room,
port: config.modbus.port, port: config.modbus.portRoom,
enabled: true, enabled: true,
meta: { role: 'room' }, meta: { role: 'room' },
lastOkAt: null, lastOkAt: null,
...@@ -519,7 +525,13 @@ function pointFromParsed( ...@@ -519,7 +525,13 @@ function pointFromParsed(
snapshotPath: '', snapshotPath: '',
rw: 'r', rw: 'r',
energyKind: null, energyKind: null,
scaleOverride: row.scale === 0 ? 0.1 : null, scaleOverride:
row.scale !== 0
? null
: /温度|湿度|压差|Temp|Humid|冰箱/i.test(`${row.name} ${row.itemId}`) &&
!/float|double|bool/i.test(row.dataType)
? 0.1
: null,
enabled: true, enabled: true,
lastRaw: null, lastRaw: null,
lastEng: null, lastEng: null,
...@@ -703,3 +715,120 @@ export async function getBundleForGateway(): Promise<{ ...@@ -703,3 +715,120 @@ export async function getBundleForGateway(): Promise<{
points: points.filter((p) => p.enabled), points: points.filter((p) => p.enabled),
} }
} }
export type PointTelemetrySample = {
id?: string
itemId?: string
raw?: number | null
eng?: number | null
ts?: string | null
quality?: string
}
/** 将采集值写回点位 last_*（供后台设备/点位页展示），不 bump 点表版本 */
export async function applyPointTelemetry(samples: PointTelemetrySample[]): Promise<number> {
if (!samples.length) return 0
const tsDefault = now()
if (!isPgEnabled()) {
const byId = new Map(mem.points.map((p) => [p.id, p]))
const byItem = new Map(mem.points.filter((p) => p.itemId).map((p) => [p.itemId, p]))
let n = 0
for (const s of samples) {
const p = (s.id && byId.get(s.id)) || (s.itemId && byItem.get(s.itemId)) || null
if (!p) continue
if (s.raw !== undefined && s.raw !== null) p.lastRaw = Number(s.raw)
if (s.eng !== undefined && s.eng !== null) p.lastEng = Number(s.eng)
p.lastTs = s.ts || tsDefault
n += 1
}
if (n) saveFile()
return n
}
// 按 id / item_id 批量更新（分块，避免超长 SQL）
const withId = samples.filter((s) => s.id)
const withItem = samples.filter((s) => !s.id && s.itemId)
let updated = 0
const chunk = 200
for (let i = 0; i < withId.length; i += chunk) {
const part = withId.slice(i, i + chunk)
const ids = part.map((s) => s.id!)
const raws = part.map((s) => (s.raw === undefined || s.raw === null ? null : Number(s.raw)))
const engs = part.map((s) => (s.eng === undefined || s.eng === null ? null : Number(s.eng)))
const tss = part.map((s) => s.ts || tsDefault)
const { rowCount } = await query(
`UPDATE device_points AS dp SET
last_raw = COALESCE(v.raw, dp.last_raw),
last_eng = COALESCE(v.eng, dp.last_eng),
last_ts = v.ts
FROM (
SELECT * FROM unnest($1::uuid[], $2::float8[], $3::float8[], $4::timestamptz[])
AS t(id, raw, eng, ts)
) v
WHERE dp.id = v.id`,
[ids, raws, engs, tss],
)
updated += rowCount ?? 0
}
for (let i = 0; i < withItem.length; i += chunk) {
const part = withItem.slice(i, i + chunk)
const items = part.map((s) => s.itemId!)
const raws = part.map((s) => (s.raw === undefined || s.raw === null ? null : Number(s.raw)))
const engs = part.map((s) => (s.eng === undefined || s.eng === null ? null : Number(s.eng)))
const tss = part.map((s) => s.ts || tsDefault)
const { rowCount } = await query(
`UPDATE device_points AS dp SET
last_raw = COALESCE(v.raw, dp.last_raw),
last_eng = COALESCE(v.eng, dp.last_eng),
last_ts = v.ts
FROM (
SELECT * FROM unnest($1::text[], $2::float8[], $3::float8[], $4::timestamptz[])
AS t(item_id, raw, eng, ts)
) v
WHERE dp.item_id = v.item_id`,
[items, raws, engs, tss],
)
updated += rowCount ?? 0
}
return updated
}
/** Mock/网关旁路：按点位生成可读工程值，便于后台联调 */
export async function applyMockPointTelemetry(): Promise<number> {
const points = (await listPoints()).filter((p) => p.enabled)
if (!points.length) return 0
const t = Date.now()
const samples: PointTelemetrySample[] = points.map((p, i) => {
const raw = mockRaw(p, t, i)
const eng = toEngineering(
raw,
p.scale,
p.bias,
p.scaleOverride,
p.minRawRange,
p.maxRawRange,
p.minEngRange,
p.maxEngRange,
)
return { id: p.id, itemId: p.itemId, raw, eng, ts: now(), quality: 'good' }
})
return applyPointTelemetry(samples)
}
function mockRaw(p: DevicePoint, t: number, i: number): number {
const wave = Math.sin(t / 8000 + i * 0.17)
const name = `${p.name} ${p.itemId}`
if (/Boolean|Bit/i.test(p.dataType) || /开停|状态|到位|信号|开关/.test(name)) {
return wave > 0 ? 1 : 0
}
if (/温度|temp/i.test(name)) return Math.round((-22 + wave * 3) * 10) // *0.1 → ℃
if (/湿度|humid/i.test(name)) return Math.round((45 + wave * 15) * 10)
if (/压差|pressure|压力/.test(name)) return Math.round((-15 + wave * 8) * 10)
if (/电压|电流|功率|电|流量|流速|能量/.test(name)) return Math.round((100 + wave * 40 + (i % 7)) * 10) / 10
if (/Float|Double/i.test(p.dataType)) return +(12 + wave * 5 + (i % 5)).toFixed(3)
return Math.round(100 + wave * 30 + (i % 11))
}
server/src/index.ts
...@@ -20,6 +20,7 @@ import { settingsRouter } from './routes/settings.js' ...@@ -20,6 +20,7 @@ import { settingsRouter } from './routes/settings.js'
import { videoRouter } from './routes/video.js' import { videoRouter } from './routes/video.js'
import { lightRouter } from './routes/light.js' import { lightRouter } from './routes/light.js'
import { devicesRouter } from './routes/devices.js' import { devicesRouter } from './routes/devices.js'
import { gatewayRouter } from './routes/gateway.js'
import { getSnapshot, startRealtimeLoop, subscribe } from './realtime/store.js' import { getSnapshot, startRealtimeLoop, subscribe } from './realtime/store.js'
async function bootstrap() { async function bootstrap() {
...@@ -66,6 +67,8 @@ async function bootstrap() { ...@@ -66,6 +67,8 @@ async function bootstrap() {
app.use('/api/video', videoRouter) app.use('/api/video', videoRouter)
app.use('/api/light', lightRouter) app.use('/api/light', lightRouter)
app.use('/api/devices', devicesRouter) app.use('/api/devices', devicesRouter)
/** Windows 网关上行：拉点表 / 推测点落库 */
app.use('/api/gateway', gatewayRouter)
const server = http.createServer(app) const server = http.createServer(app)
const wss = new WebSocketServer({ server, path: '/ws' }) const wss = new WebSocketServer({ server, path: '/ws' })
......
server/src/realtime/store.ts
...@@ -4,6 +4,11 @@ import { config } from '../config.js' ...@@ -4,6 +4,11 @@ import { config } from '../config.js'
import { persistSnapshot, persistWriteLog, pruneCollectData } from '../db/collect.js' import { persistSnapshot, persistWriteLog, pruneCollectData } from '../db/collect.js'
import { getCollectPollMs } from '../settings/store.js' import { getCollectPollMs } from '../settings/store.js'
import { fetchGatewayHealth, fetchGatewaySnapshot, resolveGatewayBase } from '../gateway/client.js' import { fetchGatewayHealth, fetchGatewaySnapshot, resolveGatewayBase } from '../gateway/client.js'
import {
applyMockPointTelemetry,
applyPointTelemetry,
type PointTelemetrySample,
} from '../devices/store.js'
let snapshot: RealtimeSnapshot = generateSnapshot() let snapshot: RealtimeSnapshot = generateSnapshot()
const listeners = new Set<(s: RealtimeSnapshot) => void>() const listeners = new Set<(s: RealtimeSnapshot) => void>()
...@@ -43,6 +48,29 @@ function useGatewayMode() { ...@@ -43,6 +48,29 @@ function useGatewayMode() {
return config.dataMode === 'gateway' || config.dataMode === 'modbus' return config.dataMode === 'gateway' || config.dataMode === 'modbus'
} }
function parseGatewayValues(gwSnap: unknown): PointTelemetrySample[] {
if (!gwSnap || typeof gwSnap !== 'object') return []
const values = (gwSnap as { values?: unknown }).values
if (!Array.isArray(values)) return []
const out: PointTelemetrySample[] = []
for (const row of values) {
if (!row || typeof row !== 'object') continue
const r = row as Record<string, unknown>
const id = typeof r.id === 'string' && r.id ? r.id : undefined
const itemId = typeof r.itemId === 'string' && r.itemId ? r.itemId : undefined
if (!id && !itemId) continue
out.push({
id,
itemId,
raw: r.raw === undefined || r.raw === null ? null : Number(r.raw),
eng: r.eng === undefined || r.eng === null ? null : Number(r.eng),
ts: typeof r.ts === 'string' ? r.ts : null,
quality: typeof r.quality === 'string' ? r.quality : undefined,
})
}
return out
}
async function tick() { async function tick() {
if (tickBusy) return if (tickBusy) return
tickBusy = true tickBusy = true
...@@ -50,14 +78,28 @@ async function tick() { ...@@ -50,14 +78,28 @@ async function tick() {
if (!useGatewayMode() || !resolveGatewayBase()) { if (!useGatewayMode() || !resolveGatewayBase()) {
snapshot = generateSnapshot() snapshot = generateSnapshot()
if (useGatewayMode()) snapshot.online.gatewayOk = false if (useGatewayMode()) snapshot.online.gatewayOk = false
// 仅 mock 模式本地造数；gateway 模式优先等 Windows 网关 POST /api/gateway/telemetry
if (config.dataMode === 'mock') {
void applyMockPointTelemetry().catch((e) =>
console.warn('[realtime] mock point telemetry failed', e),
)
}
} else { } else {
const [health, gwSnap] = await Promise.all([fetchGatewayHealth(), fetchGatewaySnapshot()]) const [health, gwSnap] = await Promise.all([fetchGatewayHealth(), fetchGatewaySnapshot()])
// 网关快照合并进业务快照：一期先保留 mock 结构，标记网关连通；点值供设备页展示 // 网关快照合并进业务快照：一期先保留 mock 结构，标记网关连通；点值供设备页展示
snapshot = generateSnapshot() snapshot = generateSnapshot()
snapshot.online.gatewayOk = health.ok snapshot.online.gatewayOk = health.ok
const samples = parseGatewayValues(gwSnap)
if (gwSnap && typeof gwSnap === 'object') { if (gwSnap && typeof gwSnap === 'object') {
const values = (gwSnap as { values?: unknown }).values ;(snapshot as RealtimeSnapshot & { gatewayTelemetry?: unknown }).gatewayTelemetry = (
;(snapshot as RealtimeSnapshot & { gatewayTelemetry?: unknown }).gatewayTelemetry = values gwSnap as { values?: unknown }
).values
}
// 拉取兜底（网关也会主动推送到 /api/gateway/telemetry）
if (samples.length) {
void applyPointTelemetry(samples).catch((e) =>
console.warn('[realtime] gateway point telemetry failed', e),
)
} }
} }
publish() publish()
......
server/src/routes/gateway.ts 0 → 100644
import { Router } from 'express'
import { config } from '../config.js'
import {
applyPointTelemetry,
getBundleForGateway,
type PointTelemetrySample,
} from '../devices/store.js'
export const gatewayRouter = Router()
function ingestAuthorized(req: { headers: Record<string, unknown> }): boolean {
const key = (process.env.GATEWAY_INGEST_KEY || '').trim()
if (!key) return true // 开发联调默认放行；生产请设置 GATEWAY_INGEST_KEY
const hdr = String(req.headers['x-gateway-key'] || '')
const auth = String(req.headers.authorization || '')
if (hdr && hdr === key) return true
if (auth === `Bearer ${key}`) return true
return false
}
/** Windows 网关拉取全量点表（与后台「下发网关」同源） */
gatewayRouter.get('/points-bundle', async (req, res) => {
if (!ingestAuthorized(req as never)) {
res.status(401).json({ ok: false, message: '网关鉴权失败' })
return
}
const bundle = await getBundleForGateway()
res.json({
ok: true,
pointsVersion: bundle.pointsVersion,
gateways: bundle.gateways,
points: bundle.points,
})
})
/**
* Windows 网关主动上报测点 → 写入 device_points.last_*（落库/内存）
* Body: { ts?, pointsVersion?, values: [{ id, itemId, raw, eng, quality, ts }] }
*/
gatewayRouter.post('/telemetry', async (req, res) => {
if (!ingestAuthorized(req as never)) {
res.status(401).json({ ok: false, message: '网关鉴权失败' })
return
}
const values = req.body?.values
if (!Array.isArray(values)) {
res.status(400).json({ ok: false, message: '缺少 values 数组' })
return
}
const samples: PointTelemetrySample[] = []
for (const row of values) {
if (!row || typeof row !== 'object') continue
const id = typeof row.id === 'string' && row.id ? row.id : undefined
const itemId = typeof row.itemId === 'string' && row.itemId ? row.itemId : undefined
if (!id && !itemId) continue
samples.push({
id,
itemId,
raw: row.raw === undefined || row.raw === null ? null : Number(row.raw),
eng: row.eng === undefined || row.eng === null ? null : Number(row.eng),
ts: typeof row.ts === 'string' ? row.ts : null,
quality: typeof row.quality === 'string' ? row.quality : undefined,
})
}
try {
const updated = await applyPointTelemetry(samples)
res.json({
ok: true,
updated,
received: samples.length,
mode: config.dataMode,
})
} catch (e) {
res.status(500).json({
ok: false,
message: e instanceof Error ? e.message : '写入失败',
})
}
})
tools/modbus_sim/requirements.txt
pymodbus>=3.6.0,<4 # 本模拟器已改为纯 socket 服务端，pymodbus 仅作可选客户端探测
pymodbus>=3.6.0,<3.7
tools/modbus_sim/simulate.py
#!/usr/bin/env python3 #!/usr/bin/env python3
""" """
三网关 Modbus TCP 模拟器：从现场协议 CSV 加载约 800 点，持续刷新 Holding 寄存器。 三网关 Modbus TCP 模拟器：从现场协议 CSV 加载约 800 点，持续刷新 Holding 寄存器。
使用纯 socket 实现 FC03，避免 pymodbus 服务端在部分 Windows 环境无法 accept 的问题。
""" """
from __future__ import annotations from __future__ import annotations
import argparse import argparse
import asyncio
import math import math
import socket
import struct import struct
import sys import sys
import threading
import time import time
from dataclasses import dataclass, field from dataclasses import dataclass, field
from pathlib import Path from pathlib import Path
try:
from pymodbus.datastore import (
ModbusSequentialDataBlock,
ModbusServerContext,
ModbusSlaveContext,
)
from pymodbus.server import ModbusTcpServer
except ImportError:
print("请先安装依赖: pip install -r requirements.txt", file=sys.stderr)
raise
try:
from pymodbus.device import ModbusDeviceIdentification # type: ignore
except ImportError: # pragma: no cover
from pymodbus.server.device import ModbusDeviceIdentification # type: ignore
@dataclass @dataclass
class PointDef: class PointDef:
...@@ -47,8 +33,7 @@ class GatewaySim: ...@@ -47,8 +33,7 @@ class GatewaySim:
label: str label: str
port: int port: int
points: list[PointDef] = field(default_factory=list) points: list[PointDef] = field(default_factory=list)
block: ModbusSequentialDataBlock | None = None regs: list[int] = field(default_factory=list)
context: ModbusServerContext | None = None
GATEWAY_FOLDERS = { GATEWAY_FOLDERS = {
...@@ -57,9 +42,7 @@ GATEWAY_FOLDERS = { ...@@ -57,9 +42,7 @@ GATEWAY_FOLDERS = {
"104": ("房间", "屏 / PLC / 冰箱 / N·E电"), "104": ("房间", "屏 / PLC / 冰箱 / N·E电"),
} }
BLOCK_SIZE = 2048 # 覆盖地址 1–882（含 Double 占 4 字） BLOCK_SIZE = 2048
# pymodbus 3.x SlaveContext 读时 address+=1，直接写 block 需同样偏移
BLOCK_ADDR_OFFSET = 1
def decode_csv_bytes(raw: bytes) -> str: def decode_csv_bytes(raw: bytes) -> str:
...@@ -79,11 +62,11 @@ def split_csv_line(line: str) -> list[str]: ...@@ -79,11 +62,11 @@ def split_csv_line(line: str) -> list[str]:
while i < len(line): while i < len(line):
ch = line[i] ch = line[i]
if in_q: if in_q:
if ch == '"' and i + 1 < len(line) and line[i + 1] == '"':
cur += '"'
i += 2
continue
if ch == '"': if ch == '"':
if i + 1 < len(line) and line[i + 1] == '"':
cur += '"'
i += 2
continue
in_q = False in_q = False
i += 1 i += 1
continue continue
...@@ -94,7 +77,7 @@ def split_csv_line(line: str) -> list[str]: ...@@ -94,7 +77,7 @@ def split_csv_line(line: str) -> list[str]:
in_q = True in_q = True
i += 1 i += 1
continue continue
if ch == ",": if ch == ',':
out.append(cur) out.append(cur)
cur = "" cur = ""
i += 1 i += 1
...@@ -119,28 +102,25 @@ def parse_points_csv(path: Path, gateway: str) -> list[PointDef]: ...@@ -119,28 +102,25 @@ def parse_points_csv(path: Path, gateway: str) -> list[PointDef]:
return lower[n.lower()] return lower[n.lower()]
return -1 return -1
i_name = col("Name") i_name = col("Name", "名称")
i_type = col("DataType") i_addr = col("Modbus Address", "Address", "地址")
i_addr = col("Modbus Address", "ModbusAddress") i_type = col("Data Type", "Type", "数据类型")
i_item = col("ItemID", "ItemId") i_item = col("ItemID", "Item Id", "Item")
i_scale = col("Scale") i_scale = col("Scale")
if i_name < 0 or i_addr < 0: if i_name < 0 or i_addr < 0:
print(f"[skip] 缺 Name/Modbus Address: {path.name}") print(f"[skip] 缺 Name/Modbus Address: {path.name}")
return [] return []
points: list[PointDef] = [] pts: list[PointDef] = []
for ln in lines[1:]: for ln in lines[1:]:
cols = split_csv_line(ln) cols = split_csv_line(ln)
# 现场表偶发多列，ItemID 在最后
item = cols[i_item].strip() if 0 <= i_item < len(cols) else "" item = cols[i_item].strip() if 0 <= i_item < len(cols) else ""
if len(cols) > len(header): if len(cols) > len(header):
last = cols[-1].strip() item = cols[-1].strip()
if last and ("RTU" in last or "." in last):
item = last
name = cols[i_name].strip() if i_name < len(cols) else "" name = cols[i_name].strip() if i_name < len(cols) else ""
try: try:
addr = int(float(cols[i_addr].strip())) address = int(float(cols[i_addr])) if i_addr < len(cols) and cols[i_addr].strip() else -1
except (ValueError, IndexError): except ValueError:
continue continue
dtype = cols[i_type].strip() if 0 <= i_type < len(cols) else "Short" dtype = cols[i_type].strip() if 0 <= i_type < len(cols) else "Short"
try: try:
...@@ -149,35 +129,35 @@ def parse_points_csv(path: Path, gateway: str) -> list[PointDef]: ...@@ -149,35 +129,35 @@ def parse_points_csv(path: Path, gateway: str) -> list[PointDef]:
scale = 0.0 scale = 0.0
if not name and not item: if not name and not item:
continue continue
points.append( if address < 0:
continue
pts.append(
PointDef( PointDef(
name=name or item or f"addr-{addr}", name=name,
address=addr, address=address,
data_type=dtype or "Short", data_type=dtype,
item_id=item or f"{name}@{addr}", item_id=item,
scale=scale, scale=scale,
gateway=gateway, gateway=gateway,
file=path.name, file=path.name,
) )
) )
return points return pts
def find_csv_root(explicit: str | None) -> Path: def find_csv_root(explicit: str | None) -> Path:
if explicit: if explicit:
p = Path(explicit).expanduser().resolve() p = Path(explicit).expanduser().resolve()
if not p.is_dir(): if not p.is_dir():
raise SystemExit(f"CSV 目录不存在: {p}") raise SystemExit(f"--csv-root 不是目录: {p}")
return p return p
here = Path(__file__).resolve() here = Path(__file__).resolve().parent
# simulate.py → modbus_sim → tools → web → code → 00-澳门镜湖医院
hospital_root = here.parents[4]
candidates = [ candidates = [
hospital_root / "设备对接" / "澳门镜湖医院协议20261019" / "澳门镜湖医院协议20261019", here / "protocol",
here.parents[2] / ".." / ".." / "设备对接" / "澳门镜湖医院协议20261019" / "澳门镜湖医院协议20261019", here.parent.parent.parent / "设备对接" / "澳门镜湖医院协议20261019" / "澳门镜湖医院协议20261019",
Path("/Users/james/Documents/2026项目/00-澳门镜湖医院/设备对接/澳门镜湖医院协议20261019/澳门镜湖医院协议20261019"),
] ]
for c in candidates: for c in candidates:
c = c.resolve()
if c.is_dir() and any(c.rglob("*.csv")): if c.is_dir() and any(c.rglob("*.csv")):
return c return c
raise SystemExit( raise SystemExit(
...@@ -196,73 +176,65 @@ def load_gateways(csv_root: Path, ports: dict[str, int]) -> list[GatewaySim]: ...@@ -196,73 +176,65 @@ def load_gateways(csv_root: Path, ports: dict[str, int]) -> list[GatewaySim]:
pts: list[PointDef] = [] pts: list[PointDef] = []
for csv_path in sorted(d.glob("*.csv")): for csv_path in sorted(d.glob("*.csv")):
pts.extend(parse_points_csv(csv_path, key)) pts.extend(parse_points_csv(csv_path, key))
# 同地址保留第一条
uniq: dict[int, PointDef] = {} uniq: dict[int, PointDef] = {}
for p in pts: for p in pts:
uniq.setdefault(p.address, p) uniq.setdefault(p.address, p)
ordered = sorted(uniq.values(), key=lambda x: x.address) ordered = sorted(uniq.values(), key=lambda x: x.address)
sims.append( sims.append(GatewaySim(key=key, label=label, port=ports[key], points=ordered))
GatewaySim(key=key, label=label, port=ports[key], points=ordered)
)
print(f"[load] .{key} {label}: {len(ordered)} 点 ← {folder}/ ({len(list(d.glob('*.csv')))} 文件)") print(f"[load] .{key} {label}: {len(ordered)} 点 ← {folder}/ ({len(list(d.glob('*.csv')))} 文件)")
return sims return sims
def float_to_regs(value: float) -> tuple[int, int]: def float_to_regs(v: float) -> tuple[int, int]:
raw = struct.pack(">f", float(value)) b = struct.pack(">f", float(v))
hi, lo = struct.unpack(">HH", raw) hi, lo = struct.unpack(">HH", b)
return hi, lo return hi, lo
def double_to_regs(value: float) -> tuple[int, int, int, int]: def double_to_regs(v: float) -> tuple[int, int, int, int]:
raw = struct.pack(">d", float(value)) b = struct.pack(">d", float(v))
return struct.unpack(">HHHH", raw) return struct.unpack(">HHHH", b)
def u16(n: int) -> int: def write_point(regs: list[int], point: PointDef, eng: float, _t: float) -> None:
return int(n) & 0xFFFF addr = point.address
if addr < 0 or addr >= len(regs):
def write_point(block: ModbusSequentialDataBlock, point: PointDef, eng: float, t: float) -> None:
"""按类型写入；CSV Modbus Address 与主站读地址一致，写入时补偿 pymodbus +1。"""
addr = point.address + BLOCK_ADDR_OFFSET
if addr < 0 or addr >= BLOCK_SIZE - 4:
return return
dtype = point.data_type.lower() dtype = point.data_type.lower()
scale = point.scale if point.scale not in (0, 0.0) else None scale = point.scale if point.scale not in (0, 0.0) else None
if "bool" in dtype: if "bool" in dtype:
block.setValues(addr, [1 if eng >= 0.5 else 0]) regs[addr] = 1 if eng >= 0.5 else 0
return return
if "float" in dtype: if "float" in dtype:
hi, lo = float_to_regs(eng) hi, lo = float_to_regs(eng)
block.setValues(addr, [hi, lo]) if addr + 1 < len(regs):
regs[addr], regs[addr + 1] = hi, lo
return return
if "double" in dtype: if "double" in dtype:
regs = list(double_to_regs(eng)) vals = list(double_to_regs(eng))
block.setValues(addr, regs) for i, w in enumerate(vals):
if addr + i < len(regs):
regs[addr + i] = w
return return
# Short / Word / 其它整数：若 scale 为 0.1 一类，寄存器存 raw=eng/scale n = point.name + point.item_id
if scale and abs(scale) > 1e-12: if scale and abs(scale) > 1e-12:
raw = int(round(eng / scale)) raw = int(round(eng / scale))
else: else:
# 默认温度/湿度类按 *0.1 存储 if any(k in n for k in ("温", "湿", "压差", "Temp", "Humid", "冰箱")):
if any(k in point.name for k in ("温", "湿", "temp", "humi")) or "冰箱" in point.item_id:
raw = int(round(eng * 10)) raw = int(round(eng * 10))
else: else:
raw = int(round(eng)) raw = int(round(eng))
if "short" in dtype: if "short" in dtype:
raw = max(-32768, min(32767, raw)) raw = max(-32768, min(32767, raw))
if raw < 0: regs[addr] = raw & 0xFFFF
raw = raw + 65536
else: else:
raw = u16(raw) raw = max(0, min(65535, raw))
block.setValues(addr, [u16(raw)]) regs[addr] = raw
def base_engineering(point: PointDef) -> float: def base_engineering(point: PointDef) -> float:
"""按点名给出合理基线工程值。"""
n = point.name + point.item_id n = point.name + point.item_id
addr = point.address addr = point.address
if any(k in n for k in ("温", "Temp", "温度")): if any(k in n for k in ("温", "Temp", "温度")):
...@@ -287,11 +259,10 @@ def base_engineering(point: PointDef) -> float: ...@@ -287,11 +259,10 @@ def base_engineering(point: PointDef) -> float:
def refresh_values(sims: list[GatewaySim], t0: float) -> None: def refresh_values(sims: list[GatewaySim], t0: float) -> None:
t = time.time() - t0 t = time.time() - t0
for sim in sims: for sim in sims:
if not sim.block: if not sim.regs:
continue continue
for i, p in enumerate(sim.points): for i, p in enumerate(sim.points):
base = base_engineering(p) base = base_engineering(p)
# 慢变正弦 + 轻微噪声感（确定性，便于复现）
wave = math.sin(t * 0.15 + i * 0.17 + p.address * 0.01) wave = math.sin(t * 0.15 + i * 0.17 + p.address * 0.01)
if "bool" in p.data_type.lower() or any( if "bool" in p.data_type.lower() or any(
k in (p.name + p.item_id) for k in ("启停", "报警", "故障") k in (p.name + p.item_id) for k in ("启停", "报警", "故障")
...@@ -299,58 +270,81 @@ def refresh_values(sims: list[GatewaySim], t0: float) -> None: ...@@ -299,58 +270,81 @@ def refresh_values(sims: list[GatewaySim], t0: float) -> None:
eng = 1.0 if (int(t / 5) + i) % 4 else 0.0 eng = 1.0 if (int(t / 5) + i) % 4 else 0.0
else: else:
eng = base * (1.0 + 0.03 * wave) eng = base * (1.0 + 0.03 * wave)
write_point(sim.block, p, eng, t) write_point(sim.regs, p, eng, t)
def build_context(sim: GatewaySim) -> ModbusServerContext: def handle_client(conn: socket.socket, regs: list[int], lock: threading.Lock) -> None:
block = ModbusSequentialDataBlock(0, [0] * BLOCK_SIZE) conn.settimeout(5)
sim.block = block try:
store = ModbusSlaveContext(hr=block) while True:
context = ModbusServerContext(slaves=store, single=True) hdr = b""
sim.context = context while len(hdr) < 7:
refresh_values([sim], time.time()) chunk = conn.recv(7 - len(hdr))
return context if not chunk:
return
hdr += chunk
tid = struct.unpack(">H", hdr[0:2])[0]
length = struct.unpack(">H", hdr[4:6])[0]
unit = hdr[6]
need = length - 1
body = b""
while len(body) < need:
chunk = conn.recv(need - len(body))
if not chunk:
return
body += chunk
if len(body) < 5 or body[0] != 0x03:
# 仅支持 FC03；其它功能码回异常
exc = struct.pack(">HHHBB", tid, 0, 3, unit, (body[0] if body else 0) | 0x80) + b"\x01"
conn.sendall(exc)
continue
addr = struct.unpack(">H", body[1:3])[0]
count = struct.unpack(">H", body[3:5])[0]
if count < 1 or count > 125 or addr + count > len(regs):
exc = struct.pack(">HHHBB", tid, 0, 3, unit, 0x83) + b"\x02"
conn.sendall(exc)
continue
with lock:
data = regs[addr : addr + count]
payload = bytes([unit, 0x03, count * 2]) + b"".join(struct.pack(">H", w & 0xFFFF) for w in data)
resp = struct.pack(">HHH", tid, 0, len(payload)) + payload
conn.sendall(resp)
except OSError:
return
finally:
try:
conn.close()
except OSError:
pass
def identity() -> ModbusDeviceIdentification: def serve_gateway(sim: GatewaySim, host: str, lock: threading.Lock) -> None:
ident = ModbusDeviceIdentification() sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
ident.VendorName = "JhLab" sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
ident.ProductCode = "MODBUS-SIM" sock.bind((host, sim.port))
ident.VendorUrl = "https://localhost" sock.listen(32)
ident.ProductName = "KiangWu Lab Gateway Simulator" print(f"[listen] .{sim.key} {sim.label} → {host}:{sim.port} points={len(sim.points)}", flush=True)
ident.ModelName = "3-Gateway-800pts" while True:
ident.MajorMinorRevision = "1.0.0" conn, _ = sock.accept()
return ident threading.Thread(target=handle_client, args=(conn, sim.regs, lock), daemon=True).start()
async def updater_loop(sims: list[GatewaySim], interval: float) -> None: def run_sync(sims: list[GatewaySim], host: str, interval: float) -> None:
lock = threading.Lock()
for sim in sims:
sim.regs = [0] * BLOCK_SIZE
refresh_values(sims, time.time())
for sim in sims:
threading.Thread(target=serve_gateway, args=(sim, host, lock), daemon=True, name=f"mb-{sim.key}").start()
print("[ready] Ctrl+C 退出", flush=True)
t0 = time.time() t0 = time.time()
while True: while True:
try: try:
refresh_values(sims, t0) with lock:
refresh_values(sims, t0)
except Exception as e: # noqa: BLE001 except Exception as e: # noqa: BLE001
print(f"[update] error: {e}") print(f"[update] error: {e}", flush=True)
await asyncio.sleep(interval) time.sleep(interval)
async def run_async(sims: list[GatewaySim], host: str, interval: float) -> None:
for sim in sims:
build_context(sim)
print(f"[listen] .{sim.key} {sim.label} → {host}:{sim.port} points={len(sim.points)}")
servers = [
ModbusTcpServer(
context=sim.context,
identity=identity(),
address=(host, sim.port),
)
for sim in sims
]
print("[ready] Ctrl+C 退出")
await asyncio.gather(
updater_loop(sims, interval),
*[srv.serve_forever() for srv in servers],
)
def main() -> None: def main() -> None:
...@@ -364,16 +358,16 @@ def main() -> None: ...@@ -364,16 +358,16 @@ def main() -> None:
args = ap.parse_args() args = ap.parse_args()
csv_root = find_csv_root(args.csv_root) csv_root = find_csv_root(args.csv_root)
print(f"[csv] {csv_root}") print(f"[csv] {csv_root}", flush=True)
ports = {"102": args.port_102, "103": args.port_103, "104": args.port_104} ports = {"102": args.port_102, "103": args.port_103, "104": args.port_104}
sims = load_gateways(csv_root, ports) sims = load_gateways(csv_root, ports)
total = sum(len(s.points) for s in sims) total = sum(len(s.points) for s in sims)
print(f"[total] {total} 唯一地址点位（三网关）") print(f"[total] {total} 唯一地址点位（三网关）", flush=True)
if total == 0: if total == 0:
raise SystemExit("未加载到任何点位") raise SystemExit("未加载到任何点位")
try: try:
asyncio.run(run_async(sims, args.host, args.interval)) run_sync(sims, args.host, args.interval)
except KeyboardInterrupt: except KeyboardInterrupt:
print("\n[exit]") print("\n[exit]")
sys.exit(0) sys.exit(0)
......
web/Dockerfile
...@@ -11,10 +11,12 @@ COPY web ./web ...@@ -11,10 +11,12 @@ COPY web ./web
WORKDIR /app/web WORKDIR /app/web
ARG VITE_API_BASE= ARG VITE_API_BASE=
ARG VITE_VIEWPORT_MODE=glb ARG VITE_VIEWPORT_MODE=rtengine
ARG VITE_UE_STREAM_URL= ARG VITE_RTENGINE_URL=http://192.168.32.121:6088/gis.html
ARG VITE_UE_STREAM_URL=http://192.168.32.121:6088/gis.html
ENV VITE_API_BASE=$VITE_API_BASE \ ENV VITE_API_BASE=$VITE_API_BASE \
VITE_VIEWPORT_MODE=$VITE_VIEWPORT_MODE \ VITE_VIEWPORT_MODE=$VITE_VIEWPORT_MODE \
VITE_RTENGINE_URL=$VITE_RTENGINE_URL \
VITE_UE_STREAM_URL=$VITE_UE_STREAM_URL VITE_UE_STREAM_URL=$VITE_UE_STREAM_URL
RUN npx vite build RUN npx vite build
......
web/src/components/scene/SceneViewport.vue
...@@ -12,8 +12,20 @@ withDefaults( ...@@ -12,8 +12,20 @@ withDefaults(
) )
const realtime = useRealtimeStore() const realtime = useRealtimeStore()
const mode = (import.meta.env.VITE_VIEWPORT_MODE || 'glb') as 'glb' | 'ue-stream'
const ueUrl = import.meta.env.VITE_UE_STREAM_URL || '' type ViewportMode = 'glb' | 'ue-stream' | 'rtengine'
const mode = (import.meta.env.VITE_VIEWPORT_MODE || 'rtengine') as ViewportMode
/** RTEngineHD / UE 云渲染页面地址（iframe） */
const cloudUrl =
(import.meta.env.VITE_RTENGINE_URL || '').trim() ||
(import.meta.env.VITE_UE_STREAM_URL || '').trim() ||
'http://192.168.32.121:6088/gis.html'
const useCloudFrame = computed(
() =>
(mode === 'rtengine' || mode === 'ue-stream') && Boolean(cloudUrl),
)
const viewerRef = ref<HTMLElement | null>(null) const viewerRef = ref<HTMLElement | null>(null)
type SmoothControlsLike = { type SmoothControlsLike = {
...@@ -89,6 +101,7 @@ const selected = computed(() => { ...@@ -89,6 +101,7 @@ const selected = computed(() => {
}) })
const markers = computed(() => { const markers = computed(() => {
if (useCloudFrame.value) return []
return roomList.value return roomList.value
.filter((r) => ['sample-1', 'lib-1', 'seq'].includes(r.id)) .filter((r) => ['sample-1', 'lib-1', 'seq'].includes(r.id))
.map((r, i) => ({ .map((r, i) => ({
...@@ -102,11 +115,18 @@ const markers = computed(() => { ...@@ -102,11 +115,18 @@ const markers = computed(() => {
})) }))
}) })
const modeLabel = computed(() => {
if (mode === 'rtengine') return 'RTEngineHD 云渲染'
if (mode === 'ue-stream') return 'UE 推流模式'
return '实景扫描模型 · 自由视角'
})
function onSelect(id: string) { function onSelect(id: string) {
realtime.selectRoom(id) realtime.selectRoom(id)
} }
onMounted(() => { onMounted(() => {
if (useCloudFrame.value) return
const el = viewerRef.value const el = viewerRef.value
el?.addEventListener('load', onViewerLoad) el?.addEventListener('load', onViewerLoad)
applyMouseMapping() applyMouseMapping()
...@@ -125,7 +145,7 @@ onBeforeUnmount(() => { ...@@ -125,7 +145,7 @@ onBeforeUnmount(() => {
<AppIcon name="pin" /> <AppIcon name="pin" />
<span class="area">{{ areaLabel || `当前区域：${selected?.name || '—'}` }}</span> <span class="area">{{ areaLabel || `当前区域：${selected?.name || '—'}` }}</span>
<span class="sep" /> <span class="sep" />
<span class="view">{{ mode === 'ue-stream' ? 'UE 推流模式' : '实景扫描模型 · 自由视角' }}</span> <span class="view">{{ modeLabel }}</span>
</div> </div>
<div class="vp-room-panel"> <div class="vp-room-panel">
...@@ -150,22 +170,27 @@ onBeforeUnmount(() => { ...@@ -150,22 +170,27 @@ onBeforeUnmount(() => {
</div> </div>
</div> </div>
<template v-if="mode === 'ue-stream' && ueUrl"> <iframe
<iframe class="ue-frame" :src="ueUrl" title="UE Stream" allow="autoplay; fullscreen" /> v-if="useCloudFrame"
</template> class="ue-frame"
<template v-else> :src="cloudUrl"
<model-viewer title="RTEngineHD Cloud Render"
ref="viewerRef" allow="autoplay; fullscreen; microphone; camera; clipboard-read; clipboard-write; display-capture"
class="scene" allowfullscreen
src="/models/scene-model.glb" referrerpolicy="no-referrer-when-downgrade"
camera-controls />
auto-rotate <model-viewer
shadow-intensity="0.6" v-else
exposure="1.1" ref="viewerRef"
interaction-prompt="none" class="scene"
@contextmenu.prevent src="/models/scene-model.glb"
/> camera-controls
</template> auto-rotate
shadow-intensity="0.6"
exposure="1.1"
interaction-prompt="none"
@contextmenu.prevent
/>
<div <div
v-for="m in markers" v-for="m in markers"
...@@ -181,7 +206,12 @@ onBeforeUnmount(() => { ...@@ -181,7 +206,12 @@ onBeforeUnmount(() => {
<div class="vp-tip"> <div class="vp-tip">
<AppIcon name="hand" class-name="hand" /> <AppIcon name="hand" class-name="hand" />
<span><b>左键</b> 拖动 · <b>右键</b> 旋转 · <b>滚轮</b> 缩放 · 右侧点选房间</span> <span v-if="useCloudFrame">
RTEngineHD 云渲染 · 在画面内操作漫游 · 右侧点选房间
</span>
<span v-else>
<b>左键</b> 拖动 · <b>右键</b> 旋转 · <b>滚轮</b> 缩放 · 右侧点选房间
</span>
</div> </div>
</div> </div>
</template> </template>
...@@ -195,6 +225,7 @@ onBeforeUnmount(() => { ...@@ -195,6 +225,7 @@ onBeforeUnmount(() => {
} }
.scene, .ue-frame { .scene, .ue-frame {
position: absolute; inset: 0; width: 100%; height: 100%; border: 0; position: absolute; inset: 0; width: 100%; height: 100%; border: 0;
background: #000;
} }
.vp-topbar, .vp-tip { .vp-topbar, .vp-tip {
position: absolute; z-index: 5; position: absolute; z-index: 5;
...@@ -203,6 +234,7 @@ onBeforeUnmount(() => { ...@@ -203,6 +234,7 @@ onBeforeUnmount(() => {
background: rgba(8, 26, 61, 0.85); background: rgba(8, 26, 61, 0.85);
border: 1px solid rgba(91, 212, 255, 0.42); border: 1px solid rgba(91, 212, 255, 0.42);
backdrop-filter: blur(10px); backdrop-filter: blur(10px);
pointer-events: none;
} }
.vp-topbar { left: 14px; top: 14px; } .vp-topbar { left: 14px; top: 14px; }
.vp-room-panel { .vp-room-panel {
......
web/src/views/settings/SettingsView.vue
...@@ -12,8 +12,11 @@ interface SystemSettings { ...@@ -12,8 +12,11 @@ interface SystemSettings {
} }
const auth = useAuthStore() const auth = useAuthStore()
const viewportMode = import.meta.env.VITE_VIEWPORT_MODE || 'glb' const viewportMode = import.meta.env.VITE_VIEWPORT_MODE || 'rtengine'
const ueUrl = import.meta.env.VITE_UE_STREAM_URL || '（未配置）' const ueUrl =
import.meta.env.VITE_RTENGINE_URL ||
import.meta.env.VITE_UE_STREAM_URL ||
'（未配置）'
const settings = ref<SystemSettings | null>(null) const settings = ref<SystemSettings | null>(null)
const pollMs = ref(2000) const pollMs = ref(2000)
...@@ -35,7 +38,7 @@ const presets = [ ...@@ -35,7 +38,7 @@ const presets = [
const rows = computed(() => [ const rows = computed(() => [
{ key: 'user', label: '当前用户', value: `${auth.user?.displayName || '—'} (${auth.user?.username || '—'})` }, { key: 'user', label: '当前用户', value: `${auth.user?.displayName || '—'} (${auth.user?.username || '—'})` },
{ key: 'viewport', label: '三维视口模式', value: viewportMode }, { key: 'viewport', label: '三维视口模式', value: viewportMode },
{ key: 'ue', label: 'UE 推流地址', value: ueUrl }, { key: 'ue', label: '云渲染 / UE 地址', value: ueUrl },
{ {
key: 'dataMode', key: 'dataMode',
label: '数据采集模式', label: '数据采集模式',
......
web/src/vite-env.d.ts
...@@ -3,8 +3,10 @@ ...@@ -3,8 +3,10 @@
interface ImportMetaEnv { interface ImportMetaEnv {
readonly VITE_API_BASE: string readonly VITE_API_BASE: string
readonly VITE_WS_URL: string readonly VITE_WS_URL: string
readonly VITE_VIEWPORT_MODE: 'glb' | 'ue-stream' readonly VITE_VIEWPORT_MODE: 'glb' | 'ue-stream' | 'rtengine'
readonly VITE_UE_STREAM_URL: string readonly VITE_UE_STREAM_URL: string
/** RTEngineHD 云渲染页面，如 http://192.168.32.121:6088/gis.html */
readonly VITE_RTENGINE_URL: string
} }
interface ImportMeta { interface ImportMeta {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment