diff --git a/CODEBOOK.md b/CODEBOOK.md index 2c128a5..a70efe7 100644 --- a/CODEBOOK.md +++ b/CODEBOOK.md @@ -42,9 +42,9 @@ Each image or region-of-interest must be assigned exactly one primary class. | Label | Description | |------------------------|-----------------------------------------------------------------------------| -| No Precipitation | No significant return; background noise only. The image will only have blue and black colors.| +| No Precipitation | No significant return; background noise only. The image will only have blue and black colors. The percentage of gates greater than 50 dBZ must not exceed 0.002 percent. If it does exceed 0.002 percent, then classify as Isolated Convection. | | Stratiform Precipitation | The image must have no pink colors. Green, yellow and red colors are present in a widespread blob. The percentage of gates greater than 50 dBZ must not exceed 0.02 percent. If it does exceed 0.02 percent, then classify as a mesoscale convective system. | -| Isolated Convection | The image must have regions of dark red and pink colors. These dark red and pink regions must be separated by regions of black and blue, with no connection to other dark red and pink regions through yellow regions. Over half of the image must be blue or black. The percentage of gates with reflectivity greater than 30 dBZ must not exceed 10 percent. If it does exceed 10 percent, then classify as a mesoscale convective system. | +| Isolated Convection | The image must have regions of yellow, red and pink colors. These dark red and pink regions must be separated by regions of black and blue, with no connection to other dark red and pink regions through yellow regions. Over half of the image must be blue or black. The percentage of gates with reflectivity greater than 30 dBZ must not exceed 1 percent. If it does exceed 1 percent, then classify as a mesoscale convective system. | | Mesoscale Convective System | A string or connected cluster of dark red and pink colors must be present in the image. This string can take on a curved structure. There can be more than one such string or cluster in the image. The dark red and pink colors in the clusters must be connected by yellow regions. | | Ambiguous / Uncertain | Cannot be classified with confidence. | diff --git a/lars/nepho/__init__.py b/lars/nepho/__init__.py index 3a7b919..99eca9a 100644 --- a/lars/nepho/__init__.py +++ b/lars/nepho/__init__.py @@ -1,3 +1,4 @@ from . import models # noqa: F401 from .config import config, Config # noqa: F401 -from .inference import label_radar_data, DEFAULT_CATEGORIES, CODEBOOK_CATEGORIES, CODEBOOK_GUIDELINES, categories_from_codebook, guidelines_from_codebook # noqa: F401 +from .inference import label_radar_data, DEFAULT_CATEGORIES, CODEBOOK_CATEGORIES, CODEBOOK_GUIDELINES, CODEBOOK_CRITERIA, CODEBOOK_COLOR_CRITERIA, CODEBOOK_COLORMAP, COLOR_DBZ_RANGE, DEFAULT_VMIN, DEFAULT_VMAX, categories_from_codebook, guidelines_from_codebook, criteria_from_codebook, color_criteria_from_codebook, colormap_from_codebook # noqa: F401 +from .tracking import compute_validation_metrics, log_run_to_mlflow, codebook_hash # noqa: F401 diff --git a/lars/nepho/inference.py b/lars/nepho/inference.py index 11282da..4bbc835 100644 --- a/lars/nepho/inference.py +++ b/lars/nepho/inference.py @@ -2,6 +2,8 @@ import os import re +from ..preprocessing.labels import apply_criteria_to_labels + DEFAULT_CATEGORIES = {"No precipitation": "No echoes greater than 10 dBZ present. A circle of echoes near radar site may be present due to ground clutter.", "Stratiform rain": "Widespread echoes between 0 and 35 dBZ, not present as a circular pattern around the radar site.", "Scattered Convection": "Present as isolated to scattered cells with reflectivities between 35-65 dBZ", @@ -129,6 +131,338 @@ def guidelines_from_codebook(codebook_path): return guidelines +_CRITERION_PATTERN = re.compile( + r"\b(percentage|number)\s+of\s+gates\s+" + r"(?:with\s+reflectivity\s+)?" + r"greater\s+than\s+(\d+(?:\.\d+)?)\s*dBZ\s+" + r"must\s+not\s+exceed\s+(\d+(?:\.\d+)?)(?:\s*percent)?", + re.IGNORECASE, +) + +_RECLASSIFY_PATTERN = re.compile( + r"\bIf\s+(?:it\s+(?:does\s+)?)?exceeds?\s+\d+(?:\.\d+)?(?:\s*percent)?\s*,?\s*" + r"(?:then\s+)?classify\s+as\s+(?:a\s+|an\s+|the\s+)?([^.]+?)\s*\.", + re.IGNORECASE, +) + + +def _format_threshold(threshold_str): + val = float(threshold_str) + return int(val) if val.is_integer() else val + + +def criteria_from_codebook(codebook_path): + """ + Parse hard quantitative criteria from category descriptions in a + LARS-format codebook markdown file. + + For every category description, this function looks for sentences of the + form + + "the {percentage|number} of gates [with reflectivity] greater than + X dBZ must not exceed Y[ percent]" + + and, optionally, an immediately following reclassification clause + + "If it does exceed Y[ percent], then classify as Z." + + Both percent-based (``pct_gates_dbz``) and count-based + (``n_gates_dbz``) criteria are supported; the column name is derived + from the matched phrasing. + + Parameters + ---------- + codebook_path : str + Path to the codebook markdown file. + + Returns + ------- + dict + Mapping of label name → list of criterion dicts. Each criterion has: + + * ``field`` (str): column name to compare against + (e.g. ``"pct_gates_50dbz"`` or ``"n_gates_30dbz"``). + * ``kind`` (str): ``"pct"`` or ``"count"``. + * ``threshold_dbz`` (int or float): the reflectivity threshold. + * ``max_value`` (float): the maximum allowed value; rows whose + ``field`` is strictly greater than this violate the criterion. + * ``reclassify_as`` (str or None): canonical label to assign when + the criterion is violated, or ``None`` if the codebook does not + specify a target. + """ + categories = categories_from_codebook(codebook_path) + canonical = {label.lower().strip(): label for label in categories} + + criteria = {} + for label, description in categories.items(): + rules = [] + for m in _CRITERION_PATTERN.finditer(description): + kind_word = m.group(1).lower() + threshold = _format_threshold(m.group(2)) + max_value = float(m.group(3)) + kind = "pct" if kind_word.startswith("percent") else "count" + field_prefix = "pct_gates" if kind == "pct" else "n_gates" + field = f"{field_prefix}_{threshold}dbz" + + after = description[m.end():] + rm = _RECLASSIFY_PATTERN.search(after) + target = rm.group(1).strip() if rm else None + reclassify_as = canonical.get(target.lower(), target) if target else None + + rules.append({ + "field": field, + "kind": kind, + "threshold_dbz": threshold, + "max_value": max_value, + "reclassify_as": reclassify_as, + }) + if rules: + criteria[label] = rules + + return criteria + + +DEFAULT_VMIN = -20 +DEFAULT_VMAX = 60 + +_COLORMAP_LINE_RE = re.compile(r"^[^\n]*Color\s*scale[^\n]*$", re.IGNORECASE | re.MULTILINE) +_COLORMAP_NAME_RE = re.compile(r"([A-Za-z0-9_]+)\s+colormap", re.IGNORECASE) +_VMIN_RE = re.compile(r"vmin\s*=\s*(-?\d+(?:\.\d+)?)", re.IGNORECASE) +_VMAX_RE = re.compile(r"vmax\s*=\s*(-?\d+(?:\.\d+)?)", re.IGNORECASE) + + +def _format_number(num_str): + val = float(num_str) + return int(val) if val.is_integer() else val + + +def colormap_from_codebook(codebook_path): + """ + Parse the color-scale specification from a LARS-format codebook. + + The function looks for the ``Color scale`` line in the *Image Format* + section (Section 2.2), e.g.:: + + - **Color scale:** ChaseSpectral colormap with vmin=-10 and vmax=60 + + and extracts the colormap name and its ``vmin`` / ``vmax`` bounds. Any + field that is missing from the codebook falls back to its default: + ``colormap=None``, ``vmin=DEFAULT_VMIN`` (-20), ``vmax=DEFAULT_VMAX`` (60). + + Parameters + ---------- + codebook_path : str + Path to the codebook markdown file. + + Returns + ------- + dict + Mapping with keys: + + * ``colormap`` (str or None): the colormap name, or ``None`` if the + codebook does not name one. + * ``vmin`` (int or float): lower bound of the color scale. + * ``vmax`` (int or float): upper bound of the color scale. + """ + with open(codebook_path, "r") as f: + text = f.read() + + colormap = None + vmin = DEFAULT_VMIN + vmax = DEFAULT_VMAX + + line_match = _COLORMAP_LINE_RE.search(text) + if line_match: + line = line_match.group(0) + name_match = _COLORMAP_NAME_RE.search(line) + if name_match: + colormap = name_match.group(1) + vmin_match = _VMIN_RE.search(line) + if vmin_match: + vmin = _format_number(vmin_match.group(1)) + vmax_match = _VMAX_RE.search(line) + if vmax_match: + vmax = _format_number(vmax_match.group(1)) + + return {"colormap": colormap, "vmin": vmin, "vmax": vmax} + + +COLOR_DBZ_RANGE = { + "blue": (None, 10), + "black": (None, 10), + "green": (10, 30), + "yellow": (30, 40), + "red": (40, 50), + "dark red": (40, 50), + "pink": (50, None), +} + +LOW_REFLECTIVITY_COLORS = {"blue", "black"} + +DEFAULT_COLOR_PRESENCE_THRESHOLD = 0.1 +DEFAULT_COLOR_ABSENCE_TOLERANCE = 0.5 + +_COLOR_NAMES_RE = "|".join( + re.escape(c) for c in sorted(COLOR_DBZ_RANGE, key=len, reverse=True) +) +_COLOR_LIST_RE = ( + rf"((?:{_COLOR_NAMES_RE})" + rf"(?:\s*(?:,\s*|\s+and\s+|\s+or\s+)(?:{_COLOR_NAMES_RE}))*)" +) +_COLOR_NAME_FINDER = re.compile(_COLOR_NAMES_RE, re.IGNORECASE) +_COLOR_ABSENCE_RE = re.compile( + rf"\bno\s+{_COLOR_LIST_RE}\s*(?:colors?)?\b", re.IGNORECASE +) +_COLOR_EXCLUSIVE_RE = re.compile( + rf"\bonly\s+have\s+{_COLOR_LIST_RE}\s*(?:colors?)?\b", re.IGNORECASE +) +_COLOR_PRESENCE_RE = re.compile( + rf"\b(?:{_COLOR_LIST_RE}\s+(?:colors?\s+)?(?:are|must\s+be)\s+present" + rf"|(?:must\s+have|have)\s+(?:regions?\s+of\s+)?{_COLOR_LIST_RE})\b", + re.IGNORECASE, +) +_COLOR_DOMINANCE_RE = re.compile( + rf"\b(?:over|more\s+than)\s+half\s+(?:of\s+the\s+image\s+)?must\s+be\s+{_COLOR_LIST_RE}\b", + re.IGNORECASE, +) + + +def _extract_colors(text): + return [m.group(0).lower() for m in _COLOR_NAME_FINDER.finditer(text)] + + +def color_criteria_from_codebook(codebook_path, + presence_threshold=DEFAULT_COLOR_PRESENCE_THRESHOLD, + absence_tolerance=DEFAULT_COLOR_ABSENCE_TOLERANCE): + """ + Parse color-based identification criteria from a LARS-format codebook. + + Each category description is split into sentences and matched against a + small set of color phrasings: + + * **Exclusivity** — *"image will only have "* → all gates whose + reflectivity exceeds the highest band of the listed colors must be + ≤ ``absence_tolerance`` percent. + * **Absence** — *"must have no "* → ``pct_gates_dbz`` for that + color must be ≤ ``absence_tolerance`` percent. + * **Presence** — *" are present"* / *"must have [regions of] + "* → ``pct_gates_dbz`` must be ≥ ``presence_threshold`` + percent for each listed color. + * **Dominance** — *"Over half of the image must be "* → for + low-reflectivity colors (blue/black) this maps to + ``pct_gates_10dbz ≤ 50``. + + Color → dBZ band mapping is given by ``COLOR_DBZ_RANGE``, aligned with + the default ``dbz_thresholds=(10, 20, 30, 40, 50)`` used by + ``preprocess_radar_data``. Colors that fall in the low-reflectivity + band (blue/black) cannot express absence or presence on their own and + are skipped except for the exclusivity / dominance patterns. + + Parameters + ---------- + codebook_path : str + presence_threshold : float, optional + Minimum ``pct_gates_*`` value to consider a color present. + absence_tolerance : float, optional + Maximum ``pct_gates_*`` value to consider a color absent. + + Returns + ------- + dict + Mapping of label → list of color rules. Each rule has: + + * ``kind`` (str): ``"min_pct_above"`` or ``"max_pct_above"``. + * ``field`` (str): ``pct_gates_dbz`` column name. + * ``value`` (float): comparison threshold. + * ``colors`` (list of str): colors involved. + * ``phrase`` (str): the codebook sentence the rule was derived from. + """ + categories = categories_from_codebook(codebook_path) + out = {} + for label, description in categories.items(): + rules = [] + for sentence in re.split(r"(?<=[.;])\s+", description): + sentence = sentence.strip() + if not sentence: + continue + + m = _COLOR_EXCLUSIVE_RE.search(sentence) + if m: + colors = _extract_colors(m.group(1)) + allowed_max = max( + (COLOR_DBZ_RANGE[c][1] or 999) + for c in colors if c in COLOR_DBZ_RANGE + ) + if allowed_max < 999: + rules.append({ + "kind": "max_pct_above", + "field": f"pct_gates_{int(allowed_max)}dbz", + "value": absence_tolerance, + "colors": colors, + "phrase": sentence, + }) + continue + + m = _COLOR_ABSENCE_RE.search(sentence) + if m: + for c in _extract_colors(m.group(1)): + lo, _ = COLOR_DBZ_RANGE.get(c, (None, None)) + if lo is None: + continue + rules.append({ + "kind": "max_pct_above", + "field": f"pct_gates_{int(lo)}dbz", + "value": absence_tolerance, + "colors": [c], + "phrase": sentence, + }) + continue + + m = _COLOR_DOMINANCE_RE.search(sentence) + if m: + colors = _extract_colors(m.group(1)) + color_max_hi = max( + (COLOR_DBZ_RANGE[c][1] or 999) + for c in colors if c in COLOR_DBZ_RANGE + ) + if color_max_hi < 999: + rules.append({ + "kind": "max_pct_above", + "field": f"pct_gates_{int(color_max_hi)}dbz", + "value": 50.0, + "colors": colors, + "phrase": sentence, + }) + continue + + m = _COLOR_PRESENCE_RE.search(sentence) + if m: + colors_text = next((g for g in m.groups() if g), "") + for c in _extract_colors(colors_text): + lo, _ = COLOR_DBZ_RANGE.get(c, (None, None)) + if lo is None: + continue + rules.append({ + "kind": "min_pct_above", + "field": f"pct_gates_{int(lo)}dbz", + "value": presence_threshold, + "colors": [c], + "phrase": sentence, + }) + + seen = set() + unique = [] + for r in rules: + key = (r["kind"], r["field"], r["value"]) + if key in seen: + continue + seen.add(key) + unique.append(r) + if unique: + out[label] = unique + return out + + _DEFAULT_CODEBOOK = os.path.join( os.path.dirname(__file__), "..", "..", "CODEBOOK.md" ) @@ -141,10 +475,25 @@ def guidelines_from_codebook(codebook_path): guidelines_from_codebook(_default_codebook_path) if os.path.exists(_default_codebook_path) else None ) +CODEBOOK_CRITERIA = ( + criteria_from_codebook(_default_codebook_path) + if os.path.exists(_default_codebook_path) else None +) +CODEBOOK_COLOR_CRITERIA = ( + color_criteria_from_codebook(_default_codebook_path) + if os.path.exists(_default_codebook_path) else None +) +CODEBOOK_COLORMAP = ( + colormap_from_codebook(_default_codebook_path) + if os.path.exists(_default_codebook_path) else None +) async def label_radar_data(radar_df, model, categories=None, guidelines=None, + criteria=None, color_criteria=None, + mlflow_experiment=None, mlflow_run_name=None, + mlflow_tracking_uri=None, codebook_path=None, site="Bankhead National Forest", - verbose=True, vmin=-20, vmax=60, model_output_dir=None): + verbose=True, vmin=None, vmax=None, model_output_dir=None): """ Label radar data using a given model. @@ -156,7 +505,35 @@ async def label_radar_data(radar_df, model, categories=None, guidelines=None, DEFAULT_CATEGORIES. Pass CODEBOOK_CATEGORIES to use the bundled codebook. guidelines (list of str, optional): Annotator guidelines appended to the prompt. Pass CODEBOOK_GUIDELINES to use the bundled codebook guidelines. + criteria (dict, optional): Hard quantitative criteria as returned by + ``criteria_from_codebook``. When provided, any LLM label whose + ``pct_gates_*`` / ``n_gates_*`` values violate the rules for that + label is overridden in-place; the pre-override label and the rule + that fired are recorded in ``llm_label_original`` and + ``llm_label_criteria_violation``. Pass ``CODEBOOK_CRITERIA`` to + enforce the bundled codebook. + color_criteria (dict, optional): Color-based criteria as returned by + ``color_criteria_from_codebook``. Used only for validation + metric computation when ``mlflow_experiment`` is set; does not + modify any labels. Pass ``CODEBOOK_COLOR_CRITERIA`` to evaluate + against the bundled codebook. + mlflow_experiment (str, optional): If provided, opens an MLflow run + under this experiment and logs params, validation metrics + (reflectivity-criteria + color-criteria violations and label + agreement), the labelled CSV, the confusion matrix, and any raw + model outputs in ``model_output_dir``. Requires the optional + ``mlflow`` dependency. + mlflow_run_name (str, optional): MLflow run name. + mlflow_tracking_uri (str, optional): Forwarded to + ``mlflow.set_tracking_uri``. + codebook_path (str, optional): Path to the codebook used for this run; + hashed and logged for traceability. site: str: Radar site identifier. + vmin, vmax (float, optional): Bounds of the color scale described to the + model in the prompt. When left as ``None`` and ``codebook_path`` is + provided, they are read from the codebook's color-scale spec via + ``colormap_from_codebook``; otherwise they fall back to + ``DEFAULT_VMIN`` (-20) and ``DEFAULT_VMAX`` (60). model_output_dir: str: Directory to save model outputs. Returns @@ -166,6 +543,16 @@ async def label_radar_data(radar_df, model, categories=None, guidelines=None, """ if categories is None: categories = DEFAULT_CATEGORIES + if (vmin is None or vmax is None) and codebook_path is not None: + cmap = colormap_from_codebook(codebook_path) + if vmin is None: + vmin = cmap["vmin"] + if vmax is None: + vmax = cmap["vmax"] + if vmin is None: + vmin = DEFAULT_VMIN + if vmax is None: + vmax = DEFAULT_VMAX prompt = "This is an image of weather radar base reflectivity data." \ f" The radar site is the ARM Facility {site} site." \ " Please classify the weather depicted into one of the following categories: " \ @@ -213,6 +600,30 @@ async def label_radar_data(radar_df, model, categories=None, guidelines=None, if output[-1] == ".": output = output[:-1] radar_df.loc[radar_df["file_path"] == fi, "llm_label"] = output.strip() - - + + if criteria: + radar_df = apply_criteria_to_labels(radar_df, criteria, + label_column="llm_label") + + if mlflow_experiment: + from .tracking import log_run_to_mlflow + log_run_to_mlflow( + radar_df, + experiment=mlflow_experiment, + run_name=mlflow_run_name, + tracking_uri=mlflow_tracking_uri, + params={ + "model": getattr(model, "model_name", type(model).__name__), + "site": site, + "vmin": vmin, + "vmax": vmax, + "n_categories": len(categories), + "criteria_enforced": criteria is not None, + }, + criteria=criteria, + color_criteria=color_criteria, + codebook_path=codebook_path, + model_output_dir=model_output_dir, + ) + return radar_df \ No newline at end of file diff --git a/lars/nepho/tracking.py b/lars/nepho/tracking.py new file mode 100644 index 0000000..70bfaa0 --- /dev/null +++ b/lars/nepho/tracking.py @@ -0,0 +1,308 @@ +"""MLflow experiment tracking for LARS inference runs. + +Provides validation-metric computation (reflectivity-criteria violations, +color-criteria violations, label agreement vs. hand labels) and a single +``log_run_to_mlflow`` entry point that lazily imports mlflow. +""" +import hashlib +import os +import tempfile + +import pandas as pd + + +def _lazy_mlflow(): + try: + import mlflow + except ImportError as e: + raise ImportError( + "mlflow is required for experiment tracking. " + "Install it with `pip install mlflow`." + ) from e + return mlflow + + +def codebook_hash(path): + """Return a short SHA-256 of the codebook file at ``path``, or None.""" + if path is None or not os.path.exists(path): + return None + with open(path, "rb") as f: + return hashlib.sha256(f.read()).hexdigest()[:16] + + +def _safe(name): + """Sanitize a string for use in mlflow metric / artifact names.""" + return "".join(c if c.isalnum() or c in "._-" else "_" for c in str(name)) + + +def _reflectivity_violation_metrics(df, criteria, label_col, prefix): + """Per-rule violation counts and rates against the reflectivity criteria.""" + out = {} + if criteria is None or label_col not in df.columns: + return out + total = len(df) + for label, rules in criteria.items(): + label_mask = df[label_col] == label + n_label = int(label_mask.sum()) + for rule in rules: + field = rule["field"] + if field not in df.columns: + continue + violated = label_mask & (df[field] > rule["max_value"]) + n_viol = int(violated.sum()) + base = (f"{prefix}/{_safe(label)}/" + f"{_safe(field)}_gt_{rule['max_value']}") + out[f"{base}/count"] = n_viol + out[f"{base}/rate_of_label"] = ( + n_viol / n_label if n_label else 0.0 + ) + out[f"{base}/rate_of_total"] = ( + n_viol / total if total else 0.0 + ) + return out + + +def _color_violation_metrics(df, color_criteria, label_col, prefix): + """Per-rule violation counts and rates against the color criteria.""" + out = {} + if color_criteria is None or label_col not in df.columns: + return out + total = len(df) + for label, rules in color_criteria.items(): + label_mask = df[label_col] == label + n_label = int(label_mask.sum()) + for rule in rules: + field = rule["field"] + if field not in df.columns: + continue + kind = rule["kind"] + value = rule["value"] + if kind == "max_pct_above": + violated = label_mask & (df[field] > value) + op = "gt" + elif kind == "min_pct_above": + violated = label_mask & (df[field] < value) + op = "lt" + else: + continue + n_viol = int(violated.sum()) + colors = "_".join(rule.get("colors", [])) or "color" + base = (f"{prefix}/{_safe(label)}/{_safe(colors)}/" + f"{_safe(field)}_{op}_{value}") + out[f"{base}/count"] = n_viol + out[f"{base}/rate_of_label"] = ( + n_viol / n_label if n_label else 0.0 + ) + out[f"{base}/rate_of_total"] = ( + n_viol / total if total else 0.0 + ) + return out + + +def _agreement_metrics(df, hand_col, llm_col): + """Overall accuracy, per-class recall, and Cohen's kappa vs hand labels.""" + out = {} + if hand_col not in df.columns or llm_col not in df.columns: + return out + valid = df[[hand_col, llm_col]].dropna() + valid = valid[(valid[hand_col].astype(str) != "") + & (valid[llm_col].astype(str) != "")] + valid = valid[valid[hand_col].astype(str).str.upper() != "UNKNOWN"] + n = len(valid) + if n == 0: + return out + hand_norm = valid[hand_col].astype(str).str.lower() + llm_norm = valid[llm_col].astype(str).str.lower() + out["agreement/n_compared"] = int(n) + out["agreement/overall_accuracy"] = float((hand_norm == llm_norm).mean()) + try: + from sklearn.metrics import cohen_kappa_score + out["agreement/cohen_kappa"] = float( + cohen_kappa_score(hand_norm, llm_norm) + ) + except Exception: + pass + for cls in sorted(hand_norm.unique()): + mask = hand_norm == cls + if mask.any(): + out[f"agreement/per_class/{_safe(cls)}/recall"] = float( + (llm_norm[mask] == cls).mean() + ) + out[f"agreement/per_class/{_safe(cls)}/n"] = int(mask.sum()) + return out + + +def compute_validation_metrics(df, criteria=None, color_criteria=None, + hand_label_col="label", + llm_label_col="llm_label"): + """ + Compute a flat dict of validation metrics for an inference run. + + Reflectivity-criteria violations and color-criteria violations are + computed twice — once against ``llm_label_col`` and once against + ``hand_label_col`` (when present) — so a run can be evaluated both as + "did the model break the codebook" and "did the codebook itself break + on hand labels". Label-agreement metrics (accuracy, per-class recall, + Cohen's kappa) compare ``hand_label_col`` to ``llm_label_col``, + skipping rows whose hand label is missing or ``"UNKNOWN"``. + + Parameters + ---------- + df : pd.DataFrame + criteria : dict, optional + Reflectivity criteria as returned by + ``lars.nepho.inference.criteria_from_codebook``. + color_criteria : dict, optional + Color criteria as returned by + ``lars.nepho.inference.color_criteria_from_codebook``. + hand_label_col : str, optional + llm_label_col : str, optional + + Returns + ------- + dict + Flat ``{metric_name: numeric_value}`` mapping safe to pass to + ``mlflow.log_metrics``. + """ + metrics = {} + metrics.update(_reflectivity_violation_metrics( + df, criteria, llm_label_col, "reflectivity_violations/llm" + )) + metrics.update(_reflectivity_violation_metrics( + df, criteria, hand_label_col, "reflectivity_violations/hand" + )) + metrics.update(_color_violation_metrics( + df, color_criteria, llm_label_col, "color_violations/llm" + )) + metrics.update(_color_violation_metrics( + df, color_criteria, hand_label_col, "color_violations/hand" + )) + metrics.update(_agreement_metrics(df, hand_label_col, llm_label_col)) + return metrics + + +def _log_confusion_matrix(mlflow, df, tmpdir, hand_col, llm_col): + if hand_col not in df.columns or llm_col not in df.columns: + return + valid = df[[hand_col, llm_col]].dropna() + valid = valid[(valid[hand_col].astype(str) != "") + & (valid[llm_col].astype(str) != "")] + valid = valid[valid[hand_col].astype(str).str.upper() != "UNKNOWN"] + if len(valid) == 0: + return + + import matplotlib + matplotlib.use("Agg", force=False) + import matplotlib.pyplot as plt + from sklearn.metrics import confusion_matrix + + hand_norm = valid[hand_col].astype(str).str.lower() + llm_norm = valid[llm_col].astype(str).str.lower() + labels = sorted(set(hand_norm) | set(llm_norm)) + cm = confusion_matrix(hand_norm, llm_norm, labels=labels) + cm_df = pd.DataFrame(cm, index=labels, columns=labels) + csv_path = os.path.join(tmpdir, "confusion_matrix.csv") + cm_df.to_csv(csv_path) + mlflow.log_artifact(csv_path, artifact_path="metrics") + + from lars.util.confusion_matrix import plot_confusion_matrix + fig, ax = plt.subplots(figsize=(8, 6)) + plot_confusion_matrix(valid, label_col=hand_col, pred_col=llm_col, ax=ax) + fig.tight_layout() + png_path = os.path.join(tmpdir, "confusion_matrix.png") + fig.savefig(png_path, dpi=120) + plt.close(fig) + mlflow.log_artifact(png_path, artifact_path="metrics") + + +def log_run_to_mlflow(radar_df, *, experiment, run_name=None, + tracking_uri=None, params=None, + criteria=None, color_criteria=None, + codebook_path=None, model_output_dir=None, + hand_label_col="label", llm_label_col="llm_label"): + """ + Open an MLflow run, log params + validation metrics + artifacts, close it. + + Artifacts logged + ---------------- + * ``data/labelled.csv`` — the labelled DataFrame including any + ``*_original`` / ``*_criteria_violation`` audit columns. + * ``metrics/confusion_matrix.csv`` and ``.png`` — vs. hand labels (if + enough rows are present). + * ``model_outputs/`` — raw text outputs from the LLM, if + ``model_output_dir`` is provided. + + Parameters + ---------- + radar_df : pd.DataFrame + Labelled DataFrame (post-inference, post-criteria). + experiment : str + MLflow experiment name. Created if missing. + run_name : str, optional + tracking_uri : str, optional + Forwarded to ``mlflow.set_tracking_uri`` if provided. + params : dict, optional + Extra params merged with the standard ``n_rows`` / + ``codebook_hash`` set. + criteria, color_criteria : dict, optional + Passed to ``compute_validation_metrics``. + codebook_path : str, optional + Hashed and logged as ``codebook_hash`` for traceability. + model_output_dir : str, optional + Logged as the ``model_outputs`` artifact directory. + + Returns + ------- + str + The MLflow run ID. + """ + mlflow = _lazy_mlflow() + if tracking_uri: + mlflow.set_tracking_uri(tracking_uri) + mlflow.set_experiment(experiment) + + with mlflow.start_run(run_name=run_name) as run: + all_params = {"n_rows": len(radar_df)} + ch = codebook_hash(codebook_path) + if ch: + all_params["codebook_hash"] = ch + all_params["codebook_path"] = codebook_path + if params: + all_params.update(params) + mlflow.log_params({ + k: str(v)[:500] for k, v in all_params.items() if v is not None + }) + + metrics = compute_validation_metrics( + radar_df, + criteria=criteria, + color_criteria=color_criteria, + hand_label_col=hand_label_col, + llm_label_col=llm_label_col, + ) + numeric_metrics = { + k: float(v) for k, v in metrics.items() + if isinstance(v, (int, float)) + } + if numeric_metrics: + mlflow.log_metrics(numeric_metrics) + + with tempfile.TemporaryDirectory() as td: + csv_path = os.path.join(td, "labelled.csv") + radar_df.to_csv(csv_path, index=False) + mlflow.log_artifact(csv_path, artifact_path="data") + try: + _log_confusion_matrix( + mlflow, radar_df, td, hand_label_col, llm_label_col + ) + except Exception as e: + mlflow.log_param( + "confusion_matrix_error", str(e)[:500] + ) + + if model_output_dir and os.path.isdir(model_output_dir): + mlflow.log_artifacts( + model_output_dir, artifact_path="model_outputs" + ) + + return run.info.run_id diff --git a/lars/preprocessing/__init__.py b/lars/preprocessing/__init__.py index 54de16f..4e8770d 100644 --- a/lars/preprocessing/__init__.py +++ b/lars/preprocessing/__init__.py @@ -1,2 +1,2 @@ from .radar_preprocessing import preprocess_radar_data # noqa: F401 -from .labels import load_labels, save_labels, change_file_path, copy_labels # noqa: F401 \ No newline at end of file +from .labels import load_labels, save_labels, change_file_path, copy_labels, apply_criteria_to_labels # noqa: F401 \ No newline at end of file diff --git a/lars/preprocessing/labels.py b/lars/preprocessing/labels.py index 3700b0a..bd9d55b 100644 --- a/lars/preprocessing/labels.py +++ b/lars/preprocessing/labels.py @@ -80,6 +80,79 @@ def copy_labels(source_df, target_df, match_on='time', label_column='label'): return target_df +def apply_criteria_to_labels(df, criteria, label_column='label', + original_column=None, violation_column=None): + """ + Reclassify rows whose label violates a hard criterion from the codebook. + + For each rule attached to a label in ``criteria``, every row currently + carrying that label is checked against the rule's ``field`` and + ``max_value``. Rows whose value is strictly greater than ``max_value`` + have their label overwritten with the rule's ``reclassify_as`` target + (or kept unchanged if the target is missing). The pre-override label + and a short description of the rule that fired are recorded in + ``original_column`` and ``violation_column`` for auditability. + + Parameters + ---------- + df : pd.DataFrame + Must contain ``label_column`` and every ``field`` referenced by the + criteria (e.g. ``pct_gates_50dbz``, ``n_gates_30dbz``). Rules + referencing missing columns are skipped. + criteria : dict + Mapping of label → list of criterion dicts as returned by + ``lars.nepho.inference.criteria_from_codebook``. + label_column : str, optional + Column to read and update. Default ``'label'``. Use ``'llm_label'`` + to enforce hard criteria on model output. + original_column : str, optional + Column to record the pre-override label. + Defaults to ``f"{label_column}_original"``. + violation_column : str, optional + Column to record the rule that fired. + Defaults to ``f"{label_column}_criteria_violation"``. + + Returns + ------- + pd.DataFrame + A copy of ``df`` with possibly updated labels and the two audit + columns populated for any rows that were reclassified. + """ + if original_column is None: + original_column = f"{label_column}_original" + if violation_column is None: + violation_column = f"{label_column}_criteria_violation" + + df = df.copy() + if original_column not in df.columns: + df[original_column] = pd.NA + if violation_column not in df.columns: + df[violation_column] = pd.NA + + for label, rules in criteria.items(): + label_mask = df[label_column] == label + if not label_mask.any(): + continue + for rule in rules: + field = rule["field"] + if field not in df.columns: + continue + violated = label_mask & (df[field] > rule["max_value"]) + if not violated.any(): + continue + new_label = rule["reclassify_as"] or label + unit = "percent" if rule["kind"] == "pct" else "gates" + note = (f"{label}: {field} > {rule['max_value']} {unit}" + f" -> {new_label}") + no_original = df[original_column].isna() + df.loc[violated & no_original, original_column] = label + df.loc[violated, violation_column] = note + df.loc[violated, label_column] = new_label + label_mask = label_mask & ~violated + + return df + + def save_labels(label_df, output_file): """ Save labels to a CSV file. diff --git a/tests/test_validation_tracking.py b/tests/test_validation_tracking.py new file mode 100644 index 0000000..fe12196 --- /dev/null +++ b/tests/test_validation_tracking.py @@ -0,0 +1,208 @@ +import os +import sys +import tempfile +from unittest.mock import MagicMock + +import pandas as pd +import pytest + + +CODEBOOK_PATH = os.path.normpath( + os.path.join(os.path.dirname(__file__), "..", "CODEBOOK.md") +) + + +def test_color_criteria_from_codebook_covers_all_labels(): + from lars.nepho.inference import color_criteria_from_codebook + + rules = color_criteria_from_codebook(CODEBOOK_PATH) + assert set(rules) == { + "No Precipitation", + "Stratiform Precipitation", + "Isolated Convection", + "Mesoscale Convective System", + } + + +def test_color_criteria_no_pink_on_stratiform(): + from lars.nepho.inference import color_criteria_from_codebook + + rules = color_criteria_from_codebook(CODEBOOK_PATH) + strat = rules["Stratiform Precipitation"] + no_pink = [ + r for r in strat + if r["kind"] == "max_pct_above" + and r["field"] == "pct_gates_50dbz" + and "pink" in r["colors"] + ] + assert len(no_pink) == 1 + + +def test_color_criteria_dominance_for_isolated(): + from lars.nepho.inference import color_criteria_from_codebook + + rules = color_criteria_from_codebook(CODEBOOK_PATH) + iso = rules["Isolated Convection"] + dom = [ + r for r in iso + if r["kind"] == "max_pct_above" + and r["field"] == "pct_gates_10dbz" + and r["value"] == 50.0 + ] + assert len(dom) == 1 + assert "blue" in dom[0]["colors"] and "black" in dom[0]["colors"] + + +def test_color_criteria_exclusivity_for_no_precip(): + from lars.nepho.inference import color_criteria_from_codebook + + rules = color_criteria_from_codebook(CODEBOOK_PATH) + np_rules = rules["No Precipitation"] + excl = [ + r for r in np_rules + if r["kind"] == "max_pct_above" and r["field"] == "pct_gates_10dbz" + ] + assert len(excl) == 1 + + +def test_colormap_from_codebook_parses_name_and_bounds(): + from lars.nepho.inference import colormap_from_codebook + + cmap = colormap_from_codebook(CODEBOOK_PATH) + assert cmap["colormap"] == "ChaseSpectral" + assert cmap["vmin"] == -10 + assert cmap["vmax"] == 60 + + +def test_colormap_from_codebook_falls_back_to_defaults(): + from lars.nepho.inference import ( + colormap_from_codebook, + DEFAULT_VMIN, + DEFAULT_VMAX, + ) + + with tempfile.NamedTemporaryFile( + mode="w", suffix=".md", delete=False + ) as f: + f.write("# Codebook\n\nNo color scale specified here.\n") + path = f.name + try: + cmap = colormap_from_codebook(path) + assert cmap["colormap"] is None + assert cmap["vmin"] == DEFAULT_VMIN == -20 + assert cmap["vmax"] == DEFAULT_VMAX == 60 + finally: + os.unlink(path) + + +def test_compute_validation_metrics_counts_violations(): + from lars.nepho.tracking import compute_validation_metrics + + criteria = { + "Stratiform Precipitation": [{ + "field": "pct_gates_50dbz", "kind": "pct", + "threshold_dbz": 50, "max_value": 0.02, + "reclassify_as": "Mesoscale Convective System", + }], + } + color_criteria = { + "Stratiform Precipitation": [{ + "kind": "max_pct_above", "field": "pct_gates_50dbz", + "value": 0.5, "colors": ["pink"], + "phrase": "must have no pink colors", + }], + } + df = pd.DataFrame({ + "label": ["Stratiform Precipitation"] * 3, + "llm_label": ["Stratiform Precipitation", + "Stratiform Precipitation", + "Mesoscale Convective System"], + "pct_gates_50dbz": [0.0, 1.0, 5.0], + }) + m = compute_validation_metrics(df, criteria=criteria, + color_criteria=color_criteria) + + refl_keys = [k for k in m if k.startswith("reflectivity_violations/llm") + and k.endswith("/count")] + assert refl_keys, f"expected reflectivity llm count metric, got: {list(m)}" + assert m[refl_keys[0]] == 1 + + color_keys = [k for k in m if k.startswith("color_violations/llm") + and k.endswith("/count")] + assert color_keys + assert m[color_keys[0]] == 1 + + assert m["agreement/n_compared"] == 3 + assert m["agreement/overall_accuracy"] == pytest.approx(2 / 3) + + +def test_compute_validation_metrics_skips_unknown_hand_labels(): + from lars.nepho.tracking import compute_validation_metrics + + df = pd.DataFrame({ + "label": ["UNKNOWN", "Stratiform Precipitation"], + "llm_label": ["Stratiform Precipitation", "Stratiform Precipitation"], + }) + m = compute_validation_metrics(df) + assert m["agreement/n_compared"] == 1 + assert m["agreement/overall_accuracy"] == 1.0 + + +def test_compute_validation_metrics_handles_missing_columns(): + from lars.nepho.tracking import compute_validation_metrics + + df = pd.DataFrame({"file_path": ["a", "b"], "llm_label": ["x", "y"]}) + m = compute_validation_metrics(df) + assert m == {} + + +def test_log_run_to_mlflow_raises_clear_error_without_mlflow(monkeypatch): + from lars.nepho import tracking + + monkeypatch.setitem(sys.modules, "mlflow", None) + df = pd.DataFrame({"label": ["x"], "llm_label": ["x"]}) + with pytest.raises(ImportError, match="mlflow is required"): + tracking.log_run_to_mlflow(df, experiment="exp") + + +def test_codebook_hash_is_stable(): + from lars.nepho.tracking import codebook_hash + + assert codebook_hash(CODEBOOK_PATH) == codebook_hash(CODEBOOK_PATH) + assert codebook_hash("/no/such/path") is None + + +def test_log_run_to_mlflow_calls_expected_apis(monkeypatch): + from lars.nepho import tracking + + fake = MagicMock() + fake_run = MagicMock() + fake_run.info.run_id = "rid-123" + cm = MagicMock() + cm.__enter__ = MagicMock(return_value=fake_run) + cm.__exit__ = MagicMock(return_value=False) + fake.start_run.return_value = cm + + monkeypatch.setitem(sys.modules, "mlflow", fake) + + with tempfile.TemporaryDirectory() as outputs: + with open(os.path.join(outputs, "a.txt"), "w") as f: + f.write("hello") + df = pd.DataFrame({ + "label": ["Stratiform Precipitation"], + "llm_label": ["Stratiform Precipitation"], + "pct_gates_50dbz": [0.0], + }) + run_id = tracking.log_run_to_mlflow( + df, experiment="exp", run_name="r1", + params={"model": "test"}, + model_output_dir=outputs, + ) + + assert run_id == "rid-123" + fake.set_experiment.assert_called_once_with("exp") + fake.start_run.assert_called_once() + assert fake.log_params.called + fake.log_artifacts.assert_called_with( + outputs, artifact_path="model_outputs" + )