パズルLESSON 58一刻ずつ進む水路・着火予防の架空パズル原作046

消防水路の作り方

消えた水は、蒸気として帳簿に残します。
小さな水路の模型で、水が1刻(時間の1区切り)に1マスずつ流れる様子と、火を消す仕組みを作ります。水そのものと、水が通ったあとに残る濡れは、分けて更新します。

完成版の遊び方25マスの架空の模型で、家を火から守ります。流れ下る水、蒸気、残った濡れ、火が自然に消えることを使います。現実の消火の手順ではありません。

消防水路の元の完成版。自動操作によるPyxelの実描画画面。
元の完成版の画面です。この講座では、仕組みがわかりやすいように小さくした「教材版」を、5つのステップで作ります。
PLAY / 試験公開

完成サンプルをブラウザーで遊ぶ

この講座のもとになった完成版のゲームを、ブラウザーでそのまま遊べます。キーボードで操作するので、PCで開いてください(スマホのタッチ操作には対応していません)。初めて開くときは、読み込みに少し時間がかかります。

完成版の主な操作:左右キーで放水口を選ぶ、上下キーで水の量を選ぶ、Zで放水、Spaceで待つ、Xで戻す、Rで再開、Pで一時停止、1〜3で状況を選ぶ

ボタンを押すと、ゲームの読み込みが始まります。

画面に「CLICK TO START」と出たらクリックします。続けてゲーム画面を一度クリックすると、キー操作が効くようになります。ゲームを終えるときは、枠の外にある「閉じる」ボタンを使ってください(Escキーの働きはゲームごとに違います)。ゲーム内のセーブは、閉じる・ページを移る・再読み込みすると消えます。

うまく表示されないとき

ゲームを動かすためのプログラム(Pyxel 2.9.9)は、外部の配信サービス(jsDelivr)から読み込みます。表示されないときは「最初から再読込」を押すか、完成版のソースをダウンロードしてPCで実行してください。ブラウザー版の録画機能は止めています。ブラウザー版で確認できている範囲は検証状況にまとめています。

同じPyxelアプリをダウンロード(.pyxapp) / 別のタブで開く

水が動く順番と使われる順番を、水の合計が変わらないこと(保存則)で確かめる。

いきなり完成版のコードを全部読む必要はありません。小さなゲームにルールを1つずつ足しながら、5つのステップで作っていきます。どのステップのファイルもそれだけで動くので、動きを確かめてから次へ進みましょう。

1. 古い配列を読んで、新しい配列に書く
2. 受け付けるかの判定、放水、水の合流、排水
3. 水が蒸気に移ることと、前からあった火だけが燃え広がること
それだけで動く小さなプログラムを、5つのステップで少しずつ大きくしていきます。
必要な環境
Python 3.11以上 / Pyxel 2.9.9
知っておくとよいこと
変数・if・for・関数・リストの基本 / 添字(何番目かを表す番号)・タプル・集合・dataclassは本文で説明します / 二重バッファ・書き換えないState・Undoについて、前もって知っておく必要はありません
教材に入っているもの
ステップごとに動くPythonファイル5本・説明・動作確認用のテスト
完成した教材の操作
左右:02以降、放水口WとEを選ぶ / 上・下:02以降、水の量2と1を選ぶ / Z:02以降、放水して1刻進む / Space:放水せずに1刻進む / X:05のUndo(元に戻す) / 1・2・3:05で状況A・B・Cを切り替えて最初に戻す / R:今の状況を最初に戻す / Esc:終了
難易度の目安
中級このページで使うプログラムの考え方の数と難しさから決めた目安です。難易度について
学習順に迷ったら
おすすめの学ぶ順番を見る
この講座の進め方① ページ下の教材ZIPをダウンロードして、すべて展開します。② はじめる準備を見ながら、PythonとPyxelを使える状態にします。③ まず 01_flow.py を、何も変えずにそのまま実行してみましょう。記事に載せているコードは説明用の一部分なので、実行するときはZIPに入っているファイルを使ってください。
コードを変更したら

コードを書き換えたら、ファイルを保存し、動いているゲームを閉じてから、もう一度同じ.pyを実行します。ゲーム中にRキーなどで「やり直し」をしても、書き換えたコードは読み込まれません。

教材のゲーム画面は、日本語フォントを用意しなくても動くように英語で表示しています。ファイルには動作確認用のコードも入っていますが、最初は Game / State・update・draw の3つに注目すれば十分です。わからない言葉は用語集で調べられます。class・self・dataclass はPythonの状態、キー入力と時間の扱いは入力・更新・描画、マス目とピクセルの違いは画面とマップの座標のページで詳しく説明しています。

STEP 01 / 05

水を一刻に一マスだけ動かす

古い配列を読んで、新しい配列に書く。

最初は、火も放水の操作も作りません。Wの位置(1,0)に置いた水4単位を、下へ流します。座標は、左上を(0,0)、右下を(4,4)とします。Spaceを押すたびにT(時刻)が1だけ増え、水は1行だけ下がります。キーを離してから、次を押しましょう。何も押さずに待っていても、Tは増えません。

25マスを、water という長さ25のタプルで持ちます。ix(x,y) の y*5+x は、2次元の座標を1次元の添字に変えます。xy(i) は、%(割り算の余り)と //(割り算の商)で座標に戻します。画面の青いW4は、水の量です。火の燃料や濡れを表す数字ではありません。

水を動かしている途中で water 自身を書き換えると、上から下へ調べる for の後半で、もう動かした水をもう一度読んでしまいます。すると、1刻で一番下の段まで落ちることがあります。そこで、moved という空の配列を作り、古い water だけを読んで moved に足します。これが二重バッファ(読む配列と書く配列を分ける方法)です。最後に新しい State を返すので、描画は状態を変えません。

下のはしから出た水は、drain(排水)に移します。水が画面から消えても、帳簿から消えてはいけません。この段階の式は sum(water)+drain=4 です。貯水槽や蒸気はまだないので、右側のSTOCKとSTEAMは0と表示されます。水が全部排水されたあとも、練習として刻を進められます。元のゲームの「火が消えたら終わる」というルールは、03から入れます。

下のコードは 01_flow.py の一部です。前のステップから追加・変更したところだけを抜き出しています。

01_flow.py / 注目する部分
    moved = [0] * 25
    drain = state.drain
    for i, amount in enumerate(state.water):

実行して確かめる

教材のZIPを展開したフォルダーで、次のコマンドを実行します。

Windows
.venv\Scripts\python.exe 01_flow.py
macOS / Linux
.venv/bin/python 01_flow.py
画面で見ることSpaceを2回押すとT2になり、(1,2)にW4があります。5回押すとT5で、GRID 0・DRAIN 4・TOTAL 4/4になります。
  • 座標と添字を行き来できる
  • 古い配列と新しい配列を分ける
  • 排水も含めた水の合計を変えない
つまずきやすいところfor で読んでいる配列に移動先を書き戻すと、水が同じ刻に何度も動いてしまいます。moved に書き、water は最後まで読むだけにします。
少し変えてみよう

START_AMOUNT = 4 を2に変えて、起動し直してみましょう。Spaceを5回押すと、DRAIN 2・TOTAL 2/2になります。終わったら4に戻します。

01_flow.pyをダウンロードこの段階だけで実行できます
このステップのコード全文(219行)
01_flow.py / 全文
# MIT License
#
# Copyright (c) 2026 OpenAI-assisted game 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.
#
# This MIT license covers original source, documentation and primitive artwork.
# Bitmap glyph data is derived from Noto Sans CJK and remains under SIL Open
# Font License 1.1. See assets/NOTO-LICENSE.txt for the separate font license.

"""消防水路・教材1/5。架空の格子規則。現実の消火手順ではありません。"""
from dataclasses import dataclass, replace

N = 5
WAIT = (-1, 0)
ACTIONS = (WAIT, (0, 1), (0, 2), (1, 1), (1, 2))

def ix(x, y):
    return y * N + x

def xy(i):
    return i % N, i // N

START_AMOUNT = 4

@dataclass(frozen=True)
class Level:
    title: str = 'FLOW'
    stock: int = START_AMOUNT
    outlets: tuple = (ix(1, 0),)

LEVELS = (Level(),)

@dataclass(frozen=True)
class State:
    t: int
    water: tuple
    drain: int = 0

def initial(level):
    water = [0] * 25
    water[level.outlets[0]] = level.stock
    return State(0, tuple(water))

def apply(state, level, action):
    if action != WAIT:
        return state, 'invalid'
    moved = [0] * 25
    drain = state.drain
    for i, amount in enumerate(state.water):
        if not amount:
            continue
        x, y = xy(i)
        if y == N - 1:
            drain += amount
        else:
            moved[ix(x, y + 1)] += amount
    return State(state.t + 1, tuple(moved), drain), 'ok'

def invariant(state, level):
    assert sum(state.water) + state.drain == level.stock
    assert all(n >= 0 for n in state.water) and state.drain >= 0


