From ad84b572d17992958fe0f9e1cef2a451f032fd8e Mon Sep 17 00:00:00 2001 From: Teresa Burlingame Date: Fri, 3 Jul 2026 14:01:56 -0700 Subject: [PATCH 01/16] testing code to get enviroscan depth data to pub --- .../wrap.concH2oSalinity.grp.split.R | 60 +++++++++++++++- .../sensor_positions/sensor_positions_file.py | 26 +++++-- .../sensor_specific_processors.py | 71 +++++++++++++++++-- 3 files changed, 147 insertions(+), 10 deletions(-) diff --git a/flow/flow.concH2oSalinity.grp.split/wrap.concH2oSalinity.grp.split.R b/flow/flow.concH2oSalinity.grp.split/wrap.concH2oSalinity.grp.split.R index ef5a1b126b..33dfe61fc3 100644 --- a/flow/flow.concH2oSalinity.grp.split/wrap.concH2oSalinity.grp.split.R +++ b/flow/flow.concH2oSalinity.grp.split/wrap.concH2oSalinity.grp.split.R @@ -124,10 +124,12 @@ wrap.concH2oSalinity.grp.split <- function(DirIn, dirOutQm <- fs::path(dirOut, 'quality_metrics') dirOutGroup <- fs::path(base::dirname(dirOut), 'group') - # Create the CFGLOC-level data directories + # Create the CFGLOC-level data directories. 'location' holds the synthesized + # per-VER CFGLOC JSON consumed by pub_files (sensor positions), sibling to + # stats and quality_metrics so it lands under DATA_TYPE_INDEX in consolidate. NEONprocIS.base::def.dir.crea( DirBgn = dirOut, - DirSub = c('stats', 'quality_metrics'), + DirSub = c('stats', 'quality_metrics', 'location'), log = log ) # Create the group directory one level up (above the CFGLOC identity) @@ -218,6 +220,60 @@ wrap.concH2oSalinity.grp.split <- function(DirIn, # Process quality_metrics files processParquetDir(dirInQm, dirOutQm, depthStr, 'quality_metrics', SchmQm) + # Synthesize the per-VER CFGLOC location JSON consumed by pub_files. Depth + # for this VER lives in the stats INPUT as depth##SoilMoistureMean (added + # as a passthrough TermStat in stats_group_and_compute.yaml since depth is + # constant per day per VER). Read from dirInStats because selectRenameCols + # drops the column from output (case-sensitive grep won't match lowercase 'd'). + # Emit a CFGLOC-shaped JSON marked override_source=enviroscan so pub_files' + # is_enviroscan_sensor branch differentiates the 8 rows by JSON contents. + depthColName <- base::sprintf('depth%02dSoilMoistureMean', depthIdx) + depthValue <- NA_real_ + statsFiles <- base::list.files(dirInStats, pattern = '\\.parquet$', + full.names = TRUE, recursive = FALSE) + if (base::length(statsFiles) > 0) { + dfStats <- base::tryCatch( + arrow::read_parquet(statsFiles[1], as_data_frame = TRUE), + error = function(e) NULL + ) + if (!base::is.null(dfStats) && depthColName %in% base::names(dfStats)) { + vals <- dfStats[[depthColName]] + vals <- vals[!base::is.na(vals)] + if (base::length(vals) > 0) { + depthValue <- base::as.numeric(vals[1]) + } + } + } + + if (base::is.na(depthValue)) { + log$warn(base::paste0( + 'No non-NA depth value found in column ', depthColName, + ' for ', nameCfgloc, ' (VER=', VER, + '). Skipping location JSON emit; sensor_positions will not include this row.' + )) + } else { + locJson <- base::list( + type = 'FeatureCollection', + override_source = 'enviroscan', + features = base::list( + base::list( + type = 'Feature', + geometry = NA, + properties = base::list(name = nameCfgloc, site = site), + HOR = HOR, + VER = VER, + depth = depthValue + ) + ) + ) + fileOutLocJson <- fs::path(dirOut, 'location', paste0(nameCfgloc, '.json')) + base::writeLines(rjson::toJSON(locJson, indent = 2), con = fileOutLocJson) + log$info(base::paste0( + 'Wrote synthesized location JSON: ', fileOutLocJson, + ' (HOR=', HOR, ' VER=', VER, ' depth=', depthValue, ')' + )) + } + # Link only this datum's group JSON into the output group/ directory. fileOutGrpJson <- fs::path(dirOutGroup, paste0(nameCfgloc, '.json')) diff --git a/modules/pub_files/output_files/sensor_positions/sensor_positions_file.py b/modules/pub_files/output_files/sensor_positions/sensor_positions_file.py index f77694c184..4aa3455122 100644 --- a/modules/pub_files/output_files/sensor_positions/sensor_positions_file.py +++ b/modules/pub_files/output_files/sensor_positions/sensor_positions_file.py @@ -1,4 +1,5 @@ import csv +import json from datetime import datetime from pathlib import Path from typing import Tuple, List, Dict, Optional @@ -30,16 +31,33 @@ def write_file(out_path: Path, location_path: Path, elements: PathElements, time location = database.get_named_location(named_location_name) (row_hor_ver, row_location_id, row_description) = get_named_location_data(database, named_location_name) geolocations = database.get_geolocations(named_location_name) - + + # Read the JSON contents to detect per-VER overrides (e.g. EnviroSCAN + # writes one JSON per depth under distinct split-group dirs; the same + # CFGLOC name yields one DB row, so JSON contents differentiate rows). + location_json = None + try: + if path.stat().st_size > 0: + with open(path) as f: + location_json = json.load(f) + except (json.JSONDecodeError, OSError): + location_json = None + for geolocation in geolocations: # Use the specified processing method - if sensor_specific_processors.is_tchain_sensor(location): + if sensor_specific_processors.is_enviroscan_sensor(location_json): + rows = sensor_specific_processors.create_enviroscan_row( + location_json, geolocation, + row_location_id, row_description, + _create_base_row_data, _add_reference_position_data, + database) + elif sensor_specific_processors.is_tchain_sensor(location): rows = sensor_specific_processors.create_tchain_rows( - database, location, geolocation, row_hor_ver, + database, location, geolocation, row_hor_ver, row_location_id, row_description, _create_base_row_data, _add_reference_position_data) else: - rows = _create_standard_rows(database, geolocation, row_hor_ver, + rows = _create_standard_rows(database, geolocation, row_hor_ver, row_location_id, row_description) # Add rows, preventing duplicates diff --git a/modules/pub_files/output_files/sensor_positions/sensor_specific_processors.py b/modules/pub_files/output_files/sensor_positions/sensor_specific_processors.py index 0fd6030743..506bf5c0e4 100644 --- a/modules/pub_files/output_files/sensor_positions/sensor_specific_processors.py +++ b/modules/pub_files/output_files/sensor_positions/sensor_specific_processors.py @@ -1,6 +1,7 @@ """ Sensor-specific processing module. -Handles thermistor depth processing for tchain sensors. +Handles thermistor depth processing for tchain sensors and per-VER depth +overrides for EnviroSCAN (concH2oSoilSalinity). """ from typing import List, Dict, Optional from pub_files.output_files.sensor_positions.sensor_position import get_property @@ -37,10 +38,10 @@ def create_tchain_rows(database: SensorPositionsDatabase, location, geolocation, hor_ver_split = complete_row_data['row_hor_ver'].split('.') hor_ver_split[1] = thermistor_id modified_hor_ver = ".".join(hor_ver_split) - + # Adjust z_offset for thermistor depth modified_z_offset = round(complete_row_data['row_z_offset'] - depth, 2) - + tchain_row = [ modified_hor_ver, complete_row_data['row_location_id'], @@ -66,5 +67,67 @@ def create_tchain_rows(database: SensorPositionsDatabase, location, geolocation, complete_row_data['row_y_azimuth'] ] tchain_rows.append(tchain_row) - + return tchain_rows + + +def is_enviroscan_sensor(location_json: Optional[dict]) -> bool: + """Check if this location JSON was synthesized by group_split for EnviroSCAN. + + The R wrapper wrap.concH2oSalinity.grp.split.R writes one JSON per (CFGLOC, + VER) with override_source == 'enviroscan'. The shared CFGLOC has one DB row + so DB values alone would collapse the 8 depths to one; the JSON supplies + per-VER HOR/VER and depth (z-offset). + """ + if not location_json: + return False + return location_json.get('override_source') == 'enviroscan' + + +def create_enviroscan_row(location_json: dict, geolocation, + row_location_id: str, row_description: str, + create_base_row_data_func, add_reference_position_data_func, + database: SensorPositionsDatabase) -> List[List]: + """Build one row per DB geolocation history entry, with HOR/VER/z_offset + overridden from the synthesized location JSON.""" + feature = location_json.get('features', [{}])[0] or {} + hor = feature.get('HOR', '') + ver = feature.get('VER', '') + override_hor_ver = f'{hor}.{ver}' + depth = feature.get('depth') + + base_data = create_base_row_data_func(database, geolocation, override_hor_ver, + row_location_id, row_description) + if depth is not None: + base_data['row_z_offset'] = round(base_data['row_z_offset'] + float(depth), 2) + + complete_rows = add_reference_position_data_func(database, base_data, geolocation, + geolocation.offset_name) + + rows = [] + for row_data in complete_rows: + rows.append([ + row_data['row_hor_ver'], + row_data['row_location_id'], + row_data['row_description'], + row_data['row_position_start_date'], + row_data['row_position_end_date'], + row_data['row_reference_location_id'], + row_data['row_reference_location_description'], + row_data['row_reference_location_start_date'], + row_data['row_reference_location_end_date'], + row_data['row_x_offset'], + row_data['row_y_offset'], + row_data['row_z_offset'], + row_data['row_pitch'], + row_data['row_roll'], + row_data['row_azimuth'], + row_data['row_reference_location_latitude'], + row_data['row_reference_location_longitude'], + row_data['row_reference_location_elevation'], + row_data['row_east_offset'], + row_data['row_north_offset'], + row_data['row_x_azimuth'], + row_data['row_y_azimuth'] + ]) + return rows From 498d8cb487914e24f1d57d1b9a3eedc2c571b4b6 Mon Sep 17 00:00:00 2001 From: Teresa Burlingame Date: Mon, 6 Jul 2026 11:34:16 -0700 Subject: [PATCH 02/16] testing new logic to include changes within a day --- .../wrap.concH2oSalinity.grp.split.R | 162 +++++++++++++++--- .../sensor_positions/sensor_positions_file.py | 32 ++-- .../sensor_specific_processors.py | 148 +++++++++++----- 3 files changed, 260 insertions(+), 82 deletions(-) diff --git a/flow/flow.concH2oSalinity.grp.split/wrap.concH2oSalinity.grp.split.R b/flow/flow.concH2oSalinity.grp.split/wrap.concH2oSalinity.grp.split.R index 33dfe61fc3..d3ddb7e4c4 100644 --- a/flow/flow.concH2oSalinity.grp.split/wrap.concH2oSalinity.grp.split.R +++ b/flow/flow.concH2oSalinity.grp.split/wrap.concH2oSalinity.grp.split.R @@ -221,33 +221,152 @@ wrap.concH2oSalinity.grp.split <- function(DirIn, processParquetDir(dirInQm, dirOutQm, depthStr, 'quality_metrics', SchmQm) # Synthesize the per-VER CFGLOC location JSON consumed by pub_files. Depth - # for this VER lives in the stats INPUT as depth##SoilMoistureMean (added - # as a passthrough TermStat in stats_group_and_compute.yaml since depth is - # constant per day per VER). Read from dirInStats because selectRenameCols - # drops the column from output (case-sensitive grep won't match lowercase 'd'). - # Emit a CFGLOC-shaped JSON marked override_source=enviroscan so pub_files' - # is_enviroscan_sensor branch differentiates the 8 rows by JSON contents. + # for this VER lives in the stats INPUT as depth##SoilMoistureMean (added as + # a passthrough TermStat in stats_group_and_compute since cal-driven depth + # varies within a day when there are mid-day cal changes or sensor swaps). + # Read from dirInStats because selectRenameCols drops the column from output + # (case-sensitive grep won't match lowercase 'd'). Emit a CFGLOC-shaped JSON + # marked override_source=enviroscan so pub_files' is_enviroscan_sensor branch + # differentiates the 8 rows by JSON contents. + # + # Segmentation (5-min bins, "both neighbors match" persistence rule): + # - Prefer the 5-min stats parquet (_005.parquet), else shortest bin. + # - A bin is 'confirmed' if both temporal neighbors share its value + # (edge bins fall back to single-neighbor match; single-bin day: non-NA). + # - Unconfirmed bins define the ambiguous gap between adjacent segments. + # - Adjacent segments meet at the midpoint of that gap (no positions gap). + # - First segment starts at dayStart; last segment ends at dayEnd; pub_files + # merges day-adjacent same-(HOR,VER,depth) features into contiguous rows. depthColName <- base::sprintf('depth%02dSoilMoistureMean', depthIdx) - depthValue <- NA_real_ + statsFiles <- base::list.files(dirInStats, pattern = '\\.parquet$', full.names = TRUE, recursive = FALSE) + + # Prefer explicit 5-min files by conventional NEON naming (..._005.parquet); + # fall back to the file with the shortest median bin duration. + fileFinest <- NULL if (base::length(statsFiles) > 0) { + filesByName <- statsFiles[base::grepl('_005\\.parquet$', statsFiles)] + if (base::length(filesByName) > 0) { + fileFinest <- filesByName[1] + } else { + minBin <- Inf + for (f in statsFiles) { + dfTry <- base::tryCatch(arrow::read_parquet(f, as_data_frame = TRUE), + error = function(e) NULL) + if (!base::is.null(dfTry) && + all(c('startDateTime','endDateTime') %in% base::names(dfTry)) && + base::nrow(dfTry) >= 1) { + bin <- stats::median(as.numeric(difftime(dfTry$endDateTime, + dfTry$startDateTime, + units = 'secs')), na.rm = TRUE) + if (!is.na(bin) && bin < minBin) { minBin <- bin; fileFinest <- f } + } + } + } + } + + features <- base::list() + + if (!base::is.null(fileFinest)) { dfStats <- base::tryCatch( - arrow::read_parquet(statsFiles[1], as_data_frame = TRUE), + arrow::read_parquet(fileFinest, as_data_frame = TRUE), error = function(e) NULL ) - if (!base::is.null(dfStats) && depthColName %in% base::names(dfStats)) { - vals <- dfStats[[depthColName]] - vals <- vals[!base::is.na(vals)] - if (base::length(vals) > 0) { - depthValue <- base::as.numeric(vals[1]) + if (!base::is.null(dfStats) && + depthColName %in% base::names(dfStats) && + all(c('startDateTime','endDateTime') %in% base::names(dfStats)) && + base::nrow(dfStats) > 0) { + ord <- base::order(dfStats$startDateTime) + dfStats <- dfStats[ord, , drop = FALSE] + # Round to 4 decimals for equality comparisons (cal depths are ~cm). + vR <- base::round(base::as.numeric(dfStats[[depthColName]]), 4) + s <- dfStats$startDateTime + e <- dfStats$endDateTime + n <- base::length(vR) + + confirmed <- base::rep(FALSE, n) + if (n == 1) { + confirmed[1] <- !base::is.na(vR[1]) + } else { + for (i in seq_len(n)) { + vi <- vR[i] + if (base::is.na(vi)) next + if (i == 1) { + confirmed[i] <- !base::is.na(vR[2]) && (vR[2] == vi) + } else if (i == n) { + confirmed[i] <- !base::is.na(vR[i-1]) && (vR[i-1] == vi) + } else { + confirmed[i] <- !base::is.na(vR[i-1]) && !base::is.na(vR[i+1]) && + (vR[i-1] == vi) && (vR[i+1] == vi) + } + } + } + + # Walk confirmed bins into runs of matching value. + segs <- base::list() + curVal <- NA_real_; curBinStart <- NA; curBinEnd <- NA + for (i in seq_len(n)) { + if (!confirmed[i]) next + if (base::is.na(curVal)) { + curVal <- vR[i]; curBinStart <- s[i]; curBinEnd <- e[i] + } else if (vR[i] == curVal) { + curBinEnd <- e[i] + } else { + segs[[length(segs) + 1]] <- base::list(value = curVal, + binStart = curBinStart, + binEnd = curBinEnd) + curVal <- vR[i]; curBinStart <- s[i]; curBinEnd <- e[i] + } + } + if (!base::is.na(curVal)) { + segs[[length(segs) + 1]] <- base::list(value = curVal, + binStart = curBinStart, + binEnd = curBinEnd) + } + + # Day-scoped bounds (from datum path, not data — full-day coverage even + # if the parquet has NA bins at day edges). InfoDirIn$time is UTC midnight. + dayStart <- InfoDirIn$time + base::attr(dayStart, 'tzone') <- 'UTC' + dayEnd <- dayStart + 86400 + + nSeg <- base::length(segs) + crossovers <- base::vector('list', base::max(0, nSeg - 1)) + if (nSeg >= 2) { + for (k in seq_len(nSeg - 1)) { + a <- segs[[k]]; b <- segs[[k + 1]] + gapSec <- base::as.numeric(difftime(b$binStart, a$binEnd, units = 'secs')) + crossovers[[k]] <- a$binEnd + gapSec / 2 + } + } + + isoFmt <- function(x) { + base::attr(x, 'tzone') <- 'UTC' + base::format(x, '%Y-%m-%dT%H:%M:%SZ') + } + + for (k in seq_len(nSeg)) { + segK <- segs[[k]] + segStart <- if (k == 1) dayStart else crossovers[[k - 1]] + segEnd <- if (k == nSeg) dayEnd else crossovers[[k]] + features[[length(features) + 1]] <- base::list( + type = 'Feature', + geometry = NA, + properties = base::list(name = nameCfgloc, site = site), + HOR = HOR, + VER = VER, + depth = segK$value, + positionStartDateTime = isoFmt(segStart), + positionEndDateTime = isoFmt(segEnd) + ) } } } - if (base::is.na(depthValue)) { + if (base::length(features) == 0) { log$warn(base::paste0( - 'No non-NA depth value found in column ', depthColName, + 'No confirmed depth segments found in column ', depthColName, ' for ', nameCfgloc, ' (VER=', VER, '). Skipping location JSON emit; sensor_positions will not include this row.' )) @@ -255,22 +374,13 @@ wrap.concH2oSalinity.grp.split <- function(DirIn, locJson <- base::list( type = 'FeatureCollection', override_source = 'enviroscan', - features = base::list( - base::list( - type = 'Feature', - geometry = NA, - properties = base::list(name = nameCfgloc, site = site), - HOR = HOR, - VER = VER, - depth = depthValue - ) - ) + features = features ) fileOutLocJson <- fs::path(dirOut, 'location', paste0(nameCfgloc, '.json')) base::writeLines(rjson::toJSON(locJson, indent = 2), con = fileOutLocJson) log$info(base::paste0( 'Wrote synthesized location JSON: ', fileOutLocJson, - ' (HOR=', HOR, ' VER=', VER, ' depth=', depthValue, ')' + ' (HOR=', HOR, ' VER=', VER, ' segments=', base::length(features), ')' )) } diff --git a/modules/pub_files/output_files/sensor_positions/sensor_positions_file.py b/modules/pub_files/output_files/sensor_positions/sensor_positions_file.py index 4aa3455122..cc43907db5 100644 --- a/modules/pub_files/output_files/sensor_positions/sensor_positions_file.py +++ b/modules/pub_files/output_files/sensor_positions/sensor_positions_file.py @@ -23,6 +23,10 @@ def write_file(out_path: Path, location_path: Path, elements: PathElements, time writer = csv.writer(file) writer.writerow(get_column_names()) file_rows = [] + # EnviroSCAN features accumulate across the monthly datum's days; + # merged into rows after the glob loop so day-adjacent same-depth + # segments collapse into contiguous ranges. + enviroscan_features = [] # Parse location file path for the datum elements. Assume we end at site (**/site/location/*/location_file.json) site = location_path.parts[-1] for path in location_path.parent.parent.rglob(f'*/{site}/location/*/*.json'): @@ -43,15 +47,15 @@ def write_file(out_path: Path, location_path: Path, elements: PathElements, time except (json.JSONDecodeError, OSError): location_json = None + if sensor_specific_processors.is_enviroscan_sensor(location_json): + sensor_specific_processors.collect_enviroscan_features( + location_json, named_location_name, + row_location_id, row_description, geolocations, + enviroscan_features) + continue + for geolocation in geolocations: - # Use the specified processing method - if sensor_specific_processors.is_enviroscan_sensor(location_json): - rows = sensor_specific_processors.create_enviroscan_row( - location_json, geolocation, - row_location_id, row_description, - _create_base_row_data, _add_reference_position_data, - database) - elif sensor_specific_processors.is_tchain_sensor(location): + if sensor_specific_processors.is_tchain_sensor(location): rows = sensor_specific_processors.create_tchain_rows( database, location, geolocation, row_hor_ver, row_location_id, row_description, @@ -59,12 +63,20 @@ def write_file(out_path: Path, location_path: Path, elements: PathElements, time else: rows = _create_standard_rows(database, geolocation, row_hor_ver, row_location_id, row_description) - + # Add rows, preventing duplicates for row in rows: if row not in file_rows: file_rows.append(row) - + + # Merge collected enviroscan features across days and emit rows. + enviro_rows = sensor_specific_processors.aggregate_enviroscan_rows( + enviroscan_features, database, + _create_base_row_data, _add_reference_position_data) + for row in enviro_rows: + if row not in file_rows: + file_rows.append(row) + writer.writerows(file_rows) return file_path diff --git a/modules/pub_files/output_files/sensor_positions/sensor_specific_processors.py b/modules/pub_files/output_files/sensor_positions/sensor_specific_processors.py index 506bf5c0e4..9d0d43a13b 100644 --- a/modules/pub_files/output_files/sensor_positions/sensor_specific_processors.py +++ b/modules/pub_files/output_files/sensor_positions/sensor_specific_processors.py @@ -3,6 +3,7 @@ Handles thermistor depth processing for tchain sensors and per-VER depth overrides for EnviroSCAN (concH2oSoilSalinity). """ +from collections import defaultdict from typing import List, Dict, Optional from pub_files.output_files.sensor_positions.sensor_position import get_property from pub_files.output_files.sensor_positions.sensor_positions_database import SensorPositionsDatabase @@ -84,50 +85,105 @@ def is_enviroscan_sensor(location_json: Optional[dict]) -> bool: return location_json.get('override_source') == 'enviroscan' -def create_enviroscan_row(location_json: dict, geolocation, - row_location_id: str, row_description: str, - create_base_row_data_func, add_reference_position_data_func, - database: SensorPositionsDatabase) -> List[List]: - """Build one row per DB geolocation history entry, with HOR/VER/z_offset - overridden from the synthesized location JSON.""" - feature = location_json.get('features', [{}])[0] or {} - hor = feature.get('HOR', '') - ver = feature.get('VER', '') - override_hor_ver = f'{hor}.{ver}' - depth = feature.get('depth') - - base_data = create_base_row_data_func(database, geolocation, override_hor_ver, - row_location_id, row_description) - if depth is not None: - base_data['row_z_offset'] = round(base_data['row_z_offset'] + float(depth), 2) - - complete_rows = add_reference_position_data_func(database, base_data, geolocation, - geolocation.offset_name) - - rows = [] - for row_data in complete_rows: - rows.append([ - row_data['row_hor_ver'], - row_data['row_location_id'], - row_data['row_description'], - row_data['row_position_start_date'], - row_data['row_position_end_date'], - row_data['row_reference_location_id'], - row_data['row_reference_location_description'], - row_data['row_reference_location_start_date'], - row_data['row_reference_location_end_date'], - row_data['row_x_offset'], - row_data['row_y_offset'], - row_data['row_z_offset'], - row_data['row_pitch'], - row_data['row_roll'], - row_data['row_azimuth'], - row_data['row_reference_location_latitude'], - row_data['row_reference_location_longitude'], - row_data['row_reference_location_elevation'], - row_data['row_east_offset'], - row_data['row_north_offset'], - row_data['row_x_azimuth'], - row_data['row_y_azimuth'] - ]) +def collect_enviroscan_features(location_json: dict, cfgloc: str, + row_location_id: str, row_description: str, + geolocations, sink: List[Dict]) -> None: + """Append each feature in the JSON to a shared collection pool for + later aggregation across days. group_split emits one JSON per (CFGLOC, + VER) per day with N features covering N depth segments within that day; + pub_files sees all monthly datum days at once and merges day-adjacent + same-(cfgloc, HOR, VER, depth) features into single rows.""" + for feature in location_json.get('features', []) or []: + if not feature: + continue + depth = feature.get('depth') + if depth is None: + continue + sink.append({ + 'cfgloc': cfgloc, + 'hor': feature.get('HOR', ''), + 'ver': feature.get('VER', ''), + 'depth': float(depth), + 'position_start_date': feature.get('positionStartDateTime', ''), + 'position_end_date': feature.get('positionEndDateTime', ''), + 'row_location_id': row_location_id, + 'row_description': row_description, + 'geolocations': geolocations, + }) + + +def aggregate_enviroscan_rows(collected: List[Dict], + database: SensorPositionsDatabase, + create_base_row_data_func, + add_reference_position_data_func) -> List[List]: + """Merge day-adjacent same-(cfgloc, HOR, VER, depth) features into + contiguous ranges, emit one row per merged range per DB geolocation. + Adjacent day-runs are considered contiguous when prev.end == next.start + (group_split ensures this: day N ends at UTC midnight, day N+1 starts + at the same instant; transition-day crossovers fall in the middle of + the ambiguous zone). z_offset overridden as DB_z + segment depth.""" + if not collected: + return [] + + # Group by (cfgloc, HOR, VER, rounded depth). Depths compared to 4 + # decimals so cal-file jitter doesn't fragment same-depth runs. + groups: Dict[tuple, List[Dict]] = defaultdict(list) + for f in collected: + key = (f['cfgloc'], f['hor'], f['ver'], round(f['depth'], 4)) + groups[key].append(f) + + rows: List[List] = [] + for (cfgloc, hor, ver, depth), feats in groups.items(): + feats.sort(key=lambda x: x['position_start_date']) + + # Merge contiguous ISO-Z strings (group_split guarantees exact match + # at day boundaries — no timezone or precision drift on the seam). + merged: List[Dict] = [dict(feats[0])] + for f in feats[1:]: + if merged[-1]['position_end_date'] == f['position_start_date']: + merged[-1]['position_end_date'] = f['position_end_date'] + else: + merged.append(dict(f)) + + override_hor_ver = f'{hor}.{ver}' + for m in merged: + for geolocation in m['geolocations']: + base_data = create_base_row_data_func( + database, geolocation, override_hor_ver, + m['row_location_id'], m['row_description']) + base_data['row_z_offset'] = round( + base_data['row_z_offset'] + float(depth), 2) + # Segment range overrides DB geolocation dates for the + # positionStart/EndDateTime CSV columns. + base_data['row_position_start_date'] = m['position_start_date'] + base_data['row_position_end_date'] = m['position_end_date'] + + complete_rows = add_reference_position_data_func( + database, base_data, geolocation, geolocation.offset_name) + + for row_data in complete_rows: + rows.append([ + row_data['row_hor_ver'], + row_data['row_location_id'], + row_data['row_description'], + row_data['row_position_start_date'], + row_data['row_position_end_date'], + row_data['row_reference_location_id'], + row_data['row_reference_location_description'], + row_data['row_reference_location_start_date'], + row_data['row_reference_location_end_date'], + row_data['row_x_offset'], + row_data['row_y_offset'], + row_data['row_z_offset'], + row_data['row_pitch'], + row_data['row_roll'], + row_data['row_azimuth'], + row_data['row_reference_location_latitude'], + row_data['row_reference_location_longitude'], + row_data['row_reference_location_elevation'], + row_data['row_east_offset'], + row_data['row_north_offset'], + row_data['row_x_azimuth'], + row_data['row_y_azimuth'] + ]) return rows From 8ec113147d5b6f967d668402409ea29ad4f1b82f Mon Sep 17 00:00:00 2001 From: Teresa Burlingame Date: Tue, 7 Jul 2026 08:36:12 -0700 Subject: [PATCH 03/16] Revert segmentation approach for enviroscan sensor_positions Reverts commits 498d8cb48 and ad84b572d. Depth per VER is calibration- driven (CVALD1 in calibration_metadata), not data-driven, so segmenting depth##SoilMoistureMean from the 5-min stats parquet was the wrong path. Restarting with a plan to join calibration by asset_uid at pub time. --- .../wrap.concH2oSalinity.grp.split.R | 170 +----------------- .../sensor_positions/sensor_positions_file.py | 42 +---- .../sensor_specific_processors.py | 127 +------------ 3 files changed, 12 insertions(+), 327 deletions(-) diff --git a/flow/flow.concH2oSalinity.grp.split/wrap.concH2oSalinity.grp.split.R b/flow/flow.concH2oSalinity.grp.split/wrap.concH2oSalinity.grp.split.R index d3ddb7e4c4..ef5a1b126b 100644 --- a/flow/flow.concH2oSalinity.grp.split/wrap.concH2oSalinity.grp.split.R +++ b/flow/flow.concH2oSalinity.grp.split/wrap.concH2oSalinity.grp.split.R @@ -124,12 +124,10 @@ wrap.concH2oSalinity.grp.split <- function(DirIn, dirOutQm <- fs::path(dirOut, 'quality_metrics') dirOutGroup <- fs::path(base::dirname(dirOut), 'group') - # Create the CFGLOC-level data directories. 'location' holds the synthesized - # per-VER CFGLOC JSON consumed by pub_files (sensor positions), sibling to - # stats and quality_metrics so it lands under DATA_TYPE_INDEX in consolidate. + # Create the CFGLOC-level data directories NEONprocIS.base::def.dir.crea( DirBgn = dirOut, - DirSub = c('stats', 'quality_metrics', 'location'), + DirSub = c('stats', 'quality_metrics'), log = log ) # Create the group directory one level up (above the CFGLOC identity) @@ -220,170 +218,6 @@ wrap.concH2oSalinity.grp.split <- function(DirIn, # Process quality_metrics files processParquetDir(dirInQm, dirOutQm, depthStr, 'quality_metrics', SchmQm) - # Synthesize the per-VER CFGLOC location JSON consumed by pub_files. Depth - # for this VER lives in the stats INPUT as depth##SoilMoistureMean (added as - # a passthrough TermStat in stats_group_and_compute since cal-driven depth - # varies within a day when there are mid-day cal changes or sensor swaps). - # Read from dirInStats because selectRenameCols drops the column from output - # (case-sensitive grep won't match lowercase 'd'). Emit a CFGLOC-shaped JSON - # marked override_source=enviroscan so pub_files' is_enviroscan_sensor branch - # differentiates the 8 rows by JSON contents. - # - # Segmentation (5-min bins, "both neighbors match" persistence rule): - # - Prefer the 5-min stats parquet (_005.parquet), else shortest bin. - # - A bin is 'confirmed' if both temporal neighbors share its value - # (edge bins fall back to single-neighbor match; single-bin day: non-NA). - # - Unconfirmed bins define the ambiguous gap between adjacent segments. - # - Adjacent segments meet at the midpoint of that gap (no positions gap). - # - First segment starts at dayStart; last segment ends at dayEnd; pub_files - # merges day-adjacent same-(HOR,VER,depth) features into contiguous rows. - depthColName <- base::sprintf('depth%02dSoilMoistureMean', depthIdx) - - statsFiles <- base::list.files(dirInStats, pattern = '\\.parquet$', - full.names = TRUE, recursive = FALSE) - - # Prefer explicit 5-min files by conventional NEON naming (..._005.parquet); - # fall back to the file with the shortest median bin duration. - fileFinest <- NULL - if (base::length(statsFiles) > 0) { - filesByName <- statsFiles[base::grepl('_005\\.parquet$', statsFiles)] - if (base::length(filesByName) > 0) { - fileFinest <- filesByName[1] - } else { - minBin <- Inf - for (f in statsFiles) { - dfTry <- base::tryCatch(arrow::read_parquet(f, as_data_frame = TRUE), - error = function(e) NULL) - if (!base::is.null(dfTry) && - all(c('startDateTime','endDateTime') %in% base::names(dfTry)) && - base::nrow(dfTry) >= 1) { - bin <- stats::median(as.numeric(difftime(dfTry$endDateTime, - dfTry$startDateTime, - units = 'secs')), na.rm = TRUE) - if (!is.na(bin) && bin < minBin) { minBin <- bin; fileFinest <- f } - } - } - } - } - - features <- base::list() - - if (!base::is.null(fileFinest)) { - dfStats <- base::tryCatch( - arrow::read_parquet(fileFinest, as_data_frame = TRUE), - error = function(e) NULL - ) - if (!base::is.null(dfStats) && - depthColName %in% base::names(dfStats) && - all(c('startDateTime','endDateTime') %in% base::names(dfStats)) && - base::nrow(dfStats) > 0) { - ord <- base::order(dfStats$startDateTime) - dfStats <- dfStats[ord, , drop = FALSE] - # Round to 4 decimals for equality comparisons (cal depths are ~cm). - vR <- base::round(base::as.numeric(dfStats[[depthColName]]), 4) - s <- dfStats$startDateTime - e <- dfStats$endDateTime - n <- base::length(vR) - - confirmed <- base::rep(FALSE, n) - if (n == 1) { - confirmed[1] <- !base::is.na(vR[1]) - } else { - for (i in seq_len(n)) { - vi <- vR[i] - if (base::is.na(vi)) next - if (i == 1) { - confirmed[i] <- !base::is.na(vR[2]) && (vR[2] == vi) - } else if (i == n) { - confirmed[i] <- !base::is.na(vR[i-1]) && (vR[i-1] == vi) - } else { - confirmed[i] <- !base::is.na(vR[i-1]) && !base::is.na(vR[i+1]) && - (vR[i-1] == vi) && (vR[i+1] == vi) - } - } - } - - # Walk confirmed bins into runs of matching value. - segs <- base::list() - curVal <- NA_real_; curBinStart <- NA; curBinEnd <- NA - for (i in seq_len(n)) { - if (!confirmed[i]) next - if (base::is.na(curVal)) { - curVal <- vR[i]; curBinStart <- s[i]; curBinEnd <- e[i] - } else if (vR[i] == curVal) { - curBinEnd <- e[i] - } else { - segs[[length(segs) + 1]] <- base::list(value = curVal, - binStart = curBinStart, - binEnd = curBinEnd) - curVal <- vR[i]; curBinStart <- s[i]; curBinEnd <- e[i] - } - } - if (!base::is.na(curVal)) { - segs[[length(segs) + 1]] <- base::list(value = curVal, - binStart = curBinStart, - binEnd = curBinEnd) - } - - # Day-scoped bounds (from datum path, not data — full-day coverage even - # if the parquet has NA bins at day edges). InfoDirIn$time is UTC midnight. - dayStart <- InfoDirIn$time - base::attr(dayStart, 'tzone') <- 'UTC' - dayEnd <- dayStart + 86400 - - nSeg <- base::length(segs) - crossovers <- base::vector('list', base::max(0, nSeg - 1)) - if (nSeg >= 2) { - for (k in seq_len(nSeg - 1)) { - a <- segs[[k]]; b <- segs[[k + 1]] - gapSec <- base::as.numeric(difftime(b$binStart, a$binEnd, units = 'secs')) - crossovers[[k]] <- a$binEnd + gapSec / 2 - } - } - - isoFmt <- function(x) { - base::attr(x, 'tzone') <- 'UTC' - base::format(x, '%Y-%m-%dT%H:%M:%SZ') - } - - for (k in seq_len(nSeg)) { - segK <- segs[[k]] - segStart <- if (k == 1) dayStart else crossovers[[k - 1]] - segEnd <- if (k == nSeg) dayEnd else crossovers[[k]] - features[[length(features) + 1]] <- base::list( - type = 'Feature', - geometry = NA, - properties = base::list(name = nameCfgloc, site = site), - HOR = HOR, - VER = VER, - depth = segK$value, - positionStartDateTime = isoFmt(segStart), - positionEndDateTime = isoFmt(segEnd) - ) - } - } - } - - if (base::length(features) == 0) { - log$warn(base::paste0( - 'No confirmed depth segments found in column ', depthColName, - ' for ', nameCfgloc, ' (VER=', VER, - '). Skipping location JSON emit; sensor_positions will not include this row.' - )) - } else { - locJson <- base::list( - type = 'FeatureCollection', - override_source = 'enviroscan', - features = features - ) - fileOutLocJson <- fs::path(dirOut, 'location', paste0(nameCfgloc, '.json')) - base::writeLines(rjson::toJSON(locJson, indent = 2), con = fileOutLocJson) - log$info(base::paste0( - 'Wrote synthesized location JSON: ', fileOutLocJson, - ' (HOR=', HOR, ' VER=', VER, ' segments=', base::length(features), ')' - )) - } - # Link only this datum's group JSON into the output group/ directory. fileOutGrpJson <- fs::path(dirOutGroup, paste0(nameCfgloc, '.json')) diff --git a/modules/pub_files/output_files/sensor_positions/sensor_positions_file.py b/modules/pub_files/output_files/sensor_positions/sensor_positions_file.py index cc43907db5..f77694c184 100644 --- a/modules/pub_files/output_files/sensor_positions/sensor_positions_file.py +++ b/modules/pub_files/output_files/sensor_positions/sensor_positions_file.py @@ -1,5 +1,4 @@ import csv -import json from datetime import datetime from pathlib import Path from typing import Tuple, List, Dict, Optional @@ -23,10 +22,6 @@ def write_file(out_path: Path, location_path: Path, elements: PathElements, time writer = csv.writer(file) writer.writerow(get_column_names()) file_rows = [] - # EnviroSCAN features accumulate across the monthly datum's days; - # merged into rows after the glob loop so day-adjacent same-depth - # segments collapse into contiguous ranges. - enviroscan_features = [] # Parse location file path for the datum elements. Assume we end at site (**/site/location/*/location_file.json) site = location_path.parts[-1] for path in location_path.parent.parent.rglob(f'*/{site}/location/*/*.json'): @@ -35,48 +30,23 @@ def write_file(out_path: Path, location_path: Path, elements: PathElements, time location = database.get_named_location(named_location_name) (row_hor_ver, row_location_id, row_description) = get_named_location_data(database, named_location_name) geolocations = database.get_geolocations(named_location_name) - - # Read the JSON contents to detect per-VER overrides (e.g. EnviroSCAN - # writes one JSON per depth under distinct split-group dirs; the same - # CFGLOC name yields one DB row, so JSON contents differentiate rows). - location_json = None - try: - if path.stat().st_size > 0: - with open(path) as f: - location_json = json.load(f) - except (json.JSONDecodeError, OSError): - location_json = None - - if sensor_specific_processors.is_enviroscan_sensor(location_json): - sensor_specific_processors.collect_enviroscan_features( - location_json, named_location_name, - row_location_id, row_description, geolocations, - enviroscan_features) - continue - + for geolocation in geolocations: + # Use the specified processing method if sensor_specific_processors.is_tchain_sensor(location): rows = sensor_specific_processors.create_tchain_rows( - database, location, geolocation, row_hor_ver, + database, location, geolocation, row_hor_ver, row_location_id, row_description, _create_base_row_data, _add_reference_position_data) else: - rows = _create_standard_rows(database, geolocation, row_hor_ver, + rows = _create_standard_rows(database, geolocation, row_hor_ver, row_location_id, row_description) - + # Add rows, preventing duplicates for row in rows: if row not in file_rows: file_rows.append(row) - - # Merge collected enviroscan features across days and emit rows. - enviro_rows = sensor_specific_processors.aggregate_enviroscan_rows( - enviroscan_features, database, - _create_base_row_data, _add_reference_position_data) - for row in enviro_rows: - if row not in file_rows: - file_rows.append(row) - + writer.writerows(file_rows) return file_path diff --git a/modules/pub_files/output_files/sensor_positions/sensor_specific_processors.py b/modules/pub_files/output_files/sensor_positions/sensor_specific_processors.py index 9d0d43a13b..0fd6030743 100644 --- a/modules/pub_files/output_files/sensor_positions/sensor_specific_processors.py +++ b/modules/pub_files/output_files/sensor_positions/sensor_specific_processors.py @@ -1,9 +1,7 @@ """ Sensor-specific processing module. -Handles thermistor depth processing for tchain sensors and per-VER depth -overrides for EnviroSCAN (concH2oSoilSalinity). +Handles thermistor depth processing for tchain sensors. """ -from collections import defaultdict from typing import List, Dict, Optional from pub_files.output_files.sensor_positions.sensor_position import get_property from pub_files.output_files.sensor_positions.sensor_positions_database import SensorPositionsDatabase @@ -39,10 +37,10 @@ def create_tchain_rows(database: SensorPositionsDatabase, location, geolocation, hor_ver_split = complete_row_data['row_hor_ver'].split('.') hor_ver_split[1] = thermistor_id modified_hor_ver = ".".join(hor_ver_split) - + # Adjust z_offset for thermistor depth modified_z_offset = round(complete_row_data['row_z_offset'] - depth, 2) - + tchain_row = [ modified_hor_ver, complete_row_data['row_location_id'], @@ -68,122 +66,5 @@ def create_tchain_rows(database: SensorPositionsDatabase, location, geolocation, complete_row_data['row_y_azimuth'] ] tchain_rows.append(tchain_row) - + return tchain_rows - - -def is_enviroscan_sensor(location_json: Optional[dict]) -> bool: - """Check if this location JSON was synthesized by group_split for EnviroSCAN. - - The R wrapper wrap.concH2oSalinity.grp.split.R writes one JSON per (CFGLOC, - VER) with override_source == 'enviroscan'. The shared CFGLOC has one DB row - so DB values alone would collapse the 8 depths to one; the JSON supplies - per-VER HOR/VER and depth (z-offset). - """ - if not location_json: - return False - return location_json.get('override_source') == 'enviroscan' - - -def collect_enviroscan_features(location_json: dict, cfgloc: str, - row_location_id: str, row_description: str, - geolocations, sink: List[Dict]) -> None: - """Append each feature in the JSON to a shared collection pool for - later aggregation across days. group_split emits one JSON per (CFGLOC, - VER) per day with N features covering N depth segments within that day; - pub_files sees all monthly datum days at once and merges day-adjacent - same-(cfgloc, HOR, VER, depth) features into single rows.""" - for feature in location_json.get('features', []) or []: - if not feature: - continue - depth = feature.get('depth') - if depth is None: - continue - sink.append({ - 'cfgloc': cfgloc, - 'hor': feature.get('HOR', ''), - 'ver': feature.get('VER', ''), - 'depth': float(depth), - 'position_start_date': feature.get('positionStartDateTime', ''), - 'position_end_date': feature.get('positionEndDateTime', ''), - 'row_location_id': row_location_id, - 'row_description': row_description, - 'geolocations': geolocations, - }) - - -def aggregate_enviroscan_rows(collected: List[Dict], - database: SensorPositionsDatabase, - create_base_row_data_func, - add_reference_position_data_func) -> List[List]: - """Merge day-adjacent same-(cfgloc, HOR, VER, depth) features into - contiguous ranges, emit one row per merged range per DB geolocation. - Adjacent day-runs are considered contiguous when prev.end == next.start - (group_split ensures this: day N ends at UTC midnight, day N+1 starts - at the same instant; transition-day crossovers fall in the middle of - the ambiguous zone). z_offset overridden as DB_z + segment depth.""" - if not collected: - return [] - - # Group by (cfgloc, HOR, VER, rounded depth). Depths compared to 4 - # decimals so cal-file jitter doesn't fragment same-depth runs. - groups: Dict[tuple, List[Dict]] = defaultdict(list) - for f in collected: - key = (f['cfgloc'], f['hor'], f['ver'], round(f['depth'], 4)) - groups[key].append(f) - - rows: List[List] = [] - for (cfgloc, hor, ver, depth), feats in groups.items(): - feats.sort(key=lambda x: x['position_start_date']) - - # Merge contiguous ISO-Z strings (group_split guarantees exact match - # at day boundaries — no timezone or precision drift on the seam). - merged: List[Dict] = [dict(feats[0])] - for f in feats[1:]: - if merged[-1]['position_end_date'] == f['position_start_date']: - merged[-1]['position_end_date'] = f['position_end_date'] - else: - merged.append(dict(f)) - - override_hor_ver = f'{hor}.{ver}' - for m in merged: - for geolocation in m['geolocations']: - base_data = create_base_row_data_func( - database, geolocation, override_hor_ver, - m['row_location_id'], m['row_description']) - base_data['row_z_offset'] = round( - base_data['row_z_offset'] + float(depth), 2) - # Segment range overrides DB geolocation dates for the - # positionStart/EndDateTime CSV columns. - base_data['row_position_start_date'] = m['position_start_date'] - base_data['row_position_end_date'] = m['position_end_date'] - - complete_rows = add_reference_position_data_func( - database, base_data, geolocation, geolocation.offset_name) - - for row_data in complete_rows: - rows.append([ - row_data['row_hor_ver'], - row_data['row_location_id'], - row_data['row_description'], - row_data['row_position_start_date'], - row_data['row_position_end_date'], - row_data['row_reference_location_id'], - row_data['row_reference_location_description'], - row_data['row_reference_location_start_date'], - row_data['row_reference_location_end_date'], - row_data['row_x_offset'], - row_data['row_y_offset'], - row_data['row_z_offset'], - row_data['row_pitch'], - row_data['row_roll'], - row_data['row_azimuth'], - row_data['row_reference_location_latitude'], - row_data['row_reference_location_longitude'], - row_data['row_reference_location_elevation'], - row_data['row_east_offset'], - row_data['row_north_offset'], - row_data['row_x_azimuth'], - row_data['row_y_azimuth'] - ]) - return rows From d3730ab1d847d93e3bc7c300dfc7c92e93b5b5f1 Mon Sep 17 00:00:00 2001 From: Teresa Burlingame Date: Tue, 7 Jul 2026 11:56:33 -0700 Subject: [PATCH 04/16] new module for envscn loc history --- .../Dockerfile | 47 +++ ...H2oSoilSalinity_position_history_loader.py | 320 ++++++++++++++++++ ...ilSalinity_position_history_loader_main.py | 45 +++ .../requirements.txt | 3 + .../get_enviroscan_asset_installs.py | 48 +++ .../data_access/get_enviroscan_cfgloc_vers.py | 46 +++ .../get_enviroscan_cvald1_calibrations.py | 59 ++++ modules/data_access/types/asset_install.py | 12 + modules/data_access/types/cfgloc_ver.py | 9 + .../data_access/types/cvald1_calibration.py | 14 + 10 files changed, 603 insertions(+) create mode 100644 modules/concH2oSoilSalinity_position_history_loader/Dockerfile create mode 100644 modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py create mode 100644 modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader_main.py create mode 100644 modules/concH2oSoilSalinity_position_history_loader/requirements.txt create mode 100644 modules/data_access/get_enviroscan_asset_installs.py create mode 100644 modules/data_access/get_enviroscan_cfgloc_vers.py create mode 100644 modules/data_access/get_enviroscan_cvald1_calibrations.py create mode 100644 modules/data_access/types/asset_install.py create mode 100644 modules/data_access/types/cfgloc_ver.py create mode 100644 modules/data_access/types/cvald1_calibration.py diff --git a/modules/concH2oSoilSalinity_position_history_loader/Dockerfile b/modules/concH2oSoilSalinity_position_history_loader/Dockerfile new file mode 100644 index 0000000000..a1748a8748 --- /dev/null +++ b/modules/concH2oSoilSalinity_position_history_loader/Dockerfile @@ -0,0 +1,47 @@ +#### +# +# This dockerfile builds the concH2oSoilSalinity position-history loader. +# Must be built from the project root so common/ and data_access/ are in context: +# docker build --no-cache \ +# -t concH2oSoilSalinity_position_history_loader:latest \ +# -f modules/concH2oSoilSalinity_position_history_loader/Dockerfile modules +# +### +FROM registry.access.redhat.com/ubi8/ubi-minimal + +ARG MODULE_DIR="./modules" +ARG APP_DIR="concH2oSoilSalinity_position_history_loader" +ARG COMMON_DIR="common" +ARG DATA_ACCESS_DIR="data_access" +ARG CONTAINER_APP_DIR="/usr/src/app" +ENV PYTHONPATH="${PYTHONPATH}:${CONTAINER_APP_DIR}" + +WORKDIR ${CONTAINER_APP_DIR} + +COPY ${MODULE_DIR}/${APP_DIR}/requirements.txt ${CONTAINER_APP_DIR}/${APP_DIR}/app-requirements.txt +COPY ${MODULE_DIR}/${DATA_ACCESS_DIR}/requirements.txt ${CONTAINER_APP_DIR}/${APP_DIR}/data-access-requirements.txt + +RUN update-ca-trust && \ + microdnf update -y --disableplugin=subscription-manager && \ + microdnf install -y --disableplugin=subscription-manager \ + shadow-utils \ + gcc \ + libzstd \ + python39 \ + python39-pip \ + python39-wheel \ + python39-devel \ + python39-setuptools && \ + python3 -mpip install --no-cache-dir --upgrade pip setuptools wheel && \ + python3 -mpip install --no-cache-dir -r ${CONTAINER_APP_DIR}/${APP_DIR}/app-requirements.txt && \ + python3 -mpip install --no-cache-dir -r ${CONTAINER_APP_DIR}/${APP_DIR}/data-access-requirements.txt && \ + microdnf remove -y --disableplugin=subscription-manager gcc cpp && \ + microdnf clean all --disableplugin=subscription-manager && \ + groupadd -g 9999 appuser && \ + useradd -r -u 9999 -g appuser appuser + +COPY ${MODULE_DIR}/${APP_DIR} ${CONTAINER_APP_DIR}/${APP_DIR} +COPY ${MODULE_DIR}/${COMMON_DIR} ${CONTAINER_APP_DIR}/${COMMON_DIR} +COPY ${MODULE_DIR}/${DATA_ACCESS_DIR} ${CONTAINER_APP_DIR}/${DATA_ACCESS_DIR} + +USER appuser diff --git a/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py b/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py new file mode 100644 index 0000000000..3b5685665e --- /dev/null +++ b/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py @@ -0,0 +1,320 @@ +#!/usr/bin/env python3 +""" +concH2oSoilSalinity position-history loader. + +Queries PDR for the full CFGLOC × asset × calibration × geolocation history for +every enviroscan probe, stitches the intersections, applies the CVALD1 depth +correction (depth_m = z_offset + CVALD1_cm / -100) per VER, and writes one JSON +file per CFGLOC. Consumers (pub_files sensor_positions) read these files instead +of hitting the DB per publish month, so every downloaded month contains the +complete position history — including moves that happened outside that month's +sensor operation window. +""" +import json +import sys +from collections import defaultdict +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable, Dict, List, Optional, Set, Tuple + +import structlog + +import common.date_formatter as date_formatter +from common.err_datum import err_datum_path +from data_access.types.asset_install import AssetInstall +from data_access.types.cfgloc_ver import CfglocVer +from data_access.types.cvald1_calibration import Cvald1Calibration + +log = structlog.get_logger() + +DateRange = Tuple[Optional[datetime], Optional[datetime]] + + +def write_files(*, + get_asset_installs: Callable[[], List[AssetInstall]], + get_calibrations: Callable[[], List[Cvald1Calibration]], + get_cfgloc_vers: Callable[[], List[CfglocVer]], + get_geolocations: Callable[[int], Any], + get_parents: Callable[[int], Optional[Dict[str, Tuple[int, str]]]], + out_path: Path, + err_path: Path, + generated_at: Optional[datetime] = None) -> None: + """ + Build and write one position-history JSON per enviroscan CFGLOC. + + :param get_asset_installs: Callable returning Q1 rows (CFGLOC × asset install). + :param get_calibrations: Callable returning Q2 rows (CVALD1 per asset × period × VSWC stream). + :param get_cfgloc_vers: Callable returning Q4 rows (VERs actually configured per CFGLOC). + :param get_geolocations: Callable returning the FeatureCollection of geolocations for a nam_locn_id + (reuses data_access.get_named_location_geolocations). + :param get_parents: Callable returning the parents dict for a nam_locn_id + (reuses data_access.get_named_location_parents). + :param out_path: Base output path. + :param err_path: Error-routing base path. + :param generated_at: Timestamp stamped into every JSON. Defaults to now (UTC). + """ + if generated_at is None: + generated_at = datetime.now(tz=timezone.utc) + generated_at_str = _fmt_dt(generated_at) + + installs = get_asset_installs() + calibrations = get_calibrations() + cfgloc_vers = get_cfgloc_vers() + + installs_by_cfgloc: Dict[str, List[AssetInstall]] = defaultdict(list) + for install in installs: + installs_by_cfgloc[install.cfgloc].append(install) + + # (asset_uid, valid_start, valid_end) -> [stream calibrations at that period] + cals_by_asset_period: Dict[Tuple[int, Optional[datetime], Optional[datetime]], List[Cvald1Calibration]] = defaultdict(list) + for cal in calibrations: + cals_by_asset_period[(cal.asset_uid, cal.valid_start_time, cal.valid_end_time)].append(cal) + + allowed_vers_by_cfgloc: Dict[str, Set[str]] = defaultdict(set) + hor_by_cfgloc: Dict[str, str] = {} + for cv in cfgloc_vers: + allowed_vers_by_cfgloc[cv.cfgloc].add(cv.ver) + hor_by_cfgloc[cv.cfgloc] = cv.hor # HOR is per-CFGLOC (all VERs share the same HOR) + + for cfgloc, cfgloc_installs in installs_by_cfgloc.items(): + allowed_vers = allowed_vers_by_cfgloc.get(cfgloc) + if not allowed_vers: + # CFGLOC has enviroscan installs but is not configured in the split group + # (e.g. sensor decommissioned before concH2oSoilSalinity DP existed). Skip. + log.debug(f'Skipping {cfgloc}: no configured VERs in split group') + continue + + hor = hor_by_cfgloc[cfgloc] + nam_locn_id = cfgloc_installs[0].nam_locn_id + cfgloc_description = cfgloc_installs[0].cfgloc_description + + parents = get_parents(nam_locn_id) or {} + site = parents.get('site', (None, None))[1] + + geo_collection = get_geolocations(nam_locn_id) + geolocations = _flatten_geolocations(geo_collection) + + rows = _build_rows( + cfgloc_installs=cfgloc_installs, + geolocations=geolocations, + cals_by_asset_period=cals_by_asset_period, + allowed_vers=allowed_vers, + hor=hor, + ) + + if not rows: + log.debug(f'{cfgloc}: no rows produced (no valid intersections)') + continue + + out_obj = { + 'source_type': 'enviroscan', + 'cfgloc': cfgloc, + 'cfgloc_description': cfgloc_description, + 'site': site, + 'generated_at': generated_at_str, + 'rows': rows, + } + _write_json(cfgloc=cfgloc, obj=out_obj, out_path=out_path, err_path=err_path) + + +def _build_rows(*, + cfgloc_installs: List[AssetInstall], + geolocations: List[Dict[str, Any]], + cals_by_asset_period: Dict[Tuple[int, Optional[datetime], Optional[datetime]], List[Cvald1Calibration]], + allowed_vers: Set[str], + hor: str) -> List[Dict[str, Any]]: + """ + For a single CFGLOC, produce the full list of (install × geolocation × cal-period × VER) + intersection rows. + """ + rows: List[Dict[str, Any]] = [] + for install in cfgloc_installs: + asset_periods = [(period, streams) + for period, streams in cals_by_asset_period.items() + if period[0] == install.asset_uid] + for geo in geolocations: + for (_, cal_start, cal_end), streams in asset_periods: + window = _intersect( + (install.install_date, install.remove_date), + (geo['start_date'], geo['end_date']), + (cal_start, cal_end), + ) + if window is None: + continue + win_start, win_end = window + for stream in streams: + ver = _stream_to_ver(stream.schema_field_name) + if ver is None or str(ver) not in allowed_vers: + continue + z_offset_adjusted = round(geo['z_offset'] + stream.cvald1_cm / -100.0, 4) + rows.append({ + 'hor': hor, + 'ver': str(ver), + 'position_start_date': _fmt_dt(win_start), + 'position_end_date': _fmt_dt(win_end), + 'x_offset': geo['x_offset'], + 'y_offset': geo['y_offset'], + 'z_offset': z_offset_adjusted, + 'pitch': geo['pitch'], + 'roll': geo['roll'], + 'azimuth': geo['azimuth'], + 'reference_location_id': geo['reference_location_id'], + 'reference_location_start_date': geo['reference_location_start_date'], + 'reference_location_end_date': geo['reference_location_end_date'], + 'reference_location_latitude': geo['reference_location_latitude'], + 'reference_location_longitude': geo['reference_location_longitude'], + 'reference_location_elevation': geo['reference_location_elevation'], + 'asset_uid': install.asset_uid, + 'cvald1_cm': stream.cvald1_cm, + 'cal_valid_start': _fmt_dt(cal_start), + 'cal_valid_end': _fmt_dt(cal_end), + 'source_stream': stream.schema_field_name, + 'cert_filename': stream.cert_filename, + }) + return rows + + +def _flatten_geolocations(feature_collection: Any) -> List[Dict[str, Any]]: + """ + Unpack the FeatureCollection returned by get_named_location_geolocations into a flat list of dicts. + + Each feature in the top-level collection is one row from `locn` for this CFGLOC. + `properties.reference_location` is a nested Feature carrying the reference (e.g. SOILPL...) + nam_locn_name and its own recursive geolocations. The reference's first geolocation + row is where lat/lon/elevation live. + """ + flat: List[Dict[str, Any]] = [] + if feature_collection is None: + return flat + features = feature_collection.get('features') if isinstance(feature_collection, dict) else getattr(feature_collection, 'features', []) + for feature in features or []: + props = feature.get('properties') if isinstance(feature, dict) else feature.properties + if props is None: + continue + ref_feature = props.get('reference_location') + ref_name = ref_lat = ref_lon = ref_elev = None + ref_start = ref_end = None + if ref_feature is not None: + ref_props = ref_feature.get('properties') if isinstance(ref_feature, dict) else ref_feature.properties + if ref_props is not None: + ref_name = ref_props.get('name') + ref_locations = ref_props.get('locations') or {} + ref_feats = ref_locations.get('features') if isinstance(ref_locations, dict) else getattr(ref_locations, 'features', []) + if ref_feats: + ref_geo_feat = ref_feats[0] + ref_geo_props = ref_geo_feat.get('properties') if isinstance(ref_geo_feat, dict) else ref_geo_feat.properties + ref_geometry = ref_geo_feat.get('geometry') if isinstance(ref_geo_feat, dict) else ref_geo_feat.geometry + ref_lon, ref_lat, ref_elev = _extract_point(ref_geometry) + if ref_geo_props is not None: + ref_start = ref_geo_props.get('start_date') + ref_end = ref_geo_props.get('end_date') + flat.append({ + 'start_date': _parse_dt(props.get('start_date')), + 'end_date': _parse_dt(props.get('end_date')), + 'x_offset': props.get('x_offset'), + 'y_offset': props.get('y_offset'), + 'z_offset': props.get('z_offset'), + 'pitch': props.get('alpha'), + 'roll': props.get('beta'), + 'azimuth': props.get('gamma'), + 'reference_location_id': ref_name, + 'reference_location_start_date': ref_start, + 'reference_location_end_date': ref_end, + 'reference_location_latitude': ref_lat, + 'reference_location_longitude': ref_lon, + 'reference_location_elevation': ref_elev, + }) + return flat + + +def _extract_point(geometry: Any) -> Tuple[Optional[float], Optional[float], Optional[float]]: + """Return (lon, lat, elevation) from a geojson Point-like structure.""" + if geometry is None: + return None, None, None + coords = geometry.get('coordinates') if isinstance(geometry, dict) else getattr(geometry, 'coordinates', None) + if not coords or len(coords) < 2: + return None, None, None + lon = float(coords[0]) + lat = float(coords[1]) + elev = float(coords[2]) if len(coords) >= 3 else None + return lon, lat, elev + + +def _intersect(*periods: DateRange) -> Optional[DateRange]: + """ + Intersect any number of half-open date ranges. `None` on either side means open-ended. + Returns None if the ranges do not overlap. + + Strips tzinfo before comparing to avoid crashes when psycopg2 returns tz-aware + datetimes for some columns while string-parsed dates come back tz-naive. + """ + starts = [_naive(p[0]) for p in periods if p[0] is not None] + ends = [_naive(p[1]) for p in periods if p[1] is not None] + win_start = max(starts) if starts else None + win_end = min(ends) if ends else None + if win_start is not None and win_end is not None and win_start >= win_end: + return None + return (win_start, win_end) + + +def _naive(dt: datetime) -> datetime: + """Drop tzinfo so tz-aware and tz-naive datetimes can be compared consistently.""" + if dt.tzinfo is not None: + return dt.replace(tzinfo=None) + return dt + + +def _stream_to_ver(schema_field_name: Optional[str]) -> Optional[int]: + """rawVSWC0 -> 501, rawVSWC7 -> 508. Returns None if the pattern doesn't match.""" + if not schema_field_name or not schema_field_name.startswith('rawVSWC'): + return None + suffix = schema_field_name[len('rawVSWC'):] + if not suffix.isdigit(): + return None + return 501 + int(suffix) + + +def _fmt_dt(value: Any) -> str: + """ + Return an ISO-8601 UTC string for a datetime, an empty string for None, + and passthrough for pre-formatted strings. + """ + if value is None: + return '' + if isinstance(value, str): + return value + if isinstance(value, datetime): + return date_formatter.to_string(value) + return str(value) + + +def _parse_dt(value: Any) -> Optional[datetime]: + """ + Return a datetime for either a datetime input or a formatted string. + Empty/None returns None (open-ended). + """ + if value is None or value == '': + return None + if isinstance(value, datetime): + return value + if isinstance(value, str): + try: + return date_formatter.to_datetime(value) + except (ValueError, AttributeError): + return None + return None + + +def _write_json(*, cfgloc: str, obj: Dict[str, Any], out_path: Path, err_path: Path) -> None: + """ + Write the position-history JSON for one CFGLOC. Path shape matches how pub_files + joins the loader output: /enviroscan//position_history/_history.json + """ + file_path = Path(out_path, 'enviroscan', cfgloc, 'position_history', f'{cfgloc}_history.json') + file_path.parent.mkdir(parents=True, exist_ok=True) + try: + with open(file_path, 'w') as f: + json.dump(obj, f, indent=2, default=str, sort_keys=False) + except Exception: + err_datum_path(err=sys.exc_info(), DirDatm=str(file_path.parent), DirErrBase=Path(err_path), + RmvDatmOut=True, DirOutBase=out_path) diff --git a/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader_main.py b/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader_main.py new file mode 100644 index 0000000000..46718220e5 --- /dev/null +++ b/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader_main.py @@ -0,0 +1,45 @@ +#!/usr/bin/env python3 +from contextlib import closing +from functools import partial +from pathlib import Path + +import environs +import structlog + +import common.log_config as log_config +from data_access.db_config_reader import read_from_mount +from data_access.db_connector import DbConnector +from data_access.get_enviroscan_asset_installs import get_enviroscan_asset_installs +from data_access.get_enviroscan_cvald1_calibrations import get_enviroscan_cvald1_calibrations +from data_access.get_enviroscan_cfgloc_vers import get_enviroscan_cfgloc_vers +from data_access.get_named_location_geolocations import get_named_location_geolocations +from data_access.get_named_location_parents import get_named_location_parents + +import concH2oSoilSalinity_position_history_loader.concH2oSoilSalinity_position_history_loader as loader + +log = structlog.get_logger() + + +def main() -> None: + env = environs.Env() + out_path: Path = env.path('OUT_PATH') + err_path: Path = env.path('ERR_PATH') + log_level: str = env.log_level('LOG_LEVEL', 'INFO') + log_config.configure(log_level) + log.debug(f'out_path: {out_path}') + + db_config = read_from_mount(Path('/var/db_secret')) + with closing(DbConnector(db_config)) as connector: + loader.write_files( + get_asset_installs=partial(get_enviroscan_asset_installs, connector), + get_calibrations=partial(get_enviroscan_cvald1_calibrations, connector), + get_cfgloc_vers=partial(get_enviroscan_cfgloc_vers, connector), + get_geolocations=partial(get_named_location_geolocations, connector), + get_parents=partial(get_named_location_parents, connector), + out_path=out_path, + err_path=err_path, + ) + + +if __name__ == '__main__': + main() diff --git a/modules/concH2oSoilSalinity_position_history_loader/requirements.txt b/modules/concH2oSoilSalinity_position_history_loader/requirements.txt new file mode 100644 index 0000000000..dc0cf40d87 --- /dev/null +++ b/modules/concH2oSoilSalinity_position_history_loader/requirements.txt @@ -0,0 +1,3 @@ +structlog==21.5.0 +environs==6.0.0 +geojson==2.4.1 diff --git a/modules/data_access/get_enviroscan_asset_installs.py b/modules/data_access/get_enviroscan_asset_installs.py new file mode 100644 index 0000000000..d7436d928e --- /dev/null +++ b/modules/data_access/get_enviroscan_asset_installs.py @@ -0,0 +1,48 @@ +#!/usr/bin/env python3 +from contextlib import closing +from typing import List + +from data_access.db_connector import DbConnector +from data_access.types.asset_install import AssetInstall + + +def get_enviroscan_asset_installs(connector: DbConnector) -> List[AssetInstall]: + """ + Return every (CFGLOC × asset install period) for enviroscan probes across all sites and time. + + Corresponds to Q1 in swc_loc_depths_notes.md. `DISTINCT` prevents the Cartesian + that arises when an asset has multiple rows in `is_asset_assignment`. + + :param connector: A database connection. + :return: One entry per (CFGLOC, asset_uid, install period). + """ + connection = connector.get_connection() + schema = connector.get_schema() + sql = f''' + select distinct + nl.nam_locn_name, + nl.nam_locn_desc, + nl.nam_locn_id, + ial.asset_uid, + ial.install_date, + ial.remove_date + from {schema}.is_asset_location ial + join {schema}.nam_locn nl on ial.nam_locn_id = nl.nam_locn_id + join {schema}.is_asset_assignment asgn on ial.asset_uid = asgn.asset_uid + join {schema}.is_asset_definition def on def.asset_definition_uuid = asgn.asset_definition_uuid + where def.sensor_type_name = 'enviroscan' + order by nl.nam_locn_name, ial.install_date desc + ''' + installs: List[AssetInstall] = [] + with closing(connection.cursor()) as cursor: + cursor.execute(sql) + for row in cursor.fetchall(): + installs.append(AssetInstall( + cfgloc=row[0], + cfgloc_description=row[1], + nam_locn_id=row[2], + asset_uid=row[3], + install_date=row[4], + remove_date=row[5], + )) + return installs diff --git a/modules/data_access/get_enviroscan_cfgloc_vers.py b/modules/data_access/get_enviroscan_cfgloc_vers.py new file mode 100644 index 0000000000..9a7637bfc9 --- /dev/null +++ b/modules/data_access/get_enviroscan_cfgloc_vers.py @@ -0,0 +1,46 @@ +#!/usr/bin/env python3 +from contextlib import closing +from typing import List + +from data_access.db_connector import DbConnector +from data_access.types.cfgloc_ver import CfglocVer + + +def get_enviroscan_cfgloc_vers(connector: DbConnector) -> List[CfglocVer]: + """ + Return the VER positions actually configured for each CFGLOC in the + `conc-h2o-soil-salinity-split_*` group prefix. + + Corresponds to Q4 in swc_loc_depths_notes.md. This governs which VERs the + loader emits: a CFGLOC missing a lower depth simply won't have that VER in + the group table, so it won't be emitted. Consistent with what `group_split` + produces upstream. + + :param connector: A database connection. + :return: One entry per (CFGLOC, HOR, VER) that appears in the split group. + """ + connection = connector.get_connection() + schema = connector.get_schema() + sql = f''' + select + nl.nam_locn_name, + g.hor, + g.ver, + g.group_name + from {schema}.named_location_group nlg + join {schema}."group" g on nlg.group_id = g.group_id + join {schema}.nam_locn nl on nlg.named_location_id = nl.nam_locn_id + where g.group_name like 'conc-h2o-soil-salinity-split\\_%%' escape '\\' + order by nl.nam_locn_name, g.hor, g.ver + ''' + entries: List[CfglocVer] = [] + with closing(connection.cursor()) as cursor: + cursor.execute(sql) + for row in cursor.fetchall(): + entries.append(CfglocVer( + cfgloc=row[0], + hor=row[1], + ver=row[2], + group_name=row[3], + )) + return entries diff --git a/modules/data_access/get_enviroscan_cvald1_calibrations.py b/modules/data_access/get_enviroscan_cvald1_calibrations.py new file mode 100644 index 0000000000..826bbe22f5 --- /dev/null +++ b/modules/data_access/get_enviroscan_cvald1_calibrations.py @@ -0,0 +1,59 @@ +#!/usr/bin/env python3 +from contextlib import closing +from typing import List + +from data_access.db_connector import DbConnector +from data_access.types.cvald1_calibration import Cvald1Calibration + + +def get_enviroscan_cvald1_calibrations(connector: DbConnector) -> List[Cvald1Calibration]: + """ + Return every CVALD1 calibration for enviroscan probes across all assets and time, + filtered to the VSWC streams only (rawVSWC0..rawVSWC7 → VER 501..508). + + Corresponds to Q2 in swc_loc_depths_notes.md. Even `sensor_stream_num` values + (0,2,4,6,8,10,12,14) are VSWC; odd are VSIC. VSIC shares the same physical + depth so is excluded here — depth math uses only VSWC. + + :param connector: A database connection. + :return: One entry per (asset × valid-period × VSWC stream). + """ + connection = connector.get_connection() + schema = connector.get_schema() + sql = f''' + select + cal.asset_uid, + cal.calibration_id, + cal.sensor_stream_num, + it.schema_field_name, + cal.valid_start_time, + cal.valid_end_time, + cal.cert_filename, + mta.value + from {schema}.calibration cal + join {schema}.calibration_metadata mta on cal.calibration_id = mta.calibration_id + join {schema}.is_asset_assignment asgn on asgn.asset_uid = cal.asset_uid + join {schema}.is_asset_definition def on def.asset_definition_uuid = asgn.asset_definition_uuid + join {schema}.is_sensor_type st on st.sensor_type_name = def.sensor_type_name + join {schema}.is_ingest_term it on it.asset_definition_uuid = def.asset_definition_uuid + and it.stream_id = cal.sensor_stream_num + where mta.name = 'CVALD1' + and st.avro_schema_name = 'enviroscan' + and it.schema_field_name like 'rawVSWC%%' + order by cal.asset_uid, cal.valid_start_time, it.schema_field_name + ''' + calibrations: List[Cvald1Calibration] = [] + with closing(connection.cursor()) as cursor: + cursor.execute(sql) + for row in cursor.fetchall(): + calibrations.append(Cvald1Calibration( + asset_uid=row[0], + calibration_id=row[1], + sensor_stream_num=row[2], + schema_field_name=row[3], + valid_start_time=row[4], + valid_end_time=row[5], + cert_filename=row[6], + cvald1_cm=float(row[7]), + )) + return calibrations diff --git a/modules/data_access/types/asset_install.py b/modules/data_access/types/asset_install.py new file mode 100644 index 0000000000..55380aeca9 --- /dev/null +++ b/modules/data_access/types/asset_install.py @@ -0,0 +1,12 @@ +#!/usr/bin/env python3 +from datetime import datetime +from typing import NamedTuple, Optional + + +class AssetInstall(NamedTuple): + cfgloc: str + cfgloc_description: Optional[str] + nam_locn_id: int + asset_uid: int + install_date: Optional[datetime] + remove_date: Optional[datetime] diff --git a/modules/data_access/types/cfgloc_ver.py b/modules/data_access/types/cfgloc_ver.py new file mode 100644 index 0000000000..48be1d784d --- /dev/null +++ b/modules/data_access/types/cfgloc_ver.py @@ -0,0 +1,9 @@ +#!/usr/bin/env python3 +from typing import NamedTuple + + +class CfglocVer(NamedTuple): + cfgloc: str + hor: str + ver: str + group_name: str diff --git a/modules/data_access/types/cvald1_calibration.py b/modules/data_access/types/cvald1_calibration.py new file mode 100644 index 0000000000..178326e081 --- /dev/null +++ b/modules/data_access/types/cvald1_calibration.py @@ -0,0 +1,14 @@ +#!/usr/bin/env python3 +from datetime import datetime +from typing import NamedTuple, Optional + + +class Cvald1Calibration(NamedTuple): + asset_uid: int + calibration_id: int + sensor_stream_num: int + schema_field_name: str + valid_start_time: Optional[datetime] + valid_end_time: Optional[datetime] + cert_filename: Optional[str] + cvald1_cm: float From ffbc5a9cc670ff5511768a1b83e4f07b0572666a Mon Sep 17 00:00:00 2001 From: Teresa Burlingame Date: Tue, 7 Jul 2026 12:01:35 -0700 Subject: [PATCH 05/16] Copy loader build workflow to swc_loc_depths for workflow_dispatch --- ...2oSoilSalinity_position_history_loader.yml | 46 +++++++++++++++++++ 1 file changed, 46 insertions(+) create mode 100644 .github/workflows/build_push_concH2oSoilSalinity_position_history_loader.yml diff --git a/.github/workflows/build_push_concH2oSoilSalinity_position_history_loader.yml b/.github/workflows/build_push_concH2oSoilSalinity_position_history_loader.yml new file mode 100644 index 0000000000..a2b0afbd56 --- /dev/null +++ b/.github/workflows/build_push_concH2oSoilSalinity_position_history_loader.yml @@ -0,0 +1,46 @@ +name: "Build-push_concH2oSoilSalinity_position_history_loader" + +on: + push: + branches: + - 'master' + paths: + - 'modules/concH2oSoilSalinity_position_history_loader/**' + workflow_dispatch: {} # Allows trigger of workflow from web interface + +env: + GITHUB_TOKEN: ${{secrets.GITHUB_TOKEN }} + # Use github and google registries + GHCR_REGISTRY: ghcr.io + GCP_ARTIFACT_HOST: ${{ vars.SHARED_WIF_LOCATON }}-docker.pkg.dev + GCP_REGISTRY: ${{ vars.SHARED_WIF_LOCATON }}-docker.pkg.dev/${{ vars.SHARED_WIF_PROJECT }}/${{ vars.SHARED_WIF_REPO }} + GCP_PROVIDER: ${{ vars.SHARED_WIF_PROVIDER }} + GCP_SERVICE_ACCOUNT: ${{ vars.SHARED_WIF_SERVICE_ACCOUNT }} + GHCR_NS: battelleecology + # Use docker.io for Docker Hub if empty + REGISTRY: ghcr.io + # Get just the repo name from the event, i.e., NEON-IS-data-processing + REPO_NAME: ${{ github.event.repository.name }} + # IS module name + MODULE_PATH: ./modules/concH2oSoilSalinity_position_history_loader + IMAGE_NAME: neon-is-conc-h2o-salinity-position-history-loader + +jobs: + build-push: + runs-on: ubuntu-latest + permissions: + contents: 'write' + id-token: 'write' + steps: + - name: "Checkout" + uses: "actions/checkout@v4.1.4" + with: + fetch-depth: '0' + + - name: Get short SHA + run: echo "short_sha=$(git rev-parse --short HEAD)" >> $GITHUB_ENV + + - name: Build and push + uses: ./.github/actions/build-push + with: + image-tag: "${short_sha}" From b75bbceaded533aa541c8bdd9d13400fdc5be54c Mon Sep 17 00:00:00 2001 From: Teresa Burlingame Date: Tue, 7 Jul 2026 12:21:58 -0700 Subject: [PATCH 06/16] req update --- .../requirements.txt | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/modules/concH2oSoilSalinity_position_history_loader/requirements.txt b/modules/concH2oSoilSalinity_position_history_loader/requirements.txt index dc0cf40d87..a36f9890a4 100644 --- a/modules/concH2oSoilSalinity_position_history_loader/requirements.txt +++ b/modules/concH2oSoilSalinity_position_history_loader/requirements.txt @@ -1,3 +1,4 @@ structlog==21.5.0 -environs==6.0.0 +environs==11.0.0 +marshmallow==3.21.3 geojson==2.4.1 From 25d271da16a3d8729240052ef1f7939a1706e385 Mon Sep 17 00:00:00 2001 From: Teresa Burlingame Date: Tue, 7 Jul 2026 12:34:16 -0700 Subject: [PATCH 07/16] debugging --- .../concH2oSoilSalinity_position_history_loader.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py b/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py index 3b5685665e..d56bf107c2 100644 --- a/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py +++ b/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py @@ -228,11 +228,19 @@ def _flatten_geolocations(feature_collection: Any) -> List[Dict[str, Any]]: def _extract_point(geometry: Any) -> Tuple[Optional[float], Optional[float], Optional[float]]: - """Return (lon, lat, elevation) from a geojson Point-like structure.""" + """Return (lon, lat, elevation) from a geojson Point-like structure. + + PDR reference-location geometries sometimes come back as MultiPoint + ([[lon, lat, elev]]) instead of Point ([lon, lat, elev]); unwrap one level. + """ if geometry is None: return None, None, None coords = geometry.get('coordinates') if isinstance(geometry, dict) else getattr(geometry, 'coordinates', None) - if not coords or len(coords) < 2: + if not coords: + return None, None, None + if isinstance(coords[0], (list, tuple)): + coords = coords[0] + if len(coords) < 2: return None, None, None lon = float(coords[0]) lat = float(coords[1]) From 5a9c9d9ede19475ee032ae381b6293579fa8a275 Mon Sep 17 00:00:00 2001 From: Teresa Burlingame Date: Tue, 7 Jul 2026 12:48:33 -0700 Subject: [PATCH 08/16] fan out issue in query, update to avoid duplicates. --- modules/data_access/get_enviroscan_cvald1_calibrations.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/modules/data_access/get_enviroscan_cvald1_calibrations.py b/modules/data_access/get_enviroscan_cvald1_calibrations.py index 826bbe22f5..1eb4f4f116 100644 --- a/modules/data_access/get_enviroscan_cvald1_calibrations.py +++ b/modules/data_access/get_enviroscan_cvald1_calibrations.py @@ -20,8 +20,13 @@ def get_enviroscan_cvald1_calibrations(connector: DbConnector) -> List[Cvald1Cal """ connection = connector.get_connection() schema = connector.get_schema() + # DISTINCT: is_ingest_term can carry multiple historical rows per + # (asset_definition_uuid, stream_id) when the asset_definition has been + # versioned. Assets on a repeatedly-updated definition would otherwise + # fan out N-fold. See CFGLOC112096 asset 33113 (3 rows) vs assets 17834/7524 + # (1 row each) for a real-world example. sql = f''' - select + select distinct cal.asset_uid, cal.calibration_id, cal.sensor_stream_num, From 63b85a46ac62739502c690ccaadd04be3cfc2168 Mon Sep 17 00:00:00 2001 From: Teresa Burlingame Date: Tue, 7 Jul 2026 13:19:53 -0700 Subject: [PATCH 09/16] update action to get other modules --- .../build_push_concH2oSoilSalinity_position_history_loader.yml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.github/workflows/build_push_concH2oSoilSalinity_position_history_loader.yml b/.github/workflows/build_push_concH2oSoilSalinity_position_history_loader.yml index a2b0afbd56..76c196f7fe 100644 --- a/.github/workflows/build_push_concH2oSoilSalinity_position_history_loader.yml +++ b/.github/workflows/build_push_concH2oSoilSalinity_position_history_loader.yml @@ -6,6 +6,8 @@ on: - 'master' paths: - 'modules/concH2oSoilSalinity_position_history_loader/**' + - 'modules/data_access/**' + - 'modules/common/**' workflow_dispatch: {} # Allows trigger of workflow from web interface env: From f5b2bdc2116e48b3718ef6f62f4fcfa4163cb1dc Mon Sep 17 00:00:00 2001 From: Teresa Burlingame Date: Tue, 7 Jul 2026 13:28:44 -0700 Subject: [PATCH 10/16] testing new pub files for envscn --- modules/pub_files/application_config.py | 10 +- modules/pub_files/main.py | 3 +- .../sensor_positions/sensor_positions_file.py | 137 +++++++++++++++++- 3 files changed, 140 insertions(+), 10 deletions(-) diff --git a/modules/pub_files/application_config.py b/modules/pub_files/application_config.py index a4d1d40ef9..ec3727df39 100644 --- a/modules/pub_files/application_config.py +++ b/modules/pub_files/application_config.py @@ -1,5 +1,5 @@ from pathlib import Path -from typing import NamedTuple +from typing import NamedTuple, Optional import environs @@ -25,6 +25,10 @@ class ApplicationConfig(NamedTuple): eml_intellectual_rights_path: str eml_unit_types_path: str eml_units_path: str + # Optional: root of a pfs input carrying per-CFGLOC position_history JSONs + # (from concH2oSoilSalinity_position_history_loader). When set, sensor_positions + # reads history from these files instead of hitting the DB per publish month. + position_history_path: Optional[Path] = None def configure_from_environment() -> ApplicationConfig: @@ -33,6 +37,7 @@ def configure_from_environment() -> ApplicationConfig: relative_path_index: int = env.int('RELATIVE_PATH_INDEX') location_path: Path = env.path('LOCATION_PATH') out_path: Path = env.path('OUT_PATH') + position_history_path: Optional[Path] = env.path('POSITION_HISTORY_PATH', None) db_secrets_path: Path = env.path('DB_SECRETS_PATH') log_level: str = env.log_level('LOG_LEVEL', 'INFO') certificate_path: Path = env.path('GITHUB_PEM_PATH') @@ -68,4 +73,5 @@ def configure_from_environment() -> ApplicationConfig: eml_contact_path=eml_contact_path, eml_intellectual_rights_path=eml_intellectual_rights_path, eml_unit_types_path=eml_unit_types_path, - eml_units_path=eml_units_path) + eml_units_path=eml_units_path, + position_history_path=position_history_path) diff --git a/modules/pub_files/main.py b/modules/pub_files/main.py index 2d83dc3b40..db828c4d8f 100644 --- a/modules/pub_files/main.py +++ b/modules/pub_files/main.py @@ -68,7 +68,8 @@ def main() -> None: out_path=file_metadata.package_output_path, elements=file_metadata.path_elements, timestamp=timestamp, - database=sensor_positions_database) + database=sensor_positions_database, + position_history_path=config.position_history_path) # write eml file eml_file_config = EmlFileConfig(out_path=file_metadata.package_output_path, metadata=file_metadata, diff --git a/modules/pub_files/output_files/sensor_positions/sensor_positions_file.py b/modules/pub_files/output_files/sensor_positions/sensor_positions_file.py index f77694c184..12a9baabe0 100644 --- a/modules/pub_files/output_files/sensor_positions/sensor_positions_file.py +++ b/modules/pub_files/output_files/sensor_positions/sensor_positions_file.py @@ -1,4 +1,5 @@ import csv +import json from datetime import datetime from pathlib import Path from typing import Tuple, List, Dict, Optional @@ -13,8 +14,16 @@ def write_file(out_path: Path, location_path: Path, elements: PathElements, timestamp: datetime, - database: SensorPositionsDatabase) -> Path: - """Write the sensor positions file to the output path.""" + database: SensorPositionsDatabase, + position_history_path: Optional[Path] = None) -> Path: + """Write the sensor positions file to the output path. + + When `position_history_path` is provided and a per-CFGLOC history JSON exists at + `///position_history/_history.json`, + the rows for that CFGLOC come from the JSON instead of the DB. This is how + concH2oSoilSalinity_position_history_loader plugs the full enviroscan CFGLOC + history into every monthly sensor_positions.csv. + """ filename = get_filename(elements, timestamp=timestamp, file_type='sensor_positions', extension='csv') file_path = Path(out_path, filename) @@ -27,30 +36,144 @@ def write_file(out_path: Path, location_path: Path, elements: PathElements, time for path in location_path.parent.parent.rglob(f'*/{site}/location/*/*.json'): if path.is_file() and path.name.startswith('CFGLOC'): named_location_name = path.stem + + # Prefer loader-emitted history JSON when available for this CFGLOC. + history_rows = _read_position_history_rows(position_history_path, named_location_name, database) + if history_rows is not None: + for row in history_rows: + if row not in file_rows: + file_rows.append(row) + continue + location = database.get_named_location(named_location_name) (row_hor_ver, row_location_id, row_description) = get_named_location_data(database, named_location_name) geolocations = database.get_geolocations(named_location_name) - + for geolocation in geolocations: # Use the specified processing method if sensor_specific_processors.is_tchain_sensor(location): rows = sensor_specific_processors.create_tchain_rows( - database, location, geolocation, row_hor_ver, + database, location, geolocation, row_hor_ver, row_location_id, row_description, _create_base_row_data, _add_reference_position_data) else: - rows = _create_standard_rows(database, geolocation, row_hor_ver, + rows = _create_standard_rows(database, geolocation, row_hor_ver, row_location_id, row_description) - + # Add rows, preventing duplicates for row in rows: if row not in file_rows: file_rows.append(row) - + writer.writerows(file_rows) return file_path +def _read_position_history_rows(position_history_path: Optional[Path], + named_location_name: str, + database: SensorPositionsDatabase) -> Optional[List[List]]: + """Load rows from the loader-emitted history JSON, or None if unavailable. + + Returns a list of CSV rows in the same column order as `get_column_names()`. + Any CFGLOC without a matching JSON falls back to the DB-driven path. + """ + if position_history_path is None: + return None + # Loader writes to ///position_history/_history.json. + # Only enviroscan is emitted today; expand the source_type list as more loaders come online. + candidates = [position_history_path / source_type / named_location_name / 'position_history' + / f'{named_location_name}_history.json' + for source_type in ('enviroscan',)] + json_path = next((p for p in candidates if p.is_file()), None) + if json_path is None: + return None + + with open(json_path, 'r') as fp: + payload = json.load(fp) + location = database.get_named_location(named_location_name) + row_description = location.description + row_location_id = location.name + + # Reference-location azimuths (for east/north offset math) live in the DB; cache per ref name. + ref_geolocation_cache: Dict[str, List] = {} + + csv_rows: List[List] = [] + for entry in payload.get('rows', []): + ref_name = entry.get('reference_location_id') + if ref_name not in ref_geolocation_cache: + ref_geolocation_cache[ref_name] = database.get_geolocations(ref_name) if ref_name else [] + ref_geolocation = _match_reference_geolocation(ref_geolocation_cache[ref_name], entry) + if ref_geolocation is not None: + reference_position = get_position(ref_geolocation, entry['x_offset'], entry['y_offset']) + east_offset = reference_position.east_offset + north_offset = reference_position.north_offset + x_azimuth = reference_position.x_azimuth + y_azimuth = reference_position.y_azimuth + ref_description = database.get_named_location(ref_name).description if ref_name else '' + else: + east_offset = north_offset = x_azimuth = y_azimuth = '' + ref_description = database.get_named_location(ref_name).description if ref_name else '' + + csv_rows.append([ + f"{entry['hor']}.{entry['ver']}", + row_location_id, + row_description, + entry.get('position_start_date', ''), + entry.get('position_end_date', ''), + ref_name or '', + ref_description, + entry.get('reference_location_start_date', '') or '', + entry.get('reference_location_end_date', '') or '', + round(entry['x_offset'], 2) if entry.get('x_offset') is not None else '', + round(entry['y_offset'], 2) if entry.get('y_offset') is not None else '', + round(entry['z_offset'], 2) if entry.get('z_offset') is not None else '', + round(entry['pitch'], 2) if entry.get('pitch') is not None else '', + round(entry['roll'], 2) if entry.get('roll') is not None else '', + round(entry['azimuth'], 2) if entry.get('azimuth') is not None else '', + round(entry['reference_location_latitude'], 6) if entry.get('reference_location_latitude') is not None else '', + round(entry['reference_location_longitude'], 6) if entry.get('reference_location_longitude') is not None else '', + round(entry['reference_location_elevation'], 2) if entry.get('reference_location_elevation') is not None else '', + round(east_offset, 2) if isinstance(east_offset, (int, float)) or hasattr(east_offset, 'quantize') else east_offset, + round(north_offset, 2) if isinstance(north_offset, (int, float)) or hasattr(north_offset, 'quantize') else north_offset, + round(x_azimuth, 2) if isinstance(x_azimuth, (int, float)) else x_azimuth, + round(y_azimuth, 2) if isinstance(y_azimuth, (int, float)) else y_azimuth, + ]) + return csv_rows + + +def _match_reference_geolocation(ref_geolocations: List, entry: Dict): + """Pick the reference geolocation whose validity window overlaps the JSON entry's position window.""" + if not ref_geolocations: + return None + entry_start = _strip_tz(_parse_iso(entry.get('position_start_date'))) + entry_end = _strip_tz(_parse_iso(entry.get('position_end_date'))) + for ref in ref_geolocations: + ref_start = _strip_tz(ref.start_date) + ref_end = _strip_tz(ref.end_date) + if entry_end is not None and ref_start is not None and entry_end <= ref_start: + continue + if entry_start is not None and ref_end is not None and entry_start >= ref_end: + continue + return ref + # Fall back to the first available so downstream still gets azimuth-based east/north math. + return ref_geolocations[0] + + +def _parse_iso(value: Optional[str]) -> Optional[datetime]: + if not value: + return None + try: + return date_formatter.to_datetime(value) + except (ValueError, AttributeError): + return None + + +def _strip_tz(dt: Optional[datetime]) -> Optional[datetime]: + if dt is None: + return None + return dt.replace(tzinfo=None) if dt.tzinfo is not None else dt + + def _create_base_row_data(database: SensorPositionsDatabase, geolocation, row_hor_ver: str, row_location_id: str, row_description: str) -> Dict: """Create the common row data used by both standard and specific sensors.""" From 31857ffabe94a62bef004980fa71a7e43c0e889b Mon Sep 17 00:00:00 2001 From: Teresa Burlingame Date: Tue, 7 Jul 2026 16:59:54 -0700 Subject: [PATCH 11/16] update to loader --- ...H2oSoilSalinity_position_history_loader.py | 91 ++++++++++++++++++- 1 file changed, 87 insertions(+), 4 deletions(-) diff --git a/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py b/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py index d56bf107c2..04c2be7a67 100644 --- a/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py +++ b/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py @@ -56,6 +56,7 @@ def write_files(*, if generated_at is None: generated_at = datetime.now(tz=timezone.utc) generated_at_str = _fmt_dt(generated_at) + now_naive = _naive(generated_at) installs = get_asset_installs() calibrations = get_calibrations() @@ -100,6 +101,7 @@ def write_files(*, cals_by_asset_period=cals_by_asset_period, allowed_vers=allowed_vers, hor=hor, + now_naive=now_naive, ) if not rows: @@ -122,10 +124,16 @@ def _build_rows(*, geolocations: List[Dict[str, Any]], cals_by_asset_period: Dict[Tuple[int, Optional[datetime], Optional[datetime]], List[Cvald1Calibration]], allowed_vers: Set[str], - hor: str) -> List[Dict[str, Any]]: + hor: str, + now_naive: datetime) -> List[Dict[str, Any]]: """ For a single CFGLOC, produce the full list of (install × geolocation × cal-period × VER) intersection rows. + + :param now_naive: 'Generated-at' timestamp (tz-naive). Any window end that + falls after this is treated as open-ended — PDR uses + placeholder future dates (e.g. 2031-12-31) for still-valid + records, and downstream consumers expect a blank end. """ rows: List[Dict[str, Any]] = [] for install in cfgloc_installs: @@ -142,6 +150,9 @@ def _build_rows(*, if window is None: continue win_start, win_end = window + # Collapse placeholder-future end dates to open-ended. + if win_end is not None and win_end > now_naive: + win_end = None for stream in streams: ver = _stream_to_ver(stream.schema_field_name) if ver is None or str(ver) not in allowed_vers: @@ -150,8 +161,9 @@ def _build_rows(*, rows.append({ 'hor': hor, 'ver': str(ver), - 'position_start_date': _fmt_dt(win_start), - 'position_end_date': _fmt_dt(win_end), + # Keep raw datetimes for the merge pass below; formatted later. + '_win_start': win_start, + '_win_end': win_end, 'x_offset': geo['x_offset'], 'y_offset': geo['y_offset'], 'z_offset': z_offset_adjusted, @@ -171,7 +183,78 @@ def _build_rows(*, 'source_stream': stream.schema_field_name, 'cert_filename': stream.cert_filename, }) - return rows + return _merge_time_ranges(rows) + + +def _merge_time_ranges(rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """ + Consolidate rows that differ only in time range. The loader's intersection + (install × geolocation × calibration) legitimately produces N sub-windows + even when the physical position doesn't change between them (e.g. a + recalibration that leaves cvald1_cm unchanged, or successive asset installs + at the same offsets). We collapse those into a single row per contiguous or + overlapping stretch where the depth-relevant fields are identical. + + Rows with different physical values (any of x/y/z, pitch/roll/azimuth, + reference location, or reference-location time range) stay separate and + keep their own intervals — even if they overlap in time, which surfaces + real data anomalies rather than hiding them. + """ + key_fields = ( + 'hor', 'ver', + 'x_offset', 'y_offset', 'z_offset', + 'pitch', 'roll', 'azimuth', + 'reference_location_id', + 'reference_location_start_date', 'reference_location_end_date', + 'reference_location_latitude', 'reference_location_longitude', + 'reference_location_elevation', + ) + NEG_INF = datetime.min + POS_INF = datetime.max + + groups: Dict[Tuple, List[Dict[str, Any]]] = defaultdict(list) + for row in rows: + groups[tuple(row.get(f) for f in key_fields)].append(row) + + merged: List[Dict[str, Any]] = [] + for _, group_rows in groups.items(): + intervals = [] + for r in group_rows: + s = _naive(r['_win_start']) if r['_win_start'] is not None else NEG_INF + e = _naive(r['_win_end']) if r['_win_end'] is not None else POS_INF + intervals.append((s, e, r)) + intervals.sort(key=lambda x: x[0]) + + cur_start, cur_end, cur_row = intervals[0] + for s, e, r in intervals[1:]: + if s <= cur_end: + if e > cur_end: + cur_end = e + else: + merged.append(_finalize_row(cur_row, cur_start, cur_end, NEG_INF, POS_INF)) + cur_start, cur_end, cur_row = s, e, r + merged.append(_finalize_row(cur_row, cur_start, cur_end, NEG_INF, POS_INF)) + return merged + + +def _finalize_row(row: Dict[str, Any], start: datetime, end: datetime, + neg_inf: datetime, pos_inf: datetime) -> Dict[str, Any]: + """Format the merged interval back into ISO strings; preserve field order.""" + out: Dict[str, Any] = { + 'hor': row['hor'], + 'ver': row['ver'], + 'position_start_date': '' if start == neg_inf else _fmt_dt(start), + 'position_end_date': '' if end == pos_inf else _fmt_dt(end), + } + for k, v in row.items(): + if k in ('hor', 'ver') or k.startswith('_'): + continue + out[k] = v + # Provenance (asset_uid, cvald1_cm, cal_valid_start/end, source_stream, + # cert_filename) is taken from the FIRST sub-interval — a merged row spans + # multiple asset installs / cal periods, so a single value here isn't + # authoritative. Consumers care about position, not provenance. + return out def _flatten_geolocations(feature_collection: Any) -> List[Dict[str, Any]]: From 5e6f221547f58ccc4afe3afa16df38c9764db8a2 Mon Sep 17 00:00:00 2001 From: Teresa Burlingame Date: Tue, 7 Jul 2026 17:17:56 -0700 Subject: [PATCH 12/16] one more update to collapse dates --- ...H2oSoilSalinity_position_history_loader.py | 43 ++++++++----------- 1 file changed, 19 insertions(+), 24 deletions(-) diff --git a/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py b/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py index 04c2be7a67..ea6d43184d 100644 --- a/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py +++ b/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py @@ -188,17 +188,16 @@ def _build_rows(*, def _merge_time_ranges(rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """ - Consolidate rows that differ only in time range. The loader's intersection - (install × geolocation × calibration) legitimately produces N sub-windows - even when the physical position doesn't change between them (e.g. a - recalibration that leaves cvald1_cm unchanged, or successive asset installs - at the same offsets). We collapse those into a single row per contiguous or - overlapping stretch where the depth-relevant fields are identical. - - Rows with different physical values (any of x/y/z, pitch/roll/azimuth, - reference location, or reference-location time range) stay separate and - keep their own intervals — even if they overlap in time, which surfaces - real data anomalies rather than hiding them. + Consolidate rows that differ only in time range into one row per unique + physical position (HOR/VER, offsets, orientation, reference-location). + Each group emits a single row spanning the earliest start to the latest + end across all its sub-windows — even across removal/reinstall gaps, + since a re-installed sensor at the same offsets is the same position + from a sensor_positions.csv consumer's point of view. + + Rows in DIFFERENT groups (e.g. a brief cvald1 anomaly that put a depth at + z=-0.36 for 20 days while the rest of the timeline had z=-0.46) stay + separate, so real position changes still surface. """ key_fields = ( 'hor', 'ver', @@ -218,22 +217,18 @@ def _merge_time_ranges(rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]: merged: List[Dict[str, Any]] = [] for _, group_rows in groups.items(): - intervals = [] + earliest_start = POS_INF + latest_end = NEG_INF + earliest_row = group_rows[0] for r in group_rows: s = _naive(r['_win_start']) if r['_win_start'] is not None else NEG_INF e = _naive(r['_win_end']) if r['_win_end'] is not None else POS_INF - intervals.append((s, e, r)) - intervals.sort(key=lambda x: x[0]) - - cur_start, cur_end, cur_row = intervals[0] - for s, e, r in intervals[1:]: - if s <= cur_end: - if e > cur_end: - cur_end = e - else: - merged.append(_finalize_row(cur_row, cur_start, cur_end, NEG_INF, POS_INF)) - cur_start, cur_end, cur_row = s, e, r - merged.append(_finalize_row(cur_row, cur_start, cur_end, NEG_INF, POS_INF)) + if s < earliest_start: + earliest_start = s + earliest_row = r + if e > latest_end: + latest_end = e + merged.append(_finalize_row(earliest_row, earliest_start, latest_end, NEG_INF, POS_INF)) return merged From fb1172f09942e31743324c36abe3a9cc2902380e Mon Sep 17 00:00:00 2001 From: Teresa Burlingame Date: Wed, 8 Jul 2026 08:43:51 -0700 Subject: [PATCH 13/16] date logic was using cal end not sensor install info --- ...H2oSoilSalinity_position_history_loader.py | 130 ++++++++++-------- 1 file changed, 73 insertions(+), 57 deletions(-) diff --git a/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py b/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py index ea6d43184d..0beecdbbb4 100644 --- a/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py +++ b/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py @@ -3,12 +3,16 @@ concH2oSoilSalinity position-history loader. Queries PDR for the full CFGLOC × asset × calibration × geolocation history for -every enviroscan probe, stitches the intersections, applies the CVALD1 depth -correction (depth_m = z_offset + CVALD1_cm / -100) per VER, and writes one JSON -file per CFGLOC. Consumers (pub_files sensor_positions) read these files instead -of hitting the DB per publish month, so every downloaded month contains the -complete position history — including moves that happened outside that month's -sensor operation window. +every enviroscan probe, picks the authoritative cvald1_cm per (asset_uid, stream) +(highest calibration_id wins — DB insertion order, so newest-inserted supersedes +older/phantom cal records), then stitches (install × geolocation) intersections +and applies the CVALD1 depth correction (depth_m = z_offset + CVALD1_cm / -100) +per VER. Cvald1 is treated as a hardware/depth property that shouldn't slice +position windows — a re-issued cert with a corrected value applies across the +whole install period. Writes one JSON file per CFGLOC. Consumers (pub_files +sensor_positions) read these files instead of hitting the DB per publish month, +so every downloaded month contains the complete position history — including +moves that happened outside that month's sensor operation window. """ import json import sys @@ -66,10 +70,24 @@ def write_files(*, for install in installs: installs_by_cfgloc[install.cfgloc].append(install) - # (asset_uid, valid_start, valid_end) -> [stream calibrations at that period] - cals_by_asset_period: Dict[Tuple[int, Optional[datetime], Optional[datetime]], List[Cvald1Calibration]] = defaultdict(list) + # Pick ONE authoritative cal per (asset_uid, stream): the row with the highest + # calibration_id (the DB's insertion sequence, so newest-inserted wins). cvald1 + # is a hardware/depth property that shouldn't change across recalibrations for + # the same sensor slot; if it does change, the newer cert is the corrected + # value and applies across the whole install window. This intentionally drops + # cal-validity windowing from the position output — position windows come + # from (install × geo) only, so phantom/superseded cal records with overlapping + # validity periods can't slice the output anymore. + latest_cal_by_asset_stream: Dict[Tuple[int, str], Cvald1Calibration] = {} for cal in calibrations: - cals_by_asset_period[(cal.asset_uid, cal.valid_start_time, cal.valid_end_time)].append(cal) + key = (cal.asset_uid, cal.schema_field_name) + existing = latest_cal_by_asset_stream.get(key) + if existing is None or cal.calibration_id > existing.calibration_id: + latest_cal_by_asset_stream[key] = cal + + streams_by_asset: Dict[int, List[Cvald1Calibration]] = defaultdict(list) + for cal in latest_cal_by_asset_stream.values(): + streams_by_asset[cal.asset_uid].append(cal) allowed_vers_by_cfgloc: Dict[str, Set[str]] = defaultdict(set) hor_by_cfgloc: Dict[str, str] = {} @@ -98,7 +116,7 @@ def write_files(*, rows = _build_rows( cfgloc_installs=cfgloc_installs, geolocations=geolocations, - cals_by_asset_period=cals_by_asset_period, + streams_by_asset=streams_by_asset, allowed_vers=allowed_vers, hor=hor, now_naive=now_naive, @@ -122,14 +140,16 @@ def write_files(*, def _build_rows(*, cfgloc_installs: List[AssetInstall], geolocations: List[Dict[str, Any]], - cals_by_asset_period: Dict[Tuple[int, Optional[datetime], Optional[datetime]], List[Cvald1Calibration]], + streams_by_asset: Dict[int, List[Cvald1Calibration]], allowed_vers: Set[str], hor: str, now_naive: datetime) -> List[Dict[str, Any]]: """ - For a single CFGLOC, produce the full list of (install × geolocation × cal-period × VER) - intersection rows. + For a single CFGLOC, produce the (install × geolocation × VER) intersection rows. + Cal-validity dates no longer slice the output — the authoritative cvald1_cm per + (asset_uid, stream) is picked upstream and applied across the entire install window. + :param streams_by_asset: asset_uid -> list of authoritative cals (one per stream). :param now_naive: 'Generated-at' timestamp (tz-naive). Any window end that falls after this is treated as open-ended — PDR uses placeholder future dates (e.g. 2031-12-31) for still-valid @@ -137,52 +157,48 @@ def _build_rows(*, """ rows: List[Dict[str, Any]] = [] for install in cfgloc_installs: - asset_periods = [(period, streams) - for period, streams in cals_by_asset_period.items() - if period[0] == install.asset_uid] + streams = streams_by_asset.get(install.asset_uid, []) + if not streams: + continue for geo in geolocations: - for (_, cal_start, cal_end), streams in asset_periods: - window = _intersect( - (install.install_date, install.remove_date), - (geo['start_date'], geo['end_date']), - (cal_start, cal_end), - ) - if window is None: + window = _intersect( + (install.install_date, install.remove_date), + (geo['start_date'], geo['end_date']), + ) + if window is None: + continue + win_start, win_end = window + # Collapse placeholder-future end dates to open-ended. + if win_end is not None and win_end > now_naive: + win_end = None + for stream in streams: + ver = _stream_to_ver(stream.schema_field_name) + if ver is None or str(ver) not in allowed_vers: continue - win_start, win_end = window - # Collapse placeholder-future end dates to open-ended. - if win_end is not None and win_end > now_naive: - win_end = None - for stream in streams: - ver = _stream_to_ver(stream.schema_field_name) - if ver is None or str(ver) not in allowed_vers: - continue - z_offset_adjusted = round(geo['z_offset'] + stream.cvald1_cm / -100.0, 4) - rows.append({ - 'hor': hor, - 'ver': str(ver), - # Keep raw datetimes for the merge pass below; formatted later. - '_win_start': win_start, - '_win_end': win_end, - 'x_offset': geo['x_offset'], - 'y_offset': geo['y_offset'], - 'z_offset': z_offset_adjusted, - 'pitch': geo['pitch'], - 'roll': geo['roll'], - 'azimuth': geo['azimuth'], - 'reference_location_id': geo['reference_location_id'], - 'reference_location_start_date': geo['reference_location_start_date'], - 'reference_location_end_date': geo['reference_location_end_date'], - 'reference_location_latitude': geo['reference_location_latitude'], - 'reference_location_longitude': geo['reference_location_longitude'], - 'reference_location_elevation': geo['reference_location_elevation'], - 'asset_uid': install.asset_uid, - 'cvald1_cm': stream.cvald1_cm, - 'cal_valid_start': _fmt_dt(cal_start), - 'cal_valid_end': _fmt_dt(cal_end), - 'source_stream': stream.schema_field_name, - 'cert_filename': stream.cert_filename, - }) + z_offset_adjusted = round(geo['z_offset'] + stream.cvald1_cm / -100.0, 4) + rows.append({ + 'hor': hor, + 'ver': str(ver), + # Keep raw datetimes for the merge pass below; formatted later. + '_win_start': win_start, + '_win_end': win_end, + 'x_offset': geo['x_offset'], + 'y_offset': geo['y_offset'], + 'z_offset': z_offset_adjusted, + 'pitch': geo['pitch'], + 'roll': geo['roll'], + 'azimuth': geo['azimuth'], + 'reference_location_id': geo['reference_location_id'], + 'reference_location_start_date': geo['reference_location_start_date'], + 'reference_location_end_date': geo['reference_location_end_date'], + 'reference_location_latitude': geo['reference_location_latitude'], + 'reference_location_longitude': geo['reference_location_longitude'], + 'reference_location_elevation': geo['reference_location_elevation'], + 'asset_uid': install.asset_uid, + 'cvald1_cm': stream.cvald1_cm, + 'source_stream': stream.schema_field_name, + 'cert_filename': stream.cert_filename, + }) return _merge_time_ranges(rows) From 3af42f78ce9257edf3ec889395567be225808201 Mon Sep 17 00:00:00 2001 From: Teresa Burlingame Date: Wed, 8 Jul 2026 09:56:19 -0700 Subject: [PATCH 14/16] tests for swc_loc_depths --- .../tests/test_position_history_loader.py | 539 ++++++++++++++++++ .../pipe_list_concH2oSoilSalinity.txt | 1 + 2 files changed, 540 insertions(+) create mode 100644 modules/concH2oSoilSalinity_position_history_loader/tests/test_position_history_loader.py diff --git a/modules/concH2oSoilSalinity_position_history_loader/tests/test_position_history_loader.py b/modules/concH2oSoilSalinity_position_history_loader/tests/test_position_history_loader.py new file mode 100644 index 0000000000..3d36273f32 --- /dev/null +++ b/modules/concH2oSoilSalinity_position_history_loader/tests/test_position_history_loader.py @@ -0,0 +1,539 @@ +#!/usr/bin/env python3 +""" +Unit tests for concH2oSoilSalinity_position_history_loader. + +Cover the pieces the loader logic hinges on: + - stream -> VER mapping + - date-range intersection + - MultiPoint / Point unwrap for reference-location geometry + - authoritative-cal selection (highest calibration_id per (asset, stream)) + - future-date placeholder collapse + - extent-based row merge (same physical position collapses across gaps) + - end-to-end write_files against mocked PDR callables +""" +import json +from datetime import datetime, timezone +from pathlib import Path + +from pyfakefs.fake_filesystem_unittest import TestCase + +from data_access.types.asset_install import AssetInstall +from data_access.types.cfgloc_ver import CfglocVer +from data_access.types.cvald1_calibration import Cvald1Calibration +from concH2oSoilSalinity_position_history_loader import ( + concH2oSoilSalinity_position_history_loader as loader, +) + + +def dt(*args) -> datetime: + return datetime(*args) + + +class StreamToVerTest(TestCase): + def test_maps_rawvswc0_to_501(self): + self.assertEqual(loader._stream_to_ver('rawVSWC0'), 501) + + def test_maps_rawvswc7_to_508(self): + self.assertEqual(loader._stream_to_ver('rawVSWC7'), 508) + + def test_returns_none_for_vsic(self): + # VSIC odd streams are not in the naming pattern we expect + self.assertIsNone(loader._stream_to_ver('rawVSIC0')) + + def test_returns_none_for_empty_and_junk(self): + self.assertIsNone(loader._stream_to_ver(None)) + self.assertIsNone(loader._stream_to_ver('')) + self.assertIsNone(loader._stream_to_ver('rawVSWC')) # empty suffix + self.assertIsNone(loader._stream_to_ver('rawVSWCX')) # non-digit + + +class IntersectTest(TestCase): + def test_overlapping_ranges(self): + result = loader._intersect( + (dt(2020, 1, 1), dt(2024, 1, 1)), + (dt(2022, 1, 1), dt(2026, 1, 1)), + ) + self.assertEqual(result, (dt(2022, 1, 1), dt(2024, 1, 1))) + + def test_open_ended_left(self): + result = loader._intersect( + (None, dt(2024, 1, 1)), + (dt(2022, 1, 1), None), + ) + self.assertEqual(result, (dt(2022, 1, 1), dt(2024, 1, 1))) + + def test_disjoint_returns_none(self): + self.assertIsNone( + loader._intersect( + (dt(2020, 1, 1), dt(2021, 1, 1)), + (dt(2022, 1, 1), dt(2023, 1, 1)), + ) + ) + + def test_touching_at_boundary_returns_none(self): + # start >= end -> None (half-open) + self.assertIsNone( + loader._intersect( + (dt(2020, 1, 1), dt(2022, 1, 1)), + (dt(2022, 1, 1), dt(2024, 1, 1)), + ) + ) + + def test_strips_tz_before_compare(self): + aware = datetime(2022, 1, 1, tzinfo=timezone.utc) + naive = dt(2020, 1, 1) + result = loader._intersect((naive, None), (aware, None)) + # tz stripped internally; earliest common start is the aware value w/o tz + self.assertEqual(result[0], dt(2022, 1, 1)) + self.assertIsNone(result[1]) + + +class ExtractPointTest(TestCase): + def test_point_shape(self): + lon, lat, elev = loader._extract_point( + {'type': 'Point', 'coordinates': [-83.5, 35.68, 573.32]} + ) + self.assertEqual((lon, lat, elev), (-83.5, 35.68, 573.32)) + + def test_multipoint_unwrap_one_level(self): + lon, lat, elev = loader._extract_point( + {'type': 'MultiPoint', 'coordinates': [[-83.5, 35.68, 573.32]]} + ) + self.assertEqual((lon, lat, elev), (-83.5, 35.68, 573.32)) + + def test_missing_elevation(self): + lon, lat, elev = loader._extract_point( + {'type': 'Point', 'coordinates': [-83.5, 35.68]} + ) + self.assertEqual(lon, -83.5) + self.assertEqual(lat, 35.68) + self.assertIsNone(elev) + + def test_none_and_empty(self): + self.assertEqual(loader._extract_point(None), (None, None, None)) + self.assertEqual( + loader._extract_point({'type': 'Point', 'coordinates': []}), + (None, None, None), + ) + + +class MergeTimeRangesTest(TestCase): + """`_merge_time_ranges` is the piece that keeps the CSV from listing every + install/reinstall boundary as its own row when the physical position hasn't + changed. The behavior we want is *extent-based*: min(start), max(end) + across all rows in a group.""" + + @staticmethod + def _base_row(**overrides): + row = { + 'hor': '004', 'ver': '501', + '_win_start': None, '_win_end': None, + 'x_offset': 0.0, 'y_offset': 0.0, 'z_offset': -0.045, + 'pitch': 1.0, 'roll': 0.0, 'azimuth': 130.0, + 'reference_location_id': 'SOILPL999', + 'reference_location_start_date': None, + 'reference_location_end_date': None, + 'reference_location_latitude': 35.68, + 'reference_location_longitude': -83.50, + 'reference_location_elevation': 573.32, + 'asset_uid': 1, + 'cvald1_cm': 4.5, + 'source_stream': 'rawVSWC0', + 'cert_filename': 'A.xml', + } + row.update(overrides) + return row + + def test_two_rows_same_position_collapse_across_gap(self): + rows = [ + self._base_row(_win_start=dt(2020, 1, 1), _win_end=dt(2021, 6, 30)), + self._base_row(_win_start=dt(2021, 7, 1), _win_end=dt(2024, 1, 1)), + ] + merged = loader._merge_time_ranges(rows) + self.assertEqual(len(merged), 1) + self.assertEqual(merged[0]['position_start_date'], '2020-01-01T00:00:00Z') + self.assertEqual(merged[0]['position_end_date'], '2024-01-01T00:00:00Z') + + def test_different_z_offset_stays_separate(self): + rows = [ + self._base_row(_win_start=dt(2020, 1, 1), _win_end=dt(2021, 6, 30), z_offset=-0.045), + self._base_row(_win_start=dt(2021, 7, 1), _win_end=dt(2024, 1, 1), z_offset=-0.055), + ] + merged = loader._merge_time_ranges(rows) + self.assertEqual(len(merged), 2) + z_values = sorted(r['z_offset'] for r in merged) + self.assertEqual(z_values, [-0.055, -0.045]) + + def test_open_start_and_open_end_map_to_blank(self): + rows = [ + self._base_row(_win_start=None, _win_end=None), + ] + merged = loader._merge_time_ranges(rows) + self.assertEqual(merged[0]['position_start_date'], '') + self.assertEqual(merged[0]['position_end_date'], '') + + def test_provenance_from_earliest_sub_interval(self): + # A merged row spans multiple asset installs; the provenance fields we + # carry (asset_uid, cvald1_cm, cert_filename) come from the earliest row. + # If we ever want per-window provenance, this test will catch the change. + rows = [ + self._base_row(_win_start=dt(2020, 1, 1), _win_end=dt(2021, 6, 30), + asset_uid=100, cert_filename='OLD.xml'), + self._base_row(_win_start=dt(2021, 7, 1), _win_end=dt(2024, 1, 1), + asset_uid=200, cert_filename='NEW.xml'), + ] + merged = loader._merge_time_ranges(rows) + self.assertEqual(len(merged), 1) + self.assertEqual(merged[0]['asset_uid'], 100) + self.assertEqual(merged[0]['cert_filename'], 'OLD.xml') + + def test_finalize_row_field_order(self): + # position_start_date and position_end_date must come right after hor/ver + # so the emitted JSON reads naturally. Everything else preserves input order. + row = self._base_row(_win_start=dt(2020, 1, 1), _win_end=dt(2021, 6, 30)) + merged = loader._merge_time_ranges([row]) + keys = list(merged[0].keys()) + self.assertEqual(keys[:4], ['hor', 'ver', 'position_start_date', 'position_end_date']) + self.assertNotIn('_win_start', keys) + self.assertNotIn('_win_end', keys) + + +class BuildRowsTest(TestCase): + """`_build_rows` is the (install × geo) intersection + cvald1-depth-adjust + + future-date collapse. We stub out streams_by_asset so this exercises the + row construction logic without needing the full PDR fake.""" + + @staticmethod + def _install(asset_uid=1, install_date=dt(2020, 1, 1), remove_date=None): + return AssetInstall( + cfgloc='CFGLOC999999', + cfgloc_description='XYZ SP1', + nam_locn_id=1, + asset_uid=asset_uid, + install_date=install_date, + remove_date=remove_date, + ) + + @staticmethod + def _geo(start=dt(2010, 1, 1), end=None, z=0.005): + return { + 'start_date': start, 'end_date': end, + 'x_offset': -0.46, 'y_offset': -1.09, 'z_offset': z, + 'pitch': 1.0, 'roll': 0.0, 'azimuth': 130.0, + 'reference_location_id': 'SOILPL999', + 'reference_location_start_date': dt(2010, 1, 1), + 'reference_location_end_date': None, + 'reference_location_latitude': 35.68, + 'reference_location_longitude': -83.50, + 'reference_location_elevation': 573.32, + } + + @staticmethod + def _cal(asset_uid=1, cid=100, stream='rawVSWC0', cvald1=5.0): + return Cvald1Calibration( + asset_uid=asset_uid, + calibration_id=cid, + sensor_stream_num=0, + schema_field_name=stream, + valid_start_time=dt(2010, 1, 1), + valid_end_time=None, + cert_filename='CERT.xml', + cvald1_cm=cvald1, + ) + + def test_z_offset_adjusted_by_cvald1(self): + install = self._install(remove_date=dt(2024, 1, 1)) + geo = self._geo(z=0.008) + rows = loader._build_rows( + cfgloc_installs=[install], + geolocations=[geo], + streams_by_asset={1: [self._cal(cvald1=7.0)]}, + allowed_vers={'501'}, + hor='004', + now_naive=dt(2026, 7, 8), + ) + # 0.008 + 7/-100 = -0.062 + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]['z_offset'], -0.062) + + def test_future_end_date_collapses_to_blank(self): + # PDR placeholder future dates (2031-12-31) should render as open-ended. + install = self._install(remove_date=dt(2031, 12, 31)) + geo = self._geo(end=dt(2031, 12, 31)) + rows = loader._build_rows( + cfgloc_installs=[install], + geolocations=[geo], + streams_by_asset={1: [self._cal()]}, + allowed_vers={'501'}, + hor='004', + now_naive=dt(2026, 7, 8), + ) + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]['position_end_date'], '') + + def test_ver_not_in_allowed_set_is_skipped(self): + install = self._install() + geo = self._geo() + rows = loader._build_rows( + cfgloc_installs=[install], + geolocations=[geo], + streams_by_asset={1: [self._cal(stream='rawVSWC7')]}, # -> VER 508 + allowed_vers={'501', '502'}, # 508 not allowed + hor='004', + now_naive=dt(2026, 7, 8), + ) + self.assertEqual(rows, []) + + def test_asset_with_no_streams_is_skipped(self): + install = self._install(asset_uid=999) + rows = loader._build_rows( + cfgloc_installs=[install], + geolocations=[self._geo()], + streams_by_asset={1: [self._cal()]}, # no key for 999 + allowed_vers={'501'}, + hor='004', + now_naive=dt(2026, 7, 8), + ) + self.assertEqual(rows, []) + + def test_disjoint_install_and_geo_produces_no_rows(self): + install = self._install(install_date=dt(2020, 1, 1), remove_date=dt(2020, 6, 30)) + geo = self._geo(start=dt(2021, 1, 1), end=dt(2022, 1, 1)) + rows = loader._build_rows( + cfgloc_installs=[install], + geolocations=[geo], + streams_by_asset={1: [self._cal()]}, + allowed_vers={'501'}, + hor='004', + now_naive=dt(2026, 7, 8), + ) + self.assertEqual(rows, []) + + def test_two_geos_two_streams_produces_four_rows(self): + # Two geolocation windows × two allowed VSWC streams = four output rows, + # each with its own z_offset depending on cvald1. + install = self._install(remove_date=dt(2030, 1, 1)) + geos = [ + self._geo(start=dt(2010, 1, 1), end=dt(2023, 1, 1), z=0.005), + self._geo(start=dt(2023, 1, 1), end=None, z=0.008), + ] + streams = [ + self._cal(cid=1, stream='rawVSWC0', cvald1=5.0), + self._cal(cid=2, stream='rawVSWC1', cvald1=15.0), # VER 502 + ] + rows = loader._build_rows( + cfgloc_installs=[install], + geolocations=geos, + streams_by_asset={1: streams}, + allowed_vers={'501', '502'}, + hor='004', + now_naive=dt(2026, 7, 8), + ) + # 2 geos × 2 streams = 4 unique (position, ver) tuples -> 4 merged groups + self.assertEqual(len(rows), 4) + vers = sorted(r['ver'] for r in rows) + self.assertEqual(vers, ['501', '501', '502', '502']) + + +class AuthoritativeCalSelectionTest(TestCase): + """`write_files` reduces multiple Cvald1Calibration rows for the same + (asset_uid, schema_field_name) to the single row with the highest + calibration_id. This is Option B: the newer-inserted cert wins across the + whole install window; older/phantom records don't slice output.""" + + def setUp(self): + self.setUpPyfakefs() + self.out_path = Path('/out') + self.err_path = Path('/err') + self.fs.create_dir(self.out_path) + self.fs.create_dir(self.err_path) + + def _run_loader(self, calibrations): + installs = [AssetInstall( + cfgloc='CFGLOC105360', cfgloc_description='BONA SP1', + nam_locn_id=105360, asset_uid=46446, + install_date=dt(2020, 3, 11), remove_date=None, + )] + cfgloc_vers = [CfglocVer(cfgloc='CFGLOC105360', hor='004', ver='501', + group_name='conc-h2o-soil-salinity-split_BONA004501')] + geolocations = _feature_collection([_feature(x_offset=0.0, y_offset=0.0, z_offset=0.005, + start_date='2020-03-11T00:00:00Z')]) + loader.write_files( + get_asset_installs=lambda: installs, + get_calibrations=lambda: calibrations, + get_cfgloc_vers=lambda: cfgloc_vers, + get_geolocations=lambda _nam_id: geolocations, + get_parents=lambda _nam_id: {'site': (1, 'BONA')}, + out_path=self.out_path, + err_path=self.err_path, + generated_at=datetime(2026, 7, 8, tzinfo=timezone.utc), + ) + json_path = self.out_path / 'enviroscan' / 'CFGLOC105360' / 'position_history' / 'CFGLOC105360_history.json' + with open(json_path) as fp: + return json.load(fp) + + def test_highest_calibration_id_wins(self): + # Real scenario from CFGLOC105360: two cal records for the same + # (asset, stream). The older one has cvald1=5.0 (correct value), the newer + # one has cvald1=99.0. Highest calibration_id must win. + calibrations = [ + Cvald1Calibration(asset_uid=46446, calibration_id=100, + sensor_stream_num=0, schema_field_name='rawVSWC0', + valid_start_time=dt(2020, 2, 19), valid_end_time=dt(2021, 4, 14), + cert_filename='OLD.xml', cvald1_cm=5.0), + Cvald1Calibration(asset_uid=46446, calibration_id=200, + sensor_stream_num=0, schema_field_name='rawVSWC0', + valid_start_time=dt(2021, 10, 13), valid_end_time=dt(2031, 12, 31), + cert_filename='NEW.xml', cvald1_cm=99.0), + ] + payload = self._run_loader(calibrations) + # 0.005 + 99/-100 = -0.985 + self.assertEqual(len(payload['rows']), 1) + self.assertEqual(payload['rows'][0]['cvald1_cm'], 99.0) + self.assertEqual(payload['rows'][0]['z_offset'], -0.985) + self.assertEqual(payload['rows'][0]['cert_filename'], 'NEW.xml') + + def test_single_row_per_stream_produces_one_span(self): + # Two cal records with the same cvald1 value must not slice the output. + # Pre-Option-B this would produce two rows split on the cal boundary. + calibrations = [ + Cvald1Calibration(asset_uid=46446, calibration_id=100, + sensor_stream_num=0, schema_field_name='rawVSWC0', + valid_start_time=dt(2020, 2, 19), valid_end_time=dt(2021, 4, 14), + cert_filename='OLD.xml', cvald1_cm=5.0), + Cvald1Calibration(asset_uid=46446, calibration_id=200, + sensor_stream_num=0, schema_field_name='rawVSWC0', + valid_start_time=dt(2021, 10, 13), valid_end_time=dt(2031, 12, 31), + cert_filename='NEW.xml', cvald1_cm=5.0), + ] + payload = self._run_loader(calibrations) + self.assertEqual(len(payload['rows']), 1) + self.assertEqual(payload['rows'][0]['position_end_date'], '') # future end -> blank + + +class EndToEndWriteFilesTest(TestCase): + """Full loader run over a synthetic CFGLOC with two geolocation windows + and one asset install. Verifies the emitted JSON has the shape pub_files + expects and matches the design doc's worked scenario.""" + + def setUp(self): + self.setUpPyfakefs() + self.out_path = Path('/out') + self.err_path = Path('/err') + self.fs.create_dir(self.out_path) + self.fs.create_dir(self.err_path) + + def test_writes_one_json_per_cfgloc_with_expected_structure(self): + installs = [AssetInstall( + cfgloc='CFGLOC999999', cfgloc_description='XYZ SP1 EnviroSCAN', + nam_locn_id=42, asset_uid=12294, + install_date=dt(2010, 1, 1), remove_date=None, + )] + cfgloc_vers = [ + CfglocVer(cfgloc='CFGLOC999999', hor='004', ver='501', + group_name='conc-h2o-soil-salinity-split_XYZ004501'), + ] + calibrations = [ + Cvald1Calibration(asset_uid=12294, calibration_id=1, + sensor_stream_num=0, schema_field_name='rawVSWC0', + valid_start_time=dt(2010, 1, 1), valid_end_time=None, + cert_filename='A.xml', cvald1_cm=5.0), + ] + geolocations = _feature_collection([ + _feature(start_date='2010-01-01T00:00:00Z', + end_date='2023-01-19T00:00:00Z', + x_offset=-0.46, y_offset=-1.09, z_offset=0.005), + _feature(start_date='2023-01-19T00:00:00Z', + end_date=None, + x_offset=2.36, y_offset=-1.13, z_offset=0.008), + ]) + loader.write_files( + get_asset_installs=lambda: installs, + get_calibrations=lambda: calibrations, + get_cfgloc_vers=lambda: cfgloc_vers, + get_geolocations=lambda _id: geolocations, + get_parents=lambda _id: {'site': (1, 'XYZ')}, + out_path=self.out_path, + err_path=self.err_path, + generated_at=datetime(2026, 7, 8, tzinfo=timezone.utc), + ) + + json_path = self.out_path / 'enviroscan' / 'CFGLOC999999' / 'position_history' / 'CFGLOC999999_history.json' + self.assertTrue(json_path.is_file()) + payload = json.loads(json_path.read_text()) + + self.assertEqual(payload['source_type'], 'enviroscan') + self.assertEqual(payload['cfgloc'], 'CFGLOC999999') + self.assertEqual(payload['site'], 'XYZ') + self.assertEqual(payload['generated_at'], '2026-07-08T00:00:00Z') + # Two distinct z-offset positions -> two rows after merge + rows = payload['rows'] + self.assertEqual(len(rows), 2) + z_values = sorted(r['z_offset'] for r in rows) + # 0.005 + 5/-100 = -0.045; 0.008 + 5/-100 = -0.042 + self.assertEqual(z_values, [-0.045, -0.042]) + for row in rows: + self.assertEqual(row['hor'], '004') + self.assertEqual(row['ver'], '501') + self.assertIn('position_start_date', row) + self.assertIn('position_end_date', row) + + def test_cfgloc_with_no_configured_ver_is_skipped(self): + installs = [AssetInstall( + cfgloc='CFGLOC_NOVER', cfgloc_description='decommissioned', + nam_locn_id=99, asset_uid=1, + install_date=dt(2010, 1, 1), remove_date=dt(2015, 1, 1), + )] + calibrations = [Cvald1Calibration( + asset_uid=1, calibration_id=1, sensor_stream_num=0, + schema_field_name='rawVSWC0', valid_start_time=dt(2010, 1, 1), + valid_end_time=None, cert_filename='A.xml', cvald1_cm=5.0, + )] + loader.write_files( + get_asset_installs=lambda: installs, + get_calibrations=lambda: calibrations, + get_cfgloc_vers=lambda: [], # nothing configured -> skip + get_geolocations=lambda _id: _feature_collection([]), + get_parents=lambda _id: {'site': (1, 'XYZ')}, + out_path=self.out_path, + err_path=self.err_path, + generated_at=datetime(2026, 7, 8, tzinfo=timezone.utc), + ) + # No JSON should be written for a CFGLOC missing from Q4 + self.assertFalse( + (self.out_path / 'enviroscan' / 'CFGLOC_NOVER').exists() + ) + + +# --- helpers ----------------------------------------------------------------- + +def _feature(*, start_date, end_date=None, x_offset=0.0, y_offset=0.0, z_offset=0.0, + alpha=1.0, beta=0.0, gamma=130.0): + """Build a Feature dict shaped like get_named_location_geolocations' output.""" + return { + 'type': 'Feature', + 'geometry': {'type': 'Point', 'coordinates': [0.0, 0.0, 0.0]}, + 'properties': { + 'start_date': start_date, 'end_date': end_date, + 'alpha': alpha, 'beta': beta, 'gamma': gamma, + 'x_offset': x_offset, 'y_offset': y_offset, 'z_offset': z_offset, + 'reference_location': { + 'type': 'Feature', 'geometry': None, + 'properties': { + 'name': 'SOILPL999', + 'locations': { + 'type': 'FeatureCollection', + 'features': [{ + 'type': 'Feature', + 'geometry': {'type': 'Point', 'coordinates': [-83.5, 35.68, 573.32]}, + 'properties': {'start_date': '2010-01-01T00:00:00Z', 'end_date': None}, + }], + }, + }, + }, + }, + } + + +def _feature_collection(features): + return {'type': 'FeatureCollection', 'features': features} diff --git a/pipe/concH2oSoilSalinity/pipe_list_concH2oSoilSalinity.txt b/pipe/concH2oSoilSalinity/pipe_list_concH2oSoilSalinity.txt index 0463219440..e2fcd6334f 100644 --- a/pipe/concH2oSoilSalinity/pipe_list_concH2oSoilSalinity.txt +++ b/pipe/concH2oSoilSalinity/pipe_list_concH2oSoilSalinity.txt @@ -12,6 +12,7 @@ concH2oSoilSalinity_stats_group_and_compute.yaml concH2oSoilSalinity_qm_join_stats.yaml concH2oSoilSalinity_group_split.yaml concH2oSoilSalinity_level1_group_consolidate_srf.yaml +concH2oSoilSalinity_position_history_loader.yaml concH2oSoilSalinity_cron_monthly_and_pub_control.yaml concH2oSoilSalinity_pub_group.yaml concH2oSoilSalinity_pub_format_and_package.yaml From fddeb90208de018c95746cc600aaf6b83a4d259a Mon Sep 17 00:00:00 2001 From: Teresa Burlingame Date: Wed, 8 Jul 2026 10:02:12 -0700 Subject: [PATCH 15/16] remove change from pr --- pipe/concH2oSoilSalinity/pipe_list_concH2oSoilSalinity.txt | 1 - 1 file changed, 1 deletion(-) diff --git a/pipe/concH2oSoilSalinity/pipe_list_concH2oSoilSalinity.txt b/pipe/concH2oSoilSalinity/pipe_list_concH2oSoilSalinity.txt index e2fcd6334f..0463219440 100644 --- a/pipe/concH2oSoilSalinity/pipe_list_concH2oSoilSalinity.txt +++ b/pipe/concH2oSoilSalinity/pipe_list_concH2oSoilSalinity.txt @@ -12,7 +12,6 @@ concH2oSoilSalinity_stats_group_and_compute.yaml concH2oSoilSalinity_qm_join_stats.yaml concH2oSoilSalinity_group_split.yaml concH2oSoilSalinity_level1_group_consolidate_srf.yaml -concH2oSoilSalinity_position_history_loader.yaml concH2oSoilSalinity_cron_monthly_and_pub_control.yaml concH2oSoilSalinity_pub_group.yaml concH2oSoilSalinity_pub_format_and_package.yaml From 775d29a12f06dca4c6419ffad4f74767f35b4a7d Mon Sep 17 00:00:00 2001 From: Teresa Burlingame <55860349+burlingamet@users.noreply.github.com> Date: Wed, 8 Jul 2026 10:19:39 -0700 Subject: [PATCH 16/16] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- modules/concH2oSoilSalinity_position_history_loader/Dockerfile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/modules/concH2oSoilSalinity_position_history_loader/Dockerfile b/modules/concH2oSoilSalinity_position_history_loader/Dockerfile index a1748a8748..e1e607b979 100644 --- a/modules/concH2oSoilSalinity_position_history_loader/Dockerfile +++ b/modules/concH2oSoilSalinity_position_history_loader/Dockerfile @@ -4,7 +4,7 @@ # Must be built from the project root so common/ and data_access/ are in context: # docker build --no-cache \ # -t concH2oSoilSalinity_position_history_loader:latest \ -# -f modules/concH2oSoilSalinity_position_history_loader/Dockerfile modules +# -f modules/concH2oSoilSalinity_position_history_loader/Dockerfile . # ### FROM registry.access.redhat.com/ubi8/ubi-minimal