233 lines
8.6 KiB
Python
233 lines
8.6 KiB
Python
import argparse
|
|
import polars as pl
|
|
from pathlib import Path
|
|
|
|
|
|
def _build_wide(
|
|
epc_pp_path: Path,
|
|
arcgis_path: Path,
|
|
iod_path: Path,
|
|
poi_proximity_path: Path,
|
|
journey_times_path: Path,
|
|
ethnicity_path: Path,
|
|
crime_path: Path ,
|
|
noise_path: Path ,
|
|
ofsted_path: Path,
|
|
broadband_path: Path,
|
|
) -> pl.DataFrame:
|
|
"""Build the wide dataframe by joining epc_pp with all auxiliary data."""
|
|
print("Scanning epc_pp...")
|
|
wide = pl.scan_parquet(epc_pp_path)
|
|
|
|
# GPS coordinates + LSOA from ArcGIS
|
|
print("Joining GPS coordinates...")
|
|
arcgis = pl.scan_parquet(arcgis_path).select(
|
|
pl.col("pcds").alias("postcode"),
|
|
"lat",
|
|
pl.col("long").alias("lon"),
|
|
"lsoa21",
|
|
"oa21",
|
|
)
|
|
wide = wide.join(arcgis, on="postcode", how="inner")
|
|
|
|
print("Joining journey times...")
|
|
journey_times = pl.scan_parquet(journey_times_path).select(
|
|
"postcode",
|
|
"public_transport_easy_minutes",
|
|
"public_transport_quick_minutes",
|
|
"cycling_minutes",
|
|
)
|
|
wide = wide.join(journey_times, on="postcode", how="left")
|
|
|
|
print("Joining IoD scores...")
|
|
iod = pl.scan_parquet(iod_path)
|
|
wide = wide.join(iod, left_on="lsoa21", right_on="LSOA code (2021)", how="left")
|
|
|
|
# Ethnicity by local authority
|
|
print("Joining ethnicity data...")
|
|
ethnicity = pl.scan_parquet(ethnicity_path)
|
|
wide = wide.join(
|
|
ethnicity,
|
|
left_on="Local Authority District code (2024)",
|
|
right_on="Geography_code",
|
|
how="left",
|
|
)
|
|
|
|
# Crime stats by LSOA
|
|
print("Joining crime data...")
|
|
crime = pl.scan_parquet(crime_path)
|
|
wide = wide.join(crime, left_on="lsoa21", right_on="LSOA code", how="left")
|
|
|
|
print("Joining POI proximity counts...")
|
|
poi_counts = pl.scan_parquet(poi_proximity_path)
|
|
wide = wide.join(poi_counts, on="postcode", how="left")
|
|
|
|
noise = pl.scan_parquet(noise_path).select("postcode", "road_noise_lden_db")
|
|
wide = wide.join(noise, on="postcode", how="left")
|
|
|
|
print("Joining Ofsted school ratings...")
|
|
ofsted = (
|
|
pl.scan_parquet(ofsted_path)
|
|
.filter(pl.col("Overall effectiveness").is_in(["1", "2", "3", "4"]))
|
|
.select(
|
|
pl.col("Postcode").alias("ofsted_postcode"),
|
|
pl.col("Overall effectiveness").cast(pl.UInt8).alias("ofsted_rating"),
|
|
)
|
|
.group_by("ofsted_postcode")
|
|
.agg(pl.col("ofsted_rating").mean().round(1).alias("ofsted_avg_rating"))
|
|
)
|
|
wide = wide.join(ofsted, left_on="postcode", right_on="ofsted_postcode", how="left")
|
|
|
|
print("Joining broadband performance...")
|
|
broadband = pl.scan_parquet(broadband_path).select(
|
|
"output_area",
|
|
pl.col("Average max download speed (Mbit/s) for lines >=900Mbit/s").alias("broadband_max_download_900plus"),
|
|
pl.col("Average max download speed (Mbit/s) for lines 100<300Mbit/s").alias("broadband_max_download_100_300"),
|
|
pl.col("Average max download speed (Mbit/s) for lines 30<100Mbit/s").alias("broadband_max_download_30_100"),
|
|
pl.col("Average max upload speed (Mbit/s) for lines >=900Mbit/s").alias("broadband_max_upload_900plus"),
|
|
)
|
|
wide = wide.join(broadband, left_on="oa21", right_on="output_area", how="left")
|
|
|
|
# Convert construction_age_band to numeric year
|
|
wide = wide.with_columns(
|
|
pl.col("construction_age_band")
|
|
.str.replace("England and Wales: ", "")
|
|
.str.replace(" onwards", "")
|
|
.str.extract(r"(\d{4})", 1)
|
|
.cast(pl.UInt16, strict=False)
|
|
.alias("construction_age_band"),
|
|
)
|
|
|
|
wide = wide.with_columns(
|
|
pl.when(pl.col("pp_property_type") == pl.col("built_form"))
|
|
.then(pl.col("pp_property_type"))
|
|
.otherwise(
|
|
pl.concat_str(
|
|
[pl.col("pp_property_type"), pl.lit("/"), pl.col("built_form")]
|
|
)
|
|
)
|
|
.alias("property_type_built_form")
|
|
)
|
|
|
|
wide = (
|
|
wide.filter(pl.col("total_floor_area") > 0).with_columns(
|
|
(pl.col("latest_price") / pl.col("total_floor_area"))
|
|
.round(0)
|
|
.cast(pl.Int32)
|
|
.alias("Price per sqm"),
|
|
)
|
|
.drop(
|
|
"date_of_transfer",
|
|
"inspection_date",
|
|
"floor_height",
|
|
"LSOA name (2021)",
|
|
"Local Authority District code (2024)",
|
|
"Local Authority District name (2024)",
|
|
"Wider Barriers Sub-domain Score",
|
|
"Geographical Barriers Sub-domain Score",
|
|
"Adult Skills Sub-domain Score",
|
|
"Children and Young People Sub-domain Score",
|
|
"Income Deprivation Affecting Older People (IDAOPI) Score (rate)",
|
|
"Income Deprivation Affecting Children Index (IDACI) Score (rate)",
|
|
"Barriers to Housing and Services Score",
|
|
"lsoa21",
|
|
"oa21",
|
|
"pp_property_type",
|
|
"built_form",
|
|
)
|
|
.rename(
|
|
{
|
|
"construction_age_band": "Approximate construction age",
|
|
"pp_address": "Address per Property Register",
|
|
"epc_address": "Address per EPC",
|
|
"postcode": "Postcode",
|
|
"duration": "Leashold/Freehold",
|
|
"current_energy_rating": "Current energy rating",
|
|
"potential_energy_rating": "Potential energy rating",
|
|
"total_floor_area": "Total floor area (sqm)",
|
|
"epc_property_type": "Property type",
|
|
"property_type_built_form": "Property type/built form",
|
|
"restaurants_2km": "Restaurants within 2km",
|
|
"groceries_2km": "Groceries within 2km",
|
|
"parks_2km": "Parks within 2km",
|
|
"public_transport_2km": "Public transport within 2km",
|
|
"latest_price": "Last known price",
|
|
"number_habitable_rooms": "Rooms (including bedrooms & bathrooms)",
|
|
"road_noise_lden_db": "Road noise Lden (dB)",
|
|
"ofsted_avg_rating": "Ofsted avg rating (1=Outstanding, 4=Inadequate)"
|
|
"broadband_max_download_900plus": "Broadband download speed 900+ Mbps",
|
|
"broadband_max_download_100_300": "Broadband download speed 100-300 Mbps",
|
|
"broadband_max_download_30_100": "Broadband download speed 30-100 Mbps",
|
|
"broadband_max_upload_900plus": "Broadband upload speed 900+ Mbps",
|
|
}
|
|
)
|
|
)
|
|
|
|
print("Collecting with streaming engine...")
|
|
return wide.collect(engine="streaming")
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Build wide property dataframe with all joins"
|
|
)
|
|
parser.add_argument(
|
|
"--epc-pp", type=Path, required=True, help="EPC-Price Paid joined parquet file"
|
|
)
|
|
parser.add_argument(
|
|
"--arcgis", type=Path, required=True, help="ArcGIS postcode data parquet file"
|
|
)
|
|
parser.add_argument(
|
|
"--iod", type=Path, required=True, help="Index of Deprivation parquet file (optional)"
|
|
)
|
|
parser.add_argument(
|
|
"--poi-proximity",
|
|
type=Path,
|
|
help="POI proximity counts parquet file (optional)",
|
|
)
|
|
parser.add_argument(
|
|
"--journey-times", required=True, type=Path, help="Journey times parquet file (optional)"
|
|
)
|
|
parser.add_argument(
|
|
"--ethnicity", type=Path, required=True, help="Ethnicity by local authority parquet file (optional)"
|
|
)
|
|
parser.add_argument(
|
|
"--crime", type=Path, required=True, help="Crime by LSOA parquet file (optional)"
|
|
)
|
|
parser.add_argument(
|
|
"--noise", type=Path, required=True, help="Road noise by postcode parquet file"
|
|
)
|
|
parser.add_argument(
|
|
"--ofsted", type=Path, required=True, help="Ofsted school inspection parquet file"
|
|
)
|
|
parser.add_argument(
|
|
"--broadband", type=Path, required=True, help="Broadband performance by output area parquet file"
|
|
)
|
|
parser.add_argument(
|
|
"--output", type=Path, required=True, help="Output parquet file path"
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
wide = _build_wide(
|
|
epc_pp_path=args.epc_pp,
|
|
arcgis_path=args.arcgis,
|
|
iod_path=args.iod,
|
|
poi_proximity_path=args.poi_proximity,
|
|
journey_times_path=args.journey_times,
|
|
ethnicity_path=args.ethnicity,
|
|
crime_path=args.crime,
|
|
noise_path=args.noise,
|
|
ofsted_path=args.ofsted,
|
|
broadband_path=args.broadband,
|
|
)
|
|
|
|
print(f"Columns: {wide.columns}")
|
|
print(f"Rows: {wide.height}")
|
|
|
|
wide.write_parquet(args.output)
|
|
size_mb = args.output.stat().st_size / (1024 * 1024)
|
|
|
|
print(f"Wrote {args.output} ({size_mb:.1f} MB)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|