class App:
    def __init__(self):
        self.level_index = 0
        self.reset()

    @property
    def level(self):
        return LEVELS[self.level_index]

    def reset(self):
        self.state = initial(self.level)
        self.outlet = 0
        self.amount = 1
        self.gated = False
        self.message = 'READY'

    
    def control(self, pressed, held):
        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
        key = next(iter(pressed))
        
        if key != 'SPACE':
            return
        if key == 'SPACE':
            action = WAIT
        elif key == 'Z':
            action = (self.outlet, self.amount)
        else:
            return
        self.state, self.message = apply(self.state, self.level, action)
        self.gated = True

    def draw(self, p):
        p.cls(1)
        s, level = self.state, self.level
        p.text(10, 8, 'FIRE WATERWAYS / STAGE 1', 10)
        outcome = getattr(s, 'outcome', 'practice').upper()
        p.text(10, 22, 'T '+str(s.t)+'  LEVEL '+chr(65+self.level_index)+'  '+outcome, 7)
        p.text(10, 36, 'FICTIONAL GRID RULES ONLY', 6)
        for i in range(25):
            x, y = xy(i)
            px, py = 12+x*40, 62+y*40
            p.rect(px, py, 38, 38, 5)
            if i in getattr(level, 'stones', ()):
                p.rect(px+2, py+2, 34, 34, 13)
            if getattr(s, 'burned', (False,)*25)[i]:
                p.rect(px+2, py+2, 34, 34, 0)
            wet = getattr(s, 'wet', (0,)*25)[i]
            if wet:
                for dx in (8,18,28):p.pset(px+dx, py+32, 12)
            water = s.water[i]
            if water:
                p.rect(px+3, py+17, 31, 12, 12)
                p.text(px+5, py+20, 'W'+str(water), 7)
            fire = getattr(s, 'fire', (0,)*25)[i]
            if fire:
                p.tri(px+18, py+8, px+7, py+28, px+30, py+28, 9)
                p.text(px+15, py+20, str(fire), 0)
            houses = getattr(level, 'houses', ())
            if i in houses:
                damage = s.damaged[houses.index(i)]
                p.rectb(px+24, py+3, 11, 10, 8 if damage else 11)
                p.text(px+27, py+5, 'H', 8 if damage else 11)
            if hasattr(level, 'heights'):
                p.text(px+2, py+2, str(level.heights[i]), 6)
            if i in level.outlets:
                label = 'W' if level.outlets.index(i)==0 else 'E'
                p.text(px+14, py+4, label, 10)
        p.text(12, 266, 'SOUTH EDGE -> DRAIN', 6)
        p.text(232, 62, 'WATER LEDGER', 10)
        ledger = (('STOCK', getattr(s, 'stock', 0)), ('GRID', sum(s.water)),
                  ('STEAM', getattr(s, 'steam', 0)), ('DRAIN', s.drain))
        for row, (label, value) in enumerate(ledger):
            p.text(232, 78+row*14, label+' '+str(value), 7)
        p.text(232, 136, 'TOTAL '+str(sum(v for _, v in ledger))+' / '+str(level.stock), 11)
        p.text(232, 152, 'WET DOTS ARE NOT WATER', 6)
        p.text(232, 176, 'SPACE: ADVANCE ONE TICK', 7)
        p.text(10, 284, self.message.upper(), 7)
        p.text(10, 300, 'SPACE: ONE TICK / WATER STARTS ON THE GRID', 6)
        p.text(10, 314, 'NO FIRE / NO TIME LIMIT', 6)
        p.text(10, 328, 'R RESET  ESC QUIT  RELEASE KEYS BETWEEN TURNS', 6)

KEYS = ('SPACE', 'R')
DEMO_KEYS = ('SPACE', 'SPACE')

def parse_args(argv=None):
    import argparse
    parser = argparse.ArgumentParser(description='Five-stage fictional grid tutorial')
    parser.add_argument('--headless-smoke', action='store_true')
    parser.add_argument('--demo-checkpoint', action='store_true')
    parser.add_argument('--screenshot')
    args = parser.parse_args(argv)
    if (args.demo_checkpoint or args.screenshot is not None) and not args.headless_smoke:
        parser.error('--demo-checkpoint/--screenshot require --headless-smoke')
    if args.screenshot is not None and not args.screenshot.lower().endswith('.png'):
        parser.error('--screenshot requires a .png path')
    return args

def main(args):
    from pathlib import Path
    import pyxel as p  # 引数を検証した後で初めて読み込む
    screenshot = str(Path(args.screenshot).resolve()) if args.screenshot else None
    if screenshot:
        Path(screenshot).parent.mkdir(parents=True, exist_ok=True)
    app = App()
    p.init(480, 352, title='Fire Waterways 1', fps=30,
           display_scale=2, headless=args.headless_smoke)
    keys = {k:getattr(p, 'KEY_'+k) for k in KEYS}
    limit = max(12, 2*len(DEMO_KEYS)+3) if args.demo_checkpoint else 12
    def update():
        if args.headless_smoke and p.frame_count >= limit:
            if screenshot:
                p.screenshot(screenshot, scale=2)
            p.quit()
            return
        if args.headless_smoke and args.demo_checkpoint:
            index = p.frame_count // 2
            pressed = {DEMO_KEYS[index]} if 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)})
    p.run(update, lambda:app.draw(p))

if __name__ == '__main__':
    main(parse_args())
STEP 02 / 05

有限の貯水と低い地形を加える

受け付けるかの判定、放水、水の合流、排水。

始めの水4は、盤面ではなく貯水槽のSTOCKに置きます。左右キーで放水口のWかEを選び、上キーで2単位、下キーで1単位を選びます。Zを押すと放水して、1刻進みます。Spaceは、水を使わずに1刻進めます。選ぶ操作だけでは、時刻も水の量も変わりません。右側には、今の選択でZを押したあとのTとSTOCKを予告します。apply は純粋な関数(同じ入力なら必ず同じ結果を返し、ほかを書き換えない関数)なので、予告しても実際の水は減りません。

apply では、まず、その行動が許された組み合わせかを調べます。次に、必要な水の量と stock を比べます。足りなければ、元の State と、stock という理由を返して終わります。水を一部だけ減らしてから断ると、画面では止まったように見えるのに、水だけが失われます。調べるのは、更新の前にまとめて行います。

flow_to は、上下左右の4つのとなり(四近傍)のうち、今のマスより低いマスだけを候補にします(同じ高さは入りません)。候補のタプルは(高さ, 優先順位, 添字)です。min は、まず高さを比べ、同じなら優先順位を比べます。一番低い高さが同じなら、下→左→右→上の順です。石は候補から外します。低いとなりがなければ、水はその場にたまります。最初の地形は、行ごとに4・3・2・1・0なので、水は下へ流れます。南のはしの排水の候補は、高さ-1です。地形を改造するときは0以上にすると、この教材での排水の意味が保てます。

同じ行き先に2つの水が来たら、moved[j] += amount_here で足し合わせます。1マスに入る量の上限を勝手に決めて、あふれた水を捨てることはしません。水は全部で4なので、最初の設定では1マスに最大4が入ります。この段階の保存式(合計が変わらないことを表す式)は stock+sum(water)+drain=4 です。水の合計は増えません。貯水から盤面へ、盤面から排水へと、居場所が変わるだけです。火や勝ち負けは、まだありません。

下のコードは 02_stock.py の一部です。前のステップから追加・変更したところだけを抜き出しています。

02_stock.py / 注目する部分
            if j < 0:
                drain += amount_here
            else:
                moved[j] += amount_here

実行して確かめる

教材のZIPを展開したフォルダーで、次のコマンドを実行します。

Windows
.venv\Scripts\python.exe 02_stock.py
macOS / Linux
.venv/bin/python 02_stock.py
画面で見ることR→上→Z→Space→Spaceと押すと、T3、STOCK 2、(1,3)にW2、DRAIN 0になります。さらにSpaceを2回押すと、DRAIN 2です。
  • 水が足りないときは、更新の前に断る
  • 低さを先に比べる
  • 合流した水は足し算でまとめ、合計を変えない
つまずきやすいところ優先順位だけで下を選ぶと、左にもっと低い場所があるときのルールが変わってしまいます。先に高さを比べ、高さが同じときに方向を比べます。
少し変えてみよう

Level の stock: int = 4 を6に変えて、起動し直してみましょう。上→Z→Z→Zと、1回ずつキーを離して押すと、T3、STOCK 0、GRID 6になります。4回目のZは、理由STOCKで断られ、T3のままです。終わったら4に戻します。

02_stock.pyをダウンロードこの段階だけで実行できます
このステップのコード全文(247行)
02_stock.py / 全文
# MIT License
#
# Copyright (c) 2026 OpenAI-assisted game 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.
#
# This MIT license covers original source, documentation and primitive artwork.
# Bitmap glyph data is derived from Noto Sans CJK and remains under SIL Open
# Font License 1.1. See assets/NOTO-LICENSE.txt for the separate font license.

"""消防水路・教材2/5。架空の格子規則。現実の消火手順ではありません。"""
from dataclasses import dataclass, replace

N = 5
WAIT = (-1, 0)
ACTIONS = (WAIT, (0, 1), (0, 2), (1, 1), (1, 2))

def ix(x, y):
    return y * N + x

def xy(i):
    return i % N, i // N

