Coverage for src / puzzletree / reconstruct / core.py: 97.47%

1from __future__ import annotations (empty)

2 

3from collections.abc import Sequence (empty)

4from typing import Dict, List, Tuple (empty)

5 

6import numpy as np (empty)

7 

8SideEdge = Tuple[int, int] # (side_label, neighbor_index) (empty)

9AdjList = List[List[SideEdge]] (empty)

10Coord = Tuple[int, int] (empty)

11COMPONENT_GAP_TILES = 3 (empty)

12 

13 

14def reverse_side(side: int) -> int: (empty)

15 return {1: 2, 2: 1, 3: 4, 4: 3}[side] (empty)

16 

17 

18def var2(arr: np.ndarray) -> float: (empty)

19 if arr.ndim == 1: 19 ↛ 20line 19 didn't jump to line 20 because the condition on line 19 was never true(empty)

20 return float(np.var(arr)) 

21 return float(np.var(arr, axis=0).sum()) (empty)

22 

23 

24def gaussian_filter1d_nearest(arr: np.ndarray, sigma: float, axis: int = 0) -> np.ndarray: (empty)

25 if sigma <= 0: 25 ↛ 26line 25 didn't jump to line 26 because the condition on line 25 was never true(empty)

26 return arr.astype(np.float64, copy=False) 

27 

28 radius = max(1, int(round(4.0 * sigma))) (empty)

29 offsets = np.arange(-radius, radius + 1, dtype=np.float64) (empty)

30 kernel = np.exp(-(offsets**2) / (2.0 * sigma * sigma)) (empty)

31 kernel /= kernel.sum() (empty)

32 

33 moved = np.moveaxis(arr.astype(np.float64, copy=False), axis, 0) (empty)

34 pad_width = [(radius, radius)] + [(0, 0)] * (moved.ndim - 1) (empty)

35 padded = np.pad(moved, pad_width, mode="edge") (empty)

36 

37 smoothed = np.empty_like(moved, dtype=np.float64) (empty)

38 for idx in range(moved.shape[0]): (empty)

39 window = padded[idx : idx + kernel.size] (empty)

40 smoothed[idx] = np.tensordot(kernel, window, axes=(0, 0)) (empty)

41 

42 return np.moveaxis(smoothed, 0, axis) (empty)

43 

44 

45def corr(edge1: np.ndarray, edge2: np.ndarray, r: float) -> float: (empty)

46 diff = gaussian_filter1d_nearest(edge1 - edge2, sigma=r, axis=0) (empty)

47 summ = gaussian_filter1d_nearest(edge1 + edge2, sigma=r, axis=0) (empty)

48 denom = max(var2(summ), 1e-12) (empty)

49 return var2(diff) / denom (empty)

50 

51 

52def edge_features(tiles: Sequence[np.ndarray]) -> List[Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]]: (empty)

53 # (left, right, top, bottom) 

54 return [(t[:, 0, :], t[:, -1, :], t[0, :, :], t[-1, :, :]) for t in tiles] (empty)

55 

56 

57def build_weight_matrices(tiles: Sequence[np.ndarray], r: float = 12.0) -> Tuple[np.ndarray, np.ndarray]: (empty)

58 n = len(tiles) (empty)

59 feats = edge_features(tiles) (empty)

60 h, w = tiles[0].shape[:2] (empty)

61 lr_sentinel = float(w * w) (empty)

62 ud_sentinel = float(h * h) (empty)

63 

64 lr = np.empty((n, n), dtype=np.float64) (empty)

65 ud = np.empty((n, n), dtype=np.float64) (empty)

66 

67 for i in range(n): (empty)

68 for j in range(n): (empty)

69 if i == j: (empty)

70 lr[i, j] = lr_sentinel (empty)

71 ud[i, j] = ud_sentinel (empty)

72 else: 

73 lr[i, j] = corr(feats[i][1], feats[j][0], r) (empty)

74 ud[i, j] = corr(feats[i][3], feats[j][2], r) (empty)

75 return lr, ud (empty)

76 

77 

78def directed_to_unlabeled(adjs: AdjList) -> List[List[int]]: (empty)

79 return [[nbr for _, nbr in edges] for edges in adjs] (empty)

80 

81 

82def fillin(adj: List[List[int]]) -> List[List[int]]: (empty)

83 out = [neighbors[:] for neighbors in adj] (empty)

84 for node, neighbors in enumerate(adj): (empty)

85 for nbr in neighbors: (empty)

86 if node not in out[nbr]: 86 ↛ 85line 86 didn't jump to line 85 because the condition on line 86 was always true(empty)

87 out[nbr].append(node) (empty)

88 return out (empty)

89 

90 

91def fillin2(adjs: AdjList) -> AdjList: (empty)

92 out: AdjList = [[(side, nbr) for side, nbr in edges] for edges in adjs] (empty)

