From 4bb846cf0772d707ad51758b928d7806e3ad6162 Mon Sep 17 00:00:00 2001 From: gaozheng Date: Tue, 23 Jun 2026 22:18:45 +0800 Subject: [PATCH] =?UTF-8?q?feat(data):=20=E6=B5=81=E5=BC=8F=E5=BB=BA?= =?UTF-8?q?=E4=BD=93=20buildGprVolumeStreaming(=E6=B2=BFX=E5=88=86slab,?= =?UTF-8?q?=E5=86=85=E5=AD=98=E6=9C=89=E7=95=8C)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 沿X按brick对齐分slab逐块建level0体:assembleGprSurveySlab→共享采样核 sampleGprPoint→writeBrick→释放,峰值内存只随单slab,不持整卷。产出与非流式 buildGprVolume+ChunkedVolumeStore::write逐brick+meta完全一致(对拍守护)。 - 真DRY:抽buildGprVolume的逐点采样核(X/Z落格+Y向1D插值+quant.toQ)为共享自由 函数geopro::core::sampleGprPoint,整卷版与流式版同调,零漂移;原对拍测试守护 buildGprVolume公开行为不变。 - 依赖方向:StreamingVolumeBuilder置src/data,命名空间geopro::data,编进 geopro_store(store增链geopro_io_gpr;io_gpr仅依赖core,无环),core保持纯净。 - 全局量化一致:扫全线全部道tile定vmin/vmax(每次只持一道块),scale/offset与 buildGprVolume同式,逐体素量化一致。 - B3 MEDIUM:StreamingVolumeWriter改持久ofstream成员(构造开/writeBrick复用/ finalize关),消除每块open/close;同步更新对应回归测试的writer作用域。 --- src/core/algo/GprVolumeBuilder.cpp | 107 +++++------ src/core/algo/GprVolumeBuilder.hpp | 12 ++ src/data/StreamingVolumeBuilder.cpp | 208 ++++++++++++++++++++++ src/data/StreamingVolumeBuilder.hpp | 31 ++++ src/data/store/CMakeLists.txt | 7 +- src/data/store/ChunkedVolumeStore.cpp | 28 +-- src/data/store/ChunkedVolumeStore.hpp | 2 + tests/CMakeLists.txt | 2 + tests/data/store/test_streaming_write.cpp | 27 +-- tests/data/test_streaming_builder.cpp | 172 ++++++++++++++++++ 10 files changed, 519 insertions(+), 77 deletions(-) create mode 100644 src/data/StreamingVolumeBuilder.cpp create mode 100644 src/data/StreamingVolumeBuilder.hpp create mode 100644 tests/data/test_streaming_builder.cpp diff --git a/src/core/algo/GprVolumeBuilder.cpp b/src/core/algo/GprVolumeBuilder.cpp index b06ceda..c4c1177 100644 --- a/src/core/algo/GprVolumeBuilder.cpp +++ b/src/core/algo/GprVolumeBuilder.cpp @@ -17,6 +17,59 @@ int nearestIndex(double world, double origin, double step, int n) { } // namespace +std::int16_t sampleGprPoint(const GprSurvey& s, const GridSpec& spec, int gi, + int gj, int gk, const Quant& quant) { + const int nChan = static_cast(s.channelY.size()); + + const double worldZ = spec.oz + gk * spec.dz; + const int sIdx = nearestIndex(worldZ, s.z0, s.dz, s.samples); + + const double worldX = spec.ox + gi * spec.dx; + const int tIdx = nearestIndex(worldX, s.x0, s.dx, s.ntraces); + + // X / Z 越界 → blank。 + if (tIdx < 0 || sIdx < 0 || nChan == 0) { + return ScalarVolumeI16::kBlank; + } + + const double worldY = spec.oy + gj * spec.dy; + + // Y → 跨通道 1D 线性插值（channelY 升序）。 + double phys = 0.0; + bool blank = false; + if (worldY <= s.channelY.front()) { + // 边界外不外推：在 maxDist 内 clamp 到首通道，否则 blank。 + if (s.channelY.front() - worldY > spec.maxDist) { + blank = true; + } else { + phys = s.at(0, tIdx, sIdx); + } + } else if (worldY >= s.channelY.back()) { + if (worldY - s.channelY.back() > spec.maxDist) { + blank = true; + } else { + phys = s.at(nChan - 1, tIdx, sIdx); + } + } else { + // 定位 worldY 落在哪两相邻通道间，线性插值。 + int lo = 0; + while (lo + 1 < nChan && s.channelY[lo + 1] < worldY) ++lo; + const int hi = lo + 1; + const double yLo = s.channelY[lo]; + const double yHi = s.channelY[hi]; + const double span = yHi - yLo; + const double w = (span > 0.0) ? (worldY - yLo) / span : 0.0; + const double vLo = s.at(lo, tIdx, sIdx); + const double vHi = s.at(hi, tIdx, sIdx); + phys = vLo + (vHi - vLo) * w; + } + + if (blank || std::isnan(phys)) { + return ScalarVolumeI16::kBlank; + } + return quant.toQ(phys); +} + BuiltI16 buildGprVolume(const GprSurvey& s, const GridSpec& spec) { BuiltI16 out; out.origin = {spec.ox, spec.oy, spec.oz}; @@ -47,61 +100,11 @@ BuiltI16 buildGprVolume(const GprSurvey& s, const GridSpec& spec) { // 3. 分配体（构造即填 0），origin/spacing 已设。 out.vol = ScalarVolumeI16(spec.nx, spec.ny, spec.nz); - const int nChan = static_cast(s.channelY.size()); - - // 4. 逐网格点落值。 + // 4. 逐网格点落值（复用共享采样核 sampleGprPoint，与流式版零漂移）。 for (int gk = 0; gk < spec.nz; ++gk) { - const double worldZ = spec.oz + gk * spec.dz; - const int sIdx = nearestIndex(worldZ, s.z0, s.dz, s.samples); - for (int gj = 0; gj < spec.ny; ++gj) { - const double worldY = spec.oy + gj * spec.dy; - for (int gi = 0; gi < spec.nx; ++gi) { - const double worldX = spec.ox + gi * spec.dx; - const int tIdx = nearestIndex(worldX, s.x0, s.dx, s.ntraces); - - // X / Z 越界 → blank。 - if (tIdx < 0 || sIdx < 0 || nChan == 0) { - out.vol.at(gi, gj, gk) = ScalarVolumeI16::kBlank; - continue; - } - - // Y → 跨通道 1D 线性插值（channelY 升序）。 - double phys = 0.0; - bool blank = false; - if (worldY <= s.channelY.front()) { - // 边界外不外推：在 maxDist 内 clamp 到首通道，否则 blank。 - if (s.channelY.front() - worldY > spec.maxDist) { - blank = true; - } else { - phys = s.at(0, tIdx, sIdx); - } - } else if (worldY >= s.channelY.back()) { - if (worldY - s.channelY.back() > spec.maxDist) { - blank = true; - } else { - phys = s.at(nChan - 1, tIdx, sIdx); - } - } else { - // 定位 worldY 落在哪两相邻通道间，线性插值。 - int lo = 0; - while (lo + 1 < nChan && s.channelY[lo + 1] < worldY) ++lo; - const int hi = lo + 1; - const double yLo = s.channelY[lo]; - const double yHi = s.channelY[hi]; - const double span = yHi - yLo; - const double w = (span > 0.0) ? (worldY - yLo) / span : 0.0; - const double vLo = s.at(lo, tIdx, sIdx); - const double vHi = s.at(hi, tIdx, sIdx); - phys = vLo + (vHi - vLo) * w; - } - - if (blank || std::isnan(phys)) { - out.vol.at(gi, gj, gk) = ScalarVolumeI16::kBlank; - } else { - out.vol.at(gi, gj, gk) = out.quant.toQ(phys); - } + out.vol.at(gi, gj, gk) = sampleGprPoint(s, spec, gi, gj, gk, out.quant); } } } diff --git a/src/core/algo/GprVolumeBuilder.hpp b/src/core/algo/GprVolumeBuilder.hpp index d4e2623..1a3bbfc 100644 --- a/src/core/algo/GprVolumeBuilder.hpp +++ b/src/core/algo/GprVolumeBuilder.hpp @@ -18,6 +18,18 @@ struct BuiltI16 { double vminPhys = 0, vmaxPhys = 0; }; +// 单网格点采样核（X/Z 落格 + 仅 Y 向跨通道 1D 线性插值），供整卷 buildGprVolume +// 与流式 buildGprVolumeStreaming 共用，确保两者逐体素一致（真 DRY，零漂移）。 +// +// (gi,gj,gk) 为网格索引；spec 提供世界坐标轴与 maxDist；s 提供道/采样标尺、通道 Y、值。 +// quant 为全局量化映射（两版必须用同一 scale/offset）。 +// 返回该点量化后的 int16；越界/超 maxDist/NaN → ScalarVolumeI16::kBlank（不外推）。 +// +// 关键性质：每点只依赖自身 (gi,gj,gk) 经由世界坐标落到的最近道/采样与跨通道插值， +// 不依赖相邻 X，故按 X 分 slab 逐块算与整卷算逐体素相同（流式精确对拍的基础）。 +std::int16_t sampleGprPoint(const GprSurvey& s, const GridSpec& spec, int gi, + int gj, int gk, const Quant& quant); + // 结构化建体：X/Z 直接落格（取最近道/采样）+ 仅 Y 向跨通道 1D 线性插值。 // 超 X/Z 范围或 Y 越界且超 maxDist 的网格点 → ScalarVolumeI16::kBlank（不外推）。 BuiltI16 buildGprVolume(const GprSurvey& s, const GridSpec& spec); diff --git a/src/data/StreamingVolumeBuilder.cpp b/src/data/StreamingVolumeBuilder.cpp new file mode 100644 index 0000000..20c6a6d --- /dev/null +++ b/src/data/StreamingVolumeBuilder.cpp @@ -0,0 +1,208 @@ +#include "data/StreamingVolumeBuilder.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "core/algo/GprVolumeBuilder.hpp" // geopro::core::sampleGprPoint, Quant +#include "core/model/GprSurvey.hpp" +#include "core/model/ScalarVolumeI16.hpp" // kBlank +#include "data/store/ChunkedVolumeStore.hpp" +#include "io/gpr/GprSurveyAssembler.hpp" +#include "io/gpr/IprHeader.hpp" + +namespace geopro::data { + +namespace { + +namespace fs = std::filesystem; +using geopro::core::GridSpec; +using geopro::core::Quant; +using geopro::core::ScalarVolumeI16; + +constexpr std::int16_t kBlank = ScalarVolumeI16::kBlank; + +int ceilDiv(int n, int brick) { return (n + brick - 1) / brick; } + +// 块尺寸（边缘块 < brick）。 +int extent(int n, int b, int brick) { + const int got = n - b * brick; + return got < brick ? got : brick; +} + +// .iprb 路径 → 同名 .iprh（取最后一个路径分隔符之后的最后一个 '.'）。 +// 与 GprSurveyAssembler 内部一致，但那是其匿名命名空间私有，故此处独立小副本。 +std::string toHeaderPath(const std::string& iprbPath) { + const std::size_t dot = iprbPath.find_last_of('.'); + const std::size_t slash = iprbPath.find_last_of("/\\"); + if (dot != std::string::npos && + (slash == std::string::npos || dot > slash)) { + return iprbPath.substr(0, dot) + ".iprh"; + } + return iprbPath + ".iprh"; +} + +std::string readFileText(const std::string& path) { + std::ifstream f(path, std::ios::binary); + if (!f) throw std::runtime_error("StreamingVolumeBuilder: 无法打开 " + path); + std::ostringstream ss; + ss << f.rdbuf(); + return ss.str(); +} + +// 全线总道数 = min 通道(fileBytes/(samples*2))，与 assembleGprSurvey 的对齐口径一致。 +// 同时回传 survey.dx（道距，各通道 header.distanceInterval 一致，取首通道）。 +std::int64_t totalTraces(const std::vector& iprb, double& surveyDx) { + if (iprb.empty()) throw std::runtime_error("StreamingVolumeBuilder: 无通道"); + std::int64_t minTraces = std::numeric_limits::max(); + for (std::size_t c = 0; c < iprb.size(); ++c) { + const geopro::io::gpr::IprHeader h = + geopro::io::gpr::parseIprHeader(readFileText(toHeaderPath(iprb[c]))); + if (h.samples <= 0) + throw std::runtime_error("StreamingVolumeBuilder: samples<=0"); + if (c == 0) surveyDx = h.distanceInterval; + const std::int64_t bytes = + static_cast(fs::file_size(fs::path(iprb[c]))); + const std::int64_t per = static_cast(h.samples) * 2; + if (per <= 0 || bytes % per != 0) + throw std::runtime_error("StreamingVolumeBuilder: .iprb 字节非整道"); + minTraces = std::min(minTraces, bytes / per); + } + return minTraces; +} + +// 网格 X 列 [gx0,gx1) 经世界坐标落到的全局道索引范围（含端点），夹到 [0,total)。 +// 返回 false 表示该列范围内无任何网格点落进 [0,total)（整 slab 全 X 越界）。 +bool traceRangeForColumns(const GridSpec& spec, double surveyDx, + std::int64_t total, int gx0, int gx1, + std::int64_t& t0, std::int64_t& t1) { + std::int64_t lo = std::numeric_limits::max(); + std::int64_t hi = std::numeric_limits::min(); + // survey.x0=0：g = lround(worldX / surveyDx)，与 nearestIndex 同式。 + for (int gi = gx0; gi < gx1; ++gi) { + const double worldX = spec.ox + gi * spec.dx; + if (surveyDx == 0.0) continue; + const std::int64_t g = std::llround(worldX / surveyDx); + if (g < 0 || g >= total) continue; // 越界点非流式即 blank，不扩 slab 道范围 + lo = std::min(lo, g); + hi = std::max(hi, g); + } + if (lo > hi) return false; + t0 = lo; + t1 = hi + 1; // [t0,t1) 半开 + return true; +} + +} // namespace + +void buildGprVolumeStreaming(const std::vector& channelIprbPaths, + const std::string& ordPath, const GridSpec& spec, + const std::string& outDir, int sliceXBricks) { + if (sliceXBricks <= 0) sliceXBricks = 1; + constexpr int kBrick = 64; + + // 0) 全线总道数 + 道距（决定 X 落格，定 slab 道范围）。 + double surveyDx = 1.0; + const std::int64_t total = totalTraces(channelIprbPaths, surveyDx); + + const int bX = ceilDiv(spec.nx, kBrick); + const int bY = ceilDiv(spec.ny, kBrick); + const int bZ = ceilDiv(spec.nz, kBrick); + + // 1) 全局量化：扫【全线全部道】的标量值定 vmin/vmax（不留整卷）。 + // 必须与 buildGprVolume 完全一致——它扫整个 assembleGprSurvey 的 values，与 + // 网格覆盖无关。故此处按固定大小道块 tile [0,total) 全扫，而非只扫网格可达道 + // （否则当网格 X 范围窄于测线时 vmin/vmax 会偏，量化漂移）。单块只持一 slab。 + double vmin = std::numeric_limits::infinity(); + double vmax = -std::numeric_limits::infinity(); + constexpr std::int64_t kScanChunk = 64; // 每次扫的道数（内存只随道块） + for (std::int64_t t0 = 0; t0 < total; t0 += kScanChunk) { + const std::int64_t t1 = std::min(total, t0 + kScanChunk); + const auto slab = + geopro::io::gpr::assembleGprSurveySlab(channelIprbPaths, ordPath, t0, t1); + for (double v : slab.values) { + if (std::isnan(v)) continue; + if (v < vmin) vmin = v; + if (v > vmax) vmax = v; + } + } + if (!(vmin <= vmax)) { // 无有效值：退化 [0,0]，同 buildGprVolume。 + vmin = 0.0; + vmax = 0.0; + } + + Quant quant; + quant.scale = (vmax > vmin) ? (vmax - vmin) / 64000.0 : 1.0; + quant.offset = 0.5 * (vmin + vmax); + + // 2) StoreMeta（dims/brick/origin/spacing/quant/vminmax 同 buildGprVolume+write）。 + StoreMeta meta; + meta.nx = spec.nx; + meta.ny = spec.ny; + meta.nz = spec.nz; + meta.brick = kBrick; + meta.origin = {spec.ox, spec.oy, spec.oz}; + meta.spacing = {spec.dx, spec.dy, spec.dz}; + meta.quant = quant; + meta.vminPhys = vmin; + meta.vmaxPhys = vmax; + + StreamingVolumeWriter w(outDir, meta); + + // 3) 沿 X 分 slab（brick 对齐）逐块写。 + for (int bcol = 0; bcol < bX; bcol += sliceXBricks) { + const int bxEnd = std::min(bX, bcol + sliceXBricks); // 该 slab 含的 brick 列 [bcol,bxEnd) + const int gx0 = bcol * kBrick; + const int gx1 = std::min(spec.nx, bxEnd * kBrick); + + // 该 slab 的全局道范围 → 局部 survey（x0=t0*dx, ntraces=t1-t0），可能全越界。 + std::int64_t t0 = 0, t1 = 0; + const bool hasTraces = + traceRangeForColumns(spec, surveyDx, total, gx0, gx1, t0, t1); + + geopro::core::GprSurvey slab; + if (hasTraces) { + slab = geopro::io::gpr::assembleGprSurveySlab(channelIprbPaths, ordPath, + t0, t1); + } + + // 逐 brick 写：该 slab 的每个 X 列 brick × 所有 Y,Z brick。 + for (int bz = 0; bz < bZ; ++bz) { + for (int by = 0; by < bY; ++by) { + for (int bx = bcol; bx < bxEnd; ++bx) { + const int bw = extent(spec.nx, bx, kBrick); + const int bh = extent(spec.ny, by, kBrick); + const int bd = extent(spec.nz, bz, kBrick); + std::vector voxels( + static_cast(bw) * bh * bd); + const int i0 = bx * kBrick, j0 = by * kBrick, k0 = bz * kBrick; + std::size_t wi = 0; + for (int kk = 0; kk < bd; ++kk) { + for (int jj = 0; jj < bh; ++jj) { + for (int ii = 0; ii < bw; ++ii) { + if (!hasTraces) { + voxels[wi++] = kBlank; // 整列 X 越界 → blank（同非流式） + } else { + voxels[wi++] = geopro::core::sampleGprPoint( + slab, spec, i0 + ii, j0 + jj, k0 + kk, quant); + } + } + } + } + w.writeBrick(bx, by, bz, voxels); + } + } + } + // slab 缓冲随 for 作用域结束释放（下一 slab 重新装配）。 + } + + w.finalize(); +} + +} // namespace geopro::data diff --git a/src/data/StreamingVolumeBuilder.hpp b/src/data/StreamingVolumeBuilder.hpp new file mode 100644 index 0000000..b5e0e5d --- /dev/null +++ b/src/data/StreamingVolumeBuilder.hpp @@ -0,0 +1,31 @@ +#ifndef GEOPRO_DATA_STREAMINGVOLUMEBUILDER_HPP +#define GEOPRO_DATA_STREAMINGVOLUMEBUILDER_HPP + +#include +#include + +#include "core/algo/IInterpolator.hpp" // geopro::core::GridSpec + +namespace geopro::data { + +// 流式建 level0 体到 outDir。沿 X 按 brick 对齐分 slab，逐 slab： +// assembleGprSurveySlab → 共享采样核 sampleGprPoint → writeBrick → 释放，不持整卷， +// 峰值内存只随单个 slab。 +// +// 量化全局一致：先扫所有 slab 的标量值定全局 vmin/vmax（不留整卷）， +// scale=(vmax-vmin)/64000、offset=值域中点（与非流式 buildGprVolume 完全相同）， +// 保证逐体素量化一致。 +// +// 产出与非流式 buildGprVolume + ChunkedVolumeStore::write 逐 brick + meta 完全一致 +// （结构化建体无 X 邻域耦合，按 X 分块算 = 整卷算逐体素）。 +// +// sliceXBricks：每 slab 含多少个 X 方向 brick（slab 的网格 X 范围 = 这些 brick 覆盖的格）。 +// <=0 视为 1。 +void buildGprVolumeStreaming( + const std::vector& channelIprbPaths, const std::string& ordPath, + const geopro::core::GridSpec& spec, const std::string& outDir, + int sliceXBricks = 8); + +} // namespace geopro::data + +#endif // GEOPRO_DATA_STREAMINGVOLUMEBUILDER_HPP diff --git a/src/data/store/CMakeLists.txt b/src/data/store/CMakeLists.txt index 516cb9a..ab139b5 100644 --- a/src/data/store/CMakeLists.txt +++ b/src/data/store/CMakeLists.txt @@ -4,13 +4,16 @@ find_package(nlohmann_json CONFIG REQUIRED) find_package(Qt6 COMPONENTS Core REQUIRED) add_library(geopro_store STATIC - ChunkedVolumeStore.cpp) + ChunkedVolumeStore.cpp + # 流式建体（B4）：编排 io_gpr+core+store，沿 X 分 slab 建 level0 体（命名空间 geopro::data）。 + ${CMAKE_CURRENT_SOURCE_DIR}/../StreamingVolumeBuilder.cpp) # include 根 = src/，使 #include "data/store/..." 与 "core/algo/..." 可解析 # （geopro_tests 链 geopro_store 后透传）。 target_include_directories(geopro_store PUBLIC ${CMAKE_SOURCE_DIR}/src) +# geopro_io_gpr 仅依赖 core（无环）：流式建体需 assembleGprSurveySlab/IprHeader。 target_link_libraries(geopro_store - PUBLIC geopro_core Qt6::Core + PUBLIC geopro_core geopro_io_gpr Qt6::Core PRIVATE nlohmann_json::nlohmann_json) target_compile_features(geopro_store PUBLIC cxx_std_17) set_target_properties(geopro_store PROPERTIES AUTOMOC OFF AUTOUIC OFF AUTORCC OFF) diff --git a/src/data/store/ChunkedVolumeStore.cpp b/src/data/store/ChunkedVolumeStore.cpp index 56fd191..12a55f2 100644 --- a/src/data/store/ChunkedVolumeStore.cpp +++ b/src/data/store/ChunkedVolumeStore.cpp @@ -582,10 +582,11 @@ StreamingVolumeWriter::StreamingVolumeWriter(const std::string& dir, Entry{}); fs::create_directories(fs::path(dir_)); - // 截断式打开一次以建立空 data.bin（writeBrick 用追加模式逐块写）。 - std::ofstream init((fs::path(dir_) / kDataFile).string(), - std::ios::binary | std::ios::trunc); - if (!init) + // B3 MEDIUM 修复：持久句柄。构造时截断式打开一次 data.bin，writeBrick 复用此句柄 + // 顺序追加（不再逐块重开），finalize 关闭。避免每块一次 open/close 的系统调用开销。 + data_.open((fs::path(dir_) / kDataFile).string(), + std::ios::binary | std::ios::trunc); + if (!data_) throw std::runtime_error( "StreamingVolumeWriter: cannot open data.bin for write"); } @@ -615,15 +616,10 @@ void StreamingVolumeWriter::writeBrick(int bx, int by, int bz, const QByteArray compressed = compressBrick(voxels); const std::int64_t clen = compressed.size(); - // 追加写 data.bin（块按 writeBrick 调用顺序物理排布，索引记录各自偏移； - // finalize 再按固定顺序写 meta → readBrick 凭索引偏移定位，物理顺序无关）。 - std::ofstream data((fs::path(dir_) / kDataFile).string(), - std::ios::binary | std::ios::app); - if (!data) - throw std::runtime_error( - "StreamingVolumeWriter: cannot open data.bin for append"); - data.write(compressed.constData(), clen); - if (!data) + // 复用持久句柄顺序追加 data.bin（块按 writeBrick 调用顺序物理排布，索引记录各自 + // 偏移；finalize 再按固定顺序写 meta → readBrick 凭索引偏移定位，物理顺序无关）。 + data_.write(compressed.constData(), clen); + if (!data_) throw std::runtime_error("StreamingVolumeWriter: data.bin write failed"); e.offset = offset_; @@ -642,6 +638,12 @@ void StreamingVolumeWriter::finalize() { if (written_ != static_cast(entries_.size())) throw std::runtime_error("StreamingVolumeWriter: missing bricks at finalize"); + // 刷新并关闭持久 data.bin 句柄（B3 MEDIUM：句柄生命周期 = 构造→finalize）。 + data_.flush(); + if (!data_) + throw std::runtime_error("StreamingVolumeWriter: data.bin flush failed"); + data_.close(); + // 按固定顺序（bz 最慢、bx 最快）输出索引，结构与 write 的 bricks 数组一致。 json bricks = json::array(); for (const Entry& e : entries_) diff --git a/src/data/store/ChunkedVolumeStore.hpp b/src/data/store/ChunkedVolumeStore.hpp index f155794..9cc6581 100644 --- a/src/data/store/ChunkedVolumeStore.hpp +++ b/src/data/store/ChunkedVolumeStore.hpp @@ -3,6 +3,7 @@ #include #include +#include #include #include #include @@ -149,6 +150,7 @@ class StreamingVolumeWriter { StoreMeta meta_; int bricksX_ = 0, bricksY_ = 0, bricksZ_ = 0; std::vector entries_; // 固定顺序索引（bz 最慢、bx 最快） + std::ofstream data_; // 持久 data.bin 句柄（构造开、writeBrick 复用、finalize 关） std::int64_t offset_ = 0; // data.bin 当前追加偏移（64 位） std::int64_t written_ = 0; // 已写块计数 bool finalized_ = false; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 86f1d82..175cf04 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -60,6 +60,8 @@ target_sources(geopro_tests PRIVATE data/store/test_chunked_volume_store.cpp) target_sources(geopro_tests PRIVATE data/store/test_pyramid.cpp) # store 层：StreamingVolumeWriter（逐块增量写 level0；与非流式 write 逐块+meta 对拍一致）。 target_sources(geopro_tests PRIVATE data/store/test_streaming_write.cpp) +# data 层：StreamingVolumeBuilder（流式建体 B4；与非流式 buildGprVolume+write 逐 brick+meta 对拍）。 +target_sources(geopro_tests PRIVATE data/test_streaming_builder.cpp) target_link_libraries(geopro_tests PRIVATE geopro_store) # net 层：RSA 加密器。测试需直接用 OpenSSL 生成/解密密钥，故显式 find_package diff --git a/tests/data/store/test_streaming_write.cpp b/tests/data/store/test_streaming_write.cpp index c17f549..634d971 100644 --- a/tests/data/store/test_streaming_write.cpp +++ b/tests/data/store/test_streaming_write.cpp @@ -109,10 +109,13 @@ TEST(StreamingVolumeWriter, DuplicateBrickThrows) { auto dirB = tmp("swDupB"); std::filesystem::remove_all(dirB); - StreamingVolumeWriter w(dirB, m); - w.writeBrick(0, 0, 0, sliceBrickFrom(b, 0, 0, 0, 64)); - EXPECT_THROW(w.writeBrick(0, 0, 0, sliceBrickFrom(b, 0, 0, 0, 64)), - std::runtime_error); + { + // 持久 data.bin 句柄随 writer 生命周期持有，需先析构 writer 再删目录（Windows 文件锁）。 + StreamingVolumeWriter w(dirB, m); + w.writeBrick(0, 0, 0, sliceBrickFrom(b, 0, 0, 0, 64)); + EXPECT_THROW(w.writeBrick(0, 0, 0, sliceBrickFrom(b, 0, 0, 0, 64)), + std::runtime_error); + } std::filesystem::remove_all(dirA); std::filesystem::remove_all(dirB); } @@ -127,9 +130,11 @@ TEST(StreamingVolumeWriter, MissingBrickFinalizeThrows) { auto dirB = tmp("swMissB"); std::filesystem::remove_all(dirB); - StreamingVolumeWriter w(dirB, m); - w.writeBrick(0, 0, 0, sliceBrickFrom(b, 0, 0, 0, 64)); // 只写 1 块，缺 (1,0,0) - EXPECT_THROW(w.finalize(), std::runtime_error); + { + StreamingVolumeWriter w(dirB, m); + w.writeBrick(0, 0, 0, sliceBrickFrom(b, 0, 0, 0, 64)); // 只写 1 块，缺 (1,0,0) + EXPECT_THROW(w.finalize(), std::runtime_error); + } std::filesystem::remove_all(dirA); std::filesystem::remove_all(dirB); } @@ -144,9 +149,11 @@ TEST(StreamingVolumeWriter, WrongVoxelCountThrows) { auto dirB = tmp("swSizeB"); std::filesystem::remove_all(dirB); - StreamingVolumeWriter w(dirB, m); - std::vector bad(10); // 远小于 64*30*20 - EXPECT_THROW(w.writeBrick(0, 0, 0, bad), std::runtime_error); + { + StreamingVolumeWriter w(dirB, m); + std::vector bad(10); // 远小于 64*30*20 + EXPECT_THROW(w.writeBrick(0, 0, 0, bad), std::runtime_error); + } std::filesystem::remove_all(dirA); std::filesystem::remove_all(dirB); } diff --git a/tests/data/test_streaming_builder.cpp b/tests/data/test_streaming_builder.cpp new file mode 100644 index 0000000..3df9678 --- /dev/null +++ b/tests/data/test_streaming_builder.cpp @@ -0,0 +1,172 @@ +#include "data/StreamingVolumeBuilder.hpp" + +#include + +#include +#include +#include +#include +#include + +#include "core/algo/GprVolumeBuilder.hpp" +#include "core/algo/IInterpolator.hpp" // GridSpec +#include "data/store/ChunkedVolumeStore.hpp" +#include "io/gpr/GprSurveyAssembler.hpp" + +using namespace geopro::data; + +namespace { + +void writeText(const std::string& p, const std::string& s) { + std::ofstream f(p); + f << s; +} + +void writeI16(const std::string& p, const std::vector& v) { + std::ofstream f(p, std::ios::binary); + f.write(reinterpret_cast(v.data()), + static_cast(v.size() * sizeof(std::int16_t))); +} + +std::string tmpDir(const char* name) { + return (std::filesystem::temp_directory_path() / name).string(); +} + +// 写两通道 .iprb/.ord(+.iprh) 到 dir，每通道 nTraces 道、nSamples 采样。 +// 值随 (channel,trace,sample) 变化，确保块内容彼此不同（真正区分逐块对拍）。 +// 返回各通道 .iprb 路径 + .ord 路径。 +struct SurveyFiles { + std::vector iprb; + std::string ord; +}; + +SurveyFiles makeTwoChannelSurveyFiles(const std::string& dir, int nTraces, + int nSamples) { + std::filesystem::create_directories(dir); + const int lastTrace = nTraces - 1; + const std::string hdr = + "SAMPLES: " + std::to_string(nSamples) + + "\nLAST TRACE: " + std::to_string(lastTrace) + + "\nCHANNELS: 2\nTIMEWINDOW: 4.0\n" + "SOIL VELOCITY: 100.000000\nDISTANCE INTERVAL: 0.05\n"; + + auto gen = [&](int chan) { + std::vector v(static_cast(nTraces) * nSamples); + for (int t = 0; t < nTraces; ++t) + for (int s = 0; s < nSamples; ++s) + v[static_cast(t) * nSamples + s] = static_cast( + (chan * 1000 + t * 7 + s * 3) % 251 - 50); // 含负值 + return v; + }; + + writeText(dir + "/A.iprh", hdr); + writeI16(dir + "/A.iprb", gen(0)); + writeText(dir + "/B.iprh", hdr); + writeI16(dir + "/B.iprb", gen(1)); + // A->横偏 1.0、B->横偏 0.0（B 的 Y 更小，验证 Y 升序重排不影响一致性）。 + writeText(dir + "/x.ord", "0 1.0 -1.5 1\n1 0.0 -1.5 1\n"); + + return {{dir + "/A.iprb", dir + "/B.iprb"}, dir + "/x.ord"}; +} + +geopro::core::GridSpec makeSpec() { + geopro::core::GridSpec spec{}; + spec.nx = 200; // > 128 → sliceXBricks=2 时跨多个 slab + spec.ny = 5; + spec.nz = 8; + spec.ox = 0.0; + spec.oy = 0.0; + spec.oz = 0.0; + spec.dx = 0.05; // 与 survey.dx 同步，使网格 X 对齐道 + spec.dy = 0.25; + spec.dz = 0.2; + spec.power = 2.0; + spec.maxDist = 2.0; + return spec; +} + +} // namespace + +// 核心验收：流式逐 slab 建体 vs 非流式整卷 buildGprVolume+write，逐块 + meta 一致。 +TEST(StreamingVolumeBuilder, MatchesNonStreaming) { + const std::string srcDir = tmpDir("svb_src"); + std::filesystem::remove_all(srcDir); + auto files = makeTwoChannelSurveyFiles(srcDir, /*nTraces*/ 250, /*nSamples*/ 8); + const auto spec = makeSpec(); + const int brick = 64; + + const std::string dirA = tmpDir("svb_nsA"); + const std::string dirB = tmpDir("svb_strB"); + std::filesystem::remove_all(dirA); + std::filesystem::remove_all(dirB); + + // 金标准：全装配 → buildGprVolume → write。 + auto full = geopro::io::gpr::assembleGprSurvey(files.iprb, files.ord); + auto built = geopro::core::buildGprVolume(full, spec); + ChunkedVolumeStore::write(dirA, built, brick); + + // 流式：sliceXBricks=2 强制多 slab。 + buildGprVolumeStreaming(files.iprb, files.ord, spec, dirB, /*sliceXBricks*/ 2); + + ChunkedVolumeStore A(dirA), B(dirB); + + // 全 meta 字段一致。 + EXPECT_EQ(B.meta().nx, A.meta().nx); + EXPECT_EQ(B.meta().ny, A.meta().ny); + EXPECT_EQ(B.meta().nz, A.meta().nz); + EXPECT_EQ(B.meta().brick, A.meta().brick); + EXPECT_EQ(B.meta().origin, A.meta().origin); + EXPECT_EQ(B.meta().spacing, A.meta().spacing); + EXPECT_EQ(B.meta().quant.scale, A.meta().quant.scale); + EXPECT_EQ(B.meta().quant.offset, A.meta().quant.offset); + EXPECT_EQ(B.meta().vminPhys, A.meta().vminPhys); + EXPECT_EQ(B.meta().vmaxPhys, A.meta().vmaxPhys); + + // 逐 brick 完全一致。 + const int bX = A.bricksX(), bY = A.bricksY(), bZ = A.bricksZ(); + EXPECT_EQ(B.bricksX(), bX); + EXPECT_EQ(B.bricksY(), bY); + EXPECT_EQ(B.bricksZ(), bZ); + for (int bz = 0; bz < bZ; ++bz) + for (int by = 0; by < bY; ++by) + for (int bx = 0; bx < bX; ++bx) + EXPECT_EQ(A.readBrick(bx, by, bz), B.readBrick(bx, by, bz)) + << "brick mismatch at " << bx << "," << by << "," << bz; + + std::filesystem::remove_all(srcDir); + std::filesystem::remove_all(dirA); + std::filesystem::remove_all(dirB); +} + +// 不同 sliceXBricks 都与非流式一致（slab 边界划分不影响结果）。 +TEST(StreamingVolumeBuilder, SliceCountInvariant) { + const std::string srcDir = tmpDir("svb_inv_src"); + std::filesystem::remove_all(srcDir); + auto files = makeTwoChannelSurveyFiles(srcDir, /*nTraces*/ 250, /*nSamples*/ 8); + const auto spec = makeSpec(); + + const std::string dirA = tmpDir("svb_inv_A"); + std::filesystem::remove_all(dirA); + auto full = geopro::io::gpr::assembleGprSurvey(files.iprb, files.ord); + auto built = geopro::core::buildGprVolume(full, spec); + ChunkedVolumeStore::write(dirA, built, 64); + ChunkedVolumeStore A(dirA); + + for (int slice : {1, 3, 8}) { + const std::string dirB = tmpDir("svb_inv_B"); + std::filesystem::remove_all(dirB); + buildGprVolumeStreaming(files.iprb, files.ord, spec, dirB, slice); + ChunkedVolumeStore B(dirB); + EXPECT_EQ(B.meta().quant.scale, A.meta().quant.scale) << "slice=" << slice; + EXPECT_EQ(B.meta().vminPhys, A.meta().vminPhys) << "slice=" << slice; + for (int bz = 0; bz < A.bricksZ(); ++bz) + for (int by = 0; by < A.bricksY(); ++by) + for (int bx = 0; bx < A.bricksX(); ++bx) + EXPECT_EQ(A.readBrick(bx, by, bz), B.readBrick(bx, by, bz)) + << "slice=" << slice << " brick " << bx << "," << by << "," << bz; + std::filesystem::remove_all(dirB); + } + + std::filesystem::remove_all(srcDir); + std::filesystem::remove_all(dirA); +}