@dataclass(frozen=True)
class Level:
    title: str = 'A'
    outlets: tuple = (ix(1, 0), ix(3, 0))
    stock: int = 4
    heights: tuple = tuple(4 - i // 5 for i in range(25))
    stones: tuple = ()

LEVELS = (Level(),)

@dataclass(frozen=True)
class State:
    t: int
    stock: int
    water: tuple
    drain: int = 0

def initial(level):
    return State(0, level.stock, (0,) * 25)

def flow_to(level, i):
    """低い隣へ一歩。同じ最低高度なら下・左・右・上。"""
    x, y = xy(i)
    options = []
    for rank, (dx, dy) in enumerate(((0, 1), (-1, 0), (1, 0), (0, -1))):
        xx, yy = x + dx, y + dy
        if yy == N and 0 <= xx < N:
            options.append((-1, rank, -1))  # 南端の排水口
        elif 0 <= xx < N and 0 <= yy < N:
            j = ix(xx, yy)
            if j not in level.stones and level.heights[j] < level.heights[i]:
                options.append((level.heights[j], rank, j))
    return min(options)[2] if options else i

def apply(state, level, action):
    if action not in ACTIONS:
        return state, 'invalid'
    outlet, amount = action
    if amount > state.stock:
        return state, 'stock'
    water = list(state.water)
    if outlet >= 0:
        water[level.outlets[outlet]] += amount
    moved = [0] * 25
    drain = state.drain
    for i, amount_here in enumerate(water):
        if amount_here:
            j = flow_to(level, i)
            if j < 0:
                drain += amount_here
            else:
                moved[j] += amount_here
    return State(state.t + 1, state.stock - amount, tuple(moved), drain), 'ok'

def invariant(state, level):
    assert state.stock + sum(state.water) + state.drain == level.stock
    assert 0 <= state.stock <= level.stock
    assert all(n >= 0 for n in state.water) and state.drain >= 0


class App:
    def __init__(self):
        self.level_index = 0
        self.reset()

    @property
    def level(self):
        return LEVELS[self.level_index]

    def reset(self):
        self.state = initial(self.level)
        self.outlet = 0
        self.amount = 1
        self.gated = False
        self.message = 'READY'

    
    def control(self, pressed, held):
        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
        key = next(iter(pressed))
        
        if key in ('LEFT', 'RIGHT', 'UP', 'DOWN'):
            if key == 'LEFT':self.outlet = 0
            if key == 'RIGHT':self.outlet = 1
            if key == 'UP':self.amount = 2
            if key == 'DOWN':self.amount = 1
            self.message = 'SELECTED'
            self.gated = True
            return
        if key == 'SPACE':
            action = WAIT
        elif key == 'Z':
            action = (self.outlet, self.amount)
        else:
            return
        self.state, self.message = apply(self.state, self.level, action)
        self.gated = True

    def draw(self, p):
        p.cls(1)
        s, level = self.state, self.level
        p.text(10, 8, 'FIRE WATERWAYS / STAGE 2', 10)
        outcome = getattr(s, 'outcome', 'practice').upper()
        p.text(10, 22, 'T '+str(s.t)+'  LEVEL '+chr(65+self.level_index)+'  '+outcome, 7)
        p.text(10, 36, 'FICTIONAL GRID RULES ONLY', 6)
        for i in range(25):
            x, y = xy(i)
            px, py = 12+x*40, 62+y*40
            p.rect(px, py, 38, 38, 5)
            if i in getattr(level, 'stones', ()):
                p.rect(px+2, py+2, 34, 34, 13)
            if getattr(s, 'burned', (False,)*25)[i]:
                p.rect(px+2, py+2, 34, 34, 0)
            wet = getattr(s, 'wet', (0,)*25)[i]
            if wet:
                for dx in (8,18,28):p.pset(px+dx, py+32, 12)
            water = s.water[i]
            if water:
                p.rect(px+3, py+17, 31, 12, 12)
                p.text(px+5, py+20, 'W'+str(water), 7)
            fire = getattr(s, 'fire', (0,)*25)[i]
            if fire:
                p.tri(px+18, py+8, px+7, py+28, px+30, py+28, 9)
                p.text(px+15, py+20, str(fire), 0)
            houses = getattr(level, 'houses', ())
            if i in houses:
                damage = s.damaged[houses.index(i)]
                p.rectb(px+24, py+3, 11, 10, 8 if damage else 11)
                p.text(px+27, py+5, 'H', 8 if damage else 11)
            if hasattr(level, 'heights'):
                p.text(px+2, py+2, str(level.heights[i]), 6)
            if i in level.outlets:
                label = 'W' if level.outlets.index(i)==0 else 'E'
                p.text(px+14, py+4, label, 10)
        p.text(12, 266, 'SOUTH EDGE -> DRAIN', 6)
        p.text(232, 62, 'WATER LEDGER', 10)
        ledger = (('STOCK', getattr(s, 'stock', 0)), ('GRID', sum(s.water)),
                  ('STEAM', getattr(s, 'steam', 0)), ('DRAIN', s.drain))
        for row, (label, value) in enumerate(ledger):
            p.text(232, 78+row*14, label+' '+str(value), 7)
        p.text(232, 136, 'TOTAL '+str(sum(v for _, v in ledger))+' / '+str(level.stock), 11)
        p.text(232, 152, 'WET DOTS ARE NOT WATER', 6)
        p.text(232, 176, 'SELECT '+('W' if self.outlet==0 else 'E')+' / '+str(self.amount), 10)
        after, reason = apply(s, level, (self.outlet, self.amount))
        p.text(232, 192, 'Z -> T'+str(after.t)+' STOCK '+str(after.stock), 7)
        p.text(232, 206, 'PREVIEW '+reason.upper(), 6)
        p.text(10, 284, self.message.upper(), 7)
        p.text(10, 300, 'LEFT/RIGHT OUTLET   UP=2 DOWN=1   Z RELEASE', 6)
        p.text(10, 314, 'NO FIRE / NO TIME LIMIT', 6)
        p.text(10, 328, 'R RESET  ESC QUIT  RELEASE KEYS BETWEEN TURNS', 6)

KEYS = ('LEFT', 'RIGHT', 'UP', 'DOWN', 'Z', 'SPACE', 'R')
DEMO_KEYS = ('UP', 'Z', 'SPACE', 'SPACE')

def parse_args(argv=None):
    import argparse
    parser = argparse.ArgumentParser(description='Five-stage fictional grid tutorial')
    parser.add_argument('--headless-smoke', action='store_true')
    parser.add_argument('--demo-checkpoint', action='store_true')
    parser.add_argument('--screenshot')
    args = parser.parse_args(argv)
    if (args.demo_checkpoint or args.screenshot is not None) and not args.headless_smoke:
        parser.error('--demo-checkpoint/--screenshot require --headless-smoke')
    if args.screenshot is not None and not args.screenshot.lower().endswith('.png'):
        parser.error('--screenshot requires a .png path')
    return args

def main(args):
    from pathlib import Path
    import pyxel as p  # 引数を検証した後で初めて読み込む
    screenshot = str(Path(args.screenshot).resolve()) if args.screenshot else None
    if screenshot:
        Path(screenshot).parent.mkdir(parents=True, exist_ok=True)
    app = App()
    p.init(480, 352, title='Fire Waterways 2', fps=30,
           display_scale=2, headless=args.headless_smoke)
    keys = {k:getattr(p, 'KEY_'+k) for k in KEYS}
    limit = max(12, 2*len(DEMO_KEYS)+3) if args.demo_checkpoint else 12
    def update():
        if args.headless_smoke and p.frame_count >= limit:
            if screenshot:
                p.screenshot(screenshot, scale=2)
            p.quit()
            return
        if args.headless_smoke and args.demo_checkpoint:
            index = p.frame_count // 2
            pressed = {DEMO_KEYS[index]} if 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)})
    p.run(update, lambda:app.draw(p))

if __name__ == '__main__':
    main(parse_args())
STEP 03 / 05

火と水を順序付きで更新する

水が蒸気に移ることと、前からあった火だけが燃え広がること。

ここから、元のゲームのAと同じ場所に火を置きます。だいだい色の1や2は、火の燃料です。水1単位で火を1マス消し、その水はSTEAM(蒸気)に1移ります。火の燃料が2でも、必要な水は1です。これは、整数で数える架空のゲームのルールです。現実の消火、防火、避難の方法ではありません。

順番は、放水→今いるマスで消火→残った水を1歩動かす→着いた先で消火、です。動く前に蒸気になった水は、下流へは行きません。一方、水が2単位ある場所なら1を使い、残った1だけが先へ進みます。消火は、水を帳簿から消す処理ではありません。water から steam へ移す処理です。保存式は stock+sum(water)+steam+drain=4 になります。

刻の初めに、火があった添字を old_fire に記録します。消火のあとも残った古い火(旧火)だけが、風下のとなり1マスへの着火を予約します。次に旧火の燃料を1減らし、そのあとで新しい火を燃料2でまとめて作ります。新しい火を、すぐに for でもう一度処理してはいけません。燃え尽きたマスは灰になり、もう燃えません。水で消えたマスには、灰を付けません。

この段階では、水が通ったあとの濡れ(濡れ残り)はまだ作りません。W1→待つでは、着いた水が火を消した直後に、となりで残った旧火から、同じマスに火がつくことがあります。W2→待つなら、残った水1が着火を防ぎます。火が全部消えたら、CLEARで止まります。残っている水を、むりやり排水することはしません。家を守る目標は次の段階で入れるので、CLEARは家を守れたという意味ではありません。

