"""THE SCRATCHES I DID NOT MAKE — 他人の傷 Several machines. One plate. Everything about how a machine decides is imported from scribe.py and is not changed by a single character: it predicts the density of the metal ahead, it learns only from being wrong, and only a failed prediction bends its line. There is no goal, no reward and no fitness. The single difference is that the plate is shared. A machine has no sensor for other machines, no channel to them, and nothing about them in its six weights. It reads a density. It cannot tell which grooves are its own. Other machines exist for it only as terrain it cannot account for. Why this run exists ------------------- TEN DAYS ended because the machine ran out of error. Eighty hours bought half a point of new coverage: it had marked everything it could reach, and a surface it made entirely is a surface it can predict entirely. The stylus saturated at the same time, which is the same fact told in metal. If error is the only thing that bends the line, error has to come from somewhere the machine cannot smooth flat. This run asks whether another machine is such a place — one running the identical rule, so that nothing has been added to the world except that some of the marks are not its own. Reading the numbers ------------------- More machines cut more metal per hour, so at equal time they face a fuller plate. That confound points the wrong way and has to be removed, so nothing here is compared at equal time. The run records the state of every machine at the moment the plate crosses 5%, 10%, ... 90% marked, whatever hour that happens to be, and reports the mean error over each coverage interval. The finer measurement is the one that carries the claim. On sampled steps the terrain a machine has just read is classified by who cut it, and its error is filed accordingly: metal it cut itself, metal cut by somebody else, a mixture, or metal nobody has touched. Same machine, same instant, same rule — only the authorship of the surface differs. Both measurements are invisible to the machines. Authorship is recorded in a parallel array that nothing in scribe.step can reach. Usage: python3 scribe_others.py --machines 4 --hours 8 --out ../plate/others/n4 """ from __future__ import annotations import argparse import json import math import os import time import numpy as np from scribe import (Machine, Metal, step, thin, PLATE_W, PLATE_H, GROOVE, SCRIBE_MM_S, CELL, DS, SENSE_R, NW, NH) START_R = 30.0 # mm, machines start on a circle this far from centre SEED_HEADING = 0.6 # rad, as in the single session FOREIGN_EVERY = 4 # classify the terrain every Nth step (measurement only) WINDOW_FRAC = 0.125 # first / last eighth of a run, for early vs late SELF_MAX = 0.05 # terrain counted as its own below this foreign fraction FOREIGN_MIN = 0.95 # ... and as somebody else's above this # where every machine is photographed, in coverage rather than in hours COV_MARKS = (0.05, 0.10, 0.20, 0.30, 0.40, 0.50, 0.60, 0.65, 0.70, 0.75, 0.78, 0.80, 0.82, 0.84, 0.86, 0.88, 0.90) class SharedMetal(Metal): """One plate, several machines. Alongside the depth grid it keeps a parallel record of which machine was first to mark each cell. No machine can read that record: what they sense is exactly the density scribe.Metal provides, with no authorship in it. The record exists so that afterwards we can ask whether a machine was more often wrong on metal it had not cut itself. """ def __init__(self, n): super().__init__() self.owner = bytearray(NW * NH) self.first_cut = [0] * (n + 2) # cells this machine was first to mark self.on_self = [0] * (n + 2) # cell visits onto its own earlier marks self.on_foreign = [0] * (n + 2) # cell visits onto marks made by others def cut(self, x, y, mid=0): cx, cy = int(x / CELL), int(y / CELL) if 0 <= cx < NW and 0 <= cy < NH and mid: i = cy * NW + cx o = self.owner[i] if o == 0: self.owner[i] = mid self.first_cut[mid] += 1 elif o == mid: self.on_self[mid] += 1 else: self.on_foreign[mid] += 1 super().cut(x, y, mid) def owner_at(self, x, y): cx, cy = int(x / CELL), int(y / CELL) if 0 <= cx < NW and 0 <= cy < NH: return self.owner[cy * NW + cx] return 0 def foreign(self, x, y, me, r=SENSE_R): """Of the cut metal around a point, what fraction was cut by others. Returns -1.0 on untouched metal, where the question has no answer. Measurement only — this is never part of what a machine senses. """ cx, cy = int(x / CELL), int(y / CELL) cut = 0 other = 0 for dy in range(-r, r + 1): yy = cy + dy if yy < 0 or yy >= NH: continue base = yy * NW for dx in range(-r, r + 1): xx = cx + dx if xx < 0 or xx >= NW: continue o = self.owner[base + xx] if o: cut += 1 if o != me: other += 1 return other / cut if cut else -1.0 AUTHOR = ("virgin", "own", "mixed", "foreign") DENS_EDGES = (0.10, 0.30, 0.50, 0.70) # bins of read density DENS_LABEL = ("0.0-0.1", "0.1-0.3", "0.3-0.5", "0.5-0.7", "0.7-1.0") def dens_bin(a): for i, e in enumerate(DENS_EDGES): if a < e: return i return len(DENS_EDGES) class Probe: """What is measured about one machine. The machine can reach none of it. Errors are filed in a table of read density against authorship, because the two are not independent: a machine is nearly always standing in a patch it has just cut itself, so its own metal is on average much denser than the metal it meets from somebody else. Comparing own against foreign without holding density fixed measures that difference in density and calls it a difference in authorship. The table lets the contrast be taken bin by bin. """ __slots__ = ("cell", "e_ae", "e_n", "e_str", "e_mm", "l_ae", "l_n", "l_str", "l_mm", "iv_ae", "iv_n", "iv_str", "iv_mm") def __init__(self): # cell[density bin][authorship] = [sum of |error|, count] self.cell = [[[0.0, 0] for _ in AUTHOR] for _ in DENS_LABEL] self.e_ae = self.e_str = self.e_mm = 0.0 self.l_ae = self.l_str = self.l_mm = 0.0 self.iv_ae = self.iv_str = self.iv_mm = 0.0 self.e_n = self.l_n = self.iv_n = 0 def file(self, actual, ff, ae): a = 0 if ff < 0.0 else (1 if ff <= SELF_MAX else (3 if ff >= FOREIGN_MIN else 2)) c = self.cell[dens_bin(actual)][a] c[0] += ae c[1] += 1 def interval(self): """Mean error and straightness since the last coverage mark, then reset.""" out = (self.iv_ae / self.iv_n if self.iv_n else 0.0, self.iv_str / self.iv_mm if self.iv_mm else 0.0) self.iv_ae = 0.0 self.iv_n = 0 self.iv_str = 0.0 self.iv_mm = 0.0 return out def table(self): return {DENS_LABEL[b]: {AUTHOR[a]: [round(c[0] / c[1], 6), c[1]] for a, c in enumerate(row) if c[1]} for b, row in enumerate(self.cell) if any(c[1] for c in row)} def pooled(probes): """Add every machine's table together, cell by cell.""" tot = [[[0.0, 0] for _ in AUTHOR] for _ in DENS_LABEL] for p in probes: for b, row in enumerate(p.cell): for a, c in enumerate(row): tot[b][a][0] += c[0] tot[b][a][1] += c[1] return tot def contrast(tot, a_hi=3, a_lo=1): """Mean error on foreign metal minus own metal, holding density fixed. Each density bin contributes in proportion to how much evidence it has (the smaller of the two counts), so bins where one side is barely sampled cannot swing the answer. """ num = den = 0.0 rows = [] for b, row in enumerate(tot): hi, lo = row[a_hi], row[a_lo] if not hi[1] or not lo[1]: continue mh, ml = hi[0] / hi[1], lo[0] / lo[1] wt = min(hi[1], lo[1]) num += wt * (mh - ml) den += wt rows.append({"density": DENS_LABEL[b], AUTHOR[a_lo]: round(ml, 6), AUTHOR[a_hi]: round(mh, 6), "ratio": round(mh / ml, 3) if ml else None, "n_" + AUTHOR[a_lo]: lo[1], "n_" + AUTHOR[a_hi]: hi[1]}) return (round(num / den, 6) if den else None), int(den), rows def start_state(n, k): """Initial positions and headings. One machine starts where the single session started: the centre of the plate, heading 0.6. Several machines are placed on a circle so that the whole arrangement maps onto itself under rotation — the configuration adds no asymmetry of its own, and no machine is given a better place to stand. """ if n == 1: return PLATE_W * 0.5, PLATE_H * 0.5, SEED_HEADING phi = 2.0 * math.pi * k / n return (PLATE_W * 0.5 + START_R * math.cos(phi), PLATE_H * 0.5 + START_R * math.sin(phi), SEED_HEADING + phi) def territories(metal, n): """Where each machine's own marks ended up, and how tangled they are. np.roll wraps at the edges of the sheet; the effect on a 420x297 grid is a single row and column and is left alone rather than special-cased. """ O = np.frombuffer(bytes(metal.owner), dtype=np.uint8).reshape(NH, NW) ys, xs = np.mgrid[0:NH, 0:NW] out = [] for mid in range(1, n + 1): sel = O == mid c = int(sel.sum()) if not c: out.append({"mid": mid, "cells": 0}) continue cx = float(xs[sel].mean()) * CELL cy = float(ys[sel].mean()) * CELL rr = float(np.sqrt(((xs[sel] * CELL - cx) ** 2 + (ys[sel] * CELL - cy) ** 2).mean())) out.append({ "mid": mid, "cells": c, "share_of_marked": round(c / max(1, int((O > 0).sum())), 4), "centroid_mm": [round(cx, 1), round(cy, 1)], "rms_radius_mm": round(rr, 1), }) border = np.zeros((NH, NW), dtype=bool) for dy in (-1, 0, 1): for dx in (-1, 0, 1): if dx == 0 and dy == 0: continue S = np.roll(np.roll(O, dy, 0), dx, 1) border |= (O > 0) & (S > 0) & (O != S) marked = int((O > 0).sum()) return out, round(int(border.sum()) / max(1, marked), 4) def run(n=4, hours=8.0, out_dir="../plate/others", log_every=4000): out = os.path.abspath(out_dir) os.makedirs(out, exist_ok=True) metal = SharedMetal(n) machines = [] for k in range(n): x, y, th = start_state(n, k) machines.append(Machine(x, y, th, mid=k + 1)) probes = [Probe() for _ in machines] steps = int(hours * 3600.0 * SCRIBE_MM_S / DS) # per machine win = max(1, int(steps * WINDOW_FRAC)) pts = [[(m.x, m.y)] for m in machines] traces = [[] for _ in machines] marks = list(COV_MARKS) checkpoints = [] wall0 = time.time() report = max(1, steps // 10) for st in range(steps): # one round = one step for every machine. The styli move together; a # machine reads whatever is on the plate at the instant it looks, which # includes marks another machine made moments earlier. for i, m in enumerate(machines): d_here, d_left, d_right, pred, actual, ae, turn, seg = step(m, metal) pts[i].append((m.x, m.y)) p = probes[i] p.iv_ae += ae p.iv_n += 1 p.iv_mm += seg if abs(turn) < 0.02: p.iv_str += seg if st < win: p.e_ae += ae p.e_n += 1 p.e_mm += seg if abs(turn) < 0.02: p.e_str += seg elif st >= steps - win: p.l_ae += ae p.l_n += 1 p.l_mm += seg if abs(turn) < 0.02: p.l_str += seg # who cut the ground it has just read, and how dense that ground is if st % FOREIGN_EVERY == 0: p.file(actual, metal.foreign(m.ax, m.ay, m.mid), ae) if st % log_every == 0: traces[i].append({ "step": st, "mm": round(m.cut_mm, 1), "coverage": round(metal.coverage(), 5), "surprise": round(ae, 5), "surprise_avg": round(m.surprise_run, 5), "turn": round(turn, 5), "pred": round(pred, 4), "actual": round(actual, 4), }) cov = metal.coverage() while marks and cov >= marks[0]: mark = marks.pop(0) per = [pr.interval() for pr in probes] checkpoints.append({ "mark": mark, "coverage": round(cov, 5), "step": st, "machine_hours_each": round(machines[0].cut_mm / SCRIBE_MM_S / 3600.0, 3), "cut_m_total": round(sum(m.cut_mm for m in machines) / 1000.0, 1), "interval_surprise": round(sum(a for a, _ in per) / len(per), 6), "interval_straight_fraction": round(sum(b for _, b in per) / len(per), 4), "per_machine_surprise": [round(a, 6) for a, _ in per], }) print(" coverage %.2f at %.2f machine-h each (%.0f m total): " "surprise %.5f, straight %.1f%%" % (mark, checkpoints[-1]["machine_hours_each"], checkpoints[-1]["cut_m_total"], checkpoints[-1]["interval_surprise"], 100 * checkpoints[-1]["interval_straight_fraction"]), flush=True) if st % report == 0: print(" step %7d/%d cov %.3f %.0fs" % (st, steps, cov, time.time() - wall0), flush=True) terr, border = territories(metal, n) mstats = [] for i, m in enumerate(machines): p = probes[i] visits = metal.first_cut[m.mid] + metal.on_self[m.mid] + metal.on_foreign[m.mid] mstats.append({ "mid": m.mid, "start_mm": [round(pts[i][0][0], 1), round(pts[i][0][1], 1)], "cut_m": round(m.cut_mm / 1000.0, 2), "machine_hours": round(m.cut_mm / SCRIBE_MM_S / 3600.0, 3), "straight_fraction": round(m.straight_mm / max(1e-9, m.cut_mm), 4), "mean_surprise": round(m.surprise_sum / max(1, steps), 6), "max_surprise": round(m.max_surprise, 5), "early_surprise": round(p.e_ae / max(1, p.e_n), 6), "late_surprise": round(p.l_ae / max(1, p.l_n), 6), "early_straight_fraction": round(p.e_str / max(1e-9, p.e_mm), 4), "late_straight_fraction": round(p.l_str / max(1e-9, p.l_mm), 4), "surprise_by_density_and_authorship": p.table(), "cells_first_marked": metal.first_cut[m.mid], "cell_visits_onto_own": metal.on_self[m.mid], "cell_visits_onto_foreign": metal.on_foreign[m.mid], "fraction_of_cutting_onto_foreign": round(metal.on_foreign[m.mid] / max(1, visits), 4), "final_weights": [round(v, 4) for v in m.w], }) # pool every machine's table, then take the contrast bin by bin tot = pooled(probes) raw = {} for a, name in enumerate(AUTHOR): s = sum(row[a][0] for row in tot) c = sum(row[a][1] for row in tot) raw[name] = [round(s / c, 6) if c else None, c] d_foreign, n_foreign, rows_foreign = contrast(tot, 3, 1) d_mixed, n_mixed, rows_mixed = contrast(tot, 2, 1) agg = { "mean_surprise_by_authorship_unmatched": raw, "note_on_unmatched": ("Not a fair comparison: own metal is on average much " "denser than foreign metal, because a machine is always " "standing in what it just cut. Use the matched figures."), "density_matched_foreign_minus_own": d_foreign, "density_matched_foreign_minus_own_weight": n_foreign, "density_matched_mixed_minus_own": d_mixed, "density_matched_mixed_minus_own_weight": n_mixed, "by_density_foreign_vs_own": rows_foreign, "by_density_mixed_vs_own": rows_mixed, } for i, m in enumerate(machines): kept = thin(pts[i]) with open(os.path.join(out, "stroke_m%02d.json" % m.mid), "w") as f: json.dump({"pts": [[round(p[0], 2), round(p[1], 2)] for p in kept]}, f) with open(os.path.join(out, "trace_m%02d.json" % m.mid), "w") as f: json.dump(traces[i], f) mstats[i]["points_stored"] = len(kept) np.save(os.path.join(out, "owner.npy"), np.frombuffer(bytes(metal.owner), dtype=np.uint8).copy()) np.save(os.path.join(out, "metal.npy"), np.frombuffer(bytes(metal.g), dtype=np.uint8).copy()) with open(os.path.join(out, "machines.json"), "w") as f: json.dump({"machines": mstats, "territories": terr}, f, ensure_ascii=False, indent=1) with open(os.path.join(out, "checkpoints.json"), "w") as f: json.dump({"marks": checkpoints}, f, ensure_ascii=False, indent=1) summary = { "machines": n, "hours_each": hours, "machine_hours_total": round(n * hours, 2), "cut_m_total": round(sum(m.cut_mm for m in machines) / 1000.0, 1), "coverage_final": round(metal.coverage(), 5), "surprise_by_authorship": agg, "early_surprise_mean": round(sum(s["early_surprise"] for s in mstats) / n, 6), "late_surprise_mean": round(sum(s["late_surprise"] for s in mstats) / n, 6), "early_straight_mean": round(sum(s["early_straight_fraction"] for s in mstats) / n, 4), "late_straight_mean": round(sum(s["late_straight_fraction"] for s in mstats) / n, 4), "fraction_of_cutting_onto_foreign_mean": round(sum(s["fraction_of_cutting_onto_foreign"] for s in mstats) / n, 4), "territory_border_fraction": border, "steps_each": steps, "window_steps": win, "foreign_sampled_every": FOREIGN_EVERY, "plate_mm": [PLATE_W, PLATE_H], "groove_mm": GROOVE, "wall_s": round(time.time() - wall0, 1), "note": ("The rule of motion is scribe.step, imported unchanged. The only " "difference from the single session is that the plate is shared. " "No machine can sense another; authorship is recorded in a parallel " "array that the rule cannot reach. Compare across runs at equal " "coverage, never at equal time."), } with open(os.path.join(out, "summary.json"), "w") as f: json.dump(summary, f, ensure_ascii=False, indent=1) print(json.dumps(summary, ensure_ascii=False)) return summary if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--machines", type=int, default=4) ap.add_argument("--hours", type=float, default=8.0) ap.add_argument("--out", default="../plate/others") ap.add_argument("--log-every", type=int, default=4000) args = ap.parse_args() run(n=args.machines, hours=args.hours, out_dir=args.out, log_every=args.log_every)