From 847893570f025445224395e9ebc6f90380ad96fe Mon Sep 17 00:00:00 2001 From: Dylan Lee Date: Tue, 9 Jun 2026 11:43:56 -0400 Subject: [PATCH 1/5] update: NHF schema and ID types for NHF v1.2.0 --- src/icefabric/hydrofabric/subset_nhf.py | 6 +++--- .../schemas/iceberg_tables/hydrofabric_update.py | 14 +++++++++++++- 2 files changed, 16 insertions(+), 4 deletions(-) diff --git a/src/icefabric/hydrofabric/subset_nhf.py b/src/icefabric/hydrofabric/subset_nhf.py index f425240..fe02838 100644 --- a/src/icefabric/hydrofabric/subset_nhf.py +++ b/src/icefabric/hydrofabric/subset_nhf.py @@ -31,9 +31,9 @@ def _build_upstream_dict_from_nexus( """Build upstream connectivity dictionary from flowpath nexus connections.""" fp_pl = flowpaths_pl.with_columns( [ - pl.col(edge_id).cast(pl.Int32), - pl.col(f"up_{node_id}").cast(pl.Int32), - pl.col(f"dn_{node_id}").cast(pl.Int32), + pl.col(edge_id).cast(pl.Int64), + pl.col(f"up_{node_id}").cast(pl.Int64), + pl.col(f"dn_{node_id}").cast(pl.Int64), ] ) nexus_to_downstream = fp_pl.select( diff --git a/src/icefabric/schemas/iceberg_tables/hydrofabric_update.py b/src/icefabric/schemas/iceberg_tables/hydrofabric_update.py index 8884f71..b2063b5 100644 --- a/src/icefabric/schemas/iceberg_tables/hydrofabric_update.py +++ b/src/icefabric/schemas/iceberg_tables/hydrofabric_update.py @@ -1451,6 +1451,10 @@ class Lakes: Reservoir index for Medium Range configuration reservoir_index_Short_Range : float Reservoir index for Short Range configuration + dam_id : str + Dam identifier + nidid : str + National Inventory of Dams identifier geometry : binary Spatial Geometry (POINT format) - stored in WKB binary format """ @@ -1484,6 +1488,8 @@ def columns(cls) -> list[str]: "reservoir_index_GDL_AK", "reservoir_index_Medium_Range", "reservoir_index_Short_Range", + "dam_id", + "nidid", "geometry", ] @@ -1516,6 +1522,8 @@ def schema(cls) -> Schema: "Reservoir index for GDL AK configuration", "Reservoir index for Medium Range configuration", "Reservoir index for Short Range configuration", + "Dam identifier", + "National Inventory of Dams identifier", "Spatial Geometry (POINT format) - stored in WKB binary format", ] return Schema( @@ -1544,7 +1552,9 @@ def schema(cls) -> Schema: NestedField(23, "reservoir_index_GDL_AK", DoubleType(), required=False, doc=desc[22]), NestedField(24, "reservoir_index_Medium_Range", DoubleType(), required=False, doc=desc[23]), NestedField(25, "reservoir_index_Short_Range", DoubleType(), required=False, doc=desc[24]), - NestedField(26, "geometry", BinaryType(), required=False, doc=desc[25]), + NestedField(26, "dam_id", StringType(), required=False, doc=desc[25]), + NestedField(27, "nidid", StringType(), required=False, doc=desc[26]), + NestedField(28, "geometry", BinaryType(), required=False, doc=desc[27]), identifier_field_ids=[1], ) @@ -1578,6 +1588,8 @@ def arrow_schema(cls) -> pa.Schema: pa.field("reservoir_index_GDL_AK", pa.float64(), nullable=True), pa.field("reservoir_index_Medium_Range", pa.float64(), nullable=True), pa.field("reservoir_index_Short_Range", pa.float64(), nullable=True), + pa.field("dam_id", pa.string(), nullable=True), + pa.field("nidid", pa.string(), nullable=True), pa.field("geometry", pa.binary(), nullable=True), ] ) From e4b3b2bbc1f4ef79366aebae0403df206c81d89a Mon Sep 17 00:00:00 2001 From: Dylan Lee Date: Tue, 9 Jun 2026 12:19:53 -0400 Subject: [PATCH 2/5] update: add additional fields to NHF schema --- .../iceberg_tables/hydrofabric_update.py | 23 +++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/src/icefabric/schemas/iceberg_tables/hydrofabric_update.py b/src/icefabric/schemas/iceberg_tables/hydrofabric_update.py index b2063b5..48f34c3 100644 --- a/src/icefabric/schemas/iceberg_tables/hydrofabric_update.py +++ b/src/icefabric/schemas/iceberg_tables/hydrofabric_update.py @@ -238,6 +238,7 @@ def columns(cls) -> list[str]: "expon", "max_gw_storage", "geometry", + "gid", ] @classmethod @@ -395,6 +396,7 @@ def schema(cls) -> Schema: NestedField(68, "expon", DoubleType(), required=False, doc=desc[67]), NestedField(69, "max_gw_storage", DoubleType(), required=False, doc=desc[68]), NestedField(70, "geometry", BinaryType(), required=False, doc=desc[69]), + NestedField(71, "gid", StringType(), required=False, doc="Geolocation Plus Code identifier"), identifier_field_ids=[1], ) @@ -480,6 +482,7 @@ def arrow_schema(cls) -> pa.Schema: pa.field("expon", pa.float64(), nullable=True), pa.field("max_gw_storage", pa.float64(), nullable=True), pa.field("geometry", pa.binary(), nullable=True), + pa.field("gid", pa.string(), nullable=True), ] ) @@ -596,6 +599,7 @@ def columns(cls) -> list[str]: "r_ml", "fp_to_id", "geometry", + "gid", ] @classmethod @@ -673,6 +677,7 @@ def schema(cls) -> Schema: NestedField(29, "r_ml", FloatType(), required=False, doc=desc[28]), NestedField(30, "fp_to_id", LongType(), required=False, doc=desc[29]), NestedField(31, "geometry", BinaryType(), required=False, doc=desc[30]), + NestedField(32, "gid", StringType(), required=False, doc="Geolocation Plus Code identifier"), identifier_field_ids=[1], ) @@ -719,6 +724,7 @@ def arrow_schema(cls) -> pa.Schema: pa.field("r_ml", pa.float32(), nullable=True), pa.field("fp_to_id", pa.int64(), nullable=True), pa.field("geometry", pa.binary(), nullable=True), + pa.field("gid", pa.string(), nullable=True), ] ) @@ -754,6 +760,7 @@ def columns(cls) -> list[str]: "dn_fp_id", "vpu_id", "geometry", + "gid", ] @classmethod @@ -777,6 +784,7 @@ def schema(cls) -> Schema: NestedField(2, "dn_fp_id", LongType(), required=False, doc=desc[1]), NestedField(3, "vpu_id", StringType(), required=False, doc=desc[2]), NestedField(4, "geometry", BinaryType(), required=False, doc=desc[3]), + NestedField(5, "gid", StringType(), required=False, doc="Geolocation Plus Code identifier"), identifier_field_ids=[1], ) @@ -796,6 +804,7 @@ def arrow_schema(cls) -> pa.Schema: pa.field("dn_fp_id", pa.int64(), nullable=True), pa.field("vpu_id", pa.string(), nullable=True), pa.field("geometry", pa.binary(), nullable=True), + pa.field("gid", pa.string(), nullable=True), ] ) @@ -837,6 +846,7 @@ def columns(cls) -> list[str]: "div_id", "mainstem_virtual_fp_id", "segment_order", + "gid", ] @classmethod @@ -864,6 +874,7 @@ def schema(cls) -> Schema: NestedField(4, "div_id", LongType(), required=False, doc=desc[3]), NestedField(5, "mainstem_virtual_fp_id", LongType(), required=False, doc=desc[4]), NestedField(6, "segment_order", LongType(), required=False, doc=desc[5]), + NestedField(7, "gid", StringType(), required=False, doc="Geolocation Plus Code identifier"), identifier_field_ids=[1], ) @@ -885,6 +896,7 @@ def arrow_schema(cls) -> pa.Schema: pa.field("div_id", pa.int64(), nullable=True), pa.field("mainstem_virtual_fp_id", pa.int64(), nullable=True), pa.field("segment_order", pa.int64(), nullable=True), + pa.field("gid", pa.string(), nullable=True), ] ) @@ -974,6 +986,7 @@ def columns(cls) -> list[str]: "dn_virtual_nex_id", "virtual_fp_id", "geometry", + "gid", ] @classmethod @@ -1070,6 +1083,7 @@ def arrow_schema(cls) -> pa.Schema: pa.field("dn_virtual_nex_id", pa.float64(), nullable=True), pa.field("virtual_fp_id", pa.float64(), nullable=True), pa.field("geometry", pa.binary(), nullable=True), + pa.field("gid", pa.string(), nullable=True), ] ) @@ -1135,6 +1149,7 @@ def columns(cls) -> list[str]: "mainstem_virtual_fp_id", "segment_order", "geometry", + "gid", ] @classmethod @@ -1178,6 +1193,7 @@ def schema(cls) -> Schema: NestedField(12, "mainstem_virtual_fp_id", DoubleType(), required=False, doc=desc[11]), NestedField(13, "segment_order", DoubleType(), required=False, doc=desc[12]), NestedField(14, "geometry", BinaryType(), required=False, doc=desc[13]), + NestedField(15, "gid", StringType(), required=False, doc="Geolocation Plus Code identifier"), identifier_field_ids=[1], ) @@ -1207,6 +1223,7 @@ def arrow_schema(cls) -> pa.Schema: pa.field("mainstem_virtual_fp_id", pa.float64(), nullable=True), pa.field("segment_order", pa.float64(), nullable=True), pa.field("geometry", pa.binary(), nullable=True), + pa.field("gid", pa.string(), nullable=True), ] ) @@ -1257,6 +1274,7 @@ def columns(cls) -> list[str]: "percentage_area_contribution", "vpu_id", "geometry", + "gid", ] @classmethod @@ -1290,6 +1308,7 @@ def schema(cls) -> Schema: NestedField(7, "percentage_area_contribution", DoubleType(), required=False, doc=desc[6]), NestedField(8, "vpu_id", StringType(), required=False, doc=desc[7]), NestedField(9, "geometry", BinaryType(), required=False, doc=desc[8]), + NestedField(10, "gid", StringType(), required=False, doc="Geolocation Plus Code identifier"), identifier_field_ids=[1], ) @@ -1314,6 +1333,7 @@ def arrow_schema(cls) -> pa.Schema: pa.field("percentage_area_contribution", pa.float64(), nullable=True), pa.field("vpu_id", pa.string(), nullable=True), pa.field("geometry", pa.binary(), nullable=True), + pa.field("gid", pa.string(), nullable=True), ] ) @@ -1349,6 +1369,7 @@ def columns(cls) -> list[str]: "dn_virtual_fp_id", "vpu_id", "geometry", + "gid", ] @classmethod @@ -1372,6 +1393,7 @@ def schema(cls) -> Schema: NestedField(2, "dn_virtual_fp_id", LongType(), required=False, doc=desc[1]), NestedField(3, "vpu_id", StringType(), required=False, doc=desc[2]), NestedField(4, "geometry", BinaryType(), required=False, doc=desc[3]), + NestedField(5, "gid", StringType(), required=False, doc="Geolocation Plus Code identifier"), identifier_field_ids=[1], ) @@ -1391,6 +1413,7 @@ def arrow_schema(cls) -> pa.Schema: pa.field("dn_virtual_fp_id", pa.int64(), nullable=True), pa.field("vpu_id", pa.string(), nullable=True), pa.field("geometry", pa.binary(), nullable=True), + pa.field("gid", pa.string(), nullable=True), ] ) From 865f132fe137eddbeec94d69944c70cb9e7e7494 Mon Sep 17 00:00:00 2001 From: Dylan Lee Date: Tue, 9 Jun 2026 13:46:45 -0400 Subject: [PATCH 3/5] feat: remove references to waterbodies layer --- src/icefabric/hydrofabric/subset_nhf.py | 18 +++++------------- .../schemas/iceberg_tables/__init__.py | 2 -- .../schemas/iceberg_tables/nhf_snapshots.py | 13 +++++-------- 3 files changed, 10 insertions(+), 23 deletions(-) diff --git a/src/icefabric/hydrofabric/subset_nhf.py b/src/icefabric/hydrofabric/subset_nhf.py index fe02838..e030928 100644 --- a/src/icefabric/hydrofabric/subset_nhf.py +++ b/src/icefabric/hydrofabric/subset_nhf.py @@ -253,7 +253,6 @@ def generate_subset_from_ids( f = { "fp": ex.submit(source.load_filtered, "flowpaths", "fp_id", flowpath_ids), "div": ex.submit(source.load_filtered, "divides", "div_id", flowpath_ids), - "wb": ex.submit(source.load_filtered, "waterbodies", "fp_id", flowpath_ids), "gages": ex.submit(source.load_filtered, "gages", "fp_id", flowpath_ids), "ref_fp": ex.submit(source.load_filtered, "reference_flowpaths", "div_id", flowpath_ids), "lakes": ex.submit(source.load_filtered, "lakes", "fp_id", flowpath_ids), @@ -261,7 +260,6 @@ def generate_subset_from_ids( } subset_fp = f["fp"].result() subset_div = f["div"].result() - subset_wb = f["wb"].result() subset_gages = f["gages"].result() subset_ref_fp = f["ref_fp"].result() subset_lakes = f["lakes"].result() @@ -292,9 +290,9 @@ def generate_subset_from_ids( + subset_fp.filter(pl.col("dn_nex_id").is_not_null())["dn_nex_id"].cast(pl.Int64).to_list() ) all_v_fp_ids = set(subset_ref_fp["virtual_fp_id"].to_list()) - wb_hy_ids = subset_wb["hy_id"].to_list() if "hy_id" in subset_wb.columns else [] + lakes_hy_ids = subset_lakes["hy_id"].to_list() if "hy_id" in subset_lakes.columns else [] gage_hy_ids = subset_gages["hy_id"].to_list() if "hy_id" in subset_gages.columns else [] - all_hy_ids = set(wb_hy_ids + gage_hy_ids) + all_hy_ids = set(lakes_hy_ids + gage_hy_ids) # Wave 2: nex_id/virtual_fp_id filtered (parallel) with ThreadPoolExecutor(max_workers=2) as ex: @@ -345,7 +343,6 @@ def generate_subset_from_ids( "divides": pl_to_gdf(subset_div, crs=crs), "virtual_nexus": pl_to_gdf(subset_v_nex, crs=crs), "virtual_flowpaths": pl_to_gdf(subset_v_fp, crs=crs), - "waterbodies": pl_to_gdf(subset_wb, crs=crs) if len(subset_wb) > 0 else subset_wb.to_pandas(), "gages": pl_to_gdf(subset_gages, crs=crs) if len(subset_gages) > 0 else subset_gages.to_pandas(), "lakes": pl_to_gdf(subset_lakes, crs=crs) if len(subset_lakes) > 0 else subset_lakes.to_pandas(), "reference_flowpaths": subset_ref_fp.to_pandas(), @@ -363,7 +360,6 @@ def generate_subset_from_ids( "divides", "virtual_nexus", "virtual_flowpaths", - "waterbodies", "gages", "lakes", ] @@ -425,14 +421,12 @@ def generate_subset_virtual_only( f = { "fp": ex.submit(source.load_filtered, "flowpaths", "div_id", div_ids), "div": ex.submit(source.load_filtered, "divides", "div_id", div_ids), - "wb": ex.submit(source.load_filtered, "waterbodies", "div_id", div_ids), "gages": ex.submit(source.load_filtered, "gages", "div_id", div_ids), "ref_fp": ex.submit(source.load_filtered, "reference_flowpaths", "div_id", div_ids), "lakes": ex.submit(source.load_filtered, "lakes", "div_id", div_ids), } subset_fp = f["fp"].result() subset_div = f["div"].result() - subset_wb = f["wb"].result() subset_gages = f["gages"].result() subset_ref_fp = f["ref_fp"].result() subset_lakes = f["lakes"].result() @@ -463,10 +457,10 @@ def generate_subset_virtual_only( .to_list() ) - # Derive hydrolocation IDs from gages / waterbodies in this divide - wb_hy_ids = subset_wb["hy_id"].to_list() if "hy_id" in subset_wb.columns else [] + # Derive hydrolocation IDs from gages and lakes in this divide + lakes_hy_ids = subset_lakes["hy_id"].to_list() if "hy_id" in subset_lakes.columns else [] gage_hy_ids = subset_gages["hy_id"].to_list() if "hy_id" in subset_gages.columns else [] - all_hy_ids = set(wb_hy_ids + gage_hy_ids) + all_hy_ids = set(lakes_hy_ids + gage_hy_ids) # Wave 2: nexus / virtual_nexus / hydrolocations with ThreadPoolExecutor(max_workers=2) as ex: @@ -503,7 +497,6 @@ def generate_subset_virtual_only( "divides": pl_to_gdf(subset_div, crs=crs), "virtual_nexus": pl_to_gdf(subset_v_nex, crs=crs), "virtual_flowpaths": pl_to_gdf(subset_v_fp, crs=crs), - "waterbodies": (pl_to_gdf(subset_wb, crs=crs) if len(subset_wb) > 0 else subset_wb.to_pandas()), "gages": (pl_to_gdf(subset_gages, crs=crs) if len(subset_gages) > 0 else subset_gages.to_pandas()), "lakes": (pl_to_gdf(subset_lakes, crs=crs) if len(subset_lakes) > 0 else subset_lakes.to_pandas()), "reference_flowpaths": subset_ref_fp.to_pandas(), @@ -521,7 +514,6 @@ def generate_subset_virtual_only( "divides", "virtual_nexus", "virtual_flowpaths", - "waterbodies", "gages", "lakes", ] diff --git a/src/icefabric/schemas/iceberg_tables/__init__.py b/src/icefabric/schemas/iceberg_tables/__init__.py index 06c375d..3ccbae3 100644 --- a/src/icefabric/schemas/iceberg_tables/__init__.py +++ b/src/icefabric/schemas/iceberg_tables/__init__.py @@ -9,7 +9,6 @@ ReferenceFlowpaths, VirtualFlowpaths, VirtualNexus, - Waterbodies, ) nhf_layers = { @@ -17,7 +16,6 @@ "flowpaths": Flowpaths, "nexus": Nexus, "reference_flowpaths": ReferenceFlowpaths, - "waterbodies": Waterbodies, "gages": Gages, "virtual_flowpaths": VirtualFlowpaths, "virtual_nexus": VirtualNexus, diff --git a/src/icefabric/schemas/iceberg_tables/nhf_snapshots.py b/src/icefabric/schemas/iceberg_tables/nhf_snapshots.py index 791d68f..fac20d5 100644 --- a/src/icefabric/schemas/iceberg_tables/nhf_snapshots.py +++ b/src/icefabric/schemas/iceberg_tables/nhf_snapshots.py @@ -35,7 +35,6 @@ def columns(cls) -> list[str]: "flowpaths", "nexus", "reference_flowpaths", - "waterbodies", "gages", "virtual_flowpaths", "virtual_nexus", @@ -57,12 +56,11 @@ def schema(cls) -> Schema: NestedField(2, "flowpaths", LongType(), required=False), NestedField(3, "nexus", LongType(), required=False), NestedField(4, "reference_flowpaths", LongType(), required=False), - NestedField(5, "waterbodies", LongType(), required=False), - NestedField(6, "gages", LongType(), required=False), - NestedField(7, "virtual_flowpaths", LongType(), required=False), - NestedField(8, "virtual_nexus", LongType(), required=False), - NestedField(9, "hydrolocations", LongType(), required=False), - NestedField(10, "lakes", LongType(), required=False), + NestedField(5, "gages", LongType(), required=False), + NestedField(6, "virtual_flowpaths", LongType(), required=False), + NestedField(7, "virtual_nexus", LongType(), required=False), + NestedField(8, "hydrolocations", LongType(), required=False), + NestedField(9, "lakes", LongType(), required=False), ) @classmethod @@ -80,7 +78,6 @@ def arrow_schema(cls) -> pa.Schema: pa.field("flowpaths", pa.int64(), nullable=True), pa.field("nexus", pa.int64(), nullable=True), pa.field("reference_flowpaths", pa.int64(), nullable=True), - pa.field("waterbodies", pa.int64(), nullable=True), pa.field("gages", pa.int64(), nullable=True), pa.field("virtual_flowpaths", pa.int64(), nullable=True), pa.field("virtual_nexus", pa.int64(), nullable=True), From cc5911fd1ffe370c728167e00fe336ea3720ad6a Mon Sep 17 00:00:00 2001 From: Dylan Lee Date: Tue, 9 Jun 2026 14:20:27 -0400 Subject: [PATCH 4/5] fix: add terminalpa to flowpaths schema --- src/icefabric/schemas/iceberg_tables/hydrofabric_update.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/icefabric/schemas/iceberg_tables/hydrofabric_update.py b/src/icefabric/schemas/iceberg_tables/hydrofabric_update.py index 48f34c3..d903a5c 100644 --- a/src/icefabric/schemas/iceberg_tables/hydrofabric_update.py +++ b/src/icefabric/schemas/iceberg_tables/hydrofabric_update.py @@ -600,6 +600,7 @@ def columns(cls) -> list[str]: "fp_to_id", "geometry", "gid", + "terminalpa", ] @classmethod @@ -678,6 +679,9 @@ def schema(cls) -> Schema: NestedField(30, "fp_to_id", LongType(), required=False, doc=desc[29]), NestedField(31, "geometry", BinaryType(), required=False, doc=desc[30]), NestedField(32, "gid", StringType(), required=False, doc="Geolocation Plus Code identifier"), + NestedField( + 33, "terminalpa", LongType(), required=False, doc="Terminal path grouping identifier" + ), identifier_field_ids=[1], ) @@ -725,6 +729,7 @@ def arrow_schema(cls) -> pa.Schema: pa.field("fp_to_id", pa.int64(), nullable=True), pa.field("geometry", pa.binary(), nullable=True), pa.field("gid", pa.string(), nullable=True), + pa.field("terminalpa", pa.int64(), nullable=True), ] ) From 89cafde4674d763ac0f219b5bd93a5ea3d08c2ca Mon Sep 17 00:00:00 2001 From: Dylan Lee Date: Tue, 9 Jun 2026 14:27:43 -0400 Subject: [PATCH 5/5] fix: hydrofabric endpoint now returns empty spatial layeers too --- app/routers/hydrofabric/router.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/app/routers/hydrofabric/router.py b/app/routers/hydrofabric/router.py index edc1953..c989e32 100644 --- a/app/routers/hydrofabric/router.py +++ b/app/routers/hydrofabric/router.py @@ -255,7 +255,7 @@ def _release_sem() -> None: spatial_names: list[str] = [] nonspatial_names: list[str] = [] for name, data in output_layers.items(): - if isinstance(data, gpd.GeoDataFrame) and len(data) > 0: + if isinstance(data, gpd.GeoDataFrame): spatial_names.append(name) elif not isinstance(data, gpd.GeoDataFrame): nonspatial_names.append(name)