"""Rebuild the parking feasibility audit from archived inputs. No rates or models. Run from the repository root. Source bytes are verified before use; reports are separate from the approved evidence release. Python stdlib + openpyxl only. """ import collections import csv import gzip import hashlib import io import json import pathlib import zipfile import xml.etree.ElementTree as ET import openpyxl ROOT = pathlib.Path(__file__).resolve().parents[1] OUT = ROOT / 'public/research/parking/v1' RELEASE = 'lens-2026-10-09-cbec43fe76c4' AUDITED = '2026-10-10' YEARS = list(range(2021, 2026)) def sha(data): return hashlib.sha256(data).hexdigest() def read_source(source): data = (ROOT / 'public' / source['archivePath'].lstrip('/')).read_bytes() if source.get('archiveSha256'): assert sha(data) == source['archiveSha256'], source['id'] + ' archive changed' raw = gzip.decompress(data) if source['archivePath'].endswith('.gz') else data assert sha(raw) == source['sha256'], source['id'] + ' source changed' return raw def parked_state(row): """Canonical recoded field is authoritative, including its unknown values.""" key = 'vehicle_manoeuvre' if 'vehicle_manoeuvre' in row else 'vehicle_manoeuvre_historic' code = row.get(key, '') if code == '2': return 'parked' valid = set(range(1, 16)) | {19, 20} if key == 'vehicle_manoeuvre' else set(range(1, 19)) if code in {str(i) for i in valid}: return 'other' return 'unknown' def collision_state(states): if 'parked' in states: return 'parked' return 'other' if states and 'unknown' not in states else 'unknown' def ods_tables(path): ns = {'t': 'urn:oasis:names:tc:opendocument:xmlns:table:1.0'} root = ET.fromstring(zipfile.ZipFile(path).read('content.xml')) return {t.get('{'+ns['t']+'}name'): [ [' '.join(c.itertext()).strip() for c in r if 'table-cell' in c.tag] for r in t.findall('t:table-row', ns)] for t in root.findall('.//t:table', ns)} def write_csv(name, rows): with (OUT / name).open('w', newline='') as f: w = csv.DictWriter(f, fieldnames=list(rows[0])) w.writeheader() w.writerows(rows) def build(): initial = json.loads((ROOT / 'db/stats19/initial.json').read_text()) geography = json.loads((ROOT / 'db/stats19/geography-candidate.json').read_text()) sources = {s['id']: s for s in initial['sources'] + geography['sources']} selected = {k: sources[k] for k in ['collision', 'vehicle', 'guide', 'geography-aadf']} raw = {k: read_source(s) for k, s in selected.items()} guide = openpyxl.load_workbook(io.BytesIO(raw['guide']), read_only=True, data_only=True) code_rows = list(guide['2024_code_list'].values) for field in ['vehicle_manoeuvre', 'vehicle_manoeuvre_historic']: assert any(r[1] == field and r[2] == 2 and r[3] == 'Parked' for r in code_rows) assert any(r[1] == 'vehicle_manoeuvre' and r[2] == 20 and r[3] == 'Parking' for r in code_rows) collisions = {} duplicate_collisions = 0 for r in csv.DictReader(io.StringIO(raw['collision'].decode('utf-8-sig'))): key = r['collision_index'] if key in collisions: duplicate_collisions += 1 collisions[key] = r states = collections.defaultdict(list) vehicle_keys = set() duplicate_vehicles = orphan_vehicles = disagreements = 0 vehicles_by_year = collections.Counter() unknown_by_year = collections.Counter() for r in csv.DictReader(io.StringIO(raw['vehicle'].decode('utf-8-sig'))): year = int(r['collision_year']) key = (r['collision_index'], r['vehicle_reference']) duplicate_vehicles += key in vehicle_keys vehicle_keys.add(key) vehicles_by_year[year] += 1 state = parked_state(r) unknown_by_year[year] += state == 'unknown' old, new = r.get('vehicle_manoeuvre_historic', ''), r.get('vehicle_manoeuvre', '') if old not in ['', '-1', '99'] and new not in ['', '-1', '99']: disagreements += (old == '2') != (new == '2') if r['collision_index'] not in collisions: orphan_vehicles += 1 else: states[r['collision_index']].append(state) annual = {y: collections.Counter() for y in YEARS} by_area = collections.defaultdict(collections.Counter) by_lsoa = collections.defaultdict(collections.Counter) missing_vehicle_collisions = vehicle_count_mismatches = 0 for key, r in collisions.items(): y = int(r['collision_year']) state = collision_state(states[key]) missing_vehicle_collisions += not states[key] vehicle_count_mismatches += len(states[key]) != int(r['number_of_vehicles']) annual[y]['gbCollisions'] += 1 annual[y]['gbParkedCollisions'] += state == 'parked' annual[y]['gbUnknownCollisions'] += state == 'unknown' area = r['local_authority_highway_current'] if not area.startswith('E'): continue lsoa = r['lsoa_of_accident_location'] for a in [annual[y], by_area[(area, y)]]: a['collisions'] += 1 a['parkedCollisions'] += state == 'parked' a['otherCollisions'] += state == 'other' a['unknownCollisions'] += state == 'unknown' a['lsoaMissing'] += not lsoa.startswith('E01') if lsoa.startswith('E01'): by_lsoa[(lsoa, y)]['collisions'] += 1 by_lsoa[(lsoa, y)]['parkedCollisions'] += state == 'parked' by_lsoa[(lsoa, y)]['unknownCollisions'] += state == 'unknown' # Count presence is not road-network completeness or a mileage denominator. traffic = collections.defaultdict(collections.Counter) traffic_keys = set() duplicate_traffic = 0 z = zipfile.ZipFile(io.BytesIO(raw['geography-aadf'])) with z.open('dft_traffic_counts_aadf.csv') as f: for r in csv.DictReader(io.TextIOWrapper(f, encoding='utf-8-sig')): y = int(r['year']) area = r['local_authority_code'] if y not in YEARS or not area.startswith('E'): continue key = (r['count_point_id'], y) duplicate_traffic += key in traffic_keys traffic_keys.add(key) a = traffic[(area, y)] major = r['road_type'] == 'Major' method = r['estimation_method'] a['majorRows' if major else 'minorRows'] += 1 a[('major' if major else 'minor') + ('Counted' if method == 'Counted' else 'Estimated' if method == 'Estimated' else 'MethodUnknown')] += 1 try: a['nonpositiveFlow'] += float(r['all_motor_vehicles']) <= 0 except ValueError: a['missingFlow'] += 1 boundary = geography['geography']['boundaryAsset'] boundary_bytes = (ROOT / 'public' / boundary['path'].lstrip('/')).read_bytes() assert sha(boundary_bytes) == boundary['sha256'] features = json.loads(boundary_bytes)['features'] areas = sorted([{'id': f['properties']['CTYUA25CD'], 'name': f['properties']['CTYUA25NM']} for f in features if f['properties']['CTYUA25CD'].startswith('E')], key=lambda x: x['name']) area_ids = {a['id'] for a in areas} area_rows = [] for y in YEARS: for area in areas: c, t = by_area[(area['id'], y)], traffic[(area['id'], y)] area_rows.append({'areaId': area['id'], 'areaName': area['name'], 'year': y, **{k: c[k] for k in ['collisions', 'parkedCollisions', 'otherCollisions', 'unknownCollisions', 'lsoaMissing']}, **{k: t[k] for k in ['majorRows', 'minorRows', 'majorCounted', 'majorEstimated', 'majorMethodUnknown', 'minorCounted', 'minorEstimated', 'minorMethodUnknown', 'nonpositiveFlow', 'missingFlow']}, 'parkingShare': None, 'residentialMotorMiles': None, 'frequencyPer100kMiles': None, 'modelReady': False}) orphan_area = [{'areaId': a, 'year': y, **dict(v)} for (a, y), v in by_area.items() if a not in area_ids and v['collisions']] unmatched_traffic = [{'areaId': a, 'year': y, **dict(v)} for (a, y), v in traffic.items() if a not in area_ids] annual_rows = [] for y in YEARS: a = annual[y] expected = next(r for r in initial['quality']['byYear'] if r['year'] == y) assert a['gbCollisions'] == expected['collisions'] assert vehicles_by_year[y] == expected['vehicles'] assert a['collisions'] == a['parkedCollisions'] + a['otherCollisions'] + a['unknownCollisions'] annual_rows.append({'year': y, **dict(a), 'gbVehicles': vehicles_by_year[y], 'gbUnknownVehicleManoeuvres': unknown_by_year[y], 'englishLsoaCodesWithCollisions': len({a for a, yr in by_lsoa if yr == y}), 'englishMinorCountRows': sum(v['minorRows'] for (a, yr), v in traffic.items() if yr == y), 'englishMinorCountedRows': sum(v['minorCounted'] for (a, yr), v in traffic.items() if yr == y)}) extra_sources = [] for name, title, url, description in [ ('nts0908', 'NTS0908 overnight parking', 'https://assets.publishing.service.gov.uk/media/6a9ed429474b8101ece46446/nts0908.ods', 'Observed survey parking use. England and urban/rural groups, not observed LSOA shares. Questions are not asked every year. The 2024 urban/rural classification is not directly comparable with earlier groups. The workbook excludes unreported parking locations.'), ('ehs-da2201', 'English Housing Survey DA2201', 'https://assets.publishing.service.gov.uk/media/6a4d8ccb895317eadc35450c/DA2201_Parking_and_mains_gas_-_dwellings.ods', 'Parking provision by dwelling type; latest worksheet 2024. Best available provision is not actual use or capacity for every household vehicle. Suppressed cells remain unavailable.')]: p = OUT / 'inputs' / (name + '.ods') extra_sources.append({'id': name, 'title': title, 'assetUrl': url, 'archivePath': '/research/parking/v1/inputs/' + p.name, 'sha256': sha(p.read_bytes()), 'bytes': p.stat().st_size, 'accessedAt': AUDITED, 'state': 'Archived and inspected', 'detail': description, 'worksheets': list(ods_tables(p))}) results = {'schemaVersion': 1, 'version': 'v1', 'phase': 1, 'auditedAt': AUDITED, 'baseEvidenceRelease': RELEASE, 'status': 'Feasibility audited; parking model not ready', 'scope': 'England. Coding audit covers all reported injury collisions, not an exposure-matched motor-only residential cohort.', 'years': YEARS, 'analysisRun': False, 'modelReady': False, 'deciles': [], 'rate': None, 'effect': None, 'boundary': boundary, 'areas': areas, 'annual': annual_rows, 'coverage': area_rows, 'quality': {'duplicateCollisions': duplicate_collisions, 'duplicateVehicles': duplicate_vehicles, 'orphanVehicles': orphan_vehicles, 'missingVehicleCollisions': missing_vehicle_collisions, 'vehicleCountMismatches': vehicle_count_mismatches, 'parkedCodingDisagreements': disagreements, 'duplicateTrafficPointYears': duplicate_traffic, 'unmappedCollisionAreas': orphan_area, 'unmatchedTrafficBoundaryCodes': unmatched_traffic}, 'sources': [{**s, 'state': 'Archived and audited'} for s in selected.values()] + extra_sources, 'build': {'script': '/research/parking/v1/audit-parking-phase1.py', 'scriptSha256': sha(pathlib.Path(__file__).read_bytes()), 'sourceBaselineCommit': '26d8df258ccc6966a684149770507aef8ac0d6b3'}, 'limits': ['No parking deciles, rate estimates or causal conclusions have been calculated.', 'STATS19 excludes damage-only incidents and does not establish insurance claims or fault.', 'A parked vehicle can be recorded without physical contact. Involvement is not causation.', 'Unknown manoeuvres are a separate classification, never a confirmed absence of parked vehicles.', 'LSOA codes are recorded-location codes. Their boundary vintage has not been verified for a Census 2021 join.', 'Map shows 2025 county/unitary boundaries, not LSOA parking estimates. Collision counts use current highway codes.', 'Traffic rows are count-point presence. Minor-road sampling does not establish complete residential-road exposure.', 'Historical traffic authority codes are not silently crosswalked to current boundaries.', 'Housing-derived parking scores need independent validation and cannot establish an independent parking effect beyond their ingredients.']} assert not any([duplicate_collisions, duplicate_vehicles, orphan_vehicles, missing_vehicle_collisions, vehicle_count_mismatches, disagreements, duplicate_traffic]) OUT.mkdir(parents=True, exist_ok=True) (OUT / 'audit.json').write_text(json.dumps(results, indent=2) + '\n') (OUT / 'audit-parking-phase1.py').write_bytes(pathlib.Path(__file__).read_bytes()) write_csv('annual-coding-audit.csv', annual_rows) write_csv('authority-coverage.csv', area_rows) write_csv('lsoa-coding-coverage.csv', [{'recordedLsoa': a, 'year': y, **{k: v[k] for k in ['collisions', 'parkedCollisions', 'unknownCollisions']}, 'boundaryVintage': 'Unverified for Census join', 'parkingShare': None, 'residentialMotorMiles': None} for (a, y), v in sorted(by_lsoa.items())]) print(json.dumps({'annual': annual_rows, 'authorities': len(areas), 'quality': results['quality']}, indent=2)) if __name__ == '__main__': build()