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: diff --git a/modules/concH2oSoilSalinity_position_history_loader/Dockerfile b/modules/concH2oSoilSalinity_position_history_loader/Dockerfile new file mode 100644 index 0000000000..e1e607b979 --- /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 . +# +### +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..0beecdbbb4 --- /dev/null +++ b/modules/concH2oSoilSalinity_position_history_loader/concH2oSoilSalinity_position_history_loader.py @@ -0,0 +1,422 @@ +#!/usr/bin/env python3 +""" +concH2oSoilSalinity position-history loader. + +Queries PDR for the full CFGLOC × asset × calibration × geolocation history for +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 +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) + now_naive = _naive(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) + + # 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: + 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] = {} + 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, + streams_by_asset=streams_by_asset, + allowed_vers=allowed_vers, + hor=hor, + now_naive=now_naive, + ) + + 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]], + 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 (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 + records, and downstream consumers expect a blank end. + """ + rows: List[Dict[str, Any]] = [] + for install in cfgloc_installs: + streams = streams_by_asset.get(install.asset_uid, []) + if not streams: + continue + for geo in geolocations: + 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 + 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) + + +def _merge_time_ranges(rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """ + 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', + '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(): + 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 + 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 + + +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]]: + """ + 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. + + 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: + 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]) + 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..a36f9890a4 --- /dev/null +++ b/modules/concH2oSoilSalinity_position_history_loader/requirements.txt @@ -0,0 +1,4 @@ +structlog==21.5.0 +environs==11.0.0 +marshmallow==3.21.3 +geojson==2.4.1 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/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..1eb4f4f116 --- /dev/null +++ b/modules/data_access/get_enviroscan_cvald1_calibrations.py @@ -0,0 +1,64 @@ +#!/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() + # 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 distinct + 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 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."""