lgtm

pipeline/download/census_population.py

@@ -0,0 +1,153 @@
+"""Download Census 2021 usual-resident counts per unit postcode.
+
+ONS Census 2021 publishes a *direct* headcount of usual residents for every
+unit postcode in England and Wales (table P001, disaggregated by sex) as a bulk
+CSV on the NOMIS webservice. We sum the two sex rows per postcode to get the
+total usual-resident population and emit one row per unit postcode.
+
+This is a display-only side table (shown in the right-hand area pane); it is NOT
+merged into the property/postcode feature frames and is never a filterable
+attribute. The Rust server loads the parquet directly via --population-path.
+
+Source: NOMIS bulk product "Postcode resident and household estimates, England
+        and Wales: Census 2021" (table P001), as at census day 21 March 2021.
+        https://www.nomisweb.co.uk/sources/census_2021_pc
+License: Open Government Licence v3.0
+
+Caveats (Census 2021 disclosure control): counts carry targeted record swapping
+and cell-key perturbation, and only postcodes with at least one usual resident
+are present, so vacant/non-residential postcodes are absent from the output.
+"""
+
+import argparse
+import tempfile
+import time
+from pathlib import Path
+
+import polars as pl
+
+from pipeline.utils import download
+
+# P001 = "Number of usual residents by postcode by sex", split alphabetically by
+# postcode into four CSVs (each: Postcode, Sex Code, Sex Label, Count). The
+# per-file minimums are ~95% of the published record counts and let us retry
+# until a download is complete (see _fetch_csv).
+BASE = "https://www.nomisweb.co.uk/output/census/2021"
+P001_FILES = [
+    ("pcd_p001_a_d.csv", 720_000),
+    ("pcd_p001_e_l.csv", 540_000),
+    ("pcd_p001_m_r.csv", 630_000),
+    ("pcd_p001_s_z.csv", 750_000),
+]
+
+# P001 covers England & Wales (~1.37M postcodes). A materially smaller result
+# means a split file was truncated or silently 404'd.
+MIN_EXPECTED_POSTCODES = 1_000_000
+
+
+def _canonical_postcode(col: pl.Expr) -> pl.Expr:
+    """Canonical spaced unit postcode (e.g. "AL1 1AG"), matching the Rust
+    server's `normalize_postcode`: uppercase, strip non-alphanumerics, then
+    insert a single space before the final three (inward-code) characters."""
+    cleaned = col.cast(pl.String).str.to_uppercase().str.replace_all(r"[^A-Z0-9]", "")
+    return (
+        cleaned.str.slice(0, cleaned.str.len_chars() - 3)
+        + pl.lit(" ")
+        + cleaned.str.slice(-3)
+    )
+
+
+def _fetch_csv(
+    url: str, dest: Path, min_rows: int, *, max_attempts: int = 20
+) -> pl.DataFrame:
+    """Download a CSV, defending against silent truncation.
+
+    The NOMIS bulk files are served with chunked transfer encoding and no
+    Content-Length, so a dropped connection ends the stream without raising —
+    yielding a short file. We retry until the parsed row count reaches the
+    file's known approximate size, keeping the largest parse seen.
+    """
+    best: pl.DataFrame | None = None
+    for attempt in range(1, max_attempts + 1):
+        try:
+            download(url, dest)
+            df = pl.read_csv(dest)
+        except Exception as exc:  # noqa: BLE001 — retry any transport/parse error
+            print(f"  {url} attempt {attempt}: error {exc}")
+            time.sleep(3)
+            continue
+        rows = df.height
+        if best is None or rows > best.height:
+            best = df
+        complete = rows >= min_rows
+        print(
+            f"  {url} attempt {attempt}: {rows} rows "
+            f"({'complete' if complete else f'< {min_rows}, retrying'})"
+        )
+        if complete:
+            return df
+        time.sleep(2)
+    raise RuntimeError(
+        f"Failed to fully download {url} after {max_attempts} attempts "
+        f"(best {best.height if best is not None else 0} rows, need >= {min_rows})"
+    )
+
+
+def download_and_convert(output_path: Path) -> None:
+    frames: list[pl.DataFrame] = []
+    with tempfile.TemporaryDirectory() as tmp:
+        for name, min_rows in P001_FILES:
+            url = f"{BASE}/{name}"
+            dest = Path(tmp) / name
+            print(f"Downloading {url} ...")
+            frames.append(_fetch_csv(url, dest, min_rows))
+
+    df = pl.concat(frames)
+    print(f"Total P001 rows (postcode x sex): {df.height}")
+
+    if "Count" not in df.columns or "Postcode" not in df.columns:
+        raise ValueError(
+            f"Unexpected P001 columns {df.columns}; expected 'Postcode' and 'Count'."
+        )
+
+    result = (
+        df.with_columns(_canonical_postcode(pl.col("Postcode")).alias("postcode"))
+        .filter(pl.col("postcode").str.len_chars() >= 5)
+        .group_by("postcode")
+        .agg(pl.col("Count").cast(pl.Int64).sum().alias("population"))
+        .filter(pl.col("population") > 0)
+        .with_columns(pl.col("population").cast(pl.UInt32))
+        .sort("postcode")
+    )
+
+    print(f"Unit postcodes with population: {result.height}")
+    if result.height < MIN_EXPECTED_POSTCODES:
+        raise ValueError(
+            f"Only {result.height} postcodes (expected >= {MIN_EXPECTED_POSTCODES}); "
+            "a NOMIS P001 split file was likely truncated or unavailable."
+        )
+    total = result["population"].sum()
+    print(f"Total usual residents (England & Wales): {total:,}")
+    print(
+        f"Per-postcode population range: {result['population'].min()} - "
+        f"{result['population'].max()}"
+    )
+
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    result.write_parquet(output_path, compression="zstd")
+    print(f"Saved to {output_path}")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Download Census 2021 usual-resident counts per unit postcode"
+    )
+    parser.add_argument(
+        "--output", type=Path, required=True, help="Output parquet file path"
+    )
+    args = parser.parse_args()
+    download_and_convert(args.output)
+
+
+if __name__ == "__main__":
+    main()

pipeline/download/development_sites.py

