From 0480009f8015fb5680354baea64f39d26dec17c4 Mon Sep 17 00:00:00 2001 From: romer8 Date: Sun, 15 Feb 2026 18:15:21 -0700 Subject: [PATCH 1/6] refactor: update data handling to support Arrow format; enhance caching and database interactions --- .../features/DataStream/lib/duckdbClient.js | 8 +- reactapp/features/DataStream/lib/opfsCache.js | 179 +++++++++++++++--- reactapp/features/DataStream/lib/queryData.js | 87 ++++----- reactapp/features/Tethys/services/api/app.js | 5 +- tethysapp/nrds/controllers.py | 5 +- tethysapp/nrds/data_utils.py | 30 ++- 6 files changed, 208 insertions(+), 106 deletions(-) diff --git a/reactapp/features/DataStream/lib/duckdbClient.js b/reactapp/features/DataStream/lib/duckdbClient.js index 4093e5b..4c0abcd 100644 --- a/reactapp/features/DataStream/lib/duckdbClient.js +++ b/reactapp/features/DataStream/lib/duckdbClient.js @@ -31,9 +31,15 @@ export function getDuckDB() { export async function getConnection() { const db = await getDuckDB(); - return await db.connect(); + + // enable OPFS support once (safe to set repeatedly) + db.config.opfs = { fileHandling: "auto" }; + + const conn = await db.connect(); + return { db, conn }; } + // OPTIONAL: wipe all DB state (tables, etc) but keep worker export async function resetDatabase() { if (!dbPromise) return; diff --git a/reactapp/features/DataStream/lib/opfsCache.js b/reactapp/features/DataStream/lib/opfsCache.js index 39fc736..af73739 100644 --- a/reactapp/features/DataStream/lib/opfsCache.js +++ b/reactapp/features/DataStream/lib/opfsCache.js @@ -1,7 +1,12 @@ +import appAPI from "features/Tethys/services/api/app"; +import { tableFromIPC } from "apache-arrow"; +import { getNCFiles } from "./s3Utils"; + + const CACHE_DIR = "nrds-arrow-cache"; let cacheDirPromise = null; -function formatBytes(bytes, decimals = 2) { +export function formatBytes(bytes, decimals = 2) { if (bytes === 0) return '0 Bytes'; const k = 1024; const dm = decimals < 0 ? 0 : decimals; @@ -30,40 +35,92 @@ async function getCacheDir() { } } -export async function saveArrowToCache(key, buffer) { +// async function saveArrowToCache(url, vpu_gpkg, writable) { +async function saveArrowToCache(url, writable) { + try{ + const ncFile = getNCFiles(url); + const buffer = await appAPI.getArrowPerVpu({ + ncFile, + }); + + let dataToWrite; + + if (buffer instanceof ArrayBuffer) { + dataToWrite = new Uint8Array(buffer); + } else if (ArrayBuffer.isView(buffer)) { + // covers Uint8Array, DataView, etc. + dataToWrite = new Uint8Array(buffer.buffer, buffer.byteOffset, buffer.byteLength); + } else if (buffer instanceof Blob) { + dataToWrite = buffer; + } else { + console.error("saveArrowToCache: unexpected buffer type", buffer); + throw new Error("saveArrowToCache: expected ArrayBuffer, TypedArray, or Blob"); + } + + await writable.write(dataToWrite); + await writable.close(); + } + catch(error){ + console.error("Error fetching Arrow data:", error); + throw error; + } +} + +async function cacheParquetToOPFS(url, writable) { + try { + const res = await fetch(url, { cache: "no-store" }); + if (!res.ok) throw new Error(`Failed to fetch ${url}: ${res.status}`); + + // Stream to disk; avoids loading the entire file in memory + if (!res.body) { + const buf = await res.arrayBuffer(); + await writable.write(new Uint8Array(buf)); + await writable.close(); + } else { + // WritableStream from OPFS supports pipeTo in modern browsers + await res.body.pipeTo(writable); + // pipeTo closes the destination by default + } + } catch (err) { + // If pipeTo fails mid-stream, attempt to close to release the lock. + try { await writable.close(); } catch (_) {} + throw err; + } +} + +const sqlIdent = (s) => `"${String(s).replace(/"/g, '""')}"`; +const sqlStr = (s) => `'${String(s).replace(/'/g, "''")}'`; + +const safeNameForKey = (key) => encodeURIComponent(key); +const tableNameForKey = (key) => String(key).replace(/\.(arrow|parquet)$/i, ""); + +function isNCFile(key) { return key.endsWith('.nc'); } + +function isArrowFile(key) { return key.endsWith('.arrow');} + +function isParquetFile(key) { return key.endsWith('.parquet'); } + +export async function saveDataToCache(key, url) { const dir = await getCacheDir(); if (!dir) return; // noop if OPFS unavailable - - const safeName = encodeURIComponent(key) + ".arrow"; + const safeName = encodeURIComponent(key); const fileHandle = await dir.getFileHandle(safeName, { create: true }); const writable = await fileHandle.createWritable(); - - // 🔍 Make sure we always pass a proper binary type to write() - let dataToWrite; - - if (buffer instanceof ArrayBuffer) { - dataToWrite = new Uint8Array(buffer); - } else if (ArrayBuffer.isView(buffer)) { - // covers Uint8Array, DataView, etc. - dataToWrite = new Uint8Array(buffer.buffer, buffer.byteOffset, buffer.byteLength); - } else if (buffer instanceof Blob) { - dataToWrite = buffer; + if (isArrowFile(key)) { + await saveArrowToCache(url, writable); } else { - console.error("saveArrowToCache: unexpected buffer type", buffer); - throw new Error("saveArrowToCache: expected ArrayBuffer, TypedArray, or Blob"); + await cacheParquetToOPFS(url, writable); } - - await writable.write(dataToWrite); - await writable.close(); const file = await fileHandle.getFile(); return formatBytes(file.size); } -export async function loadArrowFromCache(key) { + +export async function loadFromCache(key) { const dir = await getCacheDir(); if (!dir) return null; - const safeName = encodeURIComponent(key) + ".arrow"; + const safeName = encodeURIComponent(key); try { const fileHandle = await dir.getFileHandle(safeName); const file = await fileHandle.getFile(); @@ -73,6 +130,61 @@ export async function loadArrowFromCache(key) { } } + +async function doesTableExist(conn, tableName) { + const res = await conn.query(` + SELECT 1 + FROM information_schema.tables + WHERE table_schema = 'main' + AND table_name = ${sqlStr(tableName)} + LIMIT 1 + `); + return res.toArray().length > 0; +} + +async function createTableFromOPFSParquet({ db, conn, key }) { + db.config.opfs = { fileHandling: "auto" }; + + const safeName = encodeURIComponent(key); + const fileUrl = `opfs://${CACHE_DIR}/${safeName}`; + const tableName = key.replace(/\.parquet$/i, ""); + + await conn.query(` + CREATE TABLE ${sqlIdent(tableName)} AS + SELECT * FROM read_parquet(${sqlStr(fileUrl)}); + `); +} + +async function createTableFromOPFSArrow({ conn, key }) { + const buffer = await loadFromCache(key); + if (!buffer) throw new Error(`Arrow cache missing after save: ${key}`); + + const arrowTable = tableFromIPC(new Uint8Array(buffer)); + const tableName = tableNameForKey(key); + + await conn.insertArrowTable(arrowTable, { name: tableName }); +} + +export async function createTableFromOPFS({ db, conn, key, safeName }) { + const tableName = tableNameForKey(key); + + if (await doesTableExist(conn, tableName)) { + console.debug(`Table "${tableName}" already exists, skipping.`); + return; + } + + if (isArrowFile(key)) { + return createTableFromOPFSArrow({ conn, key }); + } + if (isParquetFile(key)) { + return createTableFromOPFSParquet({ db, conn, key, safeName }); + } + + throw new Error(`Unsupported file type for key: ${key}`); +} + + + export async function getFilesFromCache() { const dir = await getCacheDir(); if (!dir) return null; @@ -81,16 +193,30 @@ export async function getFilesFromCache() { for await (const handle of dir.values()) { if (handle.kind !== "file") continue; const file = await handle.getFile(); - const id = decodeURIComponent(file.name.replace(".arrow", "")); + const id = decodeURIComponent(file.name.replace(".arrow", "") || file.name.replace(".parquet", "")); files.push({id: id, name: id.replaceAll("_", "/"), size: formatBytes(file.size)}); } return files; } +export async function statFromCache(key) { + const dir = await getCacheDir(); + if (!dir) return null; + + const safeName = safeNameForKey(key); + try { + const fileHandle = await dir.getFileHandle(safeName); + const file = await fileHandle.getFile(); + return { safeName, sizeBytes: file.size }; + } catch { + return null; + } +} + export async function deleteFileFromCache(key) { const dir = await getCacheDir(); if (!dir) return; - const safeName = encodeURIComponent(key) + ".arrow"; + const safeName = encodeURIComponent(key); try { await dir.removeEntry(safeName); return true; @@ -109,8 +235,9 @@ export async function clearCache() { } export function getCacheKey(model, date, forecast, cycle, ensemble, vpu, outputFile) { + const newOutputFile = isNCFile(outputFile) ? outputFile.replace(".nc", ".arrow") : outputFile; if (!ensemble){ - return `${model}_${date}_${forecast}_${cycle}_${vpu}_${outputFile}`.replace(/\./g,'_').replace(/\//g,'_'); + return `${model}_${date}_${forecast}_${cycle}_${vpu}`.replace(/\./g,'_').replace(/\//g,'_') + `_${newOutputFile}`; ; } - return `${model}_${date}_${forecast}_${cycle}_${ensemble}_${vpu}_${outputFile}`.replace(/\./g,'_').replace(/\//g,'_'); + return `${model}_${date}_${forecast}_${cycle}_${ensemble}_${vpu}`.replace(/\./g,'_').replace(/\//g,'_') + `_${newOutputFile}`; } diff --git a/reactapp/features/DataStream/lib/queryData.js b/reactapp/features/DataStream/lib/queryData.js index 81ad357..f0ce5f5 100644 --- a/reactapp/features/DataStream/lib/queryData.js +++ b/reactapp/features/DataStream/lib/queryData.js @@ -1,9 +1,7 @@ // // nexusTimeseries.js -import { tableFromIPC } from "apache-arrow"; -import appAPI from "features/Tethys/services/api/app"; -import { saveArrowToCache, loadArrowFromCache } from "./opfsCache"; +import { statFromCache, saveDataToCache, createTableFromOPFS, formatBytes } from "./opfsCache"; + import { getConnection } from "./duckdbClient"; -import { getNCFiles } from "./s3Utils"; const DEBUG = process.env.NODE_ENV !== "production"; const debugLog = (...args) => { @@ -11,19 +9,19 @@ const debugLog = (...args) => { }; export async function getTimeseries(id, cacheKey, variable) { - const conn = await getConnection(); - + const { conn } = await getConnection(); + const tableName = cacheKey.split('.')[0]; try { const rows = []; const stream = await conn.send(` SELECT time, ${variable} - FROM ${cacheKey} + FROM ${tableName} WHERE feature_id = ${id} ORDER BY time `); debugLog("Query executed:", ` SELECT time, ${variable} - FROM ${cacheKey} + FROM ${tableName} WHERE feature_id = ${id} ORDER BY time `); @@ -54,13 +52,13 @@ export async function getTimeseries(id, cacheKey, variable) { export async function getFeatureIDs(cacheKey) { debugLog("getFeatureIDs called with cacheKey:", cacheKey); - const conn = await getConnection(); - + const { conn } = await getConnection(); + const tableName = cacheKey.split('.')[0]; try { const featureIds = []; const stream = await conn.send(` SELECT feature_id - FROM "${cacheKey}" + FROM "${tableName}" `); for await (const batch of stream) { @@ -84,7 +82,7 @@ export async function loadIndexData({ remoteUrl }) { const cacheKey = "index_data_table"; debugLog("loadIndexData called with cacheKey:", cacheKey); - const conn = await getConnection(); + const { conn } = await getConnection(); try { const tableName = cacheKey.replace(/"/g, '""'); @@ -106,7 +104,8 @@ export async function loadIndexData({ remoteUrl }) { await conn.query("INSTALL httpfs; LOAD httpfs;"); await conn.query("INSTALL parquet; LOAD parquet;"); - + await conn.query("SET enable_http_metadata_cache=true;"); + await conn.query(` CREATE TABLE "${tableName}" AS SELECT * FROM read_parquet('${remoteUrl}') @@ -122,12 +121,12 @@ export async function loadIndexData({ remoteUrl }) { export async function getFeatureProperties({ cacheKey, feature_id }) { debugLog("getFeature called with cacheKey:", cacheKey, "feature_id:", feature_id); - const conn = await getConnection(); - + const { conn } = await getConnection(); + const tableName = cacheKey.split('.')[0]; try { const stream = await conn.send(` SELECT * - FROM "${cacheKey}" + FROM "${tableName}" WHERE id = '${feature_id}' LIMIT 1 `); @@ -157,55 +156,35 @@ export async function getFeatureProperties({ cacheKey, feature_id }) { } } - export async function loadVpuData( cacheKey, prefix, - vpu_gpkg ) { debugLog("loadVpuData called with cacheKey:", cacheKey); - let buffer = await loadArrowFromCache(cacheKey); + let meta = await statFromCache(cacheKey); let fileSize; - if (!buffer) { - const ncFile = getNCFiles(prefix); - const res = await appAPI.getParquetPerVpu({ - ncFile, - vpu_gpkg, - }); - buffer = res; // ArrayBuffer from axios - fileSize = await saveArrowToCache(cacheKey, buffer); - } - const arrowTable = tableFromIPC(new Uint8Array(buffer)); - buffer = null; // this local reference can be cleared now - - const conn = await getConnection(); + if (!meta) { + fileSize = await saveDataToCache(cacheKey, prefix); + meta = await statFromCache(cacheKey); + if (!meta) throw new Error(`Saved to cache but can't stat file: ${cacheKey}`); + } else { + fileSize = formatBytes(meta.sizeBytes); + } + const { db, conn } = await getConnection(); try { - const existsResult = await conn.query(` - SELECT COUNT(*) AS cnt - FROM information_schema.tables - WHERE table_name = '${cacheKey}' - `); - - const exists = existsResult.toArray()[0].cnt > 0; - - if (!exists) { - await conn.insertArrowTable(arrowTable, { name: cacheKey }); - } else { - debugLog( - `Table "${cacheKey}" already exists, skipping insertArrowTable.` - ); - } + await createTableFromOPFS({ db, conn, key: cacheKey, safeName: meta.safeName }); } finally { await conn.close(); } + return fileSize; } export async function checkForTable(cacheKey) { - const conn = await getConnection(); + const { db, conn } = await getConnection(); try { const existsResult = await conn.query(` SELECT COUNT(*) AS cnt @@ -221,7 +200,7 @@ export async function checkForTable(cacheKey) { } export async function deleteTable(tableName){ - const conn = await getConnection(); + const { db, conn } = await getConnection(); try { await conn.query(` DROP TABLE IF EXISTS "${tableName}" @@ -233,7 +212,7 @@ export async function deleteTable(tableName){ } export async function dropAllVpuDataTables() { - const conn = await getConnection(); + const { db, conn } = await getConnection(); try { const result = await conn.query(` @@ -272,7 +251,7 @@ export async function dropAllVpuDataTables() { export async function getVariables({ cacheKey }) { debugLog("getVariables called with cacheKey:", cacheKey); - const conn = await getConnection(); + const { db, conn } = await getConnection(); try { const cols = []; @@ -300,7 +279,7 @@ export async function getVariables({ cacheKey }) { } export async function getDistinctFeatureIds(cacheKey) { - const conn = await getConnection(); + const { db, conn } = await getConnection(); try { const featureIds = []; const stream = await conn.send(` @@ -324,7 +303,7 @@ export async function getDistinctFeatureIds(cacheKey) { } export async function getDistinctTimes(cacheKey) { - const conn = await getConnection(); + const { db, conn } = await getConnection(); try { const times = []; const stream = await conn.send(` @@ -349,7 +328,7 @@ export async function getDistinctTimes(cacheKey) { // Returns a flattened array ordered by (feature_id, time) export async function getVpuVariableFlat(cacheKey, variable) { - const conn = await getConnection(); + const { db, conn } = await getConnection(); try { const countResult = await conn.query(` SELECT COUNT(*) AS n diff --git a/reactapp/features/Tethys/services/api/app.js b/reactapp/features/Tethys/services/api/app.js index bc17187..02a8bc7 100644 --- a/reactapp/features/Tethys/services/api/app.js +++ b/reactapp/features/Tethys/services/api/app.js @@ -3,15 +3,14 @@ import apiClient from "features/Tethys/services/api/client"; const APP_ROOT_URL = process.env.TETHYS_APP_ROOT_URL; const appAPI = { - getParquetPerVpu: (data) => { + getArrowPerVpu: (data) => { return apiClient.post( - `${APP_ROOT_URL}getParquetPerVpu/`, + `${APP_ROOT_URL}getArrowPerVpu/`, { ...data }, { responseType: "arraybuffer", // key point: binary, not JSON headers: { "Content-Type": "application/json"}, } - // { headers: { ...headers } } ); } diff --git a/tethysapp/nrds/controllers.py b/tethysapp/nrds/controllers.py index c1b3c2b..bb441c4 100644 --- a/tethysapp/nrds/controllers.py +++ b/tethysapp/nrds/controllers.py @@ -29,16 +29,13 @@ def home(request): @controller -def getParquetPerVpu(request): +def getArrowPerVpu(request): print("Getting parquet file per vpu...") file_prefix = json.loads(request.body.decode("utf-8"))['ncFile'] - vpu_gpkg = json.loads(request.body.decode("utf-8"))['vpu_gpkg'] print("file_prefix", file_prefix) - print("vpu_gpkg", vpu_gpkg) complete_df = convert_nc_2_df( s3_nc_url=file_prefix, - s3_gpkg_url=vpu_gpkg, ) table = pa.Table.from_pandas(complete_df) diff --git a/tethysapp/nrds/data_utils.py b/tethysapp/nrds/data_utils.py index 88ce063..4dbac0f 100644 --- a/tethysapp/nrds/data_utils.py +++ b/tethysapp/nrds/data_utils.py @@ -205,27 +205,21 @@ def get_usgs_nwm_xwalk_df() -> pd.DataFrame: return usgs_nwm_xwalk_df -def convert_nc_2_df(s3_nc_url: str, s3_gpkg_url: str) -> pd.DataFrame: +def convert_nc_2_df(s3_nc_url: str) -> pd.DataFrame: """Convert NetCDF files to Parquet format.""" df = get_troute_df(s3_nc_url) - ngen_usgs_gages = get_gages_from_hydrofabric_remote( - s3_gpkg_url, - anon=True, - ) + # Do we really need to fetch all the gages here? + # We need the gages to do the crosswalk merge and get the usgs_id and nwm_id columns in the final df, + # which is what the client needs. If we want to avoid fetching all the gages, we would need another way to get that info (e.g. a separate crosswalk file that maps feature_id to usgs_id and nwm_id without needing the full gage list). + # For now, we'll fetch the gages and do the merge here. + # ngen_usgs_gages = get_gages_from_hydrofabric_remote( + # s3_gpkg_url, + # anon=True, + # ) + # usgs_nwm_xwalk_df = get_usgs_nwm_xwalk_df() + # complete_df = merge_usgs_nwm30_crosswalk_nc(df,ngen_usgs_gages,usgs_nwm_xwalk_df) usgs_nwm_xwalk_df = get_usgs_nwm_xwalk_df() - complete_df = merge_usgs_nwm30_crosswalk_nc(df,ngen_usgs_gages,usgs_nwm_xwalk_df) + complete_df = merge_usgs_nwm30_crosswalk_nc(df,usgs_nwm_xwalk_df) return complete_df -def convert_df_2_bytes(df: pd.DataFrame) -> bytes: - """Convert NetCDF files to Parquet format.""" - - table = pa.Table.from_pandas(df) - buf = io.BytesIO() - with pa.ipc.new_stream(buf, table.schema) as writer: - writer.write_table(table) - - buf.seek(0) - return buf.read() - - From c50b0bf4865894b8d7f49d6d496b7c7c6af09b00 Mon Sep 17 00:00:00 2001 From: romer8 Date: Wed, 18 Feb 2026 18:23:12 -0700 Subject: [PATCH 2/6] refactor: enhance DuckDB integration and OPFS file handling; improve caching logic and debug output --- .../features/DataStream/lib/duckdbClient.js | 12 +-- reactapp/features/DataStream/lib/opfsCache.js | 79 ++++++++++++++++--- reactapp/features/DataStream/lib/queryData.js | 69 ++++++++++------ reactapp/features/DataStream/lib/s3Utils.js | 4 +- .../DataStream/views/DatastreamView.js | 7 +- tethysapp/nrds/data_utils.py | 6 +- 6 files changed, 130 insertions(+), 47 deletions(-) diff --git a/reactapp/features/DataStream/lib/duckdbClient.js b/reactapp/features/DataStream/lib/duckdbClient.js index 4c0abcd..1f48b36 100644 --- a/reactapp/features/DataStream/lib/duckdbClient.js +++ b/reactapp/features/DataStream/lib/duckdbClient.js @@ -1,7 +1,6 @@ import * as duckdb from "@duckdb/duckdb-wasm"; let dbPromise = null; - export function getDuckDB() { if (!dbPromise) { dbPromise = (async () => { @@ -19,6 +18,11 @@ export function getDuckDB() { const db = new duckdb.AsyncDuckDB(logger, worker); await db.instantiate(bundle.mainModule, bundle.pthreadWorker); + + await db.open({ + accessMode: duckdb.DuckDBAccessMode.READ_WRITE, + opfs: { fileHandling: "auto" }, + }); // Optional cleanup URL.revokeObjectURL(workerUrl); @@ -31,12 +35,8 @@ export function getDuckDB() { export async function getConnection() { const db = await getDuckDB(); - - // enable OPFS support once (safe to set repeatedly) - db.config.opfs = { fileHandling: "auto" }; - const conn = await db.connect(); - return { db, conn }; + return conn; } diff --git a/reactapp/features/DataStream/lib/opfsCache.js b/reactapp/features/DataStream/lib/opfsCache.js index af73739..26eb2a4 100644 --- a/reactapp/features/DataStream/lib/opfsCache.js +++ b/reactapp/features/DataStream/lib/opfsCache.js @@ -1,8 +1,10 @@ import appAPI from "features/Tethys/services/api/app"; import { tableFromIPC } from "apache-arrow"; import { getNCFiles } from "./s3Utils"; +import { DuckDBDataProtocol } from "@duckdb/duckdb-wasm"; +const BUCKET_NAME = "ciroh-community-ngen-datastream"; const CACHE_DIR = "nrds-arrow-cache"; let cacheDirPromise = null; @@ -68,16 +70,19 @@ async function saveArrowToCache(url, writable) { async function cacheParquetToOPFS(url, writable) { try { - const res = await fetch(url, { cache: "no-store" }); + const PARQUETURL=`https://ciroh-community-ngen-datastream.s3.us-east-1.amazonaws.com/${url}`; + const res = await fetch(PARQUETURL, { cache: "no-store" }); if (!res.ok) throw new Error(`Failed to fetch ${url}: ${res.status}`); // Stream to disk; avoids loading the entire file in memory if (!res.body) { const buf = await res.arrayBuffer(); + console.log(buf) await writable.write(new Uint8Array(buf)); await writable.close(); } else { // WritableStream from OPFS supports pipeTo in modern browsers + await res.body.pipeTo(writable); // pipeTo closes the destination by default } @@ -106,6 +111,7 @@ export async function saveDataToCache(key, url) { const safeName = encodeURIComponent(key); const fileHandle = await dir.getFileHandle(safeName, { create: true }); const writable = await fileHandle.createWritable(); + console.log(`Saving to cache with key: ${key}, url: ${url}`); if (isArrowFile(key)) { await saveArrowToCache(url, writable); } else { @@ -114,7 +120,33 @@ export async function saveDataToCache(key, url) { const file = await fileHandle.getFile(); return formatBytes(file.size); } +function ascii4(u8) { + return String.fromCharCode(...u8); +} +export async function inspectCachedFile(key) { + const dir = await getCacheDir(); + const safeName = encodeURIComponent(key); + const fh = await dir.getFileHandle(safeName); + const file = await fh.getFile(); + + const head = new Uint8Array(await file.slice(0, 4).arrayBuffer()); + const tail = new Uint8Array(await file.slice(Math.max(0, file.size - 4)).arrayBuffer()); + + console.log("OPFS file:", { + key, + safeName, + size: file.size, + headBytes: [...head], + headAscii: ascii4(head), + tailBytes: [...tail], + tailAscii: ascii4(tail), + }); + + // Optional: peek text (helpful if it’s an error page) + const preview = await file.slice(0, 200).text().catch(() => ""); + console.log("Preview (first 200 chars):", preview); +} export async function loadFromCache(key) { const dir = await getCacheDir(); @@ -142,19 +174,40 @@ async function doesTableExist(conn, tableName) { return res.toArray().length > 0; } -async function createTableFromOPFSParquet({ db, conn, key }) { - db.config.opfs = { fileHandling: "auto" }; - +// async function createTableFromOPFSParquet({ db, conn, key }) { +// const safeName = encodeURIComponent(key); +// const fileUrl = `opfs://${CACHE_DIR}/${safeName}`; +// const tableName = key.replace(/\.parquet$/i, ""); + +// await conn.query(` +// CREATE TABLE ${sqlIdent(tableName)} AS +// SELECT * FROM read_parquet(${sqlStr(fileUrl)}); +// `); +// } +async function createTableFromOPFSParquet({ conn, key }) { + // 1) Get the OPFS file handle from your cache directory + const cacheDir = await getCacheDir(); const safeName = encodeURIComponent(key); - const fileUrl = `opfs://${CACHE_DIR}/${safeName}`; - const tableName = key.replace(/\.parquet$/i, ""); + const fileHandle = await cacheDir.getFileHandle(safeName); + // 2) Register it in DuckDB under some virtual path/name + const duckPath = `${CACHE_DIR}/${safeName}`; // can be any string you like + const bindings = conn.bindings; // This is the AsyncDuckDB instance + + await bindings.registerFileHandle( + duckPath, + fileHandle, + DuckDBDataProtocol.BROWSER_FSACCESS, + true + ); + + // 3) Create table from that registered file name + const tableName = tableNameForKey(key); await conn.query(` CREATE TABLE ${sqlIdent(tableName)} AS - SELECT * FROM read_parquet(${sqlStr(fileUrl)}); + SELECT * FROM read_parquet(${sqlStr(duckPath)}); `); } - async function createTableFromOPFSArrow({ conn, key }) { const buffer = await loadFromCache(key); if (!buffer) throw new Error(`Arrow cache missing after save: ${key}`); @@ -165,7 +218,7 @@ async function createTableFromOPFSArrow({ conn, key }) { await conn.insertArrowTable(arrowTable, { name: tableName }); } -export async function createTableFromOPFS({ db, conn, key, safeName }) { +export async function createTableFromOPFS({ conn, key, safeName }) { const tableName = tableNameForKey(key); if (await doesTableExist(conn, tableName)) { @@ -177,7 +230,7 @@ export async function createTableFromOPFS({ db, conn, key, safeName }) { return createTableFromOPFSArrow({ conn, key }); } if (isParquetFile(key)) { - return createTableFromOPFSParquet({ db, conn, key, safeName }); + return createTableFromOPFSParquet({ conn, key, safeName }); } throw new Error(`Unsupported file type for key: ${key}`); @@ -193,7 +246,11 @@ export async function getFilesFromCache() { for await (const handle of dir.values()) { if (handle.kind !== "file") continue; const file = await handle.getFile(); - const id = decodeURIComponent(file.name.replace(".arrow", "") || file.name.replace(".parquet", "")); + // const id = decodeURIComponent(file.name.replace(".arrow", "") || file.name.replace(".parquet", "")); + const id = decodeURIComponent( + file.name.replace(/\.arrow$/i, "").replace(/\.parquet$/i, "") + ); + files.push({id: id, name: id.replaceAll("_", "/"), size: formatBytes(file.size)}); } return files; diff --git a/reactapp/features/DataStream/lib/queryData.js b/reactapp/features/DataStream/lib/queryData.js index f0ce5f5..8ccd319 100644 --- a/reactapp/features/DataStream/lib/queryData.js +++ b/reactapp/features/DataStream/lib/queryData.js @@ -1,7 +1,7 @@ // // nexusTimeseries.js -import { statFromCache, saveDataToCache, createTableFromOPFS, formatBytes } from "./opfsCache"; +import { statFromCache, saveDataToCache, createTableFromOPFS, formatBytes, inspectCachedFile } from "./opfsCache"; -import { getConnection } from "./duckdbClient"; +import { getConnection, getDuckDB } from "./duckdbClient"; const DEBUG = process.env.NODE_ENV !== "production"; const debugLog = (...args) => { @@ -9,7 +9,7 @@ const debugLog = (...args) => { }; export async function getTimeseries(id, cacheKey, variable) { - const { conn } = await getConnection(); + const conn = await getConnection(); const tableName = cacheKey.split('.')[0]; try { const rows = []; @@ -52,7 +52,7 @@ export async function getTimeseries(id, cacheKey, variable) { export async function getFeatureIDs(cacheKey) { debugLog("getFeatureIDs called with cacheKey:", cacheKey); - const { conn } = await getConnection(); + const conn = await getConnection(); const tableName = cacheKey.split('.')[0]; try { const featureIds = []; @@ -82,7 +82,7 @@ export async function loadIndexData({ remoteUrl }) { const cacheKey = "index_data_table"; debugLog("loadIndexData called with cacheKey:", cacheKey); - const { conn } = await getConnection(); + const conn = await getConnection(); try { const tableName = cacheKey.replace(/"/g, '""'); @@ -121,7 +121,7 @@ export async function loadIndexData({ remoteUrl }) { export async function getFeatureProperties({ cacheKey, feature_id }) { debugLog("getFeature called with cacheKey:", cacheKey, "feature_id:", feature_id); - const { conn } = await getConnection(); + const conn = await getConnection(); const tableName = cacheKey.split('.')[0]; try { const stream = await conn.send(` @@ -160,7 +160,7 @@ export async function loadVpuData( cacheKey, prefix, ) { - debugLog("loadVpuData called with cacheKey:", cacheKey); + debugLog("loadVpuData called with cacheKey:", cacheKey, "prefix:", prefix); let meta = await statFromCache(cacheKey); let fileSize; @@ -172,10 +172,10 @@ export async function loadVpuData( } else { fileSize = formatBytes(meta.sizeBytes); } - - const { db, conn } = await getConnection(); + inspectCachedFile(cacheKey) + const conn = await getConnection(); try { - await createTableFromOPFS({ db, conn, key: cacheKey, safeName: meta.safeName }); + await createTableFromOPFS({ conn, key: cacheKey, safeName: meta.safeName }); } finally { await conn.close(); } @@ -184,7 +184,7 @@ export async function loadVpuData( } export async function checkForTable(cacheKey) { - const { db, conn } = await getConnection(); + const conn = await getConnection(); try { const existsResult = await conn.query(` SELECT COUNT(*) AS cnt @@ -200,7 +200,7 @@ export async function checkForTable(cacheKey) { } export async function deleteTable(tableName){ - const { db, conn } = await getConnection(); + const conn = await getConnection(); try { await conn.query(` DROP TABLE IF EXISTS "${tableName}" @@ -212,7 +212,7 @@ export async function deleteTable(tableName){ } export async function dropAllVpuDataTables() { - const { db, conn } = await getConnection(); + const conn = await getConnection(); try { const result = await conn.query(` @@ -250,15 +250,15 @@ export async function dropAllVpuDataTables() { export async function getVariables({ cacheKey }) { debugLog("getVariables called with cacheKey:", cacheKey); - - const { db, conn } = await getConnection(); + const conn = await getConnection(); + const tableName = cacheKey.split('.')[0]; try { const cols = []; const stream = await conn.send(` SELECT column_name FROM information_schema.columns - WHERE table_name = '${cacheKey}' + WHERE table_name = '${tableName}' AND column_name NOT IN ( 'ngen_id', 'usgs_id', 'nwm_id', 'feature_id', 'time', 'type' ) @@ -279,15 +279,22 @@ export async function getVariables({ cacheKey }) { } export async function getDistinctFeatureIds(cacheKey) { - const { db, conn } = await getConnection(); + const conn = await getConnection(); + const tableName = cacheKey.split('.')[0]; try { const featureIds = []; + debugLog(`Getting distinct feature_ids from table "${tableName}"...`); + debugLog(` + SELECT DISTINCT feature_id + FROM "${tableName}" + ORDER BY feature_id + `); const stream = await conn.send(` SELECT DISTINCT feature_id - FROM "${cacheKey}" + FROM "${tableName}" ORDER BY feature_id `); - + for await (const batch of stream) { const ids = batch.getChild('feature_id'); if (!ids) continue; @@ -303,12 +310,19 @@ export async function getDistinctFeatureIds(cacheKey) { } export async function getDistinctTimes(cacheKey) { - const { db, conn } = await getConnection(); + const conn = await getConnection(); + const tableName = cacheKey.split('.')[0]; try { const times = []; + debugLog(`Getting distinct times from table "${cacheKey}"...`); + debugLog(` + SELECT DISTINCT time + FROM "${tableName}" + ORDER BY time + `); const stream = await conn.send(` SELECT DISTINCT time - FROM "${cacheKey}" + FROM "${tableName}" ORDER BY time `); @@ -328,11 +342,18 @@ export async function getDistinctTimes(cacheKey) { // Returns a flattened array ordered by (feature_id, time) export async function getVpuVariableFlat(cacheKey, variable) { - const { db, conn } = await getConnection(); + const conn = await getConnection(); + const tableName = cacheKey.split('.')[0]; try { + debugLog(`Getting variable "${variable}" data from table "${tableName}"...`); + debugLog(` + SELECT ${variable} AS v + FROM "${tableName}" + ORDER BY feature_id, time + `); const countResult = await conn.query(` SELECT COUNT(*) AS n - FROM "${cacheKey}" + FROM "${tableName}" `); const countCol = countResult.getChild('n'); const totalRows = Number(countCol?.get(0) ?? 0); @@ -345,7 +366,7 @@ export async function getVpuVariableFlat(cacheKey, variable) { const stream = await conn.send(` SELECT ${variable} AS v - FROM "${cacheKey}" + FROM "${tableName}" ORDER BY feature_id, time `); diff --git a/reactapp/features/DataStream/lib/s3Utils.js b/reactapp/features/DataStream/lib/s3Utils.js index 1966f14..fbdc5c3 100644 --- a/reactapp/features/DataStream/lib/s3Utils.js +++ b/reactapp/features/DataStream/lib/s3Utils.js @@ -62,7 +62,7 @@ export async function getOptionsFromURL(url, { signal } = {}) { try{ if (url.split('/').includes('troute')){ const files = await listPublicS3Files(url, { signal }); - const ncFiles = files.filter(f => f.endsWith('.nc')); + const ncFiles = files.filter(f => f.endsWith('.nc') || f.endsWith('.parquet')); // const ncFilesParsed = ncFiles.map(f => `s3://ciroh-community-ngen-datastream/${f}`); const options = ncFiles.map((d) => ({ value: d.split('/').pop(), label: d.split('/').pop() })); const sortedOptions = Array.from(options).sort().reverse(); @@ -97,6 +97,7 @@ export const makeGpkgUrl = (vpu) => { export const initialS3Data = async(vpu, { signal } = {}) => { try{ + console.log("Fetching initial S3 data with vpu:", vpu); let _models = await getOptionsFromURL(`outputs`, { signal }); if (_models.length === 0){ return {models: [], dates: [], forecasts: [], cycles: [], ensembles: [], outputFiles: []}; @@ -118,6 +119,7 @@ export const initialS3Data = async(vpu, { signal } = {}) => { return {models, dates, forecasts, cycles, ensembles:[], outputFiles: []}; } const outputFiles = await getOptionsFromURL(`outputs/${models[0]?.value}/v2.2_hydrofabric/${dates[1]?.value}/${forecasts[0]?.value}/${cycles[0]?.value}/${vpu}/ngen-run/outputs/troute/`, { signal }); + console.log("Fetched output files:", outputFiles); return {models, dates, forecasts, cycles, ensembles:[], outputFiles}; }catch(error){ throw error; diff --git a/reactapp/features/DataStream/views/DatastreamView.js b/reactapp/features/DataStream/views/DatastreamView.js index 86878e1..bb6a509 100644 --- a/reactapp/features/DataStream/views/DatastreamView.js +++ b/reactapp/features/DataStream/views/DatastreamView.js @@ -216,7 +216,7 @@ function TimeseriesLoader() { console.log('Loading VPU data for cacheKey:', cacheKey); reset(); resetVPU(); - const vpu_gpkg = makeGpkgUrl(vpu); + // const vpu_gpkg = makeGpkgUrl(vpu); set_loading(true); set_loading_text('Loading feature properties...'); let currentVariable = variable; @@ -226,7 +226,8 @@ function TimeseriesLoader() { if (!tableExists) { try{ - const fileSize = await loadVpuData(cacheKey, prefix, vpu_gpkg); + // const fileSize = await loadVpuData(cacheKey, prefix, vpu_gpkg); + const fileSize = await loadVpuData(cacheKey, prefix); if (!alive) return; add_cacheTable({id: cacheKey, name: cacheKey.replaceAll('_',' '), size: fileSize}); }catch(err){ @@ -237,9 +238,11 @@ function TimeseriesLoader() { } } const featureIDs = await getFeatureIDs(cacheKey); + console.log('Feature IDs loaded for cacheKey:', cacheKey, 'Count:', featureIDs.length); if (!alive) return; set_feature_ids(featureIDs); const variables = await getVariables({ cacheKey }); + console.log('Variables loaded for cacheKey:', cacheKey, 'Variables:', variables); if (!alive) return; set_variables(variables); set_variable(variables[0]); diff --git a/tethysapp/nrds/data_utils.py b/tethysapp/nrds/data_utils.py index 4dbac0f..fe298a4 100644 --- a/tethysapp/nrds/data_utils.py +++ b/tethysapp/nrds/data_utils.py @@ -219,7 +219,7 @@ def convert_nc_2_df(s3_nc_url: str) -> pd.DataFrame: # ) # usgs_nwm_xwalk_df = get_usgs_nwm_xwalk_df() # complete_df = merge_usgs_nwm30_crosswalk_nc(df,ngen_usgs_gages,usgs_nwm_xwalk_df) - usgs_nwm_xwalk_df = get_usgs_nwm_xwalk_df() - complete_df = merge_usgs_nwm30_crosswalk_nc(df,usgs_nwm_xwalk_df) - return complete_df + # usgs_nwm_xwalk_df = get_usgs_nwm_xwalk_df() + # complete_df = merge_usgs_nwm30_crosswalk_nc(df,usgs_nwm_xwalk_df) + return df From 967431152d44e99cc377c30e0e83bf202b767979 Mon Sep 17 00:00:00 2001 From: romer8 Date: Wed, 18 Feb 2026 19:31:27 -0700 Subject: [PATCH 3/6] refactor: remove unused BUCKET_NAME constant and simplify file name decoding in getFilesFromCache function --- reactapp/features/DataStream/lib/opfsCache.js | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/reactapp/features/DataStream/lib/opfsCache.js b/reactapp/features/DataStream/lib/opfsCache.js index 26eb2a4..ef7484d 100644 --- a/reactapp/features/DataStream/lib/opfsCache.js +++ b/reactapp/features/DataStream/lib/opfsCache.js @@ -4,7 +4,6 @@ import { getNCFiles } from "./s3Utils"; import { DuckDBDataProtocol } from "@duckdb/duckdb-wasm"; -const BUCKET_NAME = "ciroh-community-ngen-datastream"; const CACHE_DIR = "nrds-arrow-cache"; let cacheDirPromise = null; @@ -246,11 +245,7 @@ export async function getFilesFromCache() { for await (const handle of dir.values()) { if (handle.kind !== "file") continue; const file = await handle.getFile(); - // const id = decodeURIComponent(file.name.replace(".arrow", "") || file.name.replace(".parquet", "")); - const id = decodeURIComponent( - file.name.replace(/\.arrow$/i, "").replace(/\.parquet$/i, "") - ); - + const id = decodeURIComponent(file.name); files.push({id: id, name: id.replaceAll("_", "/"), size: formatBytes(file.size)}); } return files; From 921567c26f20502e10cae767022125d3d63bfde3 Mon Sep 17 00:00:00 2001 From: romer8 Date: Wed, 18 Feb 2026 19:34:33 -0700 Subject: [PATCH 4/6] refactor: remove console.log statements and clean up unused code in various components --- .../components/forecast/cacheTable.js | 5 +--- .../components/forecast/dataMenu.js | 2 -- .../DataStream/components/map/Mapg.js | 1 - .../DataStream/components/menus/CacheMenu.js | 1 - reactapp/features/DataStream/lib/opfsCache.js | 25 ------------------- reactapp/features/DataStream/lib/queryData.js | 5 ++-- reactapp/features/DataStream/lib/s3Utils.js | 2 -- .../DataStream/views/DatastreamView.js | 4 --- 8 files changed, 3 insertions(+), 42 deletions(-) diff --git a/reactapp/features/DataStream/components/forecast/cacheTable.js b/reactapp/features/DataStream/components/forecast/cacheTable.js index a730570..2cbf949 100644 --- a/reactapp/features/DataStream/components/forecast/cacheTable.js +++ b/reactapp/features/DataStream/components/forecast/cacheTable.js @@ -7,15 +7,13 @@ export const CacheTable = React.memo(({ tables }) => { const deleteCacheTable = useCacheTablesStore((state) => state.delete_cacheTable); const resetCacheTables = useCacheTablesStore((state) => state.reset); - // ✅ local loading state const [deletingAll, setDeletingAll] = useState(false); const [deletingId, setDeletingId] = useState(null); // table id currently deleting const deleteSingleCache = useCallback( async (tableId) => { if (deletingAll || deletingId) return; // avoid concurrent deletes - console.log("Delete cache table:", tableId); - setDeletingId(tableId); + setDeletingId(tableId); try { await deleteCacheTable(tableId); } finally { @@ -27,7 +25,6 @@ export const CacheTable = React.memo(({ tables }) => { const deleteAllCache = useCallback(async () => { if (deletingAll || deletingId) return; - console.log("Delete all cache tables"); setDeletingAll(true); try { await resetCacheTables(); diff --git a/reactapp/features/DataStream/components/forecast/dataMenu.js b/reactapp/features/DataStream/components/forecast/dataMenu.js index 1d77da9..a2a1dd3 100644 --- a/reactapp/features/DataStream/components/forecast/dataMenu.js +++ b/reactapp/features/DataStream/components/forecast/dataMenu.js @@ -150,11 +150,9 @@ const DataMenuControls = React.memo(function DataMenuControls() { } // reset(); const cacheKey = getCacheKey(model, date, forecast, cycle, ensemble, vpu, outputFile); - console.log('Generated cache key:', cacheKey); set_cache_key(cacheKey); const _prefix = makePrefix(model, date, forecast, cycle, ensemble, vpu, outputFile); - console.log('Generated S3 prefix:', _prefix); set_prefix(_prefix); }); diff --git a/reactapp/features/DataStream/components/map/Mapg.js b/reactapp/features/DataStream/components/map/Mapg.js index 8e6c217..3410459 100644 --- a/reactapp/features/DataStream/components/map/Mapg.js +++ b/reactapp/features/DataStream/components/map/Mapg.js @@ -119,7 +119,6 @@ const MainMap = () => { const deckLayers = useMemo(() => { if (!isFlowPathsVisible) return EMPTY_LAYERS; - // console.log('Rendering flow paths layer'); const varData = valuesByVar; const numTimes = timesArr?.length || 0; diff --git a/reactapp/features/DataStream/components/menus/CacheMenu.js b/reactapp/features/DataStream/components/menus/CacheMenu.js index 2f62d62..c6f5959 100644 --- a/reactapp/features/DataStream/components/menus/CacheMenu.js +++ b/reactapp/features/DataStream/components/menus/CacheMenu.js @@ -12,7 +12,6 @@ export const CacheMenu = () => { useEffect(() => { const fetchCacheTables = async () => { const files = await getFilesFromCache() - console.log("Fetched cache tables:", files); set_cacheTables(files); }; fetchCacheTables(); diff --git a/reactapp/features/DataStream/lib/opfsCache.js b/reactapp/features/DataStream/lib/opfsCache.js index ef7484d..6606d6a 100644 --- a/reactapp/features/DataStream/lib/opfsCache.js +++ b/reactapp/features/DataStream/lib/opfsCache.js @@ -76,7 +76,6 @@ async function cacheParquetToOPFS(url, writable) { // Stream to disk; avoids loading the entire file in memory if (!res.body) { const buf = await res.arrayBuffer(); - console.log(buf) await writable.write(new Uint8Array(buf)); await writable.close(); } else { @@ -110,7 +109,6 @@ export async function saveDataToCache(key, url) { const safeName = encodeURIComponent(key); const fileHandle = await dir.getFileHandle(safeName, { create: true }); const writable = await fileHandle.createWritable(); - console.log(`Saving to cache with key: ${key}, url: ${url}`); if (isArrowFile(key)) { await saveArrowToCache(url, writable); } else { @@ -123,29 +121,6 @@ function ascii4(u8) { return String.fromCharCode(...u8); } -export async function inspectCachedFile(key) { - const dir = await getCacheDir(); - const safeName = encodeURIComponent(key); - const fh = await dir.getFileHandle(safeName); - const file = await fh.getFile(); - - const head = new Uint8Array(await file.slice(0, 4).arrayBuffer()); - const tail = new Uint8Array(await file.slice(Math.max(0, file.size - 4)).arrayBuffer()); - - console.log("OPFS file:", { - key, - safeName, - size: file.size, - headBytes: [...head], - headAscii: ascii4(head), - tailBytes: [...tail], - tailAscii: ascii4(tail), - }); - - // Optional: peek text (helpful if it’s an error page) - const preview = await file.slice(0, 200).text().catch(() => ""); - console.log("Preview (first 200 chars):", preview); -} export async function loadFromCache(key) { const dir = await getCacheDir(); diff --git a/reactapp/features/DataStream/lib/queryData.js b/reactapp/features/DataStream/lib/queryData.js index 8ccd319..f7122d9 100644 --- a/reactapp/features/DataStream/lib/queryData.js +++ b/reactapp/features/DataStream/lib/queryData.js @@ -1,7 +1,7 @@ // // nexusTimeseries.js -import { statFromCache, saveDataToCache, createTableFromOPFS, formatBytes, inspectCachedFile } from "./opfsCache"; +import { statFromCache, saveDataToCache, createTableFromOPFS, formatBytes } from "./opfsCache"; -import { getConnection, getDuckDB } from "./duckdbClient"; +import { getConnection } from "./duckdbClient"; const DEBUG = process.env.NODE_ENV !== "production"; const debugLog = (...args) => { @@ -172,7 +172,6 @@ export async function loadVpuData( } else { fileSize = formatBytes(meta.sizeBytes); } - inspectCachedFile(cacheKey) const conn = await getConnection(); try { await createTableFromOPFS({ conn, key: cacheKey, safeName: meta.safeName }); diff --git a/reactapp/features/DataStream/lib/s3Utils.js b/reactapp/features/DataStream/lib/s3Utils.js index fbdc5c3..3ae5879 100644 --- a/reactapp/features/DataStream/lib/s3Utils.js +++ b/reactapp/features/DataStream/lib/s3Utils.js @@ -97,7 +97,6 @@ export const makeGpkgUrl = (vpu) => { export const initialS3Data = async(vpu, { signal } = {}) => { try{ - console.log("Fetching initial S3 data with vpu:", vpu); let _models = await getOptionsFromURL(`outputs`, { signal }); if (_models.length === 0){ return {models: [], dates: [], forecasts: [], cycles: [], ensembles: [], outputFiles: []}; @@ -119,7 +118,6 @@ export const initialS3Data = async(vpu, { signal } = {}) => { return {models, dates, forecasts, cycles, ensembles:[], outputFiles: []}; } const outputFiles = await getOptionsFromURL(`outputs/${models[0]?.value}/v2.2_hydrofabric/${dates[1]?.value}/${forecasts[0]?.value}/${cycles[0]?.value}/${vpu}/ngen-run/outputs/troute/`, { signal }); - console.log("Fetched output files:", outputFiles); return {models, dates, forecasts, cycles, ensembles:[], outputFiles}; }catch(error){ throw error; diff --git a/reactapp/features/DataStream/views/DatastreamView.js b/reactapp/features/DataStream/views/DatastreamView.js index bb6a509..dba824c 100644 --- a/reactapp/features/DataStream/views/DatastreamView.js +++ b/reactapp/features/DataStream/views/DatastreamView.js @@ -169,7 +169,6 @@ function TimeseriesLoader() { async function getTsData(){ if (!feature_id || loading ) return; - console.log('Loading timeseries for feature_id:', feature_id, 'variable:', variable, 'cacheKey:', cacheKey); reset_series(); const id = feature_id.split('-')[1]; set_loading(true); @@ -213,7 +212,6 @@ function TimeseriesLoader() { async function getVPUData(){ if (!cacheKey || loading ) return; - console.log('Loading VPU data for cacheKey:', cacheKey); reset(); resetVPU(); // const vpu_gpkg = makeGpkgUrl(vpu); @@ -238,11 +236,9 @@ function TimeseriesLoader() { } } const featureIDs = await getFeatureIDs(cacheKey); - console.log('Feature IDs loaded for cacheKey:', cacheKey, 'Count:', featureIDs.length); if (!alive) return; set_feature_ids(featureIDs); const variables = await getVariables({ cacheKey }); - console.log('Variables loaded for cacheKey:', cacheKey, 'Variables:', variables); if (!alive) return; set_variables(variables); set_variable(variables[0]); From 2e16b3b2143390f645ac449b7c205bd546a81cad Mon Sep 17 00:00:00 2001 From: romer8 Date: Wed, 18 Feb 2026 19:35:25 -0700 Subject: [PATCH 5/6] refactor: remove commented-out code and clarify the conversion process in convert_nc_2_df function --- tethysapp/nrds/data_utils.py | 13 ------------- 1 file changed, 13 deletions(-) diff --git a/tethysapp/nrds/data_utils.py b/tethysapp/nrds/data_utils.py index fe298a4..885f698 100644 --- a/tethysapp/nrds/data_utils.py +++ b/tethysapp/nrds/data_utils.py @@ -207,19 +207,6 @@ def get_usgs_nwm_xwalk_df() -> pd.DataFrame: def convert_nc_2_df(s3_nc_url: str) -> pd.DataFrame: """Convert NetCDF files to Parquet format.""" - df = get_troute_df(s3_nc_url) - # Do we really need to fetch all the gages here? - # We need the gages to do the crosswalk merge and get the usgs_id and nwm_id columns in the final df, - # which is what the client needs. If we want to avoid fetching all the gages, we would need another way to get that info (e.g. a separate crosswalk file that maps feature_id to usgs_id and nwm_id without needing the full gage list). - # For now, we'll fetch the gages and do the merge here. - # ngen_usgs_gages = get_gages_from_hydrofabric_remote( - # s3_gpkg_url, - # anon=True, - # ) - # usgs_nwm_xwalk_df = get_usgs_nwm_xwalk_df() - # complete_df = merge_usgs_nwm30_crosswalk_nc(df,ngen_usgs_gages,usgs_nwm_xwalk_df) - # usgs_nwm_xwalk_df = get_usgs_nwm_xwalk_df() - # complete_df = merge_usgs_nwm30_crosswalk_nc(df,usgs_nwm_xwalk_df) return df From b94cf2926e91ab5f91bac9b9aad7b1629a46a2e5 Mon Sep 17 00:00:00 2001 From: romer8 Date: Wed, 18 Feb 2026 21:08:43 -0700 Subject: [PATCH 6/6] refactor: rename CACHE_DIR to standardize cache directory naming --- reactapp/features/DataStream/lib/opfsCache.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/reactapp/features/DataStream/lib/opfsCache.js b/reactapp/features/DataStream/lib/opfsCache.js index 6606d6a..0fb2ee1 100644 --- a/reactapp/features/DataStream/lib/opfsCache.js +++ b/reactapp/features/DataStream/lib/opfsCache.js @@ -4,7 +4,7 @@ import { getNCFiles } from "./s3Utils"; import { DuckDBDataProtocol } from "@duckdb/duckdb-wasm"; -const CACHE_DIR = "nrds-arrow-cache"; +const CACHE_DIR = "nrds-cache"; let cacheDirPromise = null; export function formatBytes(bytes, decimals = 2) {