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/duckdbClient.js b/reactapp/features/DataStream/lib/duckdbClient.js index 4093e5b..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,9 +35,11 @@ export function getDuckDB() { export async function getConnection() { const db = await getDuckDB(); - return await db.connect(); + const conn = await db.connect(); + return 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..0fb2ee1 100644 --- a/reactapp/features/DataStream/lib/opfsCache.js +++ b/reactapp/features/DataStream/lib/opfsCache.js @@ -1,7 +1,13 @@ -const CACHE_DIR = "nrds-arrow-cache"; +import appAPI from "features/Tethys/services/api/app"; +import { tableFromIPC } from "apache-arrow"; +import { getNCFiles } from "./s3Utils"; +import { DuckDBDataProtocol } from "@duckdb/duckdb-wasm"; + + +const CACHE_DIR = "nrds-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 +36,97 @@ 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 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(); + 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); } +function ascii4(u8) { + return String.fromCharCode(...u8); +} -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 +136,82 @@ 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 }) { +// 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 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(duckPath)}); + `); +} +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({ 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({ 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 +220,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); 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 +262,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..f7122d9 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) => { @@ -12,18 +10,18 @@ const debugLog = (...args) => { export async function getTimeseries(id, cacheKey, variable) { 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 `); @@ -55,12 +53,12 @@ export async function getFeatureIDs(cacheKey) { debugLog("getFeatureIDs called with cacheKey:", cacheKey); 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) { @@ -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}') @@ -123,11 +122,11 @@ export async function getFeatureProperties({ cacheKey, feature_id }) { debugLog("getFeature called with cacheKey:", cacheKey, "feature_id:", feature_id); 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,50 +156,29 @@ export async function getFeatureProperties({ cacheKey, feature_id }) { } } - export async function loadVpuData( cacheKey, prefix, - vpu_gpkg ) { - debugLog("loadVpuData called with cacheKey:", cacheKey); + debugLog("loadVpuData called with cacheKey:", cacheKey, "prefix:", prefix); - 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 + 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 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({ conn, key: cacheKey, safeName: meta.safeName }); } finally { await conn.close(); } + return fileSize; } @@ -271,15 +249,15 @@ export async function dropAllVpuDataTables() { export async function getVariables({ cacheKey }) { debugLog("getVariables called with cacheKey:", cacheKey); - 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' ) @@ -300,15 +278,22 @@ export async function getVariables({ cacheKey }) { } export async function getDistinctFeatureIds(cacheKey) { - const 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; @@ -325,11 +310,18 @@ export async function getDistinctFeatureIds(cacheKey) { export async function getDistinctTimes(cacheKey) { 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 `); @@ -350,10 +342,17 @@ 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 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); @@ -366,7 +365,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..3ae5879 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(); diff --git a/reactapp/features/DataStream/views/DatastreamView.js b/reactapp/features/DataStream/views/DatastreamView.js index 86878e1..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,10 +212,9 @@ function TimeseriesLoader() { async function getVPUData(){ if (!cacheKey || loading ) return; - 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 +224,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){ 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..885f698 100644 --- a/tethysapp/nrds/data_utils.py +++ b/tethysapp/nrds/data_utils.py @@ -205,27 +205,8 @@ 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, - ) - usgs_nwm_xwalk_df = get_usgs_nwm_xwalk_df() - complete_df = merge_usgs_nwm30_crosswalk_nc(df,ngen_usgs_gages,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() - + return df