@@ -0,0 +1,302 @@
+"""Download planned/pipeline development sites for the "new developments" layer.
+
+This is the forward-looking "where new homes are coming" signal that complements
+the EPC new-build + Land Registry "first sale" delivery signals already in the
+pipeline. Two national, Open Government Licence v3.0 sources are merged into a
+single coordinate-keyed parquet that the Rust server serves as a map layer:
+
+  - MHCLG Brownfield Land register (statutory LPA brownfield registers, normalised)
+    https://www.planning.data.gov.uk/dataset/brownfield-land
+  - Homes England Land Hub (public land being disposed for development)
+    https://www.gov.uk/government/publications/homes-england-land-hub
+
+Output schema (one row per site):
+    lat, lon, source, name, min_dwellings, max_dwellings, planning_status,
+    permission_type, permission_date, hectares, local_authority, url
+"""
+
+import argparse
+import math
+import re
+import tempfile
+from pathlib import Path
+
+import httpx
+import polars as pl
+from shapely.geometry import shape
+
+from pipeline.local_temp import local_tmp_dir
+from pipeline.utils.download import download
+
+BROWNFIELD_URL = "https://files.planning.data.gov.uk/dataset/brownfield-land.parquet"
+HOMES_ENGLAND_URL = (
+    "https://services-eu1.arcgis.com/yo0w4PgP4XL49bfF/arcgis/rest/services/"
+    "Homes_England_Land_Hub_Sites/FeatureServer/0/query"
+)
+
+# Output column order; doubles as the polars schema so empty inputs still produce
+# a well-typed frame. The Rust loader (data/developments.rs) reads these names.
+SCHEMA: dict[str, pl.DataType] = {
+    "lat": pl.Float64,
+    "lon": pl.Float64,
+    "source": pl.String,
+    "name": pl.String,
+    "min_dwellings": pl.Int32,
+    "max_dwellings": pl.Int32,
+    "planning_status": pl.String,
+    "permission_type": pl.String,
+    "permission_date": pl.String,
+    "hectares": pl.Float64,
+    "local_authority": pl.String,
+    "url": pl.String,
+}
+
+_COORD = r"-?\d+(?:\.\d+)?(?:[eE][+-]?\d+)?"
+_POINT_RE = re.compile(rf"POINT\s*\(\s*({_COORD})\s+({_COORD})\s*\)", re.IGNORECASE)
+
+# 1 acre = 0.404686 hectares. The Homes England feed reports area in acres.
+_ACRES_TO_HECTARES = 0.404686
+
+# Great Britain bounding box, used to reject mis-ordered or projected coordinates
+# (e.g. a WKT written "lat lon", or stray British National Grid eastings).
+_LON_MIN, _LON_MAX = -9.0, 2.1
+_LAT_MIN, _LAT_MAX = 49.0, 61.1
+
+
+def _first(row: dict, *keys: str):
+    """First present, non-None value among ``keys`` (case-sensitive)."""
+    for key in keys:
+        if key in row and row[key] is not None:
+            return row[key]
+    return None
+
+
+def _clean_str(value) -> str | None:
+    if value is None:
+        return None
+    text = str(value).strip()
+    return text or None
+
+
+def _to_int(value) -> int | None:
+    if value is None:
+        return None
+    try:
+        number = float(value)
+    except (TypeError, ValueError):
+        return None
+    if not math.isfinite(number):  # NaN or +/-inf (int(inf) raises OverflowError)
+        return None
+    return int(number)
+
+
+def _to_float(value) -> float | None:
+    if value is None:
+        return None
+    try:
+        number = float(value)
+    except (TypeError, ValueError):
+        return None
+    if not math.isfinite(number):  # NaN or +/-inf
+        return None
+    return number
+
+
+def _valid_lonlat(lon: float, lat: float) -> bool:
+    return _LON_MIN <= lon <= _LON_MAX and _LAT_MIN <= lat <= _LAT_MAX
+
+
+def _point_lonlat(row: dict) -> tuple[float, float] | None:
+    """Extract (lon, lat) from a brownfield row's WKT point or lat/lon columns."""
+    wkt = _clean_str(_first(row, "point"))
+    if wkt:
+        match = _POINT_RE.match(wkt)
+        if match:
+            lon, lat = float(match.group(1)), float(match.group(2))
+            if _valid_lonlat(lon, lat):
+                return lon, lat
+    lon = _to_float(_first(row, "longitude", "long", "lng"))
+    lat = _to_float(_first(row, "latitude", "lat"))
+    if lon is not None and lat is not None and _valid_lonlat(lon, lat):
+        return lon, lat
+    return None
+
+
+def _geometry_centroid(geom) -> tuple[float, float] | None:
+    """Centroid (lon, lat) of a GeoJSON geometry already in WGS84."""
+    if not geom:
+        return None
+    try:
+        centroid = shape(geom).centroid
+    except (ValueError, TypeError, AttributeError):
+        return None
+    if centroid.is_empty:
+        return None
+    lon, lat = float(centroid.x), float(centroid.y)
+    return (lon, lat) if _valid_lonlat(lon, lat) else None
+
+
+def _has_dwellings(min_d: int | None, max_d: int | None) -> bool:
+    return (min_d is not None and min_d > 0) or (max_d is not None and max_d > 0)
+
+
+def _brownfield_url(entity) -> str | None:
+    """Canonical per-site page on the Planning Data platform.
+
+    The register's own ``site-plan-url`` is unreliable — many LPAs point every
+    row at a single generic register landing page — so we link to the stable,
+    always-per-site entity page instead.
+    """
+    entity_id = _to_int(entity)
+    if entity_id is None:
+        return None
+    return f"https://www.planning.data.gov.uk/entity/{entity_id}"
+
+
+def _is_residential(use: str | None, capacity: int | None) -> bool:
+    """Keep Homes England sites that will deliver homes."""
+    if capacity is not None and capacity > 0:
+        return True
+    if use is None:
+        return False
+    lowered = use.lower()
+    return any(token in lowered for token in ("resid", "housing", "mixed", "dwelling"))
+
+
+def _frame_from_rows(rows: list[dict]) -> pl.DataFrame:
+    return pl.DataFrame(rows, schema=SCHEMA, orient="row")
+
+
+def brownfield_to_frame(raw: pl.DataFrame) -> pl.DataFrame:
+    """Normalise the brownfield-land register to the unified development schema.
+
+    Drops sites that have left the register (a non-empty ``end-date`` means the
+    site was built out or withdrawn) and sites with no residential capacity, so
+    the layer shows pipeline housing rather than every parcel ever registered.
+    """
+    rows: list[dict] = []
+    for row in raw.iter_rows(named=True):
+        lonlat = _point_lonlat(row)
+        if lonlat is None:
+            continue
+        if _clean_str(_first(row, "end-date", "end_date")):
+            continue
+        min_d = _to_int(_first(row, "minimum-net-dwellings", "minimum_net_dwellings"))
+        max_d = _to_int(_first(row, "maximum-net-dwellings", "maximum_net_dwellings"))
+        if not _has_dwellings(min_d, max_d):
+            continue
+        lon, lat = lonlat
+        rows.append(
+            {
+                "lat": lat,
+                "lon": lon,
+                "source": "brownfield",
+                "name": _clean_str(_first(row, "site-address", "name")),
+                "min_dwellings": min_d,
+                "max_dwellings": max_d,
+                "planning_status": _clean_str(
+                    _first(row, "planning-permission-status")
+                ),
+                "permission_type": _clean_str(_first(row, "planning-permission-type")),
+                "permission_date": _clean_str(_first(row, "planning-permission-date")),
+                "hectares": _to_float(_first(row, "hectares")),
+                "local_authority": _clean_str(_first(row, "organisation")),
+                "url": _brownfield_url(_first(row, "entity")),
+            }
+        )
+    return _frame_from_rows(rows)
+
+
+def homes_england_to_frame(features: list[dict]) -> pl.DataFrame:
+    """Normalise Homes England Land Hub GeoJSON features to the unified schema."""
+    rows: list[dict] = []
+    for feature in features:
+        props = feature.get("properties") or {}
+        lonlat = _geometry_centroid(feature.get("geometry"))
+        if lonlat is None:
+            continue
+        capacity = _to_int(
+            _first(props, "Housing_Capacity", "HousingCapacity", "Units", "Homes")
+        )
+        use = _clean_str(_first(props, "Proposed_Use"))
+        if not _is_residential(use, capacity):
+            continue
+        lon, lat = lonlat
+        # The feed reports area in acres (`Gross_Area__Acres_`); convert to hectares
+        # so the column is consistent with the brownfield register.
+        acres = _to_float(_first(props, "Gross_Area__Acres_"))
+        hectares = round(acres * _ACRES_TO_HECTARES, 4) if acres is not None else None
+        rows.append(
+            {
+                "lat": lat,
+                "lon": lon,
+                "source": "homes-england",
+                "name": _clean_str(_first(props, "Parcel_Name", "Site_Reference")),
+                "min_dwellings": None,
+                "max_dwellings": capacity,
+                "planning_status": _clean_str(_first(props, "Planning_Status")),
+                "permission_type": None,
+                "permission_date": None,
+                "hectares": hectares,
+                "local_authority": _clean_str(_first(props, "Local_Authority")),
+                # The Land Hub exposes no stable per-site page; leave the link unset
+                # rather than pointing at a generic landing page.
+                "url": None,
+            }
+        )
+    return _frame_from_rows(rows)
+
+
+def _fetch_homes_england() -> list[dict]:
+    params = {
+        "where": "1=1",
+        "outFields": "*",
+        "outSR": "4326",
+        "f": "geojson",
+        "resultRecordCount": "2000",
+    }
+    response = httpx.get(
+        HOMES_ENGLAND_URL, params=params, follow_redirects=True, timeout=120
+    )
+    response.raise_for_status()
+    payload = response.json()
+    if payload.get("exceededTransferLimit"):
+        print(
+            "  WARNING: Homes England query hit the transfer limit; "
+            "only the first page of sites was fetched."
+        )
+    return payload.get("features", []) or []
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Download brownfield + Homes England development sites"
+    )
+    parser.add_argument(
+        "--output", type=Path, required=True, help="Output parquet file path"
+    )
+    args = parser.parse_args()
+
+    with tempfile.TemporaryDirectory(dir=local_tmp_dir()) as cache_dir:
+        brownfield_path = Path(cache_dir) / "brownfield-land.parquet"
+        print("Downloading MHCLG brownfield land register...")
+        download(BROWNFIELD_URL, brownfield_path)
+        brownfield = brownfield_to_frame(pl.read_parquet(brownfield_path))
+        print(f"  Brownfield: {brownfield.height} residential sites")
+
+        print("Downloading Homes England Land Hub...")
+        homes_england = homes_england_to_frame(_fetch_homes_england())
+        print(f"  Homes England: {homes_england.height} residential sites")
+
+    combined = pl.concat([brownfield, homes_england]).filter(
+        pl.col("lat").is_not_null() & pl.col("lon").is_not_null()
+    )
+
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    combined.write_parquet(args.output, compression="zstd")
+    size_kb = args.output.stat().st_size / 1024
+    print(f"Saved {combined.height} development sites to {args.output} ({size_kb:.0f} KB)")
+
+
+if __name__ == "__main__":
+    main()

