"""Independent image-only inverse arm for shared chamber incidence. The implementation segments only final RGB pixels. It finds the two colored gate masks, labels side-connected near-white components at a conservative four-pixel scale, discards the exterior component, and compares the enclosed component labels adjacent to the long sides of each gate. """ from __future__ import annotations from collections import deque from functools import lru_cache import numpy as np from PIL import Image SCALE = 4 def _component_labels(white: np.ndarray) -> tuple[np.ndarray, set[int]]: height, width = white.shape labels = np.full((height, width), -1, dtype=np.int32) exterior: set[int] = set() next_label = 0 for y in range(height): for x in range(width): if not white[y, x] or labels[y, x] >= 0: continue queue = deque([(y, x)]) labels[y, x] = next_label touches_border = False while queue: cy, cx = queue.popleft() if cy in {0, height - 1} or cx in {0, width - 1}: touches_border = True for ny, nx in ((cy - 1, cx), (cy + 1, cx), (cy, cx - 1), (cy, cx + 1)): if 0 <= ny < height and 0 <= nx < width and white[ny, nx] and labels[ny, nx] < 0: labels[ny, nx] = next_label queue.append((ny, nx)) if touches_border: exterior.add(next_label) next_label += 1 return labels, exterior def _gate_bbox(mask: np.ndarray, name: str) -> tuple[int, int, int, int]: ys, xs = np.nonzero(mask) if len(xs) < 120: raise ValueError(f"abstain: {name} gate is missing or too small") x0, x1 = int(xs.min()), int(xs.max()) y0, y1 = int(ys.min()), int(ys.max()) if max(x1 - x0, y1 - y0) < 30 or min(x1 - x0, y1 - y0) < 10: raise ValueError(f"abstain: {name} gate geometry is ambiguous") return x0, y0, x1, y1 def _adjacent_labels( bbox: tuple[int, int, int, int], labels: np.ndarray, exterior: set[int] ) -> set[int]: x0, y0, x1, y1 = bbox height, width = labels.shape found: set[int] = set() if (x1 - x0) > (y1 - y0): tangent_values = range(x0 + 8, x1 - 7, 4) normal_values = [y0 - offset for offset in range(3, 18, 3)] + [ y1 + offset for offset in range(3, 18, 3) ] samples = ((normal // SCALE, tangent // SCALE) for normal in normal_values for tangent in tangent_values) else: tangent_values = range(y0 + 8, y1 - 7, 4) normal_values = [x0 - offset for offset in range(3, 18, 3)] + [ x1 + offset for offset in range(3, 18, 3) ] samples = ((tangent // SCALE, normal // SCALE) for normal in normal_values for tangent in tangent_values) for sy, sx in samples: if 0 <= sy < height and 0 <= sx < width: label = int(labels[sy, sx]) if label >= 0 and label not in exterior: found.add(label) if not found: raise ValueError("abstain: a colored gate borders no resolved enclosed chamber") return found @lru_cache(maxsize=96) def _decision_from_bytes(width: int, height: int, payload: bytes) -> str: if width < 128 or height < 128 or width % SCALE or height % SCALE: raise ValueError("abstain: unsupported raster dimensions") array = np.frombuffer(payload, dtype=np.uint8).reshape((height, width, 3)) red = array[:, :, 0].astype(np.int16) green_channel = array[:, :, 1].astype(np.int16) blue = array[:, :, 2].astype(np.int16) green_mask = (green_channel > 110) & (green_channel > red + 45) & (green_channel > blue + 18) purple_mask = (blue > 130) & (red > 90) & (blue > green_channel + 40) & (red > green_channel + 20) green_box = _gate_bbox(green_mask, "green") purple_box = _gate_bbox(purple_mask, "purple") white = (array[:, :, 0] > 225) & (array[:, :, 1] > 225) & (array[:, :, 2] > 225) small_white = white.reshape(height // SCALE, SCALE, width // SCALE, SCALE).mean(axis=(1, 3)) > 0.82 labels, exterior = _component_labels(small_white) green_labels = _adjacent_labels(green_box, labels, exterior) purple_labels = _adjacent_labels(purple_box, labels, exterior) return "yes" if green_labels & purple_labels else "no" def decision_from_image(image: Image.Image) -> str: rgb = image.convert("RGB") return _decision_from_bytes(rgb.width, rgb.height, rgb.tobytes())