"""Normalise the M-DCPS School Climate Survey dashboard workbooks (students, parents, staff; 2018-19 → 2024-25) into one per-item table per source and a text-matched item crosswalk. Each workbook has a different layout: a long results sheet with different column names per year, or (2021-22) only pivot tables whose underlying records live in the workbook's pivot cache (xl/pivotCache/*.xml), which is read directly. Percentages arrive as fractions in some years and as 0–100 in others; everything is stored as 0–100. Conventions (revised after the 2026-09-16 review): * The item table holds only survey items: the Q00 (respondent counts) and Q99 (district / overall) lines are dropped BEFORE anything is aggregated, so school-level extras never read a summary row. * Percent agreeing = agree / (agree + don't know + disagree) where the file carries all three and they sum further from 100 than rounding allows (tolerance derived from each file's precision: not-answered respondents, or the publisher's 2024-25 numerator/denominator mismatch — e.g. Blue Lakes K-8 Q05 1.23 + 0.05 + 0.02 → 94.7%, which the same row's response counts confirm). Files without a don't-know column keep agreement as published. Where response counts exist (students 2024-25) the shares are computed from the counts. A value still outside [0, 100] is invalid and dropped (counted in the log). Three published shares totalling zero are missing response distributions, never zero agreement scores. * A school's respondent count n is the item rows' count where the file carries one per row, else the Q00 row's count (2018-19 parents and staff put Ncount only on Q00). Suppression: the file's Less_10 / Low_Ncnt flag, or n < 10 — the district withholds such schools and reports zeros for them. * Recommend / grade A-B / good-education extras are per-school constants (the first non-null value on any of the school's rows — the 2021-22 pivot caches carry them on the Q-00 line only); count-based files (parents 2022-23) divide the counts by the school's respondent total. * Response rate = n / eligible × 100 where the workbook carries an eligible count; rates above 100 are invalid (dropped). The students' eligible column changes definition across years (all-grade FTE vs grades 4+), so the loader recomputes the students' rate against the same year's October membership in grades 4–12 instead. Crosswalk: items are matched across years on wording (Jaccard ≥ 0.6 on stemmed content words after dropping lead-ins such as "I feel", "there is", "at my school"), one item per year per canonical wording; a hand-reviewed alias table (data/cities/miami/reference/survey_item_aliases.csv) merges rewordings the text rule misses. Domain and reverse-coding are properties of the canonical item (the same across years); `core` = asked in every year the respondent group was surveyed. Outputs (committed): data/cities/miami/reference/parsed/survey/.csv school_number, item_no, n_responses, pct_agree, pct_dk, pct_disagree, suppressed data/cities/miami/reference/parsed/survey/.school.csv school_number, respondent, year, n_responses, n_eligible, response_rate, recommend_yes_pct, grade_ab_pct, good_education_pct data/cities/miami/reference/survey_items.csv respondent, year, item_no, item_text, canonical, domain, match_jaccard, alias_group, reverse, core, excluded_reason (excluded_reason: a hand-reviewed publisher anomaly from survey_item_exclusions.csv; the item is not core that year) Run: python -m pipeline_miami.parse.climate_survey """ from __future__ import annotations import re import zipfile from pathlib import Path from xml.etree import ElementTree as ET import pandas as pd from pipeline_miami.verify import common OUT_DIR = common.REPO_ROOT / "data" / "cities" / "miami" / "reference" / "parsed" / "survey" ITEMS_OUT = common.REPO_ROOT / "data" / "cities" / "miami" / "reference" / "survey_items.csv" ALIASES = common.REPO_ROOT / "data" / "cities" / "miami" / "reference" / "survey_item_aliases.csv" EXCLUSIONS = common.REPO_ROOT / "data" / "cities" / "miami" / "reference" / "survey_item_exclusions.csv" # hand-reviewed publisher anomalies NS = {"m": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"} # Per-file column specs. Keys: sheet ("pivot" = read the pivot cache), loc, item, text, agree, dk, dis, n, # n_elig, category, suppressed, recommend (list of columns summed), grade_ab (list), good_ed (list), # counts (dict of response-count columns: with an "agree" entry all three shares come from the counts; # with dk/dis only, those two are counts over the `n` column), extras_counts (the extras are counts to be # divided by this column), label_offset (the file stores each item's values on the row labelled that many # numbers LOWER: the values on row Q-n belong to item Q-(n+offset); the label text, category and domain are # taken from the row that carries the item's own label; the first `offset` items have no values that year and # the last `offset` rows' values belong to no labelled item and are dropped). SPECS: dict[str, dict] = { "students_2018-19": dict(sheet="2019 Stud TBL", loc="2019 Loc_No", item="Qno", text="Srvy Question No and Label", agree="Agree", dk="DK", dis="Disagree", n="Ncount", category="Category"), "students_2019-20": dict(sheet="STUD 2020", loc="2020 Loc No", item="Qno", text="SURVEY QUESTION", agree="Agree", dk="DK", dis="Disagree", n="Ncount", category="CATEGORY"), "students_2020-21": dict(sheet="STUD2021", loc="LOC", item="QNO", text="QUESTIONLABEL", agree="Q5to35_Pct_Agree", dk="Q5to35_Pct_DK", dis="Q5to35_Pct_Disagree", n="N_Srvy_Responses", n_elig="Mbrship_2021", category="CATEGORY", recommend=["Recommend_pct"], grade_ab=["Grd_AB_Pct"], good_ed=["GoodEdu_Agree"]), "students_2021-22": dict(sheet="pivot", loc="LocNo", item="Q_No", text="QuestionItem", agree="Agreement", dk="DontKnow", dis="Disagreement", n="Ncount", n_elig="Grades412Count", category="Construct", recommend=["Recommend"], grade_ab=["OvrlGradeA", "OvrlGradeB"]), "students_2022-23": dict(sheet="2023_Stud_Data_12June", loc="LOC", item="SSQNO", text="SSQLabel", agree="Pct_Agree", dk="Pct_DK", dis="Pct_Disagree", n="TTL", n_elig="FTEFeb2023", recommend=["REC_DefinitelyYes", "REC_MaybeYes"], grade_ab=["grade_A", "grade_B"], good_ed=["GoodED_Yes"]), "students_2023-24": dict(sheet="2024_Student_Data", loc="LOC", item="SSQNO", text="SSQLabel", agree="Pct_Agree", dk="PCT_DK", dis="Pct_Disagree", n="TTL", n_elig="Grd_4_Plus", suppressed="Low_Ncnt", recommend=["Rec_DefinitelyYes", "Rec_MaybeYes"], grade_ab=["Grade_A", "Grade_B"], good_ed=["GoodEd_Yes"]), "students_2024-25": dict(sheet="2025_Students_Data", loc="LOC", item="SSQNO", text="SSQLabel", agree="Agree_PCT", dk="DK_PCT", dis="Disagree_PCT", n="TTL", n_elig="Grades_4_Plus", suppressed="Less_10", recommend=["Rec_Yes"], grade_ab=["Grade_A", "Grade_B"], good_ed=["GoodEd_Yes"], counts=dict(agree=["N_SA", "N_AG"], dk=["N_DK"], dis=["N_DA", "N_SD"])), "parents_2018-19": dict(sheet="Parent Raw Data", loc="Loc No", item="Q No", text="Survey Question", agree="Schl Agreement", dk="Schl DK", dis="Schl Disagreement", n="Parent Ncount", category="Category Name"), "parents_2020-21": dict(sheet="21_Par_Climate_Data", loc="Loc", item="Qno", text="Qlabel", agree="Pct_Agree", dk="Pct_DK", dis="Pct_Disagree", n="TTL_Parent_Responses", n_elig="TTL_Students", category="Category Nmae"), "parents_2021-22": dict(sheet="pivot", loc="LocNo", item="Qno", text="QuestionLabel", agree="TTLPctAgreement", dk="PctDK", dis="TTLPctDisagreement", n="Ncount", n_elig="OFFICIALFTEFEBRUARY2022", recommend=["RecommendYes"], grade_ab=["OvrlGRDA", "OvrlGRDB"], good_ed=["GoodEducationYes"]), "parents_2022-23": dict(sheet="Parent2023_Data_Table", loc="LOC", item="PQNO", text="PQ_LABEL", agree="Agreement", n="Total", counts=dict(dk=["DK"], dis=["DA", "SD"]), extras_counts="Total", recommend=["DefinitelyYes_A", "MaybeYes"], grade_ab=["A", "B"], good_ed=["DefinitelyYes"]), "parents_2023-24": dict(sheet="2024_Parent_Data", loc="LOC", item="QNO", text="Question", agree="Pct_AG", dk="PCT_DK", dis="PCT_DA", n="TTL", recommend=["Recommend_YES"], grade_ab=["GradeA", "GradeB"], good_ed=["Good_Ed_Yes"]), "parents_2024-25": dict(sheet="2025_Parents_Data", loc="LOC", item="QNO", text="Question", agree="Pct_AG", dk="PCT_DK", dis="PCT_DA", n="TTL", recommend=["Recommend_YES"], grade_ab=["GradeA", "GradeB"], good_ed=["Good_Ed_Yes"]), "staff_2019-20": dict(sheet="2020 STAFF DATA", loc="2020 Loc No", item="Q No", text="Survey Question", agree="AGREE", dk="DK", dis="DISAGREE", n="Ncount", category="Category"), "staff_2020-21": dict(sheet="2021 Staff Data", loc="LOC", item=None, text="Teacher Survey Question", agree="Pct Agree", dk="Pct Neutral", dis="Pct Disagree", n="TTL_Responses", n_elig="TTL_Teachers", category="Category Name", recommend=["PCt Recommend YES"], grade_ab=["Pct Overall Grade A", "Pct Overall Grade B"]), # 2026-09-16: the pivot cache's values sit one row above their label (row Q-n holds item Q-(n+1)'s shares; the # district's own pivot sheets render the same shifted district averages, e.g. "substance abuse 58%", "staff morale # high 37%"); measured against the adjacent years, 33 of 35 items are closer to the NEXT label's canonical than to # their own. Q-00 carries no shares (item Q01 has no 2021-22 values) and Q-36's values belong to no label. "staff_2021-22": dict(sheet="pivot", loc="LocNo", item="Qno", text="QLabel", agree="Agrmnt", dk="PctDK", dis="Disgrmnt", n="NoStaffResponses", category="CategoryName", recommend=["RecmndDefinitelyYes", "RecmndMaybeYes"], grade_ab=["OvlGradeA", "OvlGradeB"], label_offset=1), "staff_2022-23": dict(sheet="2023_Staff_Data_June21", loc="LOC", item="QNO", text="Qlabel", agree="Agree_Pct", n="TTL", counts=dict(dk=["DK"], dis=["DA", "SD"]), n_elig="FT_Instructional", recommend=["PCTYes", "PCTMaybe"], grade_ab=["PCTGradeA", "PCTGradeB"]), "staff_2023-24": dict(sheet="2024_Staff_Data", loc="Loc", item="QNO", text="Question", agree="Pct_Agree", dk="Pct_DK", dis="Pct_Disagree", n="No Responses", recommend=["PCT_Recom-Yes"], grade_ab=["PCT_Gr-A", "PCT_Gr-B"]), "staff_2024-25": dict(sheet="Staff_Data_2025", loc="LOC", item="QNO", text="Question", agree="PCT_Agree", dk="PCT_DK", dis="PCT_Disagree", n="TTL", suppressed="Less_10", recommend=["PCT_REC_YES"], grade_ab=["PCT_Gr_A", "PCT_Gr_B"]), } # Fetched workbooks that are NOT parsed or served, with the measurement behind the decision. A retired file's # parsed tables are removed and scripts/loaders/miami/climate_survey.ts deletes any cells still served from it. RETIRED_SPECS: dict[str, str] = { "staff_2018-19": "Retired 2026-09-16 (after the favorable-response scoring change). The 2018-19 staff workbook publishes " "agreement and disagreement only, with no don't-know column, and the two sum to a median of 83% " "(10th percentile 50%): a response category is missing from the file. On the negative-worded core " "items the published disagreement is 25% against 65–70% in every later year while agreement is stable " "(18% vs 20%), so the file's disagreement column is not the measure the later files publish. Scored by " "disagreement the 2018-19 staff safety, resources and teaching composites sit 13–20 pp below 2019-20; " "scored by 100 − agreement they would count the hidden category as favorable. Neither is comparable " "with the later years, so the staff series starts in 2019-20.", } DOMAIN_OF_CATEGORY = [ (r"safety|secur|order", "safety"), (r"leader", "leadership"), (r"teach|learn|instruct", "teaching"), (r"resourc|support|quality", "resources"), ] SUMMARY_ITEMS = {"Q00", "Q99"} RENORM_TOLERANCE_PP = 1.5 # largest |agree + dk + disagree − 100| three whole-percent roundings can produce (documented reference; the parser derives the tolerance per file from how many share columns are whole percents) RENORM_TOLERANCE_FINE_PP = 0.05 # allowance for columns published with decimals: any larger deviation is a real not-answered share def source_path(key: str) -> Path: return Path(common.latest_source_path(f"mdcps_climate_survey_{key}")) def read_pivot_cache(path: Path) -> pd.DataFrame: """Records of the first pivot cache as a DataFrame (shared items resolved).""" z = zipfile.ZipFile(path) d = ET.fromstring(z.read("xl/pivotCache/pivotCacheDefinition1.xml")) fields = [] for cf in d.find("m:cacheFields", NS): name = cf.get("name") si = cf.find("m:sharedItems", NS) shared = [] if si is not None: for it in si: shared.append(it.get("v")) fields.append((name, shared)) r = ET.fromstring(z.read("xl/pivotCache/pivotCacheRecords1.xml")) rows = [] for rec in r: vals = [] for i, cell in enumerate(rec): tag = cell.tag.split("}")[-1] if tag == "x": idx = int(cell.get("v")) shared = fields[i][1] vals.append(shared[idx] if idx < len(shared) else None) elif tag == "m": vals.append(None) else: vals.append(cell.get("v")) rows.append(vals) return pd.DataFrame(rows, columns=[f[0] for f in fields]) def num(v): if v is None: return None s = str(v).strip() if s in ("", "nan", "None", "---", "*", "**", "N/A", "n/a"): return None try: return float(s.replace(",", "").replace("%", "")) except ValueError: return None def to_pct(series: pd.Series) -> pd.Series: """Fractions → 0–100; already-percent values kept. Decided by the median (a stray count in a percent column must not flip the whole file).""" s = series.map(num) med = s.dropna().median() if s.notna().any() else 0 return s * 100.0 if med is not None and med <= 1.0001 else s def item_no_from_text(t: str) -> str | None: m = re.match(r"\s*Q?-?\s*(\d{1,2})", str(t)) return f"Q{int(m.group(1)):02d}" if m else None def clean_text(t: str) -> str: t = re.sub(r"^\s*Q?-?\s*\d{1,2}\s*[.:\-–]*\s*", "", str(t)).strip() return re.sub(r"\s+", " ", t) LEAD_IN = re.compile(r"^(?:(?:i feel|i think|i believe)(?: that)?\s+|there (?:is|are)(?: an?)?\s+|(?:at|in) (?:my|the|this) school,?\s+|my child'?s school\s+|(?:my|the|this) school\s+|overall,?\s+)+", re.I) STOP = {"the", "a", "an", "my", "at", "in", "of", "to", "is", "are", "and", "or", "this", "that", "i", "s", "child", "childs", "school", "schools", "problem", "problems", "feel", "there", "students", "student", "for", "with", "on", "by", "it", "be", "as", "am", "do", "does", "me", "our", "we", "they"} def stem(w: str) -> str: """Light suffix stripping: -ies → y; -sses → ss; -ing / -ed dropped; -es dropped only after s, x, z, ch, sh (boxes → box, classes → class, but provides → provide); a final -s dropped unless the word ends in -ss.""" if w.endswith("ies") and len(w) >= 5: return w[:-3] + "y" if w.endswith("sses"): return w[:-2] for suf in ("ing", "ed"): if w.endswith(suf) and len(w) - len(suf) >= 3: return w[: -len(suf)] if w.endswith("es") and len(w) >= 5 and (w[-3] in "sxz" or w.endswith(("ches", "shes"))): return w[:-2] if w.endswith("s") and not w.endswith("ss") and len(w) >= 4: return w[:-1] return w def norm_words(t: str) -> set[str]: t = clean_text(t).lower().replace("’", "'") t = LEAD_IN.sub("", t) t = re.sub(r"[^a-z ]+", " ", t) return {stem(w) for w in t.split() if w not in STOP} def school_counts(df: pd.DataFrame, spec: dict, loc: pd.Series, item: pd.Series, col_key: str = "n") -> pd.Series: """A per-school count (respondents, or the eligible population): the item rows' value where the file carries one per row, else the Q00 row's (2018-19 parents/staff Ncount, 2022-23 staff FT_Instructional).""" n = df[spec[col_key]].map(num) if spec.get(col_key) else pd.Series([None] * len(df), index=df.index) frame = pd.DataFrame({"loc": loc, "item": item, "n": n}) items = frame[frame.item.notna() & ~frame.item.isin(SUMMARY_ITEMS)] per_item = items.dropna(subset=["n"]).groupby("loc").n.median() q00 = frame[frame.item == "Q00"].dropna(subset=["n"]).groupby("loc").n.first() return per_item.combine_first(q00) def apply_label_offset(offset: int, loc: pd.Series, item: pd.Series, text: pd.Series, category: pd.Series) -> tuple[pd.Series, pd.Series, pd.Series]: """Re-attach values to the item they belong to when a file stores them `offset` rows above their label. Row Q-n's values become item Q-(n+offset) with that item's own label text and category; summary rows (Q00/Q99) keep their label; values whose target label does not exist in the file are dropped (item None). Rows for the first `offset` items are re-labelled so that the values are not misattributed: the item is still emitted (asked that year) but only from a row with no values — the caller's null handling applies.""" label_text = {} label_cat = {} for it, tx, ct in zip(item, text, category): if isinstance(it, str) and it not in label_text: label_text[it], label_cat[it] = tx, ct new_item, new_text, new_cat = [], [], [] for it, tx, ct in zip(item, text, category): if isinstance(it, str) and re.fullmatch(r"Q\d{2}", it) and it not in SUMMARY_ITEMS: target = f"Q{int(it[1:]) + offset:02d}" if target in label_text: new_item.append(target); new_text.append(label_text[target]); new_cat.append(label_cat[target]) else: new_item.append(None); new_text.append(tx); new_cat.append(ct) # values with no labelled item else: new_item.append(it); new_text.append(tx); new_cat.append(ct) return (pd.Series(new_item, index=item.index), pd.Series(new_text, index=item.index), pd.Series(new_cat, index=item.index)) def parse(key: str, spec: dict) -> tuple[pd.DataFrame, pd.DataFrame, dict]: resp, year = key.split("_") path = source_path(key) if spec["sheet"] == "pivot": df = read_pivot_cache(path) else: df = pd.read_excel(path, sheet_name=spec["sheet"], dtype=str) df.columns = [str(c).strip() for c in df.columns] df = df.rename(columns=lambda c: str(c).strip()) loc = df[spec["loc"]].map(lambda v: common.pad4(str(v).split("-")[0].strip()) if v is not None else None) text = df[spec["text"]].astype(str) item = df[spec["item"]].astype(str).str.strip() if spec.get("item") else text.map(item_no_from_text) item = item.map(lambda s: re.sub(r"^(Q|SSQ|PQ)-?", "Q", str(s).strip().upper()) if s and str(s).strip().upper() not in ("NONE", "NAN", "") else None) item = item.map(lambda s: f"Q{int(s[1:]):02d}" if isinstance(s, str) and re.fullmatch(r"Q\d{1,2}", s) else (s if isinstance(s, str) else None)) category = df[spec["category"]].astype(str).str.strip() if spec.get("category") else pd.Series([None] * len(df), index=df.index) if spec.get("label_offset"): item, text, category = apply_label_offset(int(spec["label_offset"]), loc, item, text, category) n_school = school_counts(df, spec, loc, item) elig_school = pd.DataFrame({"loc": loc, "v": df[spec["n_elig"]].map(num)}).dropna().groupby("loc").v.first() if spec.get("n_elig") else None out = pd.DataFrame({"school_number": loc, "respondent": resp, "year": year, "item_no": item, "item_text": text.map(clean_text)}) out["category"] = category out["n_responses"] = loc.map(n_school) counts = spec.get("counts") or {} if "agree" in counts: parts = {k: sum(df[c].map(num).fillna(0) for c in cols) for k, cols in counts.items()} denom = sum(parts.values()) out["pct_agree"] = (parts["agree"] / denom * 100.0).where(denom > 0) out["pct_dk"] = (parts["dk"] / denom * 100.0).where(denom > 0) out["pct_disagree"] = (parts["dis"] / denom * 100.0).where(denom > 0) else: out["pct_agree"] = to_pct(df[spec["agree"]]) if counts: tot = df[spec["n"]].map(num) out["pct_dk"] = sum(df[c].map(num).fillna(0) for c in counts["dk"]) / tot * 100.0 out["pct_disagree"] = sum(df[c].map(num).fillna(0) for c in counts["dis"]) / tot * 100.0 else: out["pct_dk"] = to_pct(df[spec["dk"]]) if spec.get("dk") else None out["pct_disagree"] = to_pct(df[spec["dis"]]) if spec.get("dis") else None # Percent agreeing among those who answered: renormalise where all three shares exist and their sum is # further from 100 than rounding can explain (three shares rounded to whole percents can sum to 98.5–101.5; # anything beyond that is a different denominator — not-answered respondents, or the publisher's # 2024-25 numerator/denominator mismatch). Rows within rounding are left as published. diag = {"renormalised": 0, "invalid_agree": 0, "extras_invalid": 0, "renorm_tolerance_pp": None, "row_n_under_10": 0, "zero_total_items": 0} if out.pct_dk.notna().any() and out.pct_disagree.notna().any(): three = out[["pct_agree", "pct_dk", "pct_disagree"]].astype(float) s = three.sum(axis=1, min_count=3) # A zero-total distribution contains no usable answers (e.g. staff # 2024-25, school 5007). Keep a genuine zero agreement when another # response category is positive; only an all-zero distribution is missing. zero_total = s.eq(0) diag["zero_total_items"] = int((zero_total & out.item_no.notna() & ~out.item_no.isin(SUMMARY_ITEMS)).sum()) out.loc[zero_total, ["pct_agree", "pct_dk", "pct_disagree"]] = float("nan") # The tolerance follows the file's precision, column by column: each share column published as whole # percents can contribute up to 0.5 pp of rounding to the sum (three such columns: 98.5–101.5); a column # published with decimals contributes only floating-point noise (0.05 pp allowed in all). The 2023-24 # staff file rounds agreement alone, so its tolerance is 0.55 pp, not 1.5 (review 2026-09-16b). n_whole = int(sum(bool(((c.dropna() - c.dropna().round()).abs() < 1e-9).all()) for _, c in three.items() if c.notna().any())) tol = 0.5 * n_whole + (RENORM_TOLERANCE_FINE_PP if n_whole < 3 else 0.0) diag["renorm_tolerance_pp"] = tol ok = s.notna() & (s > 0) & ((s - 100.0).abs() > tol) renorm = three[ok].div(s[ok], axis=0) * 100.0 diag["renormalised"] = int(ok.sum()) out.loc[ok, ["pct_agree", "pct_dk", "pct_disagree"]] = renorm.to_numpy() # Shares are stored to 6 decimals: the pivot caches hold doubles such as 100.00000000000001, which the loader's # [0, 100] range check would otherwise reject as invalid (review 2026-09-16b: 34 staff 2021-22 rows, nine # leadership composites off by up to 2.8 pp). Overshoots within 1e-6 of the bounds are the bounds. for c in ("pct_agree", "pct_dk", "pct_disagree"): v = pd.to_numeric(out[c], errors="coerce").astype(float) v = v.where(~((v > 100.0) & (v < 100.0 + 1e-6)), 100.0).where(~((v < 0.0) & (v > -1e-6)), 0.0) out[c] = v.round(6) bad = out.pct_agree.notna() & ((out.pct_agree < 0.0) | (out.pct_agree > 100.0)) diag["invalid_agree"] = int(bad.sum()) out.loc[bad, ["pct_agree", "pct_dk", "pct_disagree"]] = float("nan") if spec.get("suppressed"): flag = df[spec["suppressed"]].astype(str).str.strip().str.lower() out["suppressed"] = flag.isin(["less10", "less_10", "low", "low_ncnt", "1", "true", "yes", "y", "low_n", "lowncnt", "<10", "less than 10"]) | out.n_responses.map(lambda v: v is not None and v == v and v < 10) else: out["suppressed"] = out.n_responses.map(lambda v: v is not None and v == v and v < 10) # An item row whose own respondent count is under 10 is suppressed too, whatever the school's median (the # publisher's floor applies per published cell; review 2026-09-16b). if spec.get("n"): row_n = df[spec["n"]].map(num) low_row = row_n.notna() & (row_n < 10) & ~out.suppressed.astype(bool) diag["row_n_under_10"] = int(low_row.sum()) out.loc[low_row, "suppressed"] = True out = out[out.school_number.notna() & out.item_no.notna() & ~out.item_no.isin(SUMMARY_ITEMS)] out = out.drop_duplicates(subset=["school_number", "item_no"]) if spec.get("label_offset"): # The first `offset` items were asked (their labels are in the file) but their values are not in the file: # one value-less row per school so the crosswalk still counts the item as asked that year (it is excluded # from that year's composite through survey_item_exclusions.csv, which names the reason). labels = df[spec["item"]].astype(str).str.strip().map(lambda v: re.sub(r"^(Q|SSQ|PQ)-?", "Q", v.upper())) labels = labels.map(lambda v: f"Q{int(v[1:]):02d}" if re.fullmatch(r"Q\d{1,2}", v) else None) lost = sorted(set(labels.dropna()) - set(out.item_no) - SUMMARY_ITEMS) texts = dict(zip(labels, df[spec["text"]].astype(str).map(clean_text))) cats = dict(zip(labels, df[spec["category"]].astype(str).str.strip())) if spec.get("category") else {} blank = pd.DataFrame([{"school_number": sn, "respondent": resp, "year": year, "item_no": it, "item_text": texts[it], "category": cats.get(it), "n_responses": n_school.get(sn), "pct_agree": float("nan"), "pct_dk": float("nan"), "pct_disagree": float("nan"), "suppressed": bool(n_school.get(sn) is not None and n_school.get(sn) < 10)} for it in lost for sn in sorted(out.school_number.unique())]) diag["items_without_values"] = lost out = pd.concat([out, blank], ignore_index=True) if len(blank) else out # School-level extras: per-school constants; the 2021-22 pivot caches carry them on the Q-00 line only, # so every row of the school is scanned and the first non-null value taken (never a summed or maxed count). keep = loc.notna() ex = pd.DataFrame({"school_number": loc[keep]}) ex["n_eligible"] = loc[keep].map(elig_school) if elig_school is not None else None denom_col = spec.get("extras_counts") denom = df.loc[keep, denom_col].map(num) if denom_col else None for name, cols in (("recommend_yes_pct", spec.get("recommend")), ("grade_ab_pct", spec.get("grade_ab")), ("good_education_pct", spec.get("good_ed"))): if cols and denom_col: tot = sum(df.loc[keep, c].map(num).fillna(0) for c in cols) ex[name] = (tot / denom * 100.0).where(denom > 0) elif cols: parts = [to_pct(df.loc[keep, c]) for c in cols] ex[name] = sum(p.fillna(0) for p in parts).where(pd.concat(parts, axis=1).notna().any(axis=1)) else: ex[name] = None ex = ex.groupby("school_number").agg({"n_eligible": "first", "recommend_yes_pct": "first", "grade_ab_pct": "first", "good_education_pct": "first"}).reset_index() ex["n_responses"] = ex.school_number.map(n_school) ex["respondent"], ex["year"] = resp, year ex["response_rate"] = (ex.n_responses / ex.n_eligible * 100.0).where(ex.n_eligible > 0) for c in ("response_rate", "recommend_yes_pct", "grade_ab_pct", "good_education_pct"): badx = ex[c].notna() & ((ex[c] < -1e-9) | (ex[c] > 100.0 + 1e-9)) diag["extras_invalid"] += int(badx.sum()) ex.loc[badx, c] = None ex = ex[["school_number", "respondent", "year", "n_responses", "n_eligible", "response_rate", "recommend_yes_pct", "grade_ab_pct", "good_education_pct"]] return out, ex, diag def domain_from_category(cat: str | None) -> str | None: if not cat or str(cat).lower() in ("nan", "none", ""): return None c = str(cat).lower() for pat, dom in DOMAIN_OF_CATEGORY: if re.search(pat, c): return dom return None KEYWORD_DOMAINS = [(r"safe|secur|clean|violence|gang|bully|drug|alcohol|discipline|rules|law enforcement", "safety"), (r"principal|administrat|ideas|listened|leader|office", "leadership"), (r"teacher|learn|homework|class|instruction|skills|education|standards|prepared", "teaching"), (r"books|equipment|counselor|supplies|resources|food|cafeteria|lunch|transport|device|overload|morale|evaluation|professional|career|job", "resources")] NEGATIVE = re.compile(r"\b(is|are) (a )?problems?\b|violence problem|school violence|gang activity|substance abuse|too many|lack of|overload|overcrowd|deficien|insufficient|not enough|interfere|affects how much", re.I) NOT_NEGATIVE = re.compile(r"free of|keep the school|adequate", re.I) def load_aliases() -> list[tuple[str, str, re.Pattern]]: if not ALIASES.exists(): return [] a = pd.read_csv(ALIASES, dtype=str).fillna("") return [(r.respondent, r.group, re.compile(r.pattern, re.I)) for r in a.itertuples() if r.pattern] def build_crosswalk(items: pd.DataFrame) -> pd.DataFrame: """Canonical item = the same wording across years (Jaccard ≥ 0.6 on stemmed content words), per respondent, one item per year per canonical; hand-reviewed aliases force known rewordings together. Domain, reverse and core are canonical-level.""" aliases = load_aliases() rows = [] for resp, g in items.groupby("respondent"): canon: list[tuple[str, set[str], str]] = [] # (canonical id, words, first text) used: dict[str, set[str]] = {} # canonical id → years already assigned (one item per year) alias_cid: dict[str, str] = {} # alias group → canonical id for (year, item_no), gg in sorted(g.groupby(["year", "item_no"]), key=lambda t: t[0]): text = gg.item_text.iloc[0] words = norm_words(text) cleaned = clean_text(text).lower() group = next((grp for r_, grp, pat in aliases if r_ == resp and pat.search(cleaned)), None) best, best_j = None, 0.0 if group is not None and group in alias_cid and year not in used.get(alias_cid[group], set()): best, best_j = alias_cid[group], 1.0 else: for cid, cw, _ in canon: if year in used.get(cid, set()): continue # that wording already has this year's item: sibling items ("free of violence / gangs / drugs") stay apart j = len(words & cw) / max(1, len(words | cw)) if j > best_j: best, best_j = cid, j if best is None or best_j < 0.6: cid = f"{resp}_{len(canon) + 1:02d}" canon.append((cid, words, text)) else: cid = best if group is not None: alias_cid.setdefault(group, cid) used.setdefault(cid, set()).add(year) cats = [domain_from_category(c) for c in gg.category.dropna().unique()] if "category" in gg else [] rows.append({"respondent": resp, "year": year, "item_no": item_no, "item_text": text, "canonical": cid, "domain_from_file": next((c for c in cats if c), None), "match_jaccard": round(best_j, 3) if best is not None and best_j >= 0.6 else None, "alias_group": group, "reverse_text": bool(NEGATIVE.search(text)) and not NOT_NEGATIVE.search(text)}) xw = pd.DataFrame(rows) # The workload alias spans all seven years. Keep the older publisher's # explicit Teaching and Learning classification rather than letting the # later wording's keyword fallback move the same item to Resources. reviewed_alias_domains = {"staff_overload": "teaching"} # Canonical-level domain: reviewed alias override, then the file's category # where any year has one, else keywords over the wordings. def canon_domain(g: pd.DataFrame) -> str: overrides = {reviewed_alias_domains[a] for a in g.alias_group.dropna() if a in reviewed_alias_domains} if len(overrides) > 1: raise ValueError(f"conflicting reviewed domains for {g.canonical.iloc[0]}") if overrides: return next(iter(overrides)) cats = g.domain_from_file.dropna() if len(cats): return cats.mode().iloc[0] tl = " ".join(g.item_text).lower() for pat, dom in KEYWORD_DOMAINS: if re.search(pat, tl): return dom return "other" dom = xw.groupby(["respondent", "canonical"]).apply(canon_domain).rename("domain").reset_index() rev = xw.groupby(["respondent", "canonical"]).reverse_text.agg(lambda s: bool(s.mean() >= 0.5)).rename("reverse").reset_index() mixed = xw.groupby(["respondent", "canonical"]).reverse_text.nunique() xw = xw.merge(dom, on=["respondent", "canonical"]).merge(rev, on=["respondent", "canonical"]) if (mixed > 1).any(): for (r_, c_) in mixed[mixed > 1].index: print(f"[survey] WARNING mixed reverse-coding within {c_}: {xw[(xw.respondent == r_) & (xw.canonical == c_)].item_text.unique().tolist()}") # Core = asked in every year the respondent group was surveyed (composites are built from core items only). n_years = xw.groupby("respondent").year.nunique() per = xw.groupby(["respondent", "canonical"]).year.nunique().rename("n_years_item").reset_index() xw = xw.merge(per, on=["respondent", "canonical"]) xw["core"] = xw.apply(lambda r: r.n_years_item == n_years[r.respondent], axis=1) # Hand-reviewed publisher anomalies: the item stays in the crosswalk (so the year still counts as asked) but is # not used in that year's composite; the reason is carried in the crosswalk for the metric pages. xw["excluded_reason"] = None if EXCLUSIONS.exists(): ex = pd.read_csv(EXCLUSIONS, dtype=str).fillna("") for r in ex.itertuples(): m = (xw.respondent == r.respondent) & (xw.year == r.year) & (xw.item_no == r.item_no) if not m.any(): print(f"[survey] WARNING exclusion not found: {r.respondent} {r.year} {r.item_no}") xw.loc[m, "core"] = False xw.loc[m, "excluded_reason"] = r.reason return xw.drop(columns=["n_years_item", "reverse_text"]).sort_values(["respondent", "canonical", "year"]) def split_candidates(xw: pd.DataFrame) -> list[str]: """Pairs of canonicals within a respondent whose year sets are disjoint and whose wordings overlap — likely rewordings for the alias table.""" out = [] for resp, g in xw.groupby("respondent"): cs = g.groupby("canonical").agg(years=("year", lambda s: frozenset(s)), text=("item_text", "first"), core=("core", "first")) ids = list(cs.index) for i in range(len(ids)): for j in range(i + 1, len(ids)): a, b = cs.loc[ids[i]], cs.loc[ids[j]] if a.years & b.years or a.core or b.core: continue wa, wb = norm_words(a.text), norm_words(b.text) jac = len(wa & wb) / max(1, len(wa | wb)) if jac >= 0.3: out.append(f"{resp}: {ids[i]} {sorted(a.years)} '{a.text}' <-> {ids[j]} {sorted(b.years)} '{b.text}' (jaccard {jac:.2f})") return out def main() -> None: OUT_DIR.mkdir(parents=True, exist_ok=True) all_items = [] for key in RETIRED_SPECS: for stale in (OUT_DIR / f"mdcps_climate_survey_{key}.csv", OUT_DIR / f"mdcps_climate_survey_{key}.school.csv"): if stale.exists(): stale.unlink() print(f"[survey] {key:18s} retired: removed {stale.name}") for key, spec in SPECS.items(): sid = f"mdcps_climate_survey_{key}" try: items, extras, diag = parse(key, spec) except Exception as e: # report and continue; the loader refuses sources without a parsed table print(f"[survey] {key:18s} FAILED: {type(e).__name__}: {e}") continue items.drop(columns=["respondent", "year", "item_text", "category"]).to_csv(OUT_DIR / f"{sid}.csv", index=False, float_format="%.6f") extras.to_csv(OUT_DIR / f"{sid}.school.csv", index=False, float_format="%.4f") all_items.append(items) print(f"[survey] {key:18s} schools {items.school_number.nunique():3d} items {items.item_no.nunique():2d} rows {len(items):5d} suppressed {int(items.suppressed.sum()):4d} n<10 schools {int((extras.n_responses < 10).sum()):3d} n unknown {int(extras.n_responses.isna().sum()):3d} agree med {items.pct_agree.median():5.1f} max {items.pct_agree.max():5.1f} renormalised {diag['renormalised']:4d} invalid {diag['invalid_agree']:3d} zero-total items {diag['zero_total_items']:4d} extras: rr {extras.response_rate.notna().sum():3d} rec {extras.recommend_yes_pct.notna().sum():3d} gradeAB {extras.grade_ab_pct.notna().sum():3d} invalid {diag['extras_invalid']}") items = pd.concat(all_items) xw = build_crosswalk(items) xw.to_csv(ITEMS_OUT, index=False) for resp, g in xw.groupby("respondent"): years = sorted(g.year.unique()) core = g[g.core].drop_duplicates("canonical") print(f"[survey] {resp}: {g.canonical.nunique()} canonical items over {len(years)} years; {len(core)} core; core per domain {core.domain.value_counts().to_dict()}; reverse-coded core {int(core.reverse.sum())}") cands = split_candidates(xw) if cands: print(f"[survey] {len(cands)} possible rewordings not merged (disjoint years, overlapping words) — review for survey_item_aliases.csv:") for c in cands: print(" ", c) if __name__ == "__main__": main()