pipeline/download/education.py

@@ -0,0 +1,177 @@
+"""Download Census 2021 highest level of qualification (TS067) by LSOA.
+
+Downloads the 8-category "highest level of qualification" breakdown (TS067,
+classification C2021_HIQUAL_8) from the NOMIS API at LSOA 2021 granularity and
+emits one row per LSOA with the percentage of usual residents aged 16+ in each
+qualification band. The bands sum to 100%, so downstream they render as a
+composition/ratio (like the ethnicity and political-vote-share stacked bars)
+rather than a single headline number.
+
+We give the ONS bands colloquial labels (the census's "Level 1/2/3/4+" jargon
+means little to a homebuyer): No qualifications / Some GCSEs / Good GCSEs /
+Apprenticeship / A-levels / Degree or higher / Other qualifications. NOTE the
+census does NOT split undergraduate from postgraduate — "Level 4 and above" is a
+single bucket ("Degree or higher").
+
+The join key downstream (merge.py) is `lsoa21`, the same key used for ethnicity,
+median age, and IoD.
+
+Source: NOMIS (ONS Census 2021 — TS067 dataset, NM_2084_1)
+License: Open Government Licence v3.0
+"""
+
+import argparse
+from pathlib import Path
+
+import polars as pl
+
+from pipeline.utils import ENGLAND_LSOA_COUNT_2021, download_nomis_csv
+
+pl.Config.set_tbl_cols(-1)
+
+# NOMIS API: Census 2021 TS067 (highest level of qualification) by LSOA 2021
+# (TYPE151). c2021_hiqual_8=1..7 selects the 7 substantive bands (excluding
+# 0 = Total, which we re-derive by summing). measures=20100 selects the count.
+BASE_URL = (
+    "https://www.nomisweb.co.uk/api/v01/dataset/NM_2084_1.data.csv"
+    "?date=latest"
+    "&geography=TYPE151"
+    "&c2021_hiqual_8=1,2,3,4,5,6,7"
+    "&measures=20100"
+    "&select=GEOGRAPHY_CODE,C2021_HIQUAL_8_NAME,OBS_VALUE"
+)
+
+# Map each canonical NOMIS C2021_HIQUAL_8 band to our colloquial output bucket.
+# 1:1 mapping (no folding) — keyed on the exact NOMIS label so a relabelled or
+# missing band fails loudly in validation instead of silently dropping people.
+BAND_MAP = {
+    "No qualifications": "No qualifications",
+    "Level 1 and entry level qualifications": "Some GCSEs",
+    "Level 2 qualifications": "Good GCSEs",
+    "Apprenticeship": "Apprenticeship",
+    "Level 3 qualifications": "A-levels",
+    "Level 4 qualifications or above": "Degree or higher",
+    "Other qualifications": "Other qualifications",
+}
+
+# Output buckets in ascending-attainment order, so the stacked composition reads
+# left-to-right from least to most qualified.
+OUTPUT_BUCKETS = [
+    "No qualifications",
+    "Some GCSEs",
+    "Good GCSEs",
+    "Apprenticeship",
+    "A-levels",
+    "Degree or higher",
+    "Other qualifications",
+]
+assert set(BAND_MAP.values()) == set(OUTPUT_BUCKETS), (
+    "BAND_MAP values must be exactly the OUTPUT_BUCKETS"
+)
+
+
+def _qualification_percentages(df: pl.DataFrame) -> pl.DataFrame:
+    """Fold the 7 NOMIS bands into percentage buckets per LSOA (summing to 100).
+
+    `df` is the long-format NOMIS download with columns GEOGRAPHY_CODE,
+    C2021_HIQUAL_8_NAME (the band label) and OBS_VALUE (a count). A missing,
+    extra, or relabelled band would silently change the denominator (the sum over
+    bands) so we validate the band set against BAND_MAP first and fail otherwise.
+    Returns one row per LSOA with `lsoa21` and `% <bucket>` for each bucket.
+    """
+    found = set(df["C2021_HIQUAL_8_NAME"].unique().to_list())
+    expected = set(BAND_MAP)
+    if found != expected:
+        missing = sorted(expected - found)
+        unexpected = sorted(found - expected)
+        raise ValueError(
+            "Census qualification bands do not match the expected NOMIS "
+            "TS067 C2021_HIQUAL_8 set.\n"
+            f"  expected {len(expected)} bands, found {len(found)}\n"
+            f"  missing:    {missing}\n"
+            f"  unexpected: {unexpected}\n"
+            "Refusing to compute percentages against an unrecognised breakdown."
+        )
+
+    grouped = (
+        df.with_columns(
+            pl.col("C2021_HIQUAL_8_NAME").replace_strict(BAND_MAP).alias("bucket"),
+            pl.col("OBS_VALUE").cast(pl.Float64, strict=False).alias("_count"),
+        )
+        .group_by("GEOGRAPHY_CODE", "bucket")
+        .agg(pl.col("_count").sum())
+    )
+    wide = grouped.pivot(on="bucket", index="GEOGRAPHY_CODE", values="_count").rename(
+        {"GEOGRAPHY_CODE": "lsoa21"}
+    )
+
+    # A bucket with no people in an LSOA is absent from the long rows, so the
+    # pivot leaves a null; treat it as 0 before normalising.
+    wide = wide.with_columns(pl.col(OUTPUT_BUCKETS).fill_null(0.0))
+
+    # Normalize so each row sums to exactly 100%, then round with the
+    # largest-remainder method to preserve the sum (independent rounding of 7
+    # values can drift +/-0.3).
+    row_total = sum(pl.col(c) for c in OUTPUT_BUCKETS)
+    wide = wide.with_columns(
+        [(pl.col(c) / row_total * 100.0).alias(c) for c in OUTPUT_BUCKETS]
+    )
+    wide = wide.with_columns([pl.col(c).round(1).alias(c) for c in OUTPUT_BUCKETS])
+    rounded_sum = sum(pl.col(c) for c in OUTPUT_BUCKETS)
+    residual = (100.0 - rounded_sum).round(1)
+    largest_col = pl.concat_list(OUTPUT_BUCKETS).list.arg_max()
+    wide = wide.with_columns(
+        [
+            pl.when(largest_col == i)
+            .then(pl.col(c) + residual)
+            .otherwise(pl.col(c))
+            .alias(c)
+            for i, c in enumerate(OUTPUT_BUCKETS)
+        ]
+    )
+
+    rename_map = {col: f"% {col}" for col in OUTPUT_BUCKETS}
+    return wide.rename(rename_map).select(
+        "lsoa21", *[f"% {c}" for c in OUTPUT_BUCKETS]
+    )
+
+
+def download_and_convert(output_path: Path) -> None:
+    print("Downloading Census 2021 highest qualification (TS067) by LSOA from NOMIS...")
+    df = download_nomis_csv(BASE_URL)
+    print(f"Total rows: {df.height}")
+
+    # Filter to England only (E-prefixed LSOA codes); the merge joins on the
+    # English postcode universe and the LSOA coverage check is England-wide.
+    df = df.filter(pl.col("GEOGRAPHY_CODE").str.starts_with("E"))
+
+    wide = _qualification_percentages(df)
+
+    print(f"England LSOAs: {wide.height}")
+    if wide.height != ENGLAND_LSOA_COUNT_2021:
+        raise ValueError(
+            f"Expected {ENGLAND_LSOA_COUNT_2021} England LSOAs, "
+            f"got {wide.height}: truncated NOMIS download?"
+        )
+    print(f"Columns: {wide.columns}")
+    deg = wide["% Degree or higher"]
+    print(f"% Degree or higher: {deg.min()}% - {deg.max()}% (mean {deg.mean():.1f}%)")
+
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    wide.write_parquet(output_path, compression="zstd")
+    print(f"Saved to {output_path}")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Download Census 2021 highest level of qualification (TS067) by LSOA"
+    )
+    parser.add_argument(
+        "--output", type=Path, required=True, help="Output parquet file path"
+    )
+    args = parser.parse_args()
+    download_and_convert(args.output)
+
+
+if __name__ == "__main__":
+    main()