下のコードは 03_steam.py の一部です。前のステップから追加・変更したところだけを抜き出しています。

03_steam.py / 注目する部分
    old_fire = {i for i, n in enumerate(state.fire) if n}
    for i in range(25):
        if water[i] and fire[i]:
            water[i] -= 1
            steam += 1
            fire[i] = 0

実行して確かめる

教材のZIPを展開したフォルダーで、次のコマンドを実行します。

Windows
.venv\Scripts\python.exe 03_steam.py
macOS / Linux
.venv/bin/python 03_steam.py
画面で見ることR→Z→Spaceと押すと、T2・STOCK 3・STEAM 1になります。ただし、(1,2)に燃料2の火がまた現れ、PLAYING(遊んでいる途中)のままです。濡れ残りがない段階なので、これはねらいどおりの結果です。
  • 消火の前とあとの水の量を、帳簿に残す
  • 前からあった火だけが燃え広がる
  • 水で消えた草と、灰を区別する
つまずきやすいところ蒸気になった1単位を移動にも使うと、同じ水を2回使うことになります。また、新しい火を同じ刻に燃え広がらせると、1列全部に一瞬で広がってしまいます。
少し変えてみよう

R→上→Z→Spaceと押してみましょう。水2のうち1が蒸気になり、1が(1,2)に残って、T2・STOCK 2・GRID 1・STEAM 1のCLEARになります。R→Z→Spaceとのちがいを説明してみましょう。

03_steam.pyをダウンロードこの段階だけで実行できます
このステップのコード全文(299行)
03_steam.py / 全文
# MIT License
#
# Copyright (c) 2026 OpenAI-assisted game 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.
#
# This MIT license covers original source, documentation and primitive artwork.
# Bitmap glyph data is derived from Noto Sans CJK and remains under SIL Open
# Font License 1.1. See assets/NOTO-LICENSE.txt for the separate font license.

"""消防水路・教材3/5。架空の格子規則。現実の消火手順ではありません。"""
from dataclasses import dataclass, replace

N = 5
WAIT = (-1, 0)
ACTIONS = (WAIT, (0, 1), (0, 2), (1, 1), (1, 2))

def ix(x, y):
    return y * N + x

def xy(i):
    return i % N, i // N

@dataclass(frozen=True)
class Level:
    title: str = 'A'
    outlets: tuple = (ix(1, 0), ix(3, 0))
    fires: tuple = (ix(0, 2),)
    winds: tuple = ('E',) * 6
    stock: int = 4
    heights: tuple = tuple(4 - i // 5 for i in range(25))
    stones: tuple = ()

LEVELS = (Level(),)

DIRS = {'E': (1, 0), 'W': (-1, 0), 'S': (0, 1), 'N': (0, -1)}

@dataclass(frozen=True)
class State:
    t: int
    stock: int
    water: tuple
    fire: tuple
    burned: tuple
    steam: int = 0
    drain: int = 0
    outcome: str = 'playing'

def initial(level):
    fire = [0] * 25
    for i in level.fires:
        fire[i] = 2
    return State(0, level.stock, (0,) * 25, tuple(fire), (False,) * 25)

def flow_to(level, i):
    """低い隣へ一歩。同じ最低高度なら下・左・右・上。"""
    x, y = xy(i)
    options = []
    for rank, (dx, dy) in enumerate(((0, 1), (-1, 0), (1, 0), (0, -1))):
        xx, yy = x + dx, y + dy
        if yy == N and 0 <= xx < N:
            options.append((-1, rank, -1))  # 南端の排水口
        elif 0 <= xx < N and 0 <= yy < N:
            j = ix(xx, yy)
            if j not in level.stones and level.heights[j] < level.heights[i]:
                options.append((level.heights[j], rank, j))
    return min(options)[2] if options else i

def wind(level, t):
    return level.winds[min(t, len(level.winds) - 1)]

def apply(state, level, action):
    if state.outcome != 'playing':
        return state, 'terminal'
    if action not in ACTIONS:
        return state, 'invalid'
    outlet, amount = action
    if amount > state.stock:
        return state, 'stock'
    water, fire, burned = list(state.water), list(state.fire), list(state.burned)
    steam, drain = state.steam, state.drain
    if outlet >= 0:
        water[level.outlets[outlet]] += amount
    old_fire = {i for i, n in enumerate(state.fire) if n}
    for i in range(25):
        if water[i] and fire[i]:
            water[i] -= 1
            steam += 1
            fire[i] = 0
    moved = [0] * 25
    for i, n in enumerate(water):
        if n:
            j = flow_to(level, i)
            if j < 0:
                drain += n
            else:
                moved[j] += n
    water = moved
    for i in range(25):
        if water[i] and fire[i]:
            water[i] -= 1
            steam += 1
            fire[i] = 0
    dx, dy = DIRS[wind(level, state.t)]
    ignitions = set()
    for i in old_fire:
        if not fire[i]:
            continue
        x, y = xy(i)
        xx, yy = x + dx, y + dy
        if 0 <= xx < N and 0 <= yy < N:
            j = ix(xx, yy)
            if not fire[j] and not burned[j] and not water[j] and j not in level.stones:
                ignitions.add(j)
    for i in old_fire:
        if fire[i]:
            fire[i] -= 1
            if not fire[i]:
                burned[i] = True
    for j in ignitions:
        fire[j] = 2
    outcome = 'playing' if any(fire) else 'clear'
    return State(state.t + 1, state.stock - amount, tuple(water), tuple(fire),
                 tuple(burned), steam, drain, outcome), 'ok'

def invariant(state, level):
    assert state.stock + sum(state.water) + state.steam + state.drain == level.stock
    assert 0 <= state.stock <= level.stock
    assert state.steam >= 0 and state.drain >= 0
    assert all(0 <= n <= level.stock for n in state.water)
    assert all(n in (0, 1, 2) for n in state.fire)
    assert not any(state.fire[i] and state.burned[i] for i in range(25))


class App:
    def __init__(self):
        self.level_index = 0
        self.reset()

    @property
    def level(self):
        return LEVELS[self.level_index]

    def reset(self):
        self.state = initial(self.level)
        self.outlet = 0
        self.amount = 1
        self.gated = False
        self.message = 'READY'

    
    def control(self, pressed, held):
        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
        key = next(iter(pressed))
        
        if key in ('LEFT', 'RIGHT', 'UP', 'DOWN'):
            if key == 'LEFT':self.outlet = 0
            if key == 'RIGHT':self.outlet = 1
            if key == 'UP':self.amount = 2
            if key == 'DOWN':self.amount = 1
            self.message = 'SELECTED'
            self.gated = True
            return
        if key == 'SPACE':
            action = WAIT
        elif key == 'Z':
            action = (self.outlet, self.amount)
        else:
            return
        self.state, self.message = apply(self.state, self.level, action)
        self.gated = True

    def draw(self, p):
        p.cls(1)
        s, level = self.state, self.level
        p.text(10, 8, 'FIRE WATERWAYS / STAGE 3', 10)
        outcome = getattr(s, 'outcome', 'practice').upper()
        p.text(10, 22, 'T '+str(s.t)+'  LEVEL '+chr(65+self.level_index)+'  '+outcome, 7)
        p.text(10, 36, 'FICTIONAL GRID RULES ONLY', 6)
        for i in range(25):
            x, y = xy(i)
            px, py = 12+x*40, 62+y*40
            p.rect(px, py, 38, 38, 5)
            if i in getattr(level, 'stones', ()):
                p.rect(px+2, py+2, 34, 34, 13)
            if getattr(s, 'burned', (False,)*25)[i]:
                p.rect(px+2, py+2, 34, 34, 0)
            wet = getattr(s, 'wet', (0,)*25)[i]
            if wet:
                for dx in (8,18,28):p.pset(px+dx, py+32, 12)
            water = s.water[i]
            if water:
                p.rect(px+3, py+17, 31, 12, 12)
                p.text(px+5, py+20, 'W'+str(water), 7)
            fire = getattr(s, 'fire', (0,)*25)[i]
            if fire:
                p.tri(px+18, py+8, px+7, py+28, px+30, py+28, 9)
                p.text(px+15, py+20, str(fire), 0)
            houses = getattr(level, 'houses', ())
            if i in houses:
                damage = s.damaged[houses.index(i)]
                p.rectb(px+24, py+3, 11, 10, 8 if damage else 11)
                p.text(px+27, py+5, 'H', 8 if damage else 11)
            if hasattr(level, 'heights'):
                p.text(px+2, py+2, str(level.heights[i]), 6)
            if i in level.outlets:
                label = 'W' if level.outlets.index(i)==0 else 'E'
                p.text(px+14, py+4, label, 10)
        p.text(12, 266, 'SOUTH EDGE -> DRAIN', 6)
        p.text(232, 62, 'WATER LEDGER', 10)
        ledger = (('STOCK', getattr(s, 'stock', 0)), ('GRID', sum(s.water)),
                  ('STEAM', getattr(s, 'steam', 0)), ('DRAIN', s.drain))
        for row, (label, value) in enumerate(ledger):
            p.text(232, 78+row*14, label+' '+str(value), 7)
        p.text(232, 136, 'TOTAL '+str(sum(v for _, v in ledger))+' / '+str(level.stock), 11)
        p.text(232, 152, 'WET DOTS ARE NOT WATER', 6)
        p.text(232, 176, 'SELECT '+('W' if self.outlet==0 else 'E')+' / '+str(self.amount), 10)
        after, reason = apply(s, level, (self.outlet, self.amount))
        p.text(232, 192, 'Z -> T'+str(after.t)+' STOCK '+str(after.stock), 7)
        p.text(232, 206, 'PREVIEW '+reason.upper(), 6)
        p.text(232, 222, 'WIND '+wind(level,s.t)+'  FIRE '+str(sum(v>0 for v in s.fire)), 9)
        p.text(10, 284, self.message.upper(), 7)
        p.text(10, 300, 'LEFT/RIGHT OUTLET   UP=2 DOWN=1   Z RELEASE', 6)
        p.text(10, 314, 'SPACE WAIT   NO REAL-WORLD SAFETY ADVICE', 6)
        p.text(10, 328, 'R RESET  ESC QUIT  RELEASE KEYS BETWEEN TURNS', 6)

KEYS = ('LEFT', 'RIGHT', 'UP', 'DOWN', 'Z', 'SPACE', 'R')
DEMO_KEYS = ('Z', 'SPACE')

def parse_args(argv=None):
    import argparse
    parser = argparse.ArgumentParser(description='Five-stage fictional grid tutorial')
    parser.add_argument('--headless-smoke', action='store_true')
    parser.add_argument('--demo-checkpoint', action='store_true')
    parser.add_argument('--screenshot')
    args = parser.parse_args(argv)
    if (args.demo_checkpoint or args.screenshot is not None) and not args.headless_smoke:
        parser.error('--demo-checkpoint/--screenshot require --headless-smoke')
    if args.screenshot is not None and not args.screenshot.lower().endswith('.png'):
        parser.error('--screenshot requires a .png path')
    return args

def main(args):
    from pathlib import Path
    import pyxel as p  # 引数を検証した後で初めて読み込む
    screenshot = str(Path(args.screenshot).resolve()) if args.screenshot else None
    if screenshot:
        Path(screenshot).parent.mkdir(parents=True, exist_ok=True)
    app = App()
    p.init(480, 352, title='Fire Waterways 3', fps=30,
           display_scale=2, headless=args.headless_smoke)
    keys = {k:getattr(p, 'KEY_'+k) for k in KEYS}
    limit = max(12, 2*len(DEMO_KEYS)+3) if args.demo_checkpoint else 12
    def update():
        if args.headless_smoke and p.frame_count >= limit:
            if screenshot:
                p.screenshot(screenshot, scale=2)
            p.quit()
            return
        if args.headless_smoke and args.demo_checkpoint:
            index = p.frame_count // 2
            pressed = {DEMO_KEYS[index]} if 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)})
    p.run(update, lambda:app.draw(p))

