path solver working, but small issues

main
Harald Holtmann 2025-09-06 15:27:25 +02:00
parent 84b8b5cf23
commit 341cd92a33
1 changed files with 140 additions and 57 deletions

@ -44,6 +44,9 @@ class Path:
def __len__(self):
return len(self.path)
def __getitem__(self, idx):
return Path(self.path.__getitem__(idx))
def last(self):
return Path(self.path[:-1]), int(self.path[-1])
@ -66,12 +69,21 @@ class Explore:
self.neighbors = {}
self.unification_id = {}
self.unifications = defaultdict(lambda: set())
self.score = 0
def _path(self, path):
return self.unification_id.get(path, path)
while True:
repeat = False
for i in range(1, len(path) + 1):
if path[:i] in self.unification_id:
path = self.unification_id[path[:i]] + path[i:]
repeat = True
break
if not repeat:
break
return path
def save(self):
return {
@ -80,7 +92,6 @@ class Explore:
"room_ids": dict(self.room_ids),
"neighbours": dict(self.neighbors),
"unification_id": dict(self.unification_id),
"unifications": dict(self.unifications),
}
@classmethod
@ -91,15 +102,13 @@ class Explore:
new.room_ids.update(obj["room_ids"])
new.neighbors.update(obj["neighbours"])
new.unification_id.update(obj["unification_id"])
new.unifications.update(obj["unifications"])
return new
def explore(self, path=Path()):
path = self._path(path)
if path in self.rooms:
return self.rooms[path]
def explore(self, path=Path(), probes=None):
probes = probes or self.probes
paths = [path + probe for probe in self.probes]
path = self._path(path)
paths = [path + probe for probe in probes]
print("explore", paths)
response = api.explore([p.path for p in paths])
@ -107,37 +116,70 @@ class Explore:
self.score = response["queryCount"]
print("id", path, results)
self.update(path, [res[len(path) :] for res in results])
return [res[len(path) :] for res in results]
def update(self, path, results):
"""
path: path to update
probe: prob result from path.
"""
def _add_room(self, path, results):
label = results[0][0]
probe_ids = [r for res in (rs[1:] for rs in results) for r in res]
print(path, results, probe_ids)
room_id = str(label) + "".join(str(p) for p in probe_ids)
# print("add room", path, results, room_id)
if path in self.rooms:
rid = self.rooms[path]
assert rid == room_id, f"expected match room at {path}: {rid} != {room_id}"
else:
self.rooms[path] = room_id
self.room_ids[room_id].add(path)
self.neighbors[path] = [(path + d, None) for d in DOORS]
return room_id
def update(self, path, results):
"""
path: path to update
probe: prob result from path.
"""
path = self._path(path)
print("update", path, results)
room_id = self._add_room(path, results)
# update penultimate room
if path:
pl, dl = path.last()
p, _ = self.neighbors[pl][dl]
self.neighbors[pl][dl] = (p, room_id)
pl, d = path.last()
pl = self._path(pl)
if pl in self.rooms:
print("penult", path, pl, d, pl in self.rooms)
self.dump()
p, rid = self.neighbors[pl][d]
assert rid is None or rid == room_id, f"penultimate {path} {pl}: {rid} != {room_id}"
self.neighbors[pl][d] = (p, room_id)
self.dump()
def update_path(self, path, door, result0, result1):
path = self._path(path)
room0 = path
room1 = path + Path([door])
room_id0 = self._add_room(room0, [result0[:1], result0]) # noqa
room_id1 = self._add_room(room1, [result1[1:2], result1[1:]])
p, rid = self.neighbors[room0][door]
assert rid is None or rid == room_id1
self.neighbors[room0][door] = (p, room_id1)
def dump(self):
print(self.problem + " rooms:")
print(f"{self.problem} rooms: {len(self.rooms)}")
for r, rid in self.rooms.items():
print(f" {r:<10}: {rid} {self.neighbors[r]}")
print()
for rid, paths in self.room_ids.items():
print(f" {rid}: {paths}")
print()
# for p, pu in self.unification_id.items():
# print(f" {p}: {pu}")
# print()
def unify(self, path1, path2):
"""
@ -147,6 +189,7 @@ class Explore:
path1 = self._path(path1)
path2 = self._path(path2)
print("unify", path1, path2)
if path1 == path2:
return
@ -161,54 +204,65 @@ class Explore:
path = min(path1, path2)
pmerge = max(path1, path2)
room_id = self.rooms[path]
# unify paths
self.unification_id[pmerge] = path
merged_neighbors = []
for n, ((p, rid), (pm, rmid)) in enumerate(zip(self.neighbors[path], self.neighbors[pmerge])):
if rid and rmid and rid != rmid:
raise ExploreError(f"neighbor {n} of '{path}'({rid}) and '{pmerge}'({rmid}) do not match")
merged_neighbors.append((p.shorten(path, pmerge), rid or rmid))
# unify paths
self.unification_id[pmerge] = path
for p in self.unifications[pmerge]:
self.unification_id[p] = path
self.unifications[path] = self.unifications[path1] | self.unifications[path2] | {pmerge}
if pmerge in self.unifications:
del self.unifications[pmerge]
merged_neighbors.append((self._path(p), rid or rmid))
# fix rooms
del self.rooms[pmerge]
del self.neighbors[pmerge]
for p in list(self.rooms.keys()):
pm = self._path(p)
if p != pm:
rid = self.rooms[p]
del self.rooms[p]
self.rooms[pm] = rid
ns = self.neighbors[p]
del self.neighbors[p]
self.neighbors[pm] = ns
self.room_ids[room_id] = {p for p in self.room_ids[room_id] if p != pmerge}
for i, ps in self.room_ids.items():
self.room_ids[i] = {self._path(p) for p in ps if p != pmerge}
self.neighbors[path] = merged_neighbors
del self.neighbors[pmerge]
for p, ns in self.neighbors.items():
new = []
for np, rid in ns:
np = np.shorten(path, pmerge)
new.append((np, rid))
np_ = self._path(np)
new.append((np_, rid))
if rid:
assert np in self.rooms, f"path {np} of {(p, ns)} not in rooms"
try:
assert np_ in self.rooms, f"unify: path {np} {np_} of {(p, ns)} not in rooms"
except AssertionError as exc:
self.dump()
raise exc
self.neighbors[p] = new
def unify_all(self):
while True:
repeat = False
for rid, paths in self.room_ids.items():
if len(paths) > 1:
paths = list(paths)
print("unify", paths[0], paths[1])
# print("unify", paths[0], paths[1])
self.unify(paths[0], paths[1])
repeat = True
break
if not repeat:
break
def unexplored(self):
for path, ns in self.neighbors.items():
for p, rid in ns:
for d, (p, rid) in enumerate(ns):
if not rid:
yield p
yield d, path
def is_explored(self):
return next(self.unexplored(), None) is None and all(len(p) == 1 for p in self.room_ids.values())
@ -250,7 +304,9 @@ class Explore:
def room_solve(problem):
ex = Explore(problem, DOORS)
api.select(ex.problem)
ex.explore()
res = ex.explore(Path())
ex.update(Path(), res)
ex.dump()
while True:
@ -259,7 +315,8 @@ def room_solve(problem):
break
print("explore", unexplored)
ex.explore(unexplored)
res = ex.explore(unexplored)
ex.update(unexplored, res)
ex.dump()
print("unify")
@ -272,26 +329,52 @@ def room_solve(problem):
print("score", ex.score)
def path_solve(problem, plen):
probe = [Path([random.randrange(6) for _ in range(plen)])]
def path_solve(problem, nrooms, plen):
maxlen = 18 * nrooms
probe = Path([random.randrange(6) for _ in range(plen)])
print("probe path", probe)
ex = Explore(problem, probe)
api.select(ex.problem)
ex.explore()
ex.dump()
while True:
unexplored = next(ex.unexplored(), None)
if not unexplored:
break
def extendpath(path, maxlen):
while len(path) < maxlen - plen - 1:
path = path + Path([random.randrange(6)]) + probe
return path
print("explore", unexplored)
ex.explore(unexplored)
def target(door, path):
print("target", door, path)
probe0 = extendpath(probe, maxlen - len(path))
probe1 = extendpath(Path([door]) + probe, maxlen - len(path))
res0, res1 = ex.explore(path, [probe0, probe1])
ex.update_path(path, door, res0[: plen + 1], res1[: plen + 2])
ex.dump()
ex.unify_all()
ex.dump()
print("unify")
i = plen + 1
while i + plen < len(res0):
ex.update(path + probe0[:i], [res0[i : i + 1], res0[i : i + plen + 1]])
ex.unify_all()
i += plen + 1
i = plen + 2
while i + plen < len(res1):
ex.update(path + probe1[:i], [res1[i : i + 1], res1[i : i + plen + 1]])
ex.unify_all()
i += plen + 1
target(0, Path())
ex.dump()
while True:
door, unexplored = next(ex.unexplored(), (None, None))
print("unexplored", door, unexplored)
if unexplored is None:
break
target(door, unexplored)
ex.dump()
print("explored", ex.is_explored())
@ -313,4 +396,4 @@ if __name__ == "__main__":
if problem not in problems:
raise ExploreError(f"unknown problem {problem}")
path_solve(problem, int(sys.argv[2]))
path_solve(problem, problems[problem]["size"], int(sys.argv[2]))