完成サンプルをブラウザーで遊ぶ
この講座のもとになった完成版のゲームを、ブラウザーでそのまま遊べます。キーボードで操作するので、PCで開いてください（スマホのタッチ操作には対応していません）。初めて開くときは、読み込みに少し時間がかかります。
完成版の主な操作：左右キーで雲を動かす、上キーで日差し、下キーで雨、Zで1手戻す、Rでやり直し、Pで一時停止、Mで箱を選ぶ、Nで次の箱
ボタンを押すと、ゲームの読み込みが始まります。
画面に「CLICK TO START」と出たらクリックします。続けてゲーム画面を一度クリックすると、キー操作が効くようになります。ゲームを終えるときは、枠の外にある「閉じる」ボタンを使ってください（Escキーの働きはゲームごとに違います）。ゲーム内のセーブは、閉じる・ページを移る・再読み込みすると消えます。
うまく表示されないとき
ゲームを動かすためのプログラム（Pyxel 2.9.9）は、外部の配信サービス（jsDelivr）から読み込みます。表示されないときは「最初から再読込」を押すか、完成版のソースをダウンロードしてPCで実行してください。ブラウザー版の録画機能は止めています。ブラウザー版で確認できている範囲は検証状況にまとめています。
すべての水の行き先を、同じ古い水の量から決める。
いきなり完成版のコードを全部読む必要はありません。小さなゲームにルールを1つずつ足しながら、5つのステップで作っていきます。どのステップのファイルもそれだけで動くので、動きを確かめてから次へ進みましょう。
1. 1滴ずつの持ち主、水の合計を保つ決まり、書き換えない状態、受け付けない操作では状態を変えない
2. 縦の列を上から調べる、さえぎるもの、ノード（水の置き場所）の性質、条件しだいで通りぬけること、比べるための決まった配置
3. 古い配列と新しい配列を分ける（二重バッファ）、合流は足し算、古いKで決める水門、1刻みの区切り
- 必要な環境
- Python 3.11以上 / Pyxel 2.9.9
- 知っておくとよいこと
- Python 3.10以上、ターミナルと仮想環境 / 変数、if、for、関数、listとtuple / dataclass、replace、純粋関数（同じ入力なら必ず同じ結果を返し、ほかを書き換えない関数）は本文とコードで説明します
- 教材に入っているもの
- ステップごとに動くPythonファイル5本・説明・動作確認用のテスト
- 完成した教材の操作
- 01：上下キーで、Sと雲のあいだで1滴をやりとり ／ 02：左右キーで列を選ぶ、1・2・3で観察用の状態を表示。上下キーはSの列だけで使える ／ 03：1・2・3で実験の配置を選ぶ、Xで1刻み進める ／ 04・05：左右キーで雲を動かす、上キーで日差し、下キーで雨。05はZでUndo（元に戻す） ／ 共通：Rで最初の状態に戻す、Escapeで終了。操作キーを全部離してから、次のキーを押す
- 難易度の目安
- 中級このページで使うプログラムの考え方の数と難しさから決めた目安です。難易度について
- 学習順に迷ったら
- おすすめの学ぶ順番を見る
この講座の進め方① ページ下の教材ZIPをダウンロードして、すべて展開します。② はじめる準備を見ながら、PythonとPyxelを使える状態にします。③ まず 01_ownership.py を、何も変えずにそのまま実行してみましょう。記事に載せているコードは説明用の一部分なので、実行するときはZIPに入っているファイルを使ってください。
コードを変更したら
コードを書き換えたら、ファイルを保存し、動いているゲームを閉じてから、もう一度同じ.pyを実行します。ゲーム中にRキーなどで「やり直し」をしても、書き換えたコードは読み込まれません。
教材のゲーム画面は、日本語フォントを用意しなくても動くように英語で表示しています。ファイルには動作確認用のコードも入っていますが、最初は Game / State・update・draw の3つに注目すれば十分です。わからない言葉は用語集で調べられます。class・self・dataclass はPythonの状態、キー入力と時間の扱いは入力・更新・描画、マス目とピクセルの違いは画面とマップの座標のページで詳しく説明しています。
水の持ち主を交換する
1滴ずつの持ち主、水の合計を保つ決まり、書き換えない状態、受け付けない操作では状態を変えない。
箱Cの水は、合計3滴です。State.cloud は雲の水の量、State.water は S・F・K・gate の順に並べた水の量です。absorbed は、吸い込まれた水を記録する帳簿です。01では、雲をSの真上の x=1 に固定します。上キーでSから雲へ1滴、下キーで雲からSへ1滴を移します。水を作る処理と消す処理に分けることはしません。同じ新しい状態の中で、一方を1減らし、もう一方を1増やします。元の tuple（あとから変えられない並び）は変えずに list へコピーし、最後に tuple と replace で State を作ります。雲の容量は3です。空の雲で雨を降らせる操作と、満杯の雲に日差しを当てる操作は、accepted=False（受け付けない）になります。そのときは同じ State を返し、手数も増やしません。絵は、この帳簿を読むだけです。雲の移動、見通し、水の通り道、成功の判定は、まだありません。
下のコードは 01_ownership.py の一部です。前のステップから追加・変更したところだけを抜き出しています。
01_ownership.py / 注目する部分
def source_exchange(level, state, action):
"""01/02の実験: Sの真上だけで所有者を1滴ずつ交換する。"""
if action not in ('U','D'):
return Result(state,False,'unused')
if state.x != 1:
return Result(state,False,'inspect_only')
i = index(level,'S')
water = list(state.water)
if action == 'U':
if state.cloud == level.cloud_capacity:
return Result(state,False,'full')
if water[i] == 0:
return Result(state,False,'dry')
water[i] -= 1
cloud = state.cloud + 1
else:
if state.cloud == 0:
return Result(state,False,'empty')
water[i] += 1
cloud = state.cloud - 1
new = replace(state,cloud=cloud,water=tuple(water),moves=state.moves+1)
check(level,new)
return Result(new,True,'sun' if action=='U' else 'rain')実行して確かめる
教材のZIPを展開したフォルダーで、次のコマンドを実行します。
Windows
.venv\Scripts\python.exe 01_ownership.pymacOS / Linux
.venv/bin/python 01_ownership.py画面で見ること最初は Cloud 0、S 3、F・K・gate 0、Absorbed 0、TOTAL 3 です。上を3回押すと、Cloud 3・S 0・Moves 3 になります。4回目の上は FULL（満杯）と出て、水の量も手数も変わりません。下を1回押すと、Cloud 2・S 1・Moves 4 です。左右は UNUSED（使わない）と出て、雲は x=1 のままです。
- 雲と、すべてのノード（水の置き場所）と、吸収の帳簿の合計が3
- 元の場所から減らすことと、移動先に足すことを、同じ1回の更新で行う
- 受け付けない操作では、手数も変えない
- 描画では水を増やさない
つまずきやすいところ
cloud+=1 だけを書くと、水が増えてしまいます。描画した滴の数を別に水の合計へ足すのも、同じ水を2回数えること（二重計上）になります。absorbed は、あとで機能を広げるために書いてあります。このゲームのふつうのプレイでは、花は水を吸い込みません。この実験の source_exchange はS専用の処理です。あとで作る、盤面全体を扱う transition とは別のものです。少し変えてみよう
上→上→上→上→下と押して、最後が Cloud 2、S 1、Moves 4 になることを確かめましょう。tools/run_examples.py の01の欄でも、同じ結果を確かめています。雲が満杯で、しかもSが空のときは、元のゲームと同じ順番で調べるので、full（満杯）が dry（空）より先に出ることも確かめます。
01_ownership.pyをダウンロードこの段階だけで実行できます
このステップのコード全文（307行）
01_ownership.py / 全文
# -*- coding: utf-8 -*-
"""030 ひと箱の天気 / standalone tutorial / fictional discrete water model.
Original model: Pyxel 100 contributors. Teaching adaptation: 2026.
No external image/font asset is needed. Pyxel is imported only in main().
"""
# 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.
from dataclasses import dataclass, replace, asdict
import hashlib, json
STAGE = 1
STAGE_TITLE = 'WATER OWNERSHIP'
GUIDE = 'UP: S to cloud DOWN: cloud to S / column fixed at 1'
@dataclass(frozen=True)
class Node:
name:str
x:int
y:int
kind:str='dish'
capacity:int=0
target:str|None=None
alternate:str|None=None
catches:bool=True
shades:bool=False
@dataclass(frozen=True)
class Level:
title:str
nodes:tuple[Node,...]
start:tuple[int,...]
total:int
gate_linked:bool=True
width:int=7
cloud_capacity:int=3
@dataclass(frozen=True)
class State:
x:int
cloud:int
water:tuple[int,...]
absorbed:int=0 # Explicit accounting slot; authored soil holds water, never absorbs it.
moves:int=0
@dataclass(frozen=True)
class Result:
state:State
accepted:bool
reason:str
trace:tuple[tuple[int,...],...]=()
transfers:tuple[tuple[tuple[str,str,int],...],...]=()
# 原作Cの座標・容量・流路。列は0〜6、yは下へ増える。
BOARD = Level('C: keep one drop in K', (
Node('S',1,5,capacity=3),
Node('F',5,4,capacity=2,target='S'),
Node('K',3,5,capacity=1,target='S'),
Node('gate',5,2,'gate',target='F',alternate='S',shades=True),
), (3,0,0,0), 3)
def index(level,name):return next(i for i,n in enumerate(level.nodes) if n.name==name)
def amount(level,state,name):return state.water[index(level,name)]
def initial(level):return State(1,0,level.start)
def total(state):return state.cloud+sum(state.water)+state.absorbed
def check(level,state,stable=True):
assert len(state.water)==len(level.nodes)
assert 0<=state.x<level.width and 0<=state.cloud<=level.cloud_capacity
assert state.absorbed>=0 and all(v>=0 for v in state.water)
assert total(state)==level.total
if stable:
assert all(v<=n.capacity if n.kind=='dish' else v==0 for n,v in zip(level.nodes,state.water))
def state_hash(state):return hashlib.sha256(json.dumps(asdict(state),sort_keys=True).encode('utf-8')).hexdigest()
def source_exchange(level, state, action):
"""01/02の実験: Sの真上だけで所有者を1滴ずつ交換する。"""
if action not in ('U','D'):
return Result(state,False,'unused')
if state.x != 1:
return Result(state,False,'inspect_only')
i = index(level,'S')
water = list(state.water)
if action == 'U':
if state.cloud == level.cloud_capacity:
return Result(state,False,'full')
if water[i] == 0:
return Result(state,False,'dry')
water[i] -= 1
cloud = state.cloud + 1
else:
if state.cloud == 0:
return Result(state,False,'empty')
water[i] += 1
cloud = state.cloud - 1
new = replace(state,cloud=cloud,water=tuple(water),moves=state.moves+1)
check(level,new)
return Result(new,True,'sun' if action=='U' else 'rain')
CONTROLS = ('LEFT','RIGHT','UP','DOWN','X','Z','R','1','2','3')
REPLAY = {'L':'LEFT','R':'RIGHT','U':'UP','D':'DOWN','X':'X','Z':'Z',
'Q':'R','1':'1','2':'2','3':'3'}
class Controller:
def __init__(self):
self.lock = False
self.reset()
def reset(self):
self.level = BOARD
self.state = initial(BOARD)
self.history = []
self.trace = ()
self.transfers = ()
self.tick_count = 0
self.reason = 'ready'
self.preset = 1
# reset中にlockは消さない。R後も全キー解放が必要。
def tick(self, pressed=(), held=()):
pressed = set(pressed) & set(CONTROLS)
held = set(held) & set(CONTROLS)
if self.lock:
if not held:
self.lock = False
return
if not pressed and len(held) <= 1:
return
if len(pressed) != 1 or pressed != held:
self.lock = True
self.reason = 'chord_rejected'
return
self.lock = True
key = next(iter(pressed))
if key == 'R':
self.reset()
return
action={'UP':'U','DOWN':'D'}.get(key,'?')
r=source_exchange(self.level,self.state,action)
self.state=r.state;self.reason=r.reason
def snapshot(self):
sun, rain = preview(self.level,self.state) if STAGE >= 2 else (None,None)
return {'state':asdict(self.state),'hash':state_hash(self.state),
'total':total(self.state),'budget':self.level.total,
'gate_open':gate_open(self.level,self.state.water) if STAGE>=2 else False,
'won':won(self.level,self.state) if STAGE>=4 else False,
'sun':None if sun is None else self.level.nodes[sun].name,
'rain':None if rain is None else self.level.nodes[rain].name,
'event':self.reason,'history_depth':len(self.history),
'trace':self.trace,'transfers':self.transfers,
'ticks':self.tick_count,'preset':self.preset,'lock':self.lock}
def draw_board(p,c):
"""描画は状態を読むだけ。水滴の絵を別資源として加算しない。"""
s=c.state; level=c.level; snap=c.snapshot()
p.cls(1)
p.text(14,12,'BOX WEATHER / STEP '+str(STAGE)+' / '+STAGE_TITLE,10)
p.text(14,27,'Fictional integer water model / no automatic evaporation / silent',6)
ox,oy,dx,dy=47,105,54,42
def xy(n): return ox+n.x*dx,oy+n.y*dy
def arrow(x,y,xx,yy,color):
p.line(x,y,xx,yy,color)
if abs(xx-x)>abs(yy-y):
sign=1 if xx>x else -1
p.line(xx,yy,xx-sign*5,yy-3,color);p.line(xx,yy,xx-sign*5,yy+3,color)
else:
sign=1 if yy>y else -1
p.line(xx,yy,xx-3,yy-sign*5,color);p.line(xx,yy,xx+3,yy-sign*5,color)
p.rect(18,47,402,313,0);p.rectb(18,47,402,313,5)
for x in range(7):
p.text(ox+x*dx-2,53,str(x),13)
for y in range(110,333,10): p.pset(ox+x*dx,y,1)
for n in level.nodes:
if n.target:
dst=level.nodes[index(level,n.target)]
x,y=xy(n);xx,yy=xy(dst)
if n.kind!='dish':
color=(11 if snap['gate_open'] else 5) if n.kind=='gate' else 6
arrow(x,y+12,xx,yy-15,color)
if n.kind=='gate':
x,y=xy(n);color=5 if snap['gate_open'] else 6
arrow(x+25,y,401,y,color);arrow(401,y,401,346,color)
src=level.nodes[index(level,'S')];sx,sy=xy(src)
arrow(401,346,sx,346,6);arrow(sx,346,sx,sy+13,6)
p.text(190,349,'RETURN TO S',6)
for i,n in enumerate(level.nodes):
x,y=xy(n);value=s.water[i]
if n.kind=='dish':
p.rect(x-21,y-9,43,19,4);p.rect(x-18,y-10,37,14,1)
for j in range(n.capacity):
if j<value:p.circ(x-12+j*9,y-2,3,12)
else:p.circb(x-12+j*9,y-2,3,5)
p.text(x-22,y+16,f'{n.name}: {value}/{n.capacity}',7)
if value>n.capacity: p.text(x-22,y+27,'EXCESS '+str(value-n.capacity),8)
if n.name=='F':
p.line(x,y-12,x,y-28,11);p.circ(x,y-30,5,10 if snap['won'] else 3)
else:
color=11 if n.kind=='gate' and snap['gate_open'] else 9
p.rectb(x-23,y-10,47,20,color)
p.text(x-20,y-3,n.name,7)
p.text(x-21,y+15,'WATER '+str(value),12)
if n.kind=='gate': p.text(x-20,y-21,'OPEN' if snap['gate_open'] else 'CLOSED',color)
if n.name==snap['rain']:p.rectb(x-26,y-14,53,29,12)
if n.name==snap['sun']:p.rectb(x-24,y-12,49,25,10)
cx=ox+s.x*dx
p.elli(cx-24,76,49,20,7);p.circ(cx-10,76,11,7);p.circ(cx+9,73,12,7)
for j in range(level.cloud_capacity):
if j<s.cloud:p.circ(cx-10+j*10,86,3,12)
else:p.circb(cx-10+j*10,86,3,5)
p.rect(437,47,269,313,0)
lines=[('WATER OWNERS',10),('Cloud: '+str(s.cloud)+' / '+str(level.cloud_capacity),12)]
lines += [(n.name+': '+str(s.water[i]),7) for i,n in enumerate(level.nodes)]
lines += [('Absorbed ledger: '+str(s.absorbed),6),
('TOTAL '+str(total(s))+' = BUDGET '+str(level.total),11),
('Moves: '+str(s.moves)+' ticks: '+str(c.tick_count),7),
('History: '+str(len(c.history)),7)]
y=60
for line,color in lines:p.text(450,y,line,color);y+=17
p.text(450,275,'SUN : '+str(snap['sun']),10)
p.text(450,292,'RAIN: '+str(snap['rain']),12)
p.text(450,310,'CASE '+str(c.preset)+' / '+level.title,6)
p.text(450,332,'WON' if snap['won'] else 'OBSERVE / TRY',10)
p.text(14,376,'EVENT: '+c.reason,10)
p.text(14,394,GUIDE,7)
p.text(14,413,'R: reset ESC: quit / release ALL controls between presses',6)
if c.transfers:
p.text(14,433,'LAST FLOW: '+str(c.transfers[-1]),12)
else:p.text(14,433,'Blue=rain target Yellow=sun target / x=0..6',13)
p.text(14,458,'A flow tick sends at most one unit per emitting node. No water is deleted.',6)
p.text(14,476,'01-03 are labs. 04-05 use original C rules. This is NOT physical hydrology.',13)
def main(headless=False,sequence='',screenshot=None,report=None):
from pathlib import Path
import time
import pyxel # モデルのimportやstdlibテストにPyxelは不要。
c=Controller();codes={n:getattr(pyxel,'KEY_'+n) for n in CONTROLS}
frame=0;events=[];last=max(12,2*len(sequence)+2);start=time.perf_counter()
def update():
nonlocal frame
frame+=1
if headless:
i=(frame-1)//2
if frame%2 and i<len(sequence):
k=REPLAY[sequence[i]];c.tick((k,),(k,))
events.append({'input':sequence[i],'snapshot':c.snapshot()})
else:c.tick()
else:c.tick([n for n,k in codes.items() if pyxel.btnp(k)],
[n for n,k in codes.items() if pyxel.btn(k)])
def draw():
draw_board(pyxel,c)
if headless and frame>=last:
if screenshot:pyxel.screenshot(str(Path(screenshot).resolve()),scale=1)
if report:
Path(report).write_text(json.dumps({
'label':'actual Pyxel headless / synthetic controller input / silent',
'stage':STAGE,'frames':frame,'logical_seconds':frame/30,
'wall_seconds_before_exit':round(time.perf_counter()-start,6),
'program_sha256':hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
'sequence':sequence,'events':events,'snapshot':c.snapshot()
},ensure_ascii=False,indent=2)+'\n',encoding='utf-8')
pyxel.quit()
pyxel.init(720,500,title='Box Weather - Step '+str(STAGE),fps=30,headless=headless)
pyxel.run(update,draw)
if __name__=='__main__':
import argparse
parser=argparse.ArgumentParser(description=__doc__)
group=parser.add_mutually_exclusive_group()
group.add_argument('--headless-smoke',action='store_true')
group.add_argument('--headless-replay',help='L/R/U/D arrows; X/Z; Q reset; 1/2/3 presets')
parser.add_argument('--screenshot');parser.add_argument('--report')
args=parser.parse_args()
headless=args.headless_smoke or args.headless_replay is not None
if not headless and (args.screenshot or args.report):parser.error('Outputs require headless mode')
seq=args.headless_replay or ''
if any(k not in REPLAY for k in seq):parser.error('Unsupported replay character')
main(headless,seq,args.screenshot,args.report)雨と日差しの最初の相手を探す
縦の列を上から調べる、さえぎるもの、ノードの性質、条件しだいで通りぬけること、比べるための決まった配置。
左右キーで、雲を0〜6列に動かします。visible は、雲と同じ x にあるノードだけを見て、y がいちばん小さい（いちばん上の）ものを選びます。雨は catches=True（雨を受ける）のノードを、日差しは dish（皿）か shades=True（日差しをさえぎる）のノードを対象にします。開いた門は、どちらの候補からも外します。そのため、門の下の花Fが見えます。閉じた門は、雨を受け止め、日差しをさえぎります。細い管（細管）は、ふつう catches=False で shades=False なので、透明です。雨の候補がないときは、preview がSを戻り先にします。02の1・2・3は、元のゲームの3つの状態（最初、UUURRDのあと、UUURRDRRDのあと。Uは上、Rは右、Dは下）を読みこんで見るための、観察用のキーです。ゲームの途中で一気に移動する機能ではありません。水を上げ下げできるのは、前と同じくSの列だけです。
下のコードは 02_visibility.py の一部です。前のステップから追加・変更したところだけを抜き出しています。
02_visibility.py / 注目する部分
def visible(level,state,sun=False):
candidates=[]
for i,n in enumerate(level.nodes):
if n.x!=state.x:continue
if n.kind=='gate' and gate_open(level,state.water):continue
if (sun and (n.kind=='dish' or n.shades)) or (not sun and n.catches):candidates.append((n.y,n.name,i))
return min(candidates)[2] if candidates else None実行して確かめる
教材のZIPを展開したフォルダーで、次のコマンドを実行します。
Windows
.venv\Scripts\python.exe 02_visibility.pymacOS / Linux
.venv/bin/python 02_visibility.py画面で見ること1→右4回で x=5 になり、SUN と RAIN はどちらも gate、門は CLOSED（閉）です。2→右2回で x=5 になり、K 1 なので門は OPEN（開）、SUN と RAIN は F です。3では、Cloud 1、F 1、K 1、x=5 の途中の状態を表示します。この段階では上を押しても INSPECT_ONLY（見るだけ）と出て、水は動きません。
- 同じ列で、y がいちばん小さいものを見る
- 日差しと雨では、対象になる条件がちがう
- 開いた門は候補から外す
- 雨を受けるものがないときは、Sへ戻す
つまずきやすいところいちばん近いノードを、どれも同じように「さえぎるもの」として扱わないでください。また元のゲームでは、雨を受けるノードが同じ x,y に重なる面を、
validate_level で受け付けません。ふつうの候補は (y,name,index) の min（いちばん小さいもの）で選びます。同じ y に、雨は受けずに日差しだけさえぎるものなどがあれば、名前の順で決まります。列のはしでさらに動こうとすると edge（はし）になります。何もない列では、日差しの候補はなく、雨の戻り先はSです。少し変えてみよう
1→右4回と、2→右2回を比べて、門が閉じたときと開いたときの SUN と RAIN の名前を書きとめましょう。R→左で何もない列 x=0 に移り、SUN None・RAIN S になることを確かめます。R はリセットのキーで、右矢印とは別です。テストの pipe_is_transparent は、透明な細管をSの上に加えた、境目を調べる例です。
02_visibility.pyをダウンロードこの段階だけで実行できます
このステップのコード全文（335行）
02_visibility.py / 全文
# -*- coding: utf-8 -*-
"""030 ひと箱の天気 / standalone tutorial / fictional discrete water model.
Original model: Pyxel 100 contributors. Teaching adaptation: 2026.
No external image/font asset is needed. Pyxel is imported only in main().
"""
# 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.
from dataclasses import dataclass, replace, asdict
import hashlib, json
STAGE = 2
STAGE_TITLE = 'VERTICAL VISIBILITY'
GUIDE = 'LEFT/RIGHT: cloud 1/2/3: presets UP/DOWN: S column only'
@dataclass(frozen=True)
class Node:
name:str
x:int
y:int
kind:str='dish'
capacity:int=0
target:str|None=None
alternate:str|None=None
catches:bool=True
shades:bool=False
@dataclass(frozen=True)
class Level:
title:str
nodes:tuple[Node,...]
start:tuple[int,...]
total:int
gate_linked:bool=True
width:int=7
cloud_capacity:int=3
@dataclass(frozen=True)
class State:
x:int
cloud:int
water:tuple[int,...]
absorbed:int=0 # Explicit accounting slot; authored soil holds water, never absorbs it.
moves:int=0
@dataclass(frozen=True)
class Result:
state:State
accepted:bool
reason:str
trace:tuple[tuple[int,...],...]=()
transfers:tuple[tuple[tuple[str,str,int],...],...]=()
# 原作Cの座標・容量・流路。列は0〜6、yは下へ増える。
BOARD = Level('C: keep one drop in K', (
Node('S',1,5,capacity=3),
Node('F',5,4,capacity=2,target='S'),
Node('K',3,5,capacity=1,target='S'),
Node('gate',5,2,'gate',target='F',alternate='S',shades=True),
), (3,0,0,0), 3)
def index(level,name):return next(i for i,n in enumerate(level.nodes) if n.name==name)
def amount(level,state,name):return state.water[index(level,name)]
def initial(level):return State(1,0,level.start)
def total(state):return state.cloud+sum(state.water)+state.absorbed
def check(level,state,stable=True):
assert len(state.water)==len(level.nodes)
assert 0<=state.x<level.width and 0<=state.cloud<=level.cloud_capacity
assert state.absorbed>=0 and all(v>=0 for v in state.water)
assert total(state)==level.total
if stable:
assert all(v<=n.capacity if n.kind=='dish' else v==0 for n,v in zip(level.nodes,state.water))
def state_hash(state):return hashlib.sha256(json.dumps(asdict(state),sort_keys=True).encode('utf-8')).hexdigest()
def source_exchange(level, state, action):
"""01/02の実験: Sの真上だけで所有者を1滴ずつ交換する。"""
if action not in ('U','D'):
return Result(state,False,'unused')
if state.x != 1:
return Result(state,False,'inspect_only')
i = index(level,'S')
water = list(state.water)
if action == 'U':
if state.cloud == level.cloud_capacity:
return Result(state,False,'full')
if water[i] == 0:
return Result(state,False,'dry')
water[i] -= 1
cloud = state.cloud + 1
else:
if state.cloud == 0:
return Result(state,False,'empty')
water[i] += 1
cloud = state.cloud - 1
new = replace(state,cloud=cloud,water=tuple(water),moves=state.moves+1)
check(level,new)
return Result(new,True,'sun' if action=='U' else 'rain')
def gate_open(level,water):
return not level.gate_linked or not any(n.name=='K' for n in level.nodes) or water[index(level,'K')]>=1
def visible(level,state,sun=False):
candidates=[]
for i,n in enumerate(level.nodes):
if n.x!=state.x:continue
if n.kind=='gate' and gate_open(level,state.water):continue
if (sun and (n.kind=='dish' or n.shades)) or (not sun and n.catches):candidates.append((n.y,n.name,i))
return min(candidates)[2] if candidates else None
def preview(level,state):
sun=visible(level,state,True);rain=visible(level,state)
rain=index(level,'S') if rain is None else rain
return sun,rain
def survey_state(number):
# 原作の到達状態: 初期 / UUURRD / UUURRDRRD。
return (initial(BOARD), State(3,2,(0,0,1,0),moves=6),
State(5,1,(0,1,1,0),moves=9))[number-1]
CONTROLS = ('LEFT','RIGHT','UP','DOWN','X','Z','R','1','2','3')
REPLAY = {'L':'LEFT','R':'RIGHT','U':'UP','D':'DOWN','X':'X','Z':'Z',
'Q':'R','1':'1','2':'2','3':'3'}
class Controller:
def __init__(self):
self.lock = False
self.reset()
def reset(self):
self.level = BOARD
self.state = initial(BOARD)
self.history = []
self.trace = ()
self.transfers = ()
self.tick_count = 0
self.reason = 'ready'
self.preset = 1
# reset中にlockは消さない。R後も全キー解放が必要。
def tick(self, pressed=(), held=()):
pressed = set(pressed) & set(CONTROLS)
held = set(held) & set(CONTROLS)
if self.lock:
if not held:
self.lock = False
return
if not pressed and len(held) <= 1:
return
if len(pressed) != 1 or pressed != held:
self.lock = True
self.reason = 'chord_rejected'
return
self.lock = True
key = next(iter(pressed))
if key == 'R':
self.reset()
return
if key in ('1','2','3'):
self.preset=int(key);self.state=survey_state(self.preset);self.reason='preset';return
if key in ('LEFT','RIGHT'):
x=self.state.x+(-1 if key=='LEFT' else 1)
if 0<=x<self.level.width:
self.state=replace(self.state,x=x,moves=self.state.moves+1);self.reason='move'
else:self.reason='edge'
return
r=source_exchange(self.level,self.state,{'UP':'U','DOWN':'D'}.get(key,'?'))
self.state=r.state;self.reason=r.reason
def snapshot(self):
sun, rain = preview(self.level,self.state) if STAGE >= 2 else (None,None)
return {'state':asdict(self.state),'hash':state_hash(self.state),
'total':total(self.state),'budget':self.level.total,
'gate_open':gate_open(self.level,self.state.water) if STAGE>=2 else False,
'won':won(self.level,self.state) if STAGE>=4 else False,
'sun':None if sun is None else self.level.nodes[sun].name,
'rain':None if rain is None else self.level.nodes[rain].name,
'event':self.reason,'history_depth':len(self.history),
'trace':self.trace,'transfers':self.transfers,
'ticks':self.tick_count,'preset':self.preset,'lock':self.lock}
def draw_board(p,c):
"""描画は状態を読むだけ。水滴の絵を別資源として加算しない。"""
s=c.state; level=c.level; snap=c.snapshot()
p.cls(1)
p.text(14,12,'BOX WEATHER / STEP '+str(STAGE)+' / '+STAGE_TITLE,10)
p.text(14,27,'Fictional integer water model / no automatic evaporation / silent',6)
ox,oy,dx,dy=47,105,54,42
def xy(n): return ox+n.x*dx,oy+n.y*dy
def arrow(x,y,xx,yy,color):
p.line(x,y,xx,yy,color)
if abs(xx-x)>abs(yy-y):
sign=1 if xx>x else -1
p.line(xx,yy,xx-sign*5,yy-3,color);p.line(xx,yy,xx-sign*5,yy+3,color)
else:
sign=1 if yy>y else -1
p.line(xx,yy,xx-3,yy-sign*5,color);p.line(xx,yy,xx+3,yy-sign*5,color)
p.rect(18,47,402,313,0);p.rectb(18,47,402,313,5)
for x in range(7):
p.text(ox+x*dx-2,53,str(x),13)
for y in range(110,333,10): p.pset(ox+x*dx,y,1)
for n in level.nodes:
if n.target:
dst=level.nodes[index(level,n.target)]
x,y=xy(n);xx,yy=xy(dst)
if n.kind!='dish':
color=(11 if snap['gate_open'] else 5) if n.kind=='gate' else 6
arrow(x,y+12,xx,yy-15,color)
if n.kind=='gate':
x,y=xy(n);color=5 if snap['gate_open'] else 6
arrow(x+25,y,401,y,color);arrow(401,y,401,346,color)
src=level.nodes[index(level,'S')];sx,sy=xy(src)
arrow(401,346,sx,346,6);arrow(sx,346,sx,sy+13,6)
p.text(190,349,'RETURN TO S',6)
for i,n in enumerate(level.nodes):
x,y=xy(n);value=s.water[i]
if n.kind=='dish':
p.rect(x-21,y-9,43,19,4);p.rect(x-18,y-10,37,14,1)
for j in range(n.capacity):
if j<value:p.circ(x-12+j*9,y-2,3,12)
else:p.circb(x-12+j*9,y-2,3,5)
p.text(x-22,y+16,f'{n.name}: {value}/{n.capacity}',7)
if value>n.capacity: p.text(x-22,y+27,'EXCESS '+str(value-n.capacity),8)
if n.name=='F':
p.line(x,y-12,x,y-28,11);p.circ(x,y-30,5,10 if snap['won'] else 3)
else:
color=11 if n.kind=='gate' and snap['gate_open'] else 9
p.rectb(x-23,y-10,47,20,color)
p.text(x-20,y-3,n.name,7)
p.text(x-21,y+15,'WATER '+str(value),12)
if n.kind=='gate': p.text(x-20,y-21,'OPEN' if snap['gate_open'] else 'CLOSED',color)
if n.name==snap['rain']:p.rectb(x-26,y-14,53,29,12)
if n.name==snap['sun']:p.rectb(x-24,y-12,49,25,10)
cx=ox+s.x*dx
p.elli(cx-24,76,49,20,7);p.circ(cx-10,76,11,7);p.circ(cx+9,73,12,7)
for j in range(level.cloud_capacity):
if j<s.cloud:p.circ(cx-10+j*10,86,3,12)
else:p.circb(cx-10+j*10,86,3,5)
p.rect(437,47,269,313,0)
lines=[('WATER OWNERS',10),('Cloud: '+str(s.cloud)+' / '+str(level.cloud_capacity),12)]
lines += [(n.name+': '+str(s.water[i]),7) for i,n in enumerate(level.nodes)]
lines += [('Absorbed ledger: '+str(s.absorbed),6),
('TOTAL '+str(total(s))+' = BUDGET '+str(level.total),11),
('Moves: '+str(s.moves)+' ticks: '+str(c.tick_count),7),
('History: '+str(len(c.history)),7)]
y=60
for line,color in lines:p.text(450,y,line,color);y+=17
p.text(450,275,'SUN : '+str(snap['sun']),10)
p.text(450,292,'RAIN: '+str(snap['rain']),12)
p.text(450,310,'CASE '+str(c.preset)+' / '+level.title,6)
p.text(450,332,'WON' if snap['won'] else 'OBSERVE / TRY',10)
p.text(14,376,'EVENT: '+c.reason,10)
p.text(14,394,GUIDE,7)
p.text(14,413,'R: reset ESC: quit / release ALL controls between presses',6)
if c.transfers:
p.text(14,433,'LAST FLOW: '+str(c.transfers[-1]),12)
else:p.text(14,433,'Blue=rain target Yellow=sun target / x=0..6',13)
p.text(14,458,'A flow tick sends at most one unit per emitting node. No water is deleted.',6)
p.text(14,476,'01-03 are labs. 04-05 use original C rules. This is NOT physical hydrology.',13)
def main(headless=False,sequence='',screenshot=None,report=None):
from pathlib import Path
import time
import pyxel # モデルのimportやstdlibテストにPyxelは不要。
c=Controller();codes={n:getattr(pyxel,'KEY_'+n) for n in CONTROLS}
frame=0;events=[];last=max(12,2*len(sequence)+2);start=time.perf_counter()
def update():
nonlocal frame
frame+=1
if headless:
i=(frame-1)//2
if frame%2 and i<len(sequence):
k=REPLAY[sequence[i]];c.tick((k,),(k,))
events.append({'input':sequence[i],'snapshot':c.snapshot()})
else:c.tick()
else:c.tick([n for n,k in codes.items() if pyxel.btnp(k)],
[n for n,k in codes.items() if pyxel.btn(k)])
def draw():
draw_board(pyxel,c)
if headless and frame>=last:
if screenshot:pyxel.screenshot(str(Path(screenshot).resolve()),scale=1)
if report:
Path(report).write_text(json.dumps({
'label':'actual Pyxel headless / synthetic controller input / silent',
'stage':STAGE,'frames':frame,'logical_seconds':frame/30,
'wall_seconds_before_exit':round(time.perf_counter()-start,6),
'program_sha256':hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
'sequence':sequence,'events':events,'snapshot':c.snapshot()
},ensure_ascii=False,indent=2)+'\n',encoding='utf-8')
pyxel.quit()
pyxel.init(720,500,title='Box Weather - Step '+str(STAGE),fps=30,headless=headless)
pyxel.run(update,draw)
if __name__=='__main__':
import argparse
parser=argparse.ArgumentParser(description=__doc__)
group=parser.add_mutually_exclusive_group()
group.add_argument('--headless-smoke',action='store_true')
group.add_argument('--headless-replay',help='L/R/U/D arrows; X/Z; Q reset; 1/2/3 presets')
parser.add_argument('--screenshot');parser.add_argument('--report')
args=parser.parse_args()
headless=args.headless_smoke or args.headless_replay is not None
if not headless and (args.screenshot or args.report):parser.error('Outputs require headless mode')
seq=args.headless_replay or ''
if any(k not in REPLAY for k in seq):parser.error('Unsupported replay character')
main(headless,seq,args.screenshot,args.report)旧状態から流れを一斉に決める
古い配列と新しい配列を分ける（二重バッファ）、合流は足し算、古いKで決める水門、1刻みの区切り。
03は、X を押すと microstep（水を1刻みだけ流す関数）を1回だけ進める実験台です。old（古い配列）は読むだけにして、new=list(old) で作った新しい配列に、変化の分を足していきます。皿は、old の量が capacity（容量）を超えるときだけ、1滴を流し出します。皿以外は、old の量が0を超えるときだけ、1滴を流し出します。同じ通り道に着いたばかりの滴は、その刻みでは先へ進みません。いくつもの流れが入ってきたら、全部を足します。そのため、Fが一時的に3/2（容量2に3滴）になるのは正しい状態です。次の刻みに、あふれた1滴をSへ返します。門の行き先も、old のKを見て決めます。同じ刻みにKへ1滴が入っても、その刻みの門の水は、門が閉じているときの行き先Sへ進みます。moves はプレイヤーの行動の数なので、刻みの数とは区別します。03では moves は0のままで、ticks だけを増やします。
下のコードは 03_flow.py の一部です。前のステップから追加・変更したところだけを抜き出しています。
03_flow.py / 注目する部分
def microstep(level,old,order=None):
"""All intents use old; merge additions in new. One unit per emitting node.
Gates inspect old K, not the partially written new buffer. Converging inflows
sum without priority: excess remains and overflows on the following tick.
"""
new=list(old);moves=[];order=range(len(old)) if order is None else tuple(order)
if sorted(order)!=list(range(len(old))):raise ValueError('scan must visit each node once')
for i in order:
n=level.nodes[i]
if old[i]<= (n.capacity if n.kind=='dish' else 0):continue
target=n.target if n.kind!='gate' or gate_open(level,old) else n.alternate
if target is None:raise ValueError('unrouted overflow')
j=index(level,target);new[i]-=1;new[j]+=1;moves.append((n.name,target,1))
assert sum(new)==sum(old) and all(v>=0 for v in new)
return tuple(new),tuple(sorted(moves))実行して確かめる
教材のZIPを展開したフォルダーで、次のコマンドを実行します。
Windows
.venv\Scripts\python.exe 03_flow.pymacOS / Linux
.venv/bin/python 03_flow.py画面で見ること1は、元のゲームの面Cの gate に3滴を置いた観察用の配置です。X を押すごとに、(0,0,0,3)→(1,0,0,2)→(2,0,0,1)→(3,0,0,0) と変わります。2→X は、S・F・K・feed・gate の順で (1,0,0,1,1)→(2,0,1,0,0) です。更新のあと、門は OPEN と描かれます。しかし、運ばれた水は、FではなくSに入っています。3→X は、S・F・a・b の順で (0,1,1,1)→(0,3,0,0)、もう一度 X で (1,2,0,0) です。3は、雲の1滴も入れて合計4滴です。
- 流し出すかどうかは、必ず
oldを見て決める - 1つのノードから流れるのは、1刻みに1滴
- 合流した水は全部足し、あふれた分は次の刻みで流す
- 更新のあとで門が開いても、もう決めた行き先は変えない
つまずきやすいところ門のためのKの量を
new から読むと、ノードを調べる順番によって結果が変わってしまいます。水が入ってくるときにFの容量を見て、先に着いた滴だけを受け取ると、ほかの滴が消えてしまいます。03の1は観察用の配置です。ふつうに面Cを遊んでも、gate に同時に3滴がたまることはありません。2と3は、元のゲームのテストをもとにした、ノードを追加した実験用の面です。花の成功の判定は、ここではまだしません。少し変えてみよう
2→X で1回進めてみましょう。K 1 なのに F 0 になる理由を、古い状態から説明してみてください。3→X→X では、途中の F 3/2 を飛ばさずに確かめます。test_stages.py は、門と合流について、ノードを調べるすべての順番で結果を照らし合わせます。order（調べる順番）を逆にしても、水の量と、並べ替えた移動の記録（転送イベント）が同じになることを確かめましょう。
03_flow.pyをダウンロードこの段階だけで実行できます
このステップのコード全文（346行）
03_flow.py / 全文
# -*- coding: utf-8 -*-
"""030 ひと箱の天気 / standalone tutorial / fictional discrete water model.
Original model: Pyxel 100 contributors. Teaching adaptation: 2026.
No external image/font asset is needed. Pyxel is imported only in main().
"""
# 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.
from dataclasses import dataclass, replace, asdict
import hashlib, json
STAGE = 3
STAGE_TITLE = 'OLD-STATE FLOW'
GUIDE = '1: closed gate 2: old K 3: merge X: one flow tick'
@dataclass(frozen=True)
class Node:
name:str
x:int
y:int
kind:str='dish'
capacity:int=0
target:str|None=None
alternate:str|None=None
catches:bool=True
shades:bool=False
@dataclass(frozen=True)
class Level:
title:str
nodes:tuple[Node,...]
start:tuple[int,...]
total:int
gate_linked:bool=True
width:int=7
cloud_capacity:int=3
@dataclass(frozen=True)
class State:
x:int
cloud:int
water:tuple[int,...]
absorbed:int=0 # Explicit accounting slot; authored soil holds water, never absorbs it.
moves:int=0
@dataclass(frozen=True)
class Result:
state:State
accepted:bool
reason:str
trace:tuple[tuple[int,...],...]=()
transfers:tuple[tuple[tuple[str,str,int],...],...]=()
# 原作Cの座標・容量・流路。列は0〜6、yは下へ増える。
BOARD = Level('C: keep one drop in K', (
Node('S',1,5,capacity=3),
Node('F',5,4,capacity=2,target='S'),
Node('K',3,5,capacity=1,target='S'),
Node('gate',5,2,'gate',target='F',alternate='S',shades=True),
), (3,0,0,0), 3)
def index(level,name):return next(i for i,n in enumerate(level.nodes) if n.name==name)
def amount(level,state,name):return state.water[index(level,name)]
def initial(level):return State(1,0,level.start)
def total(state):return state.cloud+sum(state.water)+state.absorbed
def check(level,state,stable=True):
assert len(state.water)==len(level.nodes)
assert 0<=state.x<level.width and 0<=state.cloud<=level.cloud_capacity
assert state.absorbed>=0 and all(v>=0 for v in state.water)
assert total(state)==level.total
if stable:
assert all(v<=n.capacity if n.kind=='dish' else v==0 for n,v in zip(level.nodes,state.water))
def state_hash(state):return hashlib.sha256(json.dumps(asdict(state),sort_keys=True).encode('utf-8')).hexdigest()
def gate_open(level,water):
return not level.gate_linked or not any(n.name=='K' for n in level.nodes) or water[index(level,'K')]>=1
def visible(level,state,sun=False):
candidates=[]
for i,n in enumerate(level.nodes):
if n.x!=state.x:continue
if n.kind=='gate' and gate_open(level,state.water):continue
if (sun and (n.kind=='dish' or n.shades)) or (not sun and n.catches):candidates.append((n.y,n.name,i))
return min(candidates)[2] if candidates else None
def preview(level,state):
sun=visible(level,state,True);rain=visible(level,state)
rain=index(level,'S') if rain is None else rain
return sun,rain
def microstep(level,old,order=None):
"""All intents use old; merge additions in new. One unit per emitting node.
Gates inspect old K, not the partially written new buffer. Converging inflows
sum without priority: excess remains and overflows on the following tick.
"""
new=list(old);moves=[];order=range(len(old)) if order is None else tuple(order)
if sorted(order)!=list(range(len(old))):raise ValueError('scan must visit each node once')
for i in order:
n=level.nodes[i]
if old[i]<= (n.capacity if n.kind=='dish' else 0):continue
target=n.target if n.kind!='gate' or gate_open(level,old) else n.alternate
if target is None:raise ValueError('unrouted overflow')
j=index(level,target);new[i]-=1;new[j]+=1;moves.append((n.name,target,1))
assert sum(new)==sum(old) and all(v>=0 for v in new)
return tuple(new),tuple(sorted(moves))
def lab(number):
"""原作test_model.pyの意味を保つ、旧状態と合流の観察配置。"""
if number == 1:
return BOARD, State(5,0,(0,0,0,3))
if number == 2:
level = Level('old K test',(
Node('S',0,5,capacity=3),Node('F',5,4,capacity=2,target='S'),
Node('K',3,5,capacity=1,target='S'),
Node('feed',3,2,'pipe',target='K',catches=False),
Node('gate',5,2,'gate',target='F',alternate='S',shades=True),
),(3,0,0,0,0),3)
return level, State(1,0,(1,0,0,1,1))
if number == 3:
level = Level('merge then overflow',(
Node('S',0,5,capacity=4),Node('F',5,4,capacity=2,target='S'),
Node('a',1,1,'pipe',target='F',catches=False),
Node('b',2,1,'pipe',target='F',catches=False),
),(4,0,0,0),4)
return level, State(1,1,(0,1,1,1))
raise ValueError('Choose lab 1, 2 or 3')
CONTROLS = ('LEFT','RIGHT','UP','DOWN','X','Z','R','1','2','3')
REPLAY = {'L':'LEFT','R':'RIGHT','U':'UP','D':'DOWN','X':'X','Z':'Z',
'Q':'R','1':'1','2':'2','3':'3'}
class Controller:
def __init__(self):
self.lock = False
self.reset()
def reset(self):
self.level = BOARD
self.state = initial(BOARD)
self.history = []
self.trace = ()
self.transfers = ()
self.tick_count = 0
self.reason = 'ready'
self.preset = 1
self.level,self.state=lab(1)
# reset中にlockは消さない。R後も全キー解放が必要。
def tick(self, pressed=(), held=()):
pressed = set(pressed) & set(CONTROLS)
held = set(held) & set(CONTROLS)
if self.lock:
if not held:
self.lock = False
return
if not pressed and len(held) <= 1:
return
if len(pressed) != 1 or pressed != held:
self.lock = True
self.reason = 'chord_rejected'
return
self.lock = True
key = next(iter(pressed))
if key == 'R':
self.reset()
return
if key in ('1','2','3'):
self.preset=int(key);self.level,self.state=lab(self.preset)
self.trace=();self.transfers=();self.tick_count=0;self.reason='preset';return
if key=='X':
old=self.state.water
water,events=microstep(self.level,old)
self.state=replace(self.state,water=water)
check(self.level,self.state,stable=False)
self.trace=self.trace+(old,water)
self.transfers=self.transfers+(events,)
self.tick_count+=1
self.reason='flow' if events else 'stable'
else:self.reason='use_X_or_123'
def snapshot(self):
sun, rain = preview(self.level,self.state) if STAGE >= 2 else (None,None)
return {'state':asdict(self.state),'hash':state_hash(self.state),
'total':total(self.state),'budget':self.level.total,
'gate_open':gate_open(self.level,self.state.water) if STAGE>=2 else False,
'won':won(self.level,self.state) if STAGE>=4 else False,
'sun':None if sun is None else self.level.nodes[sun].name,
'rain':None if rain is None else self.level.nodes[rain].name,
'event':self.reason,'history_depth':len(self.history),
'trace':self.trace,'transfers':self.transfers,
'ticks':self.tick_count,'preset':self.preset,'lock':self.lock}
def draw_board(p,c):
"""描画は状態を読むだけ。水滴の絵を別資源として加算しない。"""
s=c.state; level=c.level; snap=c.snapshot()
p.cls(1)
p.text(14,12,'BOX WEATHER / STEP '+str(STAGE)+' / '+STAGE_TITLE,10)
p.text(14,27,'Fictional integer water model / no automatic evaporation / silent',6)
ox,oy,dx,dy=47,105,54,42
def xy(n): return ox+n.x*dx,oy+n.y*dy
def arrow(x,y,xx,yy,color):
p.line(x,y,xx,yy,color)
if abs(xx-x)>abs(yy-y):
sign=1 if xx>x else -1
p.line(xx,yy,xx-sign*5,yy-3,color);p.line(xx,yy,xx-sign*5,yy+3,color)
else:
sign=1 if yy>y else -1
p.line(xx,yy,xx-3,yy-sign*5,color);p.line(xx,yy,xx+3,yy-sign*5,color)
p.rect(18,47,402,313,0);p.rectb(18,47,402,313,5)
for x in range(7):
p.text(ox+x*dx-2,53,str(x),13)
for y in range(110,333,10): p.pset(ox+x*dx,y,1)
for n in level.nodes:
if n.target:
dst=level.nodes[index(level,n.target)]
x,y=xy(n);xx,yy=xy(dst)
if n.kind!='dish':
color=(11 if snap['gate_open'] else 5) if n.kind=='gate' else 6
arrow(x,y+12,xx,yy-15,color)
if n.kind=='gate':
x,y=xy(n);color=5 if snap['gate_open'] else 6
arrow(x+25,y,401,y,color);arrow(401,y,401,346,color)
src=level.nodes[index(level,'S')];sx,sy=xy(src)
arrow(401,346,sx,346,6);arrow(sx,346,sx,sy+13,6)
p.text(190,349,'RETURN TO S',6)
for i,n in enumerate(level.nodes):
x,y=xy(n);value=s.water[i]
if n.kind=='dish':
p.rect(x-21,y-9,43,19,4);p.rect(x-18,y-10,37,14,1)
for j in range(n.capacity):
if j<value:p.circ(x-12+j*9,y-2,3,12)
else:p.circb(x-12+j*9,y-2,3,5)
p.text(x-22,y+16,f'{n.name}: {value}/{n.capacity}',7)
if value>n.capacity: p.text(x-22,y+27,'EXCESS '+str(value-n.capacity),8)
if n.name=='F':
p.line(x,y-12,x,y-28,11);p.circ(x,y-30,5,10 if snap['won'] else 3)
else:
color=11 if n.kind=='gate' and snap['gate_open'] else 9
p.rectb(x-23,y-10,47,20,color)
p.text(x-20,y-3,n.name,7)
p.text(x-21,y+15,'WATER '+str(value),12)
if n.kind=='gate': p.text(x-20,y-21,'OPEN' if snap['gate_open'] else 'CLOSED',color)
if n.name==snap['rain']:p.rectb(x-26,y-14,53,29,12)
if n.name==snap['sun']:p.rectb(x-24,y-12,49,25,10)
cx=ox+s.x*dx
p.elli(cx-24,76,49,20,7);p.circ(cx-10,76,11,7);p.circ(cx+9,73,12,7)
for j in range(level.cloud_capacity):
if j<s.cloud:p.circ(cx-10+j*10,86,3,12)
else:p.circb(cx-10+j*10,86,3,5)
p.rect(437,47,269,313,0)
lines=[('WATER OWNERS',10),('Cloud: '+str(s.cloud)+' / '+str(level.cloud_capacity),12)]
lines += [(n.name+': '+str(s.water[i]),7) for i,n in enumerate(level.nodes)]
lines += [('Absorbed ledger: '+str(s.absorbed),6),
('TOTAL '+str(total(s))+' = BUDGET '+str(level.total),11),
('Moves: '+str(s.moves)+' ticks: '+str(c.tick_count),7),
('History: '+str(len(c.history)),7)]
y=60
for line,color in lines:p.text(450,y,line,color);y+=17
p.text(450,275,'SUN : '+str(snap['sun']),10)
p.text(450,292,'RAIN: '+str(snap['rain']),12)
p.text(450,310,'CASE '+str(c.preset)+' / '+level.title,6)
p.text(450,332,'WON' if snap['won'] else 'OBSERVE / TRY',10)
p.text(14,376,'EVENT: '+c.reason,10)
p.text(14,394,GUIDE,7)
p.text(14,413,'R: reset ESC: quit / release ALL controls between presses',6)
if c.transfers:
p.text(14,433,'LAST FLOW: '+str(c.transfers[-1]),12)
else:p.text(14,433,'Blue=rain target Yellow=sun target / x=0..6',13)
p.text(14,458,'A flow tick sends at most one unit per emitting node. No water is deleted.',6)
p.text(14,476,'01-03 are labs. 04-05 use original C rules. This is NOT physical hydrology.',13)
def main(headless=False,sequence='',screenshot=None,report=None):
from pathlib import Path
import time
import pyxel # モデルのimportやstdlibテストにPyxelは不要。
c=Controller();codes={n:getattr(pyxel,'KEY_'+n) for n in CONTROLS}
frame=0;events=[];last=max(12,2*len(sequence)+2);start=time.perf_counter()
def update():
nonlocal frame
frame+=1
if headless:
i=(frame-1)//2
if frame%2 and i<len(sequence):
k=REPLAY[sequence[i]];c.tick((k,),(k,))
events.append({'input':sequence[i],'snapshot':c.snapshot()})
else:c.tick()
else:c.tick([n for n,k in codes.items() if pyxel.btnp(k)],
[n for n,k in codes.items() if pyxel.btn(k)])
def draw():
draw_board(pyxel,c)
if headless and frame>=last:
if screenshot:pyxel.screenshot(str(Path(screenshot).resolve()),scale=1)
if report:
Path(report).write_text(json.dumps({
'label':'actual Pyxel headless / synthetic controller input / silent',
'stage':STAGE,'frames':frame,'logical_seconds':frame/30,
'wall_seconds_before_exit':round(time.perf_counter()-start,6),
'program_sha256':hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
'sequence':sequence,'events':events,'snapshot':c.snapshot()
},ensure_ascii=False,indent=2)+'\n',encoding='utf-8')
pyxel.quit()
pyxel.init(720,500,title='Box Weather - Step '+str(STAGE),fps=30,headless=headless)
pyxel.run(update,draw)
if __name__=='__main__':
import argparse
parser=argparse.ArgumentParser(description=__doc__)
group=parser.add_mutually_exclusive_group()
group.add_argument('--headless-smoke',action='store_true')
group.add_argument('--headless-replay',help='L/R/U/D arrows; X/Z; Q reset; 1/2/3 presets')
parser.add_argument('--screenshot');parser.add_argument('--report')
args=parser.parse_args()
headless=args.headless_smoke or args.headless_replay is not None
if not headless and (args.screenshot or args.report):parser.error('Outputs require headless mode')
seq=args.headless_replay or ''
if any(k not in REPLAY for k in seq):parser.error('Unsupported replay character')
main(headless,seq,args.screenshot,args.report)雨を流し切り、Cの条件で咲かせる
純粋関数の transition、必ず終わる流れ、水の合計を保つ決まり、成功したあとの入力の停止。
ここで初めて、元のゲームの面Cのルールを全部つなぎます。transition は、まず今が安定した状態（もう水が流れない状態）かを確かめます。成功したあとなら、complete で止めます。左右キーは雲の移動、上キーは見えている皿から1滴を雲へ回収、下キーは雨を最初に受けるものへ1滴を足します。そのあと settle が microstep をくり返し、安定した水の配列、途中の記録 trace、水の移動の記録 transfers を返します。水の通り道は一方通行で、ぐるりと元に戻る輪がありません（有向非循環）。くり返しの上限は (水量+1)×(ノード数+1) 回です。上限を超えたら例外（エラー）で止め、水を捨てて続けることはしません。trace には、流れ始めの配列も入れます。変化しなくなった最後の配列は、2回は入れません。刻みごとに cloud+sum(water)+absorbed が3であることも調べます。成功の条件は、F>=2 で、しかも門が開いていることです。
下のコードは 04_puzzle.py の一部です。前のステップから追加・変更したところだけを抜き出しています。
04_puzzle.py / 注目する部分
def settle(level,water,order=None):
trace=[water];transfers=[]
# DAG plus finite integer water bounds steps by total units times node count.
limit=(sum(water)+1)*(len(level.nodes)+1)
for _ in range(limit):
nxt,events=microstep(level,trace[-1],order)
if not events:return nxt,tuple(trace),tuple(transfers)
trace.append(nxt);transfers.append(events)
raise ValueError('water did not settle; no units discarded')実行して確かめる
教材のZIPを展開したフォルダーで、次のコマンドを実行します。
Windows
.venv\Scripts\python.exe 04_puzzle.pymacOS / Linux
.venv/bin/python 04_puzzle.py画面で見ること上3回→右2回→下→右2回→下2回で成功です。Cloud 0、S 0、F 2、K 1、gate 0、Moves 10 になります。上3回→右4回→下3回では、閉じた門が1滴ごとにSへ返すので、S 3・F 0・K 0 となり、成功しません。Kへ2滴降らせると、Kは1滴をためておき、あまった1滴だけがSへ戻ります。
- 1手で、安定した状態まで進む
traceのどの配列でも、水の合計が変わらない- 勝つには、F 2 と K 1 の両方が必要
- 成功したあとは、ふつうの
transitionを止める
つまずきやすいところ花に2滴あるだけで成功にしないでください。面Cでは、Kに1滴を残して門を開くことも条件です。花の滴は、花にためてあるだけの水です。まだ成功していなければ、上キーで雲へ回収できます。手数を時間のように扱って水を自動で蒸発させたり、受け付けない入力を手数に数えたりしないでください。04には、まだ Undo（元に戻す）はありません。
少し変えてみよう
UUURRDRRDD で成功させてから、さらに上を押します。complete と出て、Moves 10 のままになることを確かめましょう。R で戻して UUURRDD と押すと、K 1・S 1・Cloud 1 になります。Kからあふれた水を確かめます。余裕があれば、tools/run_examples.py の trace を読み、K 2 から K 1・S 1 になる1刻みの変化を追ってみましょう。
04_puzzle.pyをダウンロードこの段階だけで実行できます
このステップのコード全文（382行）
04_puzzle.py / 全文
# -*- coding: utf-8 -*-
"""030 ひと箱の天気 / standalone tutorial / fictional discrete water model.
Original model: Pyxel 100 contributors. Teaching adaptation: 2026.
No external image/font asset is needed. Pyxel is imported only in main().
"""
# 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.
from dataclasses import dataclass, replace, asdict
import hashlib, json
STAGE = 4
STAGE_TITLE = 'COMPLETE C RULES'
GUIDE = 'LEFT/RIGHT: cloud UP: sun DOWN: rain / F=2 and K=1 wins'
@dataclass(frozen=True)
class Node:
name:str
x:int
y:int
kind:str='dish'
capacity:int=0
target:str|None=None
alternate:str|None=None
catches:bool=True
shades:bool=False
@dataclass(frozen=True)
class Level:
title:str
nodes:tuple[Node,...]
start:tuple[int,...]
total:int
gate_linked:bool=True
width:int=7
cloud_capacity:int=3
@dataclass(frozen=True)
class State:
x:int
cloud:int
water:tuple[int,...]
absorbed:int=0 # Explicit accounting slot; authored soil holds water, never absorbs it.
moves:int=0
@dataclass(frozen=True)
class Result:
state:State
accepted:bool
reason:str
trace:tuple[tuple[int,...],...]=()
transfers:tuple[tuple[tuple[str,str,int],...],...]=()
# 原作Cの座標・容量・流路。列は0〜6、yは下へ増える。
BOARD = Level('C: keep one drop in K', (
Node('S',1,5,capacity=3),
Node('F',5,4,capacity=2,target='S'),
Node('K',3,5,capacity=1,target='S'),
Node('gate',5,2,'gate',target='F',alternate='S',shades=True),
), (3,0,0,0), 3)
def index(level,name):return next(i for i,n in enumerate(level.nodes) if n.name==name)
def amount(level,state,name):return state.water[index(level,name)]
def initial(level):return State(1,0,level.start)
def total(state):return state.cloud+sum(state.water)+state.absorbed
def check(level,state,stable=True):
assert len(state.water)==len(level.nodes)
assert 0<=state.x<level.width and 0<=state.cloud<=level.cloud_capacity
assert state.absorbed>=0 and all(v>=0 for v in state.water)
assert total(state)==level.total
if stable:
assert all(v<=n.capacity if n.kind=='dish' else v==0 for n,v in zip(level.nodes,state.water))
def state_hash(state):return hashlib.sha256(json.dumps(asdict(state),sort_keys=True).encode('utf-8')).hexdigest()
def gate_open(level,water):
return not level.gate_linked or not any(n.name=='K' for n in level.nodes) or water[index(level,'K')]>=1
def visible(level,state,sun=False):
candidates=[]
for i,n in enumerate(level.nodes):
if n.x!=state.x:continue
if n.kind=='gate' and gate_open(level,state.water):continue
if (sun and (n.kind=='dish' or n.shades)) or (not sun and n.catches):candidates.append((n.y,n.name,i))
return min(candidates)[2] if candidates else None
def preview(level,state):
sun=visible(level,state,True);rain=visible(level,state)
rain=index(level,'S') if rain is None else rain
return sun,rain
def microstep(level,old,order=None):
"""All intents use old; merge additions in new. One unit per emitting node.
Gates inspect old K, not the partially written new buffer. Converging inflows
sum without priority: excess remains and overflows on the following tick.
"""
new=list(old);moves=[];order=range(len(old)) if order is None else tuple(order)
if sorted(order)!=list(range(len(old))):raise ValueError('scan must visit each node once')
for i in order:
n=level.nodes[i]
if old[i]<= (n.capacity if n.kind=='dish' else 0):continue
target=n.target if n.kind!='gate' or gate_open(level,old) else n.alternate
if target is None:raise ValueError('unrouted overflow')
j=index(level,target);new[i]-=1;new[j]+=1;moves.append((n.name,target,1))
assert sum(new)==sum(old) and all(v>=0 for v in new)
return tuple(new),tuple(sorted(moves))
def validate_level(level):
if len(level.nodes)!=len(level.start) or len({n.name for n in level.nodes})!=len(level.nodes):raise ValueError('node identity')
if sum(level.start)!=level.total or level.cloud_capacity<1:raise ValueError('water budget')
catches=[(n.x,n.y) for n in level.nodes if n.catches]
if len(catches)!=len(set(catches)):raise ValueError('ambiguous vertical catch tie')
names={n.name for n in level.nodes};edges={n.name:[v for v in (n.target,n.alternate) if v is not None] for n in level.nodes}
if any(v not in names for es in edges.values() for v in es):raise ValueError('unknown flow destination')
if any(v<0 for v in level.start):raise ValueError('negative initial water')
visiting=set();done=set()
def visit(k):
if k in visiting:raise ValueError('cyclic flow graph')
if k in done:return
visiting.add(k)
for v in edges[k]:visit(v)
visiting.remove(k);done.add(k)
for n in level.nodes:visit(n.name)
if any(n.kind!='dish' and level.start[i] for i,n in enumerate(level.nodes)):raise ValueError('unstable start')
if any(n.kind=='dish' and level.start[i]>n.capacity for i,n in enumerate(level.nodes)):raise ValueError('overflowing start')
return True
def settle(level,water,order=None):
trace=[water];transfers=[]
# DAG plus finite integer water bounds steps by total units times node count.
limit=(sum(water)+1)*(len(level.nodes)+1)
for _ in range(limit):
nxt,events=microstep(level,trace[-1],order)
if not events:return nxt,tuple(trace),tuple(transfers)
trace.append(nxt);transfers.append(events)
raise ValueError('water did not settle; no units discarded')
def won(level,state):return amount(level,state,'F')>=2 and gate_open(level,state.water)
def transition(level,state,action,allow_terminal=False):
check(level,state)
if won(level,state) and not allow_terminal:return Result(state,False,'complete')
if action not in 'LRUD' or len(action)!=1:return Result(state,False,'invalid')
if action in 'LR':
x=state.x+(-1 if action=='L' else 1)
if not 0<=x<level.width:return Result(state,False,'edge')
return Result(replace(state,x=x,moves=state.moves+1),True,'move')
water=list(state.water);cloud=state.cloud;sun,rain=preview(level,state)
if action=='U':
if cloud==level.cloud_capacity:return Result(state,False,'full')
if sun is None:return Result(state,False,'no_dish')
if level.nodes[sun].kind!='dish':return Result(state,False,'shade')
if water[sun]==0:return Result(state,False,'dry')
water[sun]-=1;cloud+=1;reason='sun'
else:
if cloud==0:return Result(state,False,'empty')
cloud-=1;water[rain]+=1;reason='rain'
settled,trace,moves=settle(level,tuple(water))
nxt=State(state.x,cloud,settled,state.absorbed,state.moves+1);check(level,nxt)
assert all(cloud+sum(t)+state.absorbed==level.total for t in trace)
return Result(nxt,True,reason,trace,moves)
def key(state):return state.x,state.cloud,state.water,state.absorbed
def replay(level,actions):
s=initial(level)
for a in actions:
r=transition(level,s,a)
if not r.accepted:raise ValueError((a,r.reason))
s=r.state
return s
validate_level(BOARD)
CONTROLS = ('LEFT','RIGHT','UP','DOWN','X','Z','R','1','2','3')
REPLAY = {'L':'LEFT','R':'RIGHT','U':'UP','D':'DOWN','X':'X','Z':'Z',
'Q':'R','1':'1','2':'2','3':'3'}
class Controller:
def __init__(self):
self.lock = False
self.reset()
def reset(self):
self.level = BOARD
self.state = initial(BOARD)
self.history = []
self.trace = ()
self.transfers = ()
self.tick_count = 0
self.reason = 'ready'
self.preset = 1
# reset中にlockは消さない。R後も全キー解放が必要。
def tick(self, pressed=(), held=()):
pressed = set(pressed) & set(CONTROLS)
held = set(held) & set(CONTROLS)
if self.lock:
if not held:
self.lock = False
return
if not pressed and len(held) <= 1:
return
if len(pressed) != 1 or pressed != held:
self.lock = True
self.reason = 'chord_rejected'
return
self.lock = True
key = next(iter(pressed))
if key == 'R':
self.reset()
return
action={'LEFT':'L','RIGHT':'R','UP':'U','DOWN':'D'}.get(key,'?')
r=transition(self.level,self.state,action)
self.state=r.state;self.reason=r.reason;self.trace=r.trace;self.transfers=r.transfers
def snapshot(self):
sun, rain = preview(self.level,self.state) if STAGE >= 2 else (None,None)
return {'state':asdict(self.state),'hash':state_hash(self.state),
'total':total(self.state),'budget':self.level.total,
'gate_open':gate_open(self.level,self.state.water) if STAGE>=2 else False,
'won':won(self.level,self.state) if STAGE>=4 else False,
'sun':None if sun is None else self.level.nodes[sun].name,
'rain':None if rain is None else self.level.nodes[rain].name,
'event':self.reason,'history_depth':len(self.history),
'trace':self.trace,'transfers':self.transfers,
'ticks':self.tick_count,'preset':self.preset,'lock':self.lock}
def draw_board(p,c):
"""描画は状態を読むだけ。水滴の絵を別資源として加算しない。"""
s=c.state; level=c.level; snap=c.snapshot()
p.cls(1)
p.text(14,12,'BOX WEATHER / STEP '+str(STAGE)+' / '+STAGE_TITLE,10)
p.text(14,27,'Fictional integer water model / no automatic evaporation / silent',6)
ox,oy,dx,dy=47,105,54,42
def xy(n): return ox+n.x*dx,oy+n.y*dy
def arrow(x,y,xx,yy,color):
p.line(x,y,xx,yy,color)
if abs(xx-x)>abs(yy-y):
sign=1 if xx>x else -1
p.line(xx,yy,xx-sign*5,yy-3,color);p.line(xx,yy,xx-sign*5,yy+3,color)
else:
sign=1 if yy>y else -1
p.line(xx,yy,xx-3,yy-sign*5,color);p.line(xx,yy,xx+3,yy-sign*5,color)
p.rect(18,47,402,313,0);p.rectb(18,47,402,313,5)
for x in range(7):
p.text(ox+x*dx-2,53,str(x),13)
for y in range(110,333,10): p.pset(ox+x*dx,y,1)
for n in level.nodes:
if n.target:
dst=level.nodes[index(level,n.target)]
x,y=xy(n);xx,yy=xy(dst)
if n.kind!='dish':
color=(11 if snap['gate_open'] else 5) if n.kind=='gate' else 6
arrow(x,y+12,xx,yy-15,color)
if n.kind=='gate':
x,y=xy(n);color=5 if snap['gate_open'] else 6
arrow(x+25,y,401,y,color);arrow(401,y,401,346,color)
src=level.nodes[index(level,'S')];sx,sy=xy(src)
arrow(401,346,sx,346,6);arrow(sx,346,sx,sy+13,6)
p.text(190,349,'RETURN TO S',6)
for i,n in enumerate(level.nodes):
x,y=xy(n);value=s.water[i]
if n.kind=='dish':
p.rect(x-21,y-9,43,19,4);p.rect(x-18,y-10,37,14,1)
for j in range(n.capacity):
if j<value:p.circ(x-12+j*9,y-2,3,12)
else:p.circb(x-12+j*9,y-2,3,5)
p.text(x-22,y+16,f'{n.name}: {value}/{n.capacity}',7)
if value>n.capacity: p.text(x-22,y+27,'EXCESS '+str(value-n.capacity),8)
if n.name=='F':
p.line(x,y-12,x,y-28,11);p.circ(x,y-30,5,10 if snap['won'] else 3)
else:
color=11 if n.kind=='gate' and snap['gate_open'] else 9
p.rectb(x-23,y-10,47,20,color)
p.text(x-20,y-3,n.name,7)
p.text(x-21,y+15,'WATER '+str(value),12)
if n.kind=='gate': p.text(x-20,y-21,'OPEN' if snap['gate_open'] else 'CLOSED',color)
if n.name==snap['rain']:p.rectb(x-26,y-14,53,29,12)
if n.name==snap['sun']:p.rectb(x-24,y-12,49,25,10)
cx=ox+s.x*dx
p.elli(cx-24,76,49,20,7);p.circ(cx-10,76,11,7);p.circ(cx+9,73,12,7)
for j in range(level.cloud_capacity):
if j<s.cloud:p.circ(cx-10+j*10,86,3,12)
else:p.circb(cx-10+j*10,86,3,5)
p.rect(437,47,269,313,0)
lines=[('WATER OWNERS',10),('Cloud: '+str(s.cloud)+' / '+str(level.cloud_capacity),12)]
lines += [(n.name+': '+str(s.water[i]),7) for i,n in enumerate(level.nodes)]
lines += [('Absorbed ledger: '+str(s.absorbed),6),
('TOTAL '+str(total(s))+' = BUDGET '+str(level.total),11),
('Moves: '+str(s.moves)+' ticks: '+str(c.tick_count),7),
('History: '+str(len(c.history)),7)]
y=60
for line,color in lines:p.text(450,y,line,color);y+=17
p.text(450,275,'SUN : '+str(snap['sun']),10)
p.text(450,292,'RAIN: '+str(snap['rain']),12)
p.text(450,310,'CASE '+str(c.preset)+' / '+level.title,6)
p.text(450,332,'WON' if snap['won'] else 'OBSERVE / TRY',10)
p.text(14,376,'EVENT: '+c.reason,10)
p.text(14,394,GUIDE,7)
p.text(14,413,'R: reset ESC: quit / release ALL controls between presses',6)
if c.transfers:
p.text(14,433,'LAST FLOW: '+str(c.transfers[-1]),12)
else:p.text(14,433,'Blue=rain target Yellow=sun target / x=0..6',13)
p.text(14,458,'A flow tick sends at most one unit per emitting node. No water is deleted.',6)
p.text(14,476,'01-03 are labs. 04-05 use original C rules. This is NOT physical hydrology.',13)
def main(headless=False,sequence='',screenshot=None,report=None):
from pathlib import Path
import time
import pyxel # モデルのimportやstdlibテストにPyxelは不要。
c=Controller();codes={n:getattr(pyxel,'KEY_'+n) for n in CONTROLS}
frame=0;events=[];last=max(12,2*len(sequence)+2);start=time.perf_counter()
def update():
nonlocal frame
frame+=1
if headless:
i=(frame-1)//2
if frame%2 and i<len(sequence):
k=REPLAY[sequence[i]];c.tick((k,),(k,))
events.append({'input':sequence[i],'snapshot':c.snapshot()})
else:c.tick()
else:c.tick([n for n,k in codes.items() if pyxel.btnp(k)],
[n for n,k in codes.items() if pyxel.btn(k)])
def draw():
draw_board(pyxel,c)
if headless and frame>=last:
if screenshot:pyxel.screenshot(str(Path(screenshot).resolve()),scale=1)
if report:
Path(report).write_text(json.dumps({
'label':'actual Pyxel headless / synthetic controller input / silent',
'stage':STAGE,'frames':frame,'logical_seconds':frame/30,
'wall_seconds_before_exit':round(time.perf_counter()-start,6),
'program_sha256':hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
'sequence':sequence,'events':events,'snapshot':c.snapshot()
},ensure_ascii=False,indent=2)+'\n',encoding='utf-8')
pyxel.quit()
pyxel.init(720,500,title='Box Weather - Step '+str(STAGE),fps=30,headless=headless)
pyxel.run(update,draw)
if __name__=='__main__':
import argparse
parser=argparse.ArgumentParser(description=__doc__)
group=parser.add_mutually_exclusive_group()
group.add_argument('--headless-smoke',action='store_true')
group.add_argument('--headless-replay',help='L/R/U/D arrows; X/Z; Q reset; 1/2/3 presets')
parser.add_argument('--screenshot');parser.add_argument('--report')
args=parser.parse_args()
headless=args.headless_smoke or args.headless_replay is not None
if not headless and (args.screenshot or args.report):parser.error('Outputs require headless mode')
seq=args.headless_replay or ''
if any(k not in REPLAY for k in seq):parser.error('Unsupported replay character')
main(headless,seq,args.screenshot,args.report)一手全体を戻して解き直す
書き換えない履歴、受け付けた行動だけを保存、成功したあとの Undo、入力を止める仕組み（入力ゲート）。
Timeline は、今の state と、過去の状態を並べた past の tuple を持ちます。apply は transition を呼びます。行動が受け付けられたときだけ、変更前の state を最後に加えます。undo は、最後の状態を今の状態に戻し、履歴から1つ取り除きます。水が流れる小さな刻みごとには、履歴を作りません。日差し・雨・左右の移動という、プレイヤーの1手が Undo の単位です。成功したあとも Z を先に処理するので、元に戻せます。戻したあとで別の行動をしたときに、取り消した先の手順（古い未来）を残しておく Redo（やり直し）はありません。教材の Controller は、操作キーを全部一度離すまで、次の入力を受け付けません。R をふくむ同時押しも受け付けません。元のゲームにある、水が移るアニメーション、タイトル、P・M・N は省き、1手の結果はすぐに安定した状態として表示します。元のゲームのルール部分（モデル）と、簡単にした入力の仕組みは、分けて扱います。
下のコードは 05_undo.py の一部です。前のステップから追加・変更したところだけを抜き出しています。
05_undo.py / 注目する部分
def apply(level,timeline,action):
r=transition(level,timeline.state,action)
return (Timeline(r.state,timeline.past+(timeline.state,)) if r.accepted else timeline),r
def undo(timeline):return Timeline(timeline.past[-1],timeline.past[:-1]) if timeline.past else timeline実行して確かめる
教材のZIPを展開したフォルダーで、次のコマンドを実行します。
Windows
.venv\Scripts\python.exe 05_undo.pymacOS / Linux
.venv/bin/python 05_undo.py画面で見ることUUURRDRRDD と押すと、Moves 10・History 10・WON（成功）になります。Z を押すと Cloud 1、F 1、K 1、Moves 9・History 9 に戻り、下を押すともう一度成功します。空の雲で下、雲が満杯で上、箱のはしからさらに外への移動などでは、履歴は増えません。R で最初の状態に戻すと、履歴は0です。R を押したまま別のキーを押しても入力されず、キーを全部離すと、また入力できるようになります。
- 受け付けた行動だけを履歴に残す
- 左右の移動も1手。水が流れる小さな刻みは、まとめて1手
- 成功したあとも Z を使える
- キーを全部離すまで待つことと、同時押しを受け付けないことを確かめる
つまずきやすいところあとから書き換えられる
list をそのまま履歴に入れ、あとで書き換えると、過去の状態まで変わってしまいます。この State の中の水は tuple です。描画、受け付けない操作、決まった配置（プリセット）は、プレイの履歴に加えないでください。左右の移動は、元のゲームでは正しい1手なので、Undo の対象です。成功したあとに、上キーで花の水を回収することは、ふつうの操作ではできません。Z で成功する前に戻ります。少し変えてみよう
解いたあとに、Z→下でもう一度成功させましょう。次に R で最初に戻し、下を押して History 0 のままになることを確かめます。上→Z→右と押すと、Moves 1・History 1、Cloud 0・x 2 になります。取り消した「上」の手順は、もう戻ってきません。テストでは、面Cでたどりつける85の状態からの、ふつうの操作による340通りの変化すべてと、その途中の trace を確かめます。さらに、門と合流の25の決まった配置、Undo と入力の境目も確かめます。
05_undo.pyをダウンロードこの段階だけで実行できます
このステップのコード全文（398行）
05_undo.py / 全文
# -*- coding: utf-8 -*-
"""030 ひと箱の天気 / standalone tutorial / fictional discrete water model.
Original model: Pyxel 100 contributors. Teaching adaptation: 2026.
No external image/font asset is needed. Pyxel is imported only in main().
"""
# 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.
from dataclasses import dataclass, replace, asdict
import hashlib, json
STAGE = 5
STAGE_TITLE = 'UNDO THE WHOLE ACTION'
GUIDE = 'LEFT/RIGHT: cloud UP: sun DOWN: rain Z: undo (also after win)'
@dataclass(frozen=True)
class Node:
name:str
x:int
y:int
kind:str='dish'
capacity:int=0
target:str|None=None
alternate:str|None=None
catches:bool=True
shades:bool=False
@dataclass(frozen=True)
class Level:
title:str
nodes:tuple[Node,...]
start:tuple[int,...]
total:int
gate_linked:bool=True
width:int=7
cloud_capacity:int=3
@dataclass(frozen=True)
class State:
x:int
cloud:int
water:tuple[int,...]
absorbed:int=0 # Explicit accounting slot; authored soil holds water, never absorbs it.
moves:int=0
@dataclass(frozen=True)
class Result:
state:State
accepted:bool
reason:str
trace:tuple[tuple[int,...],...]=()
transfers:tuple[tuple[tuple[str,str,int],...],...]=()
@dataclass(frozen=True)
class Timeline:
state:State
past:tuple[State,...]=()
# 原作Cの座標・容量・流路。列は0〜6、yは下へ増える。
BOARD = Level('C: keep one drop in K', (
Node('S',1,5,capacity=3),
Node('F',5,4,capacity=2,target='S'),
Node('K',3,5,capacity=1,target='S'),
Node('gate',5,2,'gate',target='F',alternate='S',shades=True),
), (3,0,0,0), 3)
def index(level,name):return next(i for i,n in enumerate(level.nodes) if n.name==name)
def amount(level,state,name):return state.water[index(level,name)]
def initial(level):return State(1,0,level.start)
def total(state):return state.cloud+sum(state.water)+state.absorbed
def check(level,state,stable=True):
assert len(state.water)==len(level.nodes)
assert 0<=state.x<level.width and 0<=state.cloud<=level.cloud_capacity
assert state.absorbed>=0 and all(v>=0 for v in state.water)
assert total(state)==level.total
if stable:
assert all(v<=n.capacity if n.kind=='dish' else v==0 for n,v in zip(level.nodes,state.water))
def state_hash(state):return hashlib.sha256(json.dumps(asdict(state),sort_keys=True).encode('utf-8')).hexdigest()
def gate_open(level,water):
return not level.gate_linked or not any(n.name=='K' for n in level.nodes) or water[index(level,'K')]>=1
def visible(level,state,sun=False):
candidates=[]
for i,n in enumerate(level.nodes):
if n.x!=state.x:continue
if n.kind=='gate' and gate_open(level,state.water):continue
if (sun and (n.kind=='dish' or n.shades)) or (not sun and n.catches):candidates.append((n.y,n.name,i))
return min(candidates)[2] if candidates else None
def preview(level,state):
sun=visible(level,state,True);rain=visible(level,state)
rain=index(level,'S') if rain is None else rain
return sun,rain
def microstep(level,old,order=None):
"""All intents use old; merge additions in new. One unit per emitting node.
Gates inspect old K, not the partially written new buffer. Converging inflows
sum without priority: excess remains and overflows on the following tick.
"""
new=list(old);moves=[];order=range(len(old)) if order is None else tuple(order)
if sorted(order)!=list(range(len(old))):raise ValueError('scan must visit each node once')
for i in order:
n=level.nodes[i]
if old[i]<= (n.capacity if n.kind=='dish' else 0):continue
target=n.target if n.kind!='gate' or gate_open(level,old) else n.alternate
if target is None:raise ValueError('unrouted overflow')
j=index(level,target);new[i]-=1;new[j]+=1;moves.append((n.name,target,1))
assert sum(new)==sum(old) and all(v>=0 for v in new)
return tuple(new),tuple(sorted(moves))
def validate_level(level):
if len(level.nodes)!=len(level.start) or len({n.name for n in level.nodes})!=len(level.nodes):raise ValueError('node identity')
if sum(level.start)!=level.total or level.cloud_capacity<1:raise ValueError('water budget')
catches=[(n.x,n.y) for n in level.nodes if n.catches]
if len(catches)!=len(set(catches)):raise ValueError('ambiguous vertical catch tie')
names={n.name for n in level.nodes};edges={n.name:[v for v in (n.target,n.alternate) if v is not None] for n in level.nodes}
if any(v not in names for es in edges.values() for v in es):raise ValueError('unknown flow destination')
if any(v<0 for v in level.start):raise ValueError('negative initial water')
visiting=set();done=set()
def visit(k):
if k in visiting:raise ValueError('cyclic flow graph')
if k in done:return
visiting.add(k)
for v in edges[k]:visit(v)
visiting.remove(k);done.add(k)
for n in level.nodes:visit(n.name)
if any(n.kind!='dish' and level.start[i] for i,n in enumerate(level.nodes)):raise ValueError('unstable start')
if any(n.kind=='dish' and level.start[i]>n.capacity for i,n in enumerate(level.nodes)):raise ValueError('overflowing start')
return True
def settle(level,water,order=None):
trace=[water];transfers=[]
# DAG plus finite integer water bounds steps by total units times node count.
limit=(sum(water)+1)*(len(level.nodes)+1)
for _ in range(limit):
nxt,events=microstep(level,trace[-1],order)
if not events:return nxt,tuple(trace),tuple(transfers)
trace.append(nxt);transfers.append(events)
raise ValueError('water did not settle; no units discarded')
def won(level,state):return amount(level,state,'F')>=2 and gate_open(level,state.water)
def transition(level,state,action,allow_terminal=False):
check(level,state)
if won(level,state) and not allow_terminal:return Result(state,False,'complete')
if action not in 'LRUD' or len(action)!=1:return Result(state,False,'invalid')
if action in 'LR':
x=state.x+(-1 if action=='L' else 1)
if not 0<=x<level.width:return Result(state,False,'edge')
return Result(replace(state,x=x,moves=state.moves+1),True,'move')
water=list(state.water);cloud=state.cloud;sun,rain=preview(level,state)
if action=='U':
if cloud==level.cloud_capacity:return Result(state,False,'full')
if sun is None:return Result(state,False,'no_dish')
if level.nodes[sun].kind!='dish':return Result(state,False,'shade')
if water[sun]==0:return Result(state,False,'dry')
water[sun]-=1;cloud+=1;reason='sun'
else:
if cloud==0:return Result(state,False,'empty')
cloud-=1;water[rain]+=1;reason='rain'
settled,trace,moves=settle(level,tuple(water))
nxt=State(state.x,cloud,settled,state.absorbed,state.moves+1);check(level,nxt)
assert all(cloud+sum(t)+state.absorbed==level.total for t in trace)
return Result(nxt,True,reason,trace,moves)
def key(state):return state.x,state.cloud,state.water,state.absorbed
def replay(level,actions):
s=initial(level)
for a in actions:
r=transition(level,s,a)
if not r.accepted:raise ValueError((a,r.reason))
s=r.state
return s
validate_level(BOARD)
def apply(level,timeline,action):
r=transition(level,timeline.state,action)
return (Timeline(r.state,timeline.past+(timeline.state,)) if r.accepted else timeline),r
def undo(timeline):return Timeline(timeline.past[-1],timeline.past[:-1]) if timeline.past else timeline
CONTROLS = ('LEFT','RIGHT','UP','DOWN','X','Z','R','1','2','3')
REPLAY = {'L':'LEFT','R':'RIGHT','U':'UP','D':'DOWN','X':'X','Z':'Z',
'Q':'R','1':'1','2':'2','3':'3'}
class Controller:
def __init__(self):
self.lock = False
self.reset()
def reset(self):
self.level = BOARD
self.state = initial(BOARD)
self.history = []
self.trace = ()
self.transfers = ()
self.tick_count = 0
self.reason = 'ready'
self.preset = 1
# reset中にlockは消さない。R後も全キー解放が必要。
def tick(self, pressed=(), held=()):
pressed = set(pressed) & set(CONTROLS)
held = set(held) & set(CONTROLS)
if self.lock:
if not held:
self.lock = False
return
if not pressed and len(held) <= 1:
return
if len(pressed) != 1 or pressed != held:
self.lock = True
self.reason = 'chord_rejected'
return
self.lock = True
key = next(iter(pressed))
if key == 'R':
self.reset()
return
if key=='Z':
timeline=undo(Timeline(self.state,tuple(self.history)))
self.state=timeline.state;self.history=list(timeline.past)
self.trace=();self.transfers=();self.reason='undo';return
action={'LEFT':'L','RIGHT':'R','UP':'U','DOWN':'D'}.get(key,'?')
timeline,r=apply(self.level,Timeline(self.state,tuple(self.history)),action)
self.state=timeline.state;self.history=list(timeline.past)
self.reason=r.reason;self.trace=r.trace;self.transfers=r.transfers
def snapshot(self):
sun, rain = preview(self.level,self.state) if STAGE >= 2 else (None,None)
return {'state':asdict(self.state),'hash':state_hash(self.state),
'total':total(self.state),'budget':self.level.total,
'gate_open':gate_open(self.level,self.state.water) if STAGE>=2 else False,
'won':won(self.level,self.state) if STAGE>=4 else False,
'sun':None if sun is None else self.level.nodes[sun].name,
'rain':None if rain is None else self.level.nodes[rain].name,
'event':self.reason,'history_depth':len(self.history),
'trace':self.trace,'transfers':self.transfers,
'ticks':self.tick_count,'preset':self.preset,'lock':self.lock}
def draw_board(p,c):
"""描画は状態を読むだけ。水滴の絵を別資源として加算しない。"""
s=c.state; level=c.level; snap=c.snapshot()
p.cls(1)
p.text(14,12,'BOX WEATHER / STEP '+str(STAGE)+' / '+STAGE_TITLE,10)
p.text(14,27,'Fictional integer water model / no automatic evaporation / silent',6)
ox,oy,dx,dy=47,105,54,42
def xy(n): return ox+n.x*dx,oy+n.y*dy
def arrow(x,y,xx,yy,color):
p.line(x,y,xx,yy,color)
if abs(xx-x)>abs(yy-y):
sign=1 if xx>x else -1
p.line(xx,yy,xx-sign*5,yy-3,color);p.line(xx,yy,xx-sign*5,yy+3,color)
else:
sign=1 if yy>y else -1
p.line(xx,yy,xx-3,yy-sign*5,color);p.line(xx,yy,xx+3,yy-sign*5,color)
p.rect(18,47,402,313,0);p.rectb(18,47,402,313,5)
for x in range(7):
p.text(ox+x*dx-2,53,str(x),13)
for y in range(110,333,10): p.pset(ox+x*dx,y,1)
for n in level.nodes:
if n.target:
dst=level.nodes[index(level,n.target)]
x,y=xy(n);xx,yy=xy(dst)
if n.kind!='dish':
color=(11 if snap['gate_open'] else 5) if n.kind=='gate' else 6
arrow(x,y+12,xx,yy-15,color)
if n.kind=='gate':
x,y=xy(n);color=5 if snap['gate_open'] else 6
arrow(x+25,y,401,y,color);arrow(401,y,401,346,color)
src=level.nodes[index(level,'S')];sx,sy=xy(src)
arrow(401,346,sx,346,6);arrow(sx,346,sx,sy+13,6)
p.text(190,349,'RETURN TO S',6)
for i,n in enumerate(level.nodes):
x,y=xy(n);value=s.water[i]
if n.kind=='dish':
p.rect(x-21,y-9,43,19,4);p.rect(x-18,y-10,37,14,1)
for j in range(n.capacity):
if j<value:p.circ(x-12+j*9,y-2,3,12)
else:p.circb(x-12+j*9,y-2,3,5)
p.text(x-22,y+16,f'{n.name}: {value}/{n.capacity}',7)
if value>n.capacity: p.text(x-22,y+27,'EXCESS '+str(value-n.capacity),8)
if n.name=='F':
p.line(x,y-12,x,y-28,11);p.circ(x,y-30,5,10 if snap['won'] else 3)
else:
color=11 if n.kind=='gate' and snap['gate_open'] else 9
p.rectb(x-23,y-10,47,20,color)
p.text(x-20,y-3,n.name,7)
p.text(x-21,y+15,'WATER '+str(value),12)
if n.kind=='gate': p.text(x-20,y-21,'OPEN' if snap['gate_open'] else 'CLOSED',color)
if n.name==snap['rain']:p.rectb(x-26,y-14,53,29,12)
if n.name==snap['sun']:p.rectb(x-24,y-12,49,25,10)
cx=ox+s.x*dx
p.elli(cx-24,76,49,20,7);p.circ(cx-10,76,11,7);p.circ(cx+9,73,12,7)
for j in range(level.cloud_capacity):
if j<s.cloud:p.circ(cx-10+j*10,86,3,12)
else:p.circb(cx-10+j*10,86,3,5)
p.rect(437,47,269,313,0)
lines=[('WATER OWNERS',10),('Cloud: '+str(s.cloud)+' / '+str(level.cloud_capacity),12)]
lines += [(n.name+': '+str(s.water[i]),7) for i,n in enumerate(level.nodes)]
lines += [('Absorbed ledger: '+str(s.absorbed),6),
('TOTAL '+str(total(s))+' = BUDGET '+str(level.total),11),
('Moves: '+str(s.moves)+' ticks: '+str(c.tick_count),7),
('History: '+str(len(c.history)),7)]
y=60
for line,color in lines:p.text(450,y,line,color);y+=17
p.text(450,275,'SUN : '+str(snap['sun']),10)
p.text(450,292,'RAIN: '+str(snap['rain']),12)
p.text(450,310,'CASE '+str(c.preset)+' / '+level.title,6)
p.text(450,332,'WON' if snap['won'] else 'OBSERVE / TRY',10)
p.text(14,376,'EVENT: '+c.reason,10)
p.text(14,394,GUIDE,7)
p.text(14,413,'R: reset ESC: quit / release ALL controls between presses',6)
if c.transfers:
p.text(14,433,'LAST FLOW: '+str(c.transfers[-1]),12)
else:p.text(14,433,'Blue=rain target Yellow=sun target / x=0..6',13)
p.text(14,458,'A flow tick sends at most one unit per emitting node. No water is deleted.',6)
p.text(14,476,'01-03 are labs. 04-05 use original C rules. This is NOT physical hydrology.',13)
def main(headless=False,sequence='',screenshot=None,report=None):
from pathlib import Path
import time
import pyxel # モデルのimportやstdlibテストにPyxelは不要。
c=Controller();codes={n:getattr(pyxel,'KEY_'+n) for n in CONTROLS}
frame=0;events=[];last=max(12,2*len(sequence)+2);start=time.perf_counter()
def update():
nonlocal frame
frame+=1
if headless:
i=(frame-1)//2
if frame%2 and i<len(sequence):
k=REPLAY[sequence[i]];c.tick((k,),(k,))
events.append({'input':sequence[i],'snapshot':c.snapshot()})
else:c.tick()
else:c.tick([n for n,k in codes.items() if pyxel.btnp(k)],
[n for n,k in codes.items() if pyxel.btn(k)])
def draw():
draw_board(pyxel,c)
if headless and frame>=last:
if screenshot:pyxel.screenshot(str(Path(screenshot).resolve()),scale=1)
if report:
Path(report).write_text(json.dumps({
'label':'actual Pyxel headless / synthetic controller input / silent',
'stage':STAGE,'frames':frame,'logical_seconds':frame/30,
'wall_seconds_before_exit':round(time.perf_counter()-start,6),
'program_sha256':hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
'sequence':sequence,'events':events,'snapshot':c.snapshot()
},ensure_ascii=False,indent=2)+'\n',encoding='utf-8')
pyxel.quit()
pyxel.init(720,500,title='Box Weather - Step '+str(STAGE),fps=30,headless=headless)
pyxel.run(update,draw)
if __name__=='__main__':
import argparse
parser=argparse.ArgumentParser(description=__doc__)
group=parser.add_mutually_exclusive_group()
group.add_argument('--headless-smoke',action='store_true')
group.add_argument('--headless-replay',help='L/R/U/D arrows; X/Z; Q reset; 1/2/3 presets')
parser.add_argument('--screenshot');parser.add_argument('--report')
args=parser.parse_args()
headless=args.headless_smoke or args.headless_replay is not None
if not headless and (args.screenshot or args.report):parser.error('Outputs require headless mode')
seq=args.headless_replay or ''
if any(k not in REPLAY for k in seq):parser.error('Unsupported replay character')
main(headless,seq,args.screenshot,args.report)仕組みがわかったら、元の完成版を読んでみよう。
元のゲームの面C「水門に一滴を残す」。S・F・K・gate の容量は、3・2・1・流れの道（ためない）。水の合計3、雲の容量3、列は0〜6。Fに2滴、しかもKに1滴で成功。古い配列から1滴ずつ動かし、合流した水は全部足して、次の刻みにあふれさせる。1滴単位で数える架空の模型で、本物の天気や水の流れではない。
ここまでに作ったのは、元のゲームの仕組みだけを取り出して作り直した「簡略版」です。元のゲームのソースは手を加えずに、別のZIP（元の完成版ソース）に入れてあります。
| 比べる点 | 5ステップの教材 | 元の完成版 |
|---|---|---|
| 対象 | 01は水の持ち主、02は見通し、03は水の移動の実験。04と05で、元のゲームの面Cのふつうのルール。 | A・B・Cの3面と、門をいつも開いておく比較モード。 |
| 観察用配置 | 02は、元のゲームの途中の3つの状態。03の「gate に3滴」「古いK」「合流」は実験用で、面Cを遊ぶ手順ではない。03の合流の例だけ、水の合計が4。 | ふつうの操作では、安定した状態から1つ行動するたびに、流れが止まるまで進める。合流と古いKの例は、元のゲームの test_model.py にある。 |
| 入力とタイミング | 1手の結果はすぐに出る。キーを全部離すまで次の入力を受け付けず、Rをふくむ同時押し（和音）も受け付けない。05で Undo。 | trace の各配列を4フレームずつ表示。水の移動中に押した矢印キーはためておかず、キーを離したことを検出する。Z・R・P は移動中も使える。タイトル、P・M・N、F12 もある。 |
| 描画と音 | 720×500の画面に、図形と英字（ASCII）で描く。x は0から数える。配列と水の量をいつも表示。日本語フォント・音・外部の画像はない。 | 384×432の画面、日本語のビットマップ文字、画面の列番号は1から数える、記録した動画がある。音はない。 |
| 権利 | 元のゲームの MIT ライセンスの全文を、すべての .py ファイルに書きこみ、元の文のままの LICENSE と、参照先の NOTO-LICENSE を同梱。文字の形のデータ（字形）は使わない。 | コードと描画は MIT、Noto をもとにした文字は SIL OFL。 |
| 確認範囲 | 元のゲームの面Cでたどりつける85の状態、340通りの変化、途中の配列を照らし合わせた。25の水の移動の配置について、門と合流をすべての順番で確かめた。本物の Pyxel を画面なし（headless）で動かし、プログラムで作った入力と、1つのファイルだけで配って動くことを確かめた。 | 元のゲームそのものの25のテストと fresh demo は、別の確認の記録。教材のテストの数には入れない。 |
完成版のコードで、次に読むところ
model.pyのLEVELSに面AとBを加え、roof・inlet・pipeの通り道を画面でたどる。app.pyのanimation・anim_frameを使い、保存したtraceを4フレームずつ表示する。表示のために水を別に作らない。app.pyのupdateで、Z・R・P、action_lock、キーを全部離すまでの待ちの優先順を比べ、元のゲームの操作まで広げる。solver.pyのkeyで手数をのぞいた安定状態を探し、面Cの最短の手順が10行動であることを確かめる。gate_linked=Falseの比較を加え、Kの1滴が成功の条件にどう影響するかを確かめる。
自分のゲームに変えるなら
最初は、数字かルールを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の実際のパソコンでの操作や、音が出るかどうかはまだ確認していません。人がすべての操作を試したわけでもありません。ブラウザー版の確認範囲は確認状況をご覧ください。
うまく動かないときは：起動やエラーの対処法 / 質問用のメモを作る