"""Continuous score feedback adapted from the supplied temp.py ScoreOptimizer. This searches training quantiles using an already-trained predictor; it does not train a second model or identify causal feature harm. Text/graphic area, text-count and text-height groups are actionable. Geometry stays fixed; related measurements are updated as coherent groups. Count/height edits assume text area scales proportionally; this is a model probe, not an image edit or a guarantee of realizable design improvement. """ import hashlib import json from pathlib import Path import numpy as np import pandas as pd AREA_FEATURES = ("text_area_frac", "graphic_area_frac") GROUPS = {"word_count": ("word_count", "char_count", "text_block_count"), "mean_text_height_frac": ("mean_text_height_frac", "max_text_height_frac", "text_height_std_frac")} DERIVED = ("background_area_frac", "text_share", "text_to_graphic_ratio") RATIO_EPS = 1e-3 # Same definition as features.extract_features. class ScoreOptimizer: """Attach to a fitted predictor and learn candidate values from TRAIN only.""" def __init__(self, predictor, train_features, feature_names=None, training_ids=None): self.predictor = predictor self.feature_names = list(feature_names or train_features.columns) train = self._validate(train_features) if train.empty: raise ValueError("Training features cannot be empty") self.actionable = list(AREA_FEATURES) + [name for name, group in GROUPS.items() if set(group).issubset(self.feature_names)] self.quantiles = { name: np.unique(train[name].quantile(np.linspace(0.05, 0.95, 19))).tolist() for name in self.actionable } self.recipe = { "version": 1, "method": "training_quantile_group_counterfactuals", "fit_split": "train", "training_rows": len(train), "feature_columns": self.feature_names, "actionable_features": self.actionable, "quantiles": self.quantiles, "coupled_groups": {name: list(group) + ["text_area_frac"] + list(DERIVED) for name, group in GROUPS.items() if name in self.actionable}, "immutable_features": [c for c in self.feature_names if c not in AREA_FEATURES + DERIVED + tuple(x for group in GROUPS.values() for x in group)], "interpretation": "Model-estimated layout changes, not causal attribution or feedback labels.", } if training_ids is not None: ids = sorted(str(value) for value in training_ids) if len(ids) != len(train) or len(set(ids)) != len(ids): raise ValueError("Training IDs must uniquely match training rows") self.recipe["training_ids_sha256"] = hashlib.sha256("\n".join(ids).encode()).hexdigest() def save_recipe(self, path): Path(path).write_text(json.dumps(self.recipe, indent=2, allow_nan=False) + "\n") @classmethod def load_recipe(cls, predictor, path): """Reuse fitted candidate metadata without reading any dataset splits.""" recipe = json.loads(Path(path).read_text()) if recipe.get("fit_split") != "train" or recipe.get("version") != 1: raise ValueError("Expected a version 1 recipe fitted on train") obj = cls.__new__(cls) obj.predictor = predictor obj.recipe = recipe obj.feature_names = recipe["feature_columns"] obj.actionable = recipe["actionable_features"] obj.quantiles = recipe["quantiles"] if not set(obj.actionable).issubset(set(AREA_FEATURES) | set(GROUPS)): raise ValueError("Unknown actionable feature in recipe") for name in obj.actionable: values = np.asarray(obj.quantiles[name], dtype=float) if not len(values) or not np.isfinite(values).all() or (values < 0).any(): raise ValueError("Invalid candidate values in feedback recipe") return obj def _validate(self, frame): missing = set(self.feature_names) - set(frame.columns) if missing: raise ValueError(f"Missing required features: {sorted(missing)}") required_areas = set(AREA_FEATURES + DERIVED) if not required_areas.issubset(self.feature_names): raise ValueError("Feedback requires the five shared area features") values = frame[self.feature_names].apply(pd.to_numeric, errors="raise").astype(float) if not np.isfinite(values.to_numpy()).all(): raise ValueError("Feedback requires finite, non-missing feature values") for column in values: if (values[column] < 0).any(): raise ValueError(f"{column} cannot be negative") if (column.endswith("_frac") or column == "text_share") and (values[column] > 1).any(): raise ValueError(f"{column} must be between 0 and 1") text, graphic = values[AREA_FEATURES[0]], values[AREA_FEATURES[1]] if (text + graphic > 1 + 1e-8).any(): raise ValueError("Text and graphic areas cannot exceed the image area") expected = { "background_area_frac": np.maximum(0, 1 - text - graphic), "text_share": np.divide(text, text + graphic, out=np.zeros(len(text)), where=(text + graphic) > 0), "text_to_graphic_ratio": text / np.maximum(graphic, RATIO_EPS), } for column, expected_values in expected.items(): if not np.allclose(values[column], expected_values, rtol=1e-5, atol=1e-7): raise ValueError(f"Inconsistent derived feature: {column}") return values.reset_index(drop=True) def candidate_rows(self, features): """Return feasible rows and (feature, old value, new value) descriptions.""" base = self._validate(pd.DataFrame([features])).iloc[0] rows, descriptions = [], [] for feature in self.actionable: if feature != "graphic_area_frac" and (base["text_area_frac"] == 0 or base.get("word_count", 1) == 0): continue for value in self.quantiles[feature]: if feature == "word_count": value = float(round(value)) if np.isclose(value, base[feature]): continue row = base.copy() row[feature] = value if feature in GROUPS: if value <= 0 or base[feature] <= 0: continue factor = value / base[feature] if feature == "word_count": row["char_count"] = max(value, round(base["char_count"] * factor)) row["text_block_count"] = min(value, max(1, round(base["text_block_count"] * factor))) else: row["max_text_height_frac"] = base["max_text_height_frac"] * factor row["text_height_std_frac"] = base["text_height_std_frac"] * factor if row["max_text_height_frac"] > 1 or row["text_height_std_frac"] > row["max_text_height_frac"] / 2 + 1e-8: continue row["text_area_frac"] = base["text_area_frac"] * factor text, graphic = row["text_area_frac"], row["graphic_area_frac"] if text < 0 or graphic < 0 or text + graphic > 1 or (text == 0 and base.get("word_count", 0) > 0): continue row["background_area_frac"] = max(0.0, 1 - text - graphic) row["text_share"] = text / (text + graphic) if text + graphic > 0 else 0.0 row["text_to_graphic_ratio"] = text / max(graphic, RATIO_EPS) changes = {name: float(row[name]) for name in self.feature_names if not np.isclose(row[name], base[name])} rows.append(row) descriptions.append((feature, float(base[feature]), float(value), changes)) return pd.DataFrame(rows, columns=self.feature_names), descriptions def analyze_many(self, features, top_k=3, min_improvement=1e-6): """Score every input and all feasible candidates in one prediction batch.""" if top_k < 1 or min_improvement < 0: raise ValueError("top_k must be positive and min_improvement nonnegative") base = self._validate(features) if base.empty: return [] parts, metadata = [base], [] for index, row in base.iterrows(): candidates, descriptions = self.candidate_rows(row.to_dict()) parts.append(candidates) metadata.extend((index, *description) for description in descriptions) scores = np.asarray(self.predictor.predict(pd.concat(parts, ignore_index=True)), dtype=float) if len(scores) != sum(len(part) for part in parts) or not np.isfinite(scores).all(): raise ValueError("Predictor returned invalid scores") results = [{"score": float(score), "ranked_changes": [], "best_recommendation": None, "best_feature": None, "best_gain": None, "best_direction": None, "supported_features": self.actionable, "feedback_sentence": "The model found no tested layout change likely to improve this poster’s score.", "message": "No tested feature-group change improved the model's predicted score."} for score in scores[:len(base)]] best_by_row = [{} for _ in range(len(base))] for (index, feature, old, new, changes), score in zip(metadata, scores[len(base):]): change = {"feature": feature, "current_value": old, "suggested_value": new, "direction": "increase" if new > old else "decrease", "coupled_feature_updates": changes, "predicted_score": float(score), "gain": float(score - results[index]["score"])} previous = best_by_row[index].get(feature) if previous is None or change["gain"] > previous["gain"]: best_by_row[index][feature] = change for result, best in zip(results, best_by_row): ranked = sorted(best.values(), key=lambda item: item["gain"], reverse=True)[:top_k] result["ranked_changes"] = ranked # Signed gains include zero/negative outcomes. if ranked and ranked[0]["gain"] > min_improvement: recommendation = ranked[0] result.update(best_recommendation=recommendation, best_feature=recommendation["feature"], best_gain=recommendation["gain"], best_direction=recommendation["direction"], message="Largest estimated gain among tested feature-group changes; not a causal diagnosis.") labels = {"text_area_frac": "text coverage", "graphic_area_frac": "graphic coverage", "word_count": "the amount of text", "mean_text_height_frac": "text size"} action = "increasing" if recommendation["direction"] == "increase" else "decreasing" result["feedback_sentence"] = ( f"The model suggests {action} {labels[recommendation['feature']]} " "as the most promising change for this poster." ) return results def analyze(self, features, **kwargs): return self.analyze_many(pd.DataFrame([features]), **kwargs)[0]