pipeline_miami/parse/fish_los.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.
"""Parse an FLDOE FISH Level of Service report (per-district PDF) into a
per-facility table, with a two-parser agreement gate.
Layout (2022-23 -> 2024-25 editions, 20 pages for Dade): one row per
facility with ten fields —
FACILITY NAME | SCHOOL CAPACITY | PERMANENT CAPACITY | DINING CAPACITY |
YEAR ROUND CAPACITY | UTILIZATION FACTOR | PERCENTAGE UTILIZATION |
PRIMARY USE | TOTAL COFTE | AVAILABLE CAPACITY (SURPLUS)
Long facility names and long primary-use labels wrap onto the following
line. Percentage utilization = COFTE / school capacity (findings.md F.7).
Parser A ("text"): pdfplumber page text + a regex on the numeric tail; a
non-matching line is appended to the previous record as `continuation`
(A cannot tell a name continuation from a primary-use continuation).
Parser B ("words"): pdfplumber word boxes clustered into lines; column
membership decided by x-position against the header words, so name and
primary-use continuations are assigned correctly. B is canonical; the gate
requires A and B to agree on record count and on every numeric field, in
order, per page.
Output: data/cities/miami/reference/parsed/<source_id>.csv (+ a json gate
summary) with school numbers attached by name (parse/names.py).
Run: python -m pipeline_miami.parse.fish_los fldoe_fish_los_dade_2024-25
"""
from __future__ import annotations
import json
import pathlib
import re
import sys
import pandas as pd
import pdfplumber
from pipeline_miami.parse.names import NameMatcher
REPO_ROOT = pathlib.Path(__file__).resolve().parents[2]
SOURCES = REPO_ROOT / "data" / "cities" / "miami" / "sources"
OUT_DIR = REPO_ROOT / "data" / "cities" / "miami" / "reference" / "parsed"
NUM = r"-?[\d,]+(?:\.\d+)?"
ROW_RE = re.compile(
rf"^(?P<name>.+?)\s+(?P<school_cap>[\d,]+)\s+(?P<perm_cap>[\d,]+)\s+(?P<dining>[\d,]+)\s+(?P<year_round>[\d,]+)"
rf"\s+(?P<factor>\d\.\d{{2}})\s+(?P<util>\d+)\s+(?P<use>[A-Z][A-Z /&\-]*?)\s+(?P<cofte>{NUM})\s+(?P<avail>{NUM})$"
)
HEADER_MARKERS = ("FACILITY NAME", "FLORIDA INVENTORY", "LEVEL OF SERVICE", "ORGANIZATION:", "FACILITY :",
"FACILITY USE:", "(SURPLUS)", "Report Date:", "CAPACITY CAPACITY")
FIELDS = ["school_capacity", "permanent_capacity", "dining_capacity", "year_round_capacity",
"utilization_factor", "utilization_pct", "primary_use", "cofte", "available_capacity"]
# Name overrides for facilities whose FISH label differs from every directory
# name beyond fuzzy tolerance. Keep this list short and documented.
NAME_OVERRIDES: dict[str, str] = {
# FISH label -> M-DCPS school number (directory name in comment). Verified 2026-09-15.
"ETHEL F BECKFORD-RICHMOND ELEMENTARY": "4651", # Ethel F. Beckford/Richmond Primary Learning Center (closed in MSID; facility now ESE use)
"JAMES H BRIGHT ELEMENTARY": "0481", # James H. Bright Elementary / J.W. Johnson Elementary
"THOMAS JEFFERSON MIDDLE": "6281", # Thomas Jefferson Biscayne Gardens K-8 Academy
"YOUNG WOMEN'S ACADEMY": "7055", # Young Women's Preparatory Academy
"CITRUS GROVE ELEMENTARY": "0801", # Citrus Grove K-8 Center
"COLONIAL DRIVE ELEMENTARY": "0861", # Johnnie M. Parris Colonial Drive Elementary
"MIAMI LAKES TECHNOLOGICAL SENIOR HIGH": "7391", # Miami Lakes Educational Center (senior high)
}
def _num(s: str) -> float:
return float(s.replace(",", ""))
def _is_header(line: str) -> bool:
return any(m in line for m in HEADER_MARKERS) or re.match(r"^Page \d+ of \d+", line) is not None
def _latest_file(source_id: str) -> pathlib.Path:
d = SOURCES / source_id
for day in sorted([p for p in d.iterdir() if p.is_dir()], reverse=True):
for f in day.iterdir():
if f.suffix.lower() == ".pdf":
return f
raise FileNotFoundError(source_id)
# ── Parser A: text lines ─────────────────────────────────────────────────────
def parse_text(pdf: pdfplumber.PDF) -> list[dict]:
recs: list[dict] = []
for pno, page in enumerate(pdf.pages, start=1):
for line in (page.extract_text() or "").split("\n"):
line = line.strip()
if not line or _is_header(line):
continue
m = ROW_RE.match(line)
if m:
recs.append({
"page": pno, "name_first_line": m["name"].strip(), "continuation": "",
"school_capacity": _num(m["school_cap"]), "permanent_capacity": _num(m["perm_cap"]),
"dining_capacity": _num(m["dining"]), "year_round_capacity": _num(m["year_round"]),
"utilization_factor": float(m["factor"]), "utilization_pct": float(m["util"]),
"primary_use": m["use"].strip(), "cofte": _num(m["cofte"]), "available_capacity": _num(m["avail"]),
})
elif recs and recs[-1]["page"] == pno:
recs[-1]["continuation"] += (" " if recs[-1]["continuation"] else "") + line
return recs
# ── Parser B: word boxes ─────────────────────────────────────────────────────
def _lines_from_words(words: list[dict], tol: float = 3.0) -> list[list[dict]]:
lines: list[list[dict]] = []
for w in sorted(words, key=lambda w: (round(w["top"]), w["x0"])):
if lines and abs(lines[-1][0]["top"] - w["top"]) <= tol:
lines[-1].append(w)
else:
lines.append([w])
return [sorted(l, key=lambda w: w["x0"]) for l in lines]
def parse_words(pdf: pdfplumber.PDF) -> list[dict]:
recs: list[dict] = []
for pno, page in enumerate(pdf.pages, start=1):
words = page.extract_words(keep_blank_chars=False, use_text_flow=False, x_tolerance=1.5)
lines = _lines_from_words(words)
# Column anchors from the header line that contains "FACILITY" and "PRIMARY".
x_school = x_primary = x_total = None
for l in lines:
texts = [w["text"] for w in l]
if "FACILITY" in texts and "PRIMARY" in texts:
for w in l:
if w["text"] == "SCHOOL" and x_school is None:
x_school = w["x0"]
if w["text"] == "PRIMARY":
x_primary = w["x0"]
if w["text"] == "TOTAL":
x_total = w["x0"]
break
if x_school is None or x_primary is None or x_total is None:
raise RuntimeError(f"page {pno}: header anchors not found")
name_max = x_school - 4
use_min, use_max = x_primary - 6, x_total - 4
for l in lines:
text = " ".join(w["text"] for w in l)
if _is_header(text):
continue
numeric_tail = [w for w in l if re.fullmatch(NUM, w["text"]) and w["x0"] >= name_max]
if len(numeric_tail) >= 8: # a data line: 6 capacity/factor/util numbers + cofte + available
name_words = [w["text"] for w in l if w["x0"] < name_max]
numw = [w for w in l if w["x0"] >= name_max and re.fullmatch(NUM, w["text"])]
nums = [w["text"] for w in numw]
# order: school, perm, dining, year_round, factor, util, cofte, avail
if len(nums) != 8:
raise RuntimeError(f"page {pno}: expected 8 numbers, got {nums} in line: {text}")
# primary use = the non-numeric words between the utilization % and the COFTE tokens
use_words = [w["text"] for w in l if w["x0"] > numw[5]["x1"] and w["x1"] < numw[6]["x0"] and not re.fullmatch(NUM, w["text"])]
recs.append({
"page": pno, "name_first_line": " ".join(name_words), "name": " ".join(name_words),
"school_capacity": _num(nums[0]), "permanent_capacity": _num(nums[1]),
"dining_capacity": _num(nums[2]), "year_round_capacity": _num(nums[3]),
"utilization_factor": float(nums[4]), "utilization_pct": float(nums[5]),
"primary_use": " ".join(use_words), "cofte": _num(nums[6]), "available_capacity": _num(nums[7]),
})
elif recs and recs[-1]["page"] == pno:
# continuation: split words by column
name_more = [w["text"] for w in l if w["x0"] < name_max]
use_more = [w["text"] for w in l if use_min <= w["x0"] < use_max]
if name_more:
recs[-1]["name"] += " " + " ".join(name_more)
if use_more:
recs[-1]["primary_use"] += " " + " ".join(use_more)
return recs
# ── Gate + output ────────────────────────────────────────────────────────────
def gate(a: list[dict], b: list[dict]) -> dict:
result = {"parser_a_records": len(a), "parser_b_records": len(b), "count_match": len(a) == len(b),
"numeric_mismatches": [], "name_first_line_mismatches": []}
for i, (ra, rb) in enumerate(zip(a, b)):
for f in FIELDS:
if f == "primary_use":
continue
if abs(float(ra[f]) - float(rb[f])) > 1e-9:
result["numeric_mismatches"].append({"index": i, "field": f, "a": ra[f], "b": rb[f], "name": rb["name"]})
if ra["name_first_line"] != rb["name_first_line"]:
result["name_first_line_mismatches"].append({"index": i, "a": ra["name_first_line"], "b": rb["name_first_line"]})
result["pass"] = result["count_match"] and not result["numeric_mismatches"] and not result["name_first_line_mismatches"]
return result
def main(source_id: str) -> None:
path = _latest_file(source_id)
with pdfplumber.open(path) as pdf:
a = parse_text(pdf)
b = parse_words(pdf)
g = gate(a, b)
df = pd.DataFrame(b)
# Arithmetic check of the publisher's own figure: util = cofte / school_capacity (rounded).
df["util_recomputed"] = (df.cofte / df.school_capacity.where(df.school_capacity > 0) * 100).round(0)
df["util_arith_ok"] = (df.util_recomputed - df.utilization_pct).abs() <= 1
df.loc[df.school_capacity == 0, "util_arith_ok"] = True
# Attach school numbers by name.
nm = NameMatcher(NAME_OVERRIDES)
matched = df.name.map(nm.match)
df["school_number"] = [m[0] for m in matched]
df["match_method"] = [m[1] for m in matched]
df["match_score"] = [m[2] for m in matched]
df.insert(0, "source_id", source_id)
OUT_DIR.mkdir(parents=True, exist_ok=True)
out_csv = OUT_DIR / f"{source_id}.csv"
df.to_csv(out_csv, index=False)
summary = {
"source_id": source_id, "file": str(path.relative_to(REPO_ROOT)), "gate": g,
"records": len(df), "records_with_capacity": int((df.school_capacity > 0).sum()),
"util_arith_failures": int((~df.util_arith_ok).sum()),
"matched": int(df.school_number.notna().sum()),
"unmatched_with_capacity": df[(df.school_number.isna()) & (df.school_capacity > 0)][["name", "primary_use", "school_capacity", "match_method", "match_score"]].to_dict("records"),
"duplicate_school_numbers": df[df.school_number.notna()].school_number[df[df.school_number.notna()].school_number.duplicated(keep=False)].unique().tolist(),
}
(OUT_DIR / f"{source_id}.gate.json").write_text(json.dumps(summary, indent=1, default=str))
print(json.dumps({k: v for k, v in summary.items() if k not in ("unmatched_with_capacity",)}, indent=1, default=str))
print("unmatched with capacity:", len(summary["unmatched_with_capacity"]))
for u in summary["unmatched_with_capacity"][:40]:
print(" ", u)
if not g["pass"]:
sys.exit(2)
if __name__ == "__main__":
main(sys.argv[1] if len(sys.argv) > 1 else "fldoe_fish_los_dade_2024-25")