#!/usr/bin/env python3 """Schelling (1971) の分居モデル — 株式会社DIGIT「ひとつ変えてみる No.01 / sim-lab ep01」の独自実装。 原典: Schelling, T. C. (1971). Dynamic models of segregation. Journal of Mathematical Sociology, 1(2), 143-186. doi:10.1080/0022250x.1971.9989794 このファイルだけで、世界の定義・動かし方・測り方・全実験の実行がそろう。 数値は全部 `sim/results/*.json` に書き出し、台本と画面はそこからしか数字を取らない。 ■ 世界 - 正方格子 size×size（既定 100×100 = 1万区画）。空き区画の割合 empty（既定 10%）。 - 住民は n_colors 色（既定2色）。色ごとの人数は等分（割り切れない分は若い色へ1人ずつ）。 - 隣人 = ムーア近傍の8区画のうち、住民がいる区画。 - 街の形: bounded（長方形・端あり。角は3人、辺は5人まで）／ torus（ドーナツ型・端なし）。 ■ 満足の規則 - 住民 i の「同色隣人率」 s_i = (同じ色の隣人の数) / (住民のいる隣の区画の数)。 - s_i >= t_i なら満足、s_i < t_i なら不満。t_i は許容下限（既定は全員同じ t）。 - 隣に誰もいない住民は満足とする（比べる相手がいない）。 - 比較は分数のまま行う（same >= t*occ - 1e-9）。8人以下の分数には 0.30 と 0.33 の間の値が無いので、 t=0.30 と t=0.33 は同じ規則になる（結果も一致する。results で検査する）。 ■ 動き（1ステップ = 「1年」と呼ぶ。現実の1年とは無関係なラベル） 1. ステップの頭で不満な住民を全員リストにし、乱数で順番を決める。 2. 順番が来た住民について: a. その時点で満足になっていれば動かない（前の人の引越しで状況が変わるため、再判定する）。 b. 引越し確率 move_prob < 1 のときは、確率 move_prob でだけ動く（それ以外はこの年は我慢する）。 c. 動く場合、いったん元の区画を空け、空き区画（元の区画は除く）のうち 「そこへ移れば満足できる」区画から一様に1つ選んで移る。 満足できる空き区画が1つも無ければ、空き区画から一様に1つ選んで移る（我慢して移る）。 3. ステップの頭で不満な住民が0人なら「収束」として止める。上限 max_steps で打ち切る。 ■ 無頓着な住民（介入 b） - 住民の割合 indifferent を無作為に選び、その人の許容下限を 0 にする（決して不満にならない＝動かない）。 ■ 測り方（毎ステップ記録） - mean_same : 同色隣人率 s_i の平均（隣人0人の住民は分母から除く）。 - share_majority : s_i >= 0.5 の住民の割合（「隣の半分以上が同じ色」）。同上。 - share_all_same : s_i == 1 の住民の割合（「隣が全員同じ色」）。同上。 - unhappy : 不満な住民の割合（そのステップの頭の時点）。 - moves : そのステップで実際に引っ越した人数（満足できない引越しも含む）。 - dissimilarity : 分離度。街を district×district（既定 10×10 区画）の「地区」100個に分け、 多群の相違指数 D を計算する（Reardon & Firebaugh 2002 の式。2色では Duncan & Duncan 1955 の相違指数 D と一致する）: D = Σ_j Σ_m t_j |π_jm − π_m| / (2 T I), I = Σ_m π_m (1 − π_m) t_j = 地区 j の住民数、T = 総住民数、π_jm = 地区 j の色 m の割合、π_m = 全体の割合。 0 = どの地区も全体と同じ混ざり方、1 = どの地区も1色だけ。 無作為に混ぜても地区が有限なので 0 にはならない（参照値を results に記録）。 ■ 乱数 - シード s ごとに3本の独立な乱数列を使う: 配置 default_rng([s, 0])、無頓着の指名 default_rng([s, 1])、 動き default_rng([s, 2])。同じシードなら条件が違っても最初の街は同じ（色数が同じ限り）。 実行: .venv/bin/python sim/schelling.py --out sim/results # 全実験（12並列） .venv/bin/python sim/schelling.py --quick --out /tmp/x # 試験用の小さな実行 """ from __future__ import annotations import argparse import hashlib import json import math import os import sys import time from dataclasses import dataclass, asdict, field from multiprocessing import Pool import numpy as np NBR_OFFSETS = [(-1, -1), (-1, 0), (-1, 1), (0, -1), (0, 1), (1, -1), (1, 0), (1, 1)] EPS = 1e-9 SEEDS = [1001, 1002, 1003, 1004, 1005, 1006, 1007, 1008, 1009, 1010] DISPLAY_SEED = SEEDS[0] # 画面に出す街。10本のうち最初の1本（選んでいない） # --------------------------------------------------------------------------- # 測り方（全部ベクトル化。動きのループとは独立に、格子だけから計算できる） # --------------------------------------------------------------------------- def neighbour_counts(grid: np.ndarray, n_colors: int, torus: bool) -> np.ndarray: """counts[c-1, y, x] = 区画 (y, x) の8近傍にいる色 c の住民数。""" H, W = grid.shape out = np.zeros((n_colors, H, W), dtype=np.int16) for c in range(1, n_colors + 1): m = (grid == c).astype(np.int16) if torus: s = np.zeros_like(m) for dy, dx in NBR_OFFSETS: s += np.roll(np.roll(m, dy, axis=0), dx, axis=1) else: p = np.pad(m, 1) s = np.zeros_like(m) for dy, dx in NBR_OFFSETS: s += p[1 + dy:1 + dy + H, 1 + dx:1 + dx + W] out[c - 1] = s return out def same_shares(grid: np.ndarray, counts: np.ndarray): """(same, occ, has) を返す。has = 住民がいて隣人が1人以上いる区画。""" occ = counts.sum(axis=0) same = np.zeros(grid.shape, dtype=np.int16) for c in range(1, counts.shape[0] + 1): sel = grid == c same[sel] = counts[c - 1][sel] has = (grid > 0) & (occ > 0) return same, occ, has def dissimilarity(grid: np.ndarray, n_colors: int, district: int = 10) -> float: """多群の相違指数 D（Reardon & Firebaugh 2002）。2色なら Duncan & Duncan (1955) の D。""" H, W = grid.shape assert H % district == 0 and W % district == 0, "格子は地区の大きさで割り切れること" blocks = grid.reshape(H // district, district, W // district, district) tjm = np.stack([(blocks == c).sum(axis=(1, 3)).ravel() for c in range(1, n_colors + 1)], axis=1).astype(float) tj = tjm.sum(axis=1) T = tj.sum() pim = tjm.sum(axis=0) / T I = float((pim * (1 - pim)).sum()) nz = tj > 0 pijm = np.zeros_like(tjm) pijm[nz] = tjm[nz] / tj[nz, None] D = float((tj[:, None] * np.abs(pijm - pim[None, :])).sum() / (2 * T * I)) return D def expected_dissimilarity_random(district_sizes, p: float) -> float: """2色・各住民の色が独立に確率 p の場合の D の期待値（二項分布の平均絶対偏差を厳密に足す）。 2色の D = Σ_j |b_j − m_j p| / (2 T p (1−p)) （b_j = 地区 j の色1の人数、m_j = 地区 j の住民数） """ from math import comb T = float(sum(district_sizes)) tot = 0.0 cache = {} for m in district_sizes: m = int(m) if m not in cache: mad = 0.0 for b in range(m + 1): mad += comb(m, b) * (p ** b) * ((1 - p) ** (m - b)) * abs(b - m * p) cache[m] = mad tot += cache[m] return tot / (2 * T * p * (1 - p)) def measure(grid: np.ndarray, n_colors: int, torus: bool, tol: np.ndarray | None = None, district: int = 10) -> dict: counts = neighbour_counts(grid, n_colors, torus) same, occ, has = same_shares(grid, counts) share = same[has].astype(float) / occ[has] out = { "mean_same": float(share.mean()) if share.size else float("nan"), "share_majority": float((share >= 0.5 - EPS).mean()) if share.size else float("nan"), "share_all_same": float((share >= 1.0 - EPS).mean()) if share.size else float("nan"), "dissimilarity": dissimilarity(grid, n_colors, district), } if tol is not None: unhappy = has & (same < tol * occ - EPS) out["unhappy"] = float(unhappy.sum() / (grid > 0).sum()) return out HIST_BINS = 10 def share_hist(grid: np.ndarray, n_colors: int, torus: bool) -> list: """同色隣人率の分布（10階級: [0,0.1), [0.1,0.2), …, [0.9,1.0]）。住民の割合で返す。""" counts = neighbour_counts(grid, n_colors, torus) same, occ, has = same_shares(grid, counts) share = same[has].astype(float) / occ[has] idx = np.minimum((share * HIST_BINS + EPS).astype(int), HIST_BINS - 1) h = np.bincount(idx, minlength=HIST_BINS).astype(float) return (h / h.sum()).tolist() def grid_to_str(grid: np.ndarray) -> str: return "".join(map(str, grid.ravel().tolist())) def grid_sha(grid: np.ndarray) -> str: return hashlib.sha256(grid.astype(np.int8).tobytes()).hexdigest() # --------------------------------------------------------------------------- # モデル # --------------------------------------------------------------------------- @dataclass class Condition: key: str tol: float = 0.30 n_colors: int = 2 torus: bool = False indifferent: float = 0.0 move_prob: float = 1.0 size: int = 100 empty: float = 0.10 district: int = 10 max_steps: int = 400 class Model: def __init__(self, cond: Condition, seed: int): self.cond = cond self.seed = seed S = cond.size N = S * S self.S, self.N = S, N self.K = cond.n_colors rng_layout = np.random.default_rng([seed, 0]) rng_ind = np.random.default_rng([seed, 1]) self.rng = np.random.default_rng([seed, 2]) n_empty = int(round(N * cond.empty)) n_agents = N - n_empty labels = np.zeros(N, dtype=np.int8) per = [n_agents // self.K + (1 if i < n_agents % self.K else 0) for i in range(self.K)] pos = 0 for c, k in enumerate(per, start=1): labels[pos:pos + k] = c pos += k rng_layout.shuffle(labels) self.grid = labels # flat self.n_agents = n_agents self.tol = np.where(labels > 0, cond.tol, 0.0).astype(float) if cond.indifferent > 0: agents = np.flatnonzero(labels > 0) k = int(round(cond.indifferent * n_agents)) chosen = rng_ind.choice(agents, size=k, replace=False) self.tol[chosen] = 0.0 self.indiff_mask0 = (labels > 0) & (self.tol == 0.0) # 住民の背番号（誰が一度でも動いたかを数えるため） self.aid = np.full(N, -1, dtype=np.int32) self.aid[labels > 0] = np.arange(n_agents, dtype=np.int32) self.ever_moved = np.zeros(n_agents, dtype=bool) # 近傍表（端の外は番兵 N を指す） ys, xs = np.divmod(np.arange(N), S) nbr = np.empty((N, 8), dtype=np.int64) for k, (dy, dx) in enumerate(NBR_OFFSETS): yy, xx = ys + dy, xs + dx if cond.torus: nbr[:, k] = (yy % S) * S + (xx % S) else: ok = (yy >= 0) & (yy < S) & (xx >= 0) & (xx < S) nbr[:, k] = np.where(ok, yy * S + xx, N) self.nbr = nbr g2 = self.grid.reshape(S, S) counts = neighbour_counts(g2, self.K, cond.torus).reshape(self.K, N).astype(np.int32) self.cnt = np.zeros((self.K, N + 1), dtype=np.int32) self.cnt[:, :N] = counts self.occ = self.cnt.sum(axis=0) self.empties = np.flatnonzero(self.grid == 0).astype(np.int64) self.epos = np.full(N, -1, dtype=np.int64) self.epos[self.empties] = np.arange(self.empties.size) # -- 状態 ---------------------------------------------------------------- def grid2d(self) -> np.ndarray: return self.grid.reshape(self.S, self.S) def tol2d(self) -> np.ndarray: return self.tol.reshape(self.S, self.S) def unhappy_cells(self) -> np.ndarray: g = self.grid agents = g > 0 occ = self.occ[:self.N] same = np.zeros(self.N, dtype=np.int32) for c in range(1, self.K + 1): sel = g == c same[sel] = self.cnt[c - 1, :self.N][sel] unhappy = agents & (occ > 0) & (same < self.tol * occ - EPS) return np.flatnonzero(unhappy) def snapshot_metrics(self) -> dict: return measure(self.grid2d(), self.K, self.cond.torus, self.tol2d(), self.cond.district) # -- 1ステップ ----------------------------------------------------------- def step(self, log: list | None = None): """1ステップ進める。戻り値 (moves, moves_unsatisfied, n_unhappy_at_start)。""" unhappy = self.unhappy_cells() n0 = int(unhappy.size) if n0 == 0: return 0, 0, 0 order = self.rng.permutation(unhappy) grid, cnt, occ, nbr, tol = self.grid, self.cnt, self.occ, self.nbr, self.tol mp = self.cond.move_prob moves = 0 unsat = 0 for p in order: c = int(grid[p]) o = int(occ[p]) tp = float(tol[p]) if o == 0 or cnt[c - 1, p] >= tp * o - EPS: continue # もう満足している if mp < 1.0 and self.rng.random() >= mp: continue # 引越し費用: 今年は我慢 # 元の区画を空ける nb = nbr[p] cnt[c - 1, nb] -= 1 occ[nb] -= 1 grid[p] = 0 # 行き先 e = self.empties oe = occ[e] ok = (oe == 0) | (cnt[c - 1, e] >= tp * oe - EPS) cand = e[ok] if cand.size: dest = int(cand[self.rng.integers(cand.size)]) else: dest = int(e[self.rng.integers(e.size)]) unsat += 1 # 住む nd = nbr[dest] cnt[c - 1, nd] += 1 occ[nd] += 1 grid[dest] = c k = self.epos[dest] self.empties[k] = p self.epos[p] = k self.epos[dest] = -1 tol[dest] = tp tol[p] = 0.0 a = self.aid[p] self.aid[dest] = a self.aid[p] = -1 self.ever_moved[a] = True moves += 1 if log is not None: log.append((int(p), int(dest))) return moves, unsat, n0 def run(cond: Condition, seed: int, keep_display: bool = False) -> dict: t0 = time.perf_counter() m = Model(cond, seed) init_grid = m.grid2d().copy() met0 = m.snapshot_metrics() hist0 = share_hist(init_grid, m.K, cond.torus) init_unhappy = np.zeros(m.n_agents, dtype=bool) init_unhappy[m.aid[m.unhappy_cells()]] = True series = {k: [v] for k, v in met0.items()} series["moves"] = [0] series["ever_moved"] = [0.0] display = None if keep_display: display = {"init": grid_to_str(init_grid), "step_moves": [], "snapshots_every_k": 1, "snapshots": [grid_to_str(init_grid)]} steps = 0 converged = False total_moves = 0 total_unsat = 0 first_step_moves = None while True: log = [] if keep_display else None moves, unsat, n_unhappy = m.step(log) if n_unhappy == 0: converged = True break steps += 1 total_moves += moves total_unsat += unsat if first_step_moves is None: first_step_moves = moves met = m.snapshot_metrics() for k, v in met.items(): series[k].append(v) series["moves"].append(moves) series["ever_moved"].append(float(m.ever_moved.mean())) if keep_display: display["step_moves"].append(log) display["snapshots"].append(grid_to_str(m.grid2d())) if steps >= cond.max_steps: break final = m.snapshot_metrics() # 連鎖: 最初は満足していたのに、のちに引っ越した人／最初は不満だったのに、動かずに満足した人 cascade = m.ever_moved & ~init_unhappy resolved = init_unhappy & ~m.ever_moved # 収束時の unhappy は 0。打ち切り時は最後の測定値。 res = { "condition": cond.key, "seed": seed, "steps": steps, "converged": converged, "total_moves": total_moves, "moves_unsatisfied": total_unsat, "first_step_moves": first_step_moves or 0, "ever_moved_share": float(m.ever_moved.mean()), "init_unhappy_share": float(init_unhappy.mean()), "cascade_movers_share": float(cascade.mean()), "resolved_without_moving_share": float(resolved.mean()), "hist_initial": hist0, "hist_final": share_hist(m.grid2d(), m.K, cond.torus), "initial": met0, "final": final, "final_grid_sha256": grid_sha(m.grid2d()), "init_grid_sha256": grid_sha(init_grid), "series": series, "runtime_s": round(time.perf_counter() - t0, 3), } if keep_display: # 住民ごとの経歴を最終位置の区画に書く: 0 空き / 1 一度も不満にならず動かず / 2 最初から不満で引っ越した # / 3 最初は満足していたのに引っ越した（連鎖）/ 4 最初は不満だったが動かずに満足した status = np.zeros(m.N, dtype=np.int8) occ_cells = np.flatnonzero(m.grid > 0) a = m.aid[occ_cells] st = np.ones(a.size, dtype=np.int8) st[init_unhappy[a] & m.ever_moved[a]] = 2 st[~init_unhappy[a] & m.ever_moved[a]] = 3 st[init_unhappy[a] & ~m.ever_moved[a]] = 4 status[occ_cells] = st display["final_status"] = "".join(map(str, status.tolist())) res["display"] = display res["final_grid"] = grid_to_str(m.grid2d()) res["indifferent_cells_final"] = np.flatnonzero((m.grid > 0) & (m.tol == 0.0)).tolist() if cond.indifferent > 0 else [] res["indifferent_cells_init"] = np.flatnonzero(m.indiff_mask0).tolist() if cond.indifferent > 0 else [] return res # --------------------------------------------------------------------------- # 実験計画 # --------------------------------------------------------------------------- def experiment_plan(max_steps: int = 400, size: int = 100) -> dict: """実験群 → 条件のリスト。同じ条件は key で1回だけ回す。""" def C(key, **kw): return Condition(key=key, max_steps=max_steps, size=size, **kw) base = dict(tol=0.30, n_colors=2, torus=False, indifferent=0.0, move_prob=1.0) groups = { "baseline": [C("t030", **base), C("t033", **{**base, "tol": 0.33})], "sweep_t": [C(f"t{int(round(t*100)):03d}", **{**base, "tol": t}) for t in [0.10, 0.15, 0.20, 0.25, 0.30, 0.35, 0.40, 0.45, 0.50, 0.55, 0.60]], # 仕様（0.10〜0.60）の外側。どこで街が止まらなくなるかを見るための追加。 "sweep_t_ext": [C(f"t{int(round(t*100)):03d}", **{**base, "tol": t}) for t in [0.65, 0.70, 0.75, 0.80]], "indifferent": [C("t030", **base), C("indiff010", **{**base, "indifferent": 0.10}), C("indiff030", **{**base, "indifferent": 0.30})], "move_prob": [C("t030", **base), C("moveprob050", **{**base, "move_prob": 0.5}), C("moveprob020", **{**base, "move_prob": 0.2})], "colors_shape": [C("t030", **base), C("colors2_torus", **{**base, "torus": True}), C("colors3_bounded", **{**base, "n_colors": 3}), C("colors3_torus", **{**base, "n_colors": 3, "torus": True})], } return groups def _job(args): cond_dict, seed, keep = args cond = Condition(**cond_dict) return run(cond, seed, keep_display=keep) def agg(values): v = [x for x in values if x is not None and not (isinstance(x, float) and math.isnan(x))] return {"mean": float(np.mean(v)), "min": float(np.min(v)), "max": float(np.max(v)), "sd": float(np.std(v, ddof=1)) if len(v) > 1 else 0.0, "n": len(v), "values": v} def summarise(runs: list[dict]) -> dict: out = {"n": len(runs), "seeds": [r["seed"] for r in runs]} for k in ["mean_same", "share_majority", "share_all_same", "dissimilarity", "unhappy"]: out[f"initial.{k}"] = agg([r["initial"][k] for r in runs]) out[f"final.{k}"] = agg([r["final"][k] for r in runs]) out["steps"] = agg([r["steps"] for r in runs]) out["converged_share"] = float(np.mean([r["converged"] for r in runs])) out["converged_n"] = int(sum(r["converged"] for r in runs)) out["total_moves"] = agg([r["total_moves"] for r in runs]) out["first_step_moves"] = agg([r["first_step_moves"] for r in runs]) out["moves_unsatisfied"] = agg([r["moves_unsatisfied"] for r in runs]) out["ever_moved_share"] = agg([r["ever_moved_share"] for r in runs]) out["init_unhappy_share"] = agg([r["init_unhappy_share"] for r in runs]) out["cascade_movers_share"] = agg([r["cascade_movers_share"] for r in runs]) out["resolved_without_moving_share"] = agg([r["resolved_without_moving_share"] for r in runs]) out["hist_initial_mean"] = np.mean([r["hist_initial"] for r in runs], axis=0).tolist() out["hist_final_mean"] = np.mean([r["hist_final"] for r in runs], axis=0).tolist() out["final_grid_sha256"] = [r["final_grid_sha256"] for r in runs] return out def rnd(x, nd=6): if isinstance(x, float): return round(x, nd) if isinstance(x, dict): return {k: rnd(v, nd) for k, v in x.items()} if isinstance(x, list): return [rnd(v, nd) for v in x] return x def random_reference(cond: Condition, seeds, shuffles: int = 20) -> dict: """同じ人数・同じ空き区画のまま色だけを無作為に混ぜ直した街の指標（参照値）。""" vals = {"mean_same": [], "share_majority": [], "share_all_same": [], "dissimilarity": []} for s in seeds: m = Model(cond, s) g = m.grid2d().copy() rng = np.random.default_rng([s, 9]) occ_cells = np.flatnonzero(g.ravel() > 0) colours = g.ravel()[occ_cells].copy() for _ in range(shuffles): rng.shuffle(colours) h = np.zeros(g.size, dtype=np.int8) h[occ_cells] = colours met = measure(h.reshape(g.shape), cond.n_colors, cond.torus, None, cond.district) for k in vals: vals[k].append(met[k]) out = {k: agg(v) for k, v in vals.items()} for k in out: out[k].pop("values") if cond.n_colors == 2: g = Model(cond, seeds[0]).grid2d() d = cond.district sizes = (g.reshape(g.shape[0] // d, d, g.shape[1] // d, d) > 0).sum(axis=(1, 3)).ravel() out["dissimilarity_expected_binomial"] = expected_dissimilarity_random(sizes.tolist(), 0.5) out["n_note"] = f"{len(seeds)} シード × {shuffles} 回の混ぜ直し" return out def main(argv=None): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--out", default=os.path.join(os.path.dirname(__file__), "results")) ap.add_argument("--procs", type=int, default=min(11, os.cpu_count() or 2)) ap.add_argument("--quick", action="store_true", help="試験用: 40×40・3シード・上限60ステップ") args = ap.parse_args(argv) t_start = time.perf_counter() size, seeds, max_steps = (40, SEEDS[:3], 60) if args.quick else (100, SEEDS, 400) groups = experiment_plan(max_steps=max_steps, size=size) unique = {} for g, conds in groups.items(): for c in conds: unique[c.key] = c display_keys = set(unique.keys()) # 全条件でシード[0]の最終格子を残す。動きの記録は基準だけ。 jobs = [] for key, c in unique.items(): for s in seeds: keep = (s == seeds[0]) jobs.append((asdict(c), s, keep)) # 重い条件から先に投げる（t が高いほど遅い） jobs.sort(key=lambda j: (-j[0]["tol"], -1 * (j[0]["move_prob"] < 1))) os.makedirs(args.out, exist_ok=True) with Pool(args.procs) as pool: results = pool.map(_job, jobs, chunksize=1) by_key: dict[str, list] = {} for r in results: by_key.setdefault(r["condition"], []).append(r) for k in by_key: by_key[k].sort(key=lambda r: r["seed"]) conf = { "model": "Schelling (1971) segregation model — DIGIT own implementation (sim/schelling.py)", "size": size, "cells": size * size, "empty": 0.10, "agents": size * size - int(round(size * size * 0.10)), "empty_cells": int(round(size * size * 0.10)), "neighbourhood": "Moore 8", "district": 10, "districts": (size // 10) ** 2, "seeds": seeds, "display_seed": seeds[0], "n_per_condition": len(seeds), "max_steps": max_steps, "move_rule": "unhappy agents (re-checked at their turn) move to a uniformly random empty cell that would satisfy them; if none exists, to a uniformly random empty cell", "no_neighbour_rule": "agents with no occupied neighbour are counted as satisfied and excluded from share denominators", "conditions": {k: asdict(c) for k, c in unique.items()}, "random_reference_2col": rnd(random_reference(unique["t030"], seeds)), "random_reference_3col": rnd(random_reference(unique["colors3_bounded"], seeds)), "generated_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "numpy": np.__version__, "python": sys.version.split()[0], } # 実験群ごとの JSON summary = {"config": conf, "groups": {}} for g, conds in groups.items(): doc = {"group": g, "conditions": {}} for c in conds: runs = by_key[c.key] doc["conditions"][c.key] = { "params": asdict(c), "summary": summarise(runs), "runs": [{k: v for k, v in r.items() if k not in ("display", "final_grid", "indifferent_cells_final", "indifferent_cells_init")} for r in runs], } with open(os.path.join(args.out, f"{g}.json"), "w") as f: json.dump(rnd(doc), f, ensure_ascii=False) summary["groups"][g] = {k: rnd({kk: vv for kk, vv in v["summary"].items()}) for k, v in doc["conditions"].items()} # 画面用: 基準（t030・表示シード）の動きの全記録と、各条件の表示シードの最終格子 disp = {} for key, runs in by_key.items(): r0 = runs[0] disp[key] = {"seed": r0["seed"], "final_grid": r0["final_grid"], "init_grid": r0["display"]["init"], "steps": r0["steps"], "series": rnd(r0["series"]), "indifferent_cells_final": r0["indifferent_cells_final"]} with open(os.path.join(args.out, "display_finals.json"), "w") as f: json.dump({"size": size, "conditions": disp}, f) for key in ["t030", "moveprob020", "colors3_bounded", "indiff030"]: if key not in by_key: continue r0 = by_key[key][0] with open(os.path.join(args.out, f"display_run_{key}.json"), "w") as f: json.dump({"condition": key, "seed": r0["seed"], "size": size, "params": asdict(unique[key]), "init": r0["display"]["init"], "step_moves": r0["display"]["step_moves"], "snapshots_every_k": 1, "snapshots": r0["display"]["snapshots"], "final_status": r0["display"]["final_status"], "summary_seed": {k: r0[k] for k in ["steps", "total_moves", "first_step_moves", "ever_moved_share", "init_unhappy_share", "cascade_movers_share", "resolved_without_moving_share"]}, "series": rnd(r0["series"]), "indifferent_cells_init": r0["indifferent_cells_init"]}, f) # 検査: t030 と t033 は全シードで同じ最終格子になるはず same = [a["final_grid_sha256"] == b["final_grid_sha256"] for a, b in zip(by_key["t030"], by_key["t033"])] summary["checks"] = {"t030_equals_t033_all_seeds": all(same), "t030_equals_t033_per_seed": same} summary["runtime_s"] = round(time.perf_counter() - t_start, 2) summary["runtime_per_condition_s"] = {k: agg([r["runtime_s"] for r in v]) for k, v in by_key.items()} with open(os.path.join(args.out, "summary.json"), "w") as f: json.dump(rnd(summary), f, ensure_ascii=False, indent=1) print(f"done: {len(results)} runs, {len(unique)} conditions, {summary['runtime_s']} s -> {args.out}") return 0 if __name__ == "__main__": sys.exit(main())