シミュレーションLESSON 53二台の並行加工と段取りの工場計画原作049

町工場の一日の作り方

2台の機械を動かして、先頭の1つを仕上げる。
「切る」と「仕上げる」を同時に進め、入れられる数に限りがある順番待ちのレール、材料を変えるときの掃除、納期を扱う工場を作ります。

完成版の遊び方1人の指示で同時に動く2台の機械を使い、材料を切りかえるときの掃除と、レールに並べる順番を考えます。

町工場の一日の元の完成版。自動操作による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. 書きかえない状態と、1刻ずつ進む時間
2. 先に入れた順に出す列(FIFO)・処理の順番・品物がどこにあるか
3. 状態を持つ機械と、準備にかかるコスト
それだけで動く小さなプログラムを、5つのステップで少しずつ大きくしていきます。
必要な環境
Python 3.11以上 / Pyxel 2.9.9
知っておくとよいこと
Python 3.11以上、Pyxel 2.9.9 / 変数・if・for・関数・タプル / dataclass、状態を書きかえない関数、先に入れた順に出す列(FIFO)はステップごとに説明します
教材に入っているもの
ステップごとに動くPythonファイル5本・説明・動作確認用のテスト
完成した教材の操作
左右:一つ前/次の指示を選ぶ(時刻は進まない) / Z / Space:選択指示を確定 / R:現在の面を初期化 / 1 / 2 / 3:04/05でA/B/C面へ切替、初期化 / X:05で一刻戻す / B:05で無料素材切替比較へ切替、初期化 / Esc:Pyxelを終了
難易度の目安
中級このページで使うプログラムの考え方の数と難しさから決めた目安です。難易度について
学習順に迷ったら
おすすめの学ぶ順番を見る
この講座の進め方① ページ下の教材ZIPをダウンロードして、すべて展開します。② はじめる準備を見ながら、PythonとPyxelを使える状態にします。③ まず 01_cutting.py を、何も変えずにそのまま実行してみましょう。記事に載せているコードは説明用の一部分なので、実行するときはZIPに入っているファイルを使ってください。
コードを変更したら

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

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

STEP 01 / 05

一つの指示で一刻だけ進める

書きかえない状態と、1刻ずつ進む時間。

「町工場の一日」は、注文を切断の機械 C に入れて、できあがるまで待つゲームです。最初は C だけにしぼります。左右で枠を選び、Z か Space で決定します。a は1刻、b は3刻、c は1刻かかる仕事です。操作したら、キーをいったん離してください。画面の T は判断した回数で、本当の秒数ではありません。何も押さずに待っていても、加工は進みません。

Job は注文の決まったデータ、State は今の工場のようすです。cjob=-1 は空き、crem は残りの時間、started は着手した印、done はできあがった時刻を並べたタプルです。done=-1 はまだ、0以上ならできあがった時刻なので、False と整数の0をまちがえません。frozen=True で State を直接書きかえられないようにし、replace で次の状態を作ります。途中で入力を受け付けなかったときは、元の箱をそのまま返せます。

apply は「指示をチェック → 開始 → 残り時間を1減らす → できあがりを記録 → T を1増やす」の順番です。開始の指示を出した刻も、1刻目に数えます。b を T0 から始めると、T1 で残り2、待つと T2 で残り1、もう一度待つと T3 でできあがります。a なら、始めた1刻だけでできあがります。

同じ注文をもう一度始める操作、機械がふさがっているのに別の注文を入れる操作、知らない指示では、時間を進めません。見た目だけ無視して、裏で時刻を進めるような作りにはしません。draw は読むだけ、control はキーを指示に変え、apply が工場を更新します。この段階には、仕上げの機械 F、レール、材料、売上、勤務の終わりはありません。

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

01_cutting.py / 注目する部分
    next_state = replace(next_state, crem=max(0, next_state.crem - 1))
    if next_state.cjob >= 0 and next_state.crem == 0:
        done = list(next_state.done)
        done[next_state.cjob] = state.t + 1
        next_state = replace(next_state, cjob=-1, done=tuple(done))

実行して確かめる

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

Windows
.venv\Scripts\python.exe 01_cutting.py
macOS / Linux
.venv/bin/python 01_cutting.py
画面で見ること右を1回押して CUT b を選び、Z を押します。T1・C JOB b・REM 2 です。右を2回押して WAIT に移り、Z を2回(毎回離す)押すと、T3 で b が DONE @3(時刻3で完成)になり、C は IDLE(空き)になります。
  • 始めた刻も加工に数える
  • 受け付けない指示では時間を使わない
  • 描く処理と、状態の更新を分ける
つまずきやすいところ始める前に残り時間を減らすと、3刻の仕事に4回の指示が必要になってしまいます。開始を replace で反映してから減らしている行を確かめましょう。
少し変えてみよう

LEVELS の Job("a", 2) の部分を、元の Job("a", 1) から変えてみましょう。もう一度起動して CUT a と WAIT を1回ずつ実行したときの、T・残り時間・完成の時刻を予想してから確かめます。

01_cutting.pyをダウンロードこの段階だけで実行できます
このステップのコード全文(256行)
01_cutting.py / 全文
# MIT License
# 
# Copyright (c) 2026
# 
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
# 
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
# 
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
# 
# Third-party exception: bitmap glyph data derived from Noto Sans CJK is licensed
# under the SIL Open Font License 1.1. See assets/NOTO-LICENSE.txt.

"""町工場の一日・教材 1/5。単体起動可能、外部素材なし。"""
from dataclasses import dataclass, replace

@dataclass(frozen=True)
class Job:
    id: str
    cut: int


@dataclass(frozen=True)
class Level:
    title: str
    jobs: tuple


LEVELS = (Level("Cutting only", (Job("a", 1), Job("b", 3), Job("c", 1))),)


@dataclass(frozen=True)
class State:
    t: int = 0
    cjob: int = -1
    crem: int = 0
    started: tuple = ()
    done: tuple = ()


def initial(level):
    return State(started=(False,) * len(level.jobs), done=(-1,) * len(level.jobs))


def actions(level):
    cuts = tuple(("cut", i) for i in range(len(level.jobs)))
    return cuts + (("wait",),)


def apply(state, level, action):
    if action not in actions(level):
        return state, "invalid"
    next_state = state
    kind = action[0]
    if kind == "cut":
        index = action[1]
        job = level.jobs[index]
        if state.started[index]:
            return state, "started"
        if state.cjob >= 0:
            return state, "busy"
        started = list(state.started)
        started[index] = True
        next_state = replace(next_state, cjob=index, crem=job.cut,
                             started=tuple(started))
    # 指示を反映した後で進める。開始した刻も1刻目に数える。
    next_state = replace(next_state, crem=max(0, next_state.crem - 1))
    if next_state.cjob >= 0 and next_state.crem == 0:
        done = list(next_state.done)
        done[next_state.cjob] = state.t + 1
        next_state = replace(next_state, cjob=-1, done=tuple(done))
    next_state = replace(next_state, t=state.t + 1)
    check(next_state, level)
    return next_state, "ok"


def check(state, level):
    assert state.t >= 0 and state.crem >= 0
    assert len(state.started) == len(level.jobs) == len(state.done)
    owners = [state.cjob] if state.cjob >= 0 else []
    owners += [i for i, time in enumerate(state.done) if time >= 0]
    assert len(owners) == len(set(owners))
    assert set(owners) == {i for i, started in enumerate(state.started) if started}
    assert all(0 <= i < len(level.jobs) for i in owners)
    assert state.cjob >= 0 or state.crem == 0


