pipeline_miami/verify/connectors/source_file.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.
"""Source-layer re-readers: pull values directly from the ingested files with
code that shares nothing with the TypeScript loaders.
Each reader returns {(school_number, year): {"value": float|None, "suppressed": bool}}.
"""
from __future__ import annotations
import io
import re
import zipfile
import pandas as pd
from pipeline_miami.verify import common
TOKENS = {"*", "**", "***", "i.s.", "n.a.", "n/a", "na", "--"}
GRADES = ["PK", "KG", "1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12"]
def _cell(v) -> tuple[float | None, bool]:
if v is None or (isinstance(v, float) and pd.isna(v)):
return None, False
s = str(v).strip()
if s == "":
return None, False
if s.lower() in TOKENS:
return None, True
try:
return float(s.replace(",", "")), False
except ValueError:
return None, True
def _header_row(df: pd.DataFrame, first_cell_re: str) -> int:
for i in range(min(12, len(df))):
if re.fullmatch(first_cell_re, str(df.iloc[i, 0]).strip(), re.I):
return i
raise SystemExit(f"header {first_cell_re} not found")
# ── FLDOE membership by school by grade ─────────────────────────────────────
def membership_total(year: str) -> dict[tuple[str, str], dict]:
"""School total from the total sheet ("… Total Enrollment" from 2018-19, "… Total Membership" in 2016-17 and
2017-18; the value is the fifth column in every year), or the sum of the grade columns in a year without one."""
path = common.latest_source_path(f"fldoe_membership_by_school_grade_{year}")
xf = pd.ExcelFile(path)
total_sheet = next((s for s in xf.sheet_names if re.search(r"total (enrollment|membership)", s, re.I)), None)
out: dict[tuple[str, str], dict] = {}
if total_sheet:
raw = pd.read_excel(xf, sheet_name=total_sheet, header=None, dtype=str)
h = _header_row(raw, r"district #")
for _, r in raw.iloc[h + 1:].iterrows():
if str(r.iloc[0]).strip().replace(".0", "") != "13":
continue
num = common.pad4(r.iloc[2])
if not num:
continue
v, sup = _cell(r.iloc[4])
out[(num, year)] = {"value": v, "suppressed": sup}
return out
raw = pd.read_excel(xf, sheet_name="School", header=None, dtype=str)
h = _header_row(raw, r"district #")
hdr = [str(c).strip().upper().replace(".0", "") for c in raw.iloc[h].tolist()]
idx = {g: hdr.index(g) for g in GRADES}
for _, r in raw.iloc[h + 1:].iterrows():
if str(r.iloc[0]).strip().replace(".0", "") != "13":
continue
num = common.pad4(r.iloc[2])
if not num:
continue
cells = [_cell(r.iloc[idx[g]]) for g in GRADES]
if any(s for _, s in cells):
out[(num, year)] = {"value": None, "suppressed": True}
else:
out[(num, year)] = {"value": sum(v or 0 for v, _ in cells), "suppressed": False}
return out
def membership_grade(year: str, grade: str) -> dict[tuple[str, str], dict]:
"""One grade column; tokens resolved with the same sum-equals-total rule the loader documents."""
path = common.latest_source_path(f"fldoe_membership_by_school_grade_{year}")
xf = pd.ExcelFile(path)
raw = pd.read_excel(xf, sheet_name="School", header=None, dtype=str)
h = _header_row(raw, r"district #")
hdr = [str(c).strip().upper().replace(".0", "") for c in raw.iloc[h].tolist()]
idx = {g: hdr.index(g) for g in GRADES}
totals = membership_total(year)
out: dict[tuple[str, str], dict] = {}
for _, r in raw.iloc[h + 1:].iterrows():
if str(r.iloc[0]).strip().replace(".0", "") != "13":
continue
num = common.pad4(r.iloc[2])
if not num:
continue
cells = {g: _cell(r.iloc[idx[g]]) for g in GRADES}
numeric_sum = sum(v or 0 for v, _ in cells.values())
t = totals.get((num, year), {}).get("value")
tokens_are_zero = any(s for _, s in cells.values()) and t is not None and abs(numeric_sum - t) < 0.5
v, sup = cells[grade]
if sup and tokens_are_zero:
out[(num, year)] = {"value": 0.0, "suppressed": False}
else:
out[(num, year)] = {"value": (v if v is not None else (None if sup else 0.0)), "suppressed": sup}
return out
def ell_share(year: str, which: str = "exact") -> dict[tuple[str, str], dict]:
"""Independent source reconstruction of exact ELL shares and censoring bounds.
This checks the declared interval arithmetic, not an imputation of hidden
counts. The source's absent-row convention is zero, as recorded in the
source registry; a malformed/empty district extract must not imply zeros.
"""
if which not in {"exact", "lower", "upper"}:
raise ValueError(which)
path = common.latest_source_path(f"fldoe_ell_membership_{year}")
raw = pd.read_excel(path, sheet_name="School", header=None, dtype=str)
h = _header_row(raw, r"district #")
hdr = [re.sub(r"[^a-z0-9]+", " ", str(c).lower()).strip() for c in raw.iloc[h]]
labels = ["white", "black or african american", "hispanic latino", "asian", "native hawaiian or other pacific islander", "american indian or alaska native", "two or more races"]
positions = [hdr.index(label) for label in labels]
totals = membership_total(year)
by_school = {}
for _, row in raw.iloc[h + 1:].iterrows():
if str(row.iloc[0]).strip().replace(".0", "") != "13":
continue
school = common.pad4(row.iloc[2])
if not school:
continue
if school in by_school:
raise ValueError(f"duplicate ELL school {school} in {year}")
visible, hidden = 0.0, 0
for i in positions:
value = row.iloc[i]
if pd.isna(value) or not str(value).strip():
continue
if str(value).strip().lower() in TOKENS:
hidden += 1
continue
number = float(value) # fail on an unknown token, never silently infer zero
if number < 0 or not number.is_integer():
raise ValueError(f"invalid ELL count {value}")
visible += number
by_school[school] = (visible, hidden)
if not by_school:
raise ValueError(f"no district-13 ELL rows for {year}")
out = {}
for key, total in totals.items():
n = total["value"]
if total["suppressed"] or n is None or n <= 0:
continue
visible, hidden = by_school.get(key[0], (0.0, 0))
if visible > n:
raise ValueError(f"ELL visible count exceeds membership for {key}")
lower = 100 * visible / n
upper = 100 * min(n, visible + 9 * hidden) / n
suppressed = which == "exact" and lower != upper
out[key] = {"value": None if suppressed else upper if which == "upper" else lower, "suppressed": suppressed}
return out
# ── FISH level-of-service: parser A (text regex), numbers attached via the parsed table's name map ──
def fish_los(year: str, field: str) -> dict[tuple[str, str], dict]:
import pdfplumber
from pipeline_miami.parse.fish_los import parse_text
sid = f"fldoe_fish_los_dade_{year}"
path = common.latest_source_path(sid)
with pdfplumber.open(path) as pdf:
recs = parse_text(pdf)
# Name → number from the committed parsed table (parser B + matcher); values come from parser A.
parsed = pd.read_csv(common.PARSED_DIR / f"{sid}.csv", dtype={"school_number": str})
by_first = {}
for _, r in parsed.iterrows():
if isinstance(r.school_number, str):
by_first.setdefault(r.name_first_line, []).append((r.school_number, r.school_capacity))
out: dict[tuple[str, str], dict] = {}
best_cap: dict[str, float] = {}
for rec in recs:
cands = by_first.get(rec["name_first_line"], [])
if not cands or rec["school_capacity"] <= 0:
continue
num = cands[0][0]
if rec["school_capacity"] < best_cap.get(num, -1):
continue
best_cap[num] = rec["school_capacity"]
out[(num, year)] = {"value": float(rec[field]), "suppressed": False}
return out
# ── ESSA per-pupil expenditures ──────────────────────────────────────────────
def essa_ppe(year: str, column: int) -> dict[tuple[str, str], dict]:
path = common.latest_source_path(f"fldoe_essa_ppe_{year}")
raw = pd.read_excel(path, header=None, dtype=str)
h = _header_row(raw, r"district (code|number)")
out: dict[tuple[str, str], dict] = {}
for _, r in raw.iloc[h + 1:].iterrows():
if str(r.iloc[0]).strip() != "13":
continue
num = common.pad4(r.iloc[2])
if not num:
continue
v, sup = _cell(r.iloc[column])
if v is None and not sup:
continue
out[(num, year)] = {"value": v, "suppressed": sup}
return out
# ── ABC book: independent regex over the page text ──────────────────────────
def abc_book(source_id: str, field: str) -> dict[tuple[str, str], dict]:
import pdfplumber
path = common.latest_source_path(source_id)
label = {"enrollment_abc": r"Student Enrollment", "fish_capacity_permanent_abc": r"Perm FISH Cap", "utilization_permanent_pct_abc": r"% Utilization Perm FISH(?:\s*Cap)?"}[field]
out: dict[tuple[str, str], dict] = {}
with pdfplumber.open(path) as pdf:
for page in pdf.pages:
t = page.extract_text() or ""
if "Capacity Membership Data" not in t:
continue
t = re.sub(r"\([^)]*\)", " ", t).replace("*", " ")
h = re.search(r"^(\d{4})\s*[-–]\s*(.+?)\s*$", t, re.M)
y = re.search(r"(\d{4})-(\d{4})\s+\d{4}-\d{4}\s+\d{4}-\d{4}", t)
m = re.search(label + r"\s*(N/A|[\d,]+(?:\.\d+)?%?)", t, re.S)
if not (h and y and m):
continue
tok = m.group(1)
if tok.upper() == "N/A":
continue
out[(h.group(1), f"{y.group(1)}-{y.group(2)[2:]}")] = {"value": float(tok.replace(",", "").replace("%", "")), "suppressed": False}
return out
# ── NCES CCD membership (federal copy) ──────────────────────────────────────
def ccd_total(year: str) -> dict[tuple[str, str], dict]:
path = common.latest_source_path(f"nces_ccd_membership_{year}")
z = zipfile.ZipFile(path)
member = [m for m in z.namelist() if m.lower().endswith(".csv")][0]
out: dict[tuple[str, str], dict] = {}
with z.open(member) as fh:
for chunk in pd.read_csv(fh, dtype=str, chunksize=500_000, encoding="latin1", low_memory=False):
sel = chunk[(chunk.ST_LEAID == "FL-13") & (chunk.TOTAL_INDICATOR == "Education Unit Total")]
for _, r in sel.iterrows():
num = common.pad4(str(r.ST_SCHID).replace("FL-13-", ""))
if num:
out[(num, year)] = {"value": float(r.STUDENT_COUNT), "suppressed": False}
return out
# ── Outcome files (P1): School Grades, absenteeism, graduation, SRS ────────────
def _find_header(raw: pd.DataFrame, first_cell_re: str) -> int | None:
for i in range(min(12, len(raw))):
if re.fullmatch(first_cell_re, str(raw.iloc[i, 0]).strip(), re.I):
return i
return None
def school_grades(year: str, column_prefix: str) -> dict[tuple[str, str], dict]:
"""A School Grades component column (all sheets; 2022-23 has one sheet per school type)."""
path = common.latest_source_path(f"fldoe_school_grades_{year}")
xf = pd.ExcelFile(path)
out: dict[tuple[str, str], dict] = {}
for sh in xf.sheet_names:
raw = pd.read_excel(xf, sheet_name=sh, header=None, dtype=str)
h = _find_header(raw, r"district number")
if h is None:
continue
hdr = [re.sub(r"[^a-z0-9]+", " ", str(c).lower()).strip() for c in raw.iloc[h].tolist()]
try:
ic = next(i for i, c in enumerate(hdr) if c.startswith(column_prefix))
except StopIteration:
continue
iprov = hdr.index("virtual provider number") if "virtual provider number" in hdr else None
for _, r in raw.iloc[h + 1:].iterrows():
if str(r.iloc[0]).strip() != "13":
continue
if iprov is not None and isinstance(r.iloc[iprov], str) and r.iloc[iprov].strip():
continue
num = common.pad4(r.iloc[2])
if not num:
continue
v, sup = _cell(r.iloc[ic])
if v is None and not sup:
continue
out[(num, year)] = {"value": v, "suppressed": sup}
return out
def absenteeism(year: str, which: str) -> dict[tuple[str, str], dict]:
"""which = 'pct10' (chronic_absenteeism_rate) or 'pct21' (absent_21plus_rate); values scaled to percent."""
sid = f"fldoe_absent21_10pct_{year}" if year >= "2017-18" else f"fldoe_absent21_{year}"
path = common.latest_source_path(sid)
xf = pd.ExcelFile(path)
sheet = next((s for s in xf.sheet_names if re.search("school", s, re.I)), xf.sheet_names[-1])
raw = pd.read_excel(xf, sheet_name=sheet, header=None, dtype=str)
h = _find_header(raw, r"district #")
hdr = [re.sub(r"[^a-z0-9%]+", " ", str(c).lower()).strip() for c in raw.iloc[h].tolist()]
pat = r"^% of students absent 10%" if which == "pct10" else r"^% (of students )?absent 21"
ic = next(i for i, c in enumerate(hdr) if re.match(pat, c))
isch = next(i for i, c in enumerate(hdr) if c == "school")
out: dict[tuple[str, str], dict] = {}
for _, r in raw.iloc[h + 1:].iterrows():
if str(r.iloc[0]).strip().replace(".0", "") != "13":
continue
num = common.pad4(r.iloc[isch])
if not num:
continue
v, sup = _cell(r.iloc[ic])
if v is None and not sup:
continue
out[(num, year)] = {"value": None if v is None else v * 100, "suppressed": sup}
return out
def graduation(year: str, group: str = "total") -> dict[tuple[str, str], dict]:
"""Federal graduation rate for a group label as it appears in the category or race file header."""
sid = f"fldoe_fed_grad_rate_{'category' if group in ('total', 'ese', 'free/reduced lunch', 'ell', 'migrant', 'at risk', 'male', 'female') else 'race'}_{year}"
path = common.latest_source_path(sid)
raw = pd.read_excel(path, header=None, dtype=str)
h = _find_header(raw, r"district number")
hdr = [re.sub(r"[^a-z0-9/]+", " ", str(c).lower()).strip() for c in raw.iloc[h].tolist()]
ic = next(i for i, c in enumerate(hdr) if c in (f"{group} graduation rate", f"{group} federal graduation rate"))
iprov = hdr.index("virtual provider number")
out: dict[tuple[str, str], dict] = {}
for _, r in raw.iloc[h + 1:].iterrows():
if str(r.iloc[0]).strip() != "13":
continue
if isinstance(r.iloc[iprov], str) and r.iloc[iprov].strip():
continue
num = common.pad4(r.iloc[2])
if not num:
continue
v, sup = _cell(r.iloc[ic])
if v is None and not sup:
continue
out[(num, year)] = {"value": v, "suppressed": sup}
return out
def srs(year: str, source_id: str) -> dict[tuple[str, str], dict]:
"""Percent level 3 or above from one State Report of Schools file."""
path = common.latest_source_path(source_id)
raw = pd.read_excel(path, header=None, dtype=str)
# Header = first row with a "District Number" cell in any column (the
# 2023 science files lead with "Grade"; 2024+ files lead with the district).
h = next(
i for i in range(min(12, len(raw)))
if any(re.fullmatch(r"district number", str(c).strip(), re.I) for c in raw.iloc[i].tolist())
)
hdr = [re.sub(r"[^a-z0-9]+", " ", str(c).lower()).strip() for c in raw.iloc[h].tolist()]
ic = next(i for i, c in enumerate(hdr) if "level 3 or above" in c)
isch = hdr.index("school number")
idist = hdr.index("district number")
out: dict[tuple[str, str], dict] = {}
for _, r in raw.iloc[h + 1:].iterrows():
if str(r.iloc[idist]).strip() != "13":
continue
num = common.pad4(r.iloc[isch])
if not num:
continue
v, sup = _cell(r.iloc[ic])
if v is None and not sup:
continue
out[(num, year)] = {"value": v, "suppressed": sup}
return out
# ── NCES CCD race shares (unsuppressed): cross-publisher reference for pct_* ──
_CCD_RACE = {
"pct_white": "White", "pct_black": "Black or African American", "pct_hispanic": "Hispanic/Latino", "pct_asian": "Asian",
"pct_hawaiian_pac_isl": "Native Hawaiian or Other Pacific Islander", "pct_amer_indian": "American Indian or Alaska Native", "pct_two_or_more": "Two or more races",
}
def ccd_race(year: str, metric_key: str) -> dict[tuple[str, str], dict]:
"""Race share from the CCD membership file's race × sex subtotals (no suppression)."""
path = common.latest_source_path(f"nces_ccd_membership_{year}")
z = zipfile.ZipFile(path)
member = [m for m in z.namelist() if m.lower().endswith(".csv")][0]
counts: dict[str, dict[str, float]] = {}
with z.open(member) as fh:
for chunk in pd.read_csv(fh, dtype=str, chunksize=500_000, encoding="latin1", low_memory=False):
sel = chunk[(chunk.ST_LEAID == "FL-13") & chunk.TOTAL_INDICATOR.str.startswith("Derived - Subtotal by Race/Ethnicity and Sex")]
for _, r in sel.iterrows():
num = common.pad4(str(r.ST_SCHID).replace("FL-13-", ""))
if not num:
continue
try:
n = float(r.STUDENT_COUNT)
except (TypeError, ValueError):
continue
counts.setdefault(num, {})
counts[num][r.RACE_ETHNICITY] = counts[num].get(r.RACE_ETHNICITY, 0.0) + n
label = _CCD_RACE[metric_key]
out: dict[tuple[str, str], dict] = {}
for num, m in counts.items():
tot = sum(m.values())
if tot > 0:
out[(num, year)] = {"value": m.get(label, 0.0) / tot * 100, "suppressed": False}
return out
# ── School climate survey: independent re-read of a long-format workbook ─────
# The reader's own column map for the years the harness checks (configuration,
# duplicated on purpose so this code imports nothing from the parser).
# counts: response-count columns; with "agree" all three shares come from the
# counts, otherwise dk/dis counts are divided by the n column.
_SURVEY_COLUMNS: dict[str, dict] = {
"students_2019-20": dict(sheet="STUD 2020", loc="2020 Loc No", item="Qno", agree="Agree", dk="DK", dis="Disagree", n="Ncount"),
# 2021-22: the three workbooks carry no results sheet, only pivot caches (read here with the harness's own reader).
"students_2021-22": dict(sheet="pivot", loc="LocNo", item="Q_No", agree="Agreement", dk="DontKnow", dis="Disagreement", n="Ncount"),
"parents_2021-22": dict(sheet="pivot", loc="LocNo", item="Qno", agree="TTLPctAgreement", dk="PctDK", dis="TTLPctDisagreement", n="Ncount"),
# The staff cache stores each item's shares on the row labelled one number lower (values on row Q-n are item
# Q-(n+1)'s; review 2026-09-16b, N1): label_offset re-attaches them. The rule is copied from the parser's
# finding, so this re-read verifies the implementation of the offset, not the finding itself (which rests on
# the adjacent-year comparison recorded in docs/cities/miami/10_review_2026-09-16b.md).
"staff_2021-22": dict(sheet="pivot", loc="LocNo", item="Qno", agree="Agrmnt", dk="PctDK", dis="Disgrmnt", n="NoStaffResponses", label_offset=1),
"students_2024-25": dict(sheet="2025_Students_Data", loc="LOC", item="SSQNO", n="TTL", flag="Less_10", 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", agree="Schl Agreement", dk="Schl DK", dis="Schl Disagreement", n="Parent Ncount"),
"parents_2023-24": dict(sheet="2024_Parent_Data", loc="LOC", item="QNO", agree="Pct_AG", dk="PCT_DK", dis="PCT_DA", n="TTL"),
"staff_2020-21": dict(sheet="2021 Staff Data", loc="LOC", text="Teacher Survey Question", agree="Pct Agree", dk="Pct Neutral", dis="Pct Disagree", n="TTL_Responses"),
"staff_2022-23": dict(sheet="2023_Staff_Data_June21", loc="LOC", item="QNO", agree="Agree_Pct", n="TTL", counts=dict(dk=["DK"], dis=["DA", "SD"])),
"staff_2024-25": dict(sheet="Staff_Data_2025", loc="LOC", item="QNO", agree="PCT_Agree", dk="PCT_DK", dis="PCT_Disagree", n="TTL", flag="Less_10"),
}
_SURVEY_SUMMARY_ROWS = {"Q00", "Q99"}
_SURVEY_ROUNDING_PP = 0.5 # per share column published as whole percents: its rounding can move the sum by up to 0.5 pp
_SURVEY_FLOAT_PP = 0.05 # allowance for columns published with decimals
def _survey_pivot_cache(path) -> pd.DataFrame:
"""The harness's own reader for a workbook whose results live only in a pivot cache: the first cache's
field names and shared items from pivotCacheDefinition1.xml, the records from pivotCacheRecords1.xml
(an <x v="i"/> cell indexes the field's shared items, <m/> is missing, any other cell carries its value)."""
import zipfile
from xml.etree import ElementTree as ET
ns = "{http://schemas.openxmlformats.org/spreadsheetml/2006/main}"
z = zipfile.ZipFile(path)
definition = ET.fromstring(z.read("xl/pivotCache/pivotCacheDefinition1.xml"))
names, shared = [], []
for field in definition.find(f"{ns}cacheFields"):
names.append(field.get("name"))
items = field.find(f"{ns}sharedItems")
shared.append([it.get("v") for it in items] if items is not None else [])
records = []
for rec in ET.fromstring(z.read("xl/pivotCache/pivotCacheRecords1.xml")):
row = {}
for i, cell in enumerate(rec):
tag = cell.tag.replace(ns, "")
if tag == "x":
idx = int(cell.get("v"))
row[names[i]] = shared[i][idx] if idx < len(shared[i]) else None
elif tag == "m":
row[names[i]] = None
else:
row[names[i]] = cell.get("v")
records.append(row)
return pd.DataFrame.from_records(records, columns=names)
def _survey_num(v) -> float | None:
try:
x = float(str(v).strip().replace(",", "").replace("%", ""))
except ValueError:
return None
return None if x != x else x
def climate_survey(year: str, respondent: str, domain: str) -> dict[tuple[str, str], dict]:
"""Recompute a survey composite from the workbook's results sheet with code
that shares nothing with pipeline_miami/parse/climate_survey.py except the
committed crosswalk (canonical, domain, core, reverse per item). Its own
rules, written independently of the parser (2026-09-16):
* shares come from response counts where the sheet has them, else from
the published share columns (fractions × 100 when the column's median
is ≤ 1); where all three shares exist and sum further from 100 than
the file's own rounding allows (0.5 pp per share column published as
whole percents, plus 0.05 pp), agreement is re-expressed over
agree + dk + disagree;
* a share outside [0, 100] after that is invalid and the item is skipped;
* a three-share distribution totalling zero is missing; favorable means
agreement with a positive item or disagreement with a negative item,
with unknown answers credited in neither direction;
* the school's respondent count is the median of its item rows' counts,
else the Q00 row's count; below 10 the school is suppressed, and an
item row with its own count below 10 is skipped;
* Q00 / Q99 rows are never items;
* a 2021-22 workbook is read from its pivot cache (_survey_pivot_cache);
a label_offset in the column map moves each row's shares to the item
`offset` labels later (the staff 2021-22 cache).
Limits, stated so the check is not over-read: the conventions above were
written to the same publisher rules the parser follows, so the re-read
verifies the parser's implementation of them and would not detect a wrong
convention shared by both (the row/label offset in N1 was found by the
adjacent-year comparison, not by this reader)."""
import csv
key = f"{respondent}_{year}"
if key not in _SURVEY_COLUMNS:
raise SystemExit(f"{key}: not in the harness's own column map (_SURVEY_COLUMNS)")
cols = _SURVEY_COLUMNS[key]
xw = {}
with open(common.REPO_ROOT / "data" / "cities" / "miami" / "reference" / "survey_items.csv") as fh:
for r in csv.DictReader(fh):
if r["respondent"] == respondent and r["year"] == year:
xw[r["item_no"]] = r
core_n = len({r["canonical"] for r in xw.values() if r["core"] == "True" and r["domain"] == domain})
path = common.latest_source_path(f"mdcps_climate_survey_{key}")
raw = _survey_pivot_cache(path) if cols["sheet"] == "pivot" else pd.read_excel(path, sheet_name=cols["sheet"], dtype=str)
raw.columns = [str(c).strip() for c in raw.columns]
# item numbers
item_src = raw[cols["item"]] if cols.get("item") else raw[cols["text"]]
item_no = []
for it in item_src:
m = re.match(r"\s*(?:Q|SSQ|PQ)?-?\s*(\d{1,2})", str(it))
item_no.append(f"Q{int(m.group(1)):02d}" if m else None)
if cols.get("label_offset"):
labels = set(item_no)
shifted = []
for it in item_no:
if it and it not in _SURVEY_SUMMARY_ROWS:
target = f"Q{int(it[1:]) + int(cols['label_offset']):02d}"
shifted.append(target if target in labels else None) # values whose target label is not in the file belong to no item
else:
shifted.append(it)
item_no = shifted
locs = [common.pad4(str(v).split("-")[0].strip()) for v in raw[cols["loc"]]]
# shares
if "counts" in cols and "agree" in cols["counts"]:
parts = {k: sum(pd.to_numeric(raw[c], errors="coerce").fillna(0) for c in cs) for k, cs in cols["counts"].items()}
denom = parts["agree"] + parts["dk"] + parts["dis"]
agree = (parts["agree"] / denom * 100.0).where(denom > 0)
dk = (parts["dk"] / denom * 100.0).where(denom > 0)
dis = (parts["dis"] / denom * 100.0).where(denom > 0)
else:
def share(col):
x = pd.to_numeric(raw[col].astype(str).str.replace("%", "").str.replace(",", ""), errors="coerce")
return x * 100.0 if x.dropna().median() <= 1.0001 else x
agree = share(cols["agree"])
if "counts" in cols:
tot = pd.to_numeric(raw[cols["n"]], errors="coerce")
dk = sum(pd.to_numeric(raw[c], errors="coerce").fillna(0) for c in cols["counts"]["dk"]) / tot * 100.0
dis = sum(pd.to_numeric(raw[c], errors="coerce").fillna(0) for c in cols["counts"]["dis"]) / tot * 100.0
else:
dk = share(cols["dk"]) if cols.get("dk") else pd.Series([None] * len(raw), dtype=float)
dis = share(cols["dis"]) if cols.get("dis") else pd.Series([None] * len(raw), dtype=float)
n_col = pd.to_numeric(raw[cols["n"]], errors="coerce") if cols.get("n") else pd.Series([None] * len(raw), dtype=float)
# rounding allowance from the file's own precision, per share column
whole_cols = sum(int(((c.dropna() - c.dropna().round()).abs() < 1e-9).all()) for c in (agree, dk, dis) if c.notna().any())
rounding_pp = _SURVEY_ROUNDING_PP * whole_cols + (_SURVEY_FLOAT_PP if whole_cols < 3 else 0.0)
flag = raw[cols["flag"]].astype(str).str.strip().str.lower().isin(["less10", "less_10", "low", "low_ncnt", "1", "true", "yes", "y", "low_n", "lowncnt", "<10", "less than 10"]) if cols.get("flag") else pd.Series([False] * len(raw))
# respondent count per school: item rows' median, else the Q00 row
n_items: dict[str, list[float]] = {}
n_q00: dict[str, float] = {}
for num, it, nn in zip(locs, item_no, n_col):
if not num or nn is None or nn != nn:
continue
if it == "Q00":
n_q00[num] = float(nn)
elif it and it not in _SURVEY_SUMMARY_ROWS:
n_items.setdefault(num, []).append(float(nn))
n_school = {num: float(pd.Series(v).median()) for num, v in n_items.items()}
for num, v in n_q00.items():
n_school.setdefault(num, v)
per_school: dict[str, list[float]] = {}
flagged: set[str] = set()
seen: set[tuple[str, str]] = set()
for num, it, a, d1, d2, fl, nn in zip(locs, item_no, agree, dk, dis, flag, n_col):
if not num or not it or it in _SURVEY_SUMMARY_ROWS or (num, it) in seen:
continue
seen.add((num, it))
if fl:
flagged.add(num)
x = xw.get(it)
if not x or x["core"] != "True" or x["domain"] != domain:
continue
per_school.setdefault(num, [])
if a is None or a != a:
continue
if nn is not None and nn == nn and nn < 10:
continue # the row's own respondent count is under the publisher's floor
if d1 is not None and d1 == d1 and d2 is not None and d2 == d2:
tot = a + d1 + d2
if tot == 0:
continue # empty published response distribution, not zero agreement
if tot > 0 and abs(tot - 100.0) > rounding_pp:
a = a / tot * 100.0
d2 = d2 / tot * 100.0
favorable = d2 if x["reverse"] == "True" else a
if favorable is None or favorable != favorable or favorable < -1e-6 or favorable > 100.0 + 1e-6:
continue # invalid publisher row: skipped, as the loader skips it
per_school[num].append(round(min(100.0, max(0.0, favorable)), 6))
out: dict[tuple[str, str], dict] = {}
for num, vals in per_school.items():
n = n_school.get(num)
if num in flagged or (n is not None and n < 10):
out[(num, year)] = {"value": None, "suppressed": True}
elif len(vals) * 2 < core_n:
out[(num, year)] = {"value": None, "suppressed": True}
else:
v = sum(vals) / len(vals)
assert 0.0 <= v <= 100.0, (num, v)
out[(num, year)] = {"value": v, "suppressed": False}
return out