pipeline/download/tenure.py

@@ -0,0 +1,179 @@
+"""Download Census 2021 household tenure (TS054) by LSOA.
+
+Downloads the household-tenure breakdown (TS054, classification C2021_TENURE_9)
+from the NOMIS API at LSOA 2021 granularity and folds the 8 detailed leaf
+categories into our 3 output buckets — Owner occupied / Social rent /
+Private rent — emitting one row per LSOA with the percentage of households in
+each. The three buckets sum to 100%, so downstream they render as a
+composition/ratio (like the ethnicity and qualifications stacked bars) AND each
+percentage is independently filterable.
+
+The 3-bucket grouping follows ONS's standard tenure summary:
+  * Owner occupied = Owns outright + Owns with a mortgage/loan + Shared ownership
+  * Social rent    = Rents from council/LA + Other social rented
+  * Private rent    = Private landlord/letting agency + Other private + Lives rent free
+(Shared ownership is part-owned and rolls into owner-occupied; "lives rent free"
+rolls into private rent, mirroring ONS's "Private rented or lives rent free".)
+
+NOTE this table counts HOUSEHOLDS (not usual residents). The join key downstream
+(merge.py) is `lsoa21`, the same key used for ethnicity, qualifications,
+median age, and IoD.
+
+Source: NOMIS (ONS Census 2021 — TS054 dataset, NM_2072_1)
+License: Open Government Licence v3.0
+"""
+
+import argparse
+from pathlib import Path
+
+import polars as pl
+
+from pipeline.utils import ENGLAND_LSOA_COUNT_2021, download_nomis_csv
+
+pl.Config.set_tbl_cols(-1)
+
+# NOMIS API: Census 2021 TS054 (household tenure) by LSOA 2021 (TYPE151).
+# c2021_tenure_9=1..8 selects the 8 substantive leaf categories (excluding the
+# aggregates 0/1001-1004/9996/9997, whose components we sum ourselves).
+# measures=20100 selects the count.
+BASE_URL = (
+    "https://www.nomisweb.co.uk/api/v01/dataset/NM_2072_1.data.csv"
+    "?date=latest"
+    "&geography=TYPE151"
+    "&c2021_tenure_9=1,2,3,4,5,6,7,8"
+    "&measures=20100"
+    "&select=GEOGRAPHY_CODE,C2021_TENURE_9_NAME,OBS_VALUE"
+)
+
+# Map each canonical NOMIS C2021_TENURE_9 leaf to our 3 output buckets. Keyed on
+# the exact NOMIS label so a relabelled or missing leaf fails loudly in
+# validation instead of silently dropping households from the denominator.
+TENURE_MAP = {
+    "Owned: Owns outright": "Owner occupied",
+    "Owned: Owns with a mortgage or loan": "Owner occupied",
+    "Shared ownership: Shared ownership": "Owner occupied",
+    "Social rented: Rents from council or Local Authority": "Social rent",
+    "Social rented: Other social rented": "Social rent",
+    "Private rented: Private landlord or letting agency": "Private rent",
+    "Private rented: Other private rented": "Private rent",
+    "Lives rent free": "Private rent",
+}
+
+# Output buckets in a fixed order (own -> social rent -> private rent), so the
+# stacked composition reads left-to-right and the largest-remainder rounding
+# below is deterministic regardless of pivot column ordering.
+OUTPUT_BUCKETS = ["Owner occupied", "Social rent", "Private rent"]
+assert set(TENURE_MAP.values()) == set(OUTPUT_BUCKETS), (
+    "TENURE_MAP values must be exactly the OUTPUT_BUCKETS"
+)
+
+
+def _tenure_percentages(df: pl.DataFrame) -> pl.DataFrame:
+    """Fold the 8 NOMIS tenure leaves into 3-bucket percentages per LSOA.
+
+    `df` is the long-format NOMIS download with columns GEOGRAPHY_CODE,
+    C2021_TENURE_9_NAME (the leaf label) and OBS_VALUE (a household count). A
+    missing, extra, or relabelled leaf would silently change the denominator (the
+    sum over leaves) so we validate the leaf set against TENURE_MAP first and fail
+    otherwise. Returns one row per LSOA with `lsoa21` and `% <bucket>` for each
+    bucket.
+    """
+    found = set(df["C2021_TENURE_9_NAME"].unique().to_list())
+    expected = set(TENURE_MAP)
+    if found != expected:
+        missing = sorted(expected - found)
+        unexpected = sorted(found - expected)
+        raise ValueError(
+            "Census tenure categories do not match the expected NOMIS "
+            "TS054 C2021_TENURE_9 leaf set.\n"
+            f"  expected {len(expected)} categories, found {len(found)}\n"
+            f"  missing:    {missing}\n"
+            f"  unexpected: {unexpected}\n"
+            "Refusing to compute percentages against an unrecognised breakdown."
+        )
+
+    # Map each leaf to its output bucket and sum counts per (LSOA, bucket).
+    # Summing counts (not rounded percentages) keeps the denominator exact.
+    grouped = (
+        df.with_columns(
+            pl.col("C2021_TENURE_9_NAME").replace_strict(TENURE_MAP).alias("bucket"),
+            pl.col("OBS_VALUE").cast(pl.Float64, strict=False).alias("_count"),
+        )
+        .group_by("GEOGRAPHY_CODE", "bucket")
+        .agg(pl.col("_count").sum())
+    )
+    wide = grouped.pivot(on="bucket", index="GEOGRAPHY_CODE", values="_count").rename(
+        {"GEOGRAPHY_CODE": "lsoa21"}
+    )
+
+    # A bucket with no households in an LSOA is absent from the long rows, so the
+    # pivot leaves a null; treat it as 0 before normalising.
+    wide = wide.with_columns(pl.col(OUTPUT_BUCKETS).fill_null(0.0))
+
+    # Normalize so each row sums to exactly 100%, then round with the
+    # largest-remainder method to preserve the sum (independent rounding of 3
+    # values can drift +/-0.1).
+    row_total = sum(pl.col(c) for c in OUTPUT_BUCKETS)
+    wide = wide.with_columns(
+        [(pl.col(c) / row_total * 100.0).alias(c) for c in OUTPUT_BUCKETS]
+    )
+    wide = wide.with_columns([pl.col(c).round(1).alias(c) for c in OUTPUT_BUCKETS])
+    rounded_sum = sum(pl.col(c) for c in OUTPUT_BUCKETS)
+    residual = (100.0 - rounded_sum).round(1)
+    largest_col = pl.concat_list(OUTPUT_BUCKETS).list.arg_max()
+    wide = wide.with_columns(
+        [
+            pl.when(largest_col == i)
+            .then(pl.col(c) + residual)
+            .otherwise(pl.col(c))
+            .alias(c)
+            for i, c in enumerate(OUTPUT_BUCKETS)
+        ]
+    )
+
+    rename_map = {col: f"% {col}" for col in OUTPUT_BUCKETS}
+    return wide.rename(rename_map).select(
+        "lsoa21", *[f"% {c}" for c in OUTPUT_BUCKETS]
+    )
+
+
+def download_and_convert(output_path: Path) -> None:
+    print("Downloading Census 2021 household tenure (TS054) by LSOA from NOMIS...")
+    df = download_nomis_csv(BASE_URL)
+    print(f"Total rows: {df.height}")
+
+    # Filter to England only (E-prefixed LSOA codes); the merge joins on the
+    # English postcode universe and the LSOA coverage check is England-wide.
+    df = df.filter(pl.col("GEOGRAPHY_CODE").str.starts_with("E"))
+
+    wide = _tenure_percentages(df)
+
+    print(f"England LSOAs: {wide.height}")
+    if wide.height != ENGLAND_LSOA_COUNT_2021:
+        raise ValueError(
+            f"Expected {ENGLAND_LSOA_COUNT_2021} England LSOAs, "
+            f"got {wide.height}: truncated NOMIS download?"
+        )
+    print(f"Columns: {wide.columns}")
+    for bucket in OUTPUT_BUCKETS:
+        col = wide[f"% {bucket}"]
+        print(f"% {bucket}: {col.min()}% - {col.max()}% (mean {col.mean():.1f}%)")
+
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    wide.write_parquet(output_path, compression="zstd")
+    print(f"Saved to {output_path}")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Download Census 2021 household tenure (TS054) by LSOA"
+    )
+    parser.add_argument(
+        "--output", type=Path, required=True, help="Output parquet file path"
+    )
+    args = parser.parse_args()
+    download_and_convert(args.output)
+
+
+if __name__ == "__main__":
+    main()

