完成サンプルをブラウザーで遊ぶ
この講座のもとになった完成版のゲームを、ブラウザーでそのまま遊べます。キーボードで操作するので、PCで開いてください(スマホのタッチ操作には対応していません)。初めて開くときは、読み込みに少し時間がかかります。
完成版の主な操作:上下文、Z送信/待つ、Space一刻待つ、X一刻戻す、R再開、P説明、1〜3状況
ボタンを押すと、ゲームの読み込みが始まります。
画面に「CLICK TO START」と出たらクリックします。続けてゲーム画面を一度クリックすると、キー操作が効くようになります。ゲームを終えるときは、枠の外にある「閉じる」ボタンを使ってください(Escキーの働きはゲームごとに違います)。ゲーム内のセーブは、閉じる・ページを移る・再読み込みすると消えます。
うまく表示されないとき
ゲームを動かすためのプログラム(Pyxel 2.9.9)は、外部の配信サービス(jsDelivr)から読み込みます。表示されないときは「最初から再読込」を押すか、完成版のソースをダウンロードしてPCで実行してください。ブラウザー版の録画機能は止めています。ブラウザー版で確認できている範囲は検証状況にまとめています。
「見た時刻」と「情報を受け取った時刻」を分ける。
いきなり完成版のコードを全部読む必要はありません。小さなゲームにルールを1つずつ足しながら、5つのステップで作っていきます。どのステップのファイルもそれだけで動くので、動きを確かめてから次へ進みましょう。
- 必要な環境
- Python 3.11以上 / Pyxel 2.9.9
- 知っておくとよいこと
- 変数・if・for・関数 / タプルと辞書 / dataclass・幅優先探索・候補を絞りこむ推理は本文で説明します
- 教材に入っているもの
- ステップごとに動くPythonファイル5本・説明・動作確認用のテスト
- 完成した教材の操作
- 1:03以降で位置を質問 / 2:02以降で安全路を指示 / 3:02以降で橋T経由を指示 / 4:02以降で停止を指示 / Space:一刻待つ / X:05で一つ戻す / R:初期化 / Esc:終了
- 難易度の目安
- 中級このページで使うプログラムの考え方の数と難しさから決めた目安です。難易度について
- 学習順に迷ったら
- おすすめの学ぶ順番を見る
コードを書き換えたら、ファイルを保存し、動いているゲームを閉じてから、もう一度同じ.pyを実行します。ゲーム中にRキーなどで「やり直し」をしても、書き換えたコードは読み込まれません。
教材のゲーム画面は、日本語フォントを用意しなくても動くように英語で表示しています。ファイルには動作確認用のコードも入っていますが、最初は Game / State・update・draw の3つに注目すれば十分です。わからない言葉は用語集で調べられます。class・self・dataclass はPythonの状態、キー入力と時間の扱いは入力・更新・描画、マス目とピクセルの違いは画面とマップの座標のページで詳しく説明しています。
二刻に一歩だけ、相手を進める
ゲームの時刻と、いちばん短い道の探し方。
画面の S から H へ向かう相手を、Space で1刻ずつ進めます。T は本当の秒でも、画面を描いた回数でもなく、受け付けた判断の回数です。画面を読んだまま待っていても動きません。最初の Space で T1 になりますが、相手が動くのは 2・4・6… の偶数の刻だけです。
この段階は、相手の今いる場所が黄色い輪で見える練習です。最初の指示は safe(安全な道)に決めておき、橋 JT を使わない S→J→M→N→H を歩きます。灰色は道、黄色の二重線は橋、緑の H は避難所です。まだ無線はなく、橋が開く時刻も、移動の判定には使いません。
route は幅優先探索です。deque に「ここまでの道」を入れ、短い道から順に1つずつのばしていきます。seen は行ったことのある場所の集合で、同じ場所を回り続けるのを防ぎます。safe=True なら JT を候補から外し、同じ長さの道は sorted で場所の名前の順に選びます。道がなければ空のタプルを返し、walk はその場に止まります。
State は、時刻と World(世界のようす)をまとめた、書きかえないデータです。replace は元をこわさずに、新しい値を作ります。advance が世界を1回進め、draw は読むだけです。元のゲームの最初の指示は join で J に集まるので、この safe に決めているのは、学習のための変更です。
下のコードは 01_clock.py の一部です。前のステップから追加・変更したところだけを抜き出しています。
if tick % 2 == 0:
actual = walk(actual, level, tick)実行して確かめる
教材のZIPを展開したフォルダーで、次のコマンドを実行します。
.venv\Scripts\python.exe 01_clock.py.venv/bin/python 01_clock.py- 判断の回数と、フレームの数を分ける
- 幅優先探索で安全な道を求める
- H に着いたら状態を固定する
advance を呼ぶと、読んでいる間にも期限が来てしまいます。入力で1刻だけ進め、その新しい tick が偶数なら歩きます。R で戻して Space を8回押しましょう。T8・H・WON(成功)です。9回目からは、Space を押しても T8 のままです。次に、route(LEVEL,"S","H",safe=True) の答えが ("S","J","M","N","H") になる理由を、図で追ってみましょう。
このステップのコード全文(254行)
# MIT License
#
# Copyright (c) 2026
#
# 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.
#
# Third-party exception: bitmap glyph data derived from Noto Sans CJK is licensed
# under the SIL Open Font License 1.1. See assets/NOTO-LICENSE.txt.
"""返事の遅い無線・教材 1/5。Python 3.11+ / Pyxel 2.9.9。"""
COMMANDS = ()
from dataclasses import dataclass, replace, asdict
from collections import deque
@dataclass(frozen=True)
class Level:
edges: tuple = ('SJ', 'KU', 'UJ', 'JM', 'MN', 'NH', 'JT', 'TH')
starts: tuple = ('S', 'K')
hidden_start: str = 'S'
opens: tuple = (4, 6)
initial_mode: str = 'join'
limit: int = 12
LEVEL = Level()
@dataclass(frozen=True)
class World:
origin: str
pos: str
mode: str
via: bool = False
blocked: int = 0
def route(level, start, target, safe=False):
"""幅優先探索。同じ長さなら地点名順。安全路はJTを除く。"""
if start == target:
return (start,)
queue = deque([(start,)])
seen = {start}
while queue:
path = queue.popleft()
near = sorted({
edge[1] if edge[0] == path[-1] else edge[0]
for edge in level.edges
if path[-1] in edge
and not (safe and set(edge) == {'J', 'T'})
})
for node in near:
if node in seen:
continue
next_path = path + (node,)
if node == target:
return next_path
queue.append(next_path)
seen.add(node)
return ()
def walk(world, level, tick):
if world.pos == 'H':
return world
path = route(level, world.pos, 'H', safe=True)
return replace(world, pos=path[1]) if len(path) >= 2 else world
INITIAL_MODE = 'safe'
@dataclass(frozen=True)
class State:
t: int
actual: World
sent: int = 0
outcome: str = 'running'
def initial(level=LEVEL):
return State(0, World('S', 'S', INITIAL_MODE))
def advance(state, action, level=LEVEL):
if state.outcome != 'running':
return state, 'ended'
if action not in COMMANDS + ('wait',):
return state, 'unknown'
tick = state.t + 1
actual = state.actual
sent = state.sent
if tick % 2 == 0:
actual = walk(actual, level, tick)
outcome = ('won' if actual.pos == 'H' else
'late' if tick >= level.limit else 'running')
return State(tick, actual, sent, outcome), 'advanced'
XY = {'S': (36, 105), 'K': (36, 187), 'U': (94, 187),
'J': (159, 142), 'M': (159, 228), 'N': (215, 228),
'H': (274, 228), 'T': (274, 142)}
KEY_TO_ACTION = {'SPACE': 'wait'}
KEYS = tuple(KEY_TO_ACTION) + ('R',)
DEMO_KEYS = ('SPACE', 'SPACE')
class App:
def __init__(self):
self.reset()
def reset(self):
self.state = initial()
self.gated = False
self.message = 'READY'
def control(self, pressed, held):
# R is a fresh press; holding it does not reset every frame.
if 'R' in pressed:
self.reset()
self.gated = True
return
if self.gated:
if not pressed and not held:
self.gated = False
return
if len(pressed | held) > 1:
self.gated = True
self.message = 'RELEASE ALL KEYS'
return
if len(pressed) != 1:
return
action = KEY_TO_ACTION.get(next(iter(pressed)))
if action is None:
return
self.state, self.message = advance(self.state, action)
self.gated = True
def draw(self, p):
state = self.state
p.cls(1)
p.text(12, 12, '061 DELAYED RADIO / STAGE 1', 10)
p.text(12, 29, 'T '+str(state.t)+'/12 SENT '+str(state.sent)
+' '+state.outcome.upper(), 11)
p.text(12, 48, 'YELLOW RING = KNOWN POSITION', 6)
p.rectb(12, 72, 288, 178, 5)
for edge in LEVEL.edges:
x, y = XY[edge[0]]
u, v = XY[edge[1]]
bridge = set(edge) == {'J', 'T'}
p.line(x, y, u, v, 10 if bridge else 5)
if bridge:
p.line(x, y+3, u, v+3, 10)
locations = (state.actual.pos,)
for node, (x, y) in XY.items():
p.circ(x, y, 9, 3 if node == 'H' else 1)
p.circb(x, y, 9, 11 if node == 'H' else 6)
p.text(x-2, y-2, node, 7)
if node in locations:
p.circb(x, y, 14, 10)
p.text(180, 126, 'BRIDGE JT', 10)
p.text(310, 78, 'NEXT WALK: '+str(state.t+2-state.t%2), 7)
p.text(310, 94, 'BRIDGE NOT USED', 10)
p.text(310, 110, 'RINGS: '+','.join(locations), 6)
p.text(310, 135, 'SAFE ROUTE ONLY', 7)
p.text(310, 150, 'S-J-M-N-H', 7)
p.text(310, 165, 'NO RADIO YET', 7)
p.text(310, 196, 'SPACE WAIT', 6)
p.text(310, 211, 'RELEASE EACH KEY', 6)
p.text(310, 226, 'R RESET / ESC QUIT', 6)
p.rect(12, 260, 456, 57, 2)
p.text(20, 270, 'A LOGICAL TICK ADVANCES ONLY ON INPUT.', 7)
p.text(20, 287, 'TWO TICKS = ONE WALK. READING IS FREE.', 6)
p.text(20, 303, 'RELEASE ALL KEYS BEFORE THE NEXT ACTION.', 13)
p.text(12, 328, self.message.upper(), 11)
def parse_args(argv=None):
"""すべてのCLI検証はPyxelをimportする前に終える。"""
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('--frames', type=int)
parser.add_argument('--screenshot')
parser.add_argument('--state-output')
args = parser.parse_args(argv)
if not args.headless_smoke and (args.demo_checkpoint or args.frames is not None
or args.screenshot or args.state_output):
parser.error('QA options require --headless-smoke')
minimum = 2 * len(DEMO_KEYS) + 4 if args.demo_checkpoint else 4
if args.frames is None:
args.frames = minimum
if not minimum <= args.frames <= 600:
parser.error('--frames must be between '+str(minimum)+' and 600')
for name, suffix in (('screenshot', '.png'), ('state_output', '.json')):
value = getattr(args, name)
if value is not None and (not value.strip() or Path(value).suffix.lower() != suffix):
parser.error('--'+name.replace('_', '-')+' requires a '+suffix+' file path')
return args
def main(argv=None):
args = parse_args(argv)
from pathlib import Path
import json
import pyxel as p
app = App()
p.init(480, 352, title='061 Delayed Radio', fps=30,
display_scale=2, headless=args.headless_smoke)
keys = {key: getattr(p, 'KEY_'+key) for key in KEYS}
finished = False
def update():
if args.headless_smoke:
index = p.frame_count // 2
pressed = ({DEMO_KEYS[index]} if args.demo_checkpoint
and p.frame_count % 2 == 0 and index < len(DEMO_KEYS) else set())
app.control(pressed, pressed)
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)})
def draw():
nonlocal finished
app.draw(p)
if args.headless_smoke and p.frame_count >= args.frames-1 and not finished:
finished = True
if args.screenshot:
path = Path(args.screenshot).resolve()
path.parent.mkdir(parents=True, exist_ok=True)
p.screenshot(str(path), scale=2)
if args.state_output:
path = Path(args.state_output).resolve()
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(asdict(app.state), ensure_ascii=False, indent=2)+'\n', encoding='utf-8')
p.quit()
p.run(update, draw)
if __name__ == '__main__':
main()指示を変えてから、その刻の移動を決める
受信と移動の順番・橋が開いている時間。
ここからは、最初の指示が join で、相手は J に集まります。2 は安全な道で H へ、3 は T を通って H へ、4 はその場で止まる指示です。この段階の指示は、次の1刻に進んだ時点で反映され、返事を待つ仕組みはまだ入れません。Space なら、指示を変えずに1刻進めます。
advance は、先に instruct で mode(指示)を変え、そのあとで偶数の刻なら walk します。T1 で「止まる」を送ると、T2 の移動より先に止まる指示が反映され、相手は S に残ります。「歩いてから指示を変える」順番だと、同じ操作でも結果が J になってしまいます。
bridge の via は、「まだ T を通っていない」という印です。T に着くまでは目標が T、着いたあとは H です。JT を渡れるのは T4 と T6 だけです。閉まっていたら位置は変えず、blocked を1つ増やして、次の偶数の刻を待ちます。safe は、JT そのものを道の探索から外します。橋が閉まっているときに、安全な道に切りかえることもできます。
この段階では場所が見えるので、道と順番に集中できます。返事なしで続けて指示できるところは、元のゲームとちがいます。WON は H に着くこと、LATE は12刻までに着けないことです。最後の刻ちょうどに H に着いたら、WON を先に判定します。
下のコードは 02_commands.py の一部です。前のステップから追加・変更したところだけを抜き出しています。
if action in COMMANDS:
actual = instruct(actual, action)
sent += 1実行して確かめる
教材のZIPを展開したフォルダーで、次のコマンドを実行します。
.venv\Scripts\python.exe 02_commands.py.venv/bin/python 02_commands.py- 指示を反映してから歩く
- 橋は、表示された刻だけ渡れる
- 最後の刻での成功を、期限切れより優先する
bridge にした瞬間に、T へ瞬間移動させないようにしましょう。毎回 route で次の1区間だけを選び、JT を渡る刻だけ、橋が開いているかを確かめます。R → Space → 4 で、T2・S・止まる です。Space を2回押しても、T4・S のままです。続けて 2 → Space を7回押すと、T12・H・WON です。止まる指示が、偶数の刻の歩きより先に効くことと、期限ちょうどの成功を確かめましょう。
このステップのコード全文(277行)
# MIT License
#
# Copyright (c) 2026
#
# 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.
#
# Third-party exception: bitmap glyph data derived from Noto Sans CJK is licensed
# under the SIL Open Font License 1.1. See assets/NOTO-LICENSE.txt.
"""返事の遅い無線・教材 2/5。Python 3.11+ / Pyxel 2.9.9。"""
COMMANDS = ('safe', 'bridge', 'stop')
from dataclasses import dataclass, replace, asdict
from collections import deque
@dataclass(frozen=True)
class Level:
edges: tuple = ('SJ', 'KU', 'UJ', 'JM', 'MN', 'NH', 'JT', 'TH')
starts: tuple = ('S', 'K')
hidden_start: str = 'S'
opens: tuple = (4, 6)
initial_mode: str = 'join'
limit: int = 12
LEVEL = Level()
@dataclass(frozen=True)
class World:
origin: str
pos: str
mode: str
via: bool = False
blocked: int = 0
def route(level, start, target, safe=False):
"""幅優先探索。同じ長さなら地点名順。安全路はJTを除く。"""
if start == target:
return (start,)
queue = deque([(start,)])
seen = {start}
while queue:
path = queue.popleft()
near = sorted({
edge[1] if edge[0] == path[-1] else edge[0]
for edge in level.edges
if path[-1] in edge
and not (safe and set(edge) == {'J', 'T'})
})
for node in near:
if node in seen:
continue
next_path = path + (node,)
if node == target:
return next_path
queue.append(next_path)
seen.add(node)
return ()
def goal(world):
if world.mode == 'stop':
return '待機'
if world.mode == 'join':
return 'J'
return 'T→H' if world.mode == 'bridge' and world.via else 'H'
def instruct(world, command):
if world.pos == 'H' or command == 'ask':
return world
return replace(world, mode=command,
via=(command == 'bridge' and world.pos != 'T'))
def walk(world, level, tick):
if world.pos == 'H' or world.mode == 'stop':
return world
target = ('J' if world.mode == 'join' else
'T' if world.mode == 'bridge' and world.via else 'H')
path = route(level, world.pos, target, safe=world.mode == 'safe')
if len(path) < 2:
return world
if set((world.pos, path[1])) == {'J', 'T'} and tick not in level.opens:
return replace(world, blocked=world.blocked + 1)
return replace(world, pos=path[1], via=world.via and path[1] != 'T')
INITIAL_MODE = 'join'
@dataclass(frozen=True)
class State:
t: int
actual: World
sent: int = 0
outcome: str = 'running'
def initial(level=LEVEL):
return State(0, World('S', 'S', INITIAL_MODE))
def advance(state, action, level=LEVEL):
if state.outcome != 'running':
return state, 'ended'
if action not in COMMANDS + ('wait',):
return state, 'unknown'
tick = state.t + 1
actual = state.actual
sent = state.sent
if action in COMMANDS:
actual = instruct(actual, action)
sent += 1
if tick % 2 == 0:
actual = walk(actual, level, tick)
outcome = ('won' if actual.pos == 'H' else
'late' if tick >= level.limit else 'running')
return State(tick, actual, sent, outcome), 'advanced'
XY = {'S': (36, 105), 'K': (36, 187), 'U': (94, 187),
'J': (159, 142), 'M': (159, 228), 'N': (215, 228),
'H': (274, 228), 'T': (274, 142)}
KEY_TO_ACTION = {'SPACE': 'wait', '2': 'safe', '3': 'bridge', '4': 'stop'}
KEYS = tuple(KEY_TO_ACTION) + ('R',)
DEMO_KEYS = ('3', 'SPACE', 'SPACE', 'SPACE', 'SPACE', 'SPACE')
class App:
def __init__(self):
self.reset()
def reset(self):
self.state = initial()
self.gated = False
self.message = 'READY'
def control(self, pressed, held):
# R is a fresh press; holding it does not reset every frame.
if 'R' in pressed:
self.reset()
self.gated = True
return
if self.gated:
if not pressed and not held:
self.gated = False
return
if len(pressed | held) > 1:
self.gated = True
self.message = 'RELEASE ALL KEYS'
return
if len(pressed) != 1:
return
action = KEY_TO_ACTION.get(next(iter(pressed)))
if action is None:
return
self.state, self.message = advance(self.state, action)
self.gated = True
def draw(self, p):
state = self.state
p.cls(1)
p.text(12, 12, '061 DELAYED RADIO / STAGE 2', 10)
p.text(12, 29, 'T '+str(state.t)+'/12 SENT '+str(state.sent)
+' '+state.outcome.upper(), 11)
p.text(12, 48, 'YELLOW RING = KNOWN POSITION', 6)
p.rectb(12, 72, 288, 178, 5)
for edge in LEVEL.edges:
x, y = XY[edge[0]]
u, v = XY[edge[1]]
bridge = set(edge) == {'J', 'T'}
p.line(x, y, u, v, 10 if bridge else 5)
if bridge:
p.line(x, y+3, u, v+3, 10)
locations = (state.actual.pos,)
for node, (x, y) in XY.items():
p.circ(x, y, 9, 3 if node == 'H' else 1)
p.circb(x, y, 9, 11 if node == 'H' else 6)
p.text(x-2, y-2, node, 7)
if node in locations:
p.circb(x, y, 14, 10)
p.text(180, 126, 'BRIDGE JT', 10)
p.text(310, 78, 'NEXT WALK: '+str(state.t+2-state.t%2), 7)
p.text(310, 94, 'BRIDGE AT: 4,6', 10)
p.text(310, 110, 'RINGS: '+','.join(locations), 6)
p.text(310, 135, 'INSTANT COMMANDS', 7)
p.text(310, 150, '2 SAFE / 3 BRIDGE', 7)
p.text(310, 165, '4 STOP', 7)
p.text(310, 196, 'SPACE WAIT', 6)
p.text(310, 211, 'RELEASE EACH KEY', 6)
p.text(310, 226, 'R RESET / ESC QUIT', 6)
p.rect(12, 260, 456, 57, 2)
p.text(20, 270, 'A LOGICAL TICK ADVANCES ONLY ON INPUT.', 7)
p.text(20, 287, 'TWO TICKS = ONE WALK. READING IS FREE.', 6)
p.text(20, 303, 'RELEASE ALL KEYS BEFORE THE NEXT ACTION.', 13)
p.text(12, 328, self.message.upper(), 11)
def parse_args(argv=None):
"""すべてのCLI検証はPyxelをimportする前に終える。"""
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('--frames', type=int)
parser.add_argument('--screenshot')
parser.add_argument('--state-output')
args = parser.parse_args(argv)
if not args.headless_smoke and (args.demo_checkpoint or args.frames is not None
or args.screenshot or args.state_output):
parser.error('QA options require --headless-smoke')
minimum = 2 * len(DEMO_KEYS) + 4 if args.demo_checkpoint else 4
if args.frames is None:
args.frames = minimum
if not minimum <= args.frames <= 600:
parser.error('--frames must be between '+str(minimum)+' and 600')
for name, suffix in (('screenshot', '.png'), ('state_output', '.json')):
value = getattr(args, name)
if value is not None and (not value.strip() or Path(value).suffix.lower() != suffix):
parser.error('--'+name.replace('_', '-')+' requires a '+suffix+' file path')
return args
def main(argv=None):
args = parse_args(argv)
from pathlib import Path
import json
import pyxel as p
app = App()
p.init(480, 352, title='061 Delayed Radio', fps=30,
display_scale=2, headless=args.headless_smoke)
keys = {key: getattr(p, 'KEY_'+key) for key in KEYS}
finished = False
def update():
if args.headless_smoke:
index = p.frame_count // 2
pressed = ({DEMO_KEYS[index]} if args.demo_checkpoint
and p.frame_count % 2 == 0 and index < len(DEMO_KEYS) else set())
app.control(pressed, pressed)
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)})
def draw():
nonlocal finished
app.draw(p)
if args.headless_smoke and p.frame_count >= args.frames-1 and not finished:
finished = True
if args.screenshot:
path = Path(args.screenshot).resolve()
path.parent.mkdir(parents=True, exist_ok=True)
p.screenshot(str(path), scale=2)
if args.state_output:
path = Path(args.state_output).resolve()
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(asdict(app.state), ensure_ascii=False, indent=2)+'\n', encoding='utf-8')
p.quit()
p.run(update, draw)
if __name__ == '__main__':
main()古い位置を封筒に入れて返す
1件だけの通信待ちと、記録した時刻。
1 の質問を加えます。送った時刻を t とすると、相手が受け取るのは t+1、返事が届くのは t+2 です。1回の入力で世界が1刻進むので、送る操作が終わった画面では、もう相手が文を受け取っています。pending は、まだ届いていない返事を1件だけ入れておく箱です。
送った直後、instruct で指示を反映し、相手の位置と目標を snapshot(そのときの記録)に保存します。これは、受け取った直後で、その刻の移動の前の記録です。続いて、偶数の刻なら歩きます。返事の刻には、保存しておいた値を取り出し、今の位置で上書きはしません。Reply は、送った・受け取った・届いた・記録した時刻を、別々に持ちます。
T0 で質問すると、T1 の S を保存します。次の Space で T2 になり、相手は J へ歩きますが、返ってくる言葉は「記録 T1、位置 S」です。画面には、今の輪 J と、下の古い返事 S が同時に表示されます。この段階だけは相手の位置を見せて、時刻のちがいを直接見比べます。
返事を待っている間に、別の質問や指示を送ると、BUSY(通信中)として受け付けません。時刻・送った回数・保存した記録は、何も変わりません。Space で待って返事を受け取れば、また送れます。safe などの返事は「受け取った」という確認で、位置は報告しません。待つ返事が1件だけで足りるのは、元のゲームも同時の通信を1件に限っているからです。
下のコードは 03_delay.py の一部です。前のステップから追加・変更したところだけを抜き出しています。
pending = Pending(action, state.t, tick, tick + 1, (snapshot(actual),))実行して確かめる
教材のZIPを展開したフォルダーで、次のコマンドを実行します。
.venv\Scripts\python.exe 03_delay.py.venv/bin/python 03_delay.py- 送った・受け取った・届いたを区別する
- 歩く前の記録を保存する
- 通信待ちのときに送っても、時刻は進まない
actual.pos(今の位置)を返事に使うと、このゲームの「少し前の返事」が消えてしまいます。pending.snapshots を保存しておき、その中身を返事に移します。BUSY を表示してから tick を増やすのも、まちがいです。R → Space → 1 → Space と押してみましょう。T3 で輪は J、返事は「位置 S・記録 T2・届いた T3」です。質問が偶数の刻 T2 に届いた場合も、その刻に歩く前を保存します。別に R → 1 → 2 を試すと、BUSY で T1、送った回数は1のままです。
このステップのコード全文(325行)
# MIT License
#
# Copyright (c) 2026
#
# 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.
#
# Third-party exception: bitmap glyph data derived from Noto Sans CJK is licensed
# under the SIL Open Font License 1.1. See assets/NOTO-LICENSE.txt.
"""返事の遅い無線・教材 3/5。Python 3.11+ / Pyxel 2.9.9。"""
COMMANDS = ('ask', 'safe', 'bridge', 'stop')
from dataclasses import dataclass, replace, asdict
from collections import deque
@dataclass(frozen=True)
class Level:
edges: tuple = ('SJ', 'KU', 'UJ', 'JM', 'MN', 'NH', 'JT', 'TH')
starts: tuple = ('S', 'K')
hidden_start: str = 'S'
opens: tuple = (4, 6)
initial_mode: str = 'join'
limit: int = 12
LEVEL = Level()
@dataclass(frozen=True)
class World:
origin: str
pos: str
mode: str
via: bool = False
blocked: int = 0
def route(level, start, target, safe=False):
"""幅優先探索。同じ長さなら地点名順。安全路はJTを除く。"""
if start == target:
return (start,)
queue = deque([(start,)])
seen = {start}
while queue:
path = queue.popleft()
near = sorted({
edge[1] if edge[0] == path[-1] else edge[0]
for edge in level.edges
if path[-1] in edge
and not (safe and set(edge) == {'J', 'T'})
})
for node in near:
if node in seen:
continue
next_path = path + (node,)
if node == target:
return next_path
queue.append(next_path)
seen.add(node)
return ()
def goal(world):
if world.mode == 'stop':
return '待機'
if world.mode == 'join':
return 'J'
return 'T→H' if world.mode == 'bridge' and world.via else 'H'
def instruct(world, command):
if world.pos == 'H' or command == 'ask':
return world
return replace(world, mode=command,
via=(command == 'bridge' and world.pos != 'T'))
def walk(world, level, tick):
if world.pos == 'H' or world.mode == 'stop':
return world
target = ('J' if world.mode == 'join' else
'T' if world.mode == 'bridge' and world.via else 'H')
path = route(level, world.pos, target, safe=world.mode == 'safe')
if len(path) < 2:
return world
if set((world.pos, path[1])) == {'J', 'T'} and tick not in level.opens:
return replace(world, blocked=world.blocked + 1)
return replace(world, pos=path[1], via=world.via and path[1] != 'T')
@dataclass(frozen=True)
class Pending:
command: str
sent: int
received: int
arrives: int
snapshots: tuple
@dataclass(frozen=True)
class Reply:
command: str
sent: int
received: int
arrived: int
sample: int
phase: str
pos: str
goal: str
def snapshot(world):
return (world.origin, world.pos, goal(world))
@dataclass(frozen=True)
class State:
t: int
actual: World
pending: object = None
replies: tuple = ()
sent: int = 0
outcome: str = 'running'
def initial(level=LEVEL):
return State(0, World('S', 'S', level.initial_mode))
def advance(state, action, level=LEVEL):
if state.outcome != 'running':
return state, 'ended'
if action not in COMMANDS + ('wait',):
return state, 'unknown'
if action in COMMANDS and state.pending:
return state, 'busy'
tick = state.t + 1
actual, pending = state.actual, state.pending
replies, sent = state.replies, state.sent
if action in COMMANDS:
actual = instruct(actual, action)
pending = Pending(action, state.t, tick, tick + 1, (snapshot(actual),))
sent += 1
if tick % 2 == 0:
actual = walk(actual, level, tick)
if pending and pending.arrives == tick:
pos, dest = pending.snapshots[0][1:] if pending.command == 'ask' else ('', '')
replies += (Reply(pending.command, pending.sent, pending.received,
tick, pending.received, '移動前', pos, dest),)
pending = None
outcome = ('won' if actual.pos == 'H' else
'late' if tick >= level.limit else 'running')
return State(tick, actual, pending, replies, sent, outcome), 'advanced'
XY = {'S': (36, 105), 'K': (36, 187), 'U': (94, 187),
'J': (159, 142), 'M': (159, 228), 'N': (215, 228),
'H': (274, 228), 'T': (274, 142)}
KEY_TO_ACTION = {'SPACE': 'wait', '1': 'ask', '2': 'safe', '3': 'bridge', '4': 'stop'}
KEYS = tuple(KEY_TO_ACTION) + ('R',)
DEMO_KEYS = ('1', 'SPACE')
class App:
def __init__(self):
self.reset()
def reset(self):
self.state = initial()
self.gated = False
self.message = 'READY'
def control(self, pressed, held):
# R is a fresh press; holding it does not reset every frame.
if 'R' in pressed:
self.reset()
self.gated = True
return
if self.gated:
if not pressed and not held:
self.gated = False
return
if len(pressed | held) > 1:
self.gated = True
self.message = 'RELEASE ALL KEYS'
return
if len(pressed) != 1:
return
action = KEY_TO_ACTION.get(next(iter(pressed)))
if action is None:
return
self.state, self.message = advance(self.state, action)
self.gated = True
def draw(self, p):
state = self.state
p.cls(1)
p.text(12, 12, '061 DELAYED RADIO / STAGE 3', 10)
p.text(12, 29, 'T '+str(state.t)+'/12 SENT '+str(state.sent)
+' '+state.outcome.upper(), 11)
p.text(12, 48, 'YELLOW RING = KNOWN POSITION', 6)
p.rectb(12, 72, 288, 178, 5)
for edge in LEVEL.edges:
x, y = XY[edge[0]]
u, v = XY[edge[1]]
bridge = set(edge) == {'J', 'T'}
p.line(x, y, u, v, 10 if bridge else 5)
if bridge:
p.line(x, y+3, u, v+3, 10)
locations = (state.actual.pos,)
for node, (x, y) in XY.items():
p.circ(x, y, 9, 3 if node == 'H' else 1)
p.circb(x, y, 9, 11 if node == 'H' else 6)
p.text(x-2, y-2, node, 7)
if node in locations:
p.circb(x, y, 14, 10)
p.text(180, 126, 'BRIDGE JT', 10)
p.text(310, 78, 'NEXT WALK: '+str(state.t+2-state.t%2), 7)
p.text(310, 94, 'BRIDGE AT: 4,6', 10)
p.text(310, 110, 'RINGS: '+','.join(locations), 6)
p.text(310, 135, '1 ASK: WHERE?', 7)
p.text(310, 150, '2 SAFE / 3 BRIDGE', 7)
p.text(310, 165, '4 STOP', 7)
p.text(310, 196, 'SPACE WAIT', 6)
p.text(310, 211, 'RELEASE EACH KEY', 6)
p.text(310, 226, 'R RESET / ESC QUIT', 6)
p.rect(12, 260, 456, 57, 2)
pending = state.pending
if pending:
line = ('SENT '+str(pending.sent)+' / RECEIVED '+str(pending.received)
+' / REPLY AT '+str(pending.arrives))
else:
line = 'RADIO READY: SEND NOW'
p.text(20, 269, line, 7)
if state.replies:
reply = state.replies[-1]
if reply.command == 'ask':
dest = {'待機': 'STOP', 'T→H': 'T>H'}.get(reply.goal, reply.goal)
line = ('SAMPLE T'+str(reply.sample)+' BEFORE WALK: '
+reply.pos+' -> '+dest)
else:
line = reply.command.upper()+' RECEIVED T'+str(reply.received)
p.text(20, 287, line, 11)
p.text(20, 303, 'REPLY ARRIVED T'+str(reply.arrived)+' / THIS IS OLD DATA', 13)
else:
p.text(20, 287, 'NO REPLY YET. WAIT WITH SPACE.', 6)
p.text(20, 303, 'A SECOND SEND IS REJECTED WHILE BUSY.', 13)
p.text(12, 328, self.message.upper(), 11)
def parse_args(argv=None):
"""すべてのCLI検証はPyxelをimportする前に終える。"""
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('--frames', type=int)
parser.add_argument('--screenshot')
parser.add_argument('--state-output')
args = parser.parse_args(argv)
if not args.headless_smoke and (args.demo_checkpoint or args.frames is not None
or args.screenshot or args.state_output):
parser.error('QA options require --headless-smoke')
minimum = 2 * len(DEMO_KEYS) + 4 if args.demo_checkpoint else 4
if args.frames is None:
args.frames = minimum
if not minimum <= args.frames <= 600:
parser.error('--frames must be between '+str(minimum)+' and 600')
for name, suffix in (('screenshot', '.png'), ('state_output', '.json')):
value = getattr(args, name)
if value is not None and (not value.strip() or Path(value).suffix.lower() != suffix):
parser.error('--'+name.replace('_', '-')+' requires a '+suffix+' file path')
return args
def main(argv=None):
args = parse_args(argv)
from pathlib import Path
import json
import pyxel as p
app = App()
p.init(480, 352, title='061 Delayed Radio', fps=30,
display_scale=2, headless=args.headless_smoke)
keys = {key: getattr(p, 'KEY_'+key) for key in KEYS}
finished = False
def update():
if args.headless_smoke:
index = p.frame_count // 2
pressed = ({DEMO_KEYS[index]} if args.demo_checkpoint
and p.frame_count % 2 == 0 and index < len(DEMO_KEYS) else set())
app.control(pressed, pressed)
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)})
def draw():
nonlocal finished
app.draw(p)
if args.headless_smoke and p.frame_count >= args.frames-1 and not finished:
finished = True
if args.screenshot:
path = Path(args.screenshot).resolve()
path.parent.mkdir(parents=True, exist_ok=True)
p.screenshot(str(path), scale=2)
if args.state_output:
path = Path(args.state_output).resolve()
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(asdict(app.state), ensure_ascii=False, indent=2)+'\n', encoding='utf-8')
p.quit()
p.run(update, draw)
if __name__ == '__main__':
main()返事に合う出発点だけを残す
本当の世界と、知っている候補を分ける。
相手の今いる場所をかくして、S から出発した場合と、K から出発した場合の2つの World を candidates(候補)に持たせます。黄色い輪は、その時刻にいるかもしれない場所です。本当の出発点は、元のゲームの状況 A と同じ S に決めていますが、描く処理は actual.pos を読みません。乱数で変わるゲームではありません。
指示と歩きは、actual(本当の世界)だけでなく、すべての候補に同じ順番で当てはめます。質問を受け取ったときには、候補ごとに、出発点 origin・位置・目的地を保存します。返事が届いたら、本当の出発点の記録を答えとして選び、答えと位置・目的地が一致する過去の記録をもつ候補だけを残します。
大事なのは、過去の記録で候補を選んでも、候補の今の位置を過去に戻さないことです。T0 で質問して T2 で返事が届いたとき、S から出発した候補は、今は J にいます。返事の S と、今の候補 J を比べて捨てると、本当の答えまで消してしまいます。origin を使って、記録したときと今の、同じ世界を結びつけます。
いくつかの出発点が同じ場所に合流していると、質問しても区別できません。WORLDS は残った候補の数で、輪の数と同じとは限りません。ここで、元のゲームの状況 A のふつうの advance を再現します。状況 B・C、返事のときに今の位置を答える比較モード、日本語の文字の画面や記録の保存は、教材では扱いません。
下のコードは 04_candidates.py の一部です。前のステップから追加・変更したところだけを抜き出しています。
candidates = tuple(w for w in candidates if any(
row[0] == w.origin and row[1:] == answer[1:]
for row in pending.snapshots))実行して確かめる
教材のZIPを展開したフォルダーで、次のコマンドを実行します。
.venv\Scripts\python.exe 04_candidates.py.venv/bin/python 04_candidates.py- すべての候補を、同じ時計で進める
- 記録した時点で比べて、今の候補を残す
- 同じ場所にいても、別の出発点として残す
origin を選び、その origin の今の World を残します。R → Space を4回 → 1 → Space と押してみましょう。T6 で、返事の位置は J、今の輪も J ですが、WORLDS 2 のままです。T4 までに両方が J に合流しているので、この質問では出発点を区別できません。
このステップのコード全文(344行)
# MIT License
#
# Copyright (c) 2026
#
# 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.
#
# Third-party exception: bitmap glyph data derived from Noto Sans CJK is licensed
# under the SIL Open Font License 1.1. See assets/NOTO-LICENSE.txt.
"""返事の遅い無線・教材 4/5。Python 3.11+ / Pyxel 2.9.9。"""
COMMANDS = ('ask', 'safe', 'bridge', 'stop')
from dataclasses import dataclass, replace, asdict
from collections import deque
@dataclass(frozen=True)
class Level:
edges: tuple = ('SJ', 'KU', 'UJ', 'JM', 'MN', 'NH', 'JT', 'TH')
starts: tuple = ('S', 'K')
hidden_start: str = 'S'
opens: tuple = (4, 6)
initial_mode: str = 'join'
limit: int = 12
LEVEL = Level()
@dataclass(frozen=True)
class World:
origin: str
pos: str
mode: str
via: bool = False
blocked: int = 0
def route(level, start, target, safe=False):
"""幅優先探索。同じ長さなら地点名順。安全路はJTを除く。"""
if start == target:
return (start,)
queue = deque([(start,)])
seen = {start}
while queue:
path = queue.popleft()
near = sorted({
edge[1] if edge[0] == path[-1] else edge[0]
for edge in level.edges
if path[-1] in edge
and not (safe and set(edge) == {'J', 'T'})
})
for node in near:
if node in seen:
continue
next_path = path + (node,)
if node == target:
return next_path
queue.append(next_path)
seen.add(node)
return ()
def goal(world):
if world.mode == 'stop':
return '待機'
if world.mode == 'join':
return 'J'
return 'T→H' if world.mode == 'bridge' and world.via else 'H'
def instruct(world, command):
if world.pos == 'H' or command == 'ask':
return world
return replace(world, mode=command,
via=(command == 'bridge' and world.pos != 'T'))
def walk(world, level, tick):
if world.pos == 'H' or world.mode == 'stop':
return world
target = ('J' if world.mode == 'join' else
'T' if world.mode == 'bridge' and world.via else 'H')
path = route(level, world.pos, target, safe=world.mode == 'safe')
if len(path) < 2:
return world
if set((world.pos, path[1])) == {'J', 'T'} and tick not in level.opens:
return replace(world, blocked=world.blocked + 1)
return replace(world, pos=path[1], via=world.via and path[1] != 'T')
@dataclass(frozen=True)
class Pending:
command: str
sent: int
received: int
arrives: int
snapshots: tuple
@dataclass(frozen=True)
class Reply:
command: str
sent: int
received: int
arrived: int
sample: int
phase: str
pos: str
goal: str
def snapshot(world):
return (world.origin, world.pos, goal(world))
@dataclass(frozen=True)
class State:
t: int
actual: World
candidates: tuple
pending: object = None
replies: tuple = ()
sent: int = 0
outcome: str = 'running'
def initial(level=LEVEL):
worlds = tuple(World(node, node, level.initial_mode) for node in level.starts)
return State(0, next(w for w in worlds if w.origin == level.hidden_start), worlds)
def advance(state, action, level=LEVEL):
"""受信→偶数刻の移動→返信→終了。拒否は同じStateを返す。"""
if state.outcome != 'running':
return state, 'ended'
if action not in COMMANDS + ('wait',):
return state, 'unknown'
if action in COMMANDS and state.pending:
return state, 'busy'
tick = state.t + 1
actual, candidates = state.actual, state.candidates
pending, replies, sent = state.pending, state.replies, state.sent
if action in COMMANDS:
actual = instruct(actual, action)
candidates = tuple(instruct(w, action) for w in candidates)
pending = Pending(action, state.t, tick, tick + 1,
tuple(snapshot(w) for w in candidates))
sent += 1
if tick % 2 == 0:
actual = walk(actual, level, tick)
candidates = tuple(walk(w, level, tick) for w in candidates)
if pending and pending.arrives == tick:
pos = dest = ''
if pending.command == 'ask':
answer = next(row for row in pending.snapshots if row[0] == actual.origin)
pos, dest = answer[1:]
candidates = tuple(w for w in candidates if any(
row[0] == w.origin and row[1:] == answer[1:]
for row in pending.snapshots))
replies += (Reply(pending.command, pending.sent, pending.received,
tick, pending.received, '移動前', pos, dest),)
pending = None
outcome = ('won' if actual.pos == 'H' else
'late' if tick >= level.limit else 'running')
reason = 'rescued' if outcome == 'won' else 'deadline' if outcome == 'late' else 'advanced'
return State(tick, actual, candidates, pending, replies, sent, outcome), reason
def visible(state):
"""真の位置ではなく、同じ時計で進んだ候補だけを画面へ渡す。"""
return tuple(sorted({world.pos for world in state.candidates}))
XY = {'S': (36, 105), 'K': (36, 187), 'U': (94, 187),
'J': (159, 142), 'M': (159, 228), 'N': (215, 228),
'H': (274, 228), 'T': (274, 142)}
KEY_TO_ACTION = {'SPACE': 'wait', '1': 'ask', '2': 'safe', '3': 'bridge', '4': 'stop'}
KEYS = tuple(KEY_TO_ACTION) + ('R',)
DEMO_KEYS = ('1', 'SPACE')
class App:
def __init__(self):
self.reset()
def reset(self):
self.state = initial()
self.gated = False
self.message = 'READY'
def control(self, pressed, held):
# R is a fresh press; holding it does not reset every frame.
if 'R' in pressed:
self.reset()
self.gated = True
return
if self.gated:
if not pressed and not held:
self.gated = False
return
if len(pressed | held) > 1:
self.gated = True
self.message = 'RELEASE ALL KEYS'
return
if len(pressed) != 1:
return
action = KEY_TO_ACTION.get(next(iter(pressed)))
if action is None:
return
self.state, self.message = advance(self.state, action)
self.gated = True
def draw(self, p):
state = self.state
p.cls(1)
p.text(12, 12, '061 DELAYED RADIO / STAGE 4', 10)
p.text(12, 29, 'T '+str(state.t)+'/12 SENT '+str(state.sent)
+' '+state.outcome.upper(), 11)
p.text(12, 48, 'YELLOW RINGS = CURRENT CANDIDATES', 6)
p.rectb(12, 72, 288, 178, 5)
for edge in LEVEL.edges:
x, y = XY[edge[0]]
u, v = XY[edge[1]]
bridge = set(edge) == {'J', 'T'}
p.line(x, y, u, v, 10 if bridge else 5)
if bridge:
p.line(x, y+3, u, v+3, 10)
locations = visible(state)
for node, (x, y) in XY.items():
p.circ(x, y, 9, 3 if node == 'H' else 1)
p.circb(x, y, 9, 11 if node == 'H' else 6)
p.text(x-2, y-2, node, 7)
if node in locations:
p.circb(x, y, 14, 10)
p.text(180, 126, 'BRIDGE JT', 10)
p.text(310, 78, 'NEXT WALK: '+str(state.t+2-state.t%2), 7)
p.text(310, 94, 'BRIDGE AT: 4,6', 10)
p.text(310, 110, 'RINGS: '+','.join(locations), 6)
p.text(310, 135, '1 ASK: WHERE?', 7)
p.text(310, 150, '2 SAFE / 3 BRIDGE', 7)
p.text(310, 165, '4 STOP', 7)
p.text(310, 181, 'WORLDS: '+str(len(state.candidates)), 13)
p.text(310, 196, 'SPACE WAIT', 6)
p.text(310, 211, 'RELEASE EACH KEY', 6)
p.text(310, 226, 'R RESET / ESC QUIT', 6)
p.rect(12, 260, 456, 57, 2)
pending = state.pending
if pending:
line = ('SENT '+str(pending.sent)+' / RECEIVED '+str(pending.received)
+' / REPLY AT '+str(pending.arrives))
else:
line = 'RADIO READY: SEND NOW'
p.text(20, 269, line, 7)
if state.replies:
reply = state.replies[-1]
if reply.command == 'ask':
dest = {'待機': 'STOP', 'T→H': 'T>H'}.get(reply.goal, reply.goal)
line = ('SAMPLE T'+str(reply.sample)+' BEFORE WALK: '
+reply.pos+' -> '+dest)
else:
line = reply.command.upper()+' RECEIVED T'+str(reply.received)
p.text(20, 287, line, 11)
p.text(20, 303, 'REPLY ARRIVED T'+str(reply.arrived)+' / THIS IS OLD DATA', 13)
else:
p.text(20, 287, 'NO REPLY YET. WAIT WITH SPACE.', 6)
p.text(20, 303, 'A SECOND SEND IS REJECTED WHILE BUSY.', 13)
p.text(12, 328, self.message.upper(), 11)
def parse_args(argv=None):
"""すべてのCLI検証はPyxelをimportする前に終える。"""
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('--frames', type=int)
parser.add_argument('--screenshot')
parser.add_argument('--state-output')
args = parser.parse_args(argv)
if not args.headless_smoke and (args.demo_checkpoint or args.frames is not None
or args.screenshot or args.state_output):
parser.error('QA options require --headless-smoke')
minimum = 2 * len(DEMO_KEYS) + 4 if args.demo_checkpoint else 4
if args.frames is None:
args.frames = minimum
if not minimum <= args.frames <= 600:
parser.error('--frames must be between '+str(minimum)+' and 600')
for name, suffix in (('screenshot', '.png'), ('state_output', '.json')):
value = getattr(args, name)
if value is not None and (not value.strip() or Path(value).suffix.lower() != suffix):
parser.error('--'+name.replace('_', '-')+' requires a '+suffix+' file path')
return args
def main(argv=None):
args = parse_args(argv)
from pathlib import Path
import json
import pyxel as p
app = App()
p.init(480, 352, title='061 Delayed Radio', fps=30,
display_scale=2, headless=args.headless_smoke)
keys = {key: getattr(p, 'KEY_'+key) for key in KEYS}
finished = False
def update():
if args.headless_smoke:
index = p.frame_count // 2
pressed = ({DEMO_KEYS[index]} if args.demo_checkpoint
and p.frame_count % 2 == 0 and index < len(DEMO_KEYS) else set())
app.control(pressed, pressed)
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)})
def draw():
nonlocal finished
app.draw(p)
if args.headless_smoke and p.frame_count >= args.frames-1 and not finished:
finished = True
if args.screenshot:
path = Path(args.screenshot).resolve()
path.parent.mkdir(parents=True, exist_ok=True)
p.screenshot(str(path), scale=2)
if args.state_output:
path = Path(args.state_output).resolve()
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(asdict(app.state), ensure_ascii=False, indent=2)+'\n', encoding='utf-8')
p.quit()
p.run(update, draw)
if __name__ == '__main__':
main()届いた返事を、通信待ちの状態へ戻す
書きかえない状態の履歴と、まとめて戻すこと。
Timeline に、今の State と過去の State を持たせ、X で1つ前の受け付けた判断に戻します。時刻や相手の位置だけでなく、すべての候補・まだ届いていない返事の記録・届いた返事の列・送った回数・終わりの状態まで、1つの State として保存します。
1 → Space で T2 の返事を受け取ってから X を押すと、T1 の通信待ちに戻ります。候補は K と S に戻り、届いた返事はなくなり、元の記録をもった pending が元に戻ります。そこからもう一度 Space を押すと同じ返事になり、送った回数が余分に増えることはありません。
advance が同じ State を返した操作は、履歴に入れません。BUSY、知らない操作、終わったあとの前進は、どれも受け付けません。X だけは終わりの判定より先に処理するので、WON や LATE からも考え直せます。R では、通信も履歴も最初に戻します。
最終版は、元のゲームの状況 A のふつうのルールについて、本当の出発点が S/K それぞれの場合に、12刻までのすべての正しい送信と待ちの並びを、元のゲームの model.py と直接照らし合わせる、任意のテストを用意しています。Undo は別の往復の並びで比べます。元のゲームが手元にない場合は、その比較だけを SKIP と表示し、教材自身のテストは動きます。画面やキーは簡単な版で、元のゲームとまったく同じ作品という意味ではありません。
下のコードは 05_undo.py の一部です。前のステップから追加・変更したところだけを抜き出しています。
if state == timeline.state:
return timeline, reason
return Timeline(state, timeline.history + (timeline.state,)), reason実行して確かめる
教材のZIPを展開したフォルダーで、次のコマンドを実行します。
.venv\Scripts\python.exe 05_undo.py.venv/bin/python 05_undo.py- まだ届いていない返事も Undo する
- 受け付けなかった操作は履歴に入れない
- 終わったあとも戻して考え直せる
State 全体を保存します。R → 3 → Space を5回押して T6・H・WON にしてから X を押すと、T5・T・RUNNING(続行中)に戻ります。次の Space で、また T6・WON です。別に、R → 1 → 2(BUSY)→ X なら、受け付けなかった操作は飛ばして、最初の T0 に戻ります。
このステップのコード全文(363行)
# MIT License
#
# Copyright (c) 2026
#
# 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.
#
# Third-party exception: bitmap glyph data derived from Noto Sans CJK is licensed
# under the SIL Open Font License 1.1. See assets/NOTO-LICENSE.txt.
"""返事の遅い無線・教材 5/5。Python 3.11+ / Pyxel 2.9.9。"""
COMMANDS = ('ask', 'safe', 'bridge', 'stop')
from dataclasses import dataclass, replace, asdict
from collections import deque
@dataclass(frozen=True)
class Level:
edges: tuple = ('SJ', 'KU', 'UJ', 'JM', 'MN', 'NH', 'JT', 'TH')
starts: tuple = ('S', 'K')
hidden_start: str = 'S'
opens: tuple = (4, 6)
initial_mode: str = 'join'
limit: int = 12
LEVEL = Level()
@dataclass(frozen=True)
class World:
origin: str
pos: str
mode: str
via: bool = False
blocked: int = 0
def route(level, start, target, safe=False):
"""幅優先探索。同じ長さなら地点名順。安全路はJTを除く。"""
if start == target:
return (start,)
queue = deque([(start,)])
seen = {start}
while queue:
path = queue.popleft()
near = sorted({
edge[1] if edge[0] == path[-1] else edge[0]
for edge in level.edges
if path[-1] in edge
and not (safe and set(edge) == {'J', 'T'})
})
for node in near:
if node in seen:
continue
next_path = path + (node,)
if node == target:
return next_path
queue.append(next_path)
seen.add(node)
return ()
def goal(world):
if world.mode == 'stop':
return '待機'
if world.mode == 'join':
return 'J'
return 'T→H' if world.mode == 'bridge' and world.via else 'H'
def instruct(world, command):
if world.pos == 'H' or command == 'ask':
return world
return replace(world, mode=command,
via=(command == 'bridge' and world.pos != 'T'))
def walk(world, level, tick):
if world.pos == 'H' or world.mode == 'stop':
return world
target = ('J' if world.mode == 'join' else
'T' if world.mode == 'bridge' and world.via else 'H')
path = route(level, world.pos, target, safe=world.mode == 'safe')
if len(path) < 2:
return world
if set((world.pos, path[1])) == {'J', 'T'} and tick not in level.opens:
return replace(world, blocked=world.blocked + 1)
return replace(world, pos=path[1], via=world.via and path[1] != 'T')
@dataclass(frozen=True)
class Pending:
command: str
sent: int
received: int
arrives: int
snapshots: tuple
@dataclass(frozen=True)
class Reply:
command: str
sent: int
received: int
arrived: int
sample: int
phase: str
pos: str
goal: str
def snapshot(world):
return (world.origin, world.pos, goal(world))
@dataclass(frozen=True)
class State:
t: int
actual: World
candidates: tuple
pending: object = None
replies: tuple = ()
sent: int = 0
outcome: str = 'running'
def initial(level=LEVEL):
worlds = tuple(World(node, node, level.initial_mode) for node in level.starts)
return State(0, next(w for w in worlds if w.origin == level.hidden_start), worlds)
def advance(state, action, level=LEVEL):
"""受信→偶数刻の移動→返信→終了。拒否は同じStateを返す。"""
if state.outcome != 'running':
return state, 'ended'
if action not in COMMANDS + ('wait',):
return state, 'unknown'
if action in COMMANDS and state.pending:
return state, 'busy'
tick = state.t + 1
actual, candidates = state.actual, state.candidates
pending, replies, sent = state.pending, state.replies, state.sent
if action in COMMANDS:
actual = instruct(actual, action)
candidates = tuple(instruct(w, action) for w in candidates)
pending = Pending(action, state.t, tick, tick + 1,
tuple(snapshot(w) for w in candidates))
sent += 1
if tick % 2 == 0:
actual = walk(actual, level, tick)
candidates = tuple(walk(w, level, tick) for w in candidates)
if pending and pending.arrives == tick:
pos = dest = ''
if pending.command == 'ask':
answer = next(row for row in pending.snapshots if row[0] == actual.origin)
pos, dest = answer[1:]
candidates = tuple(w for w in candidates if any(
row[0] == w.origin and row[1:] == answer[1:]
for row in pending.snapshots))
replies += (Reply(pending.command, pending.sent, pending.received,
tick, pending.received, '移動前', pos, dest),)
pending = None
outcome = ('won' if actual.pos == 'H' else
'late' if tick >= level.limit else 'running')
reason = 'rescued' if outcome == 'won' else 'deadline' if outcome == 'late' else 'advanced'
return State(tick, actual, candidates, pending, replies, sent, outcome), reason
def visible(state):
"""真の位置ではなく、同じ時計で進んだ候補だけを画面へ渡す。"""
return tuple(sorted({world.pos for world in state.candidates}))
@dataclass(frozen=True)
class Timeline:
state: State
history: tuple = ()
def apply(timeline, action, level=LEVEL):
if action == 'undo':
if not timeline.history:
return timeline, 'no_undo'
return Timeline(timeline.history[-1], timeline.history[:-1]), 'undo'
state, reason = advance(timeline.state, action, level)
if state == timeline.state:
return timeline, reason
return Timeline(state, timeline.history + (timeline.state,)), reason
XY = {'S': (36, 105), 'K': (36, 187), 'U': (94, 187),
'J': (159, 142), 'M': (159, 228), 'N': (215, 228),
'H': (274, 228), 'T': (274, 142)}
KEY_TO_ACTION = {'SPACE': 'wait', '1': 'ask', '2': 'safe', '3': 'bridge', '4': 'stop', 'X': 'undo'}
KEYS = tuple(KEY_TO_ACTION) + ('R',)
DEMO_KEYS = ('1', 'SPACE', 'X')
class App:
def __init__(self):
self.reset()
def reset(self):
self.timeline = Timeline(initial())
self.gated = False
self.message = 'READY'
@property
def state(self):
return self.timeline.state
def control(self, pressed, held):
# R is a fresh press; holding it does not reset every frame.
if 'R' in pressed:
self.reset()
self.gated = True
return
if self.gated:
if not pressed and not held:
self.gated = False
return
if len(pressed | held) > 1:
self.gated = True
self.message = 'RELEASE ALL KEYS'
return
if len(pressed) != 1:
return
action = KEY_TO_ACTION.get(next(iter(pressed)))
if action is None:
return
self.timeline, self.message = apply(self.timeline, action)
self.gated = True
def draw(self, p):
state = self.state
p.cls(1)
p.text(12, 12, '061 DELAYED RADIO / STAGE 5', 10)
p.text(12, 29, 'T '+str(state.t)+'/12 SENT '+str(state.sent)
+' '+state.outcome.upper(), 11)
p.text(12, 48, 'YELLOW RINGS = CURRENT CANDIDATES', 6)
p.rectb(12, 72, 288, 178, 5)
for edge in LEVEL.edges:
x, y = XY[edge[0]]
u, v = XY[edge[1]]
bridge = set(edge) == {'J', 'T'}
p.line(x, y, u, v, 10 if bridge else 5)
if bridge:
p.line(x, y+3, u, v+3, 10)
locations = visible(state)
for node, (x, y) in XY.items():
p.circ(x, y, 9, 3 if node == 'H' else 1)
p.circb(x, y, 9, 11 if node == 'H' else 6)
p.text(x-2, y-2, node, 7)
if node in locations:
p.circb(x, y, 14, 10)
p.text(180, 126, 'BRIDGE JT', 10)
p.text(310, 78, 'NEXT WALK: '+str(state.t+2-state.t%2), 7)
p.text(310, 94, 'BRIDGE AT: 4,6', 10)
p.text(310, 110, 'RINGS: '+','.join(locations), 6)
p.text(310, 135, '1 ASK: WHERE?', 7)
p.text(310, 150, '2 SAFE / 3 BRIDGE', 7)
p.text(310, 165, '4 STOP', 7)
p.text(310, 181, 'WORLDS: '+str(len(state.candidates)), 13)
p.text(310, 196, 'SPACE WAIT', 6)
p.text(310, 211, 'X UNDO', 6)
p.text(310, 226, 'R RESET / ESC QUIT', 6)
p.rect(12, 260, 456, 57, 2)
pending = state.pending
if pending:
line = ('SENT '+str(pending.sent)+' / RECEIVED '+str(pending.received)
+' / REPLY AT '+str(pending.arrives))
else:
line = 'RADIO READY: SEND NOW'
p.text(20, 269, line, 7)
if state.replies:
reply = state.replies[-1]
if reply.command == 'ask':
dest = {'待機': 'STOP', 'T→H': 'T>H'}.get(reply.goal, reply.goal)
line = ('SAMPLE T'+str(reply.sample)+' BEFORE WALK: '
+reply.pos+' -> '+dest)
else:
line = reply.command.upper()+' RECEIVED T'+str(reply.received)
p.text(20, 287, line, 11)
p.text(20, 303, 'REPLY ARRIVED T'+str(reply.arrived)+' / THIS IS OLD DATA', 13)
else:
p.text(20, 287, 'NO REPLY YET. WAIT WITH SPACE.', 6)
p.text(20, 303, 'A SECOND SEND IS REJECTED WHILE BUSY.', 13)
p.text(12, 328, self.message.upper(), 11)
def parse_args(argv=None):
"""すべてのCLI検証はPyxelをimportする前に終える。"""
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('--frames', type=int)
parser.add_argument('--screenshot')
parser.add_argument('--state-output')
args = parser.parse_args(argv)
if not args.headless_smoke and (args.demo_checkpoint or args.frames is not None
or args.screenshot or args.state_output):
parser.error('QA options require --headless-smoke')
minimum = 2 * len(DEMO_KEYS) + 4 if args.demo_checkpoint else 4
if args.frames is None:
args.frames = minimum
if not minimum <= args.frames <= 600:
parser.error('--frames must be between '+str(minimum)+' and 600')
for name, suffix in (('screenshot', '.png'), ('state_output', '.json')):
value = getattr(args, name)
if value is not None and (not value.strip() or Path(value).suffix.lower() != suffix):
parser.error('--'+name.replace('_', '-')+' requires a '+suffix+' file path')
return args
def main(argv=None):
args = parse_args(argv)
from pathlib import Path
import json
import pyxel as p
app = App()
p.init(480, 352, title='061 Delayed Radio', fps=30,
display_scale=2, headless=args.headless_smoke)
keys = {key: getattr(p, 'KEY_'+key) for key in KEYS}
finished = False
def update():
if args.headless_smoke:
index = p.frame_count // 2
pressed = ({DEMO_KEYS[index]} if args.demo_checkpoint
and p.frame_count % 2 == 0 and index < len(DEMO_KEYS) else set())
app.control(pressed, pressed)
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)})
def draw():
nonlocal finished
app.draw(p)
if args.headless_smoke and p.frame_count >= args.frames-1 and not finished:
finished = True
if args.screenshot:
path = Path(args.screenshot).resolve()
path.parent.mkdir(parents=True, exist_ok=True)
p.screenshot(str(path), scale=2)
if args.state_output:
path = Path(args.state_output).resolve()
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(asdict(app.state), ensure_ascii=False, indent=2)+'\n', encoding='utf-8')
p.quit()
p.run(update, draw)
if __name__ == '__main__':
main()仕組みがわかったら、元の完成版を読んでみよう。
元のゲームの状況 A です。最初の候補は S/K、本当は S で、最初は J に集まります。偶数の刻に1区間歩きます。橋 JT は4刻と6刻だけ開きます。送ってから1刻後に受け取り、2刻後に返事が届きます。質問は、受け取った直後・移動の前のようすを記録し、返事に合う候補を残します。12刻までに H に着けば成功です。
ここまでに作ったのは、元のゲームの仕組みだけを取り出して作り直した「簡略版」です。元のゲームのソースは手を加えずに、別のZIP(元の完成版ソース)に入れてあります。
| 比べる点 | 5ステップの教材 | 元の完成版 |
|---|---|---|
| 段階 | 01 は、出発点 S が分かっていて、安全な道に決めた版です。02 は、出発点 S が分かっていて指示だけ(返事待ちなし)。03 は、遅れて届く返事を加えますが、位置は見えます。04 と 05 で、元のゲームの状況 A のふつうのルールと、候補の推理を作ります。 | 最初から候補を表示し、A・B・C の決まった状況と、記録した時刻の比較を選べます。 |
| 操作 | 数字の1〜4で直接送ります。通信中に数字を押すと BUSY、Space で待ちます。Undo は 05 だけです。全部のキーを離すと、次の入力を受け付けます。 | 上下で文を選び、Z で送ります。返事を待っている間の Z は待ちです。表紙・一時停止・状況の選択などもあります。 |
| 表示と記録 | 英数字による 480×352 の画面です。動かすのに画像や文字のファイルはいりません。テスト用の出力には状態がふくまれるので、本当の答えのネタバレになります。音は使いません。 | 640×480 の日本語の文字の表示、一時停止・表紙、入力の意味の正確な記録と保存、比較版があります。 |
完成版のコードで、次に読むところ
- model.py: advance の受信→歩行→標本で候補選別→終了
- model.py: Pending / Reply / snapshot の異なる時刻
- model.py: step の通信を含むUndo
- model.py: forecast の公開情報
- app.py: gate とselectionの操作設計
- solver.py: 期限までの全合法列探索
最初は、数字かルールを1つだけ変えてみましょう。遊ぶ前に「どう変わるか」を予想しておくと、仕組みがよくわかります。慣れてきたら、見た目や音も足してみてください。元のファイルは残しておき、コピーした別のフォルダーで改造すると、いつでも元に戻せます。
教材をダウンロードする
無料で、登録なしでダウンロードできます。ZIPファイルは必ず展開してから、中のファイルを実行してください。元の完成版は、同じZIPに入っている assets フォルダー(画像などの素材)も一緒に使います。
コードと図形はMITライセンス、元の完成版に含まれる日本語の文字データ(Noto由来)はSIL OFL 1.1です。配ったり改造したりするときは、ZIPに入っているLICENSEなどの権利表示を残してください。出典・権利について
検証について
どこまで動作を確認したかを見る
2026年10月5日以降、Linux(Python 3.12・Pyxel 2.9.9)で、教材のコードとゲームのルールを自動テストしています。全500ステップの教材が、実際のPyxelで起動して画面を描けることも確かめました。元の完成版も、自動で遊ぶデモが起動して画面を描き、正常に終わることを確認しています。
この確認は、画面を表示しない自動テストで行ったものです。Windows・Macの実際のパソコンでの操作や、音が出るかどうかはまだ確認していません。人がすべての操作を試したわけでもありません。ブラウザー版の確認範囲は確認状況をご覧ください。
うまく動かないときは:起動やエラーの対処法 / 質問用のメモを作る