if __name__ == '__main__':
    main(parse_args())
STEP 04 / 05

濡れ残と家の被害を記録する

残り時間と、あとから取り消せない達成の判定。

元のゲームのAの、ふつうのモデルを完成させます。wet は水とは別の配列です。水が通ったマスの濡れ残りを、0・1・2で記録します。各刻の最初に古い濡れを1減らし、その刻に水が触れた場所は2に戻します。触れた場所には、消火で水が全部蒸気になった、今いるマスや着いた先も含みます。画面の青い点は濡れです。水を足したり、もう一度使ったりできる資源ではありません。

着火を予約する条件に、not wet[j] を加えます。すると、水が消えても、その場所の濡れが着火を防ぎます。2を付けた刻と、その次の刻は守ります。さらに次の刻は、最初に0になるので守りません。03でもう一度火がついたW1→待つは、ここでは濡れによって止まり、T2でWONになります。水は1単位だけ使い、貯水が3残ります。

家は、level.houses に座標を並べます。damaged には、家ごとに被害があったかどうかを持ちます。一度火がついた家の被害の印は、そのあとで火が消えても False に戻しません。saved は、家の数から被害を受けた数を引くだけです。緑のHはまだ被害がない家、赤のHはもう被害を受けた家です。火が全部なくなったときだけ、守れた家の数が goal 以上ならWON、足りなければLOSTです。

元のゲームにも教材にも、時間制限や、待っていると受ける隠れた罰はありません。大事なのは、火が動く1刻をいつ進めるかです。画面をながめる時間に制限はありません。時刻が大きいというだけで、負けにしてはいけません。成功した状態では、残った水や濡れも、そのまま止めます。予告を何度描いても、時間も貯水も変わりません。終わったあとの放水も断られます。

下のコードは 04_wet_houses.py の一部です。前のステップから追加・変更したところだけを抜き出しています。

04_wet_houses.py / 注目する部分
    for i in touched:
        wet[i] = wet_life

実行して確かめる

教材のZIPを展開したフォルダーで、次のコマンドを実行します。

Windows
.venv\Scripts\python.exe 04_wet_houses.py
macOS / Linux
.venv/bin/python 04_wet_houses.py
画面で見ることR→Z→Spaceと押すと、T2・WON・SAVED 3/3・STOCK 3・GRID 0・STEAM 1になります。青い点(濡れ)は残りますが、TOTALは4/4です。
  • 濡れと水を別々に管理する
  • 刻の最初に、古い濡れを減らす
  • 火が消えたときに、守れた家の数を調べる
つまずきやすいところwet を水の残りの量に足すと、資源が増えてしまいます。濡れは、火がつくかどうかを決めるためだけに使い、保存式には入れません。被害を受けた家は、火が消えても被害を受けたままです。
少し変えてみよう

apply の定義だけを、wet_life=2 から wet_life=0 に変えて、起動し直してみましょう。R→Z→Spaceと押し、火がなくなるまでSpaceを押すと、LOST・SAVED 1/3・STOCK 3になります。W2なら成功します。確かめたら2に戻します。

04_wet_houses.pyをダウンロードこの段階だけで実行できます
このステップのコード全文(319行)
04_wet_houses.py / 全文
# MIT License
#
# Copyright (c) 2026 OpenAI-assisted game 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.
#
# This MIT license covers original source, documentation and primitive artwork.
# Bitmap glyph data is derived from Noto Sans CJK and remains under SIL Open
# Font License 1.1. See assets/NOTO-LICENSE.txt for the separate font license.

"""消防水路・教材4/5。架空の格子規則。現実の消火手順ではありません。"""
from dataclasses import dataclass, replace
WAIT = (-1, 0)
ACTIONS = (WAIT, (0, 1), (0, 2), (1, 1), (1, 2))
N = 5
DIRS = {'E': (1, 0), 'W': (-1, 0), 'S': (0, 1), 'N': (0, -1)}

def ix(x, y):
    return y * N + x

