feat: deprecate academies_master_list.csv file (#1386)

* feat: deprecate academies_master_list.csv file

- remove dependency on this file from `build_academy_data()`
- add fields to input fields for AAR data to cover those missing
- update phase-mapping to accommodate potentially fewer sources

* fix: amend phase-type mappings

updated to derive from GIAS values only.

* fix: remove gias-all-links from academies

- no longer read `gias_all_links.csv` during academies processing
  - Trust data sourced from `aar_cs.csv`
- amended join to CFO data based on Co. No. rather than URN
- `Group UID` no longer written to DB (unused and `DEFAULT NULL`)

Author: PsypherPunk

data-pipeline/src/pipeline/database.py

@@ -141,15 +141,15 @@ def insert_schools_and_trusts_and_local_authorities(
     projections = {
         "URN": "URN",
         "EstablishmentName": "SchoolName",
-        "Companies House Number": "TrustCompanyNumber",
-        "Group Name": "TrustName",
+        "Company Registration Number": "TrustCompanyNumber",
+        "Company_Name": "TrustName",
         "Federation Lead School URN": "FederationLeadURN",
         "Federation Name": "FederationLeadName",
         "LA Code": "LACode",
         "LA Name": "LAName",
         "London Weighting": "LondonWeighting",
         "Finance Type": "FinanceType",
-        "Overall Phase": "OverallPhase",
+        "SchoolPhaseType": "OverallPhase",
         "TypeOfEstablishment (name)": "SchoolType",
         "Has Sixth Form": "HasSixthForm",
         "Has Nursery": "HasNursery",
@@ -176,19 +176,18 @@ def insert_schools_and_trusts_and_local_authorities(
     logger.info(f"Wrote {len(write_frame)} rows to school {run_type} - {year}")
 
     trust_projections = {
-        "Group Name": "TrustName",
-        "Group UID": "UID",
+        "Company_Name": "TrustName",
         "CFO name": "CFOName",
         "CFO email": "CFOEmail",
         "OpenDate": "OpenDate",
-        "Companies House Number": "CompanyNumber",
+        "Company Registration Number": "CompanyNumber",
     }
 
     trusts = (
-        df[~df["Companies House Number"].isna()]
+        df[~df["Company Registration Number"].isna()]
         .reset_index()
-        .sort_values(by=["Companies House Number", "OpenDate"], ascending=False)
-        .groupby(["Companies House Number"])
+        .sort_values(by=["Company Registration Number", "OpenDate"], ascending=False)
+        .groupby(["Company Registration Number"])
         .first()
         .reset_index()
         .rename(columns=trust_projections)[[*trust_projections.values()]]

data-pipeline/src/pipeline/input_schemas.py

@@ -57,27 +57,6 @@
     "LinkEstablishedDate": "string",
 }
 
-academy_master_list_index_col = "LA Establishment Number"
-academy_master_list = {
-    "Company Registration Number": "string",
-    "Incorporation Date": "string",
-    "Academy Trust UPIN": "Int64",
-    "Academy Trust Name": "string",
-    "Academy Name": "string",
-    "Academy UPIN": "Int64",
-    "Trust Type": "string",
-    "LA Establishment Number": "string",
-    "Date Opened": "string",
-    "Type of Provision - Phase": "string",
-    "Regional School Commissioner": "string",
-    "Valid From": "string",
-    "Valid to": "string",
-    "Territory": "string",
-    "Academy Status": "string",
-    "Academy Trust Status": "string",
-    "Number of Academies in Trust": "Int64",
-}
-
 maintained_schools_master_list_index_col = "URN"
 maintained_schools_master_list = {
     "URN": "Int64",
@@ -321,11 +300,14 @@
     "BNCH21606 (Agency supply teaching staff)": "float",
     "BNCH21403 (Energy)": "float",
     "BNCH21402 (Water and sewerage)": "float",
+    "Valid To": "string",
 }
 
 aar_central_services_index_col = "Lead_UPIN"
 aar_central_services = {
     "Lead_UPIN": "Int64",
+    "Company_Number": "string",
+    "Company_Name": "string",
     "BNCH11110T (EFA Revenue Grants)": "float",
     "BNCH11131 (DfE Family Revenue Grants)": "float",
     "BNCH11141 (SEN)": "float",

data-pipeline/src/pipeline/main.py

@@ -122,6 +122,7 @@ def pre_process_academy_ar(run_type, year) -> tuple[pd.DataFrame, pd.DataFrame]:
     academy_ar_data = get_blob(
         raw_container, f"{run_type}/{year}/aar.csv", encoding="utf-8"
     )
+
     aar = prepare_aar_data(academy_ar_data, year)
 
     write_blob(
@@ -187,17 +188,7 @@ def pre_process_academies_data(run_type, year, data_ref) -> pd.DataFrame:
     logger.info("Building Academy Set")
     schools, census, sen, cdc, aar, ks2, ks4, cfo, central_services = data_ref
 
-    academies_data = get_blob(
-        raw_container, f"{run_type}/{year}/academy_master_list.csv", encoding="utf-8"
-    )
-
-    links_data = get_blob(
-        raw_container, f"{run_type}/{year}/gias_all_links.csv", encoding="cp1252"
-    )
-
     academies = build_academy_data(
-        academies_data,
-        links_data,
         year,
         schools,
         census,
@@ -209,6 +200,7 @@ def pre_process_academies_data(run_type, year, data_ref) -> pd.DataFrame:
         cfo,
         central_services,
     )
+
     write_blob(
         "pre-processed",
         f"{run_type}/{year}/academies.parquet",

data-pipeline/src/pipeline/maintained_schools.py

@@ -107,7 +107,8 @@ def map_school_type_attrs(maintained_schools: pd.DataFrame) -> pd.DataFrame:
     maintained_schools["Finance Type"] = "Maintained"
     maintained_schools["SchoolPhaseType"] = maintained_schools.apply(
         lambda df: mappings.map_phase_type(
-            df["TypeOfEstablishment (code)"], df["PhaseOfEducation (code)"], df["Overall Phase"]
+            establishment_code=df["TypeOfEstablishment (code)"],
+            phase_code=df["PhaseOfEducation (code)"],
         ),
         axis=1,
     )
@@ -169,6 +170,7 @@ def calc_rag_cost_series(
 
     return maintained_schools
 
+
 # net catering cost, not net catering income
 def calc_catering_net_costs(maintained_schools: pd.DataFrame) -> pd.DataFrame:
     maintained_schools["Catering staff and supplies_Net Costs"] = (

data-pipeline/src/pipeline/mappings.py

@@ -14,46 +14,47 @@ def map_ofsted_rating(rating: str):
             return rating
 
 
-def map_phase_type(establishment_code: int, phase_code: int, provision: str):
+def _map_secondary_phases(establishment_code: int) -> str:
+    match establishment_code:
+        case 40:
+            return "University Technical College"
+        case _:
+            return "Secondary"
+
 
-    if (pd.isna(establishment_code) or pd.isna(phase_code) or pd.isna(provision)):
-        return
-    
+def _map_not_applicable_phases(establishment_code: int) -> str:
     match establishment_code:
-        case 33 | 36 | 44:
+        case 6:
+            return "University Technical College"
+        case 7 | 12 | 33 | 36 | 44:
             return "Special"
+        case 14:
+            return "Pupil Referral Unit"
         case 38 | 42 | 43:
             return "Alternative Provision"
-        case 40:
-            return "University Technical College"
-        case 39 | 45 | 46:
-            return "Post-16"
+        case _:
+            return "Unknown"
 
+
+def map_phase_type(
+    establishment_code: int,
+    phase_code: int,
+) -> str:
     match phase_code:
-        case 7:
-            return "All-through"
+        case 0:
+            return _map_not_applicable_phases(establishment_code)
+        case 1:
+            return "Nursery"
         case 2 | 3:
             return "Primary"
         case 4 | 5:
-            return "Secondary"
-
-    match provision.lower():
-        case "16 plus" | "post-16":
+            return _map_secondary_phases(establishment_code)
+        case 6:
             return "Post-16"
-        case "secondary":
-            return "Secondary"
-        case "special":
-            return "Special"
-        case "primary":
-            return "Primary"
-        case "all through" | "all-through":
+        case 7:
             return "All-through"
-        case "nursery":
-            return "Nursery"
-        case "pupil referral unit":
-            return "Pupil Referral Unit"
         case _:
-            return "Other"
+            return "Unknown"
 
 
 def map_block_age(block_age: str):
@@ -236,8 +237,8 @@ def map_cost_series(category_name, df, basis):
 
     for sub_category in sub_categories:
         df[sub_category + "_Per Unit"] = df[sub_category].fillna(0) / basis
-        df[sub_category + "_Per Unit"].replace(
-            [np.inf, -np.inf, np.nan], 0, inplace=True
+        df[sub_category + "_Per Unit"] = df[sub_category + "_Per Unit"].replace(
+            [np.inf, -np.inf, np.nan], 0
         )
 
     return df