From ff4c149f512fe69b851f9caa013b071de8ec0456 Mon Sep 17 00:00:00 2001 From: Brandon Keepers Date: Sun, 5 Jul 2026 10:49:57 -0400 Subject: [PATCH 1/3] Work-sized build fan-outs + single vector tile-join MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The last full planet build spent its final 2h40 running contour-bundle alone: tile-join rewrites every tile of the whole archive, and the merge-then-fold-per-layer shape re-paid that planet-wide join once per sparse layer (~90 min each for soundings and drying). Bundle both layers first and fold them into vector.pmtiles in the contour merge's ONE tile-join; fold() is gone from soundings_run/drying_run. The fan-outs also sized themselves to item counts, not work, saturating the shard ceiling (256) for every phase: - downsample: striding by ancestor left one 77-minute subtree straggling behind ~250 spin-up-only shards. Ancestors now bin-pack by parent-webp count (utils.lpt_bins, heaviest-first) and the matrix self-sizes to ceil(total/heaviest) — a shard ~= the heaviest subtree, the wall-clock floor anyway since a read-closed subtree can't split. - bundle: 235 one-group chunks each paid more runner setup than bundling. Groups now bin-pack by pmtiles bytes, same self-sizing. - contours: ~200 FGBs (~10 min of tippecanoe) per shard instead of one runner per ~1 min of work. Aggregate keeps max sharding on purpose: its shard count is the 6h job-cap headroom, and the phase is concurrency-bound regardless. The shards dispatch input is now documented as the ceiling the other phases self-size under. Tile content is unchanged, so no force rebuild is needed. Expected: ~9h40 -> ~7h wall, release ~2.5h earlier. New check_weighted_shards in test_engine.py covers the partition (complete + disjoint, self-sized n, LPT bound) and the pure packer. --- .github/workflows/build.yml | 46 +++++++++++++++++++++++-------------- Justfile | 20 +++++++++------- pipelines/bundle.py | 19 ++++++++------- pipelines/contour_run.py | 14 +++++++---- pipelines/downsampling.py | 35 ++++++++++++++++++++-------- pipelines/drying_run.py | 21 +++-------------- pipelines/soundings_run.py | 18 +-------------- pipelines/test_engine.py | 42 +++++++++++++++++++++++++++++++++ pipelines/utils.py | 14 +++++++++++ 9 files changed, 147 insertions(+), 82 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 89022141..677ad8cc 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -23,7 +23,7 @@ on: required: false default: "" shards: - description: "Aggregate matrix shards (<= 256)" + description: "Fan-out ceiling (<= 256): aggregate shard count; the other phases self-size to their work below it" required: false default: "24" force: @@ -366,10 +366,13 @@ jobs: aws s3 sync store/aggregation "s3://$DATA_BUCKET/bathymetry/aggregation" --exclude '*' --include '*.done' # Overview pyramid, sharded. The deep levels partition into read-closed subtrees - # (one per ancestor at the covering's seed zoom) so they fan out with no - # coordination, exactly like aggregate; the coarse global tail (a few cheap levels - # whose tiles span subtrees) is finished on the single bundle runner below. The - # level barrier stays inside each runner. A clean rebuild still spins one no-op shard. + # (one per ancestor at the covering's seed zoom), bin-packed by work into the shards + # (a shard ≈ the heaviest single subtree — sizing to ancestor *count* left one + # 77-minute straggler blocking bundle-plan next to hundreds of spin-up-only shards), + # so they fan out with no coordination, exactly like aggregate; the coarse global + # tail (a few cheap levels whose tiles span subtrees) is finished on the single + # bundle runner below. The level barrier stays inside each runner. A clean rebuild + # still spins one no-op shard. downsample: needs: [image, aggregate, plan] if: ${{ !cancelled() && needs.aggregate.result == 'success' }} @@ -419,9 +422,11 @@ jobs: # whose footprint-sized archives outgrew a runner as sources accumulated. bundle-plan # finishes the coarse pyramid tail (it spans ancestors, so it can't be a deep shard), # verifies the pyramid is whole, and emits the matrix: each entry is a comma-joined - # CHUNK of group names (the partition rides in the matrix, so jobs can't drift from the - # plan's store view). A matrix job loops its chunk pull→bundle→push→clean one group at - # a time; bundle-merge stitches the fragments. Contours are built separately (below). + # CHUNK of group names, bin-packed by pmtiles bytes so a chunk ≈ the biggest single + # group (one-group chunks spent more runner time on setup than bundling). The + # partition rides in the matrix, so jobs can't drift from the plan's store view. A + # matrix job loops its chunk pull→bundle→push→clean one group at a time; + # bundle-merge stitches the fragments. Contours are built separately (below). bundle-plan: needs: downsample if: ${{ !cancelled() && needs.downsample.result == 'success' }} @@ -538,7 +543,7 @@ jobs: --content-type application/json # ─── Contours (sharded — one global tippecanoe blows the 6 h cap at planet scale) ─ - # Size the fan-out to the FGB count (no download — just an R2 listing). + # Size the fan-out to the work (no download — just an R2 listing). contour-plan: needs: aggregate if: ${{ !cancelled() && needs.aggregate.result == 'success' }} @@ -549,7 +554,10 @@ jobs: - id: m run: | count=$(aws s3 ls "s3://$DATA_BUCKET/bathymetry/contour/" --recursive | grep -c '\.fgb$' || true) - shards=$(python3 -c "import json,sys; c=int(sys.argv[1]); n=min(${AGG_SHARDS}, max(c,1)); print(json.dumps([{'i':i,'n':n} for i in range(n)]))" "$count") + # ~200 FGBs ≈ 10 min of tippecanoe per shard (observed ~3 s/FGB planet-wide); + # sizing to the raw count spun one runner per ~1 min of work. Retune the divisor + # if shard runtimes drift past ~30 min. AGG_SHARDS stays the ceiling. + shards=$(python3 -c "import json,sys; c=int(sys.argv[1]); n=min(${AGG_SHARDS}, max((c + 199) // 200, 1)); print(json.dumps([{'i':i,'n':n} for i in range(n)]))" "$count") echo "shards=$shards" >> "$GITHUB_OUTPUT" echo "contour FGBs: $count → matrix $shards" @@ -591,7 +599,10 @@ jobs: # it to tile the coverage layer to the same depth as the contours. aws s3 cp store/contour-maxz.txt "s3://$DATA_BUCKET/bathymetry/build/${{ github.sha }}/contour-maxz.txt" - # tile-join the per-shard pmtiles into one vector.pmtiles. + # tile-join the per-shard pmtiles + soundings + drying + coverage into one vector.pmtiles, + # in ONE join: tile-join rewrites every tile of the whole archive, so the old + # merge-then-fold-per-layer shape re-paid the planet-wide join per layer (~90 min each, + # serial, as the only job left running — the last 2 h 40 of the build). contour-bundle: needs: contours if: ${{ !cancelled() && needs.contours.result == 'success' }} @@ -607,13 +618,14 @@ jobs: aws s3 cp "s3://$DATA_BUCKET/bathymetry/build/${{ github.sha }}/contour-maxz.txt" store/contour-maxz.txt aws s3 sync "s3://$DATA_BUCKET/bathymetry/soundings" store/soundings aws s3 sync "s3://$DATA_BUCKET/bathymetry/drying" store/drying - - name: Merge contours + # Soundings + drying aren't sharded (sparse); their pmtiles must exist before the + # merge so its tile-join folds them in. + - name: Bundle soundings + drying + run: | + docker run --rm -v "$PWD/store:/app/pipelines/store" "$IMAGE:${{ github.sha }}" just soundings + docker run --rm -v "$PWD/store:/app/pipelines/store" "$IMAGE:${{ github.sha }}" just drying + - name: Merge contours + fold layers run: docker run --rm -v "$PWD/store:/app/pipelines/store" "$IMAGE:${{ github.sha }}" just contour-merge - # Soundings + drying aren't sharded (sparse) — bundle them here and fold into vector.pmtiles. - - name: Fold in soundings - run: docker run --rm -v "$PWD/store:/app/pipelines/store" "$IMAGE:${{ github.sha }}" just soundings - - name: Fold in drying - run: docker run --rm -v "$PWD/store:/app/pipelines/store" "$IMAGE:${{ github.sha }}" just drying - name: Push vector.pmtiles to R2 build run: | aws s3 cp store/bundle/vector.pmtiles "s3://$DATA_BUCKET/bathymetry/build/${{ github.sha }}/vector.pmtiles" \ diff --git a/Justfile b/Justfile index 4a22712a..d3d06f36 100644 --- a/Justfile +++ b/Justfile @@ -24,14 +24,16 @@ sources: just ../sources/"$id"/ done -# Planet build: cover -> aggregate -> downsample -> bundle -> contours (BBOX="W,S,E,N" for a region). +# Planet build: cover -> aggregate -> downsample -> bundle -> vector layers (BBOX="W,S,E,N" +# for a region). Soundings + drying bundle BEFORE contours: the contours tile-join folds +# their pmtiles into vector.pmtiles in the same single pass. planet: just cover uv run python aggregation_run.py just combine - just contours just soundings just drying + just contours # Plan the covering: slice the planet into aggregation tiles (BBOX="W,S,E,N" for a region). cover: @@ -94,26 +96,28 @@ bundle-group name: bundle-merge: uv run python bundle.py merge -# Contours, whole set (local/regional). CI shards these across runners — see below. +# Contours, whole set (local/regional); the final tile-join also folds in any +# soundings/drying pmtiles already bundled. CI shards these across runners — see below. contours: uv run python contour_run.py bundle -# Soundings: bundle the per-tile points, then fold them into vector.pmtiles (one vector source). +# Soundings: bundle the per-tile points into soundings.pmtiles. Run BEFORE the contours +# bundle/merge — its single tile-join folds the layer into vector.pmtiles (a separate +# fold re-joined the whole planet archive per layer). soundings: uv run python soundings_run.py bundle - uv run python soundings_run.py fold -# Drying areas (green foreshore): bundle the per-tile polygons, then fold into vector.pmtiles. +# Drying areas (green foreshore): bundle the per-tile polygons into drying.pmtiles +# (folded into vector.pmtiles by the contours tile-join, same as soundings). drying: uv run python drying_run.py bundle - uv run python drying_run.py fold # tippecanoe this shard's local FGBs -> contours-shard-{i}.pmtiles (CI pulls only the # shard's slice + writes store/contour-maxz.txt; merged by contour-merge). contour-shard i: uv run python contour_run.py bundle-shard {{i}} -# tile-join the per-shard contour pmtiles into vector.pmtiles. +# tile-join the per-shard contour pmtiles + coverage + soundings/drying into vector.pmtiles. contour-merge: uv run python contour_run.py bundle-merge diff --git a/pipelines/bundle.py b/pipelines/bundle.py index dad4570f..57008535 100644 --- a/pipelines/bundle.py +++ b/pipelines/bundle.py @@ -226,16 +226,19 @@ def _manifest_from_fragments(frags): def groups_matrix(maxn): - """Verify the pyramid is whole, then print the CI bundle matrix: <= maxn chunks, - each a comma-joined strided slice of the group names. The partition rides IN the - matrix (not re-derived per job from a live R2 listing), so every job bundles the - exact set this full-store runner saw — the same freeze-the-plan reasoning as the - aggregate/downsample shards.""" + """Verify the pyramid is whole, then print the CI bundle matrix: <= maxn chunks + of comma-joined group names, bin-packed by each group's local pmtiles bytes so + every chunk carries about the biggest single group (one-group chunks meant 235 + runners each spending longer on spin-up than on bundling). The partition rides + IN the matrix (not re-derived per job from a live R2 listing), so every job + bundles the exact set this full-store runner saw — the same freeze-the-plan + reasoning as the aggregate/downsample shards.""" aggregation_id = utils.get_aggregation_ids()[-1] verify_complete(aggregation_id) - names = sorted(group_filepaths(aggregation_id)) - n = min(maxn, max(len(names), 1)) - print(json.dumps([{"cells": ",".join(names[i::n])} for i in range(n)])) + groups = group_filepaths(aggregation_id) + weights = {name: sum(os.path.getsize(fp) for fp in fps) for name, fps in groups.items()} + n = min(maxn, math.ceil(sum(weights.values()) / max(max(weights.values()), 1))) if weights else 1 + print(json.dumps([{"cells": ",".join(sorted(chunk))} for chunk in utils.lpt_bins(weights, n)])) def group_keys(name): diff --git a/pipelines/contour_run.py b/pipelines/contour_run.py index 904d30cd..89cfd90d 100644 --- a/pipelines/contour_run.py +++ b/pipelines/contour_run.py @@ -302,12 +302,16 @@ def _coverage_pmtiles(maxz): def _finalize_contours(contour_pmtiles, maxz): """tile-join the contour pmtiles (one local build or the CI shards) + the coverage - layer into store/bundle/vector.pmtiles. -pk keeps every feature of both layers; - coverage is dropped from the join when no footprints are present locally.""" + layer + the prebuilt soundings/drying pmtiles (when their bundles ran first) into + store/bundle/vector.pmtiles. ONE join: tile-join rewrites every tile of the whole + archive, so folding each sparse layer in afterwards re-paid the planet-wide join + per layer (~90 min each in CI). -pk keeps every feature of every layer; a layer + whose pmtiles isn't present locally is simply not joined.""" cov = _coverage_pmtiles(maxz) - inputs = list(contour_pmtiles) + ([cov] if cov else []) - subprocess.run(["tile-join", "-o", "store/bundle/vector.pmtiles", "-f", "-pk", *inputs], - check=True) + layers = [p for p in [cov, "store/bundle/soundings.pmtiles", "store/bundle/drying.pmtiles"] + if p and os.path.isfile(p)] + subprocess.run(["tile-join", "-o", "store/bundle/vector.pmtiles", "-f", "-pk", + *contour_pmtiles, *layers], check=True) return cov is not None diff --git a/pipelines/downsampling.py b/pipelines/downsampling.py index 6d8dd871..56d90fa2 100644 --- a/pipelines/downsampling.py +++ b/pipelines/downsampling.py @@ -17,6 +17,7 @@ """ import json +import math import os import shutil import sys @@ -199,12 +200,27 @@ def shard_ancestor(filepath): return ancestor_id(z, x, y) +def ancestor_weights(): + """Dirty deep work per SHARD_ROOT_Z ancestor: each -downsampling.csv at extent z + builds 4**(parent_zoom - z) parent webps of uniform cost. Striding by ancestor + *count* ignored this and put a deep hi-res subtree (77 min) next to hundreds of + couple-of-overview shards (~2 min, all runner spin-up).""" + weights = {} + for fp in work_list(): + a = shard_ancestor(fp) + if a is None: + continue + z, _, _, parent_zoom = (int(v) for v in fp.split("/")[-1] + .replace("-downsampling.csv", "").split("-")) + weights[a] = weights.get(a, 0) + 4 ** (parent_zoom - z) + return weights + + def owned_ancestors(i, n): - """The strided slice of dirty deep ancestors shard i of n owns (the same split - run() and matrix() use), so shard-keys and run agree on what a shard touches. + """The dirty deep ancestors shard i of n owns — the same weighted bin-packing + run() and matrix() use, so shard-keys and run agree on what a shard touches. Derived from the frozen work list, so every shard computes the identical split.""" - ancestors = sorted({a for fp in work_list() if (a := shard_ancestor(fp)) is not None}) - return set(ancestors[i::n]) + return set(utils.lpt_bins(ancestor_weights(), n)[i]) def shard_keys(i, n): @@ -366,7 +382,7 @@ def run(shard=None, tail=False): """Build the dirty overview pyramid. Default = everything on one machine. Across CI runners (the cut keeps the level barrier inside one machine): - ``shard=(i, n)`` — only the deep levels under the i-th strided slice of + ``shard=(i, n)`` — only the deep levels under the i-th work-weighted bin of SHARD_ROOT_Z ancestors; each shard's subtree is read-closed, so they run concurrently with no coordination and push disjoint tiles. ``tail=True`` — only the coarse levels whose archives span ancestors @@ -382,10 +398,11 @@ def run(shard=None, tail=False): def matrix(maxn): - """Print the CI deep-shard matrix JSON: <= maxn shards, >= 1, sized to the - number of distinct ancestors in the dirty deep set.""" - ancestors = {a for fp in work_list() if (a := shard_ancestor(fp)) is not None} - n = min(maxn, max(len(ancestors), 1)) + """Print the CI deep-shard matrix JSON: <= maxn shards, >= 1, with n sized so + each shard carries about the heaviest single ancestor — the wall-clock floor + anyway, since an ancestor's read-closed subtree can't split across shards.""" + weights = ancestor_weights() + n = min(maxn, math.ceil(sum(weights.values()) / max(weights.values()))) if weights else 1 print(json.dumps([{"i": i, "n": n} for i in range(n)])) diff --git a/pipelines/drying_run.py b/pipelines/drying_run.py index 48a4dc45..572e2412 100644 --- a/pipelines/drying_run.py +++ b/pipelines/drying_run.py @@ -13,7 +13,8 @@ drying mask off (DEM, land mask) -> polygonize -> clip to the unbuffered tile bbox -> 4326 -> store/drying/{stem}.fgb. Same seam contract as contours: the mask and DEM are deterministic on the buffered grid, so neighbouring tiles' halos polygonize identically and polygon edges meet at -the clip. bundle() tippecanoes them into a `drying` layer; fold() joins it into vector.pmtiles. +the clip. bundle() tippecanoes them into a `drying` layer pmtiles; the contour merge's single +tile-join folds it into vector.pmtiles (run bundle before the merge). """ import os @@ -170,20 +171,6 @@ def bundle(): print(f"drying bundle: store/bundle/drying.pmtiles (z0-{maxz}, {len(fgbs)} FGBs)") -def fold(): - """Fold drying.pmtiles into vector.pmtiles as the `drying` layer, so the Worker serves it - from the one vector source. tile-join -pk keeps every layer's features. Runs after both - bundles; no-op if either is missing.""" - cont, dry = "store/bundle/vector.pmtiles", "store/bundle/drying.pmtiles" - if not (os.path.isfile(cont) and os.path.isfile(dry)): - print("drying fold: need both vector.pmtiles and drying.pmtiles") - return - tmp = "store/bundle/vector-with-drying.pmtiles" # tile-join can't -o over an input - subprocess.run(["tile-join", "-o", tmp, "-f", "-pk", cont, dry], check=True) - os.replace(tmp, cont) - print("drying fold: folded drying layer into vector.pmtiles") - - def _check(): """Drying mask + polygonize on a synthetic DEM/mask grid: only foreshore (0<=elev<=cap, seaward of land) turns 1; land, deep water, and above-cap topo stay 0; and the mask is @@ -279,9 +266,7 @@ def axis_frac(gs): a = sys.argv[1:] if a[:1] == ["bundle"]: bundle() - elif a[:1] == ["fold"]: - fold() elif a[:1] == ["check"]: _check() else: - sys.exit("usage: drying_run.py bundle | fold | check") + sys.exit("usage: drying_run.py bundle | check") diff --git a/pipelines/soundings_run.py b/pipelines/soundings_run.py index 1d81cb7f..63d2b84c 100644 --- a/pipelines/soundings_run.py +++ b/pipelines/soundings_run.py @@ -194,20 +194,6 @@ def bundle(): print(f"soundings bundle: store/bundle/soundings.pmtiles (z0-{maxz}, {len(gj)} tiles)") -def fold(): - """Fold soundings.pmtiles into vector.pmtiles as the `soundings` layer, so the Worker - serves it from the one vector source (contours + coverage + soundings). tile-join -pk keeps - every feature of every layer. Runs after both bundles; no-op if either is missing.""" - cont, snd = "store/bundle/vector.pmtiles", "store/bundle/soundings.pmtiles" - if not (os.path.isfile(cont) and os.path.isfile(snd)): - print("soundings fold: need both vector.pmtiles and soundings.pmtiles") - return - tmp = "store/bundle/vector-with-soundings.pmtiles" # tile-join can't -o over an input - subprocess.run(["tile-join", "-o", tmp, "-f", "-pk", cont, snd], check=True) - os.replace(tmp, cont) - print("soundings fold: folded soundings layer into vector.pmtiles") - - def _check(): """Grid shoalest per cell; extent+3: 4**3 = 64 parent webps each) among 40 + # couple-of-overview ancestors (weight 1) — the planet build's shape in miniature. + heavies = [f"store/aggregation/{aid}/{rz}-{x}-0-{rz + 3}-downsampling.csv" for x in range(3)] + lights = [f"store/aggregation/{aid}/{rz}-{x}-{y}-{rz}-downsampling.csv" + for x in range(8) for y in range(1, 6)] + with open(f"store/aggregation/{aid}/{downsampling.FROZEN}", "w") as f: + f.write("".join(fp + "\n" for fp in heavies + lights)) + w = downsampling.ancestor_weights() + assert len(w) == 43 and sorted(w.values(), reverse=True)[:3] == [64, 64, 64], w + # matrix sizing: enough bins that each ~= the heaviest subtree — not one per ancestor + n = math.ceil(sum(w.values()) / max(w.values())) + assert n == 4, f"expected 4 work-sized bins for 43 ancestors, got {n}" + owned = [downsampling.owned_ancestors(i, n) for i in range(n)] + flat = [a for s in owned for a in s] + assert len(flat) == len(set(flat)) == len(w), "bins must partition the ancestors" + loads = [sum(w[a] for a in s) for s in owned] + assert max(loads) < 2 * max(w.values()), f"a bin exceeds the LPT bound: {loads}" + # the pure packer (also chunks the bundle groups): heaviest first into lightest bin + assert utils.lpt_bins({"a": 5, "b": 3, "c": 3, "d": 1}, 2) == [["a", "d"], ["b", "c"]] + print(f"weighted-shards ok — {len(w)} ancestors -> {n} bins, loads {sorted(loads, reverse=True)}") + finally: + os.chdir(cwd) + shutil.rmtree(tmp, ignore_errors=True) + + def check_stale_overview(): """An overview must rebuild when a child it averages is newer than it (or about to be self-healed), and that staleness must cascade up the pyramid. The bug: a child rebuilt by a @@ -532,6 +573,7 @@ def main(): drying_run._check() check_priority() check_shard_partition() + check_weighted_shards() check_stale_overview() check_grid_split() check_land_clamp() diff --git a/pipelines/utils.py b/pipelines/utils.py index 0f3cb396..2d2f9a3d 100644 --- a/pipelines/utils.py +++ b/pipelines/utils.py @@ -166,6 +166,20 @@ def get_dirty_aggregation_filenames(current_aggregation_id, last_aggregation_id) return dirty_filenames +def lpt_bins(weights, n): + """Deterministically bin-pack {name: weight} into n bins, heaviest item first + into the lightest bin (LPT greedy). Callers size n to ceil(total / max), so each + bin carries about the heaviest single item — the floor any partition has, since + one item can't split across shards. Guarantees a complete, disjoint partition and + (for n <= len(weights), all weights > 0) no empty bins.""" + bins, loads = [[] for _ in range(n)], [0] * n + for name in sorted(weights, key=lambda k: (-weights[k], k)): + j = min(range(n), key=lambda k: (loads[k], k)) + bins[j].append(name) + loads[j] += weights[name] + return bins + + def get_pmtiles_folder(x, y, z): if z < 7: return 'store/pmtiles' From e0ce8078af7ecb8e0c39f3cc04332e089b51d5b1 Mon Sep 17 00:00:00 2001 From: Brandon Keepers Date: Mon, 6 Jul 2026 09:04:08 -0400 Subject: [PATCH 2/3] Shard soundings + drying with the contours MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bundled whole-set inside contour-bundle, they held its tile-join back ~26 min while the rest of the build sat finished (run 28751489111: merge started 19:45, everything else done by 19:47, job ran solo to 21:05). Each contour shard now also tippecanoes its slice of soundings + drying (three invocations — the layers need different flags), so the bundling rides the existing fan-out and the merge starts as soon as the shards land. Slices stride each layer's own sorted list; no geographic alignment needed since the join unions everything per tile. Shard archives tile to the shared global maxz (store/contour-maxz.txt) like the contours always have: a slice whose own max child_z undershoots it would otherwise vanish from deeper tiles after the join. --- .github/workflows/build.yml | 70 ++++++++++++++++++++----------------- Justfile | 13 ++++--- pipelines/contour_run.py | 19 +++++----- pipelines/drying_run.py | 23 ++++++++---- pipelines/soundings_run.py | 21 +++++++---- 5 files changed, 87 insertions(+), 59 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 677ad8cc..d5f7dd61 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -542,8 +542,10 @@ jobs: aws s3 cp store/bundle/manifest.json "s3://$DATA_BUCKET/bathymetry/build/${{ github.sha }}/manifest.json" \ --content-type application/json - # ─── Contours (sharded — one global tippecanoe blows the 6 h cap at planet scale) ─ - # Size the fan-out to the work (no download — just an R2 listing). + # ── Vector layers (sharded — one global tippecanoe blows the 6 h cap at planet scale) ─ + # Size the fan-out to the contour work — the dominant layer; soundings + drying + # slices ride the same shards for ~20% more per-shard time (no download — just an + # R2 listing). contour-plan: needs: aggregate if: ${{ !cancelled() && needs.aggregate.result == 'success' }} @@ -561,7 +563,11 @@ jobs: echo "shards=$shards" >> "$GITHUB_OUTPUT" echo "contour FGBs: $count → matrix $shards" - # tippecanoe one strided FGB slice per runner → a per-shard contours pmtiles. + # tippecanoe one strided slice of every vector layer per runner → per-shard + # contours/soundings/drying pmtiles. Soundings + drying ride the contour shards + # (separately they held contour-bundle's tile-join back ~26 min while the rest of + # the build sat finished); the slices stride each layer's own sorted list — no + # geographic alignment needed, the merge unions everything per tile anyway. contours: needs: [aggregate, contour-plan] if: ${{ !cancelled() && needs.contour-plan.result == 'success' }} @@ -574,35 +580,41 @@ jobs: steps: - uses: actions/checkout@v6 - uses: ./.github/actions/ghcr-login - # Pull ONLY this shard's FGB slice (not the whole set). List the FGB keys - # (sorted = the order Python globs), derive the global maxz (so every shard - # tiles to the same depth and tile-joins cleanly), and parallel-copy the - # strided slice this shard owns. - - name: Pull this shard's FGB slice from R2 - run: | - mkdir -p store/contour store/bundle - aws s3 ls "s3://$DATA_BUCKET/bathymetry/contour/" --recursive | awk '{print $NF}' | grep '\.fgb$' | sort > /tmp/all.txt - sed 's#.*/##; s/\.fgb$//' /tmp/all.txt | awk -F- '{print $4}' | sort -n | tail -1 > store/contour-maxz.txt + # Pull ONLY this shard's slices (not the whole sets). List each layer's keys + # (sorted = the order Python globs), derive the global maxz from the contour + # list (so every shard and layer tiles to the same depth and tile-joins + # cleanly), and parallel-copy the strided slices this shard owns. + - name: Pull this shard's layer slices from R2 + run: | + mkdir -p store/contour store/soundings store/drying store/bundle + aws s3 ls "s3://$DATA_BUCKET/bathymetry/contour/" --recursive | awk '{print $NF}' | grep '\.fgb$' | sort > /tmp/all-contour.txt + sed 's#.*/##; s/\.fgb$//' /tmp/all-contour.txt | awk -F- '{print $4}' | sort -n | tail -1 > store/contour-maxz.txt + aws s3 ls "s3://$DATA_BUCKET/bathymetry/soundings/" --recursive | awk '{print $NF}' | grep '\.geojson$' | sort > /tmp/all-soundings.txt + aws s3 ls "s3://$DATA_BUCKET/bathymetry/drying/" --recursive | awk '{print $NF}' | grep '\.fgb$' | sort > /tmp/all-drying.txt # Bounded outer retry + visible errors — see the downsample pull above. - awk -v i="${{ matrix.shard.i }}" -v n="${{ matrix.shard.n }}" 'NR % n == (i + 1) % n' /tmp/all.txt \ - | xargs -P 8 -I{} sh -c 'for a in 1 2 3 4 5; do aws s3 cp "s3://$DATA_BUCKET/$1" store/contour/ --only-show-errors && exit 0; sleep "$a"; done; echo "giving up on $1 after 5 attempts" >&2; exit 1' _ {} - echo "shard ${{ matrix.shard.i }}/${{ matrix.shard.n }}: $(ls store/contour/*.fgb 2>/dev/null | wc -l) FGBs, global maxz $(cat store/contour-maxz.txt)" - - name: Contour shard ${{ matrix.shard.i }} + for spec in "/tmp/all-contour.txt store/contour" "/tmp/all-soundings.txt store/soundings" "/tmp/all-drying.txt store/drying"; do + set -- $spec + awk -v i="${{ matrix.shard.i }}" -v n="${{ matrix.shard.n }}" 'NR % n == (i + 1) % n' "$1" \ + | xargs -P 8 -I{} sh -c 'for a in 1 2 3 4 5; do aws s3 cp "s3://$DATA_BUCKET/$1" "$2/" --only-show-errors && exit 0; sleep "$a"; done; echo "giving up on $1 after 5 attempts" >&2; exit 1' _ {} "$2" + done + echo "shard ${{ matrix.shard.i }}/${{ matrix.shard.n }}: $(ls store/contour/*.fgb 2>/dev/null | wc -l) contour FGBs, $(ls store/soundings/*.geojson 2>/dev/null | wc -l) soundings, $(ls store/drying/*.fgb 2>/dev/null | wc -l) drying, global maxz $(cat store/contour-maxz.txt)" + - name: Vector shard ${{ matrix.shard.i }} run: | docker run --rm -v "$PWD/store:/app/pipelines/store" \ - "$IMAGE:${{ github.sha }}" just contour-shard ${{ matrix.shard.i }} + "$IMAGE:${{ github.sha }}" just vector-shard ${{ matrix.shard.i }} - name: Push shard pmtiles to R2 run: | aws s3 cp store/bundle "s3://$DATA_BUCKET/bathymetry/build/${{ github.sha }}/contour-shards/" \ - --recursive --exclude '*' --include 'contours-shard-*.pmtiles' --content-type application/octet-stream + --recursive --exclude '*' --include '*-shard-*.pmtiles' --content-type application/octet-stream # Global contour maxz (identical from every shard) — contour-bundle's merge needs # it to tile the coverage layer to the same depth as the contours. aws s3 cp store/contour-maxz.txt "s3://$DATA_BUCKET/bathymetry/build/${{ github.sha }}/contour-maxz.txt" - # tile-join the per-shard pmtiles + soundings + drying + coverage into one vector.pmtiles, - # in ONE join: tile-join rewrites every tile of the whole archive, so the old - # merge-then-fold-per-layer shape re-paid the planet-wide join per layer (~90 min each, - # serial, as the only job left running — the last 2 h 40 of the build). + # tile-join the per-shard pmtiles (contours + soundings + drying slices) + coverage + # into one vector.pmtiles, in ONE join: tile-join rewrites every tile of the whole + # archive, so the old merge-then-fold-per-layer shape re-paid the planet-wide join + # per layer (~90 min each, serial, as the only job left running — the last 2 h 40 + # of the build). contour-bundle: needs: contours if: ${{ !cancelled() && needs.contours.result == 'success' }} @@ -611,20 +623,12 @@ jobs: steps: - uses: actions/checkout@v6 - uses: ./.github/actions/ghcr-login - - name: Pull shard pmtiles + soundings + drying from R2 + - name: Pull shard pmtiles from R2 run: | - mkdir -p store/bundle store/soundings store/drying + mkdir -p store/bundle aws s3 sync "s3://$DATA_BUCKET/bathymetry/build/${{ github.sha }}/contour-shards" store/bundle aws s3 cp "s3://$DATA_BUCKET/bathymetry/build/${{ github.sha }}/contour-maxz.txt" store/contour-maxz.txt - aws s3 sync "s3://$DATA_BUCKET/bathymetry/soundings" store/soundings - aws s3 sync "s3://$DATA_BUCKET/bathymetry/drying" store/drying - # Soundings + drying aren't sharded (sparse); their pmtiles must exist before the - # merge so its tile-join folds them in. - - name: Bundle soundings + drying - run: | - docker run --rm -v "$PWD/store:/app/pipelines/store" "$IMAGE:${{ github.sha }}" just soundings - docker run --rm -v "$PWD/store:/app/pipelines/store" "$IMAGE:${{ github.sha }}" just drying - - name: Merge contours + fold layers + - name: Merge shards + fold coverage run: docker run --rm -v "$PWD/store:/app/pipelines/store" "$IMAGE:${{ github.sha }}" just contour-merge - name: Push vector.pmtiles to R2 build run: | diff --git a/Justfile b/Justfile index d3d06f36..cda1955d 100644 --- a/Justfile +++ b/Justfile @@ -112,12 +112,17 @@ soundings: drying: uv run python drying_run.py bundle -# tippecanoe this shard's local FGBs -> contours-shard-{i}.pmtiles (CI pulls only the -# shard's slice + writes store/contour-maxz.txt; merged by contour-merge). -contour-shard i: +# tippecanoe this shard's local slice of every vector layer -> {contours,soundings, +# drying}-shard-{i}.pmtiles (CI pulls only the shard's slices + writes +# store/contour-maxz.txt so all layers tile to one depth; merged by contour-merge). +# Three invocations, not one -L run: the layers need different tippecanoe flags +# (soundings -r1, drying --drop-densest-as-needed, contours' per-zoom filter). +vector-shard i: uv run python contour_run.py bundle-shard {{i}} + uv run python soundings_run.py bundle-shard {{i}} + uv run python drying_run.py bundle-shard {{i}} -# tile-join the per-shard contour pmtiles + coverage + soundings/drying into vector.pmtiles. +# tile-join the per-shard pmtiles (all layers) + coverage into vector.pmtiles. contour-merge: uv run python contour_run.py bundle-merge diff --git a/pipelines/contour_run.py b/pipelines/contour_run.py index 89cfd90d..08602fec 100644 --- a/pipelines/contour_run.py +++ b/pipelines/contour_run.py @@ -300,9 +300,10 @@ def _coverage_pmtiles(maxz): return out -def _finalize_contours(contour_pmtiles, maxz): - """tile-join the contour pmtiles (one local build or the CI shards) + the coverage - layer + the prebuilt soundings/drying pmtiles (when their bundles ran first) into +def _finalize_contours(archives, maxz): + """tile-join the layer archives (a local build's contour pmtiles, or the CI shards — + which carry contours, soundings, AND drying slices) + the coverage layer + the + whole-set soundings/drying pmtiles (local path, when their bundles ran first) into store/bundle/vector.pmtiles. ONE join: tile-join rewrites every tile of the whole archive, so folding each sparse layer in afterwards re-paid the planet-wide join per layer (~90 min each in CI). -pk keeps every feature of every layer; a layer @@ -311,7 +312,7 @@ def _finalize_contours(contour_pmtiles, maxz): layers = [p for p in [cov, "store/bundle/soundings.pmtiles", "store/bundle/drying.pmtiles"] if p and os.path.isfile(p)] subprocess.run(["tile-join", "-o", "store/bundle/vector.pmtiles", "-f", "-pk", - *contour_pmtiles, *layers], check=True) + *archives, *layers], check=True) return cov is not None @@ -366,10 +367,10 @@ def bundle(shard=None): def bundle_merge(): - """tile-join the per-shard contour pmtiles + the coverage layer into one - vector.pmtiles (-pk keeps every feature; the shards are disjoint FGB slices - unioned per tile).""" - shards = sorted(glob("store/bundle/contours-shard-*.pmtiles")) + """tile-join the per-shard pmtiles — contours, soundings, drying (each shard job + bundles its slice of all three) — + the coverage layer into one vector.pmtiles + (-pk keeps every feature; the shards are disjoint file slices unioned per tile).""" + shards = sorted(glob("store/bundle/*-shard-*.pmtiles")) if not shards: print("contour merge: no shard pmtiles") return @@ -377,7 +378,7 @@ def bundle_merge(): if not os.path.isfile(maxzfile): raise SystemExit("contour merge: store/contour-maxz.txt missing (the shard jobs write it)") cov = _finalize_contours(shards, int(open(maxzfile).read().strip())) - print(f"contour merge: store/bundle/vector.pmtiles ({len(shards)} shards" + print(f"contour merge: store/bundle/vector.pmtiles ({len(shards)} shard archives" f"{', + coverage layer' if cov else ''})") diff --git a/pipelines/drying_run.py b/pipelines/drying_run.py index 572e2412..e6d46aff 100644 --- a/pipelines/drying_run.py +++ b/pipelines/drying_run.py @@ -152,23 +152,30 @@ def generate(filepath): # ── bundle ─────────────────────────────────────────────────────────────────── -def bundle(): +def bundle(shard=None): """tippecanoe the per-tile drying FGBs into store/bundle/drying.pmtiles (layer `drying`). - Sparse coastal polygons (not sharded, like soundings); the orphan filter drops FGBs left - from a re-tiled covering, same as contours/soundings.""" + The orphan filter drops FGBs left from a re-tiled covering, same as contours/soundings. + With a shard index → drying-shard-{shard}.pmtiles from this shard's local slice, tiled + to the shared global maxz (store/contour-maxz.txt, like the contour shards): a slice's + own max child_z can undershoot it, and the join would then drop the layer from tiles + deeper than the slice.""" fgbs = contour_run._live_fgbs(sorted(glob("store/drying/*.fgb")), contour_run._current_stems()) if not fgbs: print("drying bundle: no drying FGBs") return - maxz = max(int(f.split("/")[-1].replace(".fgb", "").split("-")[3]) for f in fgbs) + maxzfile = "store/contour-maxz.txt" + maxz = int(open(maxzfile).read().strip()) if os.path.isfile(maxzfile) else \ + max(int(f.split("/")[-1].replace(".fgb", "").split("-")[3]) for f in fgbs) utils.create_folder("store/bundle") + out = "store/bundle/drying.pmtiles" if shard is None \ + else f"store/bundle/drying-shard-{shard}.pmtiles" subprocess.run( - ["tippecanoe", "-o", "store/bundle/drying.pmtiles", "-f", "-l", "drying", + ["tippecanoe", "-o", out, "-f", "-l", "drying", "-n", "Drying areas", "-A", utils.ATTRIBUTION, "-Z", "0", "-z", str(maxz), "-P", "-q", "--drop-densest-as-needed", "--simplification", os.environ.get("DRYING_SIMPLIFICATION", "8"), *fgbs], check=True) - print(f"drying bundle: store/bundle/drying.pmtiles (z0-{maxz}, {len(fgbs)} FGBs)") + print(f"drying bundle: {out} (z0-{maxz}, {len(fgbs)} FGBs)") def _check(): @@ -266,7 +273,9 @@ def axis_frac(gs): a = sys.argv[1:] if a[:1] == ["bundle"]: bundle() + elif a[:1] == ["bundle-shard"]: + bundle(int(a[1])) elif a[:1] == ["check"]: _check() else: - sys.exit("usage: drying_run.py bundle | check") + sys.exit("usage: drying_run.py bundle | bundle-shard | check") diff --git a/pipelines/soundings_run.py b/pipelines/soundings_run.py index 63d2b84c..4786c495 100644 --- a/pipelines/soundings_run.py +++ b/pipelines/soundings_run.py @@ -176,22 +176,29 @@ def _live(paths, stems): return [p for p in paths if p.split("/")[-1].rsplit(".", 1)[0] in stems] -def bundle(): +def bundle(shard=None): """tippecanoe the per-tile soundings into store/bundle/soundings.pmtiles (layer `soundings`). Per-feature tippecanoe.minzoom places each point from the zoom the grid decimation assigned, - so no density dropping is needed (-r1 keeps every surviving point).""" + so no density dropping is needed (-r1 keeps every surviving point). With a shard index → + soundings-shard-{shard}.pmtiles from this shard's local slice, tiled to the shared global + maxz (store/contour-maxz.txt, like the contour shards): a slice's own max child_z can + undershoot it, and the join would then drop the layer from tiles deeper than the slice.""" gj = _live(sorted(glob("store/soundings/*.geojson")), contour_run._current_stems()) if not gj: print("soundings bundle: no soundings") return - maxz = max(int(g.split("/")[-1].replace(".geojson", "").split("-")[3]) for g in gj) + maxzfile = "store/contour-maxz.txt" + maxz = int(open(maxzfile).read().strip()) if os.path.isfile(maxzfile) else \ + max(int(g.split("/")[-1].replace(".geojson", "").split("-")[3]) for g in gj) utils.create_folder("store/bundle") + out = "store/bundle/soundings.pmtiles" if shard is None \ + else f"store/bundle/soundings-shard-{shard}.pmtiles" subprocess.run( - ["tippecanoe", "-o", "store/bundle/soundings.pmtiles", "-f", "-l", "soundings", + ["tippecanoe", "-o", out, "-f", "-l", "soundings", "-n", "Bathymetric soundings", "-A", utils.ATTRIBUTION, "-Z", "0", "-z", str(maxz), "-P", "-q", "-r1", "-y", "depth_m", "-y", "depth_ft", "-y", "depth_fm", *gj], check=True) - print(f"soundings bundle: store/bundle/soundings.pmtiles (z0-{maxz}, {len(gj)} tiles)") + print(f"soundings bundle: {out} (z0-{maxz}, {len(gj)} tiles)") def _check(): @@ -249,7 +256,9 @@ def _check(): a = sys.argv[1:] if a[:1] == ["bundle"]: bundle() + elif a[:1] == ["bundle-shard"]: + bundle(int(a[1])) elif a[:1] == ["check"]: _check() else: - sys.exit("usage: soundings_run.py bundle | check") + sys.exit("usage: soundings_run.py bundle | bundle-shard | check") From 275aefaa5a8acf0d2abbc45a044c6ac974b51d92 Mon Sep 17 00:00:00 2001 From: Brandon Keepers Date: Mon, 6 Jul 2026 22:24:43 -0400 Subject: [PATCH 3/3] =?UTF-8?q?Add=20AusSeabed=20per-survey=20bathymetry?= =?UTF-8?q?=20(Australia,=200.5=20m=E2=80=93440=20m,=20z13)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 137 survey zips (~30 GB) of gridded multibeam/lidar L3 products from Geoscience Australia, selected from the Marine Data Register WFS by sources/ausseabed/harvest.py: PUBLISHED + CC-BY 4.0 + no embargo + bathymetry-only. Backscatter products are excluded — some re-bundle their sibling survey's bathy tif and would collide at unzip — as is the compilations index (SDB; gbr30/AusBathyTopo already cover the useful ones). Register DATA_URLs pointing at the non-anonymous producthouse S3 bucket map by basename to files.ausseabed.gov.au; harvest range-reads every zip's central directory to verify fetchability and reject tif basename collisions. One source, no resolution split: merge order is per-file (priority, native maxzoom), so a 1 m harbor survey outranks gbr30 while a 100 m ocean transit yields to it; max_zoom=13 only caps the sub-metre surveys. The zips ship byproducts, so source_unzip grows generic member filters: --exclude drops the *_hs.tif hillshades, --prefer _cog.tif keeps only the COG where 21 zips include a raw twin of the same grid, and members extract with a lowercase .tif extension — every downstream step globs *.tif, so a .tiff member silently vanished from the source. Fetching ~30 GB over flaky links motivated two download fixes: http_download now streams to dest.part and resumes with a Range header across retries and re-runs (curl -C style), renaming into place only on completion, and source_download skips already-complete files — so a crashed fetch resumes instead of restarting. The filelist step's own .part handling folds into that. Ragged swath grids resurrect the NONNA polygonize pathology: pixel-exact footprint masks of two transit-corridor surveys (Wallaby-Zenith: 1,305 footprint parts) took 6:51 and unioned into a 23.5 MB polygon. The mask now caps at 1024 px on its long edge via gdalwarp -r max, which dilates: coverage is over-approximated, never lost — matching the coarse per-tile footprints streamed sources already get. Same two surveys: 11 s, 1 MB. New self-checks (unzip filters, download resume against a local Range-aware server, polygonize dilation) wired into just test-sources. --- Justfile | 3 + pipelines/source_download.py | 10 ++ pipelines/source_download_filelist.py | 8 +- pipelines/source_polygonize.py | 70 ++++++++- pipelines/source_unzip.py | 59 ++++++-- pipelines/test_source_stage.py | 70 +++++++++ pipelines/utils.py | 32 ++++- sources/ausseabed/Justfile | 19 +++ sources/ausseabed/file_list.txt | 141 ++++++++++++++++++ sources/ausseabed/harvest.py | 197 ++++++++++++++++++++++++++ sources/ausseabed/metadata.json | 9 ++ 11 files changed, 597 insertions(+), 21 deletions(-) create mode 100644 sources/ausseabed/Justfile create mode 100644 sources/ausseabed/file_list.txt create mode 100644 sources/ausseabed/harvest.py create mode 100644 sources/ausseabed/metadata.json diff --git a/Justfile b/Justfile index cda1955d..0dd08b0a 100644 --- a/Justfile +++ b/Justfile @@ -176,6 +176,9 @@ preview-local bbox="-74.30,40.40,-73.75,40.80": (preview bbox "local") test-sources: uv run python test_source_stage.py uv run python source_register_remote_geopkg.py --check + uv run python source_unzip.py --check + uv run python source_download.py --check + uv run python source_polygonize.py --check test-engine: uv run python test_engine.py uv run python aggregation_reproject.py --check diff --git a/pipelines/source_download.py b/pipelines/source_download.py index 771ead78..fc364e98 100644 --- a/pipelines/source_download.py +++ b/pipelines/source_download.py @@ -48,11 +48,21 @@ def main(): sys.exit(f"no URLs in {config.SOURCES_DIR}/{source}/file_list.txt") os.makedirs(f"store/source/{source}", exist_ok=True) print(f"downloading {source}: {len(urls)} url(s)") + skipped = 0 for i, url in enumerate(urls): dest = f"store/source/{source}/{source}_{i}.{ext_for(url)}" + # A finished file is skipped, so a re-run resumes instead of re-pulling + # everything (http_download is atomic — dest only exists complete; a zip + # sniffed by fix_archive_ext lands under the .zip name). No checksum — + # rm the dir to force a clean refetch. + if os.path.exists(dest) or os.path.exists(dest.rsplit(".", 1)[0] + ".zip"): + skipped += 1 + continue print(f" [{i}] {url} -> {dest}") utils.http_download(url, dest) fix_archive_ext(dest) + if skipped: + print(f" skipped {skipped} already-downloaded file(s)") def _check(): diff --git a/pipelines/source_download_filelist.py b/pipelines/source_download_filelist.py index 5caebe53..d3ad6528 100644 --- a/pipelines/source_download_filelist.py +++ b/pipelines/source_download_filelist.py @@ -47,10 +47,10 @@ def main(): if os.path.exists(dest): continue print(f" [{i}/{len(urls)}] {url} -> {dest}") - # .part + rename so a crash mid-stream never leaves a file the skip above - # would treat as complete. No checksum — rm the dir to force a clean refetch. - utils.http_download(url, dest + ".part") - os.replace(dest + ".part", dest) + # http_download is atomic (.part + rename) and resumes a crashed fetch, so + # the skip above only ever sees complete files. No checksum — rm the dir to + # force a clean refetch. + utils.http_download(url, dest) if __name__ == "__main__": diff --git a/pipelines/source_polygonize.py b/pipelines/source_polygonize.py index 91592247..9d7c095b 100644 --- a/pipelines/source_polygonize.py +++ b/pipelines/source_polygonize.py @@ -9,25 +9,51 @@ (filenames are sanitized upstream), not untrusted input. """ +import math import sys import os from multiprocessing import Pool import shutil +import rasterio + import utils SILENT = True +# Long-edge ceiling for the mask a file is polygonized from. Footprints only steer +# the tile covering and the coverage layer — both far coarser than native pixels — +# but a pixel-exact mask of a ragged swath grid (an AusSeabed transit corridor: +# thousands of nodata holes) takes minutes per file and unions into a ~20 MB +# polygon. Shrinking with `-r max` DILATES: any block with one valid pixel stays +# covered, so the footprint over-approximates and coverage is never lost. The +# ceiling costs fuzz of extent/1024 (~hundreds of m on an ocean corridor, a few +# tile-pixels on a harbor survey); a per-source knob is the upgrade if a source +# ever needs exact edges. +MASK_MAX_PX = 1024 + def polygonize_tif(source, filename): + src = f"store/source/{source}/{filename}" mask = f"store/polygon/{source}/{filename}" + with rasterio.open(src) as r: + factor = math.ceil(max(r.width, r.height) / MASK_MAX_PX) + size = (max(1, r.width // factor), max(1, r.height // factor)) + if factor > 1: + small = mask + ".small.tif" + utils.run_command( + f'GDAL_CACHEMAX=1024 gdalwarp -q -overwrite -ts {size[0]} {size[1]} ' + f'-r max {src} {small}', silent=SILENT) + src = small utils.run_command( - f'GDAL_CACHEMAX=1024 gdal_calc.py -A store/source/{source}/{filename} ' + f'GDAL_CACHEMAX=1024 gdal_calc.py -A {src} ' f'--outfile={mask} --calc="A*0+1" --type=Byte --overwrite', silent=SILENT) utils.run_command( f'GDAL_CACHEMAX=1024 gdal_polygonize.py {mask} -b 1 -f "GPKG" ' f'store/polygon/{source}/{filename}.gpkg -overwrite', silent=SILENT) os.remove(mask) + if factor > 1: + os.remove(src) def get_filenames(source): @@ -78,5 +104,45 @@ def main(): shutil.rmtree(f"store/polygon/{source}") +def _check(): + """A sparse raster wide enough to trigger the mask downsample keeps every + valid speck in its footprint (dilation, not erosion).""" + import json + import subprocess + import tempfile + import numpy as np + from rasterio.transform import from_origin + from shapely.geometry import shape, Point + + tmp = tempfile.mkdtemp() + cwd = os.getcwd() + os.chdir(tmp) + try: + os.makedirs("store/source/_synth") + arr = np.full((3000, 3000), -9999.0, dtype="float32") + arr[10:13, 10:13] = -5.0 # speck near one corner + arr[2900:2903, 2900:2903] = -7.0 # speck near the other + with rasterio.open("store/source/_synth/_synth_0.tif", "w", driver="GTiff", + height=3000, width=3000, count=1, dtype="float32", + nodata=-9999.0, crs="EPSG:4326", + transform=from_origin(0.0, 3.0, 0.001, 0.001)) as d: + d.write(arr, 1) + utils.create_folder("store/polygon/_synth/") + polygonize_tif("_synth", "_synth_0.tif") + geo = subprocess.run( + ["ogr2ogr", "-f", "GeoJSON", "/vsistdout/", "store/polygon/_synth/_synth_0.tif.gpkg"], + capture_output=True, check=True).stdout + polys = [shape(f["geometry"]) for f in json.loads(geo)["features"]] + for x, y in [(0.0115, 2.9885), (2.9015, 0.0985)]: # speck centers in map coords + assert any(p.contains(Point(x, y)) for p in polys), (x, y) + print("source_polygonize.py self-check ok") + finally: + os.chdir(cwd) + shutil.rmtree(tmp, ignore_errors=True) + + if __name__ == "__main__": - main() + if sys.argv[1:2] == ["--check"]: + _check() + else: + main() diff --git a/pipelines/source_unzip.py b/pipelines/source_unzip.py index 95cbad39..2d0f53e0 100644 --- a/pipelines/source_unzip.py +++ b/pipelines/source_unzip.py @@ -3,29 +3,70 @@ For sources fetched as a zip (e.g. the GEBCO global grid). Flattens the archive's *.tif/*.tiff members into store/source// and removes the zip. No ±85° clamp needed — the aggregation warp to EPSG:3857 clips the poles. + +Member filters (case-insensitive substring, for zips that bundle byproducts): + --exclude S skip members containing S (e.g. ``_hs.tif`` hillshades) + --prefer S if any member contains S, extract only those — for zips that ship + the same grid twice (AusSeabed: a raw tif beside its ``_cog`` twin) """ +import argparse import os -import sys import zipfile from glob import glob +def select_members(names, exclude=None, prefer=None): + tifs = [n for n in names if n.lower().endswith((".tif", ".tiff"))] + if exclude: + tifs = [n for n in tifs if exclude.lower() not in n.lower()] + if prefer: + preferred = [n for n in tifs if prefer.lower() in n.lower()] + if preferred: + tifs = preferred + return tifs + + +def dest_name(member): + """Flattened on-disk basename with a lowercase ``.tif`` extension — every + downstream step globs ``*.tif``, so a ``.tiff``/``.TIF`` member would + silently vanish from the source.""" + return os.path.basename(member).rsplit(".", 1)[0] + ".tif" + + def main(): - if len(sys.argv) != 2: - sys.exit("usage: source_unzip.py ") - source = sys.argv[1] - zips = sorted(glob(f"store/source/{source}/*.zip")) - print(f"unzip {source}: {len(zips)} archive(s)") + p = argparse.ArgumentParser(description="Flatten zip archives' tif members into the source dir.") + p.add_argument("source") + p.add_argument("--exclude", help="skip members containing this substring") + p.add_argument("--prefer", help="if any member contains this substring, extract only those") + a = p.parse_args() + zips = sorted(glob(f"store/source/{a.source}/*.zip")) + print(f"unzip {a.source}: {len(zips)} archive(s)") for zpath in zips: with zipfile.ZipFile(zpath) as z: - members = [n for n in z.namelist() if n.lower().endswith((".tif", ".tiff"))] + members = select_members(z.namelist(), a.exclude, a.prefer) print(f" {zpath}: {len(members)} tif(s)") for name in members: - with open(f"store/source/{source}/{os.path.basename(name)}", "wb") as f: + with open(f"store/source/{a.source}/{dest_name(name)}", "wb") as f: f.write(z.read(name)) os.remove(zpath) +def _check(): + names = ["d/a_cog.tif", "d/a_raw.tiff", "d/a_hs.tiff", "meta/x.txt"] + assert select_members(names) == ["d/a_cog.tif", "d/a_raw.tiff", "d/a_hs.tiff"] + assert select_members(names, exclude="_hs.tif") == ["d/a_cog.tif", "d/a_raw.tiff"] + assert select_members(names, exclude="_hs.tif", prefer="_cog.tif") == ["d/a_cog.tif"] + # prefer is a no-op when nothing matches — a cog-less zip keeps its members + assert select_members(["d/b.tif"], exclude="_hs.tif", prefer="_cog.tif") == ["d/b.tif"] + assert dest_name("d/a_cog.tiff") == "a_cog.tif" and dest_name("d/a.TIFF") == "a.tif" + assert dest_name("d/a_cog.tif") == "a_cog.tif" + print("source_unzip.py self-check ok") + + if __name__ == "__main__": - main() + import sys + if sys.argv[1:2] == ["--check"]: + _check() + else: + main() diff --git a/pipelines/test_source_stage.py b/pipelines/test_source_stage.py index b51fc8e8..db219f6f 100644 --- a/pipelines/test_source_stage.py +++ b/pipelines/test_source_stage.py @@ -47,8 +47,78 @@ def check_remote_parsers(): print("remote parsers ok") +def check_http_download(): + """utils.http_download resume semantics against a local Range-aware server: + fresh fetch, resume of a seeded .part (206 append), a server that ignores + Range (200 rewrite), a mid-stream cut (short read → retried with resume), + and a stale .part at EOF (416 → clean restart).""" + import http.server + import threading + import utils + + payload = bytes(range(256)) * 200 # 51_200 bytes, position-dependent content + state = {"cut_next": False, "ignore_range": False} + + class H(http.server.BaseHTTPRequestHandler): + def log_message(self, *a): + pass + + def do_GET(self): + start = 0 + rng = self.headers.get("Range") + if rng and not state["ignore_range"]: + start = int(rng.split("=")[1].rstrip("-")) + if start >= len(payload): + self.send_response(416) + self.end_headers() + return + self.send_response(206) + self.send_header("Content-Range", + f"bytes {start}-{len(payload) - 1}/{len(payload)}") + else: + self.send_response(200) + body = payload[start:] + self.send_header("Content-Length", str(len(body))) + self.end_headers() + if state["cut_next"]: + state["cut_next"] = False + self.wfile.write(body[:1000]) # lie, then hang up mid-stream + self.connection.close() + return + self.wfile.write(body) + + srv = http.server.ThreadingHTTPServer(("127.0.0.1", 0), H) + threading.Thread(target=srv.serve_forever, daemon=True).start() + url = f"http://127.0.0.1:{srv.server_port}/f" + tmp = tempfile.mkdtemp() + try: + def fetch(name, seed=None): + dest = f"{tmp}/{name}" + if seed is not None: + with open(dest + ".part", "wb") as f: + f.write(seed) + utils.http_download(url, dest, chunk=4096, retries=3) + with open(dest, "rb") as f: + assert f.read() == payload, name + assert not os.path.exists(dest + ".part"), name + + fetch("fresh") + fetch("resumed", seed=payload[:10_000]) # 206 appends the tail + state["ignore_range"] = True + fetch("range_ignored", seed=payload[:10_000]) # 200 rewrites from scratch + state["ignore_range"] = False + state["cut_next"] = True + fetch("cut_midstream") # short read → retry resumes + fetch("stale_part", seed=payload) # 416 → clean restart + print("http_download resume ok") + finally: + srv.shutdown() + shutil.rmtree(tmp, ignore_errors=True) + + def main(): check_remote_parsers() + check_http_download() tmp = tempfile.mkdtemp() try: sid = "_synth" diff --git a/pipelines/utils.py b/pipelines/utils.py index 2d2f9a3d..4db9ef0e 100644 --- a/pipelines/utils.py +++ b/pipelines/utils.py @@ -57,17 +57,37 @@ def create_folder(path): def http_download(url, dest, chunk=1 << 20, retries=5): '''Stream a URL to dest with requests (handles query-string URLs; no shell). - Retries with backoff on transient network errors — the public data servers - (EMODnet, SDFE, …) reset connections under load.''' + Downloads into dest.part and resumes it with a Range header (like curl -C -) + across retries and re-runs, renaming into place only when complete — so a + multi-GB fetch that dies at 90% doesn't restart from byte 0, and dest never + exists half-written. Retries with backoff on transient network errors — the + public data servers (EMODnet, SDFE, …) reset connections under load.''' import time import requests + part = dest + '.part' for attempt in range(retries): try: - with requests.get(url, stream=True, timeout=120) as r: + have = os.path.getsize(part) if os.path.exists(part) else 0 + headers = {'Range': f'bytes={have}-'} if have else {} + with requests.get(url, stream=True, timeout=120, headers=headers) as r: + if r.status_code == 416: # .part at/past EOF (crash before rename): restart clean + os.remove(part) + raise requests.exceptions.RequestException("stale .part (416)") r.raise_for_status() - with open(dest, 'wb') as f: - for part in r.iter_content(chunk): - f.write(part) + # 206 appends the tail; anything else means the server ignored the + # Range (or none was sent) and is streaming the whole file + resumed = r.status_code == 206 + if resumed: + total = int(r.headers['Content-Range'].rsplit('/', 1)[-1]) + else: + total = int(r.headers.get('Content-Length') or 0) + with open(part, 'ab' if resumed else 'wb') as f: + for piece in r.iter_content(chunk): + f.write(piece) + if total and os.path.getsize(part) < total: # connection died without an error + raise requests.exceptions.RequestException( + f"short read: {os.path.getsize(part)}/{total} bytes") + os.replace(part, dest) return except requests.exceptions.RequestException as e: if attempt == retries - 1: diff --git a/sources/ausseabed/Justfile b/sources/ausseabed/Justfile new file mode 100644 index 00000000..7cea95af --- /dev/null +++ b/sources/ausseabed/Justfile @@ -0,0 +1,19 @@ +# AusSeabed per-survey L3 bathymetry grids (Geoscience Australia) — CC-BY 4.0. +# Run from pipelines/: just ../sources/ausseabed/ +# ~137 survey zips (~30 GB) from files.ausseabed.gov.au (anonymous CloudFront/S3), selected +# from the Marine Data Register WFS by harvest.py (PUBLISHED + CC-BY + no embargo, SDB +# excluded). Each zip holds the grid as Float32 COG(s) — negative-down elevation with an +# embedded per-survey UTM (older: 4326) CRS → mixed_crs, normalize without --crs. Zips also +# bundle *_hs.tif hillshades and (21 of them) a raw twin of each _cog grid — the unzip +# member filters keep only the real data. Vertical datum is ~MSL (filename token; two LAT +# surveys read shallower — the conservative direction). Survey resolutions span +# 0.5 m–440 m; per-file native maxzoom from bounds.csv sets merge precedence, so the +# metadata max_zoom=13 only caps the sub-metre surveys. Re-run harvest.py for new surveys. +[no-cd] +default: + uv run python source_download.py ausseabed + uv run python source_unzip.py ausseabed --exclude _hs.tif --prefer _cog.tif + uv run python source_normalize.py ausseabed + uv run python source_bounds.py ausseabed + uv run python source_polygonize.py ausseabed 8 + uv run python source_create_tarball.py ausseabed diff --git a/sources/ausseabed/file_list.txt b/sources/ausseabed/file_list.txt new file mode 100644 index 00000000..6744a233 --- /dev/null +++ b/sources/ausseabed/file_list.txt @@ -0,0 +1,141 @@ +# AusSeabed per-survey L3 bathymetry (Geoscience Australia) — CC-BY 4.0. +# 137 survey zips (407 data tifs) on files.ausseabed.gov.au, selected from the +# Marine Data Register WFS: PUBLISHED + CC-BY + no embargo + GA-hosted, SDB excluded. +# Generated by harvest.py — re-run it when GA publishes new surveys. +https://files.ausseabed.gov.au/survey/19950001S-FremantleSoutheastIndianRidgeHobartBathymetry-8-210m-1995.zip +https://files.ausseabed.gov.au/survey/19960003S-FremantleSoutheastIndianRidgetoPortHedland-BMRG06MV.zip +https://files.ausseabed.gov.au/survey/20130019S-HelensRock.zip +https://files.ausseabed.gov.au/survey/20140011S-MerriMarineSanctuary.zip +https://files.ausseabed.gov.au/survey/20150012S_HendersonPerthCanyontoHenderson_FK150301.zip +https://files.ausseabed.gov.au/survey/20150035S-PortFairyWaveEnergy.zip +https://files.ausseabed.gov.au/survey/20180002S_PopesEye.zip +https://files.ausseabed.gov.au/survey/20180003S_PortseaHolePortPhillipBay.zip +https://files.ausseabed.gov.au/survey/20210005S-PortFairyWaveEnergySite.zip +https://files.ausseabed.gov.au/survey/20210010S_PortPhillipBayTieLinesSandringham-StLeonards.zip +https://files.ausseabed.gov.au/survey/20220124S-SouthernKangarooIslandMappingexpedition-SKI0001.zip +https://files.ausseabed.gov.au/survey/20240001S-ApolloMarinePark-ShallowWaters-SSCN-SEGMENT-2-C.zip +https://files.ausseabed.gov.au/survey/20240003S-DampierMarineParkSouth-SI-1037.zip +https://files.ausseabed.gov.au/survey/20240007S-CapeFourcroyNorth-West-SI-1049.zip +https://files.ausseabed.gov.au/survey/AbbotPointToHydrographersPassage-HIPP-SI-1007-Bathymetry-30m-2020.zip +https://files.ausseabed.gov.au/survey/ApolloMarinePark-Bathymetry-2m-2020.zip +https://files.ausseabed.gov.au/survey/ApproachesToDarwinBeagle-Gulf-HIPP-SI-1002-Bathymetry-30m-2020.zip +https://files.ausseabed.gov.au/survey/ApproachesToMoretonBay-HIPP-SI-1021-Bathymetry-30m-2023.zip +https://files.ausseabed.gov.au/survey/ApproachesToNewcastle-NSW-HIPP-SI-1001-Bathymetry-30m-2020.zip +https://files.ausseabed.gov.au/survey/ArafuraMarinePark-Bathymetry-6m-2021.zip +https://files.ausseabed.gov.au/survey/ArafuraSea-Bathymetry-5m-8m-2005.zip +https://files.ausseabed.gov.au/survey/AshmoreReefMarinePark-Bathymetry-16m-2021.zip +https://files.ausseabed.gov.au/survey/Backstairs-Passage-SA-HIPP-SI-1012-Bathymetry-30m-2021.zip +https://files.ausseabed.gov.au/survey/BanksStrait-Bathymetry-4m-2018.zip +https://files.ausseabed.gov.au/survey/BanksStrait-TAS-HIPP-SI-1020-Bathymetry-30m-2021.zip +https://files.ausseabed.gov.au/survey/BanksStraitToCapeBarrenTAS-HIPP-SI-1024-Bathymetry-30m-2021.zip +https://files.ausseabed.gov.au/survey/BassPyramidtoWrightBlock-HIPP-SI-1030-Bathymetry-30m-2023.zip +https://files.ausseabed.gov.au/survey/BeagleMarinePark-Bathymetry-1m-2018.zip +https://files.ausseabed.gov.au/survey/BeagleMarinePark_ShallowWaters_Bathymetry_5m_2024_download.zip +https://files.ausseabed.gov.au/survey/BoobyIslandtoDArcoleIslandsBonaparteArchipelago-HIPP-SI-1026-Bathymetry-30m-2022.zip +https://files.ausseabed.gov.au/survey/BremerLeeuwinAndPerthCanyons-Bathymetry-2m-64m-2020.zip +https://files.ausseabed.gov.au/survey/BrowseBasinLevequeShelf-Bathymety-3m-2013.zip +https://files.ausseabed.gov.au/survey/BunurongMarineNationalPark-Bathymetry-2m-2017.zip +https://files.ausseabed.gov.au/survey/BynoeHarbour-Bathymetry-2m-2016.zip +https://files.ausseabed.gov.au/survey/Camden-Sound-North-West-WA-HIPP-SI-1015-Bathymetry-30m-2022.zip +https://files.ausseabed.gov.au/survey/CapeBarrenToBabelIslandTAS-HIPP-SI-1035-Bathymetry-30m-2021.zip +https://files.ausseabed.gov.au/survey/CapeFourcroyWest-HIPP-SI-1043-Bathymetry-30m-2023.zip +https://files.ausseabed.gov.au/survey/CapeLeeuwinWA-HIPP-SI-1031-Bathymetry-30m-2022.zip +https://files.ausseabed.gov.au/survey/CapePasleyToPollockReef-Bathymetry-HIPP-SI-1054-30m-2025.zip +https://files.ausseabed.gov.au/survey/CapeRangeCanyon-Bathymetry-16m-64m-2020.zip +https://files.ausseabed.gov.au/survey/Carnarvon-Shelf-Bathymetry-3m-2008.zip +https://files.ausseabed.gov.au/survey/CarpentariaReef-Bathymetry-4m-7m-2005.zip +https://files.ausseabed.gov.au/survey/CaseyStationAntarctica-Bathymetry-1m-2014.zip +https://files.ausseabed.gov.au/survey/CaswellBasin-Bathymetry-1m-7m-2015.zip +https://files.ausseabed.gov.au/survey/Clarence-Strait-to-Dundas-Strait-HIPP-SI-1027-Bathymetry-2023-30m.zip +https://files.ausseabed.gov.au/survey/CoralSeaCanyonsAndReef-Bathymetry-16m-64m-2020.zip +https://files.ausseabed.gov.au/survey/CrowdyHead+Offshore-NSW-Bathymetry-5m-2021.zip +https://files.ausseabed.gov.au/survey/DarwinHarbour-Bathymetry-1m-2010.zip +https://files.ausseabed.gov.au/survey/DarwinHarbour-Bathymetry-1m-2015.zip +https://files.ausseabed.gov.au/survey/DavisHarbourAntarctica-Bathymetry-2m-2010.zip +https://files.ausseabed.gov.au/survey/DavisHarbourAntarcticaBathymetry-2m-2017.zip +https://files.ausseabed.gov.au/survey/Eastern-Recherche-Marine-Park-Bathymetry-100m-2024.zip +https://files.ausseabed.gov.au/survey/EclipseShoalstoLombadinaPointKimberleyRegion-HIPP-SI-1025-Bathymetry-30m-2022.zip +https://files.ausseabed.gov.au/survey/ElizabethMiddletonReef-Bathymetry-5m-2020.zip +https://files.ausseabed.gov.au/survey/ExmouthHoutmanSub-basin-Bathymetry-15m-25m-2014.zip +https://files.ausseabed.gov.au/survey/FaustCapelBasin-Bathymetry-20m-100m-2007.zip +https://files.ausseabed.gov.au/survey/FlindersCommonwealthMarineReserve-Bathymetry-2m-2012.zip +https://files.ausseabed.gov.au/survey/FlindersIslandNE-TAS-HIPP-SI-1036-Bathymetry-30m-2021.zip +https://files.ausseabed.gov.au/survey/ForsterCapeHawkeToBlackHead-NSW-Bathymetry-5m-2019.zip +https://files.ausseabed.gov.au/survey/ForsterPacificPalmsCapeHawke-NSW-Bathymetry-5m-2019.zip +https://files.ausseabed.gov.au/survey/FremantleHobart-SOJN05MV-Bathymetry-128m-1997.zip +https://files.ausseabed.gov.au/survey/FurneauxGroupBassStrait-Bathymetry-HIPP-SI-1032-30m-2022.zip +https://files.ausseabed.gov.au/survey/GippslandBasin-Bathymetry-1m-2015.zip +https://files.ausseabed.gov.au/survey/Great-Australian-Bight-Marine-Park-Bathymetry-100m-2024.zip +https://files.ausseabed.gov.au/survey/GreatNorthEastChannelSouthWest-TorresStraitQLD-HIPP-SI-1018-Bathymetry-30m-2022.zip +https://files.ausseabed.gov.au/survey/GreatNorthEastChannelTorresStrait-HIPP-SI-1006-Bathymetry-30m-2021.zip +https://files.ausseabed.gov.au/survey/Gulf-St-Vincent-North-SA-HIPP-SI-1008-Bathymetry-30m-2020.zip +https://files.ausseabed.gov.au/survey/HMAS-Canberra-Bathymetry-0.5m-2020.zip +https://files.ausseabed.gov.au/survey/HayPointtoHydrographersPassageQLD-HIPP-SI-1029-Bathymetry-30m-2022.zip +https://files.ausseabed.gov.au/survey/HowardChannel-HIPP-SI-1046-Bathymetry-30m-2023.zip +https://files.ausseabed.gov.au/survey/HunterMarineParkBathymetry-5m-2019.zip +https://files.ausseabed.gov.au/survey/HunterMarineParkWorimiOuterGibber-Bathymetry-5m-2020.zip +https://files.ausseabed.gov.au/survey/JervisBay-Bathymetry-2m-2008.zip +https://files.ausseabed.gov.au/survey/KangarooIslandSouth-East-HIPP-SI-1023-Bathymetry-30m-2023.zip +https://files.ausseabed.gov.au/survey/KennAndChesterfieldPlateaux-Bathymetry-64m-2021.zip +https://files.ausseabed.gov.au/survey/KingIslandNorthBassStraitVIC-HIPP-SI-1013-Bathymetry-30m-2021.zip +https://files.ausseabed.gov.au/survey/LacepedeChannelWA-HIPP-SI-1014-Bathymetry-30m-2021.zip +https://files.ausseabed.gov.au/survey/LordHoweIsland-Bathymetry-10m-50m-2008.zip +https://files.ausseabed.gov.au/survey/LordHoweRise-Bathymetry-50m-80m-2016.zip +https://files.ausseabed.gov.au/survey/LordHoweRise-Bathymetry-70m-90m-2017.zip +https://files.ausseabed.gov.au/survey/LordMayorShoaltoPittShoals-HIPP-SI-1042-Bathymetry-30m-2023.zip +https://files.ausseabed.gov.au/survey/MacquarieRidge-Bathymetry-128m-1994.zip +https://files.ausseabed.gov.au/survey/MacreadieSeagrass-Bathymetry-Depth-20cm-2015.zip +https://files.ausseabed.gov.au/survey/MavisReef-East-Bonaparte-Archipelago-HIPP-SI-1011-Bathymetry-30m-2020.zip +https://files.ausseabed.gov.au/survey/MentelleBasin-Bathymetry-50m-2005.zip +https://files.ausseabed.gov.au/survey/Mudjimba-Island-Bathymetry-0.5m-2024.zip +https://files.ausseabed.gov.au/survey/Murray-Marine-Park-Bathymetry-100m-2024.zip +https://files.ausseabed.gov.au/survey/NSWContinentalShelf-Bathymetry-50m-2006.zip +https://files.ausseabed.gov.au/survey/NelsonMarinePark-Bathymetry-100m-2024.zip +https://files.ausseabed.gov.au/survey/NorfolkIslandNearshoreAndCoastalHabitatMapping-AU420-Bathymetry-1m-2021.zip +https://files.ausseabed.gov.au/survey/NorthEastBeagleGulfAndClarenceStraitBeagleGulf-NT-HIPP-SI-1016-Bathymetry-30m-2021.zip +https://files.ausseabed.gov.au/survey/NorthPerthBasin-Bathymetry-15m-32m-2011.zip +https://files.ausseabed.gov.au/survey/NorthWollongong-BellambiPointToStanwellPark-NSW-Bathymetry-5m-2017.zip +https://files.ausseabed.gov.au/survey/NortheastTasmania-Bathymetry-2m-2011.zip +https://files.ausseabed.gov.au/survey/NorthernApproacheToBroome-HIPP-SI-1010-Bathymetry-30m-2020.zip +https://files.ausseabed.gov.au/survey/NorthernGreatBarrierReef-Bathymetry-4m-32m-2020.zip +https://files.ausseabed.gov.au/survey/OceanicShoalsCommonwealthMarineReserve-Bathymetry-2m-2012.zip +https://files.ausseabed.gov.au/survey/OffshoreBrisbaneTopasSeaTrials-5m-40m-2004.zip +https://files.ausseabed.gov.au/survey/PerthCanyon-KN145L4-Bathymetry-128-210m-1994.zip +https://files.ausseabed.gov.au/survey/PerthCanyonEM300Trials-Bathymetry-10m-40m-2003.zip +https://files.ausseabed.gov.au/survey/PeterboroughToPortFairy-2m-2018.zip +https://files.ausseabed.gov.au/survey/PetrelSub-Basin-RV-Solander-Bathymetry-2m-2012.zip +https://files.ausseabed.gov.au/survey/PetrelSubBasin-MV-Duke-Bathymetry-2m-2012.zip +https://files.ausseabed.gov.au/survey/PortFairyToPortland-2m-2020.zip +https://files.ausseabed.gov.au/survey/PortPhillipBayDriftAlgae-Bathymetry-2m-2018.zip +https://files.ausseabed.gov.au/survey/QueenslandPlateau-South-West-QLD-HIPP-SI-1047.zip +https://files.ausseabed.gov.au/survey/RaineIsland-Bathymetry-2m-64m-2021.zip +https://files.ausseabed.gov.au/survey/RechercheArchipelago-2m-2005.zip +https://files.ausseabed.gov.au/survey/RechercheArchipelago-Bathymetry-2m-2003.zip +https://files.ausseabed.gov.au/survey/RefugeCove-Bathymetry-1m-2013.zip +https://files.ausseabed.gov.au/survey/Shellharbour-Bathymetry-2m-5m-2017.zip +https://files.ausseabed.gov.au/survey/SolitaryIslandsGumbaynggirrYaeglMP-Bathymetry-5m-2022.zip +https://files.ausseabed.gov.au/survey/SouthTasmanRise-Bathymetry-100m-1994.zip +https://files.ausseabed.gov.au/survey/SouthWestCornerAndPerthCanyonMarinePark-Bathymetry-64m-128m-2023.zip +https://files.ausseabed.gov.au/survey/SouthWestCornerMarinePark-Bathymetry-100m-2024.zip +https://files.ausseabed.gov.au/survey/SouthWestCornerMarinePark-Bathymetry-5m-2021.zip +https://files.ausseabed.gov.au/survey/SoutheastTasmania-Bathymetry-1m-1.6m-2009.zip +https://files.ausseabed.gov.au/survey/SoutheastTasmaniaandSouthernMacquarieRidge-AUSTREA2-Bathymetry-120-440m-2000.zip +https://files.ausseabed.gov.au/survey/SouthernAustralia-Austrea1-Bathymetry-100m-1999.zip +https://files.ausseabed.gov.au/survey/SouthernGreatBarrierReefShelf-Bathymetry-2m-64m-2020zip +https://files.ausseabed.gov.au/survey/SydneyAlbanyTransit-Bathymetry-16m-64m-2020.zip +https://files.ausseabed.gov.au/survey/TasmanAndCoralSeasBathymetry-64m-2021.zip +https://files.ausseabed.gov.au/survey/TasmaniaEastCoast-Bathymetry-0.5m-2021.zip +https://files.ausseabed.gov.au/survey/TorresStraitUnderKeelClearanceTorresStraitQLD-HIPP-SI-1005-Bathymetry-30m-2021.zip +https://files.ausseabed.gov.au/survey/VanDiemenGulf-NorthWest-InshoreRoute-NT-HIPP-SI-1003-Bathymetry-30m-2021.zip +https://files.ausseabed.gov.au/survey/VarzinPassageToMerkaraShoal-HIPP-SI-1017-Bathymetry-30m-2023.zip +https://files.ausseabed.gov.au/survey/VernonIsland-Bathymetry-4m-2019.zip +https://files.ausseabed.gov.au/survey/VisioningTheCoralSea-Bathymetry-16m-64m-2020.zip +https://files.ausseabed.gov.au/survey/VlamingSub-Basin-Bathymetry-2m-2012.zip +https://files.ausseabed.gov.au/survey/WallabyZenithFractureZone-Bathymetry-64m-128m-2021.zip +https://files.ausseabed.gov.au/survey/WesternApproachesToBroome-Bathymetry-HIPP-SI-1048-30m-2024.zip +https://files.ausseabed.gov.au/survey/WesternApproachesToTorresStraitQLD-HIPP-SI-1004-Bathymetry-30m-2020.zip +https://files.ausseabed.gov.au/survey/WesternAustralianMargins-Bathymetry-100m-2008.zip +https://files.ausseabed.gov.au/survey/WesternEyreMarinePark-Bathymetry-100m-2024.zip +https://files.ausseabed.gov.au/survey/WilsonsPromontoryMarineNationalPark-0.2m-2013.zip +https://files.ausseabed.gov.au/survey/WilsonsPromontoryNationalPark-Bathymetry-2m-2016.zip +https://files.ausseabed.gov.au/survey/ZeehanandFranklinMarineParksWestCoastTasmania-Bathymetry-2m-2022.zip diff --git a/sources/ausseabed/harvest.py b/sources/ausseabed/harvest.py new file mode 100644 index 00000000..48f5a5da --- /dev/null +++ b/sources/ausseabed/harvest.py @@ -0,0 +1,197 @@ +#!/usr/bin/env python3 +"""Regenerate file_list.txt for the AusSeabed source (NOT part of the build). + +The build itself just fetches the URLs in file_list.txt. Re-run this when +Geoscience Australia publishes new surveys: + + python harvest.py # queries the register, verifies zips, rewrites file_list.txt + python harvest.py --check # offline self-check + +The AusSeabed Marine Data Register is a public GeoServer WFS +(warehouse.ausseabed.gov.au). Its ACQUISITIONS_INDEX layer catalogs every known +Australian survey (~3,300), most of which are register-only metadata: restricted +licences, third-party portals, or no download at all. The buildable subset is +PRODUCT_STATUS=PUBLISHED + CC-BY 4.0 + no embargo, whose DATA_URLs are zip +archives of Float32 COGs on files.ausseabed.gov.au (anonymous CloudFront/S3). +Some records instead emit s3://seabed-producthouse-open URLs, which deny +anonymous access — the same basenames resolve on the files host, so we map them. +The COMPILATIONS_INDEX layer (SDB and regional compilations) is deliberately +excluded: SDB is too noisy for a chart (see the tabled Allen Coral Atlas), and +the useful compilations (gbr30, AusBathyTopo) are already their own sources. + +Every kept zip's central directory is range-read to confirm it's fetchable and +to catch tif basename collisions — source_unzip flattens members by basename, +so a collision across zips would silently overwrite one survey with another. +Stdlib only, no pipeline coupling. +""" + +import csv +import io +import json +import struct +import sys +import urllib.parse +import urllib.request + +WFS = "https://warehouse.ausseabed.gov.au/geoserver/ows" +LAYER = "ausseabed:MARINEDATAREGISTER_ACQUISITIONS_INDEX" +FIELDS = ["NAME", "NEWGAID", "BATHY_TYPES", "DATA_TYPES", "DATA_URL", "META_URL", + "LEGAL_CONSTRAINTS", "EMBARGO", "PRODUCT_STATUS", "AREA_KM2"] +FILES_HOST = "files.ausseabed.gov.au" +CC_BY = "Creative Commons - Attribution 4.0 International" + + +def fetch_index(): + q = urllib.parse.urlencode({ + "service": "WFS", "version": "2.0.0", "request": "GetFeature", + "typeNames": LAYER, "outputFormat": "csv", + "propertyName": ",".join(FIELDS), # skip GEOM — footprints are ~100 MB + }) + with urllib.request.urlopen(f"{WFS}?{q}", timeout=120) as r: + return list(csv.DictReader(io.TextIOWrapper(r, encoding="utf-8"))) + + +def map_url(url): + """A record's DATA_URL resolved to the anonymous files host, or None if the + data isn't GA-hosted (state portals, Azure blobs, 'N/A', …).""" + host = urllib.parse.urlparse(url).netloc + if host == FILES_HOST: + return url + if "seabed-producthouse-open" in host: # 403 anonymous; same basename on the files host + return f"https://{FILES_HOST}/survey/{url.rsplit('/', 1)[-1]}" + return None + + +def select(rows): + kept, dropped = {}, {} + for r in rows: + reason = None + if r["PRODUCT_STATUS"] != "PUBLISHED": + reason = f"status {r['PRODUCT_STATUS'] or '?'}" + elif r["LEGAL_CONSTRAINTS"] != CC_BY: + reason = f"licence {r['LEGAL_CONSTRAINTS'] or '?'}" + elif r["EMBARGO"] != "No": + reason = "embargoed" + elif "satellite" in r["BATHY_TYPES"].lower(): # SDB — see the tabled ACA source + reason = "satellite-derived" + elif "bathymetry" not in r["DATA_TYPES"].lower() or \ + "backscatter" in r["DATA_URL"].rsplit("/", 1)[-1].lower(): + # backscatter/sidescan products; some re-bundle their sibling's bathy tif, + # which would collide with the real bathymetry zip at unzip time + reason = f"not a bathymetry product ({r['DATA_TYPES'] or '?'})" + elif not map_url(r["DATA_URL"]): + reason = f"not GA-hosted ({urllib.parse.urlparse(r['DATA_URL']).netloc or 'no url'})" + if reason: + dropped[reason] = dropped.get(reason, 0) + 1 + else: + kept[map_url(r["DATA_URL"])] = r["NAME"] # dedupe: some zips appear twice + return kept, dropped + + +def zip_tif_members(url): + """Data-tif member basenames from a remote zip's central directory (two range + reads, no download). Raises urllib.error.HTTPError on 403/404.""" + def ranged(spec): + req = urllib.request.Request(url, headers={"Range": f"bytes={spec}"}) + with urllib.request.urlopen(req, timeout=60) as r: + return r.read() + + tail = ranged("-66000") # EOCD + max comment + eocd = tail.rfind(b"PK\x05\x06") + cd_size, cd_off = struct.unpack("