93 for node, edges in enumerate(adjs): (empty)

94 for side, nbr in edges: (empty)

95 rev = reverse_side(side) (empty)

96 rev_edge = (rev, node) (empty)

97 if rev_edge not in out[nbr]: 97 ↛ 94line 97 didn't jump to line 94 because the condition on line 97 was always true(empty)

98 out[nbr].append(rev_edge) (empty)

99 return out (empty)

100 

101 

102def ita_path(in_adjgraph: AdjList, point1: int, point2: int) -> bool: (empty)

103 adjgraph = fillin(directed_to_unlabeled(in_adjgraph)) (empty)

104 marked = [False] * len(adjgraph) (empty)

105 marked[point1] = True (empty)

106 stack = list(adjgraph[point1]) (empty)

107 

108 if point1 == point2: 108 ↛ 109line 108 didn't jump to line 109 because the condition on line 108 was never true(empty)

109 return True 

110 

111 while stack: (empty)

112 current = stack.pop() (empty)

113 if not marked[current]: (empty)

114 if current == point2: (empty)

115 return True (empty)

116 marked[current] = True (empty)

117 stack.extend(adjgraph[current]) (empty)

118 

119 return False (empty)

120 

121 

122def charged_path(in_adjgraph: AdjList, point1: int, charge: Coord) -> bool: (empty)

123 adjgraph = fillin2(in_adjgraph) (empty)

124 marked = [False] * len(adjgraph) (empty)

125 parent: List[Tuple[Coord, int] | None] = [None] * len(adjgraph) (empty)

126 

127 marked[point1] = True (empty)

128 origin = (0, 0) (empty)

129 if origin == charge: (empty)

130 return True (empty)

131 

132 stack = list(adjgraph[point1]) (empty)

133 for _, nbr in adjgraph[point1]: (empty)

134 parent[nbr] = (origin, point1) (empty)

135 

136 side_to_delta = { (empty)

137 1: (-1, 0), 

138 2: (1, 0), 

139 3: (0, 1), 

140 4: (0, -1), 

141 } 

142 

143 while stack: (empty)

144 side, current = stack.pop() (empty)

145 if marked[current]: (empty)

146 continue (empty)

147 

148 current_parent = parent[current] (empty)

149 base = current_parent[0] if current_parent is not None else origin (empty)

150 add = side_to_delta[side] (empty)

151 current_charge = (base[0] + add[0], base[1] + add[1]) (empty)

152 

153 if current_charge == charge: (empty)

154 return True (empty)

155 

156 marked[current] = True (empty)

157 for edge in adjgraph[current]: (empty)

158 stack.append(edge) (empty)

159 parent[edge[1]] = (current_charge, current) (empty)

160 

161 return False (empty)

162 

163 

164def chargeds(in_adjgraph: AdjList, point1: int) -> List[Tuple[int, Coord]]: (empty)

165 adjgraph = fillin2(in_adjgraph) (empty)

166 marked = [False] * len(adjgraph) (empty)

167 parent: List[Tuple[Coord, int] | None] = [None] * len(adjgraph) (empty)

168 

169 marked[point1] = True (empty)

170 out: List[Tuple[int, Coord]] = [(point1, (0, 0))] (empty)

171 

172 stack = list(adjgraph[point1]) (empty)

173 for _, nbr in adjgraph[point1]: (empty)

174 parent[nbr] = ((0, 0), point1) (empty)

175 

176 side_to_delta = { (empty)

177 1: (-1, 0), 

178 2: (1, 0), 

179 3: (0, 1), 

180 4: (0, -1), 

181 } 

182 

183 while stack: (empty)

184 side, current = stack.pop() (empty)

185 if marked[current]: (empty)

186 continue (empty)

187 

188 current_parent = parent[current] (empty)

189 base = current_parent[0] if current_parent is not None else (0, 0) (empty)

190 add = side_to_delta[side] (empty)

191 current_charge = (base[0] + add[0], base[1] + add[1]) (empty)

192 out.append((current, current_charge)) (empty)

193 

194 marked[current] = True (empty)

195 for edge in adjgraph[current]: (empty)

196 stack.append(edge) (empty)

197 parent[edge[1]] = (current_charge, current) (empty)

198 

199 return out (empty)

200 

201 

202def edge_label_exists(edges: List[SideEdge], label: int) -> bool: (empty)

203 return any(side == label for side, _ in edges) (empty)

204 

205 

206def global_edge_exists(adjs: AdjList, label: int, node: int) -> bool: (empty)

207 return any((side == label and nbr == node) for edges in adjs for side, nbr in edges) (empty)

208 

209 

210def clone_adjs(adjs: AdjList) -> AdjList: (empty)

211 return [[(side, nbr) for side, nbr in edges] for edges in adjs] (empty)

212 

213 

