geo: measure block elevation and pick the DEM source

Probe 22 points with the USGS 3DEP point service: walkable surfaces span
115.65-119.59 m, a 3.94 m relief, so elevation data is required by the AGENTS
rule. Grade falls ~3.3 m eastward over the 279 m block (1.2%) and the north
sidewalk sits 0.4-0.7 m above the south. 3DEP 1 m is public domain and becomes
the chosen source; the tile itself is still to be downloaded.

Factored the shared OSM helpers into Geo/scripts/osm_common.py so the street
profile and the elevation probe read the same geometry.
This commit is contained in:
zwitschi committed 2026-11-06 22:43:13 +01:00
1 parent 255c1a9e86
commit 65fb3759c9
9 files changed
+621 -152

No files matched your search

+188
View File
@@ -0,0 +1,188 @@
#!/usr/bin/env python3
"""Probe ground elevation across the demo block with the USGS 3DEP point service.
Usage:
Geo/scripts/elevation_probe.py Geo/raw/osm-api-map-2026-11-06.osm
Samples a grid along the boulevard (centreline and both sidewalk centrelines),
the hero footprint corners, and the crossing midpoints, then reports the relief
across the demo and whether a separate elevation source is needed.
The decision rule follows `AGENTS.md`: acquire separately licensed elevation data
unless the DEM shows less than 1 m of height variation across the demo bounds.
Writes the raw responses to `Geo/raw/epqs-<date>.json` so the numbers can be
re-checked without re-querying. Service: USGS Elevation Point Query Service,
3DEP 1 m seamless DEM (CONUS). Standard library only.
"""
import argparse
import concurrent.futures
import datetime
import json
import os
import time
import urllib.parse
import urllib.request
import osm_common as oc
EPQS = "https://epqs.nationalmap.gov/v1/json"
UA = "gig-line-geo/0.1 (local development, elevation check)"
def query(lat: float, lon: float, retries: int = 5) -> dict:
params = urllib.parse.urlencode(
{"x": f"{lon:.7f}", "y": f"{lat:.7f}", "wkid": "4326", "units": "Meters", "includeDate": "false"}
)
url = f"{EPQS}?{params}"
for attempt in range(retries):
req = urllib.request.Request(url, headers={"User-Agent": UA})
try:
with urllib.request.urlopen(req, timeout=40) as resp:
payload = json.loads(resp.read().decode("utf-8"))
value = payload.get("value")
if value in (None, "", "No Data"):
raise ValueError(f"no elevation returned: {payload}")
return {
"lat": round(lat, 7),
"lon": round(lon, 7),
"elevation_m": float(value),
"raster_id": payload.get("rasterId"),
"resolution_m": payload.get("resolution"),
}
except Exception as exc: # noqa: BLE001
if attempt == retries - 1:
# a single dead point must not kill the whole probe
print(f" ! no elevation at {lat:.6f},{lon:.6f}: {exc}", flush=True)
return {"lat": round(lat, 7), "lon": round(lon, 7), "elevation_m": None,
"raster_id": None, "resolution_m": None}
time.sleep(3 * (attempt + 1))
return {}
def main():
ap = argparse.ArgumentParser()
ap.add_argument("osm")
ap.add_argument("--blvd", default="Hollywood Boulevard")
ap.add_argument("--west", default="North Orange Drive")
ap.add_argument("--east", default="North Highland Avenue")
ap.add_argument("--hero", type=int, default=424024147, help="hero building OSM way id")
ap.add_argument("--workers", type=int, default=4, help="parallel requests; the service is slow")
ap.add_argument("--pause", type=float, default=0.0)
args = ap.parse_args()
nodes, ways = oc.load(args.osm)
line, lat0, lon0, lengths, block_len = oc.block_extent(nodes, ways, args.blvd, args.west, args.east)
samples = []
grid = [block_len * i / 4 for i in range(5)]
for s in grid:
for label, offset in (("south sidewalk", -12.6), ("centreline", 0.0), ("north sidewalk", 13.6)):
lat, lon = oc.point_at(line, lengths, s, offset, lat0, lon0)
samples.append({"label": f"s={s:5.1f} m {label}", "lat": lat, "lon": lon})
hero_refs = next((refs for wid, _, refs in ways if str(wid) == str(args.hero)), None)
if hero_refs:
hero_pts = oc.way_points(nodes, hero_refs)
lats = [p[0] for p in hero_pts]
lons = [p[1] for p in hero_pts]
corners = [
("near NW", min(lats), min(lons)),
("near NE", min(lats), max(lons)),
("far SW", max(lats), min(lons)),
("far SE", max(lats), max(lons)),
]
for label, lat, lon in corners:
samples.append({"label": f"hero {label}", "lat": lat, "lon": lon})
for wid, tags, refs in ways:
if tags.get("footway") != "crossing":
continue
pts = oc.way_points(nodes, refs)
if not pts:
continue
mid = (sum(p[0] for p in pts) / len(pts), sum(p[1] for p in pts) / len(pts))
dist, s, _ = oc.project(mid, line, lat0, lon0, lengths)
# only crossings of THIS boulevard, not side-street crossings inside the window
if 0 <= s <= block_len and dist <= 4.0:
samples.append({"label": f"crossing way {wid} at s={s:.0f} m", "lat": mid[0], "lon": mid[1]})
def fetch(sample):
r = query(sample["lat"], sample["lon"])
r["label"] = sample["label"]
return r
started = time.time()
with concurrent.futures.ThreadPoolExecutor(max_workers=max(1, args.workers)) as pool:
futures = {pool.submit(fetch, s): s for s in samples}
results = []
for done, future in enumerate(concurrent.futures.as_completed(futures), start=1):
r = future.result()
results.append(r)
value = "no data" if r["elevation_m"] is None else f"{r['elevation_m']:7.2f} m"
print(f" [{done:02d}/{len(samples)}] {r['label']:28} {value:>10}", flush=True)
if args.pause:
time.sleep(args.pause)
order = {s["label"]: i for i, s in enumerate(samples)}
results.sort(key=lambda r: order.get(r["label"], 999))
print(f"query time {time.time() - started:.0f} s for {len(samples)} samples")
walkable = [r for r in results if r["elevation_m"] is not None and not r["label"].startswith("hero")]
building = [r for r in results if r["elevation_m"] is not None and r["label"].startswith("hero")]
values = [r["elevation_m"] for r in walkable]
lo, hi = min(values), max(values)
relief = hi - lo
print("\nsamples by group (walkable surfaces only below; building interiors are diagnostics):")
for key in ("crossing", "hero"):
group = [r for r in results if r["elevation_m"] is not None and key in r["label"]]
if group:
g = [r["elevation_m"] for r in group]
print(f" {key:12} n={len(group):2} min {min(g):.2f} max {max(g):.2f} spread {max(g)-min(g):.2f} m")
print("\ncross-slope check (south sidewalk minus north sidewalk):")
for s in [block_len * i / 4 for i in range(5)]:
south = next(r for r in results if r["label"] == f"s={s:5.1f} m south sidewalk")
north = next(r for r in results if r["label"] == f"s={s:5.1f} m north sidewalk")
drop = south["elevation_m"] - north["elevation_m"]
print(f" s={s:5.1f} m: south {south['elevation_m']:.2f} m, north {north['elevation_m']:.2f} m, "
f"difference {drop:+.2f} m over ~26 m")
if building:
print("\nwarning: points inside the hero footprint are not ground reference; 3DEP can return")
print(" building or void values there. They are excluded from the relief figure.")
print(f"\nelevation range on walkable surfaces: {lo:.2f} to {hi:.2f} m, relief {relief:.2f} m")
first = next(r for r in walkable)
print(f"source: USGS 3DEP point service, raster {first['raster_id']}, resolution {first['resolution_m']} m")
if relief < 1.0:
print("decision: relief under 1 m, a separate elevation source is NOT required (AGENTS.md rule)")
else:
print("decision: relief at or above 1 m, a separately licensed elevation source IS required")
print("datum: 3DEP CONUS heights are orthometric NAVD88; treat absolute values as such, not ellipsoidal")
today = datetime.date.today().isoformat()
out_path = os.path.join("Geo", "raw", f"epqs-{today}.json")
os.makedirs(os.path.dirname(out_path), exist_ok=True)
payload = {
"service": EPQS,
"processed_utc": datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds"),
"source_dem": "USGS 3DEP seamless 1 m (CONUS)",
"vertical_datum": "NAVD88 orthometric (3DEP CONUS)",
"query": {
"blvd": args.blvd,
"west": args.west,
"east": args.east,
"hero_way": args.hero,
"block_len_m": round(block_len, 2),
},
"summary": {"min_m": lo, "max_m": hi, "relief_m": round(relief, 3), "samples": len(results)},
"samples": results,
}
with open(out_path, "w", encoding="utf-8") as handle:
json.dump(payload, handle, indent=2)
print(f"wrote {out_path}")
if __name__ == "__main__":
main()
+151
View File
@@ -0,0 +1,151 @@
#!/usr/bin/env python3
"""Shared helpers for the Geo scripts: parse .osm, stitch a street, project points.
Standard library only. Imported by `block_report.py`, `street_profile.py`, and
`elevation_probe.py` when run from this directory.
"""
from __future__ import annotations
import math
import xml.etree.ElementTree as ET
EARTH_R = 6378137.0
def to_xy(lat: float, lon: float, lat0: float, lon0: float) -> tuple[float, float]:
"""Local planar metres, east/north, centred on (lat0, lon0)."""
return (
math.radians(lon - lon0) * EARTH_R * math.cos(math.radians(lat0)),
math.radians(lat - lat0) * EARTH_R,
)
def from_xy(x: float, y: float, lat0: float, lon0: float) -> tuple[float, float]:
lat = lat0 + math.degrees(y / EARTH_R)
lon = lon0 + math.degrees(x / (EARTH_R * math.cos(math.radians(lat0))))
return lat, lon
def load(path: str):
root = ET.parse(path).getroot()
nodes = {n.get("id"): (float(n.get("lat")), float(n.get("lon"))) for n in root.findall("node")}
ways = []
for w in root.findall("way"):
tags = {c.get("k"): c.get("v") for c in w.findall("tag")}
ways.append((w.get("id"), tags, [nd.get("ref") for nd in w.findall("nd")]))
return nodes, ways
def way_points(nodes, refs):
return [nodes[r] for r in refs if r in nodes]
def refs_by_name(ways, name):
return {r for _, tags, refs in ways if tags.get("name") == name for r in refs}
def stitch(nodes, ways, name, lon_min, lon_max):
"""Ordered polyline for a named street, restricted to a longitude window."""
segments = []
for _, tags, refs in ways:
if tags.get("name") != name:
continue
pts = way_points(nodes, refs)
if not pts:
continue
if max(p[1] for p in pts) < lon_min - 1e-6 or min(p[1] for p in pts) > lon_max + 1e-6:
continue
segments.append(pts)
merged = []
for pts in segments:
if not merged:
merged = list(pts)
continue
if pts[0] == merged[-1]:
merged.extend(pts[1:])
elif pts[-1] == merged[-1]:
merged.extend(reversed(pts[:-1]))
elif pts[0] == merged[0]:
merged = list(reversed(pts)) + merged[1:]
elif pts[-1] == merged[0]:
merged = pts + merged[1:]
return merged
def cumulative_lengths(points, lat0, lon0):
xs, total, prev = [], 0.0, None
for lat, lon in points:
p = to_xy(lat, lon, lat0, lon0)
if prev is not None:
total += math.dist(prev, p)
xs.append(total)
prev = p
return xs
def project(point, line, lat0, lon0, lengths):
"""Nearest point on the polyline: (distance_m, s_along_m, side) with side +1 = north."""
px, py = to_xy(point[0], point[1], lat0, lon0)
best = (float("inf"), 0.0, 0)
for i in range(len(line) - 1):
ax, ay = to_xy(line[i][0], line[i][1], lat0, lon0)
bx, by = to_xy(line[i + 1][0], line[i + 1][1], lat0, lon0)
dx, dy = bx - ax, by - ay
seg2 = dx * dx + dy * dy
if seg2 == 0:
continue
t = max(0.0, min(1.0, ((px - ax) * dx + (py - ay) * dy) / seg2))
cx, cy = ax + t * dx, ay + t * dy
dist = math.hypot(px - cx, py - cy)
if dist < best[0]:
cross = dx * (py - ay) - dy * (px - ax)
best = (dist, lengths[i] + t * (lengths[i + 1] - lengths[i]), 1 if cross > 0 else -1)
return best
def point_at(line, lengths, s, offset, lat0, lon0):
"""Point on the polyline at arc length s, pushed offset metres to the north side."""
for i in range(len(line) - 1):
if lengths[i] <= s <= lengths[i + 1]:
seg = lengths[i + 1] - lengths[i]
t = 0.0 if seg == 0 else (s - lengths[i]) / seg
ax, ay = to_xy(line[i][0], line[i][1], lat0, lon0)
bx, by = to_xy(line[i + 1][0], line[i + 1][1], lat0, lon0)
dx, dy = bx - ax, by - ay
norm = math.hypot(dx, dy) or 1.0
nx, ny = -dy / norm, dx / norm
x = ax + t * dx + offset * nx
y = ay + t * dy + offset * ny
return from_xy(x, y, lat0, lon0)
return line[-1]
def block_extent(nodes, ways, blvd, west, east):
"""Stitch and clip the boulevard between two cross streets.
Returns (line, lat0, lon0, lengths, block_len) with `line` running west to east.
"""
inter = {}
blvd_refs = refs_by_name(ways, blvd)
for label, street in (("west", west), ("east", east)):
shared = [nodes[r] for r in refs_by_name(ways, street) & blvd_refs if r in nodes]
if not shared:
raise SystemExit(f"no intersection between {blvd!r} and {street!r}")
inter[label] = min(p[1] for p in shared) if label == "west" else max(p[1] for p in shared)
line = stitch(nodes, ways, blvd, inter["west"], inter["east"])
if line[0][1] > line[-1][1]:
line = list(reversed(line))
west_pt = min(line, key=lambda p: abs(p[1] - inter["west"]))
east_pt = min(line, key=lambda p: abs(p[1] - inter["east"]))
i0, i1 = line.index(west_pt), line.index(east_pt)
if i0 > i1:
i0, i1 = i1, i0
line = line[i0 : i1 + 1]
lat0 = sum(p[0] for p in line) / len(line)
lon0 = sum(p[1] for p in line) / len(line)
lengths = cumulative_lengths(line, lat0, lon0)
return line, lat0, lon0, lengths, lengths[-1]
+28 -147
View File
@@ -7,111 +7,23 @@ Usage:
Measures, in metres, from a plain .osm extract (standard library only):
* the boulevard centreline, stitched into one ordered line between the two
boundary intersections;
* the boulevard centreline, stitched and clipped between the two boundary
intersections;
* each building's frontage setback: the distance from its nearest vertex to the
centreline, and which side of the road it is on;
* each mapped crossing (`footway=crossing`) across the boulevard: its length in
metres, which is the carriageway width at that point, and its position along
the block;
* each mapped crossing (`footway=crossing`) across the boulevard: its length,
which is the carriageway width at that point, and its position along the block;
* the offset of the mapped sidewalk centrelines from the road centreline.
What it cannot measure: carriageway width where no crossing or kerb is mapped,
and sidewalk width, because OSM models sidewalks here as centre lines with no
width tag. Treat those as recorded assumptions, not measurements.
width tag. Derive those from the numbers printed here and record them as
assumptions.
"""
import argparse
import math
import xml.etree.ElementTree as ET
EARTH_R = 6378137.0
def to_xy(lat, lon, lat0, lon0):
return (
math.radians(lon - lon0) * EARTH_R * math.cos(math.radians(lat0)),
math.radians(lat - lat0) * EARTH_R,
)
def load(path):
root = ET.parse(path).getroot()
nodes = {n.get("id"): (float(n.get("lat")), float(n.get("lon"))) for n in root.findall("node")}
ways = []
for w in root.findall("way"):
tags = {c.get("k"): c.get("v") for c in w.findall("tag")}
ways.append((w.get("id"), tags, [nd.get("ref") for nd in w.findall("nd")]))
return nodes, ways
def way_points(nodes, refs):
return [nodes[r] for r in refs if r in nodes]
def stitch(nodes, ways, name, lon_min, lon_max):
"""Ordered polyline for a named street between two longitudes."""
segments = []
for _, tags, refs in ways:
if tags.get("name") != name:
continue
pts = way_points(nodes, refs)
if not pts:
continue
if max(p[1] for p in pts) < lon_min - 1e-6 or min(p[1] for p in pts) > lon_max + 1e-6:
continue
segments.append(pts)
if not segments:
raise SystemExit(f"no geometry for {name!r} inside the block")
merged = []
for pts in segments:
if not merged:
merged = list(pts)
continue
if pts[0] == merged[-1]:
merged.extend(pts[1:])
elif pts[-1] == merged[-1]:
merged.extend(reversed(pts[:-1]))
elif pts[0] == merged[0]:
merged = list(reversed(pts)) + merged[1:]
elif pts[-1] == merged[0]:
merged = pts + merged[1:]
return merged
def cumulative_lengths(points, lat0, lon0):
xs = []
total = 0.0
prev = None
for lat, lon in points:
p = to_xy(lat, lon, lat0, lon0)
if prev is not None:
total += math.dist(prev, p)
xs.append(total)
prev = p
return xs
def project(point, line, lat0, lon0, lengths):
"""Nearest point on the polyline: returns (distance_m, s_along_m, side)."""
px, py = to_xy(point[0], point[1], lat0, lon0)
best = (float("inf"), 0.0, 0)
for i in range(len(line) - 1):
ax, ay = to_xy(line[i][0], line[i][1], lat0, lon0)
bx, by = to_xy(line[i + 1][0], line[i + 1][1], lat0, lon0)
dx, dy = bx - ax, by - ay
seg_len2 = dx * dx + dy * dy
if seg_len2 == 0:
continue
t = max(0.0, min(1.0, ((px - ax) * dx + (py - ay) * dy) / seg_len2))
cx, cy = ax + t * dx, ay + t * dy
dist = math.hypot(px - cx, py - cy)
if dist < best[0]:
cross = dx * (py - ay) - dy * (px - ax)
side = 1 if cross > 0 else -1
best = (dist, lengths[i] + t * (lengths[i + 1] - lengths[i]), side)
return best
import osm_common as oc
def main():
@@ -122,59 +34,32 @@ def main():
ap.add_argument("--east", default="North Highland Avenue")
args = ap.parse_args()
nodes, ways = load(args.osm)
inter = {}
for label, street in (("west", args.west), ("east", args.east)):
refs = {r for _, tags, rs in ways if tags.get("name") == street for r in rs}
pts = way_points(nodes, refs)
blvd = {r for _, tags, rs in ways if tags.get("name") == args.blvd for r in rs}
shared = [nodes[r] for r in refs & blvd if r in nodes]
if not shared:
raise SystemExit(f"no intersection with {street!r}")
inter[label] = min(p[1] for p in shared) if label == "west" else max(p[1] for p in shared)
line = stitch(nodes, ways, args.blvd, inter["west"], inter["east"])
# orient west -> east, then clip to the block between the two intersections
if line[0][1] > line[-1][1]:
line = list(reversed(line))
west_pt = min(line, key=lambda p: abs(p[1] - inter["west"]))
east_pt = min(line, key=lambda p: abs(p[1] - inter["east"]))
i0, i1 = line.index(west_pt), line.index(east_pt)
if i0 > i1:
i0, i1 = i1, i0
line = line[i0 : i1 + 1]
lat0 = sum(p[0] for p in line) / len(line)
lon0 = sum(p[1] for p in line) / len(line)
lengths = cumulative_lengths(line, lat0, lon0)
block_len = lengths[-1]
nodes, ways = oc.load(args.osm)
line, lat0, lon0, lengths, block_len = oc.block_extent(
nodes, ways, args.blvd, args.west, args.east
)
print(f"centreline: {len(line)} vertices, {block_len:.1f} m between the boundary intersections\n")
print("crossings across the boulevard (length = carriageway width there):")
for wid, tags, refs in ways:
if tags.get("footway") != "crossing":
continue
pts = way_points(nodes, refs)
pts = oc.way_points(nodes, refs)
if not pts:
continue
lats = [p[0] for p in pts]
lons = [p[1] for p in pts]
# must span the roadway: the along-road extent small, the across extent large
a = to_xy(pts[0][0], pts[0][1], lat0, lon0)
b = to_xy(pts[-1][0], pts[-1][1], lat0, lon0)
length = math.dist(a, b)
mid = (sum(lats) / len(lats), sum(lons) / len(lons))
dist, s, side = project(mid, line, lat0, lon0, lengths)
if min(lons) > inter["east"] or max(lons) < inter["west"]:
if min(p[1] for p in pts) > line[-1][1] or max(p[1] for p in pts) < line[0][1]:
continue
a = oc.to_xy(pts[0][0], pts[0][1], lat0, lon0)
b = oc.to_xy(pts[-1][0], pts[-1][1], lat0, lon0)
length = ((a[0] - b[0]) ** 2 + (a[1] - b[1]) ** 2) ** 0.5
mid = (sum(p[0] for p in pts) / len(pts), sum(p[1] for p in pts) / len(pts))
dist, s, _ = oc.project(mid, line, lat0, lon0, lengths)
if not (0.0 <= s <= block_len) or dist > 4.0:
continue
marks = tags.get("crossing:markings", "?")
print(
f" way {wid:>10} {length:6.1f} m across, at {s:6.1f} m along the block "
f"(offset {dist:4.1f} m) markings={marks:6} signals={tags.get('crossing')}"
f"(offset {dist:4.1f} m) markings={tags.get('crossing:markings', '-'):6} "
f"signals={tags.get('crossing')}"
)
print("\nsidewalk centrelines near the block (offset from road centreline):")
@@ -182,20 +67,17 @@ def main():
for wid, tags, refs in ways:
if tags.get("footway") != "sidewalk":
continue
pts = way_points(nodes, refs)
pts = oc.way_points(nodes, refs)
if not pts:
continue
if min(p[1] for p in pts) > inter["east"] or max(p[1] for p in pts) < inter["west"]:
if min(p[1] for p in pts) > line[-1][1] or max(p[1] for p in pts) < line[0][1]:
continue
offsets = [project(p, line, lat0, lon0, lengths) for p in pts]
offsets = [oc.project(p, line, lat0, lon0, lengths) for p in pts]
mean = sum(o[0] for o in offsets) / len(offsets)
side = offsets[0][2]
along = max(o[1] for o in offsets) - min(o[1] for o in offsets)
# keep only long lines that run roughly parallel within the block
if not (6.0 <= mean <= 20.0) or along < 30.0:
continue
target = south if side < 0 else north
target.append((mean, along, wid, tags.get("name", "")))
(south if offsets[0][2] < 0 else north).append((mean, along, wid, tags.get("name", "")))
for label, rows in (("south side", south), ("north side", north)):
if not rows:
@@ -210,16 +92,15 @@ def main():
for wid, tags, refs in ways:
if "building" not in tags:
continue
pts = way_points(nodes, refs)
pts = oc.way_points(nodes, refs)
if not pts:
continue
if min(p[1] for p in pts) > inter["east"] or max(p[1] for p in pts) < inter["west"]:
if min(p[1] for p in pts) > line[-1][1] or max(p[1] for p in pts) < line[0][1]:
continue
proj = [project(p, line, lat0, lon0, lengths) for p in pts]
proj = [oc.project(p, line, lat0, lon0, lengths) for p in pts]
nearest = min(proj, key=lambda o: o[0])
along = [o[1] for o in proj]
frontage = max(along) - min(along)
rows.append((nearest[0], nearest[2], wid, tags, frontage))
rows.append((nearest[0], nearest[2], wid, tags, max(along) - min(along)))
rows.sort()
for dist, side, wid, tags, frontage in rows:
side_name = "S" if side < 0 else "N"