pipeline/download/test_development_sites.py

@@ -0,0 +1,165 @@
+import polars as pl
+
+from pipeline.download.development_sites import (
+    SCHEMA,
+    _has_dwellings,
+    _is_residential,
+    _point_lonlat,
+    _valid_lonlat,
+    brownfield_to_frame,
+    homes_england_to_frame,
+)
+
+
+def test_point_lonlat_parses_wkt():
+    assert _point_lonlat({"point": "POINT(-0.1278 51.5074)"}) == (-0.1278, 51.5074)
+    # Tolerates extra spacing and uppercase.
+    assert _point_lonlat({"point": "POINT ( -1.5 53.8 )"}) == (-1.5, 53.8)
+
+
+def test_point_lonlat_falls_back_to_columns():
+    assert _point_lonlat({"longitude": "-2.0", "latitude": "52.4"}) == (-2.0, 52.4)
+
+
+def test_point_lonlat_rejects_out_of_range():
+    # A WKT written "lat lon" would place lat (51.5) into the lon slot -> rejected.
+    assert _point_lonlat({"point": "POINT(51.5 -0.12)"}) is None
+    assert _point_lonlat({"point": "garbage"}) is None
+    assert _point_lonlat({}) is None
+
+
+def test_valid_lonlat_bounds():
+    assert _valid_lonlat(-0.1, 51.5)
+    assert not _valid_lonlat(2.5, 51.5)  # lon east of GB
+    assert not _valid_lonlat(-0.1, 40.0)  # lat south of GB
+
+
+def test_has_dwellings():
+    assert _has_dwellings(0, 5)
+    assert _has_dwellings(3, None)
+    assert not _has_dwellings(None, None)
+    assert not _has_dwellings(0, 0)
+
+
+def test_is_residential():
+    assert _is_residential("Residential", None)
+    assert _is_residential("Mixed Use", None)
+    assert _is_residential(None, 12)
+    assert not _is_residential("Employment", None)
+    assert not _is_residential(None, None)
+    assert not _is_residential("Industrial", 0)
+
+
+def test_brownfield_to_frame_normalises_and_filters():
+    raw = pl.DataFrame(
+        {
+            "point": [
+                "POINT(-0.1 51.5)",  # kept
+                "POINT(-1.2 52.6)",  # dropped: no dwellings
+                "POINT(-2.0 53.0)",  # dropped: end-date set (built out)
+                "POINT(-3.0 54.0)",  # kept
+            ],
+            "site-address": ["10 High St", "Old Yard", "Mill Site", "Dock Road"],
+            "minimum-net-dwellings": ["5", "0", "10", None],
+            "maximum-net-dwellings": ["15", "0", "10", "40"],
+            "planning-permission-status": [
+                "permissioned",
+                "not-permissioned",
+                "permissioned",
+                "pending-decision",
+            ],
+            "planning-permission-type": ["full", "", "outline", None],
+            "planning-permission-date": ["2023-01-01", None, "2022-05-01", None],
+            "hectares": ["0.5", "0.2", "1.0", "2.0"],
+            "end-date": [None, None, "2021-06-01", ""],
+            "entity": [1700000, 1700001, 1700002, 1700003],
+            "organisation": [
+                "London Borough of Lewisham",
+                "Org B",
+                "Org C",
+                "London Borough of Newham",
+            ],
+        }
+    )
+    frame = brownfield_to_frame(raw)
+
+    assert frame.columns == list(SCHEMA.keys())
+    assert frame.schema["min_dwellings"] == pl.Int32
+    assert frame.schema["lat"] == pl.Float64
+    # Only the first and last rows survive (dwellings present, not built out).
+    assert frame.height == 2
+    assert frame["source"].to_list() == ["brownfield", "brownfield"]
+    first = frame.row(0, named=True)
+    assert first["lat"] == 51.5
+    assert first["lon"] == -0.1
+    assert first["min_dwellings"] == 5
+    assert first["max_dwellings"] == 15
+    assert first["name"] == "10 High St"
+    assert first["local_authority"] == "London Borough of Lewisham"
+    # Per-site Planning Data entity page, not the unreliable site-plan-url.
+    assert first["url"] == "https://www.planning.data.gov.uk/entity/1700000"
+    last = frame.row(1, named=True)
+    assert last["min_dwellings"] is None
+    assert last["max_dwellings"] == 40
+    assert last["url"] == "https://www.planning.data.gov.uk/entity/1700003"
+
+
+def test_brownfield_empty_input_yields_typed_empty_frame():
+    raw = pl.DataFrame(
+        {
+            "point": [],
+            "minimum-net-dwellings": [],
+            "maximum-net-dwellings": [],
+            "end-date": [],
+        }
+    )
+    frame = brownfield_to_frame(raw)
+    assert frame.height == 0
+    assert frame.columns == list(SCHEMA.keys())
+
+
+def test_homes_england_to_frame_uses_centroid_and_capacity():
+    features = [
+        {
+            "geometry": {
+                "type": "Polygon",
+                "coordinates": [[[-0.2, 51.4], [0.0, 51.4], [0.0, 51.6], [-0.2, 51.6], [-0.2, 51.4]]],
+            },
+            # Real Land Hub field names (Housing_Capacity arrives as a float string).
+            "properties": {
+                "Parcel_Name": "South West Of Old Park Mound, Telford, TF3 4AW",
+                "Proposed_Use": "Residential",
+                "Housing_Capacity": "67.0",
+                "Planning_Status": "Detailed Application Submitted",
+                "Local_Authority": "Telford and Wrekin",
+                "Gross_Area__Acres_": "2.8602",
+                "Site_Reference": "1494",
+            },
+        },
+        {
+            # Non-residential, no capacity -> dropped.
+            "geometry": {
+                "type": "Polygon",
+                "coordinates": [[[-1.0, 52.0], [-0.9, 52.0], [-0.9, 52.1], [-1.0, 52.1], [-1.0, 52.0]]],
+            },
+            "properties": {"Proposed_Use": "Employment"},
+        },
+    ]
+    frame = homes_england_to_frame(features)
+
+    assert frame.height == 1
+    row = frame.row(0, named=True)
+    assert row["source"] == "homes-england"
+    # Real name, not the bare Site_Reference fallback.
+    assert row["name"] == "South West Of Old Park Mound, Telford, TF3 4AW"
+    assert row["max_dwellings"] == 67
+    assert row["min_dwellings"] is None
+    assert row["local_authority"] == "Telford and Wrekin"
+    # Acres converted to hectares: 2.8602 * 0.404686 ≈ 1.1575.
+    assert row["hectares"] == 1.1575
+    # No stable per-site URL on the Land Hub.
+    assert row["url"] is None
+    # Centroid of the unit-ish box.
+    assert abs(row["lon"] - (-0.1)) < 1e-6
+    assert abs(row["lat"] - 51.5) < 1e-6
+    assert frame.columns == list(SCHEMA.keys())