def location(state, index):
    if state.done[index] >= 0:
        return "DONE @" + str(state.done[index])
    if state.cjob == index:
        return "C REM " + str(state.crem) if state.crem else "C HOLD"
    return "NOT STARTED"


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.selection = 0
        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))
        choices = actions(self.level)
        if key in ("LEFT", "RIGHT"):
            self.selection = (self.selection + (1 if key == "RIGHT" else -1)) % len(choices)
        elif key in ("Z", "SPACE"):
            self.state, self.message = apply(self.state, self.level, choices[self.selection])
        else:
            return
        self.gated = True

    def draw(self, p):
        p.cls(1)
        p.text(12, 10, "SMALL FACTORY / STAGE 1", 10)
        state = self.state
        p.text(12, 27, self.level.title + "  T=" + str(state.t), 7)
        p.rect(12, 48, 164, 78, 2)
        p.rectb(12, 48, 164, 78, 6)
        p.text(22, 58, "C / CUTTING", 10)
        p.text(22, 77, "IDLE" if state.cjob < 0 else "JOB " + self.level.jobs[state.cjob].id, 7)
        p.text(22, 94, "REM " + str(state.crem), 11)
        p.text(200, 72, "A STARTED JOB ADVANCES THIS TURN.", 6)
        p.text(200, 90, "WAIT ALSO ADVANCES THE MACHINE.", 6)
        p.text(12, 144, "JOB  TIMES   LOCATION", 13)
        for index, job in enumerate(self.level.jobs):
            y = 162 + index * 17
            if state.done[index] >= 0:
                p.rect(10, y - 2, 491, 13, 3)
            times = str(job.cut)
            p.text(12, y, job.id, 10)
            p.text(36, y, times, 7)
            p.text(96, y, location(state, index), 7)
        for index, action in enumerate(actions(self.level)):
            x = 12 + (index % 4) * 124
            y = 244 + (index // 4) * 23
            label = "CUT " + self.level.jobs[action[1]].id if action[0] == "cut" else " ".join(action)
            p.rectb(x, y, 118, 19, 10 if index == self.selection else 5)
            p.text(x + 5, y + 6, label.upper(), 10 if index == self.selection else 7)
        p.text(12, 305, "GAME TIME CHANGES ONLY AFTER A VALID INSTRUCTION.", 6)
        p.text(12, 340, self.message.upper(), 11)
        p.text(12, 359, "LEFT/RIGHT SELECT  Z/SPACE DO  R RESET  ESC QUIT", 7)


DEMO_ACTIONS = (('cut', 1), ('wait',), ('wait',))
KEYS = ("LEFT", "RIGHT", "Z", "SPACE", "X", "R", "1", "2", "3", "B")


def demo_keys():
    selection = 0
    choices = actions(LEVELS[0])
    result = []
    for action in DEMO_ACTIONS:
        if action == ("undo",):
            result.append("X")
            continue
        target = choices.index(action)
        while selection != target:
            result.append("RIGHT")
            selection = (selection + 1) % len(choices)
        result.append("Z")
    return tuple(result)


def parse_args(argv=None):
    import argparse
    from pathlib import Path
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--headless-smoke", action="store_true")
    parser.add_argument("--demo-checkpoint", action="store_true")
    parser.add_argument("--screenshot")
    parser.add_argument("--frames", type=int)
    args = parser.parse_args(argv)
    if (args.demo_checkpoint or args.screenshot or args.frames is not None) and not args.headless_smoke:
        parser.error("--demo-checkpoint / --screenshot / --frames require --headless-smoke")
    minimum = 2 * len(demo_keys()) + 2 if args.demo_checkpoint else 2
    if args.frames is not None and not minimum <= args.frames <= 600:
        parser.error("--frames must be between " + str(minimum) + " and 600")
    if args.screenshot:
        path = Path(args.screenshot)
        if path.suffix.lower() != ".png" or path.is_dir():
            parser.error("--screenshot must name a PNG file")
        args.screenshot = str(path.resolve())
    args.frames = args.frames or max(8, minimum)
    return args


def main(args):
    from pathlib import Path
    import pyxel as p
    if args.screenshot:
        Path(args.screenshot).parent.mkdir(parents=True, exist_ok=True)
    app = App()
    demo = demo_keys()
    frame = 0
    p.init(512, 396, title="Small Factory", fps=30, display_scale=2, headless=args.headless_smoke)
    keys = {key: getattr(p, "KEY_" + key) for key in KEYS}

    def update():
        nonlocal frame
        frame += 1
        if args.headless_smoke:
            if args.demo_checkpoint:
                index = (frame - 1) // 2
                pressed = {demo[index]} if frame % 2 == 1 and index < len(demo) else set()
                app.control(pressed, pressed)
        else:
            app.control({key for key, code in keys.items() if p.btnp(code)},
                        {key for key, code in keys.items() if p.btn(code)})

    def draw():
        app.draw(p)
        if args.headless_smoke and frame >= args.frames:
            if args.screenshot:
                p.screenshot(args.screenshot, scale=2)
            p.quit()
    p.run(update, draw)


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

二台の並行加工と二枠のレール

先に入れた順に出す列(FIFO)・処理の順番・品物がどこにあるか。

切断の C のあとに、仕上げの機械 F を加えます。C でできあがった仕事は、2つ分の枠があるレールに運ばれ、FINISH の指示で、いちばん前の1つだけを F に入れます。仕事がある場所は「まだ始めていない → C → レール → F → 完成」のどこか1つです。Job の cut と finish は必要な加工の刻数で、State には F の仕事の番号・残り時間・queue(レール)が増えます。

queue は、先に入れたものから順に出す(FIFO)タプルです。先頭は queue[0]、取り出した残りは queue[1:]、最後に足す式は queue+(番号,) です。うしろのカンマがないと、1つだけのタプルになりません。レールには2つまでしか入らないので、C でできあがっても空きがなければ、cjob に残したままにします。crem=0 なのに cjob>=0 という状態は、「できあがった品物を C が持ったまま」という意味です。捨てたり、まだ始めていない状態に戻したりはしません。

1刻にできる作業員の指示は1つですが、C と F は同時に1刻ずつ進みます。FINISH で、今あるレールの先頭を取り出したあと、2台の機械の残り時間を減らし、「F の完成」「C からレールへ運ぶ」の順に処理します。取り出して空いたレールの枠には、その刻の最後に C の完成品を入れられます。一方、指示を出したときにレールが空なら、FINISH は受け付けません。その刻に C ができあがる予定でも、先取りはできません。

check では、C・F・レール・完成の番号を集めて、重なりがないことと、着手した集合と一致することを調べます。品物が消えたり、2か所に増えたりする不具合を、画面とは関係なく見つけられます。この段階は、全部の注文が同じ材料の練習で、納期と売上はまだありません。

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

02_parallel.py / 注目する部分
    if next_state.cjob >= 0 and next_state.crem == 0 and len(next_state.queue) < CAPACITY:
        next_state = replace(next_state, queue=next_state.queue + (next_state.cjob,), cjob=-1)

実行して確かめる

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

Windows
.venv\Scripts\python.exe 02_parallel.py
macOS / Linux
.venv/bin/python 02_parallel.py
画面で見ることCUT a → CUT c → CUT d で T3 になり、レールには a と c、C は d を REM 0(できあがったまま)で持っています。FINISH を押すと T4、F は a・残り1、レールは c と d、C は空きです。
  • F が取るのは先頭だけ
  • 2台の残り時間を同じ刻に減らす
  • レールがいっぱいなら、完成品は C が持つ
  • 品物の重なりと消失をチェックする
つまずきやすいところcrem==0 だけを見て cjob=-1 に戻すと、レールがいっぱいのときに品物が消えてしまいます。空きがあるかの判定と、レールへ運ぶ処理を同じ if に置き、運べたときだけ C を空けます。
少し変えてみよう

CAPACITY を2から1に変えて、もう一度起動します。CUT a → CUT c まで実行して、レールと C の状態を確かめましょう。確かめたら2に戻します。

02_parallel.pyをダウンロードこの段階だけで実行できます
このステップのコード全文(293行)
02_parallel.py / 全文
# MIT License
# 
# Copyright (c) 2026
# 
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
# 
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
# 
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
# 
# Third-party exception: bitmap glyph data derived from Noto Sans CJK is licensed
# under the SIL Open Font License 1.1. See assets/NOTO-LICENSE.txt.

"""町工場の一日・教材 2/5。単体起動可能、外部素材なし。"""
from dataclasses import dataclass, replace

@dataclass(frozen=True)
class Job:
    id: str
    cut: int
    finish: int


@dataclass(frozen=True)
class Level:
    title: str
    jobs: tuple


LEVELS = (Level("Parallel work", (Job("a", 1, 2), Job("b", 2, 1),
                                       Job("c", 1, 1), Job("d", 1, 2))),)
CAPACITY = 2


@dataclass(frozen=True)
class State:
    t: int = 0
    cjob: int = -1
    crem: int = 0
    fjob: int = -1
    frem: int = 0
    queue: tuple = ()
    started: tuple = ()
    done: tuple = ()


def initial(level):
    return State(started=(False,) * len(level.jobs), done=(-1,) * len(level.jobs))


def actions(level):
    cuts = tuple(("cut", i) for i in range(len(level.jobs)))
    return cuts + (("finish",), ("wait",))


def apply(state, level, action):
    if action not in actions(level):
        return state, "invalid"
    next_state = state
    kind = action[0]
    if kind == "cut":
        index = action[1]
        job = level.jobs[index]
        if state.started[index]:
            return state, "started"
        if state.cjob >= 0:
            return state, "busy"
        started = list(state.started)
        started[index] = True
        next_state = replace(next_state, cjob=index, crem=job.cut,
                             started=tuple(started))
    elif kind == "finish":
        if state.fjob >= 0:
            return state, "busy"
        if not state.queue:
            return state, "empty"
        index = state.queue[0]
        job = level.jobs[index]
        next_state = replace(next_state, fjob=index, frem=job.finish,
                             queue=state.queue[1:])
    # 指示を反映した後で進める。開始した刻も1刻目に数える。
    next_state = replace(next_state, crem=max(0, next_state.crem - 1),
                         frem=max(0, next_state.frem - 1))
    # Fの取り出しは指示時。Cの搬出は同じ刻の最後。
    if next_state.fjob >= 0 and next_state.frem == 0:
        index = next_state.fjob
        done = list(next_state.done)
        done[index] = state.t + 1
        next_state = replace(next_state, fjob=-1, done=tuple(done))
    if next_state.cjob >= 0 and next_state.crem == 0 and len(next_state.queue) < CAPACITY:
        next_state = replace(next_state, queue=next_state.queue + (next_state.cjob,), cjob=-1)
    next_state = replace(next_state, t=state.t + 1)
    check(next_state, level)
    return next_state, "ok"


def check(state, level):
    assert state.t >= 0 and state.crem >= 0
    assert len(state.started) == len(level.jobs) == len(state.done)
    assert 0 <= len(state.queue) <= CAPACITY and state.frem >= 0
    owners = [i for i in (state.cjob, state.fjob) if i >= 0] + list(state.queue)
    assert state.fjob >= 0 or state.frem == 0
    owners += [i for i, time in enumerate(state.done) if time >= 0]
    assert len(owners) == len(set(owners))
    assert set(owners) == {i for i, started in enumerate(state.started) if started}
    assert all(0 <= i < len(level.jobs) for i in owners)
    assert state.cjob >= 0 or state.crem == 0


def location(state, index):
    if state.done[index] >= 0:
        return "DONE @" + str(state.done[index])
    if state.cjob == index:
        return "C REM " + str(state.crem) if state.crem else "C HOLD"
    if state.fjob == index:
        return "F REM " + str(state.frem)
    if index in state.queue:
        return "QUEUE " + str(state.queue.index(index) + 1)
    return "NOT STARTED"


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.selection = 0
        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))
        choices = actions(self.level)
        if key in ("LEFT", "RIGHT"):
            self.selection = (self.selection + (1 if key == "RIGHT" else -1)) % len(choices)
        elif key in ("Z", "SPACE"):
            self.state, self.message = apply(self.state, self.level, choices[self.selection])
        else:
            return
        self.gated = True

    def draw(self, p):
        p.cls(1)
        p.text(12, 10, "SMALL FACTORY / STAGE 2", 10)
        state = self.state
        p.text(12, 27, self.level.title + "  T=" + str(state.t), 7)
        p.rect(12, 48, 164, 78, 2)
        p.rectb(12, 48, 164, 78, 6)
        p.text(22, 58, "C / CUTTING", 10)
        p.text(22, 77, "IDLE" if state.cjob < 0 else "JOB " + self.level.jobs[state.cjob].id, 7)
        p.text(22, 94, "REM " + str(state.crem), 11)
        p.rect(336, 48, 164, 78, 2)
        p.rectb(336, 48, 164, 78, 6)
        p.text(346, 58, "F / FINISHING", 10)
        p.text(346, 77, "IDLE" if state.fjob < 0 else "JOB " + self.level.jobs[state.fjob].id, 7)
        p.text(346, 94, "REM " + str(state.frem), 11)
        p.text(194, 58, "RAIL / FIFO " + str(len(state.queue)) + "/" + str(CAPACITY), 6)
        for slot in range(CAPACITY):
            x = 197 + slot * 50
            p.rectb(x, 77, 38, 26, 5)
            if slot < len(state.queue):
                p.text(x + 12, 87, self.level.jobs[state.queue[slot]].id, 10)
        p.text(194, 115, "HEAD -> TAIL", 13)
        p.text(12, 144, "JOB  TIMES   LOCATION", 13)
        for index, job in enumerate(self.level.jobs):
            y = 162 + index * 17
            if state.done[index] >= 0:
                p.rect(10, y - 2, 491, 13, 3)
            times = str(job.cut)
            times += "/" + str(job.finish)
            p.text(12, y, job.id, 10)
            p.text(36, y, times, 7)
            p.text(96, y, location(state, index), 7)
        for index, action in enumerate(actions(self.level)):
            x = 12 + (index % 4) * 124
            y = 244 + (index // 4) * 23
            label = "CUT " + self.level.jobs[action[1]].id if action[0] == "cut" else " ".join(action)
            p.rectb(x, y, 118, 19, 10 if index == self.selection else 5)
            p.text(x + 5, y + 6, label.upper(), 10 if index == self.selection else 7)
        p.text(12, 305, "FULL RAIL? C HOLDS THE COMPLETED JOB. NOTHING IS LOST.", 6)
        p.text(12, 340, self.message.upper(), 11)
        p.text(12, 359, "LEFT/RIGHT SELECT  Z/SPACE DO  R RESET  ESC QUIT", 7)


DEMO_ACTIONS = (('cut', 0), ('cut', 2), ('cut', 3), ('finish',))
KEYS = ("LEFT", "RIGHT", "Z", "SPACE", "X", "R", "1", "2", "3", "B")


def demo_keys():
    selection = 0
    choices = actions(LEVELS[0])
    result = []
    for action in DEMO_ACTIONS:
        if action == ("undo",):
            result.append("X")
            continue
        target = choices.index(action)
        while selection != target:
            result.append("RIGHT")
            selection = (selection + 1) % len(choices)
        result.append("Z")
    return tuple(result)


def parse_args(argv=None):
    import argparse
    from pathlib import Path
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--headless-smoke", action="store_true")
    parser.add_argument("--demo-checkpoint", action="store_true")
    parser.add_argument("--screenshot")
    parser.add_argument("--frames", type=int)
    args = parser.parse_args(argv)
    if (args.demo_checkpoint or args.screenshot or args.frames is not None) and not args.headless_smoke:
        parser.error("--demo-checkpoint / --screenshot / --frames require --headless-smoke")
    minimum = 2 * len(demo_keys()) + 2 if args.demo_checkpoint else 2
    if args.frames is not None and not minimum <= args.frames <= 600:
        parser.error("--frames must be between " + str(minimum) + " and 600")
    if args.screenshot:
        path = Path(args.screenshot)
        if path.suffix.lower() != ".png" or path.is_dir():
            parser.error("--screenshot must name a PNG file")
        args.screenshot = str(path.resolve())
    args.frames = args.frames or max(8, minimum)
    return args


def main(args):
    from pathlib import Path
    import pyxel as p
    if args.screenshot:
        Path(args.screenshot).parent.mkdir(parents=True, exist_ok=True)
    app = App()
    demo = demo_keys()
    frame = 0
    p.init(512, 396, title="Small Factory", fps=30, display_scale=2, headless=args.headless_smoke)
    keys = {key: getattr(p, "KEY_" + key) for key in KEYS}

    def update():
        nonlocal frame
        frame += 1
        if args.headless_smoke:
            if args.demo_checkpoint:
                index = (frame - 1) // 2
                pressed = {demo[index]} if frame % 2 == 1 and index < len(demo) else set()
                app.control(pressed, pressed)
        else:
            app.control({key for key, code in keys.items() if p.btnp(code)},
                        {key for key, code in keys.items() if p.btn(code)})

    def draw():
        app.draw(p)
        if args.headless_smoke and frame >= args.frames:
            if args.screenshot:
                p.screenshot(args.screenshot, scale=2)
            p.quit()
    p.run(update, draw)


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

素材を切り替える清掃を一つの行動にする

状態を持つ機械と、準備にかかるコスト。

注文に、材料の W(木)と M(金属)を加えます。C と F にはそれぞれ ctype、ftype という設定があり、最初はどちらも W です。材料を変えるには、空いている機械に CLEAN(掃除)を1回指示して、設定を -(きれい)にします。次の仕事を始めると、その仕事の材料が新しい設定になります。同じ材料を続けて加工するなら、掃除はいりません。

切断と仕上げの「始められる条件」は、別々に調べます。C を掃除しただけでは、F は木用のままです。F が空いていて、レールに金属の仕事があっても、F を掃除していなければ、FINISH は kind(材料ちがい)で受け付けられません。レールのうしろに木の仕事があっても、先頭を飛ばすことはできません。だから、並べる順番が大切なのです。

掃除も1つの指示なので T が1進み、その間にもう1台の機械は加工を続けます。ただし、掃除する機械に仕事が入っていると busy(使用中)で受け付けません。残り0で完成品を持っている C も busy です。すでに - なら clean(掃除済み)で受け付けず、時刻も掃除の回数も変わりません。「掃除の回数だけ増えて、時間は進まない」というおかしな状態を防ぎます。

材料が合っているかを調べてから、started や queue を変える、という順番を守ります。受け付けない開始の操作で、レールから仕事だけが消える不具合を防ぐためです。元のゲームの面 C と同じ材料・加工時間の4つの注文を使いますが、練習なので、納期と売上は次の段階まで付けません。ただ早く動かすのではなく、空いた1刻にもう1台を掃除する計画を考えてみてください。

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

03_cleaning.py / 注目する部分
        fields = {"ctype": "-"} if machine == "C" else {"ftype": "-"}
        next_state = replace(next_state, **fields, cleans=state.cleans + 1)

実行して確かめる

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

Windows
.venv\Scripts\python.exe 03_cleaning.py
macOS / Linux
.venv/bin/python 03_cleaning.py
画面で見ることCLEAN C → CUT b → CLEAN F → FINISH と指示します。T4 で金属の b が DONE @4、掃除の回数は2、両方の機械の設定が M です。CUT b の直後の C の残り1は、CLEAN F の間に0まで進みます。
  • 機械ごとに材料の設定を持つ
  • 掃除している間も、もう1台は進む
  • 完成品を持っている機械は掃除できない
  • 受け付けないときは、レールも時刻も変えない
つまずきやすいところC の設定を F と共有すると、片方を掃除しただけで両方がきれいになってしまいます。ctype と ftype を分け、指定した1台の項目だけを replace します。
少し変えてみよう

最初の状態から、金属の b だけを完成させる4つの指示を考えてみましょう。待ち時間を、WAIT だけに使わずに実行してください。

03_cleaning.pyをダウンロードこの段階だけで実行できます
このステップのコード全文(318行)
03_cleaning.py / 全文
# MIT License
# 
# Copyright (c) 2026
# 
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
# 
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
# 
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
# 
# Third-party exception: bitmap glyph data derived from Noto Sans CJK is licensed
# under the SIL Open Font License 1.1. See assets/NOTO-LICENSE.txt.

"""町工場の一日・教材 3/5。単体起動可能、外部素材なし。"""
from dataclasses import dataclass, replace

@dataclass(frozen=True)
class Job:
    id: str
    kind: str
    cut: int
    finish: int


@dataclass(frozen=True)
class Level:
    title: str
    jobs: tuple


LEVELS = (Level("Materials", (Job("a", "W", 1, 2),
                                    Job("b", "M", 2, 1),
                                    Job("c", "W", 1, 1),
                                    Job("d", "M", 1, 2))),)
CAPACITY = 2


@dataclass(frozen=True)
class State:
    t: int = 0
    ctype: str = "W"
    ftype: str = "W"
    cjob: int = -1
    crem: int = 0
    fjob: int = -1
    frem: int = 0
    queue: tuple = ()
    started: tuple = ()
    done: tuple = ()
    cleans: int = 0


def initial(level):
    return State(started=(False,) * len(level.jobs), done=(-1,) * len(level.jobs))


def actions(level):
    cuts = tuple(("cut", i) for i in range(len(level.jobs)))
    return cuts + (("finish",), ("clean", "C"), ("clean", "F"), ("wait",))


def apply(state, level, action):
    if action not in actions(level):
        return state, "invalid"
    next_state = state
    kind = action[0]
    if kind == "cut":
        index = action[1]
        job = level.jobs[index]
        if state.started[index]:
            return state, "started"
        if state.cjob >= 0:
            return state, "busy"
        if state.ctype not in ("-", job.kind):
            return state, "kind"
        started = list(state.started)
        started[index] = True
        next_state = replace(next_state, cjob=index, crem=job.cut,
                             ctype=job.kind, started=tuple(started))
    elif kind == "finish":
        if state.fjob >= 0:
            return state, "busy"
        if not state.queue:
            return state, "empty"
        index = state.queue[0]
        job = level.jobs[index]
        if state.ftype not in ("-", job.kind):
            return state, "kind"
        next_state = replace(next_state, fjob=index, frem=job.finish,
                             ftype=job.kind, queue=state.queue[1:])
    elif kind == "clean":
        machine = action[1]
        busy = state.cjob if machine == "C" else state.fjob
        material = state.ctype if machine == "C" else state.ftype
        if busy >= 0:
            return state, "busy"
        if material == "-":
            return state, "clean"
        fields = {"ctype": "-"} if machine == "C" else {"ftype": "-"}
        next_state = replace(next_state, **fields, cleans=state.cleans + 1)
    # 指示を反映した後で進める。開始した刻も1刻目に数える。
    next_state = replace(next_state, crem=max(0, next_state.crem - 1),
                         frem=max(0, next_state.frem - 1))
    # Fの取り出しは指示時。Cの搬出は同じ刻の最後。
    if next_state.fjob >= 0 and next_state.frem == 0:
        index = next_state.fjob
        done = list(next_state.done)
        done[index] = state.t + 1
        next_state = replace(next_state, fjob=-1, done=tuple(done))
    if next_state.cjob >= 0 and next_state.crem == 0 and len(next_state.queue) < CAPACITY:
        next_state = replace(next_state, queue=next_state.queue + (next_state.cjob,), cjob=-1)
    next_state = replace(next_state, t=state.t + 1)
    check(next_state, level)
    return next_state, "ok"


def check(state, level):
    assert state.t >= 0 and state.crem >= 0
    assert len(state.started) == len(level.jobs) == len(state.done)
    assert 0 <= len(state.queue) <= CAPACITY and state.frem >= 0
    owners = [i for i in (state.cjob, state.fjob) if i >= 0] + list(state.queue)
    assert state.fjob >= 0 or state.frem == 0
    owners += [i for i, time in enumerate(state.done) if time >= 0]
    assert len(owners) == len(set(owners))
    assert set(owners) == {i for i, started in enumerate(state.started) if started}
    assert all(0 <= i < len(level.jobs) for i in owners)
    assert state.cjob >= 0 or state.crem == 0
    assert state.ctype in ("W", "M", "-") and state.ftype in ("W", "M", "-")


def location(state, index):
    if state.done[index] >= 0:
        return "DONE @" + str(state.done[index])
    if state.cjob == index:
        return "C REM " + str(state.crem) if state.crem else "C HOLD"
    if state.fjob == index:
        return "F REM " + str(state.frem)
    if index in state.queue:
        return "QUEUE " + str(state.queue.index(index) + 1)
    return "NOT STARTED"


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.selection = 0
        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))
        choices = actions(self.level)
        if key in ("LEFT", "RIGHT"):
            self.selection = (self.selection + (1 if key == "RIGHT" else -1)) % len(choices)
        elif key in ("Z", "SPACE"):
            self.state, self.message = apply(self.state, self.level, choices[self.selection])
        else:
            return
        self.gated = True

    def draw(self, p):
        p.cls(1)
        p.text(12, 10, "SMALL FACTORY / STAGE 3", 10)
        state = self.state
        p.text(12, 27, self.level.title + "  T=" + str(state.t), 7)
        p.rect(12, 48, 164, 78, 2)
        p.rectb(12, 48, 164, 78, 6)
        p.text(22, 58, "C / CUTTING", 10)
        p.text(22, 77, "IDLE" if state.cjob < 0 else "JOB " + self.level.jobs[state.cjob].id, 7)
        p.text(22, 94, "REM " + str(state.crem), 11)
        p.text(22, 111, "MATERIAL " + state.ctype, 6)
        p.rect(336, 48, 164, 78, 2)
        p.rectb(336, 48, 164, 78, 6)
        p.text(346, 58, "F / FINISHING", 10)
        p.text(346, 77, "IDLE" if state.fjob < 0 else "JOB " + self.level.jobs[state.fjob].id, 7)
        p.text(346, 94, "REM " + str(state.frem), 11)
        p.text(346, 111, "MATERIAL " + state.ftype, 6)
        p.text(194, 58, "RAIL / FIFO " + str(len(state.queue)) + "/" + str(CAPACITY), 6)
        for slot in range(CAPACITY):
            x = 197 + slot * 50
            p.rectb(x, 77, 38, 26, 5)
            if slot < len(state.queue):
                p.text(x + 12, 87, self.level.jobs[state.queue[slot]].id, 10)
        p.text(194, 115, "HEAD -> TAIL", 13)
        p.text(12, 144, "JOB  TIMES   LOCATION", 13)
        p.text(300, 144, "TYPE", 13)
        for index, job in enumerate(self.level.jobs):
            y = 162 + index * 17
            if state.done[index] >= 0:
                p.rect(10, y - 2, 491, 13, 3)
            times = str(job.cut)
            times += "/" + str(job.finish)
            p.text(12, y, job.id, 10)
            p.text(36, y, times, 7)
            p.text(96, y, location(state, index), 7)
            p.text(300, y, job.kind, 7)
        for index, action in enumerate(actions(self.level)):
            x = 12 + (index % 4) * 124
            y = 244 + (index // 4) * 23
            label = "CUT " + self.level.jobs[action[1]].id if action[0] == "cut" else " ".join(action)
            p.rectb(x, y, 118, 19, 10 if index == self.selection else 5)
            p.text(x + 5, y + 6, label.upper(), 10 if index == self.selection else 7)
        p.text(12, 305, "W = WOOD / M = METAL / - = CLEAN; CLEAN COSTS 1 TURN.", 6)
        p.text(12, 340, self.message.upper(), 11)
        p.text(12, 359, "LEFT/RIGHT SELECT  Z/SPACE DO  R RESET  ESC QUIT", 7)


DEMO_ACTIONS = (('cut', 0), ('finish',), ('cut', 2), ('finish',), ('clean', 'C'), ('cut', 1), ('clean', 'F'))
KEYS = ("LEFT", "RIGHT", "Z", "SPACE", "X", "R", "1", "2", "3", "B")


def demo_keys():
    selection = 0
    choices = actions(LEVELS[0])
    result = []
    for action in DEMO_ACTIONS:
        if action == ("undo",):
            result.append("X")
            continue
        target = choices.index(action)
        while selection != target:
            result.append("RIGHT")
            selection = (selection + 1) % len(choices)
        result.append("Z")
    return tuple(result)


def parse_args(argv=None):
    import argparse
    from pathlib import Path
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--headless-smoke", action="store_true")
    parser.add_argument("--demo-checkpoint", action="store_true")
    parser.add_argument("--screenshot")
    parser.add_argument("--frames", type=int)
    args = parser.parse_args(argv)
    if (args.demo_checkpoint or args.screenshot or args.frames is not None) and not args.headless_smoke:
        parser.error("--demo-checkpoint / --screenshot / --frames require --headless-smoke")
    minimum = 2 * len(demo_keys()) + 2 if args.demo_checkpoint else 2
    if args.frames is not None and not minimum <= args.frames <= 600:
        parser.error("--frames must be between " + str(minimum) + " and 600")
    if args.screenshot:
        path = Path(args.screenshot)
        if path.suffix.lower() != ".png" or path.is_dir():
            parser.error("--screenshot must name a PNG file")
        args.screenshot = str(path.resolve())
    args.frames = args.frames or max(8, minimum)
    return args


def main(args):
    from pathlib import Path
    import pyxel as p
    if args.screenshot:
        Path(args.screenshot).parent.mkdir(parents=True, exist_ok=True)
    app = App()
    demo = demo_keys()
    frame = 0
    p.init(512, 396, title="Small Factory", fps=30, display_scale=2, headless=args.headless_smoke)
    keys = {key: getattr(p, "KEY_" + key) for key in KEYS}

    def update():
        nonlocal frame
        frame += 1
        if args.headless_smoke:
            if args.demo_checkpoint:
                index = (frame - 1) // 2
                pressed = {demo[index]} if frame % 2 == 1 and index < len(demo) else set()
                app.control(pressed, pressed)
        else:
            app.control({key for key, code in keys.items() if p.btnp(code)},
                        {key for key, code in keys.items() if p.btn(code)})

    def draw():
        app.draw(p)
        if args.headless_smoke and frame >= args.frames:
            if args.screenshot:
                p.screenshot(args.screenshot, scale=2)
            p.quit()
    p.run(update, draw)


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

納期と勤務終了を別々に判定する

できあがる時刻の区切り・売上・ゲームの終わりの状態。

注文に value(売上)と due(納期)を、面に end(勤務の終わり)と goal(目標)を加えます。1・2・3 キーで、元のゲームの A・B・C と同じ注文・納期・勤務時間・目標を選べます。切断が終わっただけでは売上になりません。仕上げの F が終わった時刻 state.t+1 を done に記録し、その時刻が due 以下なら満額、遅れたら max(1,value-1) です。納期ちょうども満額です。

1刻の中では、F の納品と売上の加算を先に処理し、最後に勤務の終わりを判定します。面 A の c を8刻目に仕上げれば、納期8の満額3が加わってから、end=8 の目標7を達成できます。先に勤務の終わりを判定すると、この最後の正しい納品が失われてしまいます。

目標に早く届いても、勤務時間が終わるまでは running(続行中)です。追加の注文を進めても、WAIT で時間を使ってもかまいません。end に達したときだけ、won(成功)か lost(失敗)になります。終わったあとに次の操作をしても、terminal として元の状態を返します。勝ったときの演出と状態の更新をまぜず、この判定を状態を書きかえない apply に置くことで、テストできるようになります。

A の解き方は「木 a → 仕上げ → 木 b → 仕上げ → C を掃除 → 金属 c → F を掃除 → 仕上げ」です。B には、短い2件で目標5をとる道と、長い L を選ぶ道があります。全部の注文を完成させることは、勝ちの条件ではありません。C では、先頭の材料・2つの枠のレール・2台が重なって動く時間を合わせて計画します。ここで、ふつうのモードの元のゲームのルールがそろいます。

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

04_deadlines.py / 注目する部分
        reward = job.value if state.t + 1 <= job.due else max(1, job.value - 1)
        next_state = replace(next_state, fjob=-1, done=tuple(done),
                             value=next_state.value + reward)

実行して確かめる

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

Windows
.venv\Scripts\python.exe 04_deadlines.py
macOS / Linux
.venv/bin/python 04_deadlines.py
画面で見ることA で CUT a → FINISH → CUT b → FINISH → CLEAN C → CUT c → CLEAN F → FINISH と指示します。T8、VALUE 7/7、done=(2,4,8)、WON です。
  • できあがった時刻 t+1 で納期と比べる
  • 納期ちょうどは満額
  • 最後の納品を足してから、勝ち負けを決める
  • 目標に届くことと、勤務が終わることを区別する
つまずきやすいところ納期と比べるときに state.t を使うと、1刻早くできあがったことになってしまいます。指示を出す前は t、今回できあがったのは t+1 です。勤務の終わりの判定も、足したあとの next_state を読みます。
少し変えてみよう

面 A の金属 c の due を8から7に変えて、同じ8つの指示を実行してみましょう。完成の時刻・売上の合計・勝ち負けは、どう変わるでしょうか。

04_deadlines.pyをダウンロードこの段階だけで実行できます
このステップのコード全文(351行)
04_deadlines.py / 全文
# MIT License
# 
# Copyright (c) 2026
# 
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
# 
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
# 
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
# 
# Third-party exception: bitmap glyph data derived from Noto Sans CJK is licensed
# under the SIL Open Font License 1.1. See assets/NOTO-LICENSE.txt.

"""町工場の一日・教材 4/5。単体起動可能、外部素材なし。"""
from dataclasses import dataclass, replace

@dataclass(frozen=True)
class Job:
    id: str
    kind: str
    cut: int
    finish: int
    value: int
    due: int


@dataclass(frozen=True)
class Level:
    title: str
    jobs: tuple
    end: int
    goal: int


LEVELS = (
    Level("Group wood", (Job("a", "W", 1, 1, 2, 4),
                         Job("b", "W", 1, 1, 2, 5),
                         Job("c", "M", 1, 1, 3, 8)), 8, 7),
    Level("Long or two short", (Job("a", "W", 1, 1, 2, 4),
                               Job("L", "W", 3, 3, 4, 8),
                               Job("b", "M", 1, 1, 3, 7)), 9, 5),
    Level("Head of the rail", (Job("a", "W", 1, 2, 2, 4),
                              Job("b", "M", 2, 1, 3, 9),
                              Job("c", "W", 1, 1, 2, 6),
                              Job("d", "M", 1, 2, 3, 11)), 11, 7),
)
CAPACITY = 2


@dataclass(frozen=True)
class State:
    t: int = 0
    ctype: str = "W"
    ftype: str = "W"
    cjob: int = -1
    crem: int = 0
    fjob: int = -1
    frem: int = 0
    queue: tuple = ()
    started: tuple = ()
    done: tuple = ()
    value: int = 0
    cleans: int = 0
    outcome: str = "running"


def initial(level):
    return State(started=(False,) * len(level.jobs), done=(-1,) * len(level.jobs))


def actions(level):
    cuts = tuple(("cut", i) for i in range(len(level.jobs)))
    return cuts + (("finish",), ("clean", "C"), ("clean", "F"), ("wait",))


def apply(state, level, action):
    if state.outcome != "running":
        return state, "terminal"
    if action not in actions(level):
        return state, "invalid"
    next_state = state
    kind = action[0]
    if kind == "cut":
        index = action[1]
        job = level.jobs[index]
        if state.started[index]:
            return state, "started"
        if state.cjob >= 0:
            return state, "busy"
        if state.ctype not in ("-", job.kind):
            return state, "kind"
        started = list(state.started)
        started[index] = True
        next_state = replace(next_state, cjob=index, crem=job.cut,
                             ctype=job.kind, started=tuple(started))
    elif kind == "finish":
        if state.fjob >= 0:
            return state, "busy"
        if not state.queue:
            return state, "empty"
        index = state.queue[0]
        job = level.jobs[index]
        if state.ftype not in ("-", job.kind):
            return state, "kind"
        next_state = replace(next_state, fjob=index, frem=job.finish,
                             ftype=job.kind, queue=state.queue[1:])
    elif kind == "clean":
        machine = action[1]
        busy = state.cjob if machine == "C" else state.fjob
        material = state.ctype if machine == "C" else state.ftype
        if busy >= 0:
            return state, "busy"
        if material == "-":
            return state, "clean"
        fields = {"ctype": "-"} if machine == "C" else {"ftype": "-"}
        next_state = replace(next_state, **fields, cleans=state.cleans + 1)
    # 指示を反映した後で進める。開始した刻も1刻目に数える。
    next_state = replace(next_state, crem=max(0, next_state.crem - 1),
                         frem=max(0, next_state.frem - 1))
    # Fの取り出しは指示時。Cの搬出は同じ刻の最後。
    if next_state.fjob >= 0 and next_state.frem == 0:
        index = next_state.fjob
        done = list(next_state.done)
        done[index] = state.t + 1
        job = level.jobs[index]
        reward = job.value if state.t + 1 <= job.due else max(1, job.value - 1)
        next_state = replace(next_state, fjob=-1, done=tuple(done),
                             value=next_state.value + reward)
    if next_state.cjob >= 0 and next_state.crem == 0 and len(next_state.queue) < CAPACITY:
        next_state = replace(next_state, queue=next_state.queue + (next_state.cjob,), cjob=-1)
    next_state = replace(next_state, t=state.t + 1)
    if next_state.t >= level.end:
        next_state = replace(next_state, outcome="won" if next_state.value >= level.goal else "lost")
    check(next_state, level)
    return next_state, "ok"


def check(state, level):
    assert state.t >= 0 and state.crem >= 0
    assert len(state.started) == len(level.jobs) == len(state.done)
    assert 0 <= len(state.queue) <= CAPACITY and state.frem >= 0
    owners = [i for i in (state.cjob, state.fjob) if i >= 0] + list(state.queue)
    assert state.fjob >= 0 or state.frem == 0
    owners += [i for i, time in enumerate(state.done) if time >= 0]
    assert len(owners) == len(set(owners))
    assert set(owners) == {i for i, started in enumerate(state.started) if started}
    assert all(0 <= i < len(level.jobs) for i in owners)
    assert state.cjob >= 0 or state.crem == 0
    assert state.ctype in ("W", "M", "-") and state.ftype in ("W", "M", "-")
    assert state.t <= level.end
    assert state.value == sum((job.value if time <= job.due else max(1, job.value - 1))
                              for job, time in zip(level.jobs, state.done) if time >= 0)


def location(state, index):
    if state.done[index] >= 0:
        return "DONE @" + str(state.done[index])
    if state.cjob == index:
        return "C REM " + str(state.crem) if state.crem else "C HOLD"
    if state.fjob == index:
        return "F REM " + str(state.frem)
    if index in state.queue:
        return "QUEUE " + str(state.queue.index(index) + 1)
    return "NOT STARTED"


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.selection = 0
        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 ("1", "2", "3"):
            self.level_index = int(key) - 1
            self.reset()
            self.gated = True
            return
        choices = actions(self.level)
        if key in ("LEFT", "RIGHT"):
            self.selection = (self.selection + (1 if key == "RIGHT" else -1)) % len(choices)
        elif key in ("Z", "SPACE"):
            self.state, self.message = apply(self.state, self.level, choices[self.selection])
        else:
            return
        self.gated = True

    def draw(self, p):
        p.cls(1)
        p.text(12, 10, "SMALL FACTORY / STAGE 4", 10)
        state = self.state
        p.text(12, 27, self.level.title + "  T=" + str(state.t), 7)
        p.text(330, 27, "END " + str(self.level.end) + "  VALUE " + str(state.value) + "/" + str(self.level.goal), 11)
        p.rect(12, 48, 164, 78, 2)
        p.rectb(12, 48, 164, 78, 6)
        p.text(22, 58, "C / CUTTING", 10)
        p.text(22, 77, "IDLE" if state.cjob < 0 else "JOB " + self.level.jobs[state.cjob].id, 7)
        p.text(22, 94, "REM " + str(state.crem), 11)
        p.text(22, 111, "MATERIAL " + state.ctype, 6)
        p.rect(336, 48, 164, 78, 2)
        p.rectb(336, 48, 164, 78, 6)
        p.text(346, 58, "F / FINISHING", 10)
        p.text(346, 77, "IDLE" if state.fjob < 0 else "JOB " + self.level.jobs[state.fjob].id, 7)
        p.text(346, 94, "REM " + str(state.frem), 11)
        p.text(346, 111, "MATERIAL " + state.ftype, 6)
        p.text(194, 58, "RAIL / FIFO " + str(len(state.queue)) + "/" + str(CAPACITY), 6)
        for slot in range(CAPACITY):
            x = 197 + slot * 50
            p.rectb(x, 77, 38, 26, 5)
            if slot < len(state.queue):
                p.text(x + 12, 87, self.level.jobs[state.queue[slot]].id, 10)
        p.text(194, 115, "HEAD -> TAIL", 13)
        p.text(12, 144, "JOB  TIMES   LOCATION", 13)
        p.text(300, 144, "TYPE", 13)
        p.text(358, 144, "VALUE  DUE", 13)
        for index, job in enumerate(self.level.jobs):
            y = 162 + index * 17
            if state.done[index] >= 0:
                p.rect(10, y - 2, 491, 13, 3)
            times = str(job.cut)
            times += "/" + str(job.finish)
            p.text(12, y, job.id, 10)
            p.text(36, y, times, 7)
            p.text(96, y, location(state, index), 7)
            p.text(300, y, job.kind, 7)
            p.text(358, y, str(job.value) + "      " + str(job.due), 7)
        for index, action in enumerate(actions(self.level)):
            x = 12 + (index % 4) * 124
            y = 244 + (index // 4) * 23
            label = "CUT " + self.level.jobs[action[1]].id if action[0] == "cut" else " ".join(action)
            p.rectb(x, y, 118, 19, 10 if index == self.selection else 5)
            p.text(x + 5, y + 6, label.upper(), 10 if index == self.selection else 7)
        p.text(12, 305, "DUE TURN IS FULL VALUE; LATE = MAX(1, VALUE-1).", 6)
        p.text(12, 340, self.message.upper() + " / " + state.outcome.upper(), 11)
        p.text(12, 359, "LEFT/RIGHT SELECT  Z/SPACE DO  R RESET  ESC QUIT", 7)
        p.text(12, 376, "1/2/3 LEVEL", 13)


DEMO_ACTIONS = (('cut', 0), ('finish',), ('cut', 1), ('finish',), ('clean', 'C'), ('cut', 2), ('clean', 'F'), ('finish',))
KEYS = ("LEFT", "RIGHT", "Z", "SPACE", "X", "R", "1", "2", "3", "B")


def demo_keys():
    selection = 0
    choices = actions(LEVELS[0])
    result = []
    for action in DEMO_ACTIONS:
        if action == ("undo",):
            result.append("X")
            continue
        target = choices.index(action)
        while selection != target:
            result.append("RIGHT")
            selection = (selection + 1) % len(choices)
        result.append("Z")
    return tuple(result)


def parse_args(argv=None):
    import argparse
    from pathlib import Path
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--headless-smoke", action="store_true")
    parser.add_argument("--demo-checkpoint", action="store_true")
    parser.add_argument("--screenshot")
    parser.add_argument("--frames", type=int)
    args = parser.parse_args(argv)
    if (args.demo_checkpoint or args.screenshot or args.frames is not None) and not args.headless_smoke:
        parser.error("--demo-checkpoint / --screenshot / --frames require --headless-smoke")
    minimum = 2 * len(demo_keys()) + 2 if args.demo_checkpoint else 2
    if args.frames is not None and not minimum <= args.frames <= 600:
        parser.error("--frames must be between " + str(minimum) + " and 600")
    if args.screenshot:
        path = Path(args.screenshot)
        if path.suffix.lower() != ".png" or path.is_dir():
            parser.error("--screenshot must name a PNG file")
        args.screenshot = str(path.resolve())
    args.frames = args.frames or max(8, minimum)
    return args


def main(args):
    from pathlib import Path
    import pyxel as p
    if args.screenshot:
        Path(args.screenshot).parent.mkdir(parents=True, exist_ok=True)
    app = App()
    demo = demo_keys()
    frame = 0
    p.init(512, 396, title="Small Factory", fps=30, display_scale=2, headless=args.headless_smoke)
    keys = {key: getattr(p, "KEY_" + key) for key in KEYS}

    def update():
        nonlocal frame
        frame += 1
        if args.headless_smoke:
            if args.demo_checkpoint:
                index = (frame - 1) // 2
                pressed = {demo[index]} if frame % 2 == 1 and index < len(demo) else set()
                app.control(pressed, pressed)
        else:
            app.control({key for key, code in keys.items() if p.btnp(code)},
                        {key for key, code in keys.items() if p.btn(code)})

    def draw():
        app.draw(p)
        if args.headless_smoke and frame >= args.frames:
            if args.screenshot:
                p.screenshot(args.screenshot, scale=2)
            p.quit()
    p.run(update, draw)


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

予告とUndoで計画を試し直す

同じ関数の使い回しと、状態の記録(スナップショット)の履歴。

apply は元の State を書きかえず、次の State と理由を返します。この性質を使い、選んだ指示を実行する前に、forecast(予報)で同じ apply を呼びます。NEXT の欄には、次の時刻・売上・レールの数を表示し、受け付けない指示なら BUSY などの理由を出します。画面を描くたびに計算しても工場は動かず、履歴も増えません。予告と本番で、別々のルールを書く必要はありません。

Timeline は、今の State と過去の State を並べたタプルです。step は、何かが変わったときだけ、今の State を history の最後に足します。受け付けない指示まで積むと、X を押しても見た目が変わらない履歴ができてしまいます。Undo は、最後の State を取り出して、履歴を1つ短くするだけです。時間だけを引くやり方では、レール・材料の設定・残り時間・売上・完成の時刻・掃除の回数が戻りません。

X は、勤務が終わったあとにも使えます。step では Undo の処理を先に行い、そのあとで、終わったら受け付けない apply を呼びます。A を達成してから X を押すと、T7、売上4、レールに c、running に戻ります。そこでちがう指示を出すと、戻った地点から新しい履歴が続きます。やり直し(Redo)はありません。

B キーは、元のゲームの比較モードと同じく、「材料を切りかえるときの掃除がいらない」という条件だけを変えて、その面を最初に戻します。加工の時間、レールの容量、納期、目標は同じです。掃除の指示そのものは残ります。比較モードだからといって、加工の時間までただになるわけではありません。R と 1〜3 も履歴を最初に戻します。元のゲームの表紙・P での一時停止・日本語の文字・録画・イベントの記録は教材では省き、状態と入力の関係に集中しています。

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

05_undo.py / 注目する部分
    if action == ("undo",):
        if not timeline.history:
            return timeline, "no_undo"
        return Timeline(timeline.history[-1], timeline.history[:-1]), "undo"

実行して確かめる

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

Windows
.venv\Scripts\python.exe 05_undo.py
macOS / Linux
.venv/bin/python 05_undo.py
画面で見ること05 で、04 と同じ手順で A をクリアしたあとに X を押します。T7、VALUE 4/7、QUEUE c、RUNNING になります。FINISH をもう一度実行すると、T8・VALUE 7・WON に戻ります。NEXT を読むだけでは、T は変わりません。
  • 予告では状態も履歴も使わない
  • 受け付けない操作は履歴に積まない
  • すべての項目をまとめて戻す
  • 勤務が終わったあとも Undo できる
  • 比較モードは、材料の切りかえの条件だけを変える
つまずきやすいところapply で先に「終わったら受け付けない」と判定してから Undo しようとすると、失敗を戻せません。Undo を Timeline の操作として先に処理し、ふつうの指示だけを apply に渡します。
少し変えてみよう

A を達成した直後に X を押して、T・売上・レール・結果がどうなるかを予想してみましょう。続けて FINISH を押し、前と同じ結果になるか確かめます。

05_undo.pyをダウンロードこの段階だけで実行できます
このステップのコード全文(389行)
05_undo.py / 全文
# MIT License
# 
# Copyright (c) 2026
# 
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
# 
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
# 
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
# 
# Third-party exception: bitmap glyph data derived from Noto Sans CJK is licensed
# under the SIL Open Font License 1.1. See assets/NOTO-LICENSE.txt.

"""町工場の一日・教材 5/5。単体起動可能、外部素材なし。"""
from dataclasses import dataclass, replace

@dataclass(frozen=True)
class Job:
    id: str
    kind: str
    cut: int
    finish: int
    value: int
    due: int


@dataclass(frozen=True)
class Level:
    title: str
    jobs: tuple
    end: int
    goal: int


LEVELS = (
    Level("Group wood", (Job("a", "W", 1, 1, 2, 4),
                         Job("b", "W", 1, 1, 2, 5),
                         Job("c", "M", 1, 1, 3, 8)), 8, 7),
    Level("Long or two short", (Job("a", "W", 1, 1, 2, 4),
                               Job("L", "W", 3, 3, 4, 8),
                               Job("b", "M", 1, 1, 3, 7)), 9, 5),
    Level("Head of the rail", (Job("a", "W", 1, 2, 2, 4),
                              Job("b", "M", 2, 1, 3, 9),
                              Job("c", "W", 1, 1, 2, 6),
                              Job("d", "M", 1, 2, 3, 11)), 11, 7),
)
CAPACITY = 2


@dataclass(frozen=True)
class State:
    t: int = 0
    ctype: str = "W"
    ftype: str = "W"
    cjob: int = -1
    crem: int = 0
    fjob: int = -1
    frem: int = 0
    queue: tuple = ()
    started: tuple = ()
    done: tuple = ()
    value: int = 0
    cleans: int = 0
    outcome: str = "running"


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


def initial(level):
    return State(started=(False,) * len(level.jobs), done=(-1,) * len(level.jobs))


def actions(level):
    cuts = tuple(("cut", i) for i in range(len(level.jobs)))
    return cuts + (("finish",), ("clean", "C"), ("clean", "F"), ("wait",))


def apply(state, level, action, free_setup=False):
    if state.outcome != "running":
        return state, "terminal"
    if action not in actions(level):
        return state, "invalid"
    next_state = state
    kind = action[0]
    if kind == "cut":
        index = action[1]
        job = level.jobs[index]
        if state.started[index]:
            return state, "started"
        if state.cjob >= 0:
            return state, "busy"
        if not free_setup and state.ctype not in ("-", job.kind):
            return state, "kind"
        started = list(state.started)
        started[index] = True
        next_state = replace(next_state, cjob=index, crem=job.cut,
                             ctype=job.kind, started=tuple(started))
    elif kind == "finish":
        if state.fjob >= 0:
            return state, "busy"
        if not state.queue:
            return state, "empty"
        index = state.queue[0]
        job = level.jobs[index]
        if not free_setup and state.ftype not in ("-", job.kind):
            return state, "kind"
        next_state = replace(next_state, fjob=index, frem=job.finish,
                             ftype=job.kind, queue=state.queue[1:])
    elif kind == "clean":
        machine = action[1]
        busy = state.cjob if machine == "C" else state.fjob
        material = state.ctype if machine == "C" else state.ftype
        if busy >= 0:
            return state, "busy"
        if material == "-":
            return state, "clean"
        fields = {"ctype": "-"} if machine == "C" else {"ftype": "-"}
        next_state = replace(next_state, **fields, cleans=state.cleans + 1)
    # 指示を反映した後で進める。開始した刻も1刻目に数える。
    next_state = replace(next_state, crem=max(0, next_state.crem - 1),
                         frem=max(0, next_state.frem - 1))
    # Fの取り出しは指示時。Cの搬出は同じ刻の最後。
    if next_state.fjob >= 0 and next_state.frem == 0:
        index = next_state.fjob
        done = list(next_state.done)
        done[index] = state.t + 1
        job = level.jobs[index]
        reward = job.value if state.t + 1 <= job.due else max(1, job.value - 1)
        next_state = replace(next_state, fjob=-1, done=tuple(done),
                             value=next_state.value + reward)
    if next_state.cjob >= 0 and next_state.crem == 0 and len(next_state.queue) < CAPACITY:
        next_state = replace(next_state, queue=next_state.queue + (next_state.cjob,), cjob=-1)
    next_state = replace(next_state, t=state.t + 1)
    if next_state.t >= level.end:
        next_state = replace(next_state, outcome="won" if next_state.value >= level.goal else "lost")
    check(next_state, level)
    return next_state, "ok"


def check(state, level):
    assert state.t >= 0 and state.crem >= 0
    assert len(state.started) == len(level.jobs) == len(state.done)
    assert 0 <= len(state.queue) <= CAPACITY and state.frem >= 0
    owners = [i for i in (state.cjob, state.fjob) if i >= 0] + list(state.queue)
    assert state.fjob >= 0 or state.frem == 0
    owners += [i for i, time in enumerate(state.done) if time >= 0]
    assert len(owners) == len(set(owners))
    assert set(owners) == {i for i, started in enumerate(state.started) if started}
    assert all(0 <= i < len(level.jobs) for i in owners)
    assert state.cjob >= 0 or state.crem == 0
    assert state.ctype in ("W", "M", "-") and state.ftype in ("W", "M", "-")
    assert state.t <= level.end
    assert state.value == sum((job.value if time <= job.due else max(1, job.value - 1))
                              for job, time in zip(level.jobs, state.done) if time >= 0)


def forecast(state, level, action, free_setup=False):
    return apply(state, level, action, free_setup)


def step(timeline, level, action, free_setup=False):
    if action == ("undo",):
        if not timeline.history:
            return timeline, "no_undo"
        return Timeline(timeline.history[-1], timeline.history[:-1]), "undo"
    next_state, reason = apply(timeline.state, level, action, free_setup)
    if next_state == timeline.state:
        return timeline, reason
    return Timeline(next_state, timeline.history + (timeline.state,)), reason


def location(state, index):
    if state.done[index] >= 0:
        return "DONE @" + str(state.done[index])
    if state.cjob == index:
        return "C REM " + str(state.crem) if state.crem else "C HOLD"
    if state.fjob == index:
        return "F REM " + str(state.frem)
    if index in state.queue:
        return "QUEUE " + str(state.queue.index(index) + 1)
    return "NOT STARTED"


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

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

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

    def reset(self):
        self.timeline = Timeline(initial(self.level))
        self.selection = 0
        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 ("1", "2", "3"):
            self.level_index = int(key) - 1
            self.reset()
            self.gated = True
            return
        if key == "B":
            self.free_setup = not self.free_setup
            self.reset()
            self.gated = True
            return
        if key == "X":
            self.timeline, self.message = step(self.timeline, self.level, ("undo",), self.free_setup)
            self.gated = True
            return
        choices = actions(self.level)
        if key in ("LEFT", "RIGHT"):
            self.selection = (self.selection + (1 if key == "RIGHT" else -1)) % len(choices)
        elif key in ("Z", "SPACE"):
            self.timeline, self.message = step(self.timeline, self.level, choices[self.selection], self.free_setup)
        else:
            return
        self.gated = True

    def draw(self, p):
        p.cls(1)
        p.text(12, 10, "SMALL FACTORY / STAGE 5", 10)
        state = self.state
        p.text(12, 27, self.level.title + "  T=" + str(state.t), 7)
        p.text(330, 27, "END " + str(self.level.end) + "  VALUE " + str(state.value) + "/" + str(self.level.goal), 11)
        p.rect(12, 48, 164, 78, 2)
        p.rectb(12, 48, 164, 78, 6)
        p.text(22, 58, "C / CUTTING", 10)
        p.text(22, 77, "IDLE" if state.cjob < 0 else "JOB " + self.level.jobs[state.cjob].id, 7)
        p.text(22, 94, "REM " + str(state.crem), 11)
        p.text(22, 111, "MATERIAL " + state.ctype, 6)
        p.rect(336, 48, 164, 78, 2)
        p.rectb(336, 48, 164, 78, 6)
        p.text(346, 58, "F / FINISHING", 10)
        p.text(346, 77, "IDLE" if state.fjob < 0 else "JOB " + self.level.jobs[state.fjob].id, 7)
        p.text(346, 94, "REM " + str(state.frem), 11)
        p.text(346, 111, "MATERIAL " + state.ftype, 6)
        p.text(194, 58, "RAIL / FIFO " + str(len(state.queue)) + "/" + str(CAPACITY), 6)
        for slot in range(CAPACITY):
            x = 197 + slot * 50
            p.rectb(x, 77, 38, 26, 5)
            if slot < len(state.queue):
                p.text(x + 12, 87, self.level.jobs[state.queue[slot]].id, 10)
        p.text(194, 115, "HEAD -> TAIL", 13)
        p.text(12, 144, "JOB  TIMES   LOCATION", 13)
        p.text(300, 144, "TYPE", 13)
        p.text(358, 144, "VALUE  DUE", 13)
        for index, job in enumerate(self.level.jobs):
            y = 162 + index * 17
            if state.done[index] >= 0:
                p.rect(10, y - 2, 491, 13, 3)
            times = str(job.cut)
            times += "/" + str(job.finish)
            p.text(12, y, job.id, 10)
            p.text(36, y, times, 7)
            p.text(96, y, location(state, index), 7)
            p.text(300, y, job.kind, 7)
            p.text(358, y, str(job.value) + "      " + str(job.due), 7)
        for index, action in enumerate(actions(self.level)):
            x = 12 + (index % 4) * 124
            y = 244 + (index // 4) * 23
            label = "CUT " + self.level.jobs[action[1]].id if action[0] == "cut" else " ".join(action)
            p.rectb(x, y, 118, 19, 10 if index == self.selection else 5)
            p.text(x + 5, y + 6, label.upper(), 10 if index == self.selection else 7)
        preview, reason = forecast(state, self.level, actions(self.level)[self.selection], self.free_setup)
        line = "NEXT T=" + str(preview.t) + " VALUE=" + str(preview.value) + " QUEUE=" + str(len(preview.queue))
        p.text(12, 301, line if reason == "ok" else "NEXT: " + reason.upper(), 6)
        p.text(12, 317, "HISTORY " + str(len(self.timeline.history)) + "  FREE SETUP " + str(self.free_setup), 13)
        p.text(12, 340, self.message.upper() + " / " + state.outcome.upper(), 11)
        p.text(12, 359, "LEFT/RIGHT SELECT  Z/SPACE DO  R RESET  ESC QUIT", 7)
        p.text(12, 376, "1/2/3 LEVEL" + "  X UNDO  B FREE SETUP (RESETS)", 13)


DEMO_ACTIONS = (('cut', 0), ('finish',), ('cut', 1), ('finish',), ('clean', 'C'), ('cut', 2), ('clean', 'F'), ('finish',), ('undo',), ('finish',))
KEYS = ("LEFT", "RIGHT", "Z", "SPACE", "X", "R", "1", "2", "3", "B")


def demo_keys():
    selection = 0
    choices = actions(LEVELS[0])
    result = []
    for action in DEMO_ACTIONS:
        if action == ("undo",):
            result.append("X")
            continue
        target = choices.index(action)
        while selection != target:
            result.append("RIGHT")
            selection = (selection + 1) % len(choices)
        result.append("Z")
    return tuple(result)


def parse_args(argv=None):
    import argparse
    from pathlib import Path
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--headless-smoke", action="store_true")
    parser.add_argument("--demo-checkpoint", action="store_true")
    parser.add_argument("--screenshot")
    parser.add_argument("--frames", type=int)
    args = parser.parse_args(argv)
    if (args.demo_checkpoint or args.screenshot or args.frames is not None) and not args.headless_smoke:
        parser.error("--demo-checkpoint / --screenshot / --frames require --headless-smoke")
    minimum = 2 * len(demo_keys()) + 2 if args.demo_checkpoint else 2
    if args.frames is not None and not minimum <= args.frames <= 600:
        parser.error("--frames must be between " + str(minimum) + " and 600")
    if args.screenshot:
        path = Path(args.screenshot)
        if path.suffix.lower() != ".png" or path.is_dir():
            parser.error("--screenshot must name a PNG file")
        args.screenshot = str(path.resolve())
    args.frames = args.frames or max(8, minimum)
    return args


def main(args):
    from pathlib import Path
    import pyxel as p
    if args.screenshot:
        Path(args.screenshot).parent.mkdir(parents=True, exist_ok=True)
    app = App()
    demo = demo_keys()
    frame = 0
    p.init(512, 396, title="Small Factory", fps=30, display_scale=2, headless=args.headless_smoke)
    keys = {key: getattr(p, "KEY_" + key) for key in KEYS}

    def update():
        nonlocal frame
        frame += 1
        if args.headless_smoke:
            if args.demo_checkpoint:
                index = (frame - 1) // 2
                pressed = {demo[index]} if frame % 2 == 1 and index < len(demo) else set()
                app.control(pressed, pressed)
        else:
            app.control({key for key, code in keys.items() if p.btnp(code)},
                        {key for key, code in keys.items() if p.btn(code)})

    def draw():
        app.draw(p)
        if args.headless_smoke and frame >= args.frames:
            if args.screenshot:
                p.screenshot(args.screenshot, scale=2)
            p.quit()
    p.run(update, draw)


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

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

1刻に1つの指示です。始めた刻から C と F が同時に加工します。C の完成品は2つまで入る順番待ちのレールへ、いっぱいなら C が持ったままにします。F は、指示したときのレールの先頭だけを受け取ります。木と金属の切りかえには、空いた機械の掃除が1刻かかります。納期ちょうどは満額、遅れると売上-1(最低1)です。勤務が終わったときに目標の売上を判定します。Undo で1刻前の全部の状態に戻せます。

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

比べる点5ステップの教材元の完成版
段階01 は C だけで完成、02 は全部同じ材料で C と F、03 は元のゲームの面 C の材料と加工時間だけで納期なし、04 と 05 は元のゲームの3つの面のふつうのルールです。05 には比較モードも入っています。最初から A・B・C のすべてのルール・予告・Undo・比較があります。
表示と操作文字(ASCII)で横512×縦396、左右で選び、全部のキーを離すまで待ちます。R を押した瞬間を優先します。表紙・P での一時停止・上下での選択は省いています。日本語の文字、640×480、表紙と一時停止、上下左右での選択、M のメニュー、F12 の記録があります。
保存と構成ステップごとに1つのファイルです。画像の素材は読みこみません。ふつうに起動したときは記録を書かず、画面を出さないテストを指定したときだけ PNG を保存します。app.py・model.py・experiments.py に分かれていて、日本語の文字データ glyphs.json、イベントの記録と GIF の記録があります。
検証最後のプログラムは、元のゲームの1264の状態・13147通りの変化と、最後まで遊んだ48通りの記録で照らし合わせています。これは見本との照合で、すべての状態を調べつくしたものではありません。一緒に入っている探索の結果や作戦の比較は、元のゲームの別の確認です。教材の画面操作の確認とまぜないでください。

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

  1. model.py: applyのfinish開始→両countdown→F完成→C搬出→終端
  2. model.py: checkの所有権と価値保存
  3. model.py: forecast / step / state_hash
  4. experiments.py: search / policy / replay
  5. app.py: gateとprevious、表紙と停止時の入力順序
自分のゲームに変えるなら

最初は、数字かルールを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の実際のパソコンでの操作や、音が出るかどうかはまだ確認していません。人がすべての操作を試したわけでもありません。ブラウザー版の確認範囲は確認状況をご覧ください。

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