214def msgt( (empty)

215 leftright: np.ndarray, 

216 updown: np.ndarray, 

217 minset: float = 50.0, 

218 lr_side_size: int = 180, 

219 ud_side_size: int = 72, 

220 record_history: bool = False, 

221) -> AdjList | Tuple[AdjList, List[AdjList]]: 

222 n = leftright.shape[0] (empty)

223 lr = leftright.copy() (empty)

224 ud = updown.copy() (empty)

225 adjs: AdjList = [[] for _ in range(n)] (empty)

226 history: List[AdjList] | None = [clone_adjs(adjs)] if record_history else None (empty)

227 

228 k = n - 1 (empty)

229 while k > 0: (empty)

230 minlr = float(lr.min()) (empty)

231 minud = float(ud.min()) (empty)

232 

233 if minlr < minud: (empty)

234 i, j = np.unravel_index(int(np.argmin(lr)), lr.shape) (empty)

235 lr[i, j] = float(lr_side_size * lr_side_size) (empty)

236 directions = (-1, 0) (empty)

237 labels = [1, 2] (empty)

238 else: 

239 i, j = np.unravel_index(int(np.argmin(ud)), ud.shape) (empty)

240 ud[i, j] = float(ud_side_size * ud_side_size) (empty)

241 directions = (0, 1) (empty)

242 labels = [3, 4] (empty)

243 

244 location1, location2 = int(i), int(j) (empty)

245 

246 if location1 < location2: (empty)

247 location1, location2 = location2, location1 (empty)

248 labels = [labels[1], labels[0]] (empty)

249 directions = (-directions[0], -directions[1]) (empty)

250 

251 if minlr > minset and minud > minset: (empty)

252 break (empty)

253 

254 if edge_label_exists(adjs[location1], labels[-1]): (empty)

255 continue (empty)

256 if global_edge_exists(adjs, labels[-1], location2): (empty)

257 continue (empty)

258 if global_edge_exists(adjs, labels[0], location1): (empty)

259 continue (empty)

260 if edge_label_exists(adjs[location2], labels[0]): (empty)

261 continue (empty)

262 

263 if ita_path(adjs, location1, location2): (empty)

264 continue (empty)

265 

266 charges_l2 = [coord for _, coord in chargeds(adjs, location2)] (empty)

267 overlap_l1 = any( (empty)

268 charged_path(adjs, location1, (coord[0] - directions[0], coord[1] - directions[1])) for coord in charges_l2 

269 ) 

270 

271 charges_l1 = [coord for _, coord in chargeds(adjs, location1)] (empty)

272 overlap_l2 = any( (empty)

273 charged_path(adjs, location2, (coord[0] + directions[0], coord[1] + directions[1])) for coord in charges_l1 

274 ) 

275 

276 if not overlap_l1 and not overlap_l2: (empty)

277 adjs[location1].append((labels[-1], location2)) (empty)

278 if history is not None: (empty)

279 history.append(clone_adjs(adjs)) (empty)

280 k -= 1 (empty)

281 

282 if history is not None: (empty)

283 return adjs, history (empty)

284 return adjs (empty)

285 

286 

287def connected_components(adjs: AdjList) -> List[List[int]]: (empty)

288 undirected = fillin(directed_to_unlabeled(adjs)) (empty)

289 seen = [False] * len(adjs) (empty)

290 components: List[List[int]] = [] (empty)

291 

292 for start in range(len(adjs)): (empty)

293 if seen[start]: (empty)

294 continue (empty)

295 stack = [start] (empty)

296 seen[start] = True (empty)

297 comp = [] (empty)

298 while stack: (empty)

299 node = stack.pop() (empty)

300 comp.append(node) (empty)

301 for nbr in undirected[node]: (empty)

302 if not seen[nbr]: (empty)

303 seen[nbr] = True (empty)

304 stack.append(nbr) (empty)

305 components.append(comp) (empty)

306 

307 return components (empty)

308 

309 

310def reconstruct_layout(adjs: AdjList, component_gap: int = COMPONENT_GAP_TILES) -> Dict[int, Coord]: (empty)

311 components = connected_components(adjs) (empty)

312 components.sort(key=len, reverse=True) (empty)

313 

314 placements: Dict[int, Coord] = {} (empty)

315 x_offset = 0 (empty)

316 

317 for comp in components: (empty)

318 root = comp[0] (empty)

319 comp_coords = {node: coord for node, coord in chargeds(adjs, root)} (empty)

320 xs = [coord[0] for coord in comp_coords.values()] (empty)

321 ys = [coord[1] for coord in comp_coords.values()] (empty)

322 minx, maxx = min(xs), max(xs) (empty)

323 miny = min(ys) (empty)

324 

325 for node, (x, y) in comp_coords.items(): (empty)

326 placements[node] = (x - minx + x_offset, y - miny) (empty)

327 

328 width = maxx - minx + 1 (empty)

329 x_offset += width + component_gap (empty)

330 

331 return placements (empty)