diff --git a/.github/app.template.yaml b/.github/app.template.yaml index 619ba4cc5..1fe3b99a1 100644 --- a/.github/app.template.yaml +++ b/.github/app.template.yaml @@ -2,12 +2,18 @@ service: ${SERVICE_NAME} runtime: python313 entrypoint: ${ENTRYPOINT} service_account: "${CLOUD_SQL_USER}.gserviceaccount.com" -instance_class: F4 +# F4_1G (2 GB) not F4 (1 GB): the entrypoint runs gunicorn -w 8, and eight +# workers each importing the full stack (sqlalchemy + geoalchemy2 + shapely + +# cloud-sql connector + pygeoapi) exceeded 1 GB. App Engine logged "using too +# much memory and was terminated" ~44x/3h and cycled workers continuously +# (304 "Booting worker", 297 SIGTERM in 3h), so every request re-cold-loaded. +instance_class: F4_1G inbound_services: - warmup automatic_scaling: min_instances: ${MIN_INSTANCES} max_instances: ${MAX_INSTANCES} + max_concurrent_requests: 6 handlers: - url: /.* secure: always diff --git a/.github/workflows/CD_production.yml b/.github/workflows/CD_production.yml index 45f8dd2a8..a043f0b16 100644 --- a/.github/workflows/CD_production.yml +++ b/.github/workflows/CD_production.yml @@ -123,8 +123,8 @@ jobs: run: | export MAX_INSTANCES="10" export SERVICE_NAME="ocotillo-api" - export ENTRYPOINT="gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app" - export MIN_INSTANCES="0" + export ENTRYPOINT="gunicorn -w 8 -k uvicorn.workers.UvicornWorker main:app" + export MIN_INSTANCES="1" envsubst < .github/app.template.yaml > app.yaml - name: Deploy to Google Cloud diff --git a/.github/workflows/release-please.yml b/.github/workflows/release-please.yml index f6cd5bb7e..c6fb52863 100644 --- a/.github/workflows/release-please.yml +++ b/.github/workflows/release-please.yml @@ -14,8 +14,13 @@ jobs: release-please: runs-on: ubuntu-latest outputs: - release_created: ${{ steps.release.outputs.release_created }} - tag_name: ${{ steps.release.outputs.tag_name }} + # In manifest mode, release-please-action exposes the unprefixed + # `release_created` boolean but only a PATH-SCOPED `.--tag_name` (the + # unprefixed `tag_name` comes through empty). Fall back to the path-scoped + # key so the tag reaches CD_production; without it the deploy job's + # `startsWith(inputs.tag_name, 'v')` gate silently skips the release. + release_created: ${{ steps.release.outputs.release_created || steps.release.outputs['.--release_created'] }} + tag_name: ${{ steps.release.outputs.tag_name || steps.release.outputs['.--tag_name'] }} steps: - id: release uses: googleapis/release-please-action@v5 diff --git a/.release-please-manifest.json b/.release-please-manifest.json index 5fdd88304..a7fd6a0f1 100644 --- a/.release-please-manifest.json +++ b/.release-please-manifest.json @@ -1,3 +1,3 @@ { - ".": "1.1.0" + ".": "1.1.5" } diff --git a/CHANGELOG.md b/CHANGELOG.md index 270193b5e..fa65cb5e0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,47 @@ # Changelog +## [1.1.5](https://github.com/DataIntegrationGroup/OcotilloAPI/compare/v1.1.4...v1.1.5) (2026-07-07) + + +### Bug Fixes + +* **deploy:** raise instance class to F4_1G to stop OOM instance churn ([ec012d6](https://github.com/DataIntegrationGroup/OcotilloAPI/commit/ec012d60ff8b985e89c8a2ca4167b4f24cae8540)) + +## [1.1.4](https://github.com/DataIntegrationGroup/OcotilloAPI/compare/v1.1.3...v1.1.4) (2026-07-07) + + +### Bug Fixes + +* **ci:** deploy on inline workflow_call to CD (Production) ([8371f64](https://github.com/DataIntegrationGroup/OcotilloAPI/commit/8371f646b08363a578d339d99da8ebf5863d21c9)) +* **ci:** deploy on inline workflow_call to CD (Production) ([1a11ee9](https://github.com/DataIntegrationGroup/OcotilloAPI/commit/1a11ee9e5237d33ce7b9ee7f093dad5e023436c6)) + +## [1.1.3](https://github.com/DataIntegrationGroup/OcotilloAPI/compare/v1.1.2...v1.1.3) (2026-07-06) + + +### Bug Fixes + +* correct shapefile DBF schema and clean up temp dir on failure ([f63d9ce](https://github.com/DataIntegrationGroup/OcotilloAPI/commit/f63d9ce2d28ba190b63628f40c61edee09ac1a5e)) +* drop yield_per in get_thing_features (incompatible with unique) ([fa727c4](https://github.com/DataIntegrationGroup/OcotilloAPI/commit/fa727c42898e5d8791f248ff4dc0b5f43c7b3da2)) +* stop per-request OOM on /geospatial export endpoint ([d3a1358](https://github.com/DataIntegrationGroup/OcotilloAPI/commit/d3a1358f54b9b6d02a6187d0004ba44b2b365ed0)) +* stop per-request OOM on /geospatial export endpoint ([2ed12b6](https://github.com/DataIntegrationGroup/OcotilloAPI/commit/2ed12b602a080d98e2025fc0e4edefdc341412ba)) +* stream get_thing_features with yield_per instead of buffering ([877fff8](https://github.com/DataIntegrationGroup/OcotilloAPI/commit/877fff833755034e9c0f1b67ed3d6c74a92d4084)) + +## [1.1.2](https://github.com/DataIntegrationGroup/OcotilloAPI/compare/v1.1.1...v1.1.2) (2026-07-06) + + +### Bug Fixes + +* **deploy:** align hotfix migration head with production DB (unblock v1.1.x deploy) ([1ee8a4b](https://github.com/DataIntegrationGroup/OcotilloAPI/commit/1ee8a4b893027b851450d6ed98c811aea283c365)) +* **deploy:** align hotfix migration head with production DB + fix release tag passthrough ([790377f](https://github.com/DataIntegrationGroup/OcotilloAPI/commit/790377f53e57c9f7671d45efbcab617189f2af36)) + +## [1.1.1](https://github.com/DataIntegrationGroup/OcotilloAPI/compare/v1.1.0...v1.1.1) (2026-07-06) + + +### Bug Fixes + +* **deploy:** prevent App Engine request starvation under burst load ([fec7dbd](https://github.com/DataIntegrationGroup/OcotilloAPI/commit/fec7dbd2e0f59d6ce1c3319cda9fded5c6051eb8)) +* **deploy:** prevent App Engine request starvation under burst load ([385c974](https://github.com/DataIntegrationGroup/OcotilloAPI/commit/385c9743185520b82034f0b1549d51c57781f9e9)) + ## [1.1.0](https://github.com/DataIntegrationGroup/OcotilloAPI/compare/v1.0.0...v1.1.0) (2026-06-08) diff --git a/alembic/versions/c0d1e2f3a4b5_nmw_mirror_tables.py b/alembic/versions/c0d1e2f3a4b5_nmw_mirror_tables.py new file mode 100644 index 000000000..f59a760ca --- /dev/null +++ b/alembic/versions/c0d1e2f3a4b5_nmw_mirror_tables.py @@ -0,0 +1,723 @@ +"""NMW staging mirror tables and FK constraints + +Revision ID: c0d1e2f3a4b5 +Revises: t6u7v8w9x0y1 +Create Date: 2026-06-22 + +1:1 staging mirror of the legacy NM_Wells SQL Server tables needed for the +geothermal OGC collections (see db/nmw_legacy.py and docs/nm_wells-migration.md). +Faithful, column-for-column copies; the transform into the Ocotillo data model +is a later phase. + +Core well tables: + tbl_well_locations -> NMW_WellLocations + tbl_well_headers -> NMW_WellHeaders + tbl_well_records -> NMW_WellRecords + tbl_well_z_datum -> NMW_WellZDatum + tbl_well_samples -> NMW_WellSamples + +Geothermal tables: + tbl_gt_bht_headers -> NMW_GtBhtHeaders + tbl_gt_bht_data -> NMW_GtBhtData + tbl_ws_intervals -> NMW_WsIntervals + tbl_gt_conductivity -> NMW_GtConductivity + tbl_gt_heat_flow -> NMW_GtHeatFlow + tbl_gt_sum_heat_flow -> NMW_GtSumHeatFlow + tbl_gt_temp_depths -> NMW_GtTempDepths + +Drill Stem Test tables: + tbl_ws_dst_headers -> NMW_WsDstHeaders + tbl_ws_dst_intervals -> NMW_WsDstIntervals + tbl_ws_dst_flow_history -> NMW_WsDstFlowHistory + tbl_ws_dst_fluid_properties -> NMW_WsDstFluidProperties + tbl_ws_dst_pressure -> NMW_WsDstPressure + +Publication/source registry: + tbl_sources -> NMW_Sources + (Keyed by free-text SourceID; joined into ogc_heat_flow for attribution.) +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +revision: str = "c0d1e2f3a4b5" +down_revision: Union[str, Sequence[str], None] = "x2y3z4a5b6c7" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # ------------------------------------------------------------------ + # Core well tables + # ------------------------------------------------------------------ + op.create_table( + "NMW_WellLocations", + sa.Column("OBJECTID", sa.Integer(), nullable=False), + sa.Column("WellDataID", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("Well_ID", sa.String(), nullable=True), + sa.Column("Import_ID", sa.Integer(), nullable=True), + sa.Column("Township", sa.Float(), nullable=True), + sa.Column("NorS_TDir", sa.String(), nullable=True), + sa.Column("Range", sa.Float(), nullable=True), + sa.Column("EorW_RDir", sa.String(), nullable=True), + sa.Column("Sectn", sa.SmallInteger(), nullable=True), + sa.Column("SectnPart", sa.String(), nullable=True), + sa.Column("UnitLetter", sa.String(), nullable=True), + sa.Column("UTM_zone", sa.String(), nullable=True), + sa.Column("State", sa.String(), nullable=True), + sa.Column("County", sa.String(), nullable=True), + sa.Column("Basin", sa.String(), nullable=True), + sa.Column("Footage_NS", sa.Float(), nullable=True), + sa.Column("NorS_FDir", sa.String(), nullable=True), + sa.Column("Footage_EW", sa.Float(), nullable=True), + sa.Column("EorW_FDir", sa.String(), nullable=True), + sa.Column("Lat_min", sa.SmallInteger(), nullable=True), + sa.Column("Lat_sec", sa.Float(), nullable=True), + sa.Column("Long_deg", sa.SmallInteger(), nullable=True), + sa.Column("Long_min", sa.SmallInteger(), nullable=True), + sa.Column("Long_sec", sa.Float(), nullable=True), + sa.Column("Lat_dd27", sa.Float(), nullable=True), + sa.Column("Long_dd27", sa.Float(), nullable=True), + sa.Column("Lat_dd83", sa.Float(), nullable=True), + sa.Column("Long_dd83", sa.Float(), nullable=True), + sa.Column("SourceID", sa.String(), nullable=True), + sa.Column("SourceDatum", sa.String(), nullable=True), + sa.Column("SourceUnits", sa.String(), nullable=True), + sa.Column("LocAccType", sa.String(), nullable=True), + sa.Column("LocAccMeas", sa.String(), nullable=True), + sa.Column("LocAccVal", sa.Float(), nullable=True), + sa.Column("Duplicated", sa.SmallInteger(), nullable=True), + sa.Column("Exclude", sa.SmallInteger(), nullable=True), + sa.Column("Comments", sa.String(), nullable=True), + sa.Column("GlobalID", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("API", sa.String(), nullable=True), + sa.PrimaryKeyConstraint("OBJECTID"), + ) + op.create_index( + "ix_NMW_WellLocations_WellDataID", "NMW_WellLocations", ["WellDataID"] + ) + + op.create_table( + "NMW_WellHeaders", + sa.Column("OBJECTID", sa.Integer(), nullable=True), + sa.Column("WellDataID", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("WellSpotID", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("API", sa.String(), nullable=True), + sa.Column("WellClass", sa.String(), nullable=True), + sa.Column("WellType", sa.String(), nullable=True), + sa.Column("WellOrient", sa.String(), nullable=True), + sa.Column("CurWellNam", sa.String(), nullable=True), + sa.Column("CurWellNum", sa.String(), nullable=True), + sa.Column("CurStatus", sa.String(), nullable=True), + sa.Column("PrdPoolCnt", sa.SmallInteger(), nullable=True), + sa.Column("CurOperatr", sa.String(), nullable=True), + sa.Column("CurOwner", sa.String(), nullable=True), + sa.Column("TotalDepth", sa.Float(), nullable=True), + sa.Column("Well_TVD", sa.Float(), nullable=True), + sa.Column("Fm_TD", sa.String(), nullable=True), + sa.Column("Age_TD", sa.String(), nullable=True), + sa.Column("SpudDate", sa.DateTime(), nullable=True), + sa.Column("ComplDate", sa.DateTime(), nullable=True), + sa.Column("PlugDate", sa.DateTime(), nullable=True), + sa.Column("PlugBack", sa.Float(), nullable=True), + sa.Column("BridgePlug", sa.String(), nullable=True), + sa.Column("ScoutTickt", sa.SmallInteger(), nullable=True), + sa.Column("DwnHoleSur", sa.SmallInteger(), nullable=True), + sa.Column("GeolLog", sa.SmallInteger(), nullable=True), + sa.Column("Geophyslog", sa.SmallInteger(), nullable=True), + sa.Column("GthrmExist", sa.SmallInteger(), nullable=True), + sa.Column("PetroData", sa.SmallInteger(), nullable=True), + sa.Column("CoreExists", sa.SmallInteger(), nullable=True), + sa.Column("Cuttings", sa.SmallInteger(), nullable=True), + sa.Column("SampleData", sa.SmallInteger(), nullable=True), + sa.Column("Comments", sa.String(), nullable=True), + sa.Column("Import_ID", sa.String(), nullable=True), + sa.Column("Import_DB", sa.String(), nullable=True), + sa.PrimaryKeyConstraint("WellDataID"), + ) + + op.create_table( + "NMW_WellRecords", + sa.Column("OBJECTID", sa.Integer(), nullable=True), + sa.Column("RecrdSetID", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("WellDataID", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("RecrdClass", sa.String(), nullable=True), + sa.Column("SourceID", sa.String(), nullable=True), + sa.Column("ActionDate", sa.DateTime(), nullable=True), + sa.Column("WellName", sa.String(), nullable=True), + sa.Column("WellNumber", sa.String(), nullable=True), + sa.Column("API_suffix", sa.String(), nullable=True), + sa.Column("EnteredBy", sa.String(), nullable=True), + sa.Column("EntryDate", sa.DateTime(), nullable=True), + sa.Column("Comments", sa.String(), nullable=True), + sa.PrimaryKeyConstraint("RecrdSetID"), + ) + op.create_index("ix_NMW_WellRecords_WellDataID", "NMW_WellRecords", ["WellDataID"]) + + op.create_table( + "NMW_WellZDatum", + sa.Column("OBJECTID", sa.Integer(), nullable=False), + sa.Column("RecrdsetID", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("Elev_GL", sa.Float(), nullable=True), + sa.Column("Elev_DF", sa.Float(), nullable=True), + sa.Column("Elev_KB", sa.Float(), nullable=True), + sa.Column("Elev_unspc", sa.Float(), nullable=True), + sa.Column("DatumElev", sa.Float(), nullable=True), + sa.Column("DepthDatum", sa.String(), nullable=True), + sa.Column("DepthUnits", sa.String(), nullable=True), + sa.Column("Z_datum", sa.String(), nullable=True), + sa.Column("Z_units", sa.String(), nullable=True), + sa.Column("ElevSource", sa.String(), nullable=True), + sa.Column("ElvAccType", sa.String(), nullable=True), + sa.Column("ElvAccMeas", sa.String(), nullable=True), + sa.Column("ElvAccVal", sa.Float(), nullable=True), + sa.Column("Comments", sa.String(), nullable=True), + sa.Column("GlobalID", postgresql.UUID(as_uuid=True), nullable=True), + sa.PrimaryKeyConstraint("OBJECTID"), + ) + op.create_index("ix_NMW_WellZDatum_RecrdsetID", "NMW_WellZDatum", ["RecrdsetID"]) + + op.create_table( + "NMW_WellSamples", + sa.Column("OBJECTID", sa.Integer(), nullable=True), + sa.Column("SamplSetID", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("RecrdsetID", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("SmpSetName", sa.String(), nullable=True), + sa.Column("SamplClass", sa.String(), nullable=True), + sa.Column("SampleType", sa.String(), nullable=True), + sa.Column("SampleFm", sa.String(), nullable=True), + sa.Column("SampleLoc", sa.String(), nullable=True), + sa.Column("SampleDate", sa.DateTime(), nullable=True), + sa.Column("From_Depth", sa.Float(), nullable=True), + sa.Column("To_Depth", sa.Float(), nullable=True), + sa.Column("SmpDpUnt", sa.String(), nullable=True), + sa.Column("From_TVD", sa.Float(), nullable=True), + sa.Column("To_TVD", sa.Float(), nullable=True), + sa.Column("From_Elev", sa.Float(), nullable=True), + sa.Column("To_Elev", sa.Float(), nullable=True), + sa.Column("Porosity", sa.SmallInteger(), nullable=True), + sa.Column("Permeablty", sa.SmallInteger(), nullable=True), + sa.Column("Density", sa.SmallInteger(), nullable=True), + sa.Column("DST_Tests", sa.SmallInteger(), nullable=True), + sa.Column("ThinSect", sa.SmallInteger(), nullable=True), + sa.Column("Geochron", sa.SmallInteger(), nullable=True), + sa.Column("Geochem", sa.SmallInteger(), nullable=True), + sa.Column("Geothermal", sa.SmallInteger(), nullable=True), + sa.Column("WholeRock", sa.SmallInteger(), nullable=True), + sa.Column("Paleontlgy", sa.SmallInteger(), nullable=True), + sa.Column("EnteredBy", sa.String(), nullable=True), + sa.Column("EntryDate", sa.DateTime(), nullable=True), + sa.Column("Notes", sa.String(), nullable=True), + sa.PrimaryKeyConstraint("SamplSetID"), + ) + op.create_index("ix_NMW_WellSamples_RecrdsetID", "NMW_WellSamples", ["RecrdsetID"]) + + # ------------------------------------------------------------------ + # Geothermal tables + # ------------------------------------------------------------------ + op.create_table( + "NMW_GtBhtHeaders", + sa.Column("OBJECTID", sa.Integer(), nullable=True), + sa.Column("BHTGUID", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("SamplSetID", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("BoreDia", sa.Float(), nullable=True), + sa.Column("BoreUnits", sa.String(length=16), nullable=True), + sa.Column("DrillFluid", sa.String(length=16), nullable=True), + sa.Column("TempUnit", sa.String(length=1), nullable=True), + sa.Column("FldSalinity", sa.Float(), nullable=True), + sa.Column("FldRstvity", sa.Float(), nullable=True), + sa.Column("Fluid_pH", sa.Float(), nullable=True), + sa.Column("FldDensity", sa.Float(), nullable=True), + sa.Column("FldLevel", sa.Float(), nullable=True), + sa.Column("FldViscsty", sa.Float(), nullable=True), + sa.Column("FluidLoss", sa.String(length=50), nullable=True), + sa.Column("Notes", sa.String(length=255), nullable=True), + sa.PrimaryKeyConstraint("BHTGUID"), + ) + op.create_index( + "ix_NMW_GtBhtHeaders_SamplSetID", "NMW_GtBhtHeaders", ["SamplSetID"] + ) + + op.create_table( + "NMW_GtBhtData", + sa.Column("OBJECTID", sa.Integer(), nullable=False), + sa.Column("BHTGUID", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("Depth", sa.Float(), nullable=True), + sa.Column("BHT", sa.Float(), nullable=True), + sa.Column("TempUnit", sa.String(length=5), nullable=True), + sa.Column("HrsSnceCir", sa.Float(), nullable=True), + sa.Column("DateMeasrd", sa.DateTime(), nullable=True), + sa.Column("Comments", sa.String(length=255), nullable=True), + sa.Column("GlobalID", postgresql.UUID(as_uuid=True), nullable=True), + sa.PrimaryKeyConstraint("OBJECTID"), + ) + op.create_index("ix_NMW_GtBhtData_BHTGUID", "NMW_GtBhtData", ["BHTGUID"]) + + op.create_table( + "NMW_WsIntervals", + sa.Column("OBJECTID", sa.Integer(), nullable=True), + sa.Column("IntrvlGUID", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("SamplSetID", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("SampleID", sa.String(length=128), nullable=True), + sa.Column("From_Depth", sa.Float(), nullable=True), + sa.Column("To_Depth", sa.Float(), nullable=True), + sa.Column("From_TVD", sa.Float(), nullable=True), + sa.Column("To_TVD", sa.Float(), nullable=True), + sa.Column("From_Elev", sa.Float(), nullable=True), + sa.Column("To_Elev", sa.Float(), nullable=True), + sa.Column("Intv_Notes", sa.String(length=255), nullable=True), + sa.PrimaryKeyConstraint("IntrvlGUID"), + ) + op.create_index("ix_NMW_WsIntervals_SamplSetID", "NMW_WsIntervals", ["SamplSetID"]) + + op.create_table( + "NMW_GtConductivity", + sa.Column("OBJECTID", sa.Integer(), nullable=False), + sa.Column("IntrvlGUID", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("Cnductvity", sa.Float(), nullable=True), + sa.Column("CnductUnit", sa.String(length=3), nullable=True), + sa.Column("Comments", sa.String(length=255), nullable=True), + sa.Column("GlobalID", postgresql.UUID(as_uuid=True), nullable=True), + sa.PrimaryKeyConstraint("OBJECTID"), + ) + op.create_index( + "ix_NMW_GtConductivity_IntrvlGUID", "NMW_GtConductivity", ["IntrvlGUID"] + ) + + op.create_table( + "NMW_GtHeatFlow", + sa.Column("OBJECTID", sa.Integer(), nullable=False), + sa.Column("IntrvlGUID", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("Gradient", sa.Float(), nullable=True), + sa.Column("Ka", sa.Float(), nullable=True), + sa.Column("Ka_unit", sa.String(length=3), nullable=True), + sa.Column("Pm", sa.Float(), nullable=True), + sa.Column("Kpr", sa.Float(), nullable=True), + sa.Column("Kpr_unit", sa.String(length=3), nullable=True), + sa.Column("Q", sa.Float(), nullable=True), + sa.Column("Q_unit", sa.String(length=3), nullable=True), + sa.Column("Comments", sa.String(length=255), nullable=True), + sa.Column("GlobalID", postgresql.UUID(as_uuid=True), nullable=True), + sa.PrimaryKeyConstraint("OBJECTID"), + ) + op.create_index("ix_NMW_GtHeatFlow_IntrvlGUID", "NMW_GtHeatFlow", ["IntrvlGUID"]) + + op.create_table( + "NMW_GtSumHeatFlow", + sa.Column("OBJECTID", sa.Integer(), nullable=False), + sa.Column("RecrdSetID", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("SamplSetID", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("LithClass", sa.String(length=50), nullable=True), + sa.Column("UnitBasis", sa.String(length=16), nullable=True), + sa.Column("UnitName", sa.String(length=128), nullable=True), + sa.Column("GeoID", sa.String(length=16), nullable=True), + sa.Column("FromDepth", sa.Float(), nullable=True), + sa.Column("ToDepth", sa.Float(), nullable=True), + sa.Column("DepthUnit", sa.String(length=8), nullable=True), + sa.Column("From_Elev", sa.Float(), nullable=True), + sa.Column("To_Elev", sa.Float(), nullable=True), + sa.Column("ThermlGrad", sa.Float(), nullable=True), + sa.Column("TGError", sa.Float(), nullable=True), + sa.Column("GradUnit", sa.String(length=3), nullable=True), + sa.Column("TGradRange", sa.String(length=15), nullable=True), + sa.Column("SampleType", sa.String(length=50), nullable=True), + sa.Column("NumSamples", sa.SmallInteger(), nullable=True), + sa.Column("ThermlCond", sa.Float(), nullable=True), + sa.Column("TCondError", sa.Float(), nullable=True), + sa.Column("TCondUnit", sa.String(length=3), nullable=True), + sa.Column("TCondRange", sa.String(length=15), nullable=True), + sa.Column("HeatFlow", sa.Float(), nullable=True), + sa.Column("HtFlowErr", sa.Float(), nullable=True), + sa.Column("HtFlowUnit", sa.String(length=3), nullable=True), + sa.Column("HtFlowEst", sa.Float(), nullable=True), + sa.Column("Quality", sa.String(length=50), nullable=True), + sa.Column("Comments", sa.String(length=255), nullable=True), + sa.Column("GlobalID", postgresql.UUID(as_uuid=True), nullable=True), + sa.PrimaryKeyConstraint("OBJECTID"), + ) + op.create_index( + "ix_NMW_GtSumHeatFlow_RecrdSetID", "NMW_GtSumHeatFlow", ["RecrdSetID"] + ) + op.create_index( + "ix_NMW_GtSumHeatFlow_SamplSetID", "NMW_GtSumHeatFlow", ["SamplSetID"] + ) + + op.create_table( + "NMW_GtTempDepths", + sa.Column("OBJECTID", sa.Integer(), nullable=False), + sa.Column("SamplSetID", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("Depth", sa.Float(), nullable=True), + sa.Column("Temp", sa.Float(), nullable=True), + sa.Column("TempUnit", sa.String(length=1), nullable=True), + sa.Column("IntrvlGrad", sa.Float(), nullable=True), + sa.Column("Comments", sa.String(length=255), nullable=True), + sa.Column("GlobalID", postgresql.UUID(as_uuid=True), nullable=True), + sa.PrimaryKeyConstraint("OBJECTID"), + ) + op.create_index( + "ix_NMW_GtTempDepths_SamplSetID", "NMW_GtTempDepths", ["SamplSetID"] + ) + + # ------------------------------------------------------------------ + # Drill Stem Test tables + # ------------------------------------------------------------------ + op.create_table( + "NMW_WsDstHeaders", + sa.Column("OBJECTID", sa.Integer(), nullable=True), + sa.Column("DSTGUID", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("SamplSetID", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("TestType", sa.String(length=50), nullable=True), + sa.Column("DSTOprator", sa.String(length=50), nullable=True), + sa.Column("PressUnits", sa.String(length=8), nullable=True), + sa.Column("TempUnit", sa.String(length=1), nullable=True), + sa.Column("PipeDiaUnt", sa.String(length=8), nullable=True), + sa.Column("PipeLenUnt", sa.String(length=8), nullable=True), + sa.Column("ChokeSizUn", sa.String(length=8), nullable=True), + sa.Column("Notes", sa.String(length=255), nullable=True), + sa.PrimaryKeyConstraint("DSTGUID"), + ) + op.create_index( + "ix_NMW_WsDstHeaders_SamplSetID", "NMW_WsDstHeaders", ["SamplSetID"] + ) + + op.create_table( + "NMW_WsDstIntervals", + sa.Column("OBJECTID", sa.Integer(), nullable=True), + sa.Column("DSTInterval", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("DSTGUID", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("DSTName", sa.String(length=128), nullable=True), + sa.Column("TargetFm", sa.String(length=16), nullable=True), + sa.Column("DSTDate", sa.DateTime(), nullable=True), + sa.Column("DSTNumber", sa.SmallInteger(), nullable=True), + sa.Column("Status", sa.String(length=255), nullable=True), + sa.Column("StatusDate", sa.DateTime(), nullable=True), + sa.Column("PackrFrom", sa.Float(), nullable=True), + sa.Column("PackerTo", sa.Float(), nullable=True), + sa.Column("SrfChokeSz", sa.Float(), nullable=True), + sa.Column("BotChokeSz", sa.Float(), nullable=True), + sa.Column("PipeDia", sa.Float(), nullable=True), + sa.Column("PipeLength", sa.Float(), nullable=True), + sa.Column("Notes", sa.String(length=255), nullable=True), + sa.PrimaryKeyConstraint("DSTInterval"), + ) + op.create_index("ix_NMW_WsDstIntervals_DSTGUID", "NMW_WsDstIntervals", ["DSTGUID"]) + + op.create_table( + "NMW_WsDstFlowHistory", + sa.Column("OBJECTID", sa.Integer(), nullable=False), + sa.Column("DSTInterval", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("Operation", sa.String(length=255), nullable=True), + sa.Column("StartTime", sa.DateTime(), nullable=True), + sa.Column("EndTime", sa.DateTime(), nullable=True), + sa.Column("Duration", sa.Float(), nullable=True), + sa.Column("Pressure", sa.Float(), nullable=True), + sa.Column("Temp", sa.Float(), nullable=True), + sa.Column("RecovColmn", sa.Float(), nullable=True), + sa.Column("RecovType", sa.String(length=255), nullable=True), + sa.Column("Notes", sa.String(length=255), nullable=True), + sa.Column("GlobalID", postgresql.UUID(as_uuid=True), nullable=True), + sa.PrimaryKeyConstraint("OBJECTID"), + ) + op.create_index( + "ix_NMW_WsDstFlowHistory_DSTInterval", "NMW_WsDstFlowHistory", ["DSTInterval"] + ) + + op.create_table( + "NMW_WsDstFluidProperties", + sa.Column("OBJECTID", sa.Integer(), nullable=False), + sa.Column("DSTInterval", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("SourceLoc", sa.String(length=255), nullable=True), + sa.Column("Resistivty", sa.Float(), nullable=True), + sa.Column("Temp", sa.Float(), nullable=True), + sa.Column("Chlorides", sa.Float(), nullable=True), + sa.Column("Notes", sa.String(length=255), nullable=True), + sa.Column("GlobalID", postgresql.UUID(as_uuid=True), nullable=True), + sa.PrimaryKeyConstraint("OBJECTID"), + ) + op.create_index( + "ix_NMW_WsDstFluidProperties_DSTInterval", + "NMW_WsDstFluidProperties", + ["DSTInterval"], + ) + + op.create_table( + "NMW_WsDstPressure", + sa.Column("OBJECTID", sa.Integer(), nullable=False), + sa.Column("DSTInterval", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("PrsGageDpt", sa.Float(), nullable=True), + sa.Column("BlankedOff", sa.SmallInteger(), nullable=True), + sa.Column("InShtInMin", sa.Float(), nullable=True), + sa.Column("FlwPrsInMin", sa.Float(), nullable=True), + sa.Column("PrsInShtIn", sa.Float(), nullable=True), + sa.Column("PrsInitClsdIn", sa.Float(), nullable=True), + sa.Column("FnShtInMin", sa.Float(), nullable=True), + sa.Column("FlwPrsFinMin", sa.Float(), nullable=True), + sa.Column("PrsFnShtIn", sa.Float(), nullable=True), + sa.Column("ShtInPrMth", sa.String(length=255), nullable=True), + sa.Column("HydrostPrsIn", sa.Float(), nullable=True), + sa.Column("HydStPrsFl", sa.Float(), nullable=True), + sa.Column("HydstPrMth", sa.String(length=255), nullable=True), + sa.Column("EquilPress", sa.Float(), nullable=True), + sa.Column("EqlPrsMth", sa.String(length=255), nullable=True), + sa.Column("FlowPrsMin", sa.Float(), nullable=True), + sa.Column("FlowPrsMax", sa.Float(), nullable=True), + sa.Column("FlowPrsMth", sa.String(length=255), nullable=True), + sa.Column("DSTFluid", sa.String(length=128), nullable=True), + sa.Column("FmTemp", sa.Float(), nullable=True), + sa.Column("TempCorrtn", sa.Float(), nullable=True), + sa.Column("TempFlowng", sa.Float(), nullable=True), + sa.Column("TempUnit", sa.String(length=5), nullable=True), + sa.Column("Notes", sa.String(length=255), nullable=True), + sa.Column("GlobalID", postgresql.UUID(as_uuid=True), nullable=True), + sa.PrimaryKeyConstraint("OBJECTID"), + ) + op.create_index( + "ix_NMW_WsDstPressure_DSTInterval", "NMW_WsDstPressure", ["DSTInterval"] + ) + + # ------------------------------------------------------------------ + # Publication/source registry + # ------------------------------------------------------------------ + op.create_table( + "NMW_Sources", + sa.Column("OBJECTID", sa.Integer(), nullable=False), + sa.Column("SourceID", sa.String(), nullable=True), + sa.Column("FirstAuth", sa.String(), nullable=True), + sa.Column("PubYear", sa.String(), nullable=True), + sa.Column("Title", sa.String(), nullable=True), + sa.Column("Journal", sa.String(), nullable=True), + sa.Column("Volume", sa.String(), nullable=True), + sa.Column("PageNo", sa.String(), nullable=True), + sa.Column("ReportNo", sa.String(), nullable=True), + sa.Column("Publisher", sa.String(), nullable=True), + sa.Column("City", sa.String(), nullable=True), + sa.Column("URL", sa.String(), nullable=True), + sa.Column("Comments", sa.String(), nullable=True), + sa.PrimaryKeyConstraint("OBJECTID"), + ) + op.create_index("ix_NMW_Sources_SourceID", "NMW_Sources", ["SourceID"]) + + # ------------------------------------------------------------------ + # FK constraints + # ------------------------------------------------------------------ + op.create_foreign_key( + "fk_nmw_welllocations_welldataid", + "NMW_WellLocations", + "NMW_WellHeaders", + ["WellDataID"], + ["WellDataID"], + ) + op.create_foreign_key( + "fk_nmw_wellrecords_welldataid", + "NMW_WellRecords", + "NMW_WellHeaders", + ["WellDataID"], + ["WellDataID"], + ) + op.create_foreign_key( + "fk_nmw_wellzdatum_recrdsetid", + "NMW_WellZDatum", + "NMW_WellRecords", + ["RecrdsetID"], + ["RecrdSetID"], + ) + op.create_foreign_key( + "fk_nmw_wellsamples_recrdsetid", + "NMW_WellSamples", + "NMW_WellRecords", + ["RecrdsetID"], + ["RecrdSetID"], + ) + op.create_foreign_key( + "fk_nmw_gtbhtheaders_samplsetid", + "NMW_GtBhtHeaders", + "NMW_WellSamples", + ["SamplSetID"], + ["SamplSetID"], + ) + op.create_foreign_key( + "fk_nmw_gtbhtdata_bhtguid", + "NMW_GtBhtData", + "NMW_GtBhtHeaders", + ["BHTGUID"], + ["BHTGUID"], + ) + op.create_foreign_key( + "fk_nmw_wsintervals_samplsetid", + "NMW_WsIntervals", + "NMW_WellSamples", + ["SamplSetID"], + ["SamplSetID"], + ) + op.create_foreign_key( + "fk_nmw_gtconductivity_intrvlguid", + "NMW_GtConductivity", + "NMW_WsIntervals", + ["IntrvlGUID"], + ["IntrvlGUID"], + ) + op.create_foreign_key( + "fk_nmw_gtheatflow_intrvlguid", + "NMW_GtHeatFlow", + "NMW_WsIntervals", + ["IntrvlGUID"], + ["IntrvlGUID"], + ) + op.create_foreign_key( + "fk_nmw_gtsumheatflow_recrdsetid", + "NMW_GtSumHeatFlow", + "NMW_WellRecords", + ["RecrdSetID"], + ["RecrdSetID"], + ) + op.create_foreign_key( + "fk_nmw_gtsumheatflow_samplsetid", + "NMW_GtSumHeatFlow", + "NMW_WellSamples", + ["SamplSetID"], + ["SamplSetID"], + ) + op.create_foreign_key( + "fk_nmw_gttempdepths_samplsetid", + "NMW_GtTempDepths", + "NMW_WellSamples", + ["SamplSetID"], + ["SamplSetID"], + ) + op.create_foreign_key( + "fk_nmw_wsdstheaders_samplsetid", + "NMW_WsDstHeaders", + "NMW_WellSamples", + ["SamplSetID"], + ["SamplSetID"], + ) + op.create_foreign_key( + "fk_nmw_wsdstintervals_dstguid", + "NMW_WsDstIntervals", + "NMW_WsDstHeaders", + ["DSTGUID"], + ["DSTGUID"], + ) + op.create_foreign_key( + "fk_nmw_wsdstflowhistory_dstinterval", + "NMW_WsDstFlowHistory", + "NMW_WsDstIntervals", + ["DSTInterval"], + ["DSTInterval"], + ) + op.create_foreign_key( + "fk_nmw_wsdstfluidproperties_dstinterval", + "NMW_WsDstFluidProperties", + "NMW_WsDstIntervals", + ["DSTInterval"], + ["DSTInterval"], + ) + op.create_foreign_key( + "fk_nmw_wsdstpressure_dstinterval", + "NMW_WsDstPressure", + "NMW_WsDstIntervals", + ["DSTInterval"], + ["DSTInterval"], + ) + + +def downgrade() -> None: + # Drop FKs before tables (reverse order of creation) + op.drop_constraint( + "fk_nmw_wsdstpressure_dstinterval", "NMW_WsDstPressure", type_="foreignkey" + ) + op.drop_constraint( + "fk_nmw_wsdstfluidproperties_dstinterval", + "NMW_WsDstFluidProperties", + type_="foreignkey", + ) + op.drop_constraint( + "fk_nmw_wsdstflowhistory_dstinterval", + "NMW_WsDstFlowHistory", + type_="foreignkey", + ) + op.drop_constraint( + "fk_nmw_wsdstintervals_dstguid", "NMW_WsDstIntervals", type_="foreignkey" + ) + op.drop_constraint( + "fk_nmw_wsdstheaders_samplsetid", "NMW_WsDstHeaders", type_="foreignkey" + ) + op.drop_constraint( + "fk_nmw_gttempdepths_samplsetid", "NMW_GtTempDepths", type_="foreignkey" + ) + op.drop_constraint( + "fk_nmw_gtsumheatflow_samplsetid", "NMW_GtSumHeatFlow", type_="foreignkey" + ) + op.drop_constraint( + "fk_nmw_gtsumheatflow_recrdsetid", "NMW_GtSumHeatFlow", type_="foreignkey" + ) + op.drop_constraint( + "fk_nmw_gtheatflow_intrvlguid", "NMW_GtHeatFlow", type_="foreignkey" + ) + op.drop_constraint( + "fk_nmw_gtconductivity_intrvlguid", "NMW_GtConductivity", type_="foreignkey" + ) + op.drop_constraint( + "fk_nmw_wsintervals_samplsetid", "NMW_WsIntervals", type_="foreignkey" + ) + op.drop_constraint("fk_nmw_gtbhtdata_bhtguid", "NMW_GtBhtData", type_="foreignkey") + op.drop_constraint( + "fk_nmw_gtbhtheaders_samplsetid", "NMW_GtBhtHeaders", type_="foreignkey" + ) + op.drop_constraint( + "fk_nmw_wellsamples_recrdsetid", "NMW_WellSamples", type_="foreignkey" + ) + op.drop_constraint( + "fk_nmw_wellzdatum_recrdsetid", "NMW_WellZDatum", type_="foreignkey" + ) + op.drop_constraint( + "fk_nmw_wellrecords_welldataid", "NMW_WellRecords", type_="foreignkey" + ) + op.drop_constraint( + "fk_nmw_welllocations_welldataid", "NMW_WellLocations", type_="foreignkey" + ) + + # Drop tables in child-first order + op.drop_index("ix_NMW_Sources_SourceID", table_name="NMW_Sources") + op.drop_table("NMW_Sources") + op.drop_index("ix_NMW_WsDstPressure_DSTInterval", table_name="NMW_WsDstPressure") + op.drop_table("NMW_WsDstPressure") + op.drop_index( + "ix_NMW_WsDstFluidProperties_DSTInterval", table_name="NMW_WsDstFluidProperties" + ) + op.drop_table("NMW_WsDstFluidProperties") + op.drop_index( + "ix_NMW_WsDstFlowHistory_DSTInterval", table_name="NMW_WsDstFlowHistory" + ) + op.drop_table("NMW_WsDstFlowHistory") + op.drop_index("ix_NMW_WsDstIntervals_DSTGUID", table_name="NMW_WsDstIntervals") + op.drop_table("NMW_WsDstIntervals") + op.drop_index("ix_NMW_WsDstHeaders_SamplSetID", table_name="NMW_WsDstHeaders") + op.drop_table("NMW_WsDstHeaders") + op.drop_index("ix_NMW_GtConductivity_IntrvlGUID", table_name="NMW_GtConductivity") + op.drop_table("NMW_GtConductivity") + op.drop_index("ix_NMW_GtHeatFlow_IntrvlGUID", table_name="NMW_GtHeatFlow") + op.drop_table("NMW_GtHeatFlow") + op.drop_index("ix_NMW_WsIntervals_SamplSetID", table_name="NMW_WsIntervals") + op.drop_table("NMW_WsIntervals") + op.drop_index("ix_NMW_GtBhtData_BHTGUID", table_name="NMW_GtBhtData") + op.drop_table("NMW_GtBhtData") + op.drop_index("ix_NMW_GtBhtHeaders_SamplSetID", table_name="NMW_GtBhtHeaders") + op.drop_table("NMW_GtBhtHeaders") + op.drop_index("ix_NMW_GtSumHeatFlow_SamplSetID", table_name="NMW_GtSumHeatFlow") + op.drop_index("ix_NMW_GtSumHeatFlow_RecrdSetID", table_name="NMW_GtSumHeatFlow") + op.drop_table("NMW_GtSumHeatFlow") + op.drop_index("ix_NMW_GtTempDepths_SamplSetID", table_name="NMW_GtTempDepths") + op.drop_table("NMW_GtTempDepths") + op.drop_index("ix_NMW_WellSamples_RecrdsetID", table_name="NMW_WellSamples") + op.drop_table("NMW_WellSamples") + op.drop_index("ix_NMW_WellZDatum_RecrdsetID", table_name="NMW_WellZDatum") + op.drop_table("NMW_WellZDatum") + op.drop_index("ix_NMW_WellRecords_WellDataID", table_name="NMW_WellRecords") + op.drop_table("NMW_WellRecords") + op.drop_index("ix_NMW_WellLocations_WellDataID", table_name="NMW_WellLocations") + op.drop_table("NMW_WellLocations") + op.drop_table("NMW_WellHeaders") diff --git a/alembic/versions/d1e2f3a4b5c6_nmw_per_well_geothermal_ogc_views.py b/alembic/versions/d1e2f3a4b5c6_nmw_per_well_geothermal_ogc_views.py new file mode 100644 index 000000000..696b8118a --- /dev/null +++ b/alembic/versions/d1e2f3a4b5c6_nmw_per_well_geothermal_ogc_views.py @@ -0,0 +1,262 @@ +"""NMW per-well geothermal OGC views + +Revision ID: d1e2f3a4b5c6 +Revises: c0d1e2f3a4b5 +Create Date: 2026-06-22 + +Four pygeoapi point-layer views that aggregate geothermal data to one +feature per well. All geometry is built from NMW_WellLocations Lat/Long_dd83 +(WGS84). All views use integer id (row_number() OVER ()) as the pygeoapi +id_field; pygeoapi's PostgreSQL provider requires an integer PK column. + + ogc_geothermal_wells_bht (VIEW) + One feature per well with BHT aggregate stats. Links via: + NMW_GtBhtData → NMW_GtBhtHeaders.BHTGUID → NMW_WellSamples.SamplSetID + → NMW_WellRecords.RecrdSetID → NMW_WellLocations/Headers.WellDataID + + ogc_geothermal_wells_temperature_profile (MATERIALIZED VIEW) + One feature per well with temperature-depth JSON series. Materialized + because the source NMW_GtTempDepths is large (~370k rows) and the + json_agg is too heavy to recompute per request. NMW_WellLocations is + deduped via DISTINCT ON before joining because a well can have multiple + location rows (OBJECTID is the PK, not WellDataID). REFRESH after a + data reload. + + ogc_geothermal_wells_summary_heat_flow (VIEW) + One feature per well with summary heat-flow stats and a measurements + JSON series. Links via NMW_GtSumHeatFlow.RecrdSetID. + + ogc_geothermal_wells_interval_heat_flow (VIEW) + One feature per well with per-interval heat-flow stats. Links via + NMW_GtHeatFlow.IntrvlGUID → NMW_WsIntervals → NMW_WellSamples. +""" + +from alembic import op +from sqlalchemy import text + +revision = "d1e2f3a4b5c6" +down_revision = "c0d1e2f3a4b5" +branch_labels = None +depends_on = None + +_BHT_VIEW = "ogc_geothermal_wells_bht" +_PROFILE_VIEW = "ogc_geothermal_wells_temperature_profile" +_SUM_HF_VIEW = "ogc_geothermal_wells_summary_heat_flow" +_INT_HF_VIEW = "ogc_geothermal_wells_interval_heat_flow" + + +def upgrade() -> None: + # ogc_geothermal_wells_bht + op.execute(text(f'DROP VIEW IF EXISTS "{_BHT_VIEW}"')) + op.execute(text(f""" + CREATE VIEW "{_BHT_VIEW}" AS + WITH loc AS ( + SELECT DISTINCT ON ("WellDataID") + "WellDataID", "Lat_dd83", "Long_dd83" + FROM "NMW_WellLocations" + WHERE "Lat_dd83" IS NOT NULL + AND "Long_dd83" IS NOT NULL + ORDER BY "WellDataID", "OBJECTID" + ) + SELECT + row_number() OVER () AS id, + r."WellDataID"::text AS well_data_id, + hdr."CurWellNam" AS well_name, + hdr."API" AS api, + hdr."TotalDepth" AS total_depth, + count(d.*) AS bht_count, + max(d."BHT") AS max_bht, + min(d."BHT") AS min_bht, + max(d."Depth") AS max_bht_depth, + max(d."TempUnit") AS temp_unit, + ST_SetSRID( + ST_MakePoint(loc."Long_dd83", loc."Lat_dd83"), 4326 + ) AS geom + FROM "NMW_GtBhtData" AS d + JOIN "NMW_GtBhtHeaders" AS h ON h."BHTGUID" = d."BHTGUID" + JOIN "NMW_WellSamples" AS s ON s."SamplSetID" = h."SamplSetID" + JOIN "NMW_WellRecords" AS r ON r."RecrdSetID" = s."RecrdsetID" + JOIN loc ON loc."WellDataID" = r."WellDataID" + LEFT JOIN "NMW_WellHeaders" AS hdr ON hdr."WellDataID" = r."WellDataID" + GROUP BY + r."WellDataID", + loc."Lat_dd83", + loc."Long_dd83", + hdr."CurWellNam", + hdr."API", + hdr."TotalDepth" + """)) + + # ogc_geothermal_wells_temperature_profile (materialized) + op.execute(text(f'DROP MATERIALIZED VIEW IF EXISTS "{_PROFILE_VIEW}"')) + op.execute(text(f""" + CREATE MATERIALIZED VIEW "{_PROFILE_VIEW}" AS + WITH loc AS ( + SELECT DISTINCT ON ("WellDataID") + "WellDataID", "Lat_dd83", "Long_dd83" + FROM "NMW_WellLocations" + WHERE "Lat_dd83" IS NOT NULL + AND "Long_dd83" IS NOT NULL + ORDER BY "WellDataID", "OBJECTID" + ) + SELECT + row_number() OVER () AS id, + r."WellDataID"::text AS well_data_id, + hdr."CurWellNam" AS well_name, + hdr."API" AS api, + count(td.*) AS reading_count, + min(td."Depth") AS min_depth, + max(td."Depth") AS max_depth, + min(td."Temp") AS min_temp, + max(td."Temp") AS max_temp, + max(td."TempUnit") AS temp_unit, + json_agg( + json_build_object('depth', td."Depth", 'temp', td."Temp") + ORDER BY td."Depth" + ) AS series, + ST_SetSRID( + ST_MakePoint(loc."Long_dd83", loc."Lat_dd83"), 4326 + ) AS geom + FROM "NMW_GtTempDepths" AS td + JOIN "NMW_WellSamples" AS s ON s."SamplSetID" = td."SamplSetID" + JOIN "NMW_WellRecords" AS r ON r."RecrdSetID" = s."RecrdsetID" + JOIN loc ON loc."WellDataID" = r."WellDataID" + LEFT JOIN "NMW_WellHeaders" AS hdr ON hdr."WellDataID" = r."WellDataID" + WHERE td."Depth" IS NOT NULL + AND td."Temp" IS NOT NULL + GROUP BY + r."WellDataID", + loc."Lat_dd83", + loc."Long_dd83", + hdr."CurWellNam", + hdr."API" + """)) + op.execute( + text(f'CREATE UNIQUE INDEX ux_{_PROFILE_VIEW}_id ON "{_PROFILE_VIEW}" (id)') + ) + op.execute( + text( + f'CREATE INDEX ix_{_PROFILE_VIEW}_geom ON "{_PROFILE_VIEW}" USING GIST (geom)' + ) + ) + + # ogc_geothermal_wells_summary_heat_flow + op.execute(text(f'DROP VIEW IF EXISTS "{_SUM_HF_VIEW}"')) + op.execute(text(f""" + CREATE VIEW "{_SUM_HF_VIEW}" AS + WITH loc AS ( + SELECT DISTINCT ON ("WellDataID") + "WellDataID", "Lat_dd83", "Long_dd83" + FROM "NMW_WellLocations" + WHERE "Lat_dd83" IS NOT NULL + AND "Long_dd83" IS NOT NULL + ORDER BY "WellDataID", "OBJECTID" + ) + SELECT + row_number() OVER () AS id, + r."WellDataID"::text AS well_data_id, + hdr."CurWellNam" AS well_name, + hdr."API" AS api, + count(shf.*) AS heat_flow_count, + max(shf."HeatFlow") AS max_heat_flow, + avg(shf."HeatFlow") AS avg_heat_flow, + max(shf."HtFlowUnit") AS heat_flow_unit, + max(shf."ThermlGrad") AS max_thermal_gradient, + max(shf."GradUnit") AS gradient_unit, + max(shf."ThermlCond") AS max_thermal_conductivity, + max(shf."TCondUnit") AS conductivity_unit, + max(shf."Quality") AS quality, + json_agg( + json_build_object( + 'from_depth', shf."FromDepth", + 'to_depth', shf."ToDepth", + 'depth_unit', shf."DepthUnit", + 'heat_flow', shf."HeatFlow", + 'heat_flow_error', shf."HtFlowErr", + 'heat_flow_unit', shf."HtFlowUnit", + 'thermal_gradient', shf."ThermlGrad", + 'gradient_unit', shf."GradUnit", + 'thermal_conductivity', shf."ThermlCond", + 'conductivity_unit', shf."TCondUnit", + 'quality', shf."Quality" + ) + ORDER BY shf."FromDepth" + ) AS measurements, + ST_SetSRID( + ST_MakePoint(loc."Long_dd83", loc."Lat_dd83"), 4326 + ) AS geom + FROM "NMW_GtSumHeatFlow" AS shf + JOIN "NMW_WellRecords" AS r ON r."RecrdSetID" = shf."RecrdSetID" + JOIN loc ON loc."WellDataID" = r."WellDataID" + LEFT JOIN "NMW_WellHeaders" AS hdr ON hdr."WellDataID" = r."WellDataID" + GROUP BY + r."WellDataID", + loc."Lat_dd83", + loc."Long_dd83", + hdr."CurWellNam", + hdr."API" + """)) + + # ogc_geothermal_wells_interval_heat_flow + op.execute(text(f'DROP VIEW IF EXISTS "{_INT_HF_VIEW}"')) + op.execute(text(f""" + CREATE VIEW "{_INT_HF_VIEW}" AS + WITH loc AS ( + SELECT DISTINCT ON ("WellDataID") + "WellDataID", "Lat_dd83", "Long_dd83" + FROM "NMW_WellLocations" + WHERE "Lat_dd83" IS NOT NULL + AND "Long_dd83" IS NOT NULL + ORDER BY "WellDataID", "OBJECTID" + ) + SELECT + row_number() OVER () AS id, + r."WellDataID"::text AS well_data_id, + hdr."CurWellNam" AS well_name, + hdr."API" AS api, + count(hf.*) AS interval_count, + max(hf."Q") AS max_heat_flow, + avg(hf."Q") AS avg_heat_flow, + max(hf."Q_unit") AS heat_flow_unit, + max(hf."Gradient") AS max_gradient, + max(hf."Kpr") AS max_thermal_conductivity, + max(hf."Kpr_unit") AS conductivity_unit, + max(hf."Ka") AS max_diffusivity, + max(hf."Ka_unit") AS diffusivity_unit, + json_agg( + json_build_object( + 'from_depth', i."From_Depth", + 'to_depth', i."To_Depth", + 'heat_flow', hf."Q", + 'heat_flow_unit', hf."Q_unit", + 'gradient', hf."Gradient", + 'thermal_conductivity', hf."Kpr", + 'conductivity_unit', hf."Kpr_unit", + 'diffusivity', hf."Ka", + 'diffusivity_unit', hf."Ka_unit" + ) + ORDER BY i."From_Depth" + ) AS measurements, + ST_SetSRID( + ST_MakePoint(loc."Long_dd83", loc."Lat_dd83"), 4326 + ) AS geom + FROM "NMW_GtHeatFlow" AS hf + JOIN "NMW_WsIntervals" AS i ON i."IntrvlGUID" = hf."IntrvlGUID" + JOIN "NMW_WellSamples" AS s ON s."SamplSetID" = i."SamplSetID" + JOIN "NMW_WellRecords" AS r ON r."RecrdSetID" = s."RecrdsetID" + JOIN loc ON loc."WellDataID" = r."WellDataID" + LEFT JOIN "NMW_WellHeaders" AS hdr ON hdr."WellDataID" = r."WellDataID" + GROUP BY + r."WellDataID", + loc."Lat_dd83", + loc."Long_dd83", + hdr."CurWellNam", + hdr."API" + """)) + + +def downgrade() -> None: + op.execute(text(f'DROP VIEW IF EXISTS "{_INT_HF_VIEW}"')) + op.execute(text(f'DROP VIEW IF EXISTS "{_SUM_HF_VIEW}"')) + op.execute(text(f'DROP MATERIALIZED VIEW IF EXISTS "{_PROFILE_VIEW}"')) + op.execute(text(f'DROP VIEW IF EXISTS "{_BHT_VIEW}"')) diff --git a/alembic/versions/e2f3a4b5c6d7_nmw_measurement_ogc_views.py b/alembic/versions/e2f3a4b5c6d7_nmw_measurement_ogc_views.py new file mode 100644 index 000000000..f54ac694f --- /dev/null +++ b/alembic/versions/e2f3a4b5c6d7_nmw_measurement_ogc_views.py @@ -0,0 +1,279 @@ +"""NMW individual-measurement and analytical OGC views + +Revision ID: e2f3a4b5c6d7 +Revises: d1e2f3a4b5c6 +Create Date: 2026-06-22 + +Four pygeoapi point-layer views exposing individual measurement rows (not +aggregated per well) and analytical summaries from the NMW staging mirror. +All translated from legacy MSSQL queries against NM_Aquifer. Integer OBJECTID +columns are used as the pygeoapi id_field directly. + + ogc_bht_measurements (VIEW) + One row per BHT measurement with well header and location. + Translated from the legacy MSSQL BHT query. + + ogc_temp_depth_measurements (VIEW) + One row per downhole temperature reading with well header, location, + and elevation. Translated from TempDepth2_SortedWellName query. + Locations with Exclude=1 are filtered out. + + ogc_heat_flow (VIEW) + One row per NMW_GtSumHeatFlow record with well header, location, + elevation, and publication attribution from NMW_Sources. Unit + conversions (ft→m, HFU→mW/m², TCU→W/m·K) applied inline via + CASE WHEN. Translated from the legacy MSSQL HeatFlow query. + + ogc_dst (VIEW) + One row per DST interval with pressure, flow history, and well + header. The original Access query referenced DST_flwHstryConcat + (a broken saved query); replaced with a string_agg() CTE over + NMW_WsDstFlowHistory. DISTINCT used in place of the original + GROUP BY with no aggregate functions. Translated from the legacy + MSSQL DST query. +""" + +from alembic import op +from sqlalchemy import text + +revision = "e2f3a4b5c6d7" +down_revision = "d1e2f3a4b5c6" +branch_labels = None +depends_on = None + +_BHT_MEAS_VIEW = "ogc_bht_measurements" +_TEMP_DEPTH_VIEW = "ogc_temp_depth_measurements" +_HEAT_FLOW_VIEW = "ogc_heat_flow" +_DST_VIEW = "ogc_dst" + + +def upgrade() -> None: + # ogc_bht_measurements + op.execute(text(f'DROP VIEW IF EXISTS "{_BHT_MEAS_VIEW}"')) + op.execute(text(f""" + CREATE VIEW "{_BHT_MEAS_VIEW}" AS + SELECT + d."OBJECTID" AS id, + hdr."API" AS api, + hdr."CurWellNam" AS well_name, + hdr."CurWellNum" AS well_num, + hdr."CurOperatr" AS operator, + hdr."WellType" AS well_type, + hdr."Well_TVD" AS well_tvd, + hdr."ComplDate" AS completion_date, + hdr."CurStatus" AS current_status, + hdr."TotalDepth" AS total_depth, + hdr."Cuttings" AS cuttings, + hdr."CoreExists" AS core_exists, + loc."County" AS county, + d."Depth" AS bht_depth, + d."BHT" AS bht, + d."HrsSnceCir" AS hours_since_circulation, + d."DateMeasrd" AS date_measured, + ST_SetSRID( + ST_MakePoint(loc."Long_dd83", loc."Lat_dd83"), 4326 + ) AS geom + FROM "NMW_GtBhtData" AS d + JOIN "NMW_GtBhtHeaders" AS bh ON bh."BHTGUID" = d."BHTGUID" + JOIN "NMW_WellSamples" AS s ON s."SamplSetID" = bh."SamplSetID" + JOIN "NMW_WellRecords" AS r ON r."RecrdSetID" = s."RecrdsetID" + JOIN "NMW_WellZDatum" AS z ON z."RecrdsetID" = r."RecrdSetID" + JOIN "NMW_WellHeaders" AS hdr ON hdr."WellDataID" = r."WellDataID" + JOIN "NMW_WellLocations" AS loc ON loc."WellDataID" = r."WellDataID" + WHERE loc."Lat_dd83" IS NOT NULL + AND loc."Long_dd83" IS NOT NULL + """)) + + # ogc_temp_depth_measurements + op.execute(text(f'DROP VIEW IF EXISTS "{_TEMP_DEPTH_VIEW}"')) + op.execute(text(f""" + CREATE VIEW "{_TEMP_DEPTH_VIEW}" AS + SELECT + td."OBJECTID" AS id, + hdr."CurWellNam" AS well_name, + hdr."CurWellNum" AS well_num, + hdr."API" AS api, + r."SourceID" AS source_id, + s."SampleFm" AS sample_fm, + loc."County" AS county, + loc."State" AS state, + loc."Lat_dd27" AS lat_dd27, + loc."Long_dd27" AS long_dd27, + loc."Lat_dd83" AS lat_dd83, + loc."Long_dd83" AS long_dd83, + loc."LocAccVal" AS loc_acc_val, + s."EnteredBy" AS entered_by, + s."EntryDate" AS entry_date, + td."Depth" AS depth, + s."SmpDpUnt" AS depth_unit, + td."Temp" AS temp, + td."TempUnit" AS temp_unit, + z."Elev_GL" AS elev_gl, + z."Elev_unspc" AS elev_unspc, + z."Elev_KB" AS elev_kb, + s."SampleDate" AS sample_date, + ST_SetSRID( + ST_MakePoint(loc."Long_dd83", loc."Lat_dd83"), 4326 + ) AS geom + FROM "NMW_GtTempDepths" AS td + JOIN "NMW_WellSamples" AS s ON s."SamplSetID" = td."SamplSetID" + JOIN "NMW_WellRecords" AS r ON r."RecrdSetID" = s."RecrdsetID" + JOIN "NMW_WellZDatum" AS z ON z."RecrdsetID" = r."RecrdSetID" + JOIN "NMW_WellHeaders" AS hdr ON hdr."WellDataID" = r."WellDataID" + JOIN "NMW_WellLocations" AS loc ON loc."WellDataID" = r."WellDataID" + WHERE loc."Exclude" = 0 + AND loc."Lat_dd83" IS NOT NULL + AND loc."Long_dd83" IS NOT NULL + """)) + + # ogc_heat_flow + op.execute(text(f'DROP VIEW IF EXISTS "{_HEAT_FLOW_VIEW}"')) + op.execute(text(f""" + CREATE VIEW "{_HEAT_FLOW_VIEW}" AS + SELECT + shf."OBJECTID" AS id, + hdr."CurWellNam" AS well_name, + hdr."CurWellNum" AS well_num, + hdr."API" AS api, + loc."County" AS county, + loc."State" AS state, + loc."Lat_dd27" AS lat_dd27, + loc."Long_dd27" AS long_dd27, + loc."Lat_dd83" AS lat_dd83, + loc."Long_dd83" AS long_dd83, + r."SourceID" AS source_id, + z."Elev_GL" AS elev_gl, + z."Elev_KB" AS elev_kb, + z."Elev_unspc" AS elev_unspc, + CASE WHEN z."DepthUnits" = 'ft' + THEN 0.3048 * z."Elev_unspc" + ELSE z."Elev_unspc" + END AS elevation_m, + z."DepthUnits" AS depth_units, + hdr."TotalDepth" AS total_depth, + CASE WHEN z."DepthUnits" = 'ft' + THEN 0.3048 * hdr."TotalDepth" + ELSE hdr."TotalDepth" + END AS total_depth_m, + shf."FromDepth" AS from_depth, + shf."ToDepth" AS to_depth, + shf."ThermlCond" AS therml_cond, + shf."TCondRange" AS tcond_range, + shf."TCondError" AS tcond_error, + shf."TCondUnit" AS tcond_unit, + CASE WHEN shf."TCondUnit" = 'TCU' + THEN 0.4184 * shf."ThermlCond" + ELSE shf."ThermlCond" + END AS tc_si, + shf."SampleType" AS sample_type, + shf."NumSamples" AS num_samples, + shf."ThermlGrad" AS therml_grad, + shf."TGradRange" AS tgrad_range, + shf."TGError" AS tg_error, + shf."GradUnit" AS grad_unit, + shf."HeatFlow" AS heat_flow, + shf."HtFlowUnit" AS ht_flow_unit, + CASE WHEN shf."HtFlowUnit" = 'HFU' + THEN 41.84 * shf."HeatFlow" + ELSE shf."HeatFlow" + END AS heat_flow_si, + shf."Quality" AS quality, + src."FirstAuth" AS first_auth, + src."PubYear" AS pub_year, + src."Title" AS title, + src."Journal" AS journal, + src."Volume" AS volume, + src."PageNo" AS page_no, + shf."HtFlowEst" AS ht_flow_est, + r."EntryDate" AS entry_date, + CASE WHEN shf."HtFlowUnit" = 'HFU' + THEN 41.84 * shf."HtFlowEst" + ELSE shf."HtFlowEst" + END AS ht_flow_est_si, + ST_SetSRID( + ST_MakePoint(loc."Long_dd83", loc."Lat_dd83"), 4326 + ) AS geom + FROM "NMW_GtSumHeatFlow" AS shf + JOIN "NMW_WellRecords" AS r ON r."RecrdSetID" = shf."RecrdSetID" + JOIN "NMW_WellHeaders" AS hdr ON hdr."WellDataID" = r."WellDataID" + JOIN "NMW_WellLocations" AS loc ON loc."WellDataID" = r."WellDataID" + LEFT JOIN "NMW_WellZDatum" AS z ON z."RecrdsetID" = r."RecrdSetID" + JOIN "NMW_Sources" AS src ON src."SourceID" = r."SourceID" + WHERE loc."Exclude" = 0 + AND loc."Lat_dd83" IS NOT NULL + AND loc."Long_dd83" IS NOT NULL + """)) + + # ogc_dst + op.execute(text(f'DROP VIEW IF EXISTS "{_DST_VIEW}"')) + op.execute(text(f""" + CREATE VIEW "{_DST_VIEW}" AS + WITH flow_history AS ( + SELECT + "DSTInterval", + string_agg("Operation", '; ' ORDER BY "OBJECTID") AS flow_history + FROM "NMW_WsDstFlowHistory" + GROUP BY "DSTInterval" + ) + SELECT DISTINCT + i."OBJECTID" AS id, + hdr."CurWellNam" AS well_name, + hdr."CurWellNum" AS well_num, + hdr."API" AS api, + i."DSTName" AS dst_name, + dh."DSTOprator" AS dst_operator, + i."DSTNumber" AS dst_number, + i."DSTDate" AS dst_date, + loc."County" AS county, + loc."State" AS state, + loc."Lat_dd83" AS lat_dd83, + loc."Long_dd83" AS long_dd83, + s."From_Depth" AS from_depth, + s."To_Depth" AS to_depth, + i."TargetFm" AS target_fm, + i."PackrFrom" AS packer_from, + i."PackerTo" AS packer_to, + i."SrfChokeSz" AS srf_choke_sz, + i."BotChokeSz" AS bot_choke_sz, + s."SmpDpUnt" AS depth_unit, + z."Elev_GL" AS elev_gl, + z."Elev_unspc" AS elev_unspc, + p."PrsGageDpt" AS prs_gage_dpt, + i."PipeDia" AS pipe_dia, + i."PipeLength" AS pipe_length, + fh.flow_history AS flow_history, + p."PrsInShtIn" AS init_flow, + p."FlwPrsInMin" AS flw_prs_in_min, + p."PrsFnShtIn" AS fin_flow, + p."FlwPrsFinMin" AS flw_prs_fin_min, + p."PrsInitClsdIn" AS prs_init_clsd_in, + p."InShtInMin" AS in_sht_in_min, + p."EquilPress" AS fin_shut_in, + p."FnShtInMin" AS fn_sht_in_min, + p."HydrostPrsIn" AS hydrost_prs_in, + p."HydStPrsFl" AS hyd_st_prs_fl, + dh."PressUnits" AS press_units, + p."BlankedOff" AS blanked_off, + p."FmTemp" AS fm_temp, + ST_SetSRID( + ST_MakePoint(loc."Long_dd83", loc."Lat_dd83"), 4326 + ) AS geom + FROM "NMW_WsDstIntervals" AS i + JOIN "NMW_WsDstHeaders" AS dh ON dh."DSTGUID" = i."DSTGUID" + JOIN "NMW_WellSamples" AS s ON s."SamplSetID" = dh."SamplSetID" + JOIN "NMW_WellRecords" AS r ON r."RecrdSetID" = s."RecrdsetID" + JOIN "NMW_WellHeaders" AS hdr ON hdr."WellDataID" = r."WellDataID" + LEFT JOIN "NMW_WellLocations" AS loc ON loc."WellDataID" = r."WellDataID" + LEFT JOIN "NMW_WellZDatum" AS z ON z."RecrdsetID" = r."RecrdSetID" + LEFT JOIN "NMW_WsDstPressure" AS p ON p."DSTInterval" = i."DSTInterval" + LEFT JOIN flow_history AS fh ON fh."DSTInterval" = i."DSTInterval" + WHERE loc."Lat_dd83" IS NOT NULL + AND loc."Long_dd83" IS NOT NULL + """)) + + +def downgrade() -> None: + op.execute(text(f'DROP VIEW IF EXISTS "{_DST_VIEW}"')) + op.execute(text(f'DROP VIEW IF EXISTS "{_HEAT_FLOW_VIEW}"')) + op.execute(text(f'DROP VIEW IF EXISTS "{_TEMP_DEPTH_VIEW}"')) + op.execute(text(f'DROP VIEW IF EXISTS "{_BHT_MEAS_VIEW}"')) diff --git a/alembic/versions/u8v9w0x1y2z3_add_ngwmn_views_from_ocotillo_model.py b/alembic/versions/u8v9w0x1y2z3_add_ngwmn_views_from_ocotillo_model.py new file mode 100644 index 000000000..bf76b540e --- /dev/null +++ b/alembic/versions/u8v9w0x1y2z3_add_ngwmn_views_from_ocotillo_model.py @@ -0,0 +1,213 @@ +"""add NGWMN views sourced from the new Ocotillo data model + +Replaces the legacy NMA_view_NGWMN_* copy tables as the source for NGWMN +exports. These views reproduce the original AMPAPI (SQL Server) view +definitions but read from the new Ocotillo tables: + +- NGWMN_WaterLevels: observation/sample/field_activity/field_event/thing +- NGWMN_WellConstruction: thing/well_screen/well_casing_material +- NGWMN_Lithology: thing_geologic_formation_association/geologic_formation + +Revision ID: u8v9w0x1y2z3 +Revises: t6u7v8w9x0y1 +Create Date: 2026-06-11 00:00:00.000000 +""" + +from typing import Sequence, Union + +from alembic import op +from sqlalchemy import inspect, text + +# revision identifiers, used by Alembic. +revision: str = "u8v9w0x1y2z3" +down_revision: Union[str, Sequence[str], None] = "t6u7v8w9x0y1" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +REQUIRED_TABLES = { + "thing", + "well_screen", + "well_casing_material", + "observation", + "sample", + "field_activity", + "field_event", + "parameter", + "thing_geologic_formation_association", + "geologic_formation", +} + +DROP_WATERLEVELS_SQL = 'DROP VIEW IF EXISTS "NGWMN_WaterLevels"' +DROP_WELLCONSTRUCTION_SQL = 'DROP VIEW IF EXISTS "NGWMN_WellConstruction"' +DROP_LITHOLOGY_SQL = 'DROP VIEW IF EXISTS "NGWMN_Lithology"' + + +def _create_waterlevels_view() -> str: + # Mirrors dbo.view_NGWMN_WaterLevels: + # SELECT PointID, DateMeasured, DepthToWaterBGS, 'ft bgs', CASE + # MeasurementMethod..., CASE DataQuality..., PublicRelease + # FROM WaterLevels WHERE PublicRelease = 1 + # + # The transfer stored observation.value as depth-to-water below the + # measuring point (DepthToWater) and measuring_point_height separately, + # so DepthToWaterBGS = value - measuring_point_height (0 when no MP + # height was recorded). + # + # observation_datetime was converted to UTC during transfer; rows whose + # legacy timestamp had no time component were stored at 00:00 UTC, so + # only rows with a real time component are shifted back to local time + # before taking the date. + # + # MeasurementMethod/WLAccuracy reproduce the legacy CASE expressions, + # keyed on the lexicon meanings the transfer wrote (LU_MeasurementMethod + # and LU_DataQuality), including the legacy quirk mapping code O + # ("Observed...") to 'Acoustic Sounder'. + return """ + CREATE VIEW "NGWMN_WaterLevels" AS + SELECT + t.name AS "PointID", + CASE + WHEN (o.observation_datetime AT TIME ZONE 'UTC')::time = '00:00:00' + THEN (o.observation_datetime AT TIME ZONE 'UTC')::date + ELSE (o.observation_datetime AT TIME ZONE 'America/Denver')::date + END AS "DateMeasured", + o.value - COALESCE(o.measuring_point_height, 0) AS "DepthToWaterBGS", + 'ft bgs' AS "WLUnits", + CASE s.sample_method + WHEN 'Steel-tape measurement' THEN 'Steel tape' + WHEN 'Electric tape measurement (E-probe)' THEN 'Electric tape' + WHEN 'Observed (required for F, N, and W water level status)' THEN 'Acoustic Sounder' + WHEN 'Estimated' THEN 'Estimated' + WHEN 'Reported, method not known' THEN 'Reported' + WHEN 'Pressure-gage measurement' THEN 'Pressure gauge' + WHEN 'Unknown (for legacy data only; not for new data entry)' THEN 'Unknown; from legacy data' + ELSE NULL + END AS "MeasurementMethod", + CASE o.nma_data_quality + WHEN 'Water level accurate to within two hundreths of a foot' THEN '0.02 ft' + WHEN 'Water level accurate to within one foot' THEN '1.0 ft' + WHEN 'Water level accuracy not to nearest foot or water level not repeatable' THEN 'Unknown' + ELSE NULL + END AS "WLAccuracy", + TRUE AS "PublicRelease" + FROM observation AS o + JOIN sample AS s ON s.id = o.sample_id + JOIN field_activity AS fa ON fa.id = s.field_activity_id + JOIN field_event AS fe ON fe.id = fa.field_event_id + JOIN thing AS t ON t.id = fe.thing_id + JOIN parameter AS p ON p.id = o.parameter_id + WHERE p.parameter_name = 'groundwater level' + AND o.release_status = 'public' + AND s.release_status = 'public' + AND fa.release_status = 'public' + AND fe.release_status = 'public' + AND t.release_status = 'public' + """ + + +def _create_wellconstruction_view() -> str: + # Mirrors dbo.view_NGWMN_WellConstruction: + # WellData LEFT JOIN WellScreens ON WellData.WellID = WellScreens.WellID + # CasingTop is 0 whenever a casing depth exists (casing assumed to start + # at ground surface). The legacy free-text CasingDescription was reduced + # to controlled material terms during transfer, so it is rebuilt here as + # a comma-separated list of well_casing_material terms. + return """ + CREATE VIEW "NGWMN_WellConstruction" AS + SELECT + t.name AS "PointID", + CASE WHEN t.well_casing_depth IS NOT NULL THEN 0::double precision END AS "CasingTop", + t.well_casing_depth AS "CasingBottom", + CASE WHEN t.well_casing_depth IS NOT NULL THEN 'ft bgs' END AS "CasingDepthUnits", + ws.screen_depth_top AS "ScreenTop", + ws.screen_depth_bottom AS "ScreenBottom", + CASE WHEN ws.screen_depth_bottom IS NOT NULL THEN 'ft bgs' END AS "ScreenBottomUnit", + ws.screen_description AS "ScreenDescription", + cm.materials AS "CasingDescription" + FROM thing AS t + LEFT JOIN well_screen AS ws + ON ws.thing_id = t.id AND ws.release_status = 'public' + LEFT JOIN LATERAL ( + SELECT string_agg(wcm.material, ', ' ORDER BY wcm.material) AS materials + FROM well_casing_material AS wcm + WHERE wcm.thing_id = t.id AND wcm.release_status = 'public' + ) AS cm ON TRUE + WHERE t.thing_type = 'water well' + AND t.release_status = 'public' + """ + + +def _create_lithology_view() -> str: + # Mirrors dbo.view_NGWMN_Lithology: + # Stratigraphy INNER JOIN LU_Lithology ON Lithology = ABBREVIATION + # The new model keeps the resolved lithology term on geologic_formation + # (the abbreviation code was not migrated), so the term backs both the + # Lithology and TERM columns. The inner join is reproduced by requiring + # a non-null lithology. StratSource was not migrated and is NULL. + return """ + CREATE VIEW "NGWMN_Lithology" AS + SELECT + tgfa.id AS "OBJECTID", + t.name AS "PointID", + gf.lithology AS "Lithology", + gf.lithology AS "TERM", + NULL::character varying AS "StratSource", + tgfa.top_depth AS "StratTop", + CASE WHEN tgfa.top_depth IS NOT NULL THEN 'ft bgs' END AS "StratTopUnit", + tgfa.bottom_depth AS "StratBottom", + CASE WHEN tgfa.bottom_depth IS NOT NULL THEN 'ft bgs' END AS "StratBottomUnit" + FROM thing_geologic_formation_association AS tgfa + JOIN thing AS t ON t.id = tgfa.thing_id + JOIN geologic_formation AS gf ON gf.id = tgfa.geologic_formation_id + WHERE gf.lithology IS NOT NULL + AND tgfa.release_status = 'public' + AND t.release_status = 'public' + """ + + +def upgrade() -> None: + bind = op.get_bind() + inspector = inspect(bind) + existing_tables = set(inspector.get_table_names(schema="public")) + missing = REQUIRED_TABLES - existing_tables + if missing: + raise RuntimeError( + "Cannot create NGWMN views. Missing required tables: " + f"{', '.join(sorted(missing))}" + ) + + op.execute(text(DROP_WATERLEVELS_SQL)) + op.execute(text(_create_waterlevels_view())) + op.execute( + text( + 'COMMENT ON VIEW "NGWMN_WaterLevels" IS ' + "'Public manual groundwater level measurements in the NGWMN " + "exchange format, sourced from the Ocotillo observation model.'" + ) + ) + + op.execute(text(DROP_WELLCONSTRUCTION_SQL)) + op.execute(text(_create_wellconstruction_view())) + op.execute( + text( + 'COMMENT ON VIEW "NGWMN_WellConstruction" IS ' + "'Well casing and screen intervals in the NGWMN exchange format, " + "sourced from the Ocotillo thing/well_screen model.'" + ) + ) + + op.execute(text(DROP_LITHOLOGY_SQL)) + op.execute(text(_create_lithology_view())) + op.execute( + text( + 'COMMENT ON VIEW "NGWMN_Lithology" IS ' + "'Lithology intervals in the NGWMN exchange format, sourced from " + "the Ocotillo geologic formation associations.'" + ) + ) + + +def downgrade() -> None: + op.execute(text(DROP_LITHOLOGY_SQL)) + op.execute(text(DROP_WELLCONSTRUCTION_SQL)) + op.execute(text(DROP_WATERLEVELS_SQL)) diff --git a/alembic/versions/v0w1x2y3z4a5_add_transducer_daily_data_materialized_view.py b/alembic/versions/v0w1x2y3z4a5_add_transducer_daily_data_materialized_view.py new file mode 100644 index 000000000..c8d859847 --- /dev/null +++ b/alembic/versions/v0w1x2y3z4a5_add_transducer_daily_data_materialized_view.py @@ -0,0 +1,116 @@ +"""add transducer daily data materialized view + +Aggregates raw transducer observations into one row per well, parameter, +day, and QC status. This is the new-model replacement for the legacy +NMA_WaterLevelsContinuous_Pressure_Daily table, which AMPAPI rebuilt +nightly from the raw continuous pressure record. + +Revision ID: v0w1x2y3z4a5 +Revises: u8v9w0x1y2z3 +Create Date: 2026-06-11 00:00:00.000000 +""" + +from typing import Sequence, Union + +from alembic import op +from sqlalchemy import inspect, text + +# revision identifiers, used by Alembic. +revision: str = "v0w1x2y3z4a5" +down_revision: Union[str, Sequence[str], None] = "u8v9w0x1y2z3" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +REQUIRED_TABLES = { + "transducer_observation", + "deployment", + "thing", + "parameter", +} + +DROP_VIEW_SQL = "DROP MATERIALIZED VIEW IF EXISTS transducer_daily_data" + + +def _create_transducer_daily_data_view() -> str: + # transducer_observation.value is depth to water in feet below ground + # surface (the transfer wrote DepthToWaterBGS directly), so the daily + # depth columns need no measuring-point correction. + # + # The transfer wrote legacy timestamps unshifted (naive local clock + # readings stored as UTC), so bucketing on the UTC date preserves the + # original measurement dates. + # + # QC status: the transfer marked QCed rows release_status='public' and + # un-reviewed rows 'private', so qced mirrors the legacy QCed flag. + return """ + CREATE MATERIALIZED VIEW transducer_daily_data AS + SELECT + d.thing_id, + t.name AS point_id, + tob.parameter_id, + p.parameter_name, + (tob.observation_datetime AT TIME ZONE 'UTC')::date AS date_measured, + (tob.release_status = 'public') AS qced, + avg(tob.value) AS depth_to_water_bgs, + min(tob.value) AS depth_to_water_bgs_min, + max(tob.value) AS depth_to_water_bgs_max, + count(*) AS measurement_count, + min(tob.observation_datetime) AS first_measurement_at, + max(tob.observation_datetime) AS last_measurement_at, + avg(tob.nma_waterlevelscontinuous_pressure_temperature_water) AS temperature_water, + avg(tob.nma_waterlevelscontinuous_pressure_water_head) AS water_head, + avg(tob.nma_waterlevelscontinuous_pressure_water_head_adjusted) AS water_head_adjusted, + avg(tob.nma_waterlevelscontinuous_pressure_conddl_ms_cm) AS conddl_ms_cm + FROM transducer_observation AS tob + JOIN deployment AS d ON d.id = tob.deployment_id + JOIN thing AS t ON t.id = d.thing_id + JOIN parameter AS p ON p.id = tob.parameter_id + GROUP BY + d.thing_id, + t.name, + tob.parameter_id, + p.parameter_name, + (tob.observation_datetime AT TIME ZONE 'UTC')::date, + (tob.release_status = 'public') + """ + + +def upgrade() -> None: + bind = op.get_bind() + inspector = inspect(bind) + existing_tables = set(inspector.get_table_names(schema="public")) + missing = REQUIRED_TABLES - existing_tables + if missing: + raise RuntimeError( + "Cannot create transducer_daily_data. Missing required tables: " + f"{', '.join(sorted(missing))}" + ) + + op.execute(text(DROP_VIEW_SQL)) + op.execute(text(_create_transducer_daily_data_view())) + op.execute( + text( + "COMMENT ON MATERIALIZED VIEW transducer_daily_data IS " + "'Daily aggregates of transducer observations per well, parameter, " + "and QC status. Replacement for the legacy " + "NMA_WaterLevelsContinuous_Pressure_Daily table. Refresh with " + "REFRESH MATERIALIZED VIEW CONCURRENTLY transducer_daily_data.'" + ) + ) + # Unique index required for REFRESH MATERIALIZED VIEW CONCURRENTLY. + op.execute( + text( + "CREATE UNIQUE INDEX ux_transducer_daily_data_key " + "ON transducer_daily_data (thing_id, parameter_id, date_measured, qced)" + ) + ) + op.execute( + text( + "CREATE INDEX ix_transducer_daily_data_point_id_date " + "ON transducer_daily_data (point_id, date_measured)" + ) + ) + + +def downgrade() -> None: + op.execute(text(DROP_VIEW_SQL)) diff --git a/alembic/versions/w1x2y3z4a5b6_drop_child_release_filters_from_ngwmn_views.py b/alembic/versions/w1x2y3z4a5b6_drop_child_release_filters_from_ngwmn_views.py new file mode 100644 index 000000000..5fe815a7c --- /dev/null +++ b/alembic/versions/w1x2y3z4a5b6_drop_child_release_filters_from_ngwmn_views.py @@ -0,0 +1,133 @@ +"""drop child-row release filters from NGWMN construction/lithology views + +The release_status filters added to NGWMN_WellConstruction and +NGWMN_Lithology for well_screen, well_casing_material, and +thing_geologic_formation_association rows emptied those exports in +practice: the transfers never set release_status on those child tables, +so every row defaults to 'draft' (verified 3005/3005 well_screen, +63/63 well_casing_material, 993/993 associations on transferred data). + +Construction and lithology rows are attributes of the well, and the +well's own release_status (genuinely managed: public/private) remains +the gate. The thing-level filters are kept; only the child-row +predicates are removed. NGWMN_WaterLevels is unchanged, since the whole +field-data chain it filters on does carry real release values. + +Revision ID: w1x2y3z4a5b6 +Revises: v0w1x2y3z4a5 +Create Date: 2026-06-12 00:00:00.000000 +""" + +from typing import Sequence, Union + +from alembic import op +from sqlalchemy import inspect, text + +# revision identifiers, used by Alembic. +revision: str = "w1x2y3z4a5b6" +down_revision: Union[str, Sequence[str], None] = "v0w1x2y3z4a5" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +REQUIRED_TABLES = { + "thing", + "well_screen", + "well_casing_material", + "thing_geologic_formation_association", + "geologic_formation", +} + +DROP_WELLCONSTRUCTION_SQL = 'DROP VIEW IF EXISTS "NGWMN_WellConstruction"' +DROP_LITHOLOGY_SQL = 'DROP VIEW IF EXISTS "NGWMN_Lithology"' + + +def _create_wellconstruction_view(with_child_filters: bool) -> str: + screen_filter = " AND ws.release_status = 'public'" if with_child_filters else "" + material_filter = " AND wcm.release_status = 'public'" if with_child_filters else "" + return f""" + CREATE VIEW "NGWMN_WellConstruction" AS + SELECT + t.name AS "PointID", + CASE WHEN t.well_casing_depth IS NOT NULL THEN 0::double precision END AS "CasingTop", + t.well_casing_depth AS "CasingBottom", + CASE WHEN t.well_casing_depth IS NOT NULL THEN 'ft bgs' END AS "CasingDepthUnits", + ws.screen_depth_top AS "ScreenTop", + ws.screen_depth_bottom AS "ScreenBottom", + CASE WHEN ws.screen_depth_bottom IS NOT NULL THEN 'ft bgs' END AS "ScreenBottomUnit", + ws.screen_description AS "ScreenDescription", + cm.materials AS "CasingDescription" + FROM thing AS t + LEFT JOIN well_screen AS ws + ON ws.thing_id = t.id{screen_filter} + LEFT JOIN LATERAL ( + SELECT string_agg(wcm.material, ', ' ORDER BY wcm.material) AS materials + FROM well_casing_material AS wcm + WHERE wcm.thing_id = t.id{material_filter} + ) AS cm ON TRUE + WHERE t.thing_type = 'water well' + AND t.release_status = 'public' + """ + + +def _create_lithology_view(with_child_filters: bool) -> str: + association_filter = ( + " AND tgfa.release_status = 'public'\n" if with_child_filters else "" + ) + return f""" + CREATE VIEW "NGWMN_Lithology" AS + SELECT + tgfa.id AS "OBJECTID", + t.name AS "PointID", + gf.lithology AS "Lithology", + gf.lithology AS "TERM", + NULL::character varying AS "StratSource", + tgfa.top_depth AS "StratTop", + CASE WHEN tgfa.top_depth IS NOT NULL THEN 'ft bgs' END AS "StratTopUnit", + tgfa.bottom_depth AS "StratBottom", + CASE WHEN tgfa.bottom_depth IS NOT NULL THEN 'ft bgs' END AS "StratBottomUnit" + FROM thing_geologic_formation_association AS tgfa + JOIN thing AS t ON t.id = tgfa.thing_id + JOIN geologic_formation AS gf ON gf.id = tgfa.geologic_formation_id + WHERE gf.lithology IS NOT NULL +{association_filter} AND t.release_status = 'public' + """ + + +def _recreate_views(with_child_filters: bool) -> None: + bind = op.get_bind() + inspector = inspect(bind) + existing_tables = set(inspector.get_table_names(schema="public")) + missing = REQUIRED_TABLES - existing_tables + if missing: + raise RuntimeError( + "Cannot recreate NGWMN views. Missing required tables: " + f"{', '.join(sorted(missing))}" + ) + + op.execute(text(DROP_WELLCONSTRUCTION_SQL)) + op.execute(text(_create_wellconstruction_view(with_child_filters))) + op.execute( + text( + 'COMMENT ON VIEW "NGWMN_WellConstruction" IS ' + "'Well casing and screen intervals in the NGWMN exchange format, " + "sourced from the Ocotillo thing/well_screen model.'" + ) + ) + + op.execute(text(DROP_LITHOLOGY_SQL)) + op.execute(text(_create_lithology_view(with_child_filters))) + op.execute( + text( + 'COMMENT ON VIEW "NGWMN_Lithology" IS ' + "'Lithology intervals in the NGWMN exchange format, sourced from " + "the Ocotillo geologic formation associations.'" + ) + ) + + +def upgrade() -> None: + _recreate_views(with_child_filters=False) + + +def downgrade() -> None: + _recreate_views(with_child_filters=True) diff --git a/alembic/versions/x2y3z4a5b6c7_schedule_nightly_matview_refresh_pg_cron.py b/alembic/versions/x2y3z4a5b6c7_schedule_nightly_matview_refresh_pg_cron.py new file mode 100644 index 000000000..0e50fdef9 --- /dev/null +++ b/alembic/versions/x2y3z4a5b6c7_schedule_nightly_matview_refresh_pg_cron.py @@ -0,0 +1,132 @@ +"""schedule nightly materialized-view refresh via pg_cron + +Registers a pg_cron job that refreshes the materialized views once a +night. The job calls a SQL helper function, +``public.refresh_materialized_views()``, which discovers every +materialized view in the public schema from the catalog at run time -- so +this migration stays immutable and self-contained, and views added by +later migrations are refreshed without any rescheduling. + +pg_cron is a *production-only* dependency. It requires the extension to be +loaded via ``shared_preload_libraries`` on the database server, which the +development docker-compose Postgres image does not do. To avoid breaking +``alembic upgrade head`` in development (and in test/CI), this migration is a +no-op unless ``ENABLE_PG_CRON`` is truthy in the environment. Production sets +``ENABLE_PG_CRON=1``; everywhere else the migration records itself as applied +without touching pg_cron. See ``docs/pg_cron-nightly-refresh.md``. + +Revision ID: x2y3z4a5b6c7 +Revises: w1x2y3z4a5b6 +Create Date: 2026-06-17 00:00:00.000000 +""" + +from typing import Sequence, Union + +from alembic import op +from sqlalchemy import text + +from services.env import get_bool_env + +# revision identifiers, used by Alembic. +revision: str = "x2y3z4a5b6c7" +down_revision: Union[str, Sequence[str], None] = "w1x2y3z4a5b6" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +# Name of the pg_cron job. Used to (re)register and to unschedule. +CRON_JOB_NAME = "refresh-materialized-views" + +# Nightly schedule in standard cron syntax. pg_cron interprets this in the +# database server's timezone (UTC on Cloud SQL), so 09:00 UTC is roughly +# 02:00-03:00 in US Mountain time -- comfortably off-peak. +CRON_SCHEDULE = "0 9 * * *" + + +# Helper function the cron job calls. It discovers every materialized view in +# the public schema from the catalog at run time rather than from a baked-in +# list. This keeps the migration immutable and self-contained -- it does not +# depend on mutable application code, and views added by later migrations are +# picked up automatically without rescheduling. +# +# Plain (non-concurrent) REFRESH is used deliberately: REFRESH ... CONCURRENTLY +# cannot run inside the implicit transaction of a PL/pgSQL function, and the +# nightly window tolerates the brief exclusive lock. +_REFRESH_FUNCTION_SQL = r""" +CREATE OR REPLACE FUNCTION public.refresh_materialized_views() +RETURNS void +LANGUAGE plpgsql +AS $func$ +DECLARE + r record; +BEGIN + FOR r IN + SELECT matviewname + FROM pg_matviews + WHERE schemaname = 'public' + ORDER BY matviewname + LOOP + EXECUTE format('REFRESH MATERIALIZED VIEW %I', r.matviewname); + END LOOP; +END; +$func$; +""" + + +def _pg_cron_enabled() -> bool: + """pg_cron is only wired up where the server explicitly enables it.""" + return get_bool_env("ENABLE_PG_CRON", False) is True + + +def upgrade() -> None: + if not _pg_cron_enabled(): + print( + "ENABLE_PG_CRON is not set; skipping pg_cron job registration " + "(expected in development, test, and CI)." + ) + return + + bind = op.get_bind() + + # Requires shared_preload_libraries to include 'pg_cron' and the extension + # to be creatable in this database (cron.database_name = this DB). See docs. + op.execute(text("CREATE EXTENSION IF NOT EXISTS pg_cron")) + + # (Re)create the refresh helper. + op.execute(text(_REFRESH_FUNCTION_SQL)) + + # Drop any previously registered job with the same name so re-running this + # migration (or a re-deploy) does not accumulate duplicate schedules. + op.execute( + text( + "SELECT cron.unschedule(jobid) FROM cron.job " "WHERE jobname = :name" + ).bindparams(name=CRON_JOB_NAME) + ) + + bind.execute( + text("SELECT cron.schedule(:name, :sched, :cmd)").bindparams( + name=CRON_JOB_NAME, + sched=CRON_SCHEDULE, + cmd="SELECT public.refresh_materialized_views();", + ) + ) + + print( + f"Registered pg_cron job '{CRON_JOB_NAME}' " + f"(schedule '{CRON_SCHEDULE}', server timezone)." + ) + + +def downgrade() -> None: + if not _pg_cron_enabled(): + print("ENABLE_PG_CRON is not set; nothing to unschedule.") + return + + op.execute( + text( + "SELECT cron.unschedule(jobid) FROM cron.job " "WHERE jobname = :name" + ).bindparams(name=CRON_JOB_NAME) + ) + op.execute(text("DROP FUNCTION IF EXISTS public.refresh_materialized_views()")) + # The pg_cron extension itself is left installed: it is a server-level + # capability that other jobs may depend on, and dropping it is not the + # inverse of "schedule a job". diff --git a/api/geospatial.py b/api/geospatial.py index 082979f8a..2d649d334 100644 --- a/api/geospatial.py +++ b/api/geospatial.py @@ -14,16 +14,21 @@ # limitations under the License. # =============================================================================== import json +import os +import shutil +import tempfile from typing import Annotated, List from fastapi import APIRouter, Query, HTTPException from fastapi.responses import FileResponse from geoalchemy2.shape import to_shape from shapely.io import to_geojson -from starlette.responses import JSONResponse +from starlette.background import BackgroundTask +from starlette.responses import StreamingResponse from core.dependencies import session_dependency, viewer_dependency from db import Group +from db.engine import session_ctx from schemas.thing import FeatureCollectionResponse from services.geospatial_helper import create_shapefile, get_thing_features from services.query_helper import simple_get_by_id @@ -54,8 +59,7 @@ def get_geospatial( """ if format_ == "geojson": - content = get_feature_collection(session, thing_type, group) - return JSONResponse(content=content, media_type="application/geo+json") + return get_feature_collection(thing_type, group) else: return get_location_shapefile(session, thing_type, group) @@ -89,17 +93,17 @@ def get_project_area( def get_feature_collection( - session: session_dependency, thing_type: List[str] | None = None, group: Annotated[ str | int, Query(title="group", description="group", alias="group") ] = None, -) -> FeatureCollectionResponse: - """ - Endpoint to retrieve a GeoJSON FeatureCollection. +) -> StreamingResponse: """ + Retrieve a GeoJSON FeatureCollection. - things = get_thing_features(session, thing_type, group) + Streamed feature-by-feature so the entire result set is never buffered in + memory at once. + """ def make_feature_dict(thing, geometry, elevation, *other): geometry = json.loads(geometry) @@ -115,12 +119,18 @@ def make_feature_dict(thing, geometry, elevation, *other): "geometry": geometry, } - features = [make_feature_dict(*item) for item in things] + def generate(): + # The request-scoped session is closed before this response body + # streams, so open a dedicated session scoped to the stream. + with session_ctx() as stream_session: + yield '{"type": "FeatureCollection", "features": [' + first = True + for item in get_thing_features(stream_session, thing_type, group): + yield ("" if first else ",") + json.dumps(make_feature_dict(*item)) + first = False + yield "]}" - return { - "type": "FeatureCollection", - "features": features, - } + return StreamingResponse(generate(), media_type="application/geo+json") def get_location_shapefile( @@ -133,16 +143,32 @@ def get_location_shapefile( """ things = get_thing_features(session, thing_type, group) - create_shapefile(things, "things.shp") - # Return the shapefile as a zip (optional: zip the .shp, .shx, .dbf files) - import zipfile + # Write into a temp dir: the App Engine app directory is read-only, and /tmp + # is RAM-backed, so build here and clean up after the response is sent. + tmpdir = tempfile.mkdtemp() + try: + shp_path = os.path.join(tmpdir, "things.shp") + zip_path = os.path.join(tmpdir, "things.zip") + + create_shapefile(things, shp_path) + + import zipfile + + with zipfile.ZipFile(zip_path, "w") as zf: + for ext in ["shp", "shx", "dbf"]: + zf.write(os.path.join(tmpdir, f"things.{ext}"), arcname=f"things.{ext}") + except Exception: + # BackgroundTask only runs on a successful response, so clean up here to + # avoid leaking temp dirs when generation fails. + shutil.rmtree(tmpdir, ignore_errors=True) + raise - with zipfile.ZipFile("things.zip", "w") as zf: - for ext in ["shp", "shx", "dbf"]: - zf.write(f"things.{ext}") return FileResponse( - "things.zip", media_type="application/zip", filename="things.zip" + zip_path, + media_type="application/zip", + filename="things.zip", + background=BackgroundTask(shutil.rmtree, tmpdir, ignore_errors=True), ) diff --git a/pyproject.toml b/pyproject.toml index 8f3e8cd2f..db6314a20 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "OcotilloAPI" -version = "1.1.0" +version = "1.1.5" description = "FastAPI backend and CLI for managing Ocotillo groundwater locations, wells, assets, and bulk water-level data transfers." readme = "README.md" requires-python = ">=3.13" diff --git a/services/geospatial_helper.py b/services/geospatial_helper.py index fc1118aa8..fb8777178 100644 --- a/services/geospatial_helper.py +++ b/services/geospatial_helper.py @@ -13,6 +13,8 @@ # See the License for the specific language governing permissions and # limitations under the License. # =============================================================================== +from typing import Iterator + import shapefile from geoalchemy2.functions import ST_GeomFromText, ST_Within, ST_AsGeoJSON from geoalchemy2.shape import to_shape @@ -31,7 +33,7 @@ def get_thing_features( session, thing_type: list | str | None, group: str | int | None -) -> list: +) -> Iterator: # sql = ( # select(Thing, ST_AsGeoJSON(Location.point).label("geojson")) # .join(LocationThingAssociation, Thing.id == LocationThingAssociation.thing_id) @@ -87,15 +89,29 @@ def get_thing_features( else: sql = sql.where(Group.id == group) - # unique needs to be invoked to prevent duplicates from eager loading - return session.execute(sql).unique().all() + # Stream the result with yield_per so the whole table is never buffered in + # memory at once. Thing has no eager-loaded collections (all relationships + # are lazy), so unique() is unnecessary -- and unique() is incompatible with + # yield_per anyway. Dedup defensively by id with a bounded set of ints in + # case the joins ever produce duplicate rows. + seen = set() + result = session.execute(sql.execution_options(yield_per=1000)) + for row in result: + thing_id = row[0].id + if thing_id in seen: + continue + seen.add(thing_id) + yield row def create_shapefile(things: list, filename: str = "things.shp") -> None: # Create a point shapefile with shapefile.Writer(filename, shapeType=shapefile.POINT) as shp: - shp.field("id", "L") + # Field schema must match the values written in shp.record() below: + # id (numeric), name (char), elevation (numeric). + shp.field("id", "N") shp.field("name", "C") + shp.field("elevation", "N", decimal=3) for thing, point, elevation in things: # Assume loc.point is WKT or a Shapely geometry or GeoJSON diff --git a/uv.lock b/uv.lock index e1ac33e08..fb65415f3 100644 --- a/uv.lock +++ b/uv.lock @@ -1489,7 +1489,7 @@ wheels = [ [[package]] name = "ocotilloapi" -version = "1.0.0" +version = "1.1.5" source = { editable = "." } dependencies = [ { name = "aiofiles" },