pipeline/download/test_education.py

@@ -0,0 +1,107 @@
+import polars as pl
+import pytest
+
+from pipeline.download.education import (
+    BAND_MAP,
+    OUTPUT_BUCKETS,
+    _qualification_percentages,
+)
+
+
+def _long_rows(geo: str, counts: dict[str, int]) -> list[dict]:
+    """Build NOMIS-shaped long rows for one LSOA from {band_label: count}.
+
+    NOMIS emits a 0-count row when an LSOA has none in a band, so bands not
+    given default to 0 to mirror that.
+    """
+    return [
+        {
+            "GEOGRAPHY_CODE": geo,
+            "C2021_HIQUAL_8_NAME": label,
+            "OBS_VALUE": counts.get(label, 0),
+        }
+        for label in BAND_MAP
+    ]
+
+
+def test_qualification_percentages_keyed_by_lsoa_with_seven_buckets():
+    df = pl.DataFrame(
+        _long_rows(
+            "E01000001",
+            {
+                "No qualifications": 10,
+                "Level 2 qualifications": 20,
+                "Level 3 qualifications": 20,
+                "Level 4 qualifications or above": 50,
+            },
+        )
+    )
+
+    result = _qualification_percentages(df)
+
+    assert result.columns[0] == "lsoa21"
+    assert set(result.columns) == {"lsoa21", *(f"% {b}" for b in OUTPUT_BUCKETS)}
+    row = result.filter(pl.col("lsoa21") == "E01000001").to_dicts()[0]
+    assert row["% Degree or higher"] == 50.0
+    assert row["% No qualifications"] == 10.0
+    assert row["% Good GCSEs"] == 20.0
+    assert row["% A-levels"] == 20.0
+    # Percentages always sum to exactly 100 (largest-remainder rounding).
+    assert round(sum(row[f"% {b}"] for b in OUTPUT_BUCKETS), 1) == 100.0
+
+
+def test_qualification_colloquial_band_mapping():
+    """ONS jargon bands fold into the colloquial buckets, not 'Level N' names."""
+    df = pl.DataFrame(
+        _long_rows(
+            "E01000002",
+            {
+                "Level 1 and entry level qualifications": 40,
+                "Apprenticeship": 60,
+            },
+        )
+    )
+
+    row = _qualification_percentages(df).to_dicts()[0]
+    assert row["% Some GCSEs"] == 40.0
+    assert row["% Apprenticeship"] == 60.0
+    # No raw ONS "Level N" column names leak through.
+    assert not any("Level" in c for c in row)
+
+
+def test_qualification_percentages_independent_per_lsoa():
+    df = pl.concat(
+        [
+            pl.DataFrame(_long_rows("E01000010", {"Level 4 qualifications or above": 100})),
+            pl.DataFrame(_long_rows("E01000011", {"No qualifications": 100})),
+        ]
+    )
+
+    result = _qualification_percentages(df).sort("lsoa21")
+    assert result["% Degree or higher"].to_list() == [100.0, 0.0]
+    assert result["% No qualifications"].to_list() == [0.0, 100.0]
+
+
+def test_qualification_percentages_rejects_unexpected_band():
+    rows = _long_rows("E01000003", {"Level 4 qualifications or above": 10})
+    rows.append(
+        {
+            "GEOGRAPHY_CODE": "E01000003",
+            "C2021_HIQUAL_8_NAME": "Level 5 brand new census band",
+            "OBS_VALUE": 5,
+        }
+    )
+
+    with pytest.raises(ValueError, match="do not match the expected"):
+        _qualification_percentages(pl.DataFrame(rows))
+
+
+def test_qualification_percentages_rejects_missing_band():
+    rows = [
+        r
+        for r in _long_rows("E01000004", {"Level 4 qualifications or above": 10})
+        if r["C2021_HIQUAL_8_NAME"] != "Apprenticeship"
+    ]
+
+    with pytest.raises(ValueError, match="missing"):
+        _qualification_percentages(pl.DataFrame(rows))

