# MIT License # # Copyright (c) 2026 Pyxel 100 contributors # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in all # copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. # # The bundled Noto-derived bitmap glyphs are separately licensed under SIL OFL. # See assets/NOTO-LICENSE.txt. """引っ越しカニ合戦・独立教材 5/5。日本語の詳しい説明はLESSON.md。""" from dataclasses import dataclass, replace, asdict from collections import deque import hashlib, json from functools import lru_cache DIRS = ((0, -1), (0, 1), (-1, 0), (1, 0)) TIDE = 450 SMALL_INTERVAL = 15 BARE_INTERVAL = 8 @dataclass(frozen=True) class Level: name: str food: tuple water: tuple starts: tuple shells: tuple LEVELS = (Level('今の殻で届く', ((1, 3), (2, 2), (2, 4), (5, 3), (6, 2), (7, 3)), (), ((2, 3), (6, 3)), ((0, 1, (2, 3), 0, 0), (1, 1, (6, 3), 1, 1), (2, 2, (4, 1), -1, -1))), Level('大きい殻の届く湾', ((2, 2), (3, 1), (4, 2), (5, 1), (6, 2), (6, 4), (7, 4)), ((3, 4), (4, 4), (5, 4)), ((2, 3), (7, 3)), ((0, 1, (2, 3), 0, 0), (1, 1, (7, 3), 1, 1), (2, 2, (4, 2), -1, -1))), Level('空けた殻も陣地', ((1, 2), (2, 2), (3, 2), (5, 4), (6, 4), (7, 4), (4, 3)), ((4, 2), (4, 4)), ((2, 2), (6, 4)), ((0, 1, (2, 2), 0, 0), (1, 2, (6, 4), 1, 1), (2, 1, (4, 1), -1, -1)))) @dataclass(frozen=True) class Shell: id: int size: int pos: tuple owner: int occupant: int @dataclass(frozen=True) class Observation: positions: tuple shells: tuple food: tuple water: tuple @dataclass(frozen=True) class Memory: target: tuple | None = None goal_shell: int = -1 mode: str = 'stay' @dataclass(frozen=True) class State: positions: tuple shells: tuple ready: tuple = (0, 0) tick: int = 0 scores: tuple = (0, 0) tides: int = 0 memory: Memory = Memory() ended: bool = False event: str = 'ready' @dataclass(frozen=True) class Command: direction: tuple | None = None toggle: bool = False def initial(level): return State(level.starts, tuple((Shell(*s) for s in level.shells))) @lru_cache(maxsize=4096) def cells(level): return tuple(((x, y) for y in range(1, 6) for x in range(1, 8) if (x, y) not in level.water)) @lru_cache(maxsize=4096) def distances(level, start): out = {start: 0} q = deque([start]) valid = set(cells(level)) while q: p = q.popleft() for d in DIRS: n = (p[0] + d[0], p[1] + d[1]) if n in valid and n not in out: out[n] = out[p] + 1 q.append(n) return out def owned_shell(state, actor): return next((s for s in state.shells if s.occupant == actor), None) def observe(level, state): return Observation(state.positions, state.shells, level.food, level.water) @lru_cache(maxsize=4096) def coverage(level, pos, size): return sum((distances(level, pos).get(f, 999) <= size for f in level.food)) @lru_cache(maxsize=4096) def best_coverage(level, size): return max((coverage(level, p, size) for p in cells(level))) def policy(obs, memory): level = Level('', obs.food, obs.water, (), ()) pos = obs.positions[1] dist = distances(level, pos) inside = next((s for s in obs.shells if s.occupant == 1), None) empty = [s for s in obs.shells if s.occupant < 0] if inside: upgrades = [s for s in empty if best_coverage(level, s.size) >= best_coverage(level, inside.size) + 2 and dist.get(s.pos, 999) <= 4] if upgrades: s = max(upgrades, key=lambda s: (best_coverage(level, s.size) * 2 - dist.get(s.pos, 999), -dist.get(s.pos, 999), -s.id)) return Memory(s.pos, s.id, 'leave') available = [p for p in cells(level) if not any((sh.pos == p and sh.id != inside.id for sh in obs.shells))] target = max(available, key=lambda p: (coverage(level, p, inside.size) * 3 - dist[p], -dist[p], -p[1], -p[0])) return Memory(target, -1, 'place') if not empty: return Memory(pos, -1, 'stay') s = max(empty, key=lambda s: (best_coverage(level, s.size) * 2 - dist.get(s.pos, 999), -dist.get(s.pos, 999), -s.id)) return Memory(s.pos, s.id, 'enter') def next_step(level, start, target, blocked, forbidden=()): if target is None or start == target: return None valid = set(cells(level)) q = deque([(start, None)]) seen = {start} while q: p, first = q.popleft() for d in DIRS: n = (p[0] + d[0], p[1] + d[1]) if n not in valid or n in seen or n == blocked or (n in forbidden): continue f = first or n if n == target: return f seen.add(n) q.append((n, f)) return None def points(level, state): totals = [0, 0] owners = [] for food in level.food: influence = [] for sh in state.shells: if sh.owner < 0: continue d = distances(level, sh.pos).get(food, 999) if d <= sh.size: influence.append((d, sh.owner)) who = -1 if influence: near = min((d for d, o in influence)) c = {o for d, o in influence if d == near} if len(c) == 1: who = c.pop() totals[who] += 1 owners.append(who) return (tuple(totals), tuple(owners)) def simultaneous_entries(candidates): return tuple((v if v >= 0 and candidates.count(v) == 1 else -1 for v in candidates)) def step(level, state, command=Command(), large_interval=24): if state.ended: return state tick = state.tick mem = policy(observe(level, state), state.memory) if tick % 15 == 0 else state.memory old = state.positions inside = [owned_shell(state, i) for i in (0, 1)] dest = list(old) toggle = [command.toggle, False] if mem.mode == 'leave' and tick % 15 == 0 and inside[1]: toggle[1] = True elif mem.mode == 'enter' and inside[1] is None and (mem.target == old[1]): toggle[1] = True if not toggle[0] and command.direction in DIRS and (tick >= state.ready[0]): p = (old[0][0] + command.direction[0], old[0][1] + command.direction[1]) if p in cells(level) and p != old[1] and (not (inside[0] and any((sh.pos == p and sh.id != inside[0].id for sh in state.shells)))): dest[0] = p if not toggle[1] and tick >= state.ready[1]: p = next_step(level, old[1], mem.target, old[0], tuple((sh.pos for sh in state.shells if inside[1] and sh.id != inside[1].id))) if p: dest[1] = p event = '' if dest[0] == dest[1]: dest = list(old) event = 'blocked' shells = list(state.shells) ready = list(state.ready) for actor in (0, 1): if dest[actor] != old[actor]: sh = inside[actor] interval = BARE_INTERVAL if sh is None else SMALL_INTERVAL if sh.size == 1 else large_interval ready[actor] = tick + interval if sh: shells[sh.id] = replace(sh, pos=dest[actor]) requests = [-1, -1] for actor in (0, 1): if not toggle[actor]: continue sh = inside[actor] if sh: shells[sh.id] = replace(shells[sh.id], occupant=-1) event = 'left' else: requests[actor] = next((s.id for s in shells if s.pos == dest[actor] and s.occupant < 0), -1) for actor, sid in enumerate(simultaneous_entries(tuple(requests))): if sid >= 0: shells[sid] = replace(shells[sid], owner=actor, occupant=actor) event = 'entered' s = replace(state, positions=tuple(dest), shells=tuple(shells), ready=tuple(ready), tick=tick + 1, memory=mem, event=event) if s.tick % TIDE == 0: pts, _ = points(level, s) s = replace(s, scores=tuple((s.scores[i] + pts[i] for i in (0, 1))), tides=s.tides + 1, event='tide') if s.tides == 3: s = replace(s, ended=True, event='won' if s.scores[0] > s.scores[1] else 'draw' if s.scores[0] == s.scores[1] else 'lost') validate(level, s) return s def validate(level, s): assert s.positions[0] != s.positions[1] assert all((p in cells(level) for p in s.positions)) assert len({sh.pos for sh in s.shells}) == 3 assert len(s.shells) == 3 and {sh.id for sh in s.shells} == {0, 1, 2} for sh in s.shells: assert sh.pos in cells(level) and sh.owner in (-1, 0, 1) and (sh.occupant in (-1, 0, 1)) if sh.occupant >= 0: assert sh.owner == sh.occupant and sh.pos == s.positions[sh.occupant] assert all((sum((sh.occupant == i for sh in s.shells)) <= 1 for i in (0, 1))) assert 0 <= s.tides <= 3 and s.ended == (s.tides == 3) assert sum(s.scores) <= len(level.food) * s.tides def state_hash(s): return hashlib.sha256(json.dumps(asdict(s), sort_keys=True).encode('utf-8')).hexdigest()[:16] def run(level, actions, large_interval=24): s = initial(level) log = [] for frame in range(TIDE * 3): cmd = actions.get(frame, Command()) old = s s = step(level, s, cmd, large_interval) if cmd != Command() or s.event or s.positions != old.positions: log.append({'tick': s.tick, 'command': asdict(cmd), 'hash': state_hash(s), 'scores': s.scores, 'positions': s.positions, 'event': s.event}) return (s, log) STAGE = 5 LABEL = 'THREE TIDES / FINAL RULES' # The model above never imports Pyxel. Only main opens a window. WIDTH, HEIGHT = 640, 464 LARGE_INTERVAL = 24 DIRECTIONS = {'UP': (0, -1), 'DOWN': (0, 1), 'LEFT': (-1, 0), 'RIGHT': (1, 0)} class App: def __init__(self): self.level_index = 1 if STAGE == 4 else 0 self.lock = False self.reset() @property def level(self): return LEVELS[self.level_index] def reset(self): self.state = initial(self.level) self.pause = False self.reason = 'ready' def control(self, pressed, held): # Original priority: release lock -> reset -> pause -> level -> play. if self.lock: if not held and not pressed: self.lock = False return if 'R' in pressed: self.reset() self.lock = True return if 'P' in pressed: self.pause = not self.pause self.lock = True return if self.pause: return if STAGE == 5: for index, key in enumerate(('1', '2', '3')): if key in pressed: self.level_index = index self.reset() self.lock = True return if getattr(self.state, 'ended', False): return directions = [d for key, d in DIRECTIONS.items() if key in held] direction = directions[0] if len(directions) == 1 else None toggle = STAGE >= 2 and 'Z' in pressed before = self.state self.state = step(self.level, before, Command(direction, toggle), LARGE_INTERVAL) if self.state.event: self.reason = self.state.event def draw(self, p): s = self.state p.cls(1) p.text(12, 10, 'MOVING SHELLS / STAGE '+str(STAGE)+' / BEACH '+'ABC'[self.level_index], 10) p.text(12, 27, LABEL, 7) owners = points(self.level, s)[1] if STAGE >= 3 else (-1,)*len(self.level.food) for y in range(7): for x in range(9): pos = (x, y) px, py = 12+x*40, 52+y*40 p.rect(px, py, 39, 39, 4 if pos in cells(self.level) else 5) if pos in self.level.food: own = owners[self.level.food.index(pos)] p.circ(px+19, py+19, 6, 7) p.circb(px+19, py+19, 10, (6, 8)[own] if own >= 0 else 13) p.text(px+17, py+17, 'PE'[own] if own >= 0 else '-', 1) for sh in s.shells: x, y = 12+sh.pos[0]*40+19, 52+sh.pos[1]*40+19 radius = 15 if sh.size == 2 else 12 color = (6, 8)[sh.owner] if sh.owner >= 0 else 13 p.circ(x, y, radius, color) p.circb(x, y, radius, 7) p.text(x-8, y-7, str(sh.id)+':'+str(sh.size), 1) if sh.occupant < 0: p.text(x-9, y+4, 'EMPTY', 7) for actor, pos in enumerate(s.positions): x, y = 12+pos[0]*40+19, 52+pos[1]*40+19 p.circ(x-10, y-11, 3, (6, 8)[actor]) p.circ(x+10, y-11, 3, (6, 8)[actor]) p.text(x-2, y+3, 'PE'[actor], 7) if STAGE >= 4 and s.memory.target: x, y = s.memory.target p.rectb(12+x*40+1, 52+y*40+1, 37, 37, 8) p.text(388, 58, 'P: BLUE / YOU', 6) p.text(388, 72, 'E: RED / RIVAL', 8) p.text(388, 94, 'TICK '+str(s.tick), 7) p.text(388, 108, '30 TICKS = 1 SECOND', 13) p.text(388, 130, 'ID SIZE OWNER OCCUPANT', 7) for row, sh in enumerate(s.shells): def who(actor): return 'PE'[actor] if actor >= 0 else '-' p.text(388, 146+row*17, f'{sh.id} {sh.size} {who(sh.owner)} {who(sh.occupant)}', (6, 8)[sh.owner] if sh.owner >= 0 else 13) inside = owned_shell(s, 0) interval = BARE_INTERVAL if inside is None else SMALL_INTERVAL if inside.size == 1 else LARGE_INTERVAL p.text(388, 209, 'MOVE INTERVAL '+str(interval), 7) p.text(388, 225, 'READY IN '+str(max(0, s.ready[0]-s.tick)), 7) if STAGE >= 3: totals, _ = points(self.level, s) p.text(388, 250, f'IF TIDE NOW: {totals[0]} : {totals[1]}', 10) else: p.text(388, 250, 'FOOD: NOT SCORED YET', 13) if STAGE >= 4: p.text(388, 272, 'RIVAL: '+s.memory.mode.upper(), 8) else: p.text(388, 272, 'RIVAL: FIXED FOR PRACTICE', 13) if STAGE == 5: p.text(12, 345, f'TIDES {s.tides}/3 TOTAL {s.scores[0]} : {s.scores[1]}', 10) remain = TIDE-s.tick % TIDE if not s.ended else 0 p.text(12, 361, f'NEXT TIDE IN {remain} TICKS / RESULT '+(s.event.upper() if s.ended else 'PLAY'), 7) else: p.text(12, 345, 'PRACTICE: NO TIDES, SCORE TOTALS OR ENDING', 13) p.text(12, 361, '01 CARRY > 02 ENTER/LEAVE > 03 INFLUENCE > 04 RIVAL > 05 TIDES', 13) p.text(12, 388, ('PAUSED / RELEASE KEYS THEN P' if self.pause else 'EVENT: '+self.reason.upper()), 10) p.text(12, 407, 'ARROWS: HOLD TO MOVE / '+('Z: ENTER OR LEAVE' if STAGE >= 2 else 'Z: NOT ADDED YET'), 7) p.text(12, 424, 'P: PAUSE / R: LIVE RESET / ESC: QUIT'+(' / 1-3: BEACH' if STAGE == 5 else ''), 7) p.text(12, 443, 'AFTER CODE EDIT: SAVE > QUIT > RUN AGAIN (R DOES NOT RELOAD)', 13) def demo_keys(frame): # Finite synthetic key sequence, through exactly the same controller. if STAGE == 4: return set(), set() if STAGE == 1: return set(), {'RIGHT'} # Leave small shell; move north twice then east twice; enter large shell. if frame in (0, 34): return {'Z'}, {'Z'} if 1 <= frame <= 9: return set(), {'UP'} if 10 <= frame <= 25: return set(), {'RIGHT'} if STAGE >= 3 and 35 <= frame <= 74: return set(), {'DOWN'} return set(), set() def main(headless=False, screenshot=None, demo=False, report=None): # Resolve validated paths before importing Pyxel: import can change cwd. import pyxel as p p.init(WIDTH, HEIGHT, title='Moving shells lesson '+str(STAGE), fps=30, display_scale=2, headless=headless) app = App() frame = 0 limit = (46 if STAGE == 1 else 36 if STAGE == 2 else 90 if STAGE == 3 else 180 if STAGE == 4 else TIDE*3) if demo else 2 keys = {k: getattr(p, 'KEY_'+k) for k in ('UP', 'DOWN', 'LEFT', 'RIGHT', 'Z', 'P', 'R', '1', '2', '3')} def update(): nonlocal frame if p.btnp(p.KEY_ESCAPE) or headless and frame >= limit: if screenshot: app.draw(p) p.screen.save(screenshot, 1) if report: from pathlib import Path Path(report).write_text(json.dumps({'stage': STAGE, 'synthetic': True, 'state': asdict(app.state), 'hash': state_hash(app.state)}, ensure_ascii=False, indent=2)+'\n', encoding='utf-8') p.quit() return if headless: if demo: pressed, held = demo_keys(frame) app.control(pressed, held) else: app.control({k for k, v in keys.items() if p.btnp(v)}, {k for k, v in keys.items() if p.btn(v)}) frame += 1 p.run(update, lambda: app.draw(p)) if __name__ == '__main__': import argparse from pathlib import Path parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--headless-smoke', action='store_true') parser.add_argument('--demo-checkpoint', action='store_true') parser.add_argument('--screenshot') parser.add_argument('--report') args = parser.parse_args() if (args.demo_checkpoint or args.screenshot or args.report) and not args.headless_smoke: parser.error('demo and outputs require --headless-smoke') for option, suffix in (('screenshot', '.png'), ('report', '.json')): value = getattr(args, option) if value: path = Path(value) if path.suffix.lower() != suffix or not path.parent.is_dir() or path.exists(): parser.error('use a new '+suffix+' in an existing directory') setattr(args, option, str(path.resolve())) main(args.headless_smoke, args.screenshot, args.demo_checkpoint, args.report)