def xy(i):
    return (i % N, i // N)

@dataclass(frozen=True)
class Level:
    title: str
    outlets: tuple
    fires: tuple
    houses: tuple
    winds: tuple
    goal: int
    stock: int = 4
    heights: tuple = tuple((4 - i // 5 for i in range(25)))
    stones: tuple = ()
LEVELS = (Level('下へ残る水', (ix(1, 0), ix(3, 0)), (ix(0, 2),), (ix(2, 2), ix(4, 2), ix(3, 4)), ('E',) * 6, 3),)

@dataclass(frozen=True)
class State:
    t: int
    stock: int
    water: tuple
    wet: tuple
    fire: tuple
    burned: tuple
    damaged: tuple
    steam: int = 0
    drain: int = 0
    outcome: str = 'playing'

def initial(l):
    fire = [0] * 25
    for i in l.fires:
        fire[i] = 2
    return State(0, l.stock, (0,) * 25, (0,) * 25, tuple(fire), (False,) * 25, tuple((i in l.fires for i in l.houses)))

def flow_to(l, i):
    """Strictly lower neighbor. Equal lowest: down,left,right,up. South edge drain."""
    x, y = xy(i)
    options = []
    for rank, (dx, dy) in enumerate(((0, 1), (-1, 0), (1, 0), (0, -1))):
        xx, yy = (x + dx, y + dy)
        if yy == 5 and 0 <= xx < 5:
            options.append((-1, rank, -1))
        elif 0 <= xx < 5 and 0 <= yy < 5:
            j = ix(xx, yy)
            if j not in l.stones and l.heights[j] < l.heights[i]:
                options.append((l.heights[j], rank, j))
    return min(options)[2] if options else i

def wind(l, t):
    return l.winds[min(t, len(l.winds) - 1)]

def saved(s, l):
    return len(l.houses) - sum(s.damaged)

def apply(s, l, a, wet_life=2):
    if s.outcome != 'playing':
        return (s, 'terminal')
    if a not in ((-1, 0), (0, 1), (0, 2), (1, 1), (1, 2)):
        return (s, 'invalid')
    outlet, amount = a
    if amount > s.stock:
        return (s, 'stock')
    water = list(s.water)
    wet = [max(0, n - 1) for n in s.wet]
    fire = list(s.fire)
    burned = list(s.burned)
    damaged = list(s.damaged)
    steam = s.steam
    drain = s.drain
    if outlet >= 0:
        water[l.outlets[outlet]] += amount
    touched = {i for i, n in enumerate(water) if n}
    old_fire = {i for i, n in enumerate(s.fire) if n}
    for i in range(25):
        if water[i] and fire[i]:
            water[i] -= 1
            steam += 1
            fire[i] = 0
    moved = [0] * 25
    for i, n in enumerate(water):
        if n:
            j = flow_to(l, i)
            if j < 0:
                drain += n
            else:
                moved[j] += n
                touched.add(j)
    water = moved
    for i in range(25):
        if water[i] and fire[i]:
            water[i] -= 1
            steam += 1
            fire[i] = 0
    for i in touched:
        wet[i] = wet_life
    dx, dy = DIRS[wind(l, s.t)]
    ignitions = set()
    for i in old_fire:
        if not fire[i]:
            continue
        x, y = xy(i)
        xx, yy = (x + dx, y + dy)
        if 0 <= xx < 5 and 0 <= yy < 5:
            j = ix(xx, yy)
            if not fire[j] and (not burned[j]) and (not wet[j]) and (not water[j]) and (j not in l.stones):
                ignitions.add(j)
    for i in old_fire:
        if fire[i]:
            fire[i] -= 1
            if not fire[i]:
                burned[i] = True
    for j in ignitions:
        fire[j] = 2
        if j in l.houses:
            damaged[l.houses.index(j)] = True
    outcome = 'playing' if any(fire) else 'won' if len(l.houses) - sum(damaged) >= l.goal else 'lost'
    ns = State(s.t + 1, s.stock - amount, tuple(water), tuple(wet), tuple(fire), tuple(burned), tuple(damaged), steam, drain, outcome)
    return (ns, 'ok')

def forecast(s, l, a, wet_life=2):
    return apply(s, l, a, wet_life)

def invariant(s, l, wet_life=2):
    assert s.stock + sum(s.water) + s.steam + s.drain == l.stock
    assert 0 <= s.stock <= l.stock and s.steam >= 0 and (s.drain >= 0)
    assert all((0 <= n <= l.stock for n in s.water))
    assert all((0 <= n <= wet_life for n in s.wet))
    assert all((n in (0, 1, 2) for n in s.fire))
    assert not any((s.fire[i] and s.burned[i] for i in range(25)))
    assert len(s.damaged) == len(l.houses)


class App:
    def __init__(self):
        self.level_index = 0
        self.reset()

    @property
    def level(self):
        return LEVELS[self.level_index]

    def reset(self):
        self.state = initial(self.level)
        self.outlet = 0
        self.amount = 1
        self.gated = False
        self.message = 'READY'

    
    def control(self, pressed, held):
        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
        key = next(iter(pressed))
        
        if key in ('LEFT', 'RIGHT', 'UP', 'DOWN'):
            if key == 'LEFT':self.outlet = 0
            if key == 'RIGHT':self.outlet = 1
            if key == 'UP':self.amount = 2
            if key == 'DOWN':self.amount = 1
            self.message = 'SELECTED'
            self.gated = True
            return
        if key == 'SPACE':
            action = WAIT
        elif key == 'Z':
            action = (self.outlet, self.amount)
        else:
            return
        self.state, self.message = apply(self.state, self.level, action)
        self.gated = True

    def draw(self, p):
        p.cls(1)
        s, level = self.state, self.level
        p.text(10, 8, 'FIRE WATERWAYS / STAGE 4', 10)
        outcome = getattr(s, 'outcome', 'practice').upper()
        p.text(10, 22, 'T '+str(s.t)+'  LEVEL '+chr(65+self.level_index)+'  '+outcome, 7)
        p.text(10, 36, 'FICTIONAL GRID RULES ONLY', 6)
        for i in range(25):
            x, y = xy(i)
            px, py = 12+x*40, 62+y*40
            p.rect(px, py, 38, 38, 5)
            if i in getattr(level, 'stones', ()):
                p.rect(px+2, py+2, 34, 34, 13)
            if getattr(s, 'burned', (False,)*25)[i]:
                p.rect(px+2, py+2, 34, 34, 0)
            wet = getattr(s, 'wet', (0,)*25)[i]
            if wet:
                for dx in (8,18,28):p.pset(px+dx, py+32, 12)
            water = s.water[i]
            if water:
                p.rect(px+3, py+17, 31, 12, 12)
                p.text(px+5, py+20, 'W'+str(water), 7)
            fire = getattr(s, 'fire', (0,)*25)[i]
            if fire:
                p.tri(px+18, py+8, px+7, py+28, px+30, py+28, 9)
                p.text(px+15, py+20, str(fire), 0)
            houses = getattr(level, 'houses', ())
            if i in houses:
                damage = s.damaged[houses.index(i)]
                p.rectb(px+24, py+3, 11, 10, 8 if damage else 11)
                p.text(px+27, py+5, 'H', 8 if damage else 11)
            if hasattr(level, 'heights'):
                p.text(px+2, py+2, str(level.heights[i]), 6)
            if i in level.outlets:
                label = 'W' if level.outlets.index(i)==0 else 'E'
                p.text(px+14, py+4, label, 10)
        p.text(12, 266, 'SOUTH EDGE -> DRAIN', 6)
        p.text(232, 62, 'WATER LEDGER', 10)
        ledger = (('STOCK', getattr(s, 'stock', 0)), ('GRID', sum(s.water)),
                  ('STEAM', getattr(s, 'steam', 0)), ('DRAIN', s.drain))
        for row, (label, value) in enumerate(ledger):
            p.text(232, 78+row*14, label+' '+str(value), 7)
        p.text(232, 136, 'TOTAL '+str(sum(v for _, v in ledger))+' / '+str(level.stock), 11)
        p.text(232, 152, 'WET DOTS ARE NOT WATER', 6)
        p.text(232, 176, 'SELECT '+('W' if self.outlet==0 else 'E')+' / '+str(self.amount), 10)
        after, reason = apply(s, level, (self.outlet, self.amount))
        p.text(232, 192, 'Z -> T'+str(after.t)+' STOCK '+str(after.stock), 7)
        p.text(232, 206, 'PREVIEW '+reason.upper(), 6)
        p.text(232, 222, 'WIND '+wind(level,s.t)+'  FIRE '+str(sum(v>0 for v in s.fire)), 9)
        p.text(232, 238, 'SAVED '+str(saved(s,level))+'/'+str(len(level.houses))+'  GOAL '+str(level.goal), 11)
        p.text(10, 284, self.message.upper(), 7)
        p.text(10, 300, 'LEFT/RIGHT OUTLET   UP=2 DOWN=1   Z RELEASE', 6)
        p.text(10, 314, 'SPACE WAIT   NO REAL-WORLD SAFETY ADVICE', 6)
        p.text(10, 328, 'R RESET  ESC QUIT  RELEASE KEYS BETWEEN TURNS', 6)

KEYS = ('LEFT', 'RIGHT', 'UP', 'DOWN', 'Z', 'SPACE', 'R')
DEMO_KEYS = ('Z', 'SPACE')

def parse_args(argv=None):
    import argparse
    parser = argparse.ArgumentParser(description='Five-stage fictional grid tutorial')
    parser.add_argument('--headless-smoke', action='store_true')
    parser.add_argument('--demo-checkpoint', action='store_true')
    parser.add_argument('--screenshot')
    args = parser.parse_args(argv)
    if (args.demo_checkpoint or args.screenshot is not None) and not args.headless_smoke:
        parser.error('--demo-checkpoint/--screenshot require --headless-smoke')
    if args.screenshot is not None and not args.screenshot.lower().endswith('.png'):
        parser.error('--screenshot requires a .png path')
    return args

def main(args):
    from pathlib import Path
    import pyxel as p  # 引数を検証した後で初めて読み込む
    screenshot = str(Path(args.screenshot).resolve()) if args.screenshot else None
    if screenshot:
        Path(screenshot).parent.mkdir(parents=True, exist_ok=True)
    app = App()
    p.init(480, 352, title='Fire Waterways 4', fps=30,
           display_scale=2, headless=args.headless_smoke)
    keys = {k:getattr(p, 'KEY_'+k) for k in KEYS}
    limit = max(12, 2*len(DEMO_KEYS)+3) if args.demo_checkpoint else 12
    def update():
        if args.headless_smoke and p.frame_count >= limit:
            if screenshot:
                p.screenshot(screenshot, scale=2)
            p.quit()
            return
        if args.headless_smoke and args.demo_checkpoint:
            index = p.frame_count // 2
            pressed = {DEMO_KEYS[index]} if 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)})
    p.run(update, lambda:app.draw(p))

if __name__ == '__main__':
    main(parse_args())
STEP 05 / 05

三状況を比べ、判断を丸ごと戻す

書き換えない State と、受け付けられた1刻ごとの履歴。

1・2・3で、元のゲームの状況A・B・Cを切り替えます。切り替えると、その状況を最初の状態にし、選んでいた放水口・水の量・Undoの履歴もリセットします。Rは、今の状況だけを最初に戻します。Cでは、3番目の刻で風がSに変わり、守る目標は3軒のうち2軒です。決めてある最後の風よりあとの刻では、最後の風を使い続けます。くり返して先頭の風に戻ることはありません。

