pipeline_miami/parse/climate_survey.py
Served verbatim from the project repository. Internal working document conventions apply: documents may reference file paths, branch names, and findings-ledger anchors from the repo.
"""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/<source_id>.csv
school_number, item_no, n_responses, pct_agree, pct_dk, pct_disagree, suppressed
data/cities/miami/reference/parsed/survey/<source_id>.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()