pipeline/download/test_tenure.py

@@ -0,0 +1,121 @@
+import polars as pl
+import pytest
+
+from pipeline.download.tenure import OUTPUT_BUCKETS, TENURE_MAP, _tenure_percentages
+
+
+def _long_rows(geo: str, counts: dict[str, int]) -> list[dict]:
+    """Build NOMIS-shaped long rows for one LSOA from {leaf_label: count}.
+
+    Every one of the 8 leaf categories must be present in the download (NOMIS
+    emits a 0-count row when an LSOA has none), so categories not given default
+    to 0 to mirror that.
+    """
+    return [
+        {
+            "GEOGRAPHY_CODE": geo,
+            "C2021_TENURE_9_NAME": label,
+            "OBS_VALUE": counts.get(label, 0),
+        }
+        for label in TENURE_MAP
+    ]
+
+
+def test_tenure_percentages_keyed_by_lsoa_with_three_buckets():
+    df = pl.DataFrame(
+        _long_rows(
+            "E01000001",
+            {
+                "Owned: Owns outright": 40,
+                "Owned: Owns with a mortgage or loan": 15,
+                "Shared ownership: Shared ownership": 5,
+                "Social rented: Rents from council or Local Authority": 15,
+                "Social rented: Other social rented": 5,
+                "Private rented: Private landlord or letting agency": 18,
+                "Private rented: Other private rented": 1,
+                "Lives rent free": 1,
+            },
+        )
+    )
+
+    result = _tenure_percentages(df)
+
+    assert result.columns[0] == "lsoa21"
+    assert set(result.columns) == {"lsoa21", *(f"% {b}" for b in OUTPUT_BUCKETS)}
+    row = result.filter(pl.col("lsoa21") == "E01000001").to_dicts()[0]
+    # Owner occupied = outright + mortgage + shared ownership = 60.
+    assert row["% Owner occupied"] == 60.0
+    # Social rent = council + other social = 20.
+    assert row["% Social rent"] == 20.0
+    # Private rent = private landlord + other private + lives rent free = 20.
+    assert row["% Private rent"] == 20.0
+    # Percentages always sum to exactly 100 (largest-remainder rounding).
+    assert round(sum(row[f"% {b}"] for b in OUTPUT_BUCKETS), 1) == 100.0
+
+
+def test_tenure_shared_ownership_rolls_into_owner_occupied():
+    """Shared ownership is part-owned, so it counts as owner-occupied, not rent."""
+    df = pl.DataFrame(_long_rows("E01000002", {"Shared ownership: Shared ownership": 100}))
+
+    row = _tenure_percentages(df).to_dicts()[0]
+
+    assert row["% Owner occupied"] == 100.0
+    assert row["% Social rent"] == 0.0
+    assert row["% Private rent"] == 0.0
+
+
+def test_tenure_lives_rent_free_rolls_into_private_rent():
+    """'Lives rent free' folds into private rent (ONS 'private rented or rent free')."""
+    df = pl.DataFrame(_long_rows("E01000003", {"Lives rent free": 100}))
+
+    row = _tenure_percentages(df).to_dicts()[0]
+
+    assert row["% Private rent"] == 100.0
+    assert row["% Owner occupied"] == 0.0
+    assert row["% Social rent"] == 0.0
+
+
+def test_tenure_percentages_independent_per_lsoa():
+    """Two LSOAs get independent profiles — the LSOA granularity is the point."""
+    df = pl.concat(
+        [
+            pl.DataFrame(_long_rows("E01000010", {"Owned: Owns outright": 100})),
+            pl.DataFrame(
+                _long_rows(
+                    "E01000011",
+                    {"Social rented: Rents from council or Local Authority": 100},
+                )
+            ),
+        ]
+    )
+
+    result = _tenure_percentages(df).sort("lsoa21")
+
+    assert result["% Owner occupied"].to_list() == [100.0, 0.0]
+    assert result["% Social rent"].to_list() == [0.0, 100.0]
+
+
+def test_tenure_percentages_rejects_unexpected_category():
+    rows = _long_rows("E01000004", {"Owned: Owns outright": 10})
+    rows.append(
+        {
+            "GEOGRAPHY_CODE": "E01000004",
+            "C2021_TENURE_9_NAME": "A Brand New Census Tenure Category",
+            "OBS_VALUE": 5,
+        }
+    )
+
+    with pytest.raises(ValueError, match="do not match the expected"):
+        _tenure_percentages(pl.DataFrame(rows))
+
+
+def test_tenure_percentages_rejects_missing_category():
+    # Drop one leaf entirely: its households would vanish from the denominator.
+    rows = [
+        r
+        for r in _long_rows("E01000005", {"Owned: Owns outright": 10})
+        if r["C2021_TENURE_9_NAME"] != "Lives rent free"
+    ]
+
+    with pytest.raises(ValueError, match="missing"):
+        _tenure_percentages(pl.DataFrame(rows))