feat(cluster): 输出簇分布图,直观展示簇间距离

在 preview 拼图的同级目录生成 cluster_scatter.png:
- 左:聚类空间的 PCA 二维投影,散点按簇着色 + 质心标记,标题标注两个主成分的方差占比
- 右:质心两两距离矩阵(在聚类空间里算,不是投影后),带数值标注

投影用 PCA 而不是 t-SNE/UMAP——后两者为了把簇拉开会扭曲簇间距离,
而这张图的目的就是看距离。

- ClusterResult 新增 points 字段,保留 KMeans 实际使用的空间供可视化复用,
  质心与散点用同一组主成分变换,避免质心飘到簇外
- summary.json 新增 centroid_distances
- 新增 scatter 配置块(enabled / max_points / point_size / alpha / figsize / dpi)
- 中文字体按 PingFang / Noto Sans CJK 等顺序探测,找不到自动退回英文标签,
  不会画成豆腐块
- 新增依赖 matplotlib

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Shaun Zhou
2026-09-04 15:09:35 +08:00
co-authored by Claude Opus 5
parent 4e6c8aa2ea
commit e4bd794025
8 changed files with 1693 additions and 1071 deletions
+40
View File
@@ -6,11 +6,13 @@ torch = pytest.importorskip("torch")
from engine.cluster import ( # noqa: E402
ClusterResult,
centroid_distances,
cluster_features,
cluster_stability,
kmeans,
normalized_mutual_info,
pca_reduce,
project_2d,
random_fit_index,
silhouette_score,
)
@@ -124,3 +126,41 @@ def test_silhouette_subsamples_large_inputs(blobs):
labels = kmeans(signal, 4, n_init=3, seed=0)[0]
assert silhouette_score(signal, labels, max_samples=50) > 0.5
assert torch.isnan(torch.tensor(silhouette_score(signal, torch.zeros_like(labels))))
def test_cluster_result_keeps_the_space_kmeans_ran_in(blobs):
"""可视化要复用 KMeans 实际使用的空间,否则画出来的距离和聚类依据对不上。"""
_, noisy = blobs
result = cluster_features(noisy, 4, pca_dim=16)
assert result.points is not None
assert result.points.shape == (200, 16)
assert result.centroids.shape[1] == result.points.shape[1]
def test_centroid_distances_is_a_proper_metric_matrix(blobs):
signal, _ = blobs
result = cluster_features(signal, 4, pca_dim=0)
matrix = centroid_distances(result.centroids)
assert matrix.shape == (4, 4)
assert torch.allclose(matrix, matrix.T) # 对称
assert torch.allclose(matrix.diagonal(), torch.zeros(4), atol=1e-5) # 对角为 0
assert matrix[matrix > 0].min() > 0
def test_project_2d_shapes_and_shared_basis(blobs):
signal, _ = blobs
result = cluster_features(signal, 4, pca_dim=0)
points, centroids, ratio = project_2d(result.points, result.centroids)
assert points.shape == (200, 2)
assert centroids.shape == (4, 2)
assert len(ratio) == 2 and 0 <= sum(ratio) <= 1.0001
# 分离良好的团:前两个主成分应当解释大部分方差
assert sum(ratio) > 0.5
# 质心必须用与散点相同的变换,否则图上质心会飘到簇外
for cluster in range(4):
members = points[result.labels == cluster]
assert torch.allclose(members.mean(dim=0), centroids[cluster], atol=1e-4)
+44
View File
@@ -337,6 +337,50 @@ def test_cluster_script_simclr_branch_end_to_end(tiny_image_dir, tmp_path, monke
assert (work_dir / "clusters" / "cluster_00").is_dir()
def test_scatter_plot_is_written_next_to_previews(tiny_image_dir, tmp_path, monkeypatch):
"""cluster_scatter.png 要和 preview 拼图落在同一级目录。"""
sys.path.insert(0, "scripts")
import cluster_unlabeled as cu
work_dir = tmp_path / "run"
monkeypatch.setattr(
sys, "argv",
["cluster_unlabeled.py", "--config", "configs/cluster/film.yaml",
"--work-dir", str(work_dir), "--root", str(tiny_image_dir), "--set",
"device=cpu", "loader.num_workers=0", "loader.batch_size=4",
"num_clusters=2", "n_init=2"],
)
assert cu.main() == 0
scatter = work_dir / "cluster_scatter.png"
assert scatter.exists() and scatter.stat().st_size > 5000
assert (work_dir / "cluster_00_preview.png").parent == scatter.parent
import json
summary = json.loads((work_dir / "summary.json").read_text(encoding="utf-8"))
matrix = summary["centroid_distances"]
assert len(matrix) == 2 and matrix[0][0] == pytest.approx(0.0)
assert matrix[0][1] == pytest.approx(matrix[1][0])
def test_scatter_can_be_disabled(tiny_image_dir, tmp_path, monkeypatch):
sys.path.insert(0, "scripts")
import cluster_unlabeled as cu
work_dir = tmp_path / "run_off"
monkeypatch.setattr(
sys, "argv",
["cluster_unlabeled.py", "--config", "configs/cluster/film.yaml",
"--work-dir", str(work_dir), "--root", str(tiny_image_dir), "--set",
"device=cpu", "loader.num_workers=0", "loader.batch_size=4",
"num_clusters=2", "n_init=2", "scatter.enabled=False"],
)
assert cu.main() == 0
assert not (work_dir / "cluster_scatter.png").exists()
assert (work_dir / "cluster_00_preview.png").exists()
def test_simclr_runs_on_the_resolved_device(tiny_image_dir, tmp_path, monkeypatch):
"""method=simclr 在本机自动选中的后端上(MPS / CUDA / CPU)都应能跑通。"""
sys.path.insert(0, "scripts")