Timeline は、今の State と、過去の State のタプルを持ちます。apply がちがう State を返したときだけ、前の状態を履歴に積みます。Xは、最後に積んだ状態を取り出します。水が足りない放水・正しくない放水・終わったあとの放水・予告・選ぶ操作は、1刻を進めないので積みません。結果画面でもXを先に処理するので、成功や失敗の直前に戻れます。

Undoは、時刻だけでなく、貯水・水・濡れ・火の燃料・灰・家の被害・蒸気・排水・結果をまとめて戻します。選んでいるWかEと水の量は、画面に出している案です。State には入れていないので、Xでは戻しません。Rや状況の切り替えなら、選択もWと1に戻ります。Undoのあとで同じ判断をすると同じ結果になるので、水を増やすことはできません。

Aは、W1→待つで3軒全部を守れます。Bは、先にE1を使って待つと、3軒全部を守れて貯水も3残ります。でも、W1→E1では東側に間に合わず、2軒です。Cは、3刻待つ→E1→3回待つ、で2軒を守れます。元のゲームのデータで、たどり着けるすべての状態を調べると、Cで3軒を守ることはできません。ただし、自分で改造した面のどれでもそうなる、と教材が保証しているわけではありません。

04は元のゲームのA、05はA・B・Cの純粋なモデルと照らし合わせてあります。最後の段階には、元のゲームの表紙・一時停止・M・濡れを比べるBキー・日本語の文字・再現の記録のファイル保存はありません。濡れの比較は、04の改造の課題で行います。次は、元の model.py の apply と step、experiments.py の全探索、app.py の予告と入力の扱いを読み、モデルと画面を分ける作り方へ進みましょう。

下のコードは 05_undo.py の一部です。前のステップから追加・変更したところだけを抜き出しています。

05_undo.py / 注目する部分
    state, reason = apply(timeline.state, level, action, wet_life)
    if state == timeline.state:
        return timeline, reason
    return Timeline(state, timeline.history + (timeline.state,)), reason

実行して確かめる

教材のZIPを展開したフォルダーで、次のコマンドを実行します。

Windows
.venv\Scripts\python.exe 05_undo.py
macOS / Linux
.venv/bin/python 05_undo.py
画面で見ること2→右→Z→Spaceを5回と押すと、BでT6・WON・SAVED 3/3・STOCK 3になります。Xを押すとT5・PLAYING、履歴5に戻り、Spaceで同じWONへ進みます。自動チェックポイントは、Xを押したあとのT5です。
  • State 全体をUndoで戻す
  • 受け付けられない操作は、履歴に積まない
  • 風で広がる火と、放水した水が着く時刻を比べる
つまずきやすいところ水だけを戻して、蒸気や被害を戻さないと、同じ判断をやり直したときに、帳簿と結果が合わなくなります。State 全体を1つの履歴として扱います。
少し変えてみよう

Bで、R→Z→右→Zと押したあと、火が消えるまでSpaceをくり返すと、SAVED 2/3・LOSTです。R→右→Z→左→Zと押したあと、火が消えるまでSpaceをくり返すと、SAVED 3/3・WONです。同じ2単位の水でも、順番がちがうと結果が変わります。E1だけなら、貯水3を残して成功します。

05_undo.pyをダウンロードこの段階だけで実行できます
このステップのコード全文(348行)
05_undo.py / 全文
# MIT License
#
# Copyright (c) 2026 OpenAI-assisted game 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.
#
# This MIT license covers original source, documentation and primitive artwork.
# Bitmap glyph data is derived from Noto Sans CJK and remains under SIL Open
# Font License 1.1. See assets/NOTO-LICENSE.txt for the separate font license.

"""消防水路・教材5/5。架空の格子規則。現実の消火手順ではありません。"""
from dataclasses import dataclass, replace

WAIT = (-1, 0)
ACTIONS = (WAIT, (0, 1), (0, 2), (1, 1), (1, 2))

N = 5
DIRS = {'E': (1, 0), 'W': (-1, 0), 'S': (0, 1), 'N': (0, -1)}

def ix(x, y):
    return y * N + x

