#!/usr/bin/env python3 """Build aggregate care-estate evidence from a retained CSV. No row data is exported.""" import argparse import csv import hashlib import json from collections import Counter from pathlib import Path def build_baseline(path): required = {'Location ID', 'Care home?', 'Dormant (Y/N)', 'Primary ID (Dual registration locations)', 'Care homes beds', 'Provider ID', 'Service user band - Learning disabilities or autistic spectrum disorder'} with path.open(encoding='utf-8-sig', newline='') as handle: reader = csv.DictReader(handle) if not required.issubset(reader.fieldnames or []): raise ValueError('The required source columns are missing.') rows = list(reader) columns = len(reader.fieldnames) ids = [r['Location ID'].strip() for r in rows] if len(set(ids)) != len(ids) or not all(ids): raise ValueError('Source location identifiers must be present and unique.') homes = [r for r in rows if r['Care home?'].strip() == 'Y'] dormant = [r for r in homes if r['Dormant (Y/N)'].strip() == 'Y'] nondormant = [r for r in homes if r['Dormant (Y/N)'].strip() == 'N'] if len(dormant) + len(nondormant) != len(homes): raise ValueError('Care-home dormancy flags require review.') secondary = [r for r in nondormant if r['Primary ID (Dual registration locations)'].strip() == 'N'] cohort = [r for r in nondormant if r['Primary ID (Dual registration locations)'].strip() != 'N'] providers = Counter(r['Provider ID'].strip() for r in cohort) if '' in providers: raise ValueError('A provider identifier is missing.') for r in cohort: if not r['Care homes beds'].strip().isdigit() or int(r['Care homes beds']) <= 0: raise ValueError('Care-home bed capacity requires review.') small = [r for r in cohort if int(r['Care homes beds']) < 10] small_multi = sum(providers[r['Provider ID'].strip()] >= 2 for r in small) small_band = sum(r['Service user band - Learning disabilities or autistic spectrum disorder'].strip() == 'Y' for r in small) multihome = sum(v for v in providers.values() if v >= 2) bins = [] for label, lo, hi in [('1–9 beds', 1, 9), ('10–29 beds', 10, 29), ('30–59 beds', 30, 59), ('60 or more beds', 60, None)]: count = sum(int(r['Care homes beds']) >= lo and (hi is None or int(r['Care homes beds']) <= hi) for r in cohort) bins.append({'label': label, 'locations': count, 'sharePercent': round(count / len(cohort) * 100, 1)}) under30 = bins[0]['locations'] + bins[1]['locations'] digest = hashlib.sha256() with path.open('rb') as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b''): digest.update(chunk) return { 'title': 'Care Circle care-estate baseline', 'sourcePeriod': 'CQC 4–5 August 2026, as labelled in the retained intelligence master', 'analysisDate': '2026-10-19', 'geography': 'CQC-derived England care-home registrations in the supplied file', 'unit': 'Registration record after the stated filters; not a verified unique building', 'inputSha256': digest.hexdigest(), 'population': {'inputRows': len(rows), 'inputColumns': columns, 'careHomeRows': len(homes), 'dormantExcluded': len(dormant), 'nonDormantRows': len(nondormant), 'secondaryDualExcluded': len(secondary), 'workingCohort': len(cohort), 'providerIds': len(providers)}, 'capacityBands': bins, 'findings': [ {'id': 'under30', 'label': 'Registered for fewer than 30 beds', 'numerator': under30, 'denominator': len(cohort), 'percent': round(under30 / len(cohort) * 100, 1)}, {'id': 'smallMulti', 'label': 'Sub-10-bed locations in provider IDs with two or more cohort care-home locations', 'numerator': small_multi, 'denominator': len(small), 'percent': round(small_multi / len(small) * 100, 1)}, {'id': 'smallBand', 'label': 'Sub-10-bed locations with learning-disability/autism service-user flag', 'numerator': small_band, 'denominator': len(small), 'percent': round(small_band / len(small) * 100, 1)}, {'id': 'multiLocations', 'label': 'Locations in provider IDs with two or more cohort care-home locations', 'numerator': multihome, 'denominator': len(cohort), 'percent': round(multihome / len(cohort) * 100, 1)}], 'method': [ "Select Care home? = Y, then Dormant (Y/N) = N.", "Exclude Primary ID (Dual registration locations) = N. This field is a primary/secondary flag, not a join identifier.", 'Count unique Location IDs. Retain other records without claiming complete physical-building deduplication.', 'Group by Provider ID within the filtered cohort. Multiple locations means at least two care-home registration records in that cohort.', 'Use registered bed capacity to define size bands. Capacity does not measure occupancy or occupied bed days.', 'Treat service-user flags as registered service scope, not a diagnosis, resident census or measured energy requirement.', 'Use original full-precision integer counts; round displayed percentages to one decimal place.'], 'exclusionsAndLimits': [ 'An August baseline analysed on 19 October; not a current national census or a time series of changes.', 'Provider IDs are regulatory entities, not verified ultimate operating groups or property owners.', 'No EPC, UPRN, Land Registry, commercial score or opportunity classification determines the published findings.', 'No source contact details, personal names, site records or row-level identifiers are included in this aggregate.', 'No energy quantities, bills, actual savings, project performance or current installed equipment are inferred.'], 'attribution': 'Care Circle analysis using CQC-derived information. Contains public sector information licensed under the Open Government Licence v3.0.', 'sourceUrl': 'https://www.cqc.org.uk/about-us/transparency/using-cqc-data', 'licenceUrl': 'https://www.nationalarchives.gov.uk/doc/open-government-licence/version/3/' } if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('input', type=Path) parser.add_argument('output', type=Path) args = parser.parse_args() result = build_baseline(args.input) args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2) + '\n') print(json.dumps({'workingCohort': result['population']['workingCohort'], 'findings': result['findings']}, ensure_ascii=False))