Update app.py

app.py

@@ -70,126 +70,377 @@ def _sym(grid,axis):
             s[y,:]=(grid[y-r:y,:]==grid[y+1:y+r+1,:][::-1,:]).mean()
     return s
 
-# ── Im-side: candidate transforms ────────────────────────────────────────────
-
-def _h_mirror(grid):
-    H,W=grid.shape; ax,sc=_sym_axis(grid,'h')
-    lm=(grid[:,:ax]>0).sum(); rm=(grid[:,ax:]>0).sum()
-    if lm==0 or rm>=lm*0.7: return None,0.0
-    c=grid.copy()
-    for col in range(ax):
-        mir=W-1-col
-        if mir<W:
-            mask=c[:,mir]==0; c[mask,mir]=grid[mask,col]
-    return c,(1-rm/max(lm,1))*sc*0.95
-
-def _v_mirror(grid):
-    H,W=grid.shape; ax,sc=_sym_axis(grid,'v')
-    tm=(grid[:ax,:]>0).sum(); bm=(grid[ax:,:]>0).sum()
-    if tm==0 or bm>=tm*0.7: return None,0.0
-    c=grid.copy()
-    for row in range(ax):
-        mir=H-1-row
-        if mir<H:
-            mask=c[mir,:]==0; c[mir,mask]=grid[row,mask]
-    return c,(1-bm/max(tm,1))*sc*0.90
-
-def _boundary_only(grid):
-    if not (grid>0).any(): return None,0.0
-    solid=(grid>0).sum(); b=_boundary(grid); bpx=b.sum()
-    if solid==0 or bpx/solid>0.6: return None,0.0
-    c=np.zeros_like(grid); c[b>0]=grid[b>0]
-    return c,(1-bpx/solid)*0.85
-
-def _hollow_fill(grid):
-    b=_boundary(grid); interior=(grid==0)&(b==0)
-    if not interior.any() or not (grid>0).any(): return None,0.0
-    dom=np.argmax(np.bincount(grid[grid>0].flatten(),minlength=10)[1:])+1
-    c=grid.copy(); c[interior]=dom
-    return c,interior.sum()/max(1,(grid==0).sum())*0.80
-
-def _gravity(grid,d='down'):
-    H,W=grid.shape; c=np.zeros_like(grid)
-    if d=='down':
-        for col in range(W):
-            v=grid[:,col][grid[:,col]>0]
-            if len(v): c[H-len(v):H,col]=v
-    elif d=='up':
-        for col in range(W):
-            v=grid[:,col][grid[:,col]>0]
-            if len(v): c[:len(v),col]=v
-    elif d=='right':
-        for row in range(H):
-            v=grid[row,:][grid[row,:]>0]
-            if len(v): c[row,W-len(v):W]=v
-    elif d=='left':
-        for row in range(H):
-            v=grid[row,:][grid[row,:]>0]
-            if len(v): c[row,:len(v)]=v
-    if np.array_equal(c,grid) or not (grid>0).any(): return None,0.0
-    moved=(c!=grid).sum()
-    return c,min(0.75,moved/max(1,(grid>0).sum())*0.8)
-
-def _color_shift(grid,d=1):
-    if not (grid>0).any(): return None,0.0
-    c=grid.copy(); mask=grid>0
-    c[mask]=((grid[mask]-1+d)%9)+1
-    return c,0.45
-
-def _rotate(grid,k): return np.rot90(grid,k),0.30
-def _hflip(grid): return np.fliplr(grid),0.25
-def _vflip(grid): return np.flipud(grid),0.25
-
-def _4fold(grid):
-    c=grid.copy()
-    for k in [1,2,3]:
-        rot=np.rot90(grid,k)
-        if rot.shape==grid.shape:
-            mask=c==0; c[mask]=rot[mask]
-    return (c,0.55) if not np.array_equal(c,grid) else (None,0.0)
-
-TRANSFORMS=[
-    ('h_mirror_complete', _h_mirror),
-    ('v_mirror_complete', _v_mirror),
-    ('boundary_only',     _boundary_only),
-    ('hollow_fill',       _hollow_fill),
-    ('gravity_down',      lambda g: _gravity(g,'down')),
-    ('gravity_up',        lambda g: _gravity(g,'up')),
-    ('gravity_right',     lambda g: _gravity(g,'right')),
-    ('gravity_left',      lambda g: _gravity(g,'left')),
-    ('4fold_symmetry',    _4fold),
-    ('color_shift_+1',    lambda g: _color_shift(g,1)),
-    ('color_shift_+2',    lambda g: _color_shift(g,2)),
-    ('rotate_90',         lambda g: _rotate(g,1)),
-    ('rotate_180',        lambda g: _rotate(g,2)),
-    ('rotate_270',        lambda g: _rotate(g,3)),
-    ('h_flip',            _hflip),
-    ('v_flip',            _vflip),
-]
-
-def get_candidates(grid):
-    out=[]
-    for name,fn in TRANSFORMS:
-        try:
-            c,conf=fn(grid)
-            if c is not None and conf>0.05: out.append((name,c,conf))
-        except: pass
-    return sorted(out,key=lambda x:-x[2])
-
-def pixel_diff(cur,tgt):
-    if cur.shape!=tgt.shape: return []
-    return [(r,c,int(tgt[r,c]))
-            for r in range(cur.shape[0]) for c in range(cur.shape[1])
-            if cur[r,c]!=tgt[r,c]]
-
-def most_urgent_diff(cur,tgt):
-    diffs=pixel_diff(cur,tgt)
+# ── Re/Im board reader (inlined from arc_solver.py) ─────────────────────────
+
+"""
+arc_solver.py — Re/Im board reader for ARC-AGI-3
+=================================================
+
+The board IS a complex object M = Re(M) + i·Im(M).
+
+  Re(M) = multiplicative structure: what colors exist, how many pixels,
+          where objects are, their bounding boxes, centroids, density.
+
+  Im(M) = additive structure: symmetry axes, boundary contour, gradient
+          flow direction, winding/curl — the "where pointed" information.
+
+log separates them. i· swaps them.
+
+The answer is what you get by applying i· to the board:
+  "Read the Im side of the board → that tells you what the Re side
+   of the answer must look like → find the cells that need to change."
+
+Pipeline:
+  read_board(grid) → (re, im, answer_grid, confidence, reasoning)
+  pixel_diff(current, answer) → list of (r, c, target_color)
+  most_urgent_diff(current, answer) → single most important cell (Re coords)
+  try_analytic_action(frame, available) → (action_id, data, name, confidence)
+"""
+
+import numpy as np
+from typing import Optional, Tuple, List, Dict
+
+# ── Primitives ────────────────────────────────────────────────────────────────
+
+def _sobel(f):
+    p = np.pad(f, 1, mode='edge')
+    gx = (-p[:-2,:-2]-2*p[1:-1,:-2]-p[2:,:-2]+p[:-2,2:]+2*p[1:-1,2:]+p[2:,2:])/8
+    gy = (-p[:-2,:-2]-2*p[:-2,1:-1]-p[:-2,2:]+p[2:,:-2]+2*p[2:,1:-1]+p[2:,2:])/8
+    return gx, gy
+
+def _boundary(grid):
+    """Color-change boundary (Im side)."""
+    p = np.pad(grid, 1, mode='edge')
+    return ((p[1:-1,1:-1]!=p[:-2,1:-1])|(p[1:-1,1:-1]!=p[2:,1:-1])|
+            (p[1:-1,1:-1]!=p[1:-1,:-2])|(p[1:-1,1:-1]!=p[1:-1,2:])).astype(np.float32)
+
+def _perimeter(grid):
+    """
+    Object perimeter — cells at the edge of nonzero regions.
+    Handles solid blocks (all same color, no color-change boundary).
+    This is the Cauchy contour for the Re side.
+    """
+    H, W = grid.shape
+    p = np.zeros((H,W), dtype=np.float32)
+    mask = grid > 0
+    if not mask.any(): return p
+    padded = np.pad(mask.astype(int), 1, constant_values=0)
+    for dy,dx in [(-1,0),(1,0),(0,-1),(0,1)]:
+        shifted = padded[1+dy:H+1+dy, 1+dx:W+1+dx]
+        p[mask & (shifted==0)] = 1
+    # Solid block: no cell has a zero neighbor — use outer ring
+    if p.sum() == 0:
+        p[0,:]  = mask[0,:] .astype(float)
+        p[-1,:] = mask[-1,:].astype(float)
+        p[:,0]  = mask[:,0] .astype(float)
+        p[:,-1] = mask[:,-1].astype(float)
+    return p
+
+def _cc(mask):
+    labels = np.zeros_like(mask, dtype=np.int32); cur = 0; H,W = mask.shape
+    for r in range(H):
+        for c in range(W):
+            if mask[r,c] and labels[r,c]==0:
+                cur+=1; q=[(r,c)]; labels[r,c]=cur
+                while q:
+                    y,x=q.pop()
+                    for dy,dx in [(-1,0),(1,0),(0,-1),(0,1)]:
+                        ny,nx=y+dy,x+dx
+                        if 0<=ny<H and 0<=nx<W and mask[ny,nx] and labels[ny,nx]==0:
+                            labels[ny,nx]=cur; q.append((ny,nx))
+    return labels
+
+def _h_sym_at(grid, x):
+    r = min(x, grid.shape[1]-1-x)
+    if r <= 0: return 0.0
+    return float((grid[:, x-r:x] == grid[:, x+1:x+r+1][:,::-1]).mean())
+
+def _v_sym_at(grid, y):
+    r = min(y, grid.shape[0]-1-y)
+    if r <= 0: return 0.0
+    return float((grid[y-r:y, :] == grid[y+1:y+r+1, :][::-1, :]).mean())
+
+
+# ── Re/Im board reader ────────────────────────────────────────────────────────
+
+def read_board(grid: np.ndarray):
+    """
+    Read the board as a complex object.
+
+    Returns
+    -------
+    re         : dict — Re-side structure per color
+    im         : dict — Im-side structure (symmetry, boundary, flow, curl)
+    answer     : np.ndarray — derived answer grid (or None)
+    confidence : float 0-1
+    reasoning  : list of strings, one per signal that fired
+    """
+    H, W = grid.shape
+    gx, gy = _sobel(grid.astype(np.float32) / 9)
+    bound  = _boundary(grid)
+
+    # ── Re side ───────────────────────────────────────────────────────────
+    colors = [c for c in range(1, 10) if (grid == c).any()]
+    re = {}
+    for color in colors:
+        mask = (grid == color)
+        ys, xs = np.where(mask)
+        labels = _cc(mask)
+        re[color] = {
+            'count':    int(mask.sum()),
+            'objects':  int(labels.max()),
+            'centroid': (float(ys.mean()), float(xs.mean())),
+            'bbox':     (int(ys.min()),int(xs.min()),int(ys.max()),int(xs.max())),
+        }
+    total_px = int((grid > 0).sum())
+
+    # ── Im side ───────────────────────────────────────────────────────────
+    # Symmetry axes
+    h_scores = [(x, _h_sym_at(grid, x)) for x in range(1, W-1)]
+    v_scores = [(y, _v_sym_at(grid, y)) for y in range(1, H-1)]
+    best_h = max(h_scores, key=lambda x:x[1]) if h_scores else (W//2, 0.0)
+    best_v = max(v_scores, key=lambda x:x[1]) if v_scores else (H//2, 0.0)
+
+    # Boundary (Cauchy contour)
+    b_px    = int(bound.sum())
+    b_ratio = b_px / max(total_px, 1)
+
+    # Gradient field
+    gx_mag = float(np.abs(gx).mean())
+    gy_mag = float(np.abs(gy).mean())
+
+    # Suspended pixels (Re gives count, Im gives direction)
+    suspended = sum(
+        1 for c in range(W)
+        for r in np.where(grid[:, c] > 0)[0]
+        if r < H-1 and grid[r+1, c] == 0
+    )
+
+    # Winding / curl
+    p_gx = np.pad(gx,1,mode='edge'); p_gy = np.pad(gy,1,mode='edge')
+    curl = ((p_gy[1:-1,2:]-p_gy[1:-1,:-2])/2 -
+            (p_gx[2:,1:-1]-p_gx[:-2,1:-1])/2)
+    curl_max = float(np.abs(curl).max())
+
+    im = {
+        'best_h':    best_h,   # (x, score)
+        'best_v':    best_v,   # (y, score)
+        'b_ratio':   b_ratio,
+        'b_px':      b_px,
+        'gx_mag':    gx_mag,
+        'gy_mag':    gy_mag,
+        'suspended': suspended,
+        'curl_max':  curl_max,
+    }
+
+    # ── i· column swap: apply Im → derive answer (Re coords) ──────────────
+    reasoning = []
+    answer    = None
+    confidence = 0.0
+
+    # ── Signal 1: One dominant color fills everything — check boundary first
+    # Re: single color, high density → Im: boundary ratio is the key signal
+    if len(colors) == 1 and total_px > 0:
+        single_color = colors[0]
+        filled_ratio = total_px / (H * W)
+        # Check if grid is already hollow (has interior zeros)
+        interior = (grid == 0) & (_perimeter(grid) == 0)
+        already_hollow = interior.any()
+        if filled_ratio > 0.5 and not already_hollow:
+            # Solid block — Im says extract the perimeter
+            perim = _perimeter(grid)
+            answer = np.zeros_like(grid)
+            answer[perim > 0] = grid[perim > 0]
+            perim_ratio = float(perim.sum()) / max(total_px, 1)
+            confidence = filled_ratio * 0.88  # solid fill is the signal
+            reasoning.append(
+                f"Re: 1 color={single_color}, fill={filled_ratio:.2f}  "
+                f"Im: perimeter={int(perim.sum())}px (ratio={perim_ratio:.2f})  "
+                f"i·: Cauchy perimeter IS the answer")
+
+    # ── Signal 2: Strong H-symmetry + asymmetric Re mass
+    if confidence < 0.75:
+        hx, hs = best_h
+        if hs > 0.65:
+            left_px  = int((grid[:, :hx] > 0).sum())
+            right_px = int((grid[:, hx:] > 0).sum())
+            asymmetry = abs(left_px - right_px) / max(left_px + right_px, 1)
+            if asymmetry > 0.25:
+                answer = grid.copy()
+                if left_px > right_px:
+                    # Mirror left → right (fill empty right side)
+                    for c in range(hx):
+                        mir = W - 1 - c
+                        if 0 <= mir < W:
+                            mask = answer[:, mir] == 0
+                            answer[mask, mir] = grid[mask, c]
+                else:
+                    # Mirror right → left
+                    for c in range(hx+1, W):
+                        mir = W - 1 - c
+                        if 0 <= mir < W:
+                            mask = answer[:, mir] == 0
+                            answer[mask, mir] = grid[mask, c]
+                confidence = hs * asymmetry * 0.95
+                reasoning.append(
+                    f"Im H-sym={hs:.2f} at x={hx}  "
+                    f"Re left={left_px} right={right_px} (asym={asymmetry:.2f})  "
+                    f"i·: complete H mirror")
+
+    # ── Signal 3: Strong V-symmetry + asymmetric top/bottom
+    if confidence < 0.55:
+        vy, vs = best_v
+        if vs > 0.65:
+            top_px = int((grid[:vy, :] > 0).sum())
+            bot_px = int((grid[vy:, :] > 0).sum())
+            asymmetry = abs(top_px - bot_px) / max(top_px + bot_px, 1)
+            if asymmetry > 0.25:
+                answer = grid.copy()
+                if top_px > bot_px:
+                    for r in range(vy):
+                        mir = H - 1 - r
+                        if 0 <= mir < H:
+                            mask = answer[mir, :] == 0
+                            answer[mir, mask] = grid[r, mask]
+                else:
+                    for r in range(vy+1, H):
+                        mir = H - 1 - r
+                        if 0 <= mir < H:
+                            mask = answer[mir, :] == 0
+                            answer[mir, mask] = grid[r, mask]
+                confidence = vs * asymmetry * 0.90
+                reasoning.append(
+                    f"Im V-sym={vs:.2f} at y={vy}  "
+                    f"Re top={top_px} bot={bot_px} (asym={asymmetry:.2f})  "
+                    f"i·: complete V mirror")
+
+    # ── Signal 4: Hollow Re structure + unfilled interior
+    if confidence < 0.5:
+        interior = (grid == 0) & (bound == 0)
+        if interior.any() and total_px > 0:
+            dom = np.argmax(np.bincount(grid[grid>0].flatten(),minlength=10)[1:])+1
+            answer = grid.copy()
+            answer[interior] = dom
+            conf = interior.sum() / max(1, (grid==0).sum()) * 0.80
+            if conf > confidence:
+                confidence = conf
+                reasoning.append(
+                    f"Re: hollow object, dom_color={dom}  "
+                    f"Im: interior={int(interior.sum())}px unfilled  "
+                    f"i·: fill interior — Im interior → Re color")
+
+    # ── Signal 5: Suspended pixels + Im gradient direction → gravity
+    if confidence < 0.45 and suspended > 0:
+        direction = 'down' if gy_mag >= gx_mag else 'right'
+        answer = np.zeros_like(grid)
+        if direction == 'down':
+            for c in range(W):
+                vals = grid[:, c][grid[:, c] > 0]
+                if len(vals): answer[H-len(vals):H, c] = vals
+        else:
+            for r in range(H):
+                vals = grid[r, :][grid[r, :] > 0]
+                if len(vals): answer[r, W-len(vals):W] = vals
+        confidence = min(0.80, suspended / max(total_px, 1) * 2.5)
+        reasoning.append(
+            f"Re suspended={suspended}px  "
+            f"Im gy_mag={gy_mag:.3f} gx_mag={gx_mag:.3f}  "
+            f"i·: gravity {direction}")
+
+    # ── Signal 6: Color shift (Re count ratio between colors suggests increment)
+    if confidence < 0.35 and len(colors) > 0:
+        if all(v < 9 for v in colors):
+            answer = grid.copy()
+            mask = grid > 0
+            answer[mask] = ((grid[mask] - 1 + 1) % 9) + 1
+            confidence = 0.35
+            reasoning.append(
+                f"Re: colors {colors} all < 9  "
+                f"Im: uniform distribution suggests Re→Im shift  "
+                f"i·: increment all colors +1")
+
+    return re, im, answer, confidence, reasoning
+
+
+# ── Re-side: exact cell targeting ────────────────────────────────────────────
+
+def pixel_diff(current: np.ndarray, target: np.ndarray):
+    """All differing cells: [(r, c, target_color), ...]"""
+    if current.shape != target.shape: return []
+    return [(int(r), int(c), int(target[r,c]))
+            for r in range(current.shape[0])
+            for c in range(current.shape[1])
+            if current[r,c] != target[r,c]]
+
+def most_urgent_diff(current: np.ndarray, target: np.ndarray):
+    """
+    Re-side: find the single most important cell to fix.
+    Cauchy boundary-first: the boundary determines the interior,
+    so fix boundary cells before interior cells.
+    """
+    diffs = pixel_diff(current, target)
     if not diffs: return None
-    b=_boundary(cur)
-    bdiffs=[(r,c,v) for r,c,v in diffs if b[r,c]>0]
-    pool=bdiffs if bdiffs else diffs
+    bound = _boundary(current)
+    boundary_diffs = [(r,c,v) for r,c,v in diffs if bound[r,c] > 0]
+    pool = boundary_diffs if boundary_diffs else diffs
     return pool[np.random.randint(len(pool))]
 
+
+# ── Main entry point ─────────────────────────────────────────────────────────
+
+CONFIDENCE_THRESHOLD = 0.38
+
+def try_analytic_action(
+    frame_2d: np.ndarray,
+    available_actions,
+) -> Tuple[Optional[int], Optional[dict], str, float]:
+    """
+    Read the board as a complex object, derive the answer via i· column swap,
+    then use the Re-side diff to find the exact cell to click.
+
+    Returns (action_id, action_data, solver_name, confidence)
+    """
+    if frame_2d is None: return None, None, 'none', 0.0
+
+    avail_ids = set(
+        int(a.value if hasattr(a,'value') else a)
+        for a in (available_actions or range(1,7))
+    )
+
+    # Read the board
+    re, im, answer, confidence, reasoning = read_board(frame_2d)
+
+    if answer is None or confidence < CONFIDENCE_THRESHOLD:
+        return None, None, 'low_confidence', confidence
+
+    # Find the diff (Re side: exact wrong cells)
+    diffs = pixel_diff(frame_2d, answer)
+    if not diffs:
+        return None, None, 'already_matches', confidence
+
+    solver_name = reasoning[0].split('→')[0].strip() if reasoning else 're_im'
+
+    # Emit ACTION6 at the most urgent Re-side coordinate
+    if 6 in avail_ids:
+        cell = most_urgent_diff(frame_2d, answer)
+        if cell is not None:
+            r, c, _ = cell
+            H, W = frame_2d.shape
+            game_y = min(63, max(0, int(r * 64 / H + 32 / H)))
+            game_x = min(63, max(0, int(c * 64 / W + 32 / W)))
+            return 6, {'x': game_x, 'y': game_y}, solver_name, confidence
+
+    # Fallback: button actions
+    BUTTONS = {'h_mirror': 1, 'v_mirror': 2, 'rotate': 3, 'gravity': 4}
+    for key, a_id in BUTTONS.items():
+        if any(key in r for r in reasoning) and a_id in avail_ids:
+            return a_id, None, solver_name, confidence
+
+    return None, None, 'no_action_mapping', confidence
+
+
+
+
 # ── Feature extractor ─────────────────────────────────────────────────────────
 
 def extract_features(grid,num_colours=10):
@@ -337,17 +588,12 @@ class TinyAgent:
         feat=extract_features(grid).to(self.device)
         cur_state=str(state) if state else None
 
-        # ── Im side: rank hypotheses ──────────────────────────────────────
-        candidates=get_candidates(grid)
-        best_name,best_cand,best_conf=(candidates[0] if candidates
-                                        else ('none',grid,0.0))
+        # ── Re/Im: read the board as a complex object ────────────────────
+        re,im,cand_answer,cand_conf,cand_reasoning=read_board(grid)
 
         # Candidate proximity bonus
-        if candidates:
-            nn_name,nn_cand,nn_conf=min(
-                candidates,
-                key=lambda c:(grid!=c[1]).mean() if grid.shape==c[1].shape else 1.0)
-            curr_dist=(grid!=nn_cand).mean() if grid.shape==nn_cand.shape else 1.0
+        if cand_answer is not None and cand_conf>=0.35:
+            curr_dist=(grid!=cand_answer).mean() if grid.shape==cand_answer.shape else 1.0
             if curr_dist==0.0:
                 cand_bonus=self.candidate_win_reward
             elif curr_dist<self.prev_candidate_dist:
@@ -356,7 +602,7 @@ class TinyAgent:
                 cand_bonus=0.0
             self.prev_candidate_dist=curr_dist
         else:
-            cand_bonus=0.0
+            cand_bonus=0.0; cand_answer=None
 
         # Store shaped experience
         if self.prev_feat is not None:
@@ -386,21 +632,23 @@ class TinyAgent:
         if self.step_count%10==0 and len(self.buf)>=16:
             self._train()
 
-        # ── Im → Re bridge: analytic action ──────────────────────────────
+        # ── Im → Re bridge: read board → derive answer → click exact cell ──
         analytic_action=None; analytic_meta={}
-        if best_conf>=CONF_THRESHOLD and candidates:
-            diffs=pixel_diff(grid,best_cand)
+        if cand_answer is not None and cand_conf>=CONF_THRESHOLD:
+            diffs=pixel_diff(grid,cand_answer)
             if diffs:
-                cell=most_urgent_diff(grid,best_cand)
+                cell=most_urgent_diff(grid,cand_answer)
                 if cell is not None:
                     r,c,tgt_color=cell
                     H,W=grid.shape
                     gy=min(63,max(0,int(r*64/H+32/H)))
                     gx=min(63,max(0,int(c*64/W+32/W)))
                     analytic_action=6
+                    reasoning_str=' | '.join(cand_reasoning[:2]) if cand_reasoning else 'read_board'
                     analytic_meta={'x':gx,'y':gy,'cell':(r,c,tgt_color),
-                                   'hypothesis':best_name,'conf':best_conf,
-                                   'n_diffs':len(diffs),'candidates':candidates[:4]}
+                                   'hypothesis':reasoning_str,'conf':cand_conf,
+                                   'n_diffs':len(diffs),
+                                   'candidates':[(reasoning_str,cand_answer,cand_conf)]}
 
         # ── CNN fallback ──────────────────────────────────────────────────
         with torch.no_grad():