def xy(i):
    return (i % N, i // N)

@dataclass(frozen=True)
class Level:
    title: str
    outlets: tuple
    fires: tuple
    houses: tuple
    winds: tuple
    goal: int
    stock: int = 4
    heights: tuple = tuple((4 - i // 5 for i in range(25)))
    stones: tuple = ()
LEVELS = (Level('下へ残る水', (ix(1, 0), ix(3, 0)), (ix(0, 2),), (ix(2, 2), ix(4, 2), ix(3, 4)), ('E',) * 6, 3), Level('近い火は一手待てる', (ix(1, 0), ix(3, 1)), (ix(0, 0), ix(2, 3)), (ix(4, 3), ix(4, 4), ix(2, 2)), ('E',) * 6, 3), Level('曲がる風', (ix(1, 0), ix(3, 2)), (ix(0, 1), ix(0, 3)), (ix(3, 3), ix(2, 4), ix(4, 4)), ('E', 'E', 'S', 'E', 'E', 'E'), 2))

@dataclass(frozen=True)
class State:
    t: int
    stock: int
    water: tuple
    wet: tuple
    fire: tuple
    burned: tuple
    damaged: tuple
    steam: int = 0
    drain: int = 0
    outcome: str = 'playing'

def initial(l):
    fire = [0] * 25
    for i in l.fires:
        fire[i] = 2
    return State(0, l.stock, (0,) * 25, (0,) * 25, tuple(fire), (False,) * 25, tuple((i in l.fires for i in l.houses)))

def flow_to(l, i):
    """Strictly lower neighbor. Equal lowest: down,left,right,up. South edge drain."""
    x, y = xy(i)
    options = []
    for rank, (dx, dy) in enumerate(((0, 1), (-1, 0), (1, 0), (0, -1))):
        xx, yy = (x + dx, y + dy)
        if yy == 5 and 0 <= xx < 5:
            options.append((-1, rank, -1))
        elif 0 <= xx < 5 and 0 <= yy < 5:
            j = ix(xx, yy)
            if j not in l.stones and l.heights[j] < l.heights[i]:
                options.append((l.heights[j], rank, j))
    return min(options)[2] if options else i

def wind(l, t):
    return l.winds[min(t, len(l.winds) - 1)]

def saved(s, l):
    return len(l.houses) - sum(s.damaged)

def apply(s, l, a, wet_life=2):
    if s.outcome != 'playing':
        return (s, 'terminal')
    if a not in ((-1, 0), (0, 1), (0, 2), (1, 1), (1, 2)):
        return (s, 'invalid')
    outlet, amount = a
    if amount > s.stock:
        return (s, 'stock')
    water = list(s.water)
    wet = [max(0, n - 1) for n in s.wet]
    fire = list(s.fire)
    burned = list(s.burned)
    damaged = list(s.damaged)
    steam = s.steam
    drain = s.drain
    if outlet >= 0:
        water[l.outlets[outlet]] += amount
    touched = {i for i, n in enumerate(water) if n}
    old_fire = {i for i, n in enumerate(s.fire) if n}
    for i in range(25):
        if water[i] and fire[i]:
            water[i] -= 1
            steam += 1
            fire[i] = 0
    moved = [0] * 25
    for i, n in enumerate(water):
        if n:
            j = flow_to(l, i)
            if j < 0:
                drain += n
            else:
                moved[j] += n
                touched.add(j)
    water = moved
    for i in range(25):
        if water[i] and fire[i]:
            water[i] -= 1
            steam += 1
            fire[i] = 0
    for i in touched:
        wet[i] = wet_life
    dx, dy = DIRS[wind(l, s.t)]
    ignitions = set()
    for i in old_fire:
        if not fire[i]:
            continue
        x, y = xy(i)
        xx, yy = (x + dx, y + dy)
        if 0 <= xx < 5 and 0 <= yy < 5:
            j = ix(xx, yy)
            if not fire[j] and (not burned[j]) and (not wet[j]) and (not water[j]) and (j not in l.stones):
                ignitions.add(j)
    for i in old_fire:
        if fire[i]:
            fire[i] -= 1
            if not fire[i]:
                burned[i] = True
    for j in ignitions:
        fire[j] = 2
        if j in l.houses:
            damaged[l.houses.index(j)] = True
    outcome = 'playing' if any(fire) else 'won' if len(l.houses) - sum(damaged) >= l.goal else 'lost'
    ns = State(s.t + 1, s.stock - amount, tuple(water), tuple(wet), tuple(fire), tuple(burned), tuple(damaged), steam, drain, outcome)
    return (ns, 'ok')

def forecast(s, l, a, wet_life=2):
    return apply(s, l, a, wet_life)

def invariant(s, l, wet_life=2):
    assert s.stock + sum(s.water) + s.steam + s.drain == l.stock
    assert 0 <= s.stock <= l.stock and s.steam >= 0 and (s.drain >= 0)
    assert all((0 <= n <= l.stock for n in s.water))
    assert all((0 <= n <= wet_life for n in s.wet))
    assert all((n in (0, 1, 2) for n in s.fire))
    assert not any((s.fire[i] and s.burned[i] for i in range(25)))
    assert len(s.damaged) == len(l.houses)

@dataclass(frozen=True)
class Timeline:
    state: State
    history: tuple = ()

def step(timeline, level, action, wet_life=2):
    if action == ('undo',):
        if not timeline.history:
            return timeline, 'no_undo'
        return Timeline(timeline.history[-1], timeline.history[:-1]), 'undo'
    state, reason = apply(timeline.state, level, action, wet_life)
    if state == timeline.state:
        return timeline, reason
    return Timeline(state, timeline.history + (timeline.state,)), reason


class App:
    def __init__(self):
        self.level_index = 0
        self.reset()

    @property
    def level(self):
        return LEVELS[self.level_index]

    def reset(self):
        self.timeline = Timeline(initial(self.level))
        self.outlet = 0
        self.amount = 1
        self.gated = False
        self.message = 'READY'

    @property
    def state(self):
        return self.timeline.state

    def control(self, pressed, held):
        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
        key = next(iter(pressed))
        if key in ('1', '2', '3'):
            self.level_index = int(key)-1
            self.reset()
            self.gated = True
            return
        if key == 'X':
            self.timeline, self.message = step(self.timeline, self.level, ('undo',))
            self.gated = True
            return
        if key in ('LEFT', 'RIGHT', 'UP', 'DOWN'):
            if key == 'LEFT':self.outlet = 0
            if key == 'RIGHT':self.outlet = 1
            if key == 'UP':self.amount = 2
            if key == 'DOWN':self.amount = 1
            self.message = 'SELECTED'
            self.gated = True
            return
        if key == 'SPACE':
            action = WAIT
        elif key == 'Z':
            action = (self.outlet, self.amount)
        else:
            return
        self.timeline, self.message = step(self.timeline, self.level, action)
        self.gated = True

    def draw(self, p):
        p.cls(1)
        s, level = self.state, self.level
        p.text(10, 8, 'FIRE WATERWAYS / STAGE 5', 10)
        outcome = getattr(s, 'outcome', 'practice').upper()
        p.text(10, 22, 'T '+str(s.t)+'  LEVEL '+chr(65+self.level_index)+'  '+outcome, 7)
        p.text(10, 36, 'FICTIONAL GRID RULES ONLY', 6)
        for i in range(25):
            x, y = xy(i)
            px, py = 12+x*40, 62+y*40
            p.rect(px, py, 38, 38, 5)
            if i in getattr(level, 'stones', ()):
                p.rect(px+2, py+2, 34, 34, 13)
            if getattr(s, 'burned', (False,)*25)[i]:
                p.rect(px+2, py+2, 34, 34, 0)
            wet = getattr(s, 'wet', (0,)*25)[i]
            if wet:
                for dx in (8,18,28):p.pset(px+dx, py+32, 12)
            water = s.water[i]
            if water:
                p.rect(px+3, py+17, 31, 12, 12)
                p.text(px+5, py+20, 'W'+str(water), 7)
            fire = getattr(s, 'fire', (0,)*25)[i]
            if fire:
                p.tri(px+18, py+8, px+7, py+28, px+30, py+28, 9)
                p.text(px+15, py+20, str(fire), 0)
            houses = getattr(level, 'houses', ())
            if i in houses:
                damage = s.damaged[houses.index(i)]
                p.rectb(px+24, py+3, 11, 10, 8 if damage else 11)
                p.text(px+27, py+5, 'H', 8 if damage else 11)
            if hasattr(level, 'heights'):
                p.text(px+2, py+2, str(level.heights[i]), 6)
            if i in level.outlets:
                label = 'W' if level.outlets.index(i)==0 else 'E'
                p.text(px+14, py+4, label, 10)
        p.text(12, 266, 'SOUTH EDGE -> DRAIN', 6)
        p.text(232, 62, 'WATER LEDGER', 10)
        ledger = (('STOCK', getattr(s, 'stock', 0)), ('GRID', sum(s.water)),
                  ('STEAM', getattr(s, 'steam', 0)), ('DRAIN', s.drain))
        for row, (label, value) in enumerate(ledger):
            p.text(232, 78+row*14, label+' '+str(value), 7)
        p.text(232, 136, 'TOTAL '+str(sum(v for _, v in ledger))+' / '+str(level.stock), 11)
        p.text(232, 152, 'WET DOTS ARE NOT WATER', 6)
        p.text(232, 176, 'SELECT '+('W' if self.outlet==0 else 'E')+' / '+str(self.amount), 10)
        after, reason = apply(s, level, (self.outlet, self.amount))
        p.text(232, 192, 'Z -> T'+str(after.t)+' STOCK '+str(after.stock), 7)
        p.text(232, 206, 'PREVIEW '+reason.upper(), 6)
        p.text(232, 222, 'WIND '+wind(level,s.t)+'  FIRE '+str(sum(v>0 for v in s.fire)), 9)
        p.text(232, 238, 'SAVED '+str(saved(s,level))+'/'+str(len(level.houses))+'  GOAL '+str(level.goal), 11)
        p.text(232, 254, 'UNDO '+str(len(self.timeline.history)), 6)
        p.text(10, 284, self.message.upper(), 7)
        p.text(10, 300, 'LEFT/RIGHT OUTLET   UP=2 DOWN=1   Z RELEASE', 6)
        p.text(10, 314, 'SPACE WAIT   1/2/3 LEVEL   X UNDO', 6)
        p.text(10, 328, 'R RESET  ESC QUIT  RELEASE KEYS BETWEEN TURNS', 6)

KEYS = ('LEFT', 'RIGHT', 'UP', 'DOWN', 'Z', 'SPACE', 'R', 'X', '1', '2', '3')
DEMO_KEYS = ('2', 'RIGHT', 'Z', 'SPACE', 'SPACE', 'SPACE', 'SPACE', 'SPACE', 'X')

def parse_args(argv=None):
    import argparse
    parser = argparse.ArgumentParser(description='Five-stage fictional grid tutorial')
    parser.add_argument('--headless-smoke', action='store_true')
    parser.add_argument('--demo-checkpoint', action='store_true')
    parser.add_argument('--screenshot')
    args = parser.parse_args(argv)
    if (args.demo_checkpoint or args.screenshot is not None) and not args.headless_smoke:
        parser.error('--demo-checkpoint/--screenshot require --headless-smoke')
    if args.screenshot is not None and not args.screenshot.lower().endswith('.png'):
        parser.error('--screenshot requires a .png path')
    return args

def main(args):
    from pathlib import Path
    import pyxel as p  # 引数を検証した後で初めて読み込む
    screenshot = str(Path(args.screenshot).resolve()) if args.screenshot else None
    if screenshot:
        Path(screenshot).parent.mkdir(parents=True, exist_ok=True)
    app = App()
    p.init(480, 352, title='Fire Waterways 5', fps=30,
           display_scale=2, headless=args.headless_smoke)
    keys = {k:getattr(p, 'KEY_'+k) for k in KEYS}
    limit = max(12, 2*len(DEMO_KEYS)+3) if args.demo_checkpoint else 12
    def update():
        if args.headless_smoke and p.frame_count >= limit:
            if screenshot:
                p.screenshot(screenshot, scale=2)
            p.quit()
            return
        if args.headless_smoke and args.demo_checkpoint:
            index = p.frame_count // 2
            pressed = {DEMO_KEYS[index]} if 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)})
    p.run(update, lambda:app.draw(p))

if __name__ == '__main__':
    main(parse_args())
AFTER STEP 05

仕組みがわかったら、元の完成版を読んでみよう。

25マス、貯水4、火の燃料2です。1刻の順番は、今いるマスで消火→水が1歩動く→着いた先で消火→水が触れたマスの濡れを2にする→旧火が燃え広がる先を予約→旧火の燃料を減らす→新しい火、です。燃え尽きてできた灰は、もう燃えません。水で消えた草は、濡れがなくなったあとで、また燃えることがあります。火が全部消えたときに、家を守る目標を達成したかを調べます。期限はありません。

ここまでに作ったのは、元のゲームの仕組みだけを取り出して作り直した「簡略版」です。元のゲームのソースは手を加えずに、別のZIP(元の完成版ソース)に入れてあります。

比べる点5ステップの教材元の完成版
段階01は水4を盤面に置いて、真下へ流す練習。02は貯水と地形だけ。03は濡れと家がない。04は元のゲームのA、05は元のゲームのA・B・C。最初から、水・火・濡れ・家がそろった3つの状況。
表示と操作英字(ASCII)と図形。操作のたびに、キーが全部離されるまで待つ。新しく押したRを優先。05の状況の切り替えは数字キー。選ぶ操作と予告では、時間は進まない。日本語の文字、表紙、一時停止、M、濡れを比べるB、F12で画像と入力の記録。
記録と比較State のUndo、次にZを押したときのTと貯水の予告。濡れの比較は改造の課題。ファイルに残る記録はない。Undoに加えて、自動のデモ、画像と入力の記録、濡れの続く長さ2と0を比べる管理用の比較。

完成版のコードで、次に読むところ

  1. model.py の apply(消火・移動・濡れ・燃え広がり・結果の順番)
  2. model.py の step(状態全体のUndo)と、forecast が何も書き換えないこと
  3. experiments.py の search と replay(限られた数の状態をすべて調べる全探索)
  4. app.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の実際のパソコンでの操作や、音が出るかどうかはまだ確認していません。人がすべての操作を試したわけでもありません。ブラウザー版の確認範囲は確認状況をご覧ください。

うまく動かないときは:起動やエラーの対処法 / 質問用のメモを作る