add medical_center flag: clinical/HCOMP faculty and medical-center clinical staff

Orthogonal boolean column alongside category, computed in classifier.py's
new is_medical_center(). Always flags MED CTR-tagged titles and the
existing HS CLIN PROF/-HCOMP patterns regardless of campus; additionally
flags generic clinical-workforce titles (physicians, nurses, dentists,
pharmacists, therapists, techs, etc.) only at the five campuses that
operate a UC-owned academic medical center (UCSF, UCLA, UCD, UCI, UCSD),
since those titles are indistinguishable from Student Health Services
staff elsewhere. SHS-tagged titles are always excluded.

Threaded through as a --group-by dimension for --aggregate-output and
as a column on --top-earners-output rows, alongside the per-employee CSV.

classifier.py

@@ -376,3 +376,69 @@ def classify_title(raw_title: str, gross_pay: float = 0.0,
         return "PROF_STAFF"
     else:
         return "SUPPORT_STAFF"
+
+
+# ---------------------------------------------------------------------------
+# MEDICAL CENTER FLAG
+# ---------------------------------------------------------------------------
+# Orthogonal to classify_title()'s category: marks employees working at a UC
+# academic medical center (clinical/HCOMP faculty, physicians, nurses, and
+# the broader allied-health/hospital-operations workforce), regardless of
+# which of the 8 categories they land in above.
+#
+# Built from actual UC-wages title text, not guessed. Two signals:
+#
+# 1. ALWAYS (any campus): titles that are unambiguous wherever they appear.
+#      - "MED CTR" literal substring -- only ever seen on hospital-building
+#        facilities/exec titles (MED CTR COOK, CEO MED CTR, etc.), never on
+#        clinical professional titles, never at non-medical-center campuses.
+#      - HS CLIN PROF / HS CLIN INSTR series (Health Sciences clinical
+#        faculty -- same titles already routed to NON_SENATE_ACADEMIC above).
+#      - Bare "-HCOMP" ladder titles (Health Sciences Compensation Plan --
+#        same titles already routed to SENATE_FACULTY above).
+#
+# 2. ONLY at the 5 campuses that operate a UC-owned academic medical center
+#    (UCSF, UCLA, UCD, UCI, UCSD -- UC Riverside has a medical school but no
+#    UC-owned teaching hospital of its own): generic clinical-workforce
+#    titles (physicians, nurses, dentists, pharmacists, therapists, techs,
+#    etc.). These same titles show up identically at every campus (student
+#    health services exists everywhere), so outside these 5 campuses they
+#    can't be told apart from student health staff by title text alone.
+#
+# EXCLUDED everywhere: any title containing "SHS" (Student Health Services)
+# -- present even at UCSF (e.g. "HC ADM SHS MGR 1"), a distinct campus-wide
+# operation from the hospital system.
+MEDICAL_CENTER_CAMPUSES = frozenset({"UCSF", "UCLA", "UCD", "UCI", "UCSD"})
+
+_MEDICAL_ALWAYS_RX = _rx(
+    r"MED CTR",
+    r"^HS\s+(?:ASST\s+|ASSOC\s+)?CLIN (?:PROF|INSTR)\b",
+    r"-HCOMP\b",
+)
+
+_MEDICAL_CLINICAL_STAFF_RX = _rx(
+    r"\bPHYSICIAN\b", r"\bPHYSCN\b", r"\bNURSE\b", r"\bANESTHET",
+    r"\bDENTIST\b", r"\bPHARM", r"\bTHERAPIST\b", r"\bRESP THER",
+    r"\bPERFUSION", r"\bMIDWIFE\b", r"\bAUDIOLOG", r"\bOPTOMETR",
+    r"\bDIETIT", r"\bCLIN SOCIAL", r"\bRADIOLOG", r"\bCLIN LAB",
+    r"\bPHLEBOTOM", r"\bMED TCHN\b", r"\bNUC MED TCHNO\b", r"\bDIALYSIS\b",
+    r"\bULTRASOUND\b", r"\bECHOCARD", r"\bEEG\b", r"\bCATH LAB\b",
+    r"\bPSYCHIAT",
+)
+
+_MEDICAL_EXCLUDE_RX = _rx(r"\bSHS\b")
+
+
+def is_medical_center(raw_title: str, campus: str) -> bool:
+    t = normalise(raw_title)
+
+    if any(rx.search(t) for rx in _MEDICAL_EXCLUDE_RX):
+        return False
+
+    if any(rx.search(t) for rx in _MEDICAL_ALWAYS_RX):
+        return True
+
+    if campus in MEDICAL_CENTER_CAMPUSES:
+        return any(rx.search(t) for rx in _MEDICAL_CLINICAL_STAFF_RX)
+
+    return False

uc_salary_pipeline.py

@@ -33,18 +33,23 @@ expects, and writes the result.
 OUTPUT
 ------
 Per-employee CSV (--output), one row per employee per year:
-  year, campus, title, category, category_label,
+  year, campus, medical_center, title, category, category_label,
   gross_pay, regular_pay, overtime_pay, other_pay, first_name, last_name
 
+`medical_center` (true/false) marks employees working at a UC academic
+medical center -- clinical/HCOMP faculty, physicians, nurses, and the
+broader hospital workforce -- independent of `category`. See classifier.py's
+is_medical_center() docstring for how it's determined.
+
 Optionally, an aggregate CSV (--aggregate-output) with one row per
-category (broken down further by --group-by year/campus):
+category (broken down further by --group-by year/campus/medical_center):
-  [year] [campus] category, category_label,
+  [year] [campus] [medical_center] category, category_label,
   employee_count, total_gross_pay, mean_gross_pay, median_gross_pay
 
 Optionally, a top-earners CSV (--top-earners-output) with the --top-n
 (default 10) highest gross_pay employees per category (also broken down
 further by --group-by, e.g. per year):
-  category, category_label, rank, year, campus, title,
+  category, category_label, rank, year, campus, medical_center, title,
   gross_pay, regular_pay, overtime_pay, other_pay, first_name, last_name
 """
 
@@ -76,11 +81,13 @@ logging.basicConfig(
 log = logging.getLogger(__name__)
 
 OUTPUT_COLUMNS = [
-    "year", "campus", "title", "category", "category_label",
+    "year", "campus", "medical_center", "title", "category", "category_label",
     "gross_pay", "regular_pay", "overtime_pay", "other_pay",
     "first_name", "last_name",
 ]
 
+GROUP_BY_DIMS = ("year", "campus", "medical_center")
+
 
 def _dollar(s) -> float:
     try:
@@ -140,10 +147,6 @@ def process_file(path: Path, campus: str, year: int, writer: csv.writer,
     with open(path, encoding="utf-8") as fh:
         data = json.load(fh)
 
-    group_prefix = tuple(
-        year if dim == "year" else campus for dim in group_dims
-    )
-
     count = 0
     for record in data.get("records", []):
         parsed = parse_uc_wages_record(record)
@@ -155,21 +158,26 @@ def process_file(path: Path, campus: str, year: int, writer: csv.writer,
             senior_threshold=senior_threshold,
             support_threshold=support_threshold,
         )
+        medical_center = classifier.is_medical_center(parsed["title"], campus)
         writer.writerow([
-            year, campus, parsed["title"], category,
+            year, campus, medical_center, parsed["title"], category,
             labels[category],
             round(parsed["gross_pay"], 2), round(parsed["regular_pay"], 2),
             round(parsed["overtime_pay"], 2), round(parsed["other_pay"], 2),
             parsed["first_name"], parsed["last_name"],
         ])
         count += 1
+
+        group_values = {"year": year, "campus": campus, "medical_center": medical_center}
+        group_prefix = tuple(group_values[dim] for dim in group_dims)
+
         if agg is not None:
             agg.setdefault(group_prefix + (category,), []).append(parsed["gross_pay"])
         if console_stats is not None:
             console_stats.setdefault((year, category), []).append(parsed["gross_pay"])
         if top_earners is not None:
             entry = (parsed["gross_pay"], next(tie_breaker),
-                    (year, campus, parsed["title"], parsed["gross_pay"],
+                    (year, campus, medical_center, parsed["title"], parsed["gross_pay"],
                      parsed["regular_pay"], parsed["overtime_pay"],
                      parsed["other_pay"], parsed["first_name"], parsed["last_name"]))
             heap = top_earners.setdefault(group_prefix + (category,), [])
@@ -206,8 +214,8 @@ def write_aggregate_csv(agg: dict, group_dims: tuple, labels: dict, outpath: Pat
 def write_top_earners_csv(top_earners: dict, labels: dict, outpath: Path) -> None:
     """Write the top-N earners per (group_dims..., category) key, ranked
     highest gross_pay first within each group."""
-    header = ["category", "category_label", "rank", "year", "campus", "title",
+    header = ["category", "category_label", "rank", "year", "campus", "medical_center",
-              "gross_pay", "regular_pay", "overtime_pay", "other_pay",
+              "title", "gross_pay", "regular_pay", "overtime_pay", "other_pay",
               "first_name", "last_name"]
 
     with open(outpath, "w", newline="", encoding="utf-8") as fh:
@@ -217,10 +225,10 @@ def write_top_earners_csv(top_earners: dict, labels: dict, outpath: Path) -> Non
             category = key[-1]
             ranked = sorted(top_earners[key], key=lambda e: e[0], reverse=True)
             for rank, (gross_pay, _, row) in enumerate(ranked, start=1):
-                year, campus, title, gross, regular, overtime, other, first, last = row
+                year, campus, medical_center, title, gross, regular, overtime, other, first, last = row
                 writer.writerow([
                     category, labels[category], rank,
-                    year, campus, title, round(gross, 2), round(regular, 2),
+                    year, campus, medical_center, title, round(gross, 2), round(regular, 2),
                     round(overtime, 2), round(other, 2), first, last,
                 ])
     log.info("Top earners CSV -> %s", outpath)
@@ -293,14 +301,14 @@ def main():
                         help="If given, also write an aggregate CSV here with "
                              "employee count, total/mean/median gross pay per "
                              "category (see --group-by).")
-    parser.add_argument("--group-by", nargs="*", choices=["year", "campus"],
+    parser.add_argument("--group-by", nargs="*", choices=list(GROUP_BY_DIMS),
                         default=["year"],
                         help="Additional breakdown dimensions for "
                              "--aggregate-output and --top-earners-output, "
-                             "alongside category (default: year). Pass both "
+                             "alongside category (default: year). E.g. "
-                             "for a year x campus x category breakdown, or "
+                             "--group-by year campus medical_center for the "
-                             "nothing to collapse everything to one group "
+                             "finest breakdown, or nothing to collapse "
-                             "per category.")
+                             "everything to one group per category.")
     parser.add_argument("--top-earners-output",
                         help="If given, also write a CSV of the --top-n "
                              "highest gross_pay employees per category (see "
@@ -332,7 +340,7 @@ def main():
         log.info("--skip-download set, classifying cached data only.")
 
     labels = classifier.category_labels(args.senior_threshold, args.support_threshold)
-    group_dims = tuple(dim for dim in ("year", "campus") if dim in args.group_by)
+    group_dims = tuple(dim for dim in GROUP_BY_DIMS if dim in args.group_by)
     agg = {} if args.aggregate_output else None
     top_earners = {} if args.top_earners_output else None
     tie_breaker = itertools.count() if top_earners is not None else None