"""Visualization tools for data version comparisons.""" from pathlib import Path from typing import Any, Dict, List import matplotlib.pyplot as plt import numpy as np def plot_id_changes(comparison: Dict[str, Any], output_path: Path) -> None: """Create bar chart showing ID changes between versions. Args: comparison: Comparison results dictionary output_path: Path to save the plot """ summary = comparison["summary"] id_changes = comparison["id_changes"] fig, ax = plt.subplots(figsize=(10, 6)) categories = ["Added", "Removed", "Common"] counts = [ id_changes["added_count"], id_changes["removed_count"], id_changes["common_count"], ] colors = ["#2ecc71", "#e74c3c", "#3498db"] bars = ax.bar(categories, counts, color=colors, alpha=0.7, edgecolor="black") # Add value labels on bars for bar in bars: height = bar.get_height() ax.text( bar.get_x() + bar.get_width() / 2.0, height, f"{int(height):,}", ha="center", va="bottom", fontsize=10, fontweight="bold", ) ax.set_ylabel("Number of Records", fontsize=12) ax.set_title( f"ID Changes: {summary['old_version']} → {summary['new_version']}", fontsize=14, fontweight="bold", ) ax.set_ylim(0, max(counts) * 1.15) ax.yaxis.set_major_formatter(plt.FuncFormatter(lambda x, p: f"{int(x):,}")) ax.grid(axis="y", alpha=0.3) plt.tight_layout() plt.savefig(output_path, dpi=150, bbox_inches="tight") plt.close() print(f"ID changes plot saved to {output_path}") def plot_tag_changes(comparison: Dict[str, Any], output_path: Path) -> None: """Create bar chart showing top tag count changes. Args: comparison: Comparison results dictionary output_path: Path to save the plot """ tag_changes = comparison["tag_changes"] top_movers = tag_changes.get("top_movers", {}) if not top_movers: print("No tag changes to plot") return # Get top 15 movers by absolute change top_items = list(top_movers.items())[:15] tags = [item[0] for item in top_items] changes = [item[1]["change"] for item in top_items] # Color bars by positive/negative change colors = ["#2ecc71" if c > 0 else "#e74c3c" for c in changes] fig, ax = plt.subplots(figsize=(12, 8)) y_pos = np.arange(len(tags)) bars = ax.barh(y_pos, changes, color=colors, alpha=0.7, edgecolor="black") # Add value labels for i, (bar, change) in enumerate(zip(bars, changes)): width = bar.get_width() label_x = width + (max(abs(c) for c in changes) * 0.02) if width < 0: label_x = width - (max(abs(c) for c in changes) * 0.02) ha = "right" else: ha = "left" ax.text( label_x, i, f"{int(change):+,}", ha=ha, va="center", fontsize=9, fontweight="bold", ) ax.set_yticks(y_pos) ax.set_yticklabels(tags, fontsize=9) ax.set_xlabel("Change in Count", fontsize=12) ax.set_title( f"Top Tag Count Changes: {comparison['summary']['old_version']} → {comparison['summary']['new_version']}", fontsize=14, fontweight="bold", ) ax.axvline(x=0, color="black", linestyle="-", linewidth=0.8) ax.xaxis.set_major_formatter(plt.FuncFormatter(lambda x, p: f"{int(x):,}")) ax.grid(axis="x", alpha=0.3) plt.tight_layout() plt.savefig(output_path, dpi=150, bbox_inches="tight") plt.close() print(f"Tag changes plot saved to {output_path}") def plot_quality_metrics(comparison: Dict[str, Any], output_path: Path) -> None: """Create grouped bar chart for quality metrics. Args: comparison: Comparison results dictionary output_path: Path to save the plot """ quality = comparison["quality_metrics"] metrics = ["Duplicates", "Mismatch Rate\n(×10000)", "Path Anomalies", "Empty Texts"] old_values = [ quality["duplicates"]["old"], quality["language_mismatch_rate"]["old"] * 10000, # Scale for visibility quality["path_anomalies"]["old"], quality["empty_texts"]["old"], ] new_values = [ quality["duplicates"]["new"], quality["language_mismatch_rate"]["new"] * 10000, quality["path_anomalies"]["new"], quality["empty_texts"]["new"], ] x = np.arange(len(metrics)) width = 0.35 fig, ax = plt.subplots(figsize=(12, 6)) bars1 = ax.bar( x - width / 2, old_values, width, label=comparison["summary"]["old_version"], color="#3498db", alpha=0.7, edgecolor="black", ) bars2 = ax.bar( x + width / 2, new_values, width, label=comparison["summary"]["new_version"], color="#e67e22", alpha=0.7, edgecolor="black", ) # Add value labels for bars in [bars1, bars2]: for bar in bars: height = bar.get_height() if height > 0: ax.text( bar.get_x() + bar.get_width() / 2.0, height, f"{height:,.0f}", ha="center", va="bottom", fontsize=9, ) ax.set_ylabel("Count / Rate", fontsize=12) ax.set_title( f"Quality Metrics Comparison: {comparison['summary']['old_version']} → {comparison['summary']['new_version']}", fontsize=14, fontweight="bold", ) ax.set_xticks(x) ax.set_xticklabels(metrics, fontsize=10) ax.legend(fontsize=11) ax.grid(axis="y", alpha=0.3) plt.tight_layout() plt.savefig(output_path, dpi=150, bbox_inches="tight") plt.close() print(f"Quality metrics plot saved to {output_path}") def plot_coverage_metrics(comparison: Dict[str, Any], output_path: Path) -> None: """Create grouped bar chart for coverage metrics. Args: comparison: Comparison results dictionary output_path: Path to save the plot """ old_summary = comparison["old_summary"] new_summary = comparison["new_summary"] total_old = old_summary["total_records"] total_new = new_summary["total_records"] metrics = ["Tags", "Text", "Language", "Weight", "Stems"] old_pct = [ (old_summary["records_with_tags"] / total_old * 100) if total_old > 0 else 0, (old_summary["records_with_text"] / total_old * 100) if total_old > 0 else 0, (old_summary["records_with_lang"] / total_old * 100) if total_old > 0 else 0, (old_summary["records_with_weight"] / total_old * 100) if total_old > 0 else 0, (old_summary["records_with_stems"] / total_old * 100) if total_old > 0 else 0, ] new_pct = [ (new_summary["records_with_tags"] / total_new * 100) if total_new > 0 else 0, (new_summary["records_with_text"] / total_new * 100) if total_new > 0 else 0, (new_summary["records_with_lang"] / total_new * 100) if total_new > 0 else 0, (new_summary["records_with_weight"] / total_new * 100) if total_new > 0 else 0, (new_summary["records_with_stems"] / total_new * 100) if total_new > 0 else 0, ] x = np.arange(len(metrics)) width = 0.35 fig, ax = plt.subplots(figsize=(12, 6)) bars1 = ax.bar( x - width / 2, old_pct, width, label=comparison["summary"]["old_version"], color="#9b59b6", alpha=0.7, edgecolor="black", ) bars2 = ax.bar( x + width / 2, new_pct, width, label=comparison["summary"]["new_version"], color="#1abc9c", alpha=0.7, edgecolor="black", ) # Add value labels for bars in [bars1, bars2]: for bar in bars: height = bar.get_height() ax.text( bar.get_x() + bar.get_width() / 2.0, height, f"{height:.1f}%", ha="center", va="bottom", fontsize=9, ) ax.set_ylabel("Coverage (%)", fontsize=12) ax.set_title( f"Field Coverage Comparison: {comparison['summary']['old_version']} → {comparison['summary']['new_version']}", fontsize=14, fontweight="bold", ) ax.set_xticks(x) ax.set_xticklabels(metrics, fontsize=10) ax.set_ylim(0, 105) ax.legend(fontsize=11) ax.grid(axis="y", alpha=0.3) plt.tight_layout() plt.savefig(output_path, dpi=150, bbox_inches="tight") plt.close() print(f"Coverage metrics plot saved to {output_path}") def create_all_plots(comparison: Dict[str, Any], output_dir: Path) -> None: """Create all visualization plots for a comparison. Args: comparison: Comparison results dictionary output_dir: Directory to save plots """ plots_dir = output_dir / "plots" plots_dir.mkdir(parents=True, exist_ok=True) print("\n" + "=" * 60) print("CREATING PLOTS") print("=" * 60) # ID changes plot plot_id_changes(comparison, plots_dir / "id_changes.png") # Tag changes plot plot_tag_changes(comparison, plots_dir / "tag_changes.png") # Quality metrics plot plot_quality_metrics(comparison, plots_dir / "quality_metrics.png") # Coverage metrics plot plot_coverage_metrics(comparison, plots_dir / "coverage_metrics.png") def plot_overall_trends(all_comparisons: List[Dict[str, Any]], output_dir: Path) -> None: """Create line plots showing trends across all versions. Args: all_comparisons: List of comparison dictionaries output_dir: Directory to save plots """ if not all_comparisons: print("No comparisons to plot") return plots_dir = output_dir / "plots" plots_dir.mkdir(parents=True, exist_ok=True) # Extract version labels and data versions = [all_comparisons[0]["summary"]["old_version"]] for comp in all_comparisons: versions.append(comp["summary"]["new_version"]) total_records = [all_comparisons[0]["old_summary"]["total_records"]] for comp in all_comparisons: total_records.append(comp["new_summary"]["total_records"]) duplicates = [all_comparisons[0]["quality_metrics"]["duplicates"]["old"]] for comp in all_comparisons: duplicates.append(comp["quality_metrics"]["duplicates"]["new"]) mismatch_rates = [all_comparisons[0]["quality_metrics"]["language_mismatch_rate"]["old"]] for comp in all_comparisons: mismatch_rates.append(comp["quality_metrics"]["language_mismatch_rate"]["new"]) # Plot 1: Dataset size trend fig, ax = plt.subplots(figsize=(12, 6)) ax.plot( versions, [r / 1e6 for r in total_records], marker="o", linewidth=2, markersize=8, color="#3498db", ) ax.set_ylabel("Total Records (Millions)", fontsize=12) ax.set_xlabel("Version", fontsize=12) ax.set_title("Dataset Size Trend Across Versions", fontsize=14, fontweight="bold") ax.grid(True, alpha=0.3) plt.xticks(rotation=45) plt.tight_layout() plt.savefig(plots_dir / "dataset_size_trend.png", dpi=150, bbox_inches="tight") plt.close() print(f"Dataset size trend plot saved to {plots_dir / 'dataset_size_trend.png'}") # Plot 2: Quality metrics trend fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 10)) # Duplicates ax1.plot(versions, duplicates, marker="o", linewidth=2, markersize=8, color="#e74c3c") ax1.set_ylabel("Duplicate IDs", fontsize=12) ax1.set_title("Quality Metrics Trends Across Versions", fontsize=14, fontweight="bold") ax1.grid(True, alpha=0.3) plt.setp(ax1.xaxis.get_majorticklabels(), rotation=45) # Mismatch rates ax2.plot( versions, [r * 100 for r in mismatch_rates], marker="s", linewidth=2, markersize=8, color="#f39c12", ) ax2.set_ylabel("Language Mismatch Rate (%)", fontsize=12) ax2.set_xlabel("Version", fontsize=12) ax2.grid(True, alpha=0.3) plt.setp(ax2.xaxis.get_majorticklabels(), rotation=45) plt.tight_layout() plt.savefig(plots_dir / "quality_trends.png", dpi=150, bbox_inches="tight") plt.close() print(f"Quality trends plot saved to {plots_dir / 'quality_trends.png'}")