← Files ClaraARCHIVED FILE
modules/set-overlap-analysis/scripts/set_overlap_core.py
60.1 KB · Oct 5, 2026 · 00:02 UTC
"""Deterministic Venn and UpSet overlap analysis for chart-family runs."""
from __future__ import annotations
import argparse
import html
import json
import logging
import os
import sys
import tempfile
import warnings
import zipfile
from dataclasses import dataclass
from datetime import datetime, timezone
from itertools import combinations
from pathlib import Path
from typing import Any, Mapping, Sequence
import polars as pl
try:
from .review_session import write_review_session_artifacts, write_run_intake
except ImportError: # pragma: no cover - supports direct script imports
import importlib.util
_review_session_path = Path(__file__).resolve().parent / "review_session.py"
_review_session_spec = importlib.util.spec_from_file_location(
"mparanza_set_overlap_review_session",
_review_session_path,
)
assert _review_session_spec and _review_session_spec.loader
_review_session = importlib.util.module_from_spec(_review_session_spec)
sys.modules[_review_session_spec.name] = _review_session
_review_session_spec.loader.exec_module(_review_session)
write_review_session_artifacts = _review_session.write_review_session_artifacts
write_run_intake = _review_session.write_run_intake
__all__ = [
"InspectionResult",
"SetOverlapRunResult",
"add_common_args",
"build_overlap_tables",
"build_recipe",
"configure_logging",
"inspect_set_overlap_inputs",
"read_json",
"read_table",
"run_set_overlap",
"validate_recipe",
"write_json",
]
LOGGER = logging.getLogger(__name__)
SCHEMA_VERSION = "1.0"
PLUGIN_NAME = "set-overlap-analysis"
CANONICAL_ITEM = "item"
CANONICAL_SET = "set"
CANONICAL_PERIOD = "period"
CANONICAL_FACET = "facet"
ALL_PERIOD_LABEL = "All"
SUPPORTED_CHARTS = {"upset", "venn", "upset_small_multiples"}
ARTIFACT_MODE_DATA_ONLY = "data_only"
ARTIFACT_MODE_DATA_AND_RENDER = "data_and_render"
ARTIFACT_MODES = {
ARTIFACT_MODE_DATA_ONLY,
ARTIFACT_MODE_DATA_AND_RENDER,
}
CSV_EXTENSIONS = {".csv", ".txt", ".tsv", ".psv"}
EXCEL_EXTENSIONS = {".xlsx", ".xlsm", ".xls"}
REPO_ROOT = Path(__file__).resolve().parents[3]
SHARED_VENDOR_ROOT = REPO_ROOT / "plugins" / "_shared" / "vendor"
VENDOR_ROOT = Path(__file__).resolve().parents[1] / "vendor"
def _legacy_import_parent() -> Path:
"""Return shared modules in the repo or this component's packaged vendor."""
if (SHARED_VENDOR_ROOT / "modules" / "__init__.py").exists():
return SHARED_VENDOR_ROOT
return VENDOR_ROOT
def _activate_legacy_import_parent() -> Path:
"""Prioritize the selected vendor and evict incompatible ``modules`` imports."""
legacy_parent = _legacy_import_parent()
legacy_text = str(legacy_parent)
while legacy_text in sys.path:
sys.path.remove(legacy_text)
sys.path.insert(0, legacy_text)
module_root = (legacy_parent / "modules").resolve()
for name, module in list(sys.modules.items()):
if name == "modules" or name.startswith("modules."):
module_file = getattr(module, "__file__", None)
if not module_file or not Path(module_file).resolve().is_relative_to(
module_root
):
del sys.modules[name]
return legacy_parent
_activate_legacy_import_parent()
from modules.chart_harness import ( # noqa: E402
reporting_subject_label_from_recipe,
reporting_title_html,
)
from modules.charting.static_export import ( # noqa: E402
normalize_plotly_figure_for_static_export,
)
UPSET_TITLE_TOP_MARGIN = 96
UPSET_TITLE_BOTTOM_MARGIN = 45
SET_OVERLAP_CHART_FONT_SIZE = 12
UPSET_TITLE_FONT_SIZE = SET_OVERLAP_CHART_FONT_SIZE
UPSET_TITLE_COLOR = "#1F2328"
UPSET_TITLE_Y = 0.94
DEFAULT_SMALL_MULTIPLES_MAX_PANELS = 6
OTHER_RANK_LABEL_PREFIX = "Other rank >"
ITEM_NAME_HINTS = (
"sku",
"product",
"item",
"article",
"ean",
"upc",
"gtin",
"id",
"customer",
"client",
"store",
"name",
)
SET_NAME_HINTS = (
"set",
"group",
"company",
"manufacturer",
"retailer",
"channel",
"region",
"segment",
"brand",
"category",
"scenario",
"market",
"period",
)
PERIOD_NAME_HINTS = ("period", "scenario", "month", "week", "date", "year")
VENN_EXPORT_WIDTH = 1400
VENN_EXPORT_HEIGHT = 900
VENN_EXPORT_DPI = 100
VENN_FALLBACK_COLORS = ("#343434", "#999A9A", "#818284")
PALETTE_OPTION_KEYS = ("colorpalette", "color_palette", "palette", "chart_palette")
@dataclass(frozen=True)
class InspectionResult:
"""Inspection payload and suggested recipe."""
payload: dict[str, Any]
recipe: dict[str, Any]
output_dir: Path
@dataclass(frozen=True)
class SetOverlapRunResult:
"""Set-overlap run result."""
canonical_frame: pl.DataFrame
context: dict[str, Any]
audit: dict[str, Any]
artifact_paths: list[str]
review_session: dict[str, Any] | None = None
def configure_logging(verbose: bool = False) -> None:
"""Configure command-line logging."""
logging.basicConfig(
level=logging.DEBUG if verbose else logging.WARNING,
format="%(levelname)s:%(name)s:%(message)s",
)
def add_common_args(parser: argparse.ArgumentParser) -> None:
"""Add common plugin CLI arguments."""
parser.add_argument("input_file", type=Path)
parser.add_argument("--output-dir", type=Path, required=True)
parser.add_argument("--recipe", type=Path)
parser.add_argument("--language", default="en")
parser.add_argument(
"--artifact-mode",
choices=sorted(ARTIFACT_MODES),
default=ARTIFACT_MODE_DATA_AND_RENDER,
help=(
"Write chart data/context only or keep the legacy data-and-render behavior."
),
)
parser.add_argument("--verbose", action="store_true")
def utc_now() -> str:
"""Return an ISO timestamp for audit outputs."""
return datetime.now(timezone.utc).isoformat()
def read_json(path: Path | None) -> dict[str, Any] | None:
"""Read a JSON object from ``path`` when present."""
if path is None:
return None
payload = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(payload, dict):
raise ValueError(f"Expected JSON object: {path}")
return payload
def write_json(path: Path, payload: Mapping[str, Any]) -> None:
"""Write a deterministic JSON file."""
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
json.dumps(payload, indent=2, sort_keys=True, default=str) + "\n",
encoding="utf-8",
)
def write_csv(path: Path, frame: pl.DataFrame) -> None:
"""Write a CSV, flattening list columns for CSV compatibility."""
path.parent.mkdir(parents=True, exist_ok=True)
expressions: list[pl.Expr] = []
for name, dtype in frame.schema.items():
if str(dtype).startswith("List"):
expressions.append(pl.col(name).list.join(" | ").alias(name))
else:
expressions.append(pl.col(name))
frame.select(expressions).write_csv(path)
def _collect_csv_scan(path: Path, *, separator: str) -> pl.DataFrame:
"""Read delimited input through a lazy scan and collect once."""
lf = pl.scan_csv(path, separator=separator, infer_schema_length=10000)
try:
return lf.collect(engine="streaming")
except pl.exceptions.PolarsError:
return lf.collect()
def read_table(path: Path) -> pl.DataFrame:
"""Read CSV, TSV, PSV, TXT, XLSX, XLSM or XLS into Polars."""
suffix = path.suffix.lower()
if suffix in CSV_EXTENSIONS:
separator = ","
if suffix == ".tsv":
separator = "\t"
elif suffix == ".psv":
separator = "|"
return _collect_csv_scan(path, separator=separator)
if suffix in EXCEL_EXTENSIONS:
return pl.read_excel(path)
raise ValueError(f"Unsupported input extension: {suffix}")
def _schema(frame: pl.DataFrame) -> dict[str, pl.DataType]:
return dict(frame.schema)
def _column_names(frame: pl.DataFrame) -> list[str]:
return list(frame.schema.keys())
def _is_numeric_dtype(dtype: pl.DataType) -> bool:
checker = getattr(dtype, "is_numeric", None)
if callable(checker):
return bool(checker())
return dtype in {
pl.Int8,
pl.Int16,
pl.Int32,
pl.Int64,
pl.UInt8,
pl.UInt16,
pl.UInt32,
pl.UInt64,
pl.Float32,
pl.Float64,
}
def _safe_unique_count(frame: pl.DataFrame, column: str) -> int:
try:
value = frame.select(pl.col(column).n_unique().alias("n")).item()
except (pl.exceptions.PolarsError, TypeError, ValueError):
return 0
return int(value or 0)
def _column_profile(frame: pl.DataFrame) -> list[dict[str, Any]]:
schema = _schema(frame)
row_count = frame.height
profiles: list[dict[str, Any]] = []
for column in _column_names(frame):
unique_count = _safe_unique_count(frame, column)
profiles.append(
{
"column": column,
"dtype": str(schema[column]),
"unique_count": unique_count,
"unique_share": unique_count / row_count if row_count else 0.0,
"is_numeric": _is_numeric_dtype(schema[column]),
}
)
return profiles
def _normalized_name(column: str) -> str:
return column.lower().replace("_", " ").replace("-", " ")
def _has_hint(column: str, hints: tuple[str, ...]) -> bool:
normalized = _normalized_name(column)
return any(hint in normalized for hint in hints)
def _infer_period_column(profiles: list[dict[str, Any]]) -> str | None:
candidates = [
item
for item in profiles
if _has_hint(str(item["column"]), PERIOD_NAME_HINTS)
and int(item["unique_count"]) >= 2
]
if not candidates:
return None
candidates.sort(
key=lambda item: (
0 if _normalized_name(str(item["column"])) in {"period", "scenario"} else 1,
int(item["unique_count"]),
)
)
return str(candidates[0]["column"])
def _infer_item_column(profiles: list[dict[str, Any]]) -> str | None:
candidates: list[tuple[float, str]] = []
for item in profiles:
column = str(item["column"])
unique_count = int(item["unique_count"])
if unique_count < 2:
continue
score = float(unique_count)
if _has_hint(column, ITEM_NAME_HINTS):
score += 1000.0
if bool(item["is_numeric"]) and not _has_hint(
column, ("id", "sku", "upc", "ean", "gtin")
):
score -= 500.0
candidates.append((score, column))
if not candidates:
return None
candidates.sort(reverse=True)
return candidates[0][1]
def _infer_set_column(
profiles: list[dict[str, Any]],
*,
item_column: str | None,
period_column: str | None,
) -> str | None:
candidates: list[tuple[float, str]] = []
for item in profiles:
column = str(item["column"])
if column == item_column:
continue
unique_count = int(item["unique_count"])
if unique_count < 2:
continue
score = 0.0
if 2 <= unique_count <= 12:
score += 500.0
elif 13 <= unique_count <= 50:
score += 150.0
else:
score -= float(unique_count)
if _has_hint(column, SET_NAME_HINTS):
score += 250.0
if column == period_column:
score -= 80.0
if bool(item["is_numeric"]):
score -= 300.0
candidates.append((score, column))
if not candidates:
return None
candidates.sort(reverse=True)
return candidates[0][1]
def _unique_values(
frame: pl.DataFrame, column: str | None, *, limit: int = 200
) -> list[str]:
if not column or column not in frame.schema:
return []
values = (
frame.select(pl.col(column).cast(pl.Utf8).drop_nulls().unique().sort())
.to_series()
.to_list()
)
return [str(value) for value in values[:limit]]
def _default_selected_period(values: list[str]) -> str | None:
if not values:
return None
normalized = {value.strip().lower(): value for value in values}
for candidate in ("ac", "actual", "current"):
if candidate in normalized:
return normalized[candidate]
return values[-1]
def _bool_option(value: Any, *, default: bool) -> bool:
"""Return a strict bool for recipe options that may arrive as strings."""
if value is None:
return default
if isinstance(value, bool):
return value
text = str(value).strip().lower()
if text in {"1", "true", "yes", "y", "on"}:
return True
if text in {"0", "false", "no", "n", "off"}:
return False
return default
def _first_option(options: Mapping[str, Any], keys: tuple[str, ...]) -> Any:
for key in keys:
value = options.get(key)
if value is not None and value != "" and value != [] and value != {}:
return value
return None
def build_recipe(
input_path: Path,
frame: pl.DataFrame,
*,
language: str = "en",
existing_recipe: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Build or merge a set-overlap recipe."""
profiles = _column_profile(frame)
mappings = dict((existing_recipe or {}).get("mappings") or {})
period_column = mappings.get("period_column") or _infer_period_column(profiles)
item_column = mappings.get("item_column") or _infer_item_column(profiles)
set_column = mappings.get("set_column") or _infer_set_column(
profiles,
item_column=str(item_column) if item_column else None,
period_column=str(period_column) if period_column else None,
)
if set_column == period_column:
period_column = None
dimensions = [
item["column"]
for item in profiles
if item["column"] not in {item_column, set_column, period_column}
and not item["is_numeric"]
][:6]
period_values = _unique_values(frame, str(period_column) if period_column else None)
options = dict((existing_recipe or {}).get("options") or {})
small_multiples_dimension = _first_option(
options,
("small_multiples_dimension", "small_multiple_dimension", "facet_column"),
)
recipe = {
"schema_version": SCHEMA_VERSION,
"plugin": PLUGIN_NAME,
"source_file": str(input_path),
"language": (existing_recipe or {}).get("language") or language,
"mappings": {
"item_column": item_column,
"set_column": set_column,
"period_column": period_column,
"dimensions": mappings.get("dimensions") or dimensions,
},
"options": {
"charts": options.get("charts") or ["upset", "venn"],
"selected_period": options.get("selected_period")
or _default_selected_period(period_values),
"set_values": options.get("set_values") or [],
"max_sets": int(options.get("max_sets") or 5),
"min_intersection_size": int(options.get("min_intersection_size") or 1),
"highlighted_sets": options.get("highlighted_sets") or [],
"write_html": bool(options.get("write_html", True)),
"aggregate_other_sets": _bool_option(
options.get("aggregate_other_sets"), default=True
),
"include_other_rank_with_explicit_sets": _bool_option(
options.get("include_other_rank_with_explicit_sets"), default=False
),
"small_multiples_dimension": (
str(small_multiples_dimension) if small_multiples_dimension else None
),
"small_multiples_max_panels": int(
options.get("small_multiples_max_panels")
or options.get("max_small_multiples")
or DEFAULT_SMALL_MULTIPLES_MAX_PANELS
),
},
"inspection": {
"row_count": frame.height,
"column_count": frame.width,
"columns": _column_names(frame),
"schema": {name: str(dtype) for name, dtype in frame.schema.items()},
"column_profiles": profiles,
"period_values": period_values,
},
}
if options.get("reporting_entity_label"):
recipe["options"]["reporting_entity_label"] = str(
options["reporting_entity_label"]
)
return validate_recipe(frame, recipe)
def validate_recipe(frame: pl.DataFrame, recipe: dict[str, Any]) -> dict[str, Any]:
"""Validate recipe mappings and options against ``frame``."""
columns = set(_column_names(frame))
mappings = recipe.setdefault("mappings", {})
options = recipe.setdefault("options", {})
item_column = mappings.get("item_column")
set_column = mappings.get("set_column")
if not item_column or item_column not in columns:
raise ValueError("A valid mappings.item_column is required.")
if not set_column or set_column not in columns:
raise ValueError("A valid mappings.set_column is required.")
period_column = mappings.get("period_column")
if period_column and period_column not in columns:
mappings["period_column"] = None
charts = [
str(chart).lower().replace("-", "_") for chart in options.get("charts") or []
]
unsupported = [chart for chart in charts if chart not in SUPPORTED_CHARTS]
if unsupported:
raise ValueError("Unsupported set-overlap chart(s): " + ", ".join(unsupported))
options["charts"] = charts or ["upset", "venn"]
options["max_sets"] = max(2, int(options.get("max_sets") or 5))
options["min_intersection_size"] = max(
1, int(options.get("min_intersection_size") or 1)
)
options["small_multiples_max_panels"] = max(
1,
int(
options.get("small_multiples_max_panels")
or DEFAULT_SMALL_MULTIPLES_MAX_PANELS
),
)
options["aggregate_other_sets"] = _bool_option(
options.get("aggregate_other_sets"), default=True
)
options["include_other_rank_with_explicit_sets"] = _bool_option(
options.get("include_other_rank_with_explicit_sets"), default=False
)
options["set_values"] = [str(value) for value in options.get("set_values") or []]
options["highlighted_sets"] = [
str(value) for value in options.get("highlighted_sets") or []
]
small_multiples_dimension = _first_option(
options,
("small_multiples_dimension", "small_multiple_dimension", "facet_column"),
)
if small_multiples_dimension and str(small_multiples_dimension) in columns:
options["small_multiples_dimension"] = str(small_multiples_dimension)
else:
options["small_multiples_dimension"] = None
return recipe
def _ensure_legacy_import_path() -> None:
"""Make the shared legacy modules importable."""
_activate_legacy_import_parent()
def _cleanup_legacy_imports() -> None:
"""Remove shared/vendored ``modules`` imports loaded for this plugin."""
roots = [
(SHARED_VENDOR_ROOT / "modules").resolve(),
(VENDOR_ROOT / "modules").resolve(),
]
for name, module in list(sys.modules.items()):
if name == "modules" or name.startswith("modules."):
module_file = getattr(module, "__file__", None)
module_path = Path(module_file).resolve() if module_file else None
if module_path and any(module_path.is_relative_to(root) for root in roots):
del sys.modules[name]
for root in (str(SHARED_VENDOR_ROOT), str(VENDOR_ROOT)):
while root in sys.path:
sys.path.remove(root)
def _apply_recipe_filters(
frame: pl.DataFrame,
recipe: dict[str, Any],
) -> tuple[pl.DataFrame, dict[str, Any]]:
"""Apply shared recipe filters when available."""
_ensure_legacy_import_path()
try:
from modules.chart_harness.recipe_filters import apply_recipe_filters
return apply_recipe_filters(frame, recipe)
finally:
_cleanup_legacy_imports()
def _preserve_recipe_filters(
recipe: dict[str, Any],
source_recipe: Mapping[str, Any] | None,
) -> dict[str, Any]:
"""Preserve shared filter keys when the caller supplies a recipe."""
_ensure_legacy_import_path()
try:
from modules.chart_harness.recipe_filters import preserve_recipe_filters
return preserve_recipe_filters(recipe, source_recipe)
finally:
_cleanup_legacy_imports()
def _available_analysis_context(frame: pl.DataFrame) -> dict[str, Any]:
"""Return shared deterministic time/scenario availability metadata."""
_ensure_legacy_import_path()
try:
from modules.chart_harness import available_analysis_context
return available_analysis_context(frame)
finally:
_cleanup_legacy_imports()
def prepare_canonical_frame(
frame: pl.DataFrame, recipe: dict[str, Any]
) -> pl.DataFrame:
"""Return canonical item/set/period membership rows."""
mappings = recipe["mappings"]
options = recipe["options"]
item_column = str(mappings["item_column"])
set_column = str(mappings["set_column"])
period_column = mappings.get("period_column")
facet_column = options.get("small_multiples_dimension")
expressions = [
pl.col(item_column).cast(pl.Utf8, strict=False).alias(CANONICAL_ITEM),
pl.col(set_column).cast(pl.Utf8, strict=False).alias(CANONICAL_SET),
]
if period_column:
expressions.append(
pl.col(str(period_column))
.cast(pl.Utf8, strict=False)
.fill_null(ALL_PERIOD_LABEL)
.alias(CANONICAL_PERIOD)
)
else:
expressions.append(pl.lit(ALL_PERIOD_LABEL).alias(CANONICAL_PERIOD))
if facet_column:
expressions.append(
pl.col(str(facet_column))
.cast(pl.Utf8, strict=False)
.fill_null("Unspecified")
.alias(CANONICAL_FACET)
)
canonical = (
frame.select(expressions)
.drop_nulls(subset=[CANONICAL_ITEM, CANONICAL_SET])
.with_columns(
[
pl.col(CANONICAL_ITEM).str.strip_chars(),
pl.col(CANONICAL_SET).str.strip_chars(),
pl.col(CANONICAL_PERIOD).str.strip_chars(),
*([pl.col(CANONICAL_FACET).str.strip_chars()] if facet_column else []),
]
)
.filter((pl.col(CANONICAL_ITEM) != "") & (pl.col(CANONICAL_SET) != ""))
.unique()
)
if facet_column:
canonical = canonical.with_columns(
pl.when(pl.col(CANONICAL_FACET) == "")
.then(pl.lit("Unspecified"))
.otherwise(pl.col(CANONICAL_FACET))
.alias(CANONICAL_FACET)
)
selected_period = options.get("selected_period")
if period_column and selected_period:
canonical = canonical.filter(pl.col(CANONICAL_PERIOD) == str(selected_period))
if canonical.is_empty():
raise ValueError("No item/set membership rows remain after filters.")
return canonical
def _other_rank_label(rank_limit: int) -> str:
return f"{OTHER_RANK_LABEL_PREFIX}{rank_limit}"
def _set_summary(frame: pl.DataFrame) -> pl.DataFrame:
return (
frame.group_by(CANONICAL_SET)
.agg(pl.col(CANONICAL_ITEM).n_unique().alias("item_count"))
.sort(["item_count", CANONICAL_SET], descending=[True, False])
.with_row_index("rank", offset=1)
)
def _rank_sets_for_upset(
canonical: pl.DataFrame,
recipe: dict[str, Any],
*,
force_per_panel_ranking: bool = False,
) -> tuple[pl.DataFrame, pl.DataFrame, list[str], dict[str, Any]]:
"""Rank set values and optionally collapse lower ranks into Other."""
options = recipe["options"]
original_summary = _set_summary(canonical)
rank_limit = int(options["max_sets"])
explicit = (
[]
if force_per_panel_ranking
else [str(value) for value in options.get("set_values") or []]
)
available = set(original_summary[CANONICAL_SET].to_list())
if explicit:
missing = [value for value in explicit if value not in available]
if missing:
raise ValueError(
"Requested set_values not found after filters/period selection: "
+ ", ".join(missing)
)
top_sets = [value for value in explicit if value in available]
aggregate_other = bool(options.get("include_other_rank_with_explicit_sets"))
else:
top_sets = [
str(value)
for value in original_summary.head(rank_limit)[CANONICAL_SET].to_list()
]
aggregate_other = bool(options.get("aggregate_other_sets"))
lower_sets = [
str(value)
for value in original_summary[CANONICAL_SET].to_list()
if str(value) not in set(top_sets)
]
other_label = _other_rank_label(len(top_sets))
ranked = canonical
selected = list(top_sets)
if aggregate_other and lower_sets:
ranked = ranked.with_columns(
pl.when(pl.col(CANONICAL_SET).is_in(lower_sets))
.then(pl.lit(other_label))
.otherwise(pl.col(CANONICAL_SET))
.alias(CANONICAL_SET)
)
selected.append(other_label)
else:
ranked = ranked.filter(pl.col(CANONICAL_SET).is_in(selected))
if ranked.is_empty() or len(selected) < 2:
raise ValueError("Set-overlap charts require at least two populated sets.")
ranked_summary = _set_summary(ranked)
selected_order = {value: index for index, value in enumerate(selected, start=1)}
original_rank = {
str(row[CANONICAL_SET]): int(row["rank"]) for row in original_summary.to_dicts()
}
summary_rows: list[dict[str, Any]] = []
for row in ranked_summary.to_dicts():
set_name = str(row[CANONICAL_SET])
is_other = set_name == other_label and bool(lower_sets)
summary_rows.append(
{
"set": set_name,
"item_count": int(row["item_count"]),
"selected": set_name in selected_order,
"rank": selected_order.get(set_name, int(row["rank"])),
"original_rank": None if is_other else original_rank.get(set_name),
"is_other_rank": is_other,
"aggregated_set_count": len(lower_sets) if is_other else 0,
}
)
summary_rows.sort(key=lambda row: int(row["rank"]))
set_summary = pl.DataFrame(summary_rows)
ranking_audit = {
"rank_limit": rank_limit,
"mode": "per_panel" if force_per_panel_ranking else "regular",
"aggregate_other_sets": bool(aggregate_other),
"other_rank_label": other_label if lower_sets and aggregate_other else None,
"selected_sets": selected,
"top_sets": top_sets,
"aggregated_sets": lower_sets if aggregate_other else [],
"available_set_count": original_summary.height,
}
return ranked, set_summary, selected, ranking_audit
def _item_sets_and_intersections(
ranked_canonical: pl.DataFrame,
selected: list[str],
) -> tuple[pl.DataFrame, pl.DataFrame]:
membership = (
ranked_canonical.filter(pl.col(CANONICAL_SET).is_in(selected))
.select([CANONICAL_ITEM, CANONICAL_SET])
.unique()
)
item_sets = (
membership.group_by(CANONICAL_ITEM)
.agg(pl.col(CANONICAL_SET).unique().alias("sets"))
.with_columns(
[
pl.col("sets").list.sort(),
pl.col("sets").list.len().alias("set_count"),
]
)
.with_columns(pl.col("sets").list.join(" & ").alias("intersection"))
.sort(CANONICAL_ITEM)
)
intersections = (
item_sets.group_by(["intersection", "set_count"])
.agg(pl.len().alias("item_count"))
.sort(
["item_count", "set_count", "intersection"], descending=[True, True, False]
)
)
return item_sets, intersections
def build_overlap_tables(
canonical: pl.DataFrame,
recipe: dict[str, Any],
) -> tuple[
pl.DataFrame,
pl.DataFrame,
pl.DataFrame,
pl.DataFrame,
list[str],
dict[str, Any],
]:
"""Build deterministic set, item-membership and intersection tables."""
ranked_canonical, set_summary, selected, ranking_audit = _rank_sets_for_upset(
canonical, recipe
)
item_sets, intersections = _item_sets_and_intersections(ranked_canonical, selected)
return (
ranked_canonical,
set_summary,
item_sets,
intersections,
selected,
ranking_audit,
)
def _set_pair_table(item_sets: pl.DataFrame, selected_sets: list[str]) -> pl.DataFrame:
"""Return pairwise overlap counts for selected sets."""
rows: list[dict[str, Any]] = []
item_rows = item_sets.select([CANONICAL_ITEM, "sets"]).to_dicts()
for left, right in combinations(selected_sets, 2):
count = sum(
1
for row in item_rows
if left in set(row["sets"]) and right in set(row["sets"])
)
rows.append({"left_set": left, "right_set": right, "item_count": count})
return pl.DataFrame(rows)
def _normalize_artifact_mode(artifact_mode: str) -> str:
"""Return a supported artifact mode or raise for invalid contract input."""
normalized = str(artifact_mode or ARTIFACT_MODE_DATA_AND_RENDER).strip().lower()
if normalized not in ARTIFACT_MODES:
allowed = ", ".join(sorted(ARTIFACT_MODES))
raise ValueError(f"Unsupported artifact_mode {artifact_mode!r}; use {allowed}.")
return normalized
def _output_language(recipe: Mapping[str, Any]) -> str:
text = str(recipe.get("language") or "en").strip().lower().replace("_", "-")
return "es" if text.split("-", 1)[0] == "es" else "en"
def _localized_chart_name(chart_name: str, language: str) -> str:
if language != "es":
return chart_name
return {
"UpSet small multiples": "Múltiples gráficos UpSet",
"UpSet": "UpSet",
"Venn": "Venn",
}.get(chart_name, chart_name)
def _chart_title_lines(recipe: dict[str, Any], *, chart_name: str) -> list[str]:
mappings = recipe["mappings"]
options = recipe["options"]
language = _output_language(recipe)
period = options.get("selected_period") or (
"Todos los periodos" if language == "es" else ALL_PERIOD_LABEL
)
first_line = reporting_subject_label_from_recipe(recipe)
relationship = "solapamiento por" if language == "es" else "overlap by"
return [
line
for line in (
first_line,
f"{_localized_chart_name(chart_name, language)}: "
f"{mappings['item_column']} {relationship} {mappings['set_column']}",
str(period),
)
if line
]
def _chart_title(recipe: dict[str, Any], *, chart_name: str, html: bool = False) -> str:
lines = _chart_title_lines(recipe, chart_name=chart_name)
if html and len(lines) >= 3:
return reporting_title_html(lines[0], lines[1], lines[2])
return "\n".join(lines)
def _small_multiple_panel_title(
recipe: dict[str, Any],
*,
facet_dimension: str,
facet_value: str,
html_title: bool = False,
) -> str:
mappings = recipe["mappings"]
options = recipe["options"]
language = _output_language(recipe)
period = options.get("selected_period") or (
"Todos los periodos" if language == "es" else ALL_PERIOD_LABEL
)
relationship = "solapamiento por" if language == "es" else "overlap by"
lines = [
f"{facet_dimension}: {facet_value}",
f"UpSet: {mappings['item_column']} {relationship} {mappings['set_column']}",
str(period),
]
if html_title:
return reporting_title_html(lines[0], lines[1], lines[2])
return "\n".join(lines)
def _apply_upset_reporting_title(fig: Any, title_html: str) -> None:
"""Apply the compact three-row reporting title used by UpSet exports."""
existing_margin = fig.layout.margin.to_plotly_json() if fig.layout.margin else {}
margin = {
"l": int(existing_margin.get("l") or 35),
"r": int(existing_margin.get("r") or 25),
"t": max(int(existing_margin.get("t") or 0), UPSET_TITLE_TOP_MARGIN),
"b": max(int(existing_margin.get("b") or 0), UPSET_TITLE_BOTTOM_MARGIN),
}
fig.update_layout(
title={
"text": title_html,
"x": 0.01,
"xanchor": "left",
"y": UPSET_TITLE_Y,
"yanchor": "top",
"font": {"size": UPSET_TITLE_FONT_SIZE, "color": UPSET_TITLE_COLOR},
},
margin=margin,
)
def _facet_values_for_small_multiples(
canonical: pl.DataFrame,
recipe: dict[str, Any],
) -> list[str]:
if CANONICAL_FACET not in canonical.schema:
return []
max_panels = int(recipe["options"].get("small_multiples_max_panels") or 1)
rows = (
canonical.group_by(CANONICAL_FACET)
.agg(pl.col(CANONICAL_ITEM).n_unique().alias("item_count"))
.sort(["item_count", CANONICAL_FACET], descending=[True, False])
.head(max_panels)
.to_dicts()
)
return [str(row[CANONICAL_FACET]) for row in rows]
def _requested_palette_name(
recipe: Mapping[str, Any],
color_dict: Mapping[str, Any],
default_name: str,
) -> str:
"""Return the requested chart palette, falling back to the default palette."""
def is_palette_value(value: Any) -> bool:
return isinstance(value, (list, tuple)) and any(
isinstance(color, str) for color in value
)
options = recipe.get("options")
requested: str | None = None
if isinstance(options, Mapping):
for key in PALETTE_OPTION_KEYS:
value = options.get(key)
if isinstance(value, str) and value.strip():
requested = value.strip()
break
if requested:
if requested in color_dict and is_palette_value(color_dict[requested]):
return requested
requested_lower = requested.lower()
for palette_name, palette in color_dict.items():
if str(palette_name).lower() == requested_lower and is_palette_value(
palette
):
return str(palette_name)
if default_name in color_dict and is_palette_value(color_dict[default_name]):
return default_name
for palette_name, palette in color_dict.items():
if is_palette_value(palette):
return str(palette_name)
return default_name
def _venn_colors_from_palette(
palette: Any,
selected_set_count: int,
) -> tuple[str, ...]:
"""Return enough Venn colors from a legacy chart palette."""
colors = [str(color) for color in palette if isinstance(color, str) and color]
fallback_index = 0
while len(colors) < selected_set_count:
colors.append(VENN_FALLBACK_COLORS[fallback_index % len(VENN_FALLBACK_COLORS)])
fallback_index += 1
return tuple(colors[:selected_set_count])
def _resolve_chart_palette(
recipe: Mapping[str, Any],
selected_set_count: int,
) -> tuple[str, tuple[str, ...]]:
"""Resolve the plugin chart palette through the shared legacy palette map."""
_ensure_legacy_import_path()
try:
from modules.charting.chart_primitives import get_color_dictionary
from modules.utilities.config import get_naming_params
names = get_naming_params()
color_dict = get_color_dictionary({})
palette_name = _requested_palette_name(
recipe,
color_dict,
names["bainColorpalette"],
)
return (
palette_name,
_venn_colors_from_palette(color_dict[palette_name], selected_set_count),
)
except (ImportError, ModuleNotFoundError, KeyError, TypeError, ValueError) as exc:
LOGGER.debug("Falling back to default Venn colors: %s", exc)
return (
"fallback",
_venn_colors_from_palette(VENN_FALLBACK_COLORS, selected_set_count),
)
finally:
_cleanup_legacy_imports()
def _make_upset_figure(
canonical: pl.DataFrame,
recipe: dict[str, Any],
selected_sets: list[str],
) -> tuple[Any, str, dict[str, Any]]:
"""Return one UpSet Plotly figure from already-ranked canonical rows."""
_ensure_legacy_import_path()
try:
from modules.charting.chart_primitives import get_color_dictionary
from modules.charting.upset_helpers import build_upset_matrix
from modules.charting.upset_plot import plot_upset
from modules.utilities.config import get_naming_params
names = get_naming_params()
color_dict = get_color_dictionary({})
palette_name = _requested_palette_name(
recipe,
color_dict,
names["bainColorpalette"],
)
mapping = (
canonical.filter(pl.col(CANONICAL_SET).is_in(selected_sets))
.select(
[
pl.col(CANONICAL_ITEM).alias("Name"),
pl.col(CANONICAL_SET).alias("set"),
]
)
.lazy()
)
matrix = build_upset_matrix(mapping, selected_sets)
chart_dict = {
names["minIntersectionSize"]: int(
recipe["options"].get("min_intersection_size") or 1
),
names["highlightedDimension"]: recipe["options"].get("highlighted_sets")
or [],
names["colorChoice"]: names["redToGreen"],
names["colorpalette"]: palette_name,
}
return (
plot_upset(matrix, chart_dict),
palette_name,
{
"chart_dict": chart_dict,
"source_functions": [
"modules.charting.upset_helpers.build_upset_matrix",
"modules.charting.upset_plot.plot_upset",
],
},
)
finally:
_cleanup_legacy_imports()
def _write_upset_chart(
canonical: pl.DataFrame,
recipe: dict[str, Any],
output_dir: Path,
selected_sets: list[str],
*,
render: bool = True,
) -> tuple[list[str], dict[str, Any]]:
"""Write UpSet PNG/HTML artifacts through the legacy matrix/Plotly helper."""
if not render:
return [], {
"chart": "upset",
"status": "data_written",
"title": _chart_title(recipe, chart_name="UpSet"),
"artifacts": [],
"selected_sets": selected_sets,
"renderer": "not_rendered",
"source_functions": [
"modules.charting.upset_helpers.build_upset_matrix",
"modules.charting.upset_plot.plot_upset",
],
}
_ensure_legacy_import_path()
paths: list[str] = []
try:
fig, palette_name, figure_audit = _make_upset_figure(
canonical, recipe, selected_sets
)
title = _chart_title(recipe, chart_name="UpSet")
title_html = _chart_title(recipe, chart_name="UpSet", html=True)
_apply_upset_reporting_title(fig, title_html)
fig.update_layout(
paper_bgcolor="white",
plot_bgcolor="white",
font={"family": "Arial", "size": 12, "color": "#1F2328"},
)
export_fig, normalization_audit = normalize_plotly_figure_for_static_export(fig)
html_path = output_dir / "upset.html"
if bool(recipe["options"].get("write_html", True)):
export_fig.write_html(str(html_path), include_plotlyjs="cdn")
paths.append(str(html_path))
png_path = output_dir / "upset.png"
try:
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
export_width = int(export_fig.layout.width or 1000)
export_height = int(export_fig.layout.height or 650)
export_fig.write_image(
str(png_path),
format="png",
width=export_width,
height=export_height,
)
paths.append(str(png_path))
status = "written"
renderer = "plotly+kaleido"
error = None
except (OSError, RuntimeError, ValueError) as exc:
status = "written_html_only" if paths else "not_written"
renderer = "plotly_html"
error = str(exc)
return paths, {
"chart": "upset",
"status": status,
"title": title,
"artifacts": [Path(path).name for path in paths],
"selected_sets": selected_sets,
"palette": palette_name,
"chart_font_size": SET_OVERLAP_CHART_FONT_SIZE,
"renderer": renderer,
"error": error,
"figure_export_normalization": normalization_audit,
"source_functions": figure_audit["source_functions"],
}
finally:
_cleanup_legacy_imports()
def _write_upset_small_multiples_chart(
canonical: pl.DataFrame,
recipe: dict[str, Any],
output_dir: Path,
*,
render: bool = True,
) -> tuple[list[str], dict[str, Any]]:
"""Write a stacked HTML small-multiple UpSet artifact."""
facet_dimension = recipe["options"].get("small_multiples_dimension")
if not facet_dimension or CANONICAL_FACET not in canonical.schema:
return [], {
"chart": "upset_small_multiples",
"status": "not_written_missing_facet",
"reason": "small_multiples_dimension is not set or is unavailable.",
}
facets = _facet_values_for_small_multiples(canonical, recipe)
if not facets:
return [], {
"chart": "upset_small_multiples",
"status": "not_written_no_facets",
"small_multiples_dimension": facet_dimension,
}
panel_html: list[str] = []
panel_audits: list[dict[str, Any]] = []
set_summary_rows: list[dict[str, Any]] = []
intersection_rows: list[dict[str, Any]] = []
paths: list[str] = []
for facet_value in facets:
panel_canonical = canonical.filter(pl.col(CANONICAL_FACET) == facet_value)
try:
ranked, set_summary, selected_sets, ranking = _rank_sets_for_upset(
panel_canonical,
recipe,
force_per_panel_ranking=True,
)
item_sets, intersections = _item_sets_and_intersections(
ranked,
selected_sets,
)
except ValueError as exc:
panel_audits.append(
{
"facet": facet_value,
"status": "not_written",
"error": str(exc),
}
)
continue
for row in set_summary.to_dicts():
set_summary_rows.append({"facet": facet_value, **row})
for row in intersections.to_dicts():
intersection_rows.append({"facet": facet_value, **row})
if not render:
panel_audits.append(
{
"facet": facet_value,
"status": "data_written",
"selected_sets": selected_sets,
"set_ranking": ranking,
"item_count": item_sets.height,
"intersection_count": intersections.height,
"source_functions": [
"modules.charting.upset_helpers.build_upset_matrix",
"modules.charting.upset_plot.plot_upset",
],
}
)
continue
fig, palette_name, figure_audit = _make_upset_figure(
ranked,
recipe,
selected_sets,
)
panel_title = _small_multiple_panel_title(
recipe,
facet_dimension=str(facet_dimension),
facet_value=facet_value,
html_title=True,
)
_apply_upset_reporting_title(fig, panel_title)
fig.update_layout(
paper_bgcolor="white",
plot_bgcolor="white",
font={"family": "Arial", "size": 12, "color": "#1F2328"},
)
panel_html.append(
'<section class="upset-panel">'
+ fig.to_html(
full_html=False,
include_plotlyjs="cdn" if not panel_html else False,
config={"responsive": True},
)
+ "</section>"
)
panel_audits.append(
{
"facet": facet_value,
"status": "written",
"selected_sets": selected_sets,
"set_ranking": ranking,
"item_count": item_sets.height,
"intersection_count": intersections.height,
"palette": palette_name,
"source_functions": figure_audit["source_functions"],
}
)
if set_summary_rows:
summary_path = output_dir / "set_overlap_small_multiples_set_summary.csv"
write_csv(summary_path, pl.DataFrame(set_summary_rows))
paths.append(str(summary_path))
if intersection_rows:
intersections_path = (
output_dir / "set_overlap_small_multiples_intersections.csv"
)
write_csv(intersections_path, pl.DataFrame(intersection_rows))
paths.append(str(intersections_path))
if not render:
return paths, {
"chart": "upset_small_multiples",
"status": "data_written" if panel_audits else "not_written_no_valid_panels",
"title": _chart_title(recipe, chart_name="UpSet small multiples"),
"artifacts": [Path(path).name for path in paths],
"small_multiples_dimension": facet_dimension,
"panel_count": len(panel_audits),
"requested_panel_count": len(facets),
"per_panel_ranking": True,
"facets": panel_audits,
"renderer": "not_rendered",
"source_functions": [
"modules.charting.upset_helpers.build_upset_matrix",
"modules.charting.upset_plot.plot_upset",
],
}
if not panel_html:
return paths, {
"chart": "upset_small_multiples",
"status": "not_written_no_valid_panels",
"small_multiples_dimension": facet_dimension,
"facets": panel_audits,
}
page_title = _chart_title(recipe, chart_name="UpSet small multiples")
html_language = _output_language(recipe)
html_path = output_dir / "upset_small_multiples.html"
html_path.write_text(
"\n".join(
[
"<!doctype html>",
f'<html lang="{html_language}">',
"<head>",
'<meta charset="utf-8">',
f"<title>{html.escape(page_title)}</title>",
"<style>",
"body{font-family:Arial,sans-serif;font-size:12px;color:#1F2328;background:#fff;margin:0;padding:16px}",
".upset-small-multiples{display:inline-block;padding:8px 18px 18px 8px}",
"h1{font-size:12px;line-height:1.25;margin:0 0 16px;font-weight:700;white-space:pre-line}",
".upset-panel{break-inside:avoid;margin:0 0 24px}",
"</style>",
"</head>",
"<body>",
'<main class="upset-small-multiples" data-gallery-screenshot>',
f"<h1>{html.escape(page_title)}</h1>",
*panel_html,
"</main>",
"</body>",
"</html>",
]
)
+ "\n",
encoding="utf-8",
)
paths.append(str(html_path))
return paths, {
"chart": "upset_small_multiples",
"status": "written",
"title": page_title,
"artifacts": [Path(path).name for path in paths],
"small_multiples_dimension": facet_dimension,
"panel_count": len(panel_html),
"requested_panel_count": len(facets),
"per_panel_ranking": True,
"chart_font_size": SET_OVERLAP_CHART_FONT_SIZE,
"facets": panel_audits,
"renderer": "plotly_html",
}
def _write_venn_chart(
canonical: pl.DataFrame,
recipe: dict[str, Any],
output_dir: Path,
selected_sets: list[str],
*,
render: bool = True,
) -> tuple[list[str], dict[str, Any]]:
"""Write a Venn PNG for two or three selected sets."""
if len(selected_sets) not in {2, 3}:
return [], {
"chart": "venn",
"status": "not_written_unsupported_set_count",
"reason": "Venn charts are only readable for two or three sets.",
"selected_set_count": len(selected_sets),
"selected_sets": selected_sets,
}
if not render:
return [], {
"chart": "venn",
"status": "data_written",
"title": _chart_title(recipe, chart_name="Venn"),
"artifacts": [],
"selected_sets": selected_sets,
"renderer": "not_rendered",
"source_functions": [
"modules.data.misc_charts_data_prep.prepare_data_for_venn_plot",
"modules.charting.draw_venn_upset.draw_venn_chart",
],
}
try:
cache_dir = Path(tempfile.gettempdir()) / "mparanza-matplotlib-cache"
cache_dir.mkdir(parents=True, exist_ok=True)
os.environ.setdefault("MPLCONFIGDIR", str(cache_dir))
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
from matplotlib_venn import venn2, venn3
except (ImportError, ModuleNotFoundError) as exc:
return [], {
"chart": "venn",
"status": "not_written_missing_dependency",
"error": str(exc),
}
membership = (
canonical.filter(pl.col(CANONICAL_SET).is_in(selected_sets))
.select([CANONICAL_ITEM, CANONICAL_SET])
.unique()
)
grouped = membership.group_by(CANONICAL_SET).agg(
pl.col(CANONICAL_ITEM).unique().alias("items")
)
set_payload = {
str(row[CANONICAL_SET]): set(str(item) for item in row["items"])
for row in grouped.to_dicts()
}
data = [set_payload.get(name, set()) for name in selected_sets]
palette_name, set_colors = _resolve_chart_palette(recipe, len(selected_sets))
fig, ax = plt.subplots(
figsize=(
VENN_EXPORT_WIDTH / VENN_EXPORT_DPI,
VENN_EXPORT_HEIGHT / VENN_EXPORT_DPI,
),
facecolor="white",
)
fig.subplots_adjust(left=0.06, right=0.97, top=0.76, bottom=0.08)
if len(selected_sets) == 2:
venn = venn2(
data,
set_labels=tuple(selected_sets),
set_colors=set_colors,
alpha=0.55,
ax=ax,
)
else:
venn = venn3(
data,
set_labels=tuple(selected_sets),
set_colors=set_colors,
alpha=0.55,
ax=ax,
)
if venn is not None:
for text in [*venn.set_labels, *venn.subset_labels]:
if text is not None:
text.set_fontfamily("Arial")
text.set_fontsize(SET_OVERLAP_CHART_FONT_SIZE)
text.set_color("#1F2328")
title = _chart_title(recipe, chart_name="Venn")
fig.text(
0.04,
0.96,
title,
ha="left",
va="top",
fontfamily="Arial",
fontsize=SET_OVERLAP_CHART_FONT_SIZE,
color="#1F2328",
)
png_path = output_dir / "venn.png"
fig.savefig(png_path, dpi=VENN_EXPORT_DPI, facecolor="white")
plt.close(fig)
return [str(png_path)], {
"chart": "venn",
"status": "written",
"title": title,
"artifacts": [png_path.name],
"selected_sets": selected_sets,
"palette": palette_name,
"colors": list(set_colors),
"chart_font_size": SET_OVERLAP_CHART_FONT_SIZE,
"dimensions": {"width": VENN_EXPORT_WIDTH, "height": VENN_EXPORT_HEIGHT},
"renderer": "matplotlib_venn",
"source_functions": [
"modules.data.misc_charts_data_prep.prepare_data_for_venn_plot",
"modules.charting.draw_venn_upset.draw_venn_chart",
],
}
def inspect_set_overlap_inputs(
input_path: Path,
output_dir: Path,
recipe_path: Path | None = None,
*,
language: str = "en",
) -> InspectionResult:
"""Inspect inputs and write suggested recipe files."""
frame = read_table(input_path)
existing_recipe = read_json(recipe_path)
recipe = build_recipe(
input_path,
frame,
language=language,
existing_recipe=existing_recipe,
)
recipe = _preserve_recipe_filters(recipe, existing_recipe)
payload = {
"schema_version": SCHEMA_VERSION,
"plugin": PLUGIN_NAME,
"input_file": str(input_path),
"row_count": frame.height,
"column_count": frame.width,
"columns": _column_names(frame),
"schema": {name: str(dtype) for name, dtype in frame.schema.items()},
"available_analysis_context": _available_analysis_context(frame),
"suggested_mappings": recipe["mappings"],
"suggested_options": recipe["options"],
"column_profiles": recipe["inspection"]["column_profiles"],
}
output_dir.mkdir(parents=True, exist_ok=True)
write_json(output_dir / "inspection.json", payload)
write_json(output_dir / "suggested_recipe.json", recipe)
return InspectionResult(payload=payload, recipe=recipe, output_dir=output_dir)
def run_set_overlap(
input_path: Path,
output_dir: Path,
recipe_path: Path | None = None,
*,
language: str = "en",
artifact_mode: str = ARTIFACT_MODE_DATA_AND_RENDER,
) -> SetOverlapRunResult:
"""Run deterministic Venn/UpSet overlap analysis."""
artifact_mode = _normalize_artifact_mode(artifact_mode)
frame = read_table(input_path)
existing_recipe = read_json(recipe_path)
recipe = build_recipe(
input_path,
frame,
language=language,
existing_recipe=existing_recipe,
)
recipe = _preserve_recipe_filters(recipe, existing_recipe)
recipe = validate_recipe(frame, recipe)
filtered, filter_audit = _apply_recipe_filters(frame, recipe)
recipe.setdefault("options", {})["recipe_filter_audit"] = filter_audit
canonical = prepare_canonical_frame(filtered, recipe)
(
ranked_canonical,
set_summary,
item_sets,
intersections,
selected_sets,
ranking_audit,
) = build_overlap_tables(canonical, recipe)
pair_table = _set_pair_table(item_sets, selected_sets)
output_dir.mkdir(parents=True, exist_ok=True)
run_intake = write_run_intake(
output_dir,
input_path,
recipe_path=recipe_path,
recipe=recipe,
source_row_count=frame.height,
)
canonical_path = output_dir / "set_overlap_canonical.csv"
ranked_canonical_path = output_dir / "set_overlap_ranked_canonical.csv"
set_summary_path = output_dir / "set_overlap_set_summary.csv"
item_sets_path = output_dir / "set_overlap_item_sets.csv"
intersections_path = output_dir / "set_overlap_intersections.csv"
pair_path = output_dir / "set_overlap_pairs.csv"
write_csv(canonical_path, canonical)
write_csv(ranked_canonical_path, ranked_canonical)
write_csv(set_summary_path, set_summary)
write_csv(item_sets_path, item_sets)
write_csv(intersections_path, intersections)
write_csv(pair_path, pair_table)
artifact_paths = [
str(canonical_path),
str(ranked_canonical_path),
str(set_summary_path),
str(item_sets_path),
str(intersections_path),
str(pair_path),
]
chart_audits: dict[str, Any] = {}
chart_names = [
str(chart)
for chart in recipe["options"].get("charts") or []
if str(chart) in SUPPORTED_CHARTS
]
charts = set(chart_names)
render_charts = artifact_mode != ARTIFACT_MODE_DATA_ONLY
if "upset" in charts:
paths, chart_audit = _write_upset_chart(
ranked_canonical, recipe, output_dir, selected_sets, render=render_charts
)
artifact_paths.extend(paths)
chart_audits["upset"] = chart_audit
if "upset_small_multiples" in charts:
paths, chart_audit = _write_upset_small_multiples_chart(
canonical,
recipe,
output_dir,
render=render_charts,
)
artifact_paths.extend(paths)
chart_audits["upset_small_multiples"] = chart_audit
if "venn" in charts:
paths, chart_audit = _write_venn_chart(
ranked_canonical, recipe, output_dir, selected_sets, render=render_charts
)
artifact_paths.extend(paths)
chart_audits["venn"] = chart_audit
context = {
"schema_version": SCHEMA_VERSION,
"plugin": PLUGIN_NAME,
"analysis_type": "set_overlap",
"source_file": str(input_path),
"language": recipe.get("language") or language,
"artifact_mode": artifact_mode,
"mappings": recipe["mappings"],
"options": recipe["options"],
"recipe_filters": filter_audit,
"selected_sets": selected_sets,
"row_counts": {
"source": frame.height,
"filtered": filtered.height,
"canonical_memberships": canonical.height,
"ranked_canonical_memberships": ranked_canonical.height,
"item_count": item_sets.height,
"intersection_count": intersections.height,
},
"set_summary": set_summary.to_dicts(),
"set_ranking": ranking_audit,
"intersections": intersections.to_dicts(),
"pairwise_overlap": pair_table.to_dicts(),
"chart_audits": chart_audits,
"codex_interpretation_contract": {
"must_review_when_written": True,
"required_points": [
"State the selected item column, set column, period and filters.",
"Identify the largest exact intersections from set_overlap_intersections.csv.",
"Use UpSet for more than three sets; use Venn only as a simple communication aid for two or three sets.",
],
},
}
context_path = output_dir / "set_overlap_context.json"
used_recipe_path = output_dir / "used_recipe.json"
write_json(context_path, context)
write_json(used_recipe_path, recipe)
artifact_paths.extend([str(context_path), str(used_recipe_path)])
audit = {
"schema_version": SCHEMA_VERSION,
"plugin": PLUGIN_NAME,
"created_at": utc_now(),
"input_file": str(input_path),
"recipe": recipe,
"artifact_mode": artifact_mode,
"checks": {
"source_row_count": frame.height,
"filtered_row_count": filtered.height,
"canonical_row_count": canonical.height,
"selected_set_count": len(selected_sets),
"legacy_chart_attempt_count": len(chart_audits),
"legacy_chart_written_count": sum(
1
for item in chart_audits.values()
if item.get("status") in {"written", "written_html_only"}
),
"legacy_chart_data_count": sum(
1
for item in chart_audits.values()
if item.get("status")
in {"written", "written_html_only", "data_written"}
),
},
"charts": chart_audits,
"outputs": {
Path(path).relative_to(output_dir).as_posix(): "written"
for path in artifact_paths
if Path(path).exists() and Path(path).is_file()
},
}
audit_path = output_dir / "set_overlap_audit.json"
write_json(audit_path, audit)
artifact_paths.append(str(audit_path))
zip_path = output_dir / "set_overlap_artifacts.zip"
with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as archive:
for artifact in artifact_paths:
path = Path(artifact)
if path.exists() and path.is_file():
archive.write(path, path.relative_to(output_dir))
artifact_paths.append(str(zip_path))
review_session = write_review_session_artifacts(
output_dir,
input_path,
run_id=run_intake.run_id,
run_intake_path=run_intake.path,
recipe_path=recipe_path,
recipe=recipe,
context=context,
audit=audit,
)
audit["review_session"] = {
"run_id": review_session.run_id,
"run_intake_path": str(review_session.run_intake_path),
"review_payload_path": str(review_session.review_payload_path),
"ui_decisions_path": str(review_session.ui_decisions_path),
"final_artifacts_path": str(review_session.final_artifacts_path),
"review_item_count": review_session.review_item_count,
}
write_json(audit_path, audit)
return SetOverlapRunResult(
canonical_frame=canonical,
context=context,
audit=audit,
artifact_paths=artifact_paths,
review_session=audit["review_session"],
)
SHA-256: f759bb0d3394a824bf47c7b5815333c3ee284bad9b85a1875198c0abebfd1929