完成サンプルをブラウザーで遊ぶ
この講座のもとになった完成版のゲームを、ブラウザーでそのまま遊べます。キーボードで操作するので、PCで開いてください（スマホのタッチ操作には対応していません）。初めて開くときは、読み込みに少し時間がかかります。
完成版の主な操作：A/S/D打鍵または左右+Z、P停止、X前小節、R再開。表紙T方式/V速度/左右補正
ボタンを押すと、ゲームの読み込みが始まります。
画面に「CLICK TO START」と出たらクリックします。続けてゲーム画面を一度クリックすると、キー操作が効くようになります。ゲームを終えるときは、枠の外にある「閉じる」ボタンを使ってください（Escキーの働きはゲームごとに違います）。ゲーム内のセーブは、閉じる・ページを移る・再読み込みすると消えます。
うまく表示されないとき
ゲームを動かすためのプログラム（Pyxel 2.9.9）は、外部の配信サービス（jsDelivr）から読み込みます。表示されないときは「最初から再読込」を押すか、完成版のソースをダウンロードしてPCで実行してください。ブラウザー版の録画機能は止めています。ブラウザー版で確認できている範囲は検証状況にまとめています。
判定に使う時計と、音の予約を分ける。
いきなり完成版のコードを全部読む必要はありません。小さなゲームにルールを1つずつ足しながら、5つのステップで作っていきます。どのステップのファイルもそれだけで動くので、動きを確かめてから次へ進みましょう。
1. 今処理する刻と次の刻、主拍を受け付ける幅、押した瞬間
2. キーを押した刻（raw）、ずれを補正した判定の刻、譜面の刻の3つ
3. 押した時刻ではなく、譜面の刻に遅れを足す予約の列
- 必要な環境
- Python 3.11以上 / Pyxel 2.9.9
- 知っておくとよいこと
- Python 3.11以上、Pyxel 2.9.9、ウィンドウ表示（GUI/OpenGL）ができるパソコン / 変数・if・for・関数・
list/tuple/set/ 書き換えないdataclassの記録とreplace（本文で説明します） - 教材に入っているもの
- ステップごとに動くPythonファイル5本・説明・動作確認用のテスト
- 完成した教材の操作
- Enter：表紙からSTEP、演奏中AUTO/STEP切替 ／ Space：STEPで一刻だけ進める ／ A/S/D：木/瓦/トタン（01/02はAのみ） ／ 左右：02以降、表紙で判定補正-6～+6 ／ 1/2/3：04以降、表紙で原作譜面A/B/C ／ B：04以降、表紙でトタン返事45/15を比較 ／ V：04以降、表紙で速度100/75% ／ P：05のみ停止、もう一度で現在小節再開 ／ X：05のみ前小節から再開 ／ R：設定を維持して試行を初期化、ソースは読み直さない ／ Esc：終了
- 難易度の目安
- 中級このページで使うプログラムの考え方の数と難しさから決めた目安です。難易度について
- 学習順に迷ったら
- おすすめの学ぶ順番を見る
この講座の進め方① ページ下の教材ZIPをダウンロードして、すべて展開します。② はじめる準備を見ながら、PythonとPyxelを使える状態にします。③ まず 01_score_clock.py を、何も変えずにそのまま実行してみましょう。記事に載せているコードは説明用の一部分なので、実行するときはZIPに入っているファイルを使ってください。
コードを変更したら
コードを書き換えたら、ファイルを保存し、動いているゲームを閉じてから、もう一度同じ.pyを実行します。ゲーム中にRキーなどで「やり直し」をしても、書き換えたコードは読み込まれません。
教材のゲーム画面は、日本語フォントを用意しなくても動くように英語で表示しています。ファイルには動作確認用のコードも入っていますが、最初は Game / State・update・draw の3つに注目すれば十分です。わからない言葉は用語集で調べられます。class・self・dataclass はPythonの状態、キー入力と時間の扱いは入力・更新・描画、マス目とピクセルの違いは画面とマップの座標のページで詳しく説明しています。
入力を採点する60刻の時計
今処理する刻と次の刻、主拍を受け付ける幅、押した瞬間。
雨が屋根を叩く譜面を、画面とは別の整数の時計として持ちます。元のゲームの状況 A は4小節で、主拍（必ず叩く拍）は0・30・60・90刻、そのあとも120刻ごとに同じ並びです。BAR=120 が1小節です。ルール上の時計の1刻はふつうの速さで1/60秒、画面は1秒に30回の更新なので、自動で進めるときは、1回の画面の更新で2刻を処理します。75%の速さはあとの段階で扱います。始まる前の -120〜-1刻はカウントイン（数え始め）で、0刻が最初の主拍です。画面の NEXT RAW は、直前に終わった刻ではなく、次に処理する刻です。
最初に Enter を押すと、表紙から STEP（1刻ずつ進める）になります。Space を1回押して離すと、1刻だけ進みます。もう一度 Enter を押して離すと AUTO（自動で進める）になり、また Enter で STEP に戻ります。A は木の屋根です。STEP の間に A を押すのも、その入力を付けて1刻進める操作です。S/D・飾りの採点・返事は、まだ使いません。01/02 では最後の返事を待つ必要がないので、終わりの余韻も475刻（処理後の表示は476）に短くしています。画面と State の基本の形はあとの段階と共通なので、decor や reservations という空の入れ物も先に用意しています。
step は、今の t を保存してから、入力の判定、カウント、見逃しを処理し、最後に tick=t+1 の新しい State を返します。主拍30なら、26〜34刻（±4刻）で受け付けます。34刻ならまだ成功でき、押さないまま35刻を処理すると miss（見逃し）です。0はまだ判定していない、1は手で叩いて成功、-1は見逃しです。同じ主拍を2回叩いても、2回目は wrong で、点は増えません。
fresh=down-previous は、前の画面の更新では押していなかったキーの集まりです。長押ししても、叩いたのは1回だけです。Enter や R の直後は、キーが全部離されるまで入力を止める仕組み（gate）で受け付けず、始める操作に混じった A が演奏に流れこむのを防ぎます。draw は状態を読むだけです。屋根が光るのは、音が鳴った画面の更新を age0 として、0〜4の5回の更新だけです。ただし、STEP で時計を止めても画面の経過は進むので、光り方と採点の時間は別のものです。
下のコードは 01_score_clock.py の一部です。前のステップから追加・変更したところだけを抜き出しています。
01_score_clock.py / 注目する部分
corrected = s.tick + settings.correction
for i, t in enumerate(level.main):
if abs(corrected - t) <= 4:
return ('main', i, t, s.main[i] == 0)実行して確かめる
教材のZIPを展開したフォルダーで、次のコマンドを実行します。
Windows
.venv\Scripts\python.exe 01_score_clock.pymacOS / Linux
.venv/bin/python 01_score_clock.py画面で見ること起動したときは TITLE・NEXT RAW -120 です。Enter を押して離し、Space を押すと -119 になります。AUTO で進めて、主拍0の前後で A を押すと、緑の四角と MAIN が増えます。自動確認では、0/30/60 を木の屋根で叩き、NEXT RAW 76・MAIN 3/16・DECOR 0・ECHO 0 です。元のゲームと同じく、主拍を12/16以上叩けば、基本のクリアです。
tickは「次に処理する刻」- 主拍の成功と見逃しの境目を分ける
- 押しっぱなしを、連打として数えない
つまずきやすいところ画面の76は、75刻まで処理し終えたという意味です。描いた回数を、そのまま採点の刻にしないでください。また、R は状態を最初に戻す操作で、保存した Python のコードを読み直す操作ではありません。
少し変えてみよう
match_slot の「abs(corrected - t) <= 4」を「<= 2」に1か所変え、保存 → ゲームを閉じる → 同じ 01 のファイルを実行し直してみましょう。0刻の主拍は、2刻ずれなら受け付け、3刻ずれなら wrong です。この実験は受け付ける幅を狭めるだけなので、押さなかったときに miss になる時刻は、元の5刻ずれのままです。確認したら4に戻します。テストでは、2刻と3刻のずれを、別々の最初の状態から比べています。
01_score_clock.pyをダウンロードこの段階だけで実行できます
このステップのコード全文（538行）
01_score_clock.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。"""
"""066 Deterministic 60-tick musical clock. No rendering or audio dependencies."""
from dataclasses import dataclass, asdict, replace
import hashlib, json
BAR = 120
ROOFS = ('木', '瓦', 'トタン')
@dataclass(frozen=True)
class Level:
id: str
title: str
main: tuple[int, ...]
decor: tuple[int, ...]
rests: tuple[int, ...]
bars: int = 4
def build(id, title, patterns):
main = tuple((b * BAR + t for b, pat in enumerate(patterns) for t in pat))
decor = tuple((b * BAR + t for b in range(4) for t in (15, 45, 75, 105)))
rests = tuple((b * BAR + t for b, pat in enumerate(patterns) for t in (0, 30, 60, 90) if t not in pat))
return Level(id, title, main, decor, rests)
LEVELS = (build('066-A-v1', 'まっすぐの雨', ((0, 30, 60, 90),) * 4), build('066-B-v1', '空いた三拍目', ((0, 30, 90),) * 4), build('066-C-v1', '返事を待つ軒', ((0, 60, 90), (0, 30, 90), (0, 60, 90), (0, 30, 90))))
@dataclass(frozen=True)
class Settings:
correction: int = 0
tin_delay: int = 45
speed: int = 100
control: str = 'asd'
def __post_init__(self):
if not -6 <= self.correction <= 6:
raise ValueError('correction -6..6')
if self.tin_delay not in (15, 45):
raise ValueError('tin delay 15 or45')
if self.speed not in (75, 100):
raise ValueError('speed 75 or100')
if self.control not in ('asd', 'arrows'):
raise ValueError('unknown input mode')
@dataclass(frozen=True)
class Reservation:
id: int
roof: int
target: int
reserved_at: int
source_slot: int
@dataclass(frozen=True)
class Event:
tick: int
kind: str
roof: int = -1
slot: int = -1
reservation: int = -1
@dataclass(frozen=True)
class Input:
tick: int
roof: int
corrected: int
kind: str
slot: int
accepted: bool
@dataclass(frozen=True)
class State:
tick: int
main: tuple[int, ...]
decor: tuple[bool, ...]
reservations: tuple[Reservation, ...] = ()
events: tuple[Event, ...] = ()
inputs: tuple[Input, ...] = ()
done: bool = False
def initial(level):
return State(-BAR, tuple((0 for _ in level.main)), tuple((False for _ in level.decor)))
def delay(roof, settings):
return (0, 15, settings.tin_delay)[roof]
def close_tick(level, settings):
return max(max(level.main) + 4, max(level.decor) + 3) - settings.correction
def ending_tick(level, settings):
return max(max(level.main), max(level.decor)) + 10
def match_slot(s, level, settings):
corrected = s.tick + settings.correction
for i, t in enumerate(level.main):
if abs(corrected - t) <= 4:
return ('main', i, t, s.main[i] == 0)
return ('outside', -1, -1, False)
def future(s):
groups = {}
for r in s.reservations:
if r.target >= s.tick:
groups.setdefault(r.target, []).append(r)
out = []
for target, rs in sorted(groups.items()):
order = sorted(rs, key=lambda r: (r.reserved_at, r.roof, r.id))
out.extend(((r, r == order[0]) for r in order))
return tuple(out)
def reservation_preview(s, level, settings, roof):
corrected = s.tick + settings.correction
slots = [t for t, v in zip(level.main, s.main) if v == 0 and corrected <= t + 4]
slots += [t for t, v in zip(level.decor, s.decor) if not v and corrected <= t + 3]
if not slots:
return None
slot = min(slots)
d = 0 # 01/02: no scheduled replies yet
target = slot + d if d else None
return {'slot': slot, 'target': target, 'overlap': bool(d and any((r.target == target for r in s.reservations)))}
def step(s, level, settings, roof=None):
"""Process raw tick t, manual input, due echo, misses; then advance to t+1."""
if s.done:
return s
if roof is not None and roof not in (0, 1, 2):
raise ValueError('roof must be0/1/2')
t = s.tick
main = list(s.main)
decor = list(s.decor)
reservations = list(s.reservations)
events = list(s.events)
inputs = list(s.inputs)
if roof is not None and t <= close_tick(level, settings):
kind, i, slot, accepted = match_slot(s, level, settings)
inputs.append(Input(t, roof, t + settings.correction, kind, slot, accepted))
if accepted:
if kind == 'main':
main[i] = 1
else:
decor[i] = True
events.append(Event(t, 'manual', roof, slot))
d = 0 # 01/02: no scheduled replies yet
if d:
assert slot + d > t, 'A legal reply must be future, never retroactive.'
reservations.append(Reservation(len(reservations), roof, slot + d, t, slot))
else:
events.append(Event(t, 'wrong', roof, slot))
elif roof is not None:
events.append(Event(t, 'closed', roof))
if t < 0 and t % 30 == 0:
events.append(Event(t, 'count'))
due = sorted((r for r in reservations if r.target == t), key=lambda r: (r.reserved_at, r.roof, r.id))
if due:
events.append(Event(t, 'echo', due[0].roof, due[0].source_slot, due[0].id))
for r in due[1:]:
events.append(Event(t, 'suppressed', r.roof, r.source_slot, r.id))
for i, slot in enumerate(level.main):
if main[i] == 0 and t + settings.correction > slot + 4:
main[i] = -1
events.append(Event(t, 'miss', -1, slot))
return State(t + 1, tuple(main), tuple(decor), tuple(reservations), tuple(events), tuple(inputs), t >= ending_tick(level, settings))
def stats(s, level):
hit = sum((x == 1 for x in s.main))
dec = sum(s.decor)
echoes = [e for e in s.events if e.kind == 'echo']
manual = [e for e in s.events if e.kind == 'manual']
echo_targets = {e.tick for e in echoes}
overlaps = sum((1 for e in manual if e.slot in level.decor and e.slot in echo_targets))
used = {e.slot for e in manual} | echo_targets
return {'hits': hit, 'total': len(level.main), 'hit_percent': 100 * hit // len(level.main), 'complete': hit * 4 >= len(level.main) * 3, 'decorations': dec, 'automatic_replies': len(echoes), 'overlaid_slots': overlaps, 'blank_decorations': sum((t not in used for t in level.decor)), 'reservation_collisions': sum((e.kind == 'suppressed' for e in s.events)), 'wrong': sum((e.kind == 'wrong' for e in s.events)), 'manual_hits': len(manual)}
def normalized(v):
return json.dumps(v, ensure_ascii=False, sort_keys=True, separators=(',', ':'))
def digest(v):
return hashlib.sha256(normalized(v).encode('utf-8')).hexdigest()
STAGE = 1
# Display uses ASCII so this one file needs no glyph or image assets.
# The original sound definitions are retained, without claiming audible testing.
SOUNDS = (('c2c1', 't', '64', 'ff', 4), ('f4c4', 'p', '54', 'ff', 3),
('c3c2', 'n', '43', 'ff', 4), ('c4', 't', '3', 'f', 3))
COLORS = (9, 12, 6)
class Controller:
"""Pyxel-independent input edges and score clock. tick is the NEXT raw tick."""
def __init__(self):
self.level_index = 0
self.settings = Settings()
self.frame = 0
self.previous = set()
self.pending = []
self.audio_log = []
self.reset(stop=False)
@property
def level(self):
return LEVELS[self.level_index]
@property
def state(self):
return self.session.state if STAGE == 5 else self._state
def reset(self, stop=True):
if stop:
self.stop_audio()
if STAGE == 5:
self.session = new_session(self.level, self.settings)
else:
self._state = initial(self.level)
self.running = False
self.title = True
self.paused = False
self.gate = False
self.countdown = 0
self.accum = 0
self.flashes = [-1000000] * 3
self.caption = 'PRESS ENTER TO START. SPACE = ONE SCORE TICK.'
def emit(self, kind, tick, channel=None, sound=None):
item = {'kind': kind, 'tick': tick, 'frame': self.frame,
'channel': channel, 'sound': sound}
self.pending.append(item)
self.audio_log.append(item.copy())
def stop_audio(self):
self.emit('stop', self.state.tick)
self.flashes = [-1000000] * 3
def apply_tick(self, roof=None):
if self.state.done:
return
before = len(self.state.events)
if STAGE == 5:
self.session = session_step(self.session, self.level, roof)
else:
self._state = step(self.state, self.level, self.settings, roof)
for event in self.state.events[before:]:
if event.kind == 'manual':
self.emit('manual', event.tick, event.roof, event.roof)
self.flashes[event.roof] = self.frame
self.caption = 'MANUAL raw %d / score %d' % (event.tick, event.slot)
elif event.kind == 'echo':
self.emit('echo', event.tick, 3, event.roof)
self.flashes[event.roof] = self.frame
self.caption = 'ECHO target %d / source %d' % (event.tick, event.slot)
elif event.kind == 'count':
self.emit('count', event.tick, 3, 3)
elif event.kind == 'suppressed':
self.caption = 'SAME TARGET: EARLIER RESERVATION WINS'
elif event.kind == 'wrong':
self.caption = 'OUTSIDE WINDOW OR SLOT ALREADY USED'
if self.state.done:
self.running = False
self.gate = True
def rewind(self, previous=False):
if STAGE != 5:
return
self.stop_audio()
self.session = restart_bar(self.session, self.level, previous)
self.countdown = BAR
self.accum = 0
self.running = True
self.paused = False
self.gate = True
self.caption = 'RESTART: SCORE FROZEN DURING 120-TICK COUNT'
def control(self, down):
"""One call per renderer frame; fresh is set difference, never key repeat."""
self.frame += 1
down = set(down)
fresh = down - self.previous
self.previous = down
if self.gate:
if not down:
self.gate = False
return
if 'R' in fresh:
self.reset()
self.gate = True
return
if self.title:
changed = False
if STAGE >= 2:
correction = self.settings.correction
if 'LEFT' in fresh:
correction = max(-6, correction - 1)
elif 'RIGHT' in fresh:
correction = min(6, correction + 1)
changed = correction != self.settings.correction
self.settings = replace(self.settings, correction=correction)
if STAGE >= 4:
for i, key in enumerate(('1', '2', '3')):
if key in fresh:
self.level_index = i
changed = True
break
if 'B' in fresh:
self.settings = replace(self.settings, tin_delay=15 if self.settings.tin_delay == 45 else 45)
changed = True
if 'V' in fresh:
self.settings = replace(self.settings, speed=75 if self.settings.speed == 100 else 100)
changed = True
if changed:
self.reset()
self.gate = True
return
if 'RETURN' in fresh:
self.title = False
self.gate = True
self.caption = 'SPACE: STEP / ENTER: AUTO / A S D: ONE FRESH HIT'
return
if STAGE == 5 and 'P' in fresh:
if self.paused:
self.rewind(False)
else:
self.paused = True
self.running = False
self.stop_audio()
self.gate = True
return
if self.paused:
return
if STAGE == 5 and 'X' in fresh:
self.rewind(True)
return
if self.state.done:
return
if 'RETURN' in fresh:
self.running = not self.running
if not self.running:
self.stop_audio()
self.gate = True
return
# A chord is one hit; the canonical order is A, then S, then D.
roof = next((i for i, key in enumerate(('A', 'S', 'D'))
if key in fresh and (STAGE >= 3 or i == 0)), None)
if not self.running:
if 'SPACE' in fresh or roof is not None:
self.apply_tick(roof)
return
# 30 renderer frames/sec; 2 logical ticks/frame (or 1,2,1,2 at 75%).
self.accum += 4 if self.settings.speed == 100 else 3
dt, self.accum = divmod(self.accum, 2)
if self.countdown:
if self.countdown % 30 < dt:
self.emit('restart_count', self.state.tick, 3, 3)
self.countdown = max(0, self.countdown - dt)
if not self.countdown:
self.gate = True
return
for i in range(dt):
self.apply_tick(roof if i == 0 else None)
def flush_audio(self, pyxel):
"""Drain only the new effects; drawing or draining twice never replays them."""
pending, self.pending = self.pending, []
for event in pending:
if event['kind'] == 'stop':
pyxel.stop()
else:
pyxel.play(event['channel'], event['sound'])
return pending
def demo(self):
self.title = False
inputs = {0: 0, 30: 0, 60: 0} if STAGE < 3 else {0: 2, 30: 1, 60: 0}
if STAGE == 2:
inputs[15] = 0
while self.state.tick <= 75:
self.frame += 1
self.apply_tick(inputs.get(self.state.tick))
self.caption = ('CHECKPOINT: 3 MAIN HITS' if STAGE < 3 else
'0+45 = 30+15: SAME TARGET 45')
def draw(self, p):
s = self.state
stat = stats(s, self.level)
p.cls(1)
p.text(14, 10, 'RAIN ROOF DRUMS / STAGE %d / %s' % (STAGE, self.level.id), 10)
p.text(14, 27, 'NEXT RAW %d / CORRECTED %d / SPEED %d%%' %
(s.tick, s.tick + self.settings.correction, self.settings.speed), 7)
mode = 'TITLE' if self.title else 'PAUSED' if self.paused else 'RESULT' if s.done else 'AUTO' if self.running else 'STEP'
p.text(14, 44, mode + ' / 60 SCORE TICKS = 1 SEC AT 100% / 120 = ONE BAR', 6)
p.text(14, 63, 'MAIN %d/%d DECOR %d ECHO %d SUPPRESSED %d' %
(stat['hits'], stat['total'], stat['decorations'], stat['automatic_replies'], stat['reservation_collisions']), 11)
bar = max(0, min(3, max(0, s.tick) // BAR))
for row in range(2):
start = (bar + row) * BAR
y = 88 + row * 80
p.rect(12, y, 616, 71, 0)
label = 'BAR %d' % (bar + row + 1) if bar + row < 4 else 'TAIL'
p.text(18, y + 4, '%s / SCORE %d..%d' % (label, start, start + 119), 6)
for t in range(start, start + 120, 15):
x = 85 + (t - start) * 4
p.line(x, y + 20, x, y + 63, 5)
p.text(x - 4, y + 11, str(t), 5)
if t in self.level.main:
value = s.main[self.level.main.index(t)]
p.rectb(x - 4, y + 26, 9, 9, 11 if value == 1 else 8 if value == -1 else 7)
elif t in self.level.rests:
p.text(x - 5, y + 28, '-', 13)
if STAGE >= 2 and t in self.level.decor:
p.circb(x, y + 44, 3, 10 if s.decor[self.level.decor.index(t)] else 5)
if STAGE >= 3:
for reservation, active in future(s):
if start <= reservation.target < start + 120:
x = 85 + (reservation.target - start) * 4
color = COLORS[reservation.roof] if active or STAGE == 3 else 5
if reservation.roof == 1:
p.circb(x, y + 54, 5, color)
else:
p.tri(x, y + 49, x - 4, y + 57, x + 4, y + 57, color)
if start <= s.tick < start + 120:
x = 85 + (s.tick - start) * 4
p.line(x, y + 21, x, y + 64, 10)
for i, name in enumerate(('A WOOD', 'S TILE +15', 'D TIN +%d' % self.settings.tin_delay)):
x = 18 + i * 207
enabled = i == 0 or STAGE >= 3
# Age is zero on the event frame; ages 0..4 light five frames.
lit = 0 <= self.frame - self.flashes[i] < 5
p.tri(x, 286, x + 92, 250, x + 183, 286, COLORS[i] if lit else 5)
p.rectb(x + 12, 287, 158, 28, COLORS[i] if enabled else 5)
p.text(x + 22, 298, name if enabled else 'ADDED IN STAGE 03', 7 if enabled else 5)
p.text(14, 326, self.caption, 10)
p.text(14, 344, 'LAST EVENTS: raw tick / kind / roof / source slot', 6)
for i, e in enumerate(s.events[-6:]):
p.text(22, 358 + i * 12, '%4d %-10s roof %2d score %4d' % (e.tick, e.kind, e.roof, e.slot), 7)
if STAGE >= 3:
p.text(365, 344, 'NEXT REPLIES', 6)
pending = future(s)[:4]
for i, (reservation, active) in enumerate(pending):
p.text(365, 360 + i * 12, '%d %s %s' % (reservation.target, ('WOOD', 'TILE', 'TIN')[reservation.roof], 'ACTIVE' if active or STAGE == 3 else 'MUTED'), COLORS[reservation.roof] if active or STAGE == 3 else 5)
if not pending:
p.text(365, 360, '(none)', 5)
if self.countdown:
p.text(365, 415, 'RESTART COUNT %d' % self.countdown, 10)
if s.done:
p.text(365, 415, 'COMPLETE' if stat['complete'] else 'TRY AGAIN', 11)
p.text(14, 438, 'ENTER: START / AUTO-STEP TOGGLE SPACE: ONE TICK R: RESET', 7)
p.text(14, 453, ('A: WOOD' if STAGE < 3 else 'A/S/D: ROOFS') + ' ESC: QUIT HOLD = NO REPEAT', 7)
p.text(14, 468, 'TITLE: LEFT/RIGHT CORRECTION' if STAGE >= 2 else 'STAGE 01: MAIN BEATS ONLY / NO REPLIES', 6)
if STAGE >= 4:
p.text(300, 468, '1/2/3 CHART B DELAY V SPEED', 6)
p.text(14, 484, 'P: PAUSE / RESTART CURRENT BAR X: PREVIOUS BAR' if STAGE == 5 else 'SAVE EDIT -> QUIT -> RUN THIS SAME FILE AGAIN', 13)
if self.title:
p.rect(70, 106, 500, 100, 1)
p.rectb(70, 106, 500, 100, 12)
p.text(85, 120, 'PRESS ENTER. THEN SPACE STEPS OR ENTER RUNS.', 10)
p.text(85, 140, 'MAIN: +/-4 TICKS. DECOR FROM STAGE 02: +/-3.', 7)
p.text(85, 160, 'NEXT RAW = TICK THAT THE NEXT UPDATE WILL PROCESS.', 6)
p.text(85, 180, 'R RESETS STATE. IT DOES NOT RELOAD SAVED CODE.', 13)
def main(headless=False, screenshot=None, report=None, checkpoint=False):
# CLI validates and resolves paths before reaching this import.
import pyxel as p
p.init(640, 504, title='Rain roof lesson %d' % STAGE, fps=30,
display_scale=1, headless=headless)
for i, sound in enumerate(SOUNDS):
p.sounds[i].set(*sound)
app = Controller()
actual_calls = []
if checkpoint:
app.title = False
keys = {k: getattr(p, 'KEY_' + k) for k in
('RETURN', 'SPACE', 'R', 'A', 'S', 'D', 'P', 'X', 'LEFT', 'RIGHT', '1', '2', '3', 'B', 'V')}
def update():
if not headless:
app.control({k for k, value in keys.items() if p.btn(value)})
elif checkpoint and app.state.tick <= 75:
app.frame += 1
inputs = {0: 0, 30: 0, 60: 0} if STAGE < 3 else {0: 2, 30: 1, 60: 0}
if STAGE == 2:
inputs[15] = 0
for _ in range(2):
if app.state.tick <= 75:
app.apply_tick(inputs.get(app.state.tick))
actual_calls.extend(app.flush_audio(p))
if p.btnp(p.KEY_ESCAPE) or headless and p.frame_count >= 3 and (not checkpoint or app.state.tick > 75):
app.stop_audio()
actual_calls.extend(app.flush_audio(p))
app.draw(p)
if screenshot:
p.screen.save(screenshot, 1)
if report:
from pathlib import Path
payload = {'stage': STAGE, 'state': asdict(app.state),
'stats': stats(app.state, app.level), 'audio_calls': actual_calls,
'scope': 'Actual Pyxel API calls; dummy driver is not audible verification',
'screenshot': screenshot}
Path(report).write_text(json.dumps(payload, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
p.quit()
p.run(update, lambda: app.draw(p))
if __name__ == '__main__':
import argparse
from pathlib import Path
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--headless-smoke', action='store_true')
parser.add_argument('--demo-checkpoint', action='store_true')
parser.add_argument('--screenshot')
parser.add_argument('--report')
args = parser.parse_args()
if (args.demo_checkpoint or args.screenshot or args.report) and not args.headless_smoke:
parser.error('checkpoint and output files require --headless-smoke')
for name, suffix in (('screenshot', '.png'), ('report', '.json')):
value = getattr(args, name)
if value:
path = Path(value).resolve()
if path.suffix.lower() != suffix or not path.parent.is_dir() or path.exists():
parser.error('use a NEW ' + suffix + ' file in an existing directory')
if path.parent == Path(__file__).resolve().parent:
parser.error('output must be outside the script directory')
setattr(args, name, str(path))
main(args.headless_smoke, args.screenshot, args.report, args.demo_checkpoint)装飾と入力補正を、音の時刻から分ける
キーを押した刻（raw）、ずれを補正した判定の刻、譜面の刻の3つ。
小さな丸の飾りの音を採点に加えます。飾りは15・45・75・105刻と、各小節の同じ位置にあります。主拍の間を埋めてもいいし、空けてもいい、自由に足せる音です。成功の条件は「主拍の75%以上」だけで、飾りを全部取る必要はありません。この段階でも使う屋根は A の木だけで、返事はまだ予約しません。
受け付ける幅は、主拍が±4、飾りが±3です。譜面15刻の飾りは、補正したあとの12〜18刻で受け付けます。match_slot は、主拍を先に、飾りをあとに調べます。元のゲームの譜面では、この2種類の受け付ける幅は重なりませんが、調べる順番もルールの一部です。もう取った飾りをもう一度叩くと wrong で、False から True に変わるのは1回だけです。入力が失敗しても、Inputs には raw・corrected・候補の枠（slot）・受け付けたか（accepted）を記録します。
表紙で左右キーを押すと、ずれの補正 correction を -6〜+6刻に設定できます。判定に使う値は raw+correction です。たとえば補正 +3 で raw 12 なら、corrected は15なので、飾り15に当たります。ただし、手で叩いた音（manual）の記録の tick は raw の12のままです。音を鳴らす命令を呼ぶのも、その入力を処理した時点です。補正は、音をあとの時刻にずらす設定ではなく、譜面と照らし合わせるときだけ、ずらす設定です。
データと音をつなぐのは Controller.apply_tick です。step が新しく追加した出来事（イベント）だけを取り出し、manual なら、屋根の番号と同じチャンネル・音を鳴らす予定にします。flush_audio は一度取り出した通知を空にするので、draw を何回呼んでも、音を鳴らし直しません。これは、入力を二重に採点しないこととは別の、「鳴らす通知を二重に実行しない」ための仕組みです。設定の変更は表紙で行い、挑戦を最初から作り直します。
下のコードは 02_judgment.py の一部です。前のステップから追加・変更したところだけを抜き出しています。
02_judgment.py / 注目する部分
inputs.append(Input(t, roof, t + settings.correction, kind, slot, accepted))実行して確かめる
教材のZIPを展開したフォルダーで、次のコマンドを実行します。
Windows
.venv\Scripts\python.exe 02_judgment.pymacOS / Linux
.venv/bin/python 02_judgment.py画面で見ること自動確認では、主拍 0/30/60 と飾り15を木で叩き、NEXT RAW 76・MAIN 3/16・DECOR 1・ECHO 0 です。表紙で補正を +3 にすると、NEXT RAW と CORRECTED の差はいつも3になります。R を押したあとも、補正の値はそのままです。
- 自由な飾りの音は、基本のクリアの条件に入れない
- raw と corrected と slot（譜面の枠）を分ける
- 鳴らす通知は、一度だけ使う
つまずきやすいところcorrected を、手で叩いた音の時刻として記録しないでください。入力が raw 12 なら、手で叩いた音は12、成功した譜面の枠は15、という2つの情報を残します。
少し変えてみよう
Controller.__init__ の「self.settings = Settings()」を「self.settings = Settings(correction=3)」に変えてみましょう。保存・終了・実行し直したら、表紙の補正が +3 になっていることを確かめます。テストでは、raw 12 の A が飾り15に成功し、入力の corrected 15、手で叩いた音の tick 12、予約0件になることを確かめました。終わったら、最初の値0に戻します。
02_judgment.pyをダウンロードこの段階だけで実行できます
このステップのコード全文（541行）
02_judgment.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。"""
"""066 Deterministic 60-tick musical clock. No rendering or audio dependencies."""
from dataclasses import dataclass, asdict, replace
import hashlib, json
BAR = 120
ROOFS = ('木', '瓦', 'トタン')
@dataclass(frozen=True)
class Level:
id: str
title: str
main: tuple[int, ...]
decor: tuple[int, ...]
rests: tuple[int, ...]
bars: int = 4
def build(id, title, patterns):
main = tuple((b * BAR + t for b, pat in enumerate(patterns) for t in pat))
decor = tuple((b * BAR + t for b in range(4) for t in (15, 45, 75, 105)))
rests = tuple((b * BAR + t for b, pat in enumerate(patterns) for t in (0, 30, 60, 90) if t not in pat))
return Level(id, title, main, decor, rests)
LEVELS = (build('066-A-v1', 'まっすぐの雨', ((0, 30, 60, 90),) * 4), build('066-B-v1', '空いた三拍目', ((0, 30, 90),) * 4), build('066-C-v1', '返事を待つ軒', ((0, 60, 90), (0, 30, 90), (0, 60, 90), (0, 30, 90))))
@dataclass(frozen=True)
class Settings:
correction: int = 0
tin_delay: int = 45
speed: int = 100
control: str = 'asd'
def __post_init__(self):
if not -6 <= self.correction <= 6:
raise ValueError('correction -6..6')
if self.tin_delay not in (15, 45):
raise ValueError('tin delay 15 or45')
if self.speed not in (75, 100):
raise ValueError('speed 75 or100')
if self.control not in ('asd', 'arrows'):
raise ValueError('unknown input mode')
@dataclass(frozen=True)
class Reservation:
id: int
roof: int
target: int
reserved_at: int
source_slot: int
@dataclass(frozen=True)
class Event:
tick: int
kind: str
roof: int = -1
slot: int = -1
reservation: int = -1
@dataclass(frozen=True)
class Input:
tick: int
roof: int
corrected: int
kind: str
slot: int
accepted: bool
@dataclass(frozen=True)
class State:
tick: int
main: tuple[int, ...]
decor: tuple[bool, ...]
reservations: tuple[Reservation, ...] = ()
events: tuple[Event, ...] = ()
inputs: tuple[Input, ...] = ()
done: bool = False
def initial(level):
return State(-BAR, tuple((0 for _ in level.main)), tuple((False for _ in level.decor)))
def delay(roof, settings):
return (0, 15, settings.tin_delay)[roof]
def close_tick(level, settings):
return max(max(level.main) + 4, max(level.decor) + 3) - settings.correction
def ending_tick(level, settings):
return max(max(level.main), max(level.decor)) + 10
def match_slot(s, level, settings):
corrected = s.tick + settings.correction
for i, t in enumerate(level.main):
if abs(corrected - t) <= 4:
return ('main', i, t, s.main[i] == 0)
for i, t in enumerate(level.decor):
if abs(corrected - t) <= 3:
return ('decor', i, t, not s.decor[i])
return ('outside', -1, -1, False)
def future(s):
groups = {}
for r in s.reservations:
if r.target >= s.tick:
groups.setdefault(r.target, []).append(r)
out = []
for target, rs in sorted(groups.items()):
order = sorted(rs, key=lambda r: (r.reserved_at, r.roof, r.id))
out.extend(((r, r == order[0]) for r in order))
return tuple(out)
def reservation_preview(s, level, settings, roof):
corrected = s.tick + settings.correction
slots = [t for t, v in zip(level.main, s.main) if v == 0 and corrected <= t + 4]
slots += [t for t, v in zip(level.decor, s.decor) if not v and corrected <= t + 3]
if not slots:
return None
slot = min(slots)
d = 0 # 01/02: no scheduled replies yet
target = slot + d if d else None
return {'slot': slot, 'target': target, 'overlap': bool(d and any((r.target == target for r in s.reservations)))}
def step(s, level, settings, roof=None):
"""Process raw tick t, manual input, due echo, misses; then advance to t+1."""
if s.done:
return s
if roof is not None and roof not in (0, 1, 2):
raise ValueError('roof must be0/1/2')
t = s.tick
main = list(s.main)
decor = list(s.decor)
reservations = list(s.reservations)
events = list(s.events)
inputs = list(s.inputs)
if roof is not None and t <= close_tick(level, settings):
kind, i, slot, accepted = match_slot(s, level, settings)
inputs.append(Input(t, roof, t + settings.correction, kind, slot, accepted))
if accepted:
if kind == 'main':
main[i] = 1
else:
decor[i] = True
events.append(Event(t, 'manual', roof, slot))
d = 0 # 01/02: no scheduled replies yet
if d:
assert slot + d > t, 'A legal reply must be future, never retroactive.'
reservations.append(Reservation(len(reservations), roof, slot + d, t, slot))
else:
events.append(Event(t, 'wrong', roof, slot))
elif roof is not None:
events.append(Event(t, 'closed', roof))
if t < 0 and t % 30 == 0:
events.append(Event(t, 'count'))
due = sorted((r for r in reservations if r.target == t), key=lambda r: (r.reserved_at, r.roof, r.id))
if due:
events.append(Event(t, 'echo', due[0].roof, due[0].source_slot, due[0].id))
for r in due[1:]:
events.append(Event(t, 'suppressed', r.roof, r.source_slot, r.id))
for i, slot in enumerate(level.main):
if main[i] == 0 and t + settings.correction > slot + 4:
main[i] = -1
events.append(Event(t, 'miss', -1, slot))
return State(t + 1, tuple(main), tuple(decor), tuple(reservations), tuple(events), tuple(inputs), t >= ending_tick(level, settings))
def stats(s, level):
hit = sum((x == 1 for x in s.main))
dec = sum(s.decor)
echoes = [e for e in s.events if e.kind == 'echo']
manual = [e for e in s.events if e.kind == 'manual']
echo_targets = {e.tick for e in echoes}
overlaps = sum((1 for e in manual if e.slot in level.decor and e.slot in echo_targets))
used = {e.slot for e in manual} | echo_targets
return {'hits': hit, 'total': len(level.main), 'hit_percent': 100 * hit // len(level.main), 'complete': hit * 4 >= len(level.main) * 3, 'decorations': dec, 'automatic_replies': len(echoes), 'overlaid_slots': overlaps, 'blank_decorations': sum((t not in used for t in level.decor)), 'reservation_collisions': sum((e.kind == 'suppressed' for e in s.events)), 'wrong': sum((e.kind == 'wrong' for e in s.events)), 'manual_hits': len(manual)}
def normalized(v):
return json.dumps(v, ensure_ascii=False, sort_keys=True, separators=(',', ':'))
def digest(v):
return hashlib.sha256(normalized(v).encode('utf-8')).hexdigest()
STAGE = 2
# Display uses ASCII so this one file needs no glyph or image assets.
# The original sound definitions are retained, without claiming audible testing.
SOUNDS = (('c2c1', 't', '64', 'ff', 4), ('f4c4', 'p', '54', 'ff', 3),
('c3c2', 'n', '43', 'ff', 4), ('c4', 't', '3', 'f', 3))
COLORS = (9, 12, 6)
class Controller:
"""Pyxel-independent input edges and score clock. tick is the NEXT raw tick."""
def __init__(self):
self.level_index = 0
self.settings = Settings()
self.frame = 0
self.previous = set()
self.pending = []
self.audio_log = []
self.reset(stop=False)
@property
def level(self):
return LEVELS[self.level_index]
@property
def state(self):
return self.session.state if STAGE == 5 else self._state
def reset(self, stop=True):
if stop:
self.stop_audio()
if STAGE == 5:
self.session = new_session(self.level, self.settings)
else:
self._state = initial(self.level)
self.running = False
self.title = True
self.paused = False
self.gate = False
self.countdown = 0
self.accum = 0
self.flashes = [-1000000] * 3
self.caption = 'PRESS ENTER TO START. SPACE = ONE SCORE TICK.'
def emit(self, kind, tick, channel=None, sound=None):
item = {'kind': kind, 'tick': tick, 'frame': self.frame,
'channel': channel, 'sound': sound}
self.pending.append(item)
self.audio_log.append(item.copy())
def stop_audio(self):
self.emit('stop', self.state.tick)
self.flashes = [-1000000] * 3
def apply_tick(self, roof=None):
if self.state.done:
return
before = len(self.state.events)
if STAGE == 5:
self.session = session_step(self.session, self.level, roof)
else:
self._state = step(self.state, self.level, self.settings, roof)
for event in self.state.events[before:]:
if event.kind == 'manual':
self.emit('manual', event.tick, event.roof, event.roof)
self.flashes[event.roof] = self.frame
self.caption = 'MANUAL raw %d / score %d' % (event.tick, event.slot)
elif event.kind == 'echo':
self.emit('echo', event.tick, 3, event.roof)
self.flashes[event.roof] = self.frame
self.caption = 'ECHO target %d / source %d' % (event.tick, event.slot)
elif event.kind == 'count':
self.emit('count', event.tick, 3, 3)
elif event.kind == 'suppressed':
self.caption = 'SAME TARGET: EARLIER RESERVATION WINS'
elif event.kind == 'wrong':
self.caption = 'OUTSIDE WINDOW OR SLOT ALREADY USED'
if self.state.done:
self.running = False
self.gate = True
def rewind(self, previous=False):
if STAGE != 5:
return
self.stop_audio()
self.session = restart_bar(self.session, self.level, previous)
self.countdown = BAR
self.accum = 0
self.running = True
self.paused = False
self.gate = True
self.caption = 'RESTART: SCORE FROZEN DURING 120-TICK COUNT'
def control(self, down):
"""One call per renderer frame; fresh is set difference, never key repeat."""
self.frame += 1
down = set(down)
fresh = down - self.previous
self.previous = down
if self.gate:
if not down:
self.gate = False
return
if 'R' in fresh:
self.reset()
self.gate = True
return
if self.title:
changed = False
if STAGE >= 2:
correction = self.settings.correction
if 'LEFT' in fresh:
correction = max(-6, correction - 1)
elif 'RIGHT' in fresh:
correction = min(6, correction + 1)
changed = correction != self.settings.correction
self.settings = replace(self.settings, correction=correction)
if STAGE >= 4:
for i, key in enumerate(('1', '2', '3')):
if key in fresh:
self.level_index = i
changed = True
break
if 'B' in fresh:
self.settings = replace(self.settings, tin_delay=15 if self.settings.tin_delay == 45 else 45)
changed = True
if 'V' in fresh:
self.settings = replace(self.settings, speed=75 if self.settings.speed == 100 else 100)
changed = True
if changed:
self.reset()
self.gate = True
return
if 'RETURN' in fresh:
self.title = False
self.gate = True
self.caption = 'SPACE: STEP / ENTER: AUTO / A S D: ONE FRESH HIT'
return
if STAGE == 5 and 'P' in fresh:
if self.paused:
self.rewind(False)
else:
self.paused = True
self.running = False
self.stop_audio()
self.gate = True
return
if self.paused:
return
if STAGE == 5 and 'X' in fresh:
self.rewind(True)
return
if self.state.done:
return
if 'RETURN' in fresh:
self.running = not self.running
if not self.running:
self.stop_audio()
self.gate = True
return
# A chord is one hit; the canonical order is A, then S, then D.
roof = next((i for i, key in enumerate(('A', 'S', 'D'))
if key in fresh and (STAGE >= 3 or i == 0)), None)
if not self.running:
if 'SPACE' in fresh or roof is not None:
self.apply_tick(roof)
return
# 30 renderer frames/sec; 2 logical ticks/frame (or 1,2,1,2 at 75%).
self.accum += 4 if self.settings.speed == 100 else 3
dt, self.accum = divmod(self.accum, 2)
if self.countdown:
if self.countdown % 30 < dt:
self.emit('restart_count', self.state.tick, 3, 3)
self.countdown = max(0, self.countdown - dt)
if not self.countdown:
self.gate = True
return
for i in range(dt):
self.apply_tick(roof if i == 0 else None)
def flush_audio(self, pyxel):
"""Drain only the new effects; drawing or draining twice never replays them."""
pending, self.pending = self.pending, []
for event in pending:
if event['kind'] == 'stop':
pyxel.stop()
else:
pyxel.play(event['channel'], event['sound'])
return pending
def demo(self):
self.title = False
inputs = {0: 0, 30: 0, 60: 0} if STAGE < 3 else {0: 2, 30: 1, 60: 0}
if STAGE == 2:
inputs[15] = 0
while self.state.tick <= 75:
self.frame += 1
self.apply_tick(inputs.get(self.state.tick))
self.caption = ('CHECKPOINT: 3 MAIN HITS' if STAGE < 3 else
'0+45 = 30+15: SAME TARGET 45')
def draw(self, p):
s = self.state
stat = stats(s, self.level)
p.cls(1)
p.text(14, 10, 'RAIN ROOF DRUMS / STAGE %d / %s' % (STAGE, self.level.id), 10)
p.text(14, 27, 'NEXT RAW %d / CORRECTED %d / SPEED %d%%' %
(s.tick, s.tick + self.settings.correction, self.settings.speed), 7)
mode = 'TITLE' if self.title else 'PAUSED' if self.paused else 'RESULT' if s.done else 'AUTO' if self.running else 'STEP'
p.text(14, 44, mode + ' / 60 SCORE TICKS = 1 SEC AT 100% / 120 = ONE BAR', 6)
p.text(14, 63, 'MAIN %d/%d DECOR %d ECHO %d SUPPRESSED %d' %
(stat['hits'], stat['total'], stat['decorations'], stat['automatic_replies'], stat['reservation_collisions']), 11)
bar = max(0, min(3, max(0, s.tick) // BAR))
for row in range(2):
start = (bar + row) * BAR
y = 88 + row * 80
p.rect(12, y, 616, 71, 0)
label = 'BAR %d' % (bar + row + 1) if bar + row < 4 else 'TAIL'
p.text(18, y + 4, '%s / SCORE %d..%d' % (label, start, start + 119), 6)
for t in range(start, start + 120, 15):
x = 85 + (t - start) * 4
p.line(x, y + 20, x, y + 63, 5)
p.text(x - 4, y + 11, str(t), 5)
if t in self.level.main:
value = s.main[self.level.main.index(t)]
p.rectb(x - 4, y + 26, 9, 9, 11 if value == 1 else 8 if value == -1 else 7)
elif t in self.level.rests:
p.text(x - 5, y + 28, '-', 13)
if STAGE >= 2 and t in self.level.decor:
p.circb(x, y + 44, 3, 10 if s.decor[self.level.decor.index(t)] else 5)
if STAGE >= 3:
for reservation, active in future(s):
if start <= reservation.target < start + 120:
x = 85 + (reservation.target - start) * 4
color = COLORS[reservation.roof] if active or STAGE == 3 else 5
if reservation.roof == 1:
p.circb(x, y + 54, 5, color)
else:
p.tri(x, y + 49, x - 4, y + 57, x + 4, y + 57, color)
if start <= s.tick < start + 120:
x = 85 + (s.tick - start) * 4
p.line(x, y + 21, x, y + 64, 10)
for i, name in enumerate(('A WOOD', 'S TILE +15', 'D TIN +%d' % self.settings.tin_delay)):
x = 18 + i * 207
enabled = i == 0 or STAGE >= 3
# Age is zero on the event frame; ages 0..4 light five frames.
lit = 0 <= self.frame - self.flashes[i] < 5
p.tri(x, 286, x + 92, 250, x + 183, 286, COLORS[i] if lit else 5)
p.rectb(x + 12, 287, 158, 28, COLORS[i] if enabled else 5)
p.text(x + 22, 298, name if enabled else 'ADDED IN STAGE 03', 7 if enabled else 5)
p.text(14, 326, self.caption, 10)
p.text(14, 344, 'LAST EVENTS: raw tick / kind / roof / source slot', 6)
for i, e in enumerate(s.events[-6:]):
p.text(22, 358 + i * 12, '%4d %-10s roof %2d score %4d' % (e.tick, e.kind, e.roof, e.slot), 7)
if STAGE >= 3:
p.text(365, 344, 'NEXT REPLIES', 6)
pending = future(s)[:4]
for i, (reservation, active) in enumerate(pending):
p.text(365, 360 + i * 12, '%d %s %s' % (reservation.target, ('WOOD', 'TILE', 'TIN')[reservation.roof], 'ACTIVE' if active or STAGE == 3 else 'MUTED'), COLORS[reservation.roof] if active or STAGE == 3 else 5)
if not pending:
p.text(365, 360, '(none)', 5)
if self.countdown:
p.text(365, 415, 'RESTART COUNT %d' % self.countdown, 10)
if s.done:
p.text(365, 415, 'COMPLETE' if stat['complete'] else 'TRY AGAIN', 11)
p.text(14, 438, 'ENTER: START / AUTO-STEP TOGGLE SPACE: ONE TICK R: RESET', 7)
p.text(14, 453, ('A: WOOD' if STAGE < 3 else 'A/S/D: ROOFS') + ' ESC: QUIT HOLD = NO REPEAT', 7)
p.text(14, 468, 'TITLE: LEFT/RIGHT CORRECTION' if STAGE >= 2 else 'STAGE 01: MAIN BEATS ONLY / NO REPLIES', 6)
if STAGE >= 4:
p.text(300, 468, '1/2/3 CHART B DELAY V SPEED', 6)
p.text(14, 484, 'P: PAUSE / RESTART CURRENT BAR X: PREVIOUS BAR' if STAGE == 5 else 'SAVE EDIT -> QUIT -> RUN THIS SAME FILE AGAIN', 13)
if self.title:
p.rect(70, 106, 500, 100, 1)
p.rectb(70, 106, 500, 100, 12)
p.text(85, 120, 'PRESS ENTER. THEN SPACE STEPS OR ENTER RUNS.', 10)
p.text(85, 140, 'MAIN: +/-4 TICKS. DECOR FROM STAGE 02: +/-3.', 7)
p.text(85, 160, 'NEXT RAW = TICK THAT THE NEXT UPDATE WILL PROCESS.', 6)
p.text(85, 180, 'R RESETS STATE. IT DOES NOT RELOAD SAVED CODE.', 13)
def main(headless=False, screenshot=None, report=None, checkpoint=False):
# CLI validates and resolves paths before reaching this import.
import pyxel as p
p.init(640, 504, title='Rain roof lesson %d' % STAGE, fps=30,
display_scale=1, headless=headless)
for i, sound in enumerate(SOUNDS):
p.sounds[i].set(*sound)
app = Controller()
actual_calls = []
if checkpoint:
app.title = False
keys = {k: getattr(p, 'KEY_' + k) for k in
('RETURN', 'SPACE', 'R', 'A', 'S', 'D', 'P', 'X', 'LEFT', 'RIGHT', '1', '2', '3', 'B', 'V')}
def update():
if not headless:
app.control({k for k, value in keys.items() if p.btn(value)})
elif checkpoint and app.state.tick <= 75:
app.frame += 1
inputs = {0: 0, 30: 0, 60: 0} if STAGE < 3 else {0: 2, 30: 1, 60: 0}
if STAGE == 2:
inputs[15] = 0
for _ in range(2):
if app.state.tick <= 75:
app.apply_tick(inputs.get(app.state.tick))
actual_calls.extend(app.flush_audio(p))
if p.btnp(p.KEY_ESCAPE) or headless and p.frame_count >= 3 and (not checkpoint or app.state.tick > 75):
app.stop_audio()
actual_calls.extend(app.flush_audio(p))
app.draw(p)
if screenshot:
p.screen.save(screenshot, 1)
if report:
from pathlib import Path
payload = {'stage': STAGE, 'state': asdict(app.state),
'stats': stats(app.state, app.level), 'audio_calls': actual_calls,
'scope': 'Actual Pyxel API calls; dummy driver is not audible verification',
'screenshot': screenshot}
Path(report).write_text(json.dumps(payload, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
p.quit()
p.run(update, lambda: app.draw(p))
if __name__ == '__main__':
import argparse
from pathlib import Path
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--headless-smoke', action='store_true')
parser.add_argument('--demo-checkpoint', action='store_true')
parser.add_argument('--screenshot')
parser.add_argument('--report')
args = parser.parse_args()
if (args.demo_checkpoint or args.screenshot or args.report) and not args.headless_smoke:
parser.error('checkpoint and output files require --headless-smoke')
for name, suffix in (('screenshot', '.png'), ('report', '.json')):
value = getattr(args, name)
if value:
path = Path(value).resolve()
if path.suffix.lower() != suffix or not path.parent.is_dir() or path.exists():
parser.error('use a NEW ' + suffix + ' file in an existing directory')
if path.parent == Path(__file__).resolve().parent:
parser.error('output must be outside the script directory')
setattr(args, name, str(path))
main(args.headless_smoke, args.screenshot, args.report, args.demo_checkpoint)譜面から十五刻・四十五刻先へ返事を予約
押した時刻ではなく、譜面の刻に遅れを足す予約の列。
A の木に加えて、S の瓦と D のトタンを使えるようにします。木は手で叩いた音だけ、瓦は譜面の刻+15、トタンは譜面の刻+45 で、自動の返事を1回返します。返事の基準は、押した raw の時刻ではなく、受け付けた譜面の枠（slot）です。補正 +6 で raw -6 の D が主拍0に成功しても、手で叩いた音は -6、返事は45です。raw+45=39 にしてはいけません。
予約 Reservation には、id、roof（屋根）、target、reserved_at、source_slot を記録します。target は実際に返事を鳴らす刻、reserved_at は入力の raw の刻、source_slot は元の譜面の枠です。毎刻、target==t の予約だけを取り出して、返事（echo）の出来事を作ります。処理した予約を履歴から消さないかわりに、「等しい」で1刻だけ鳴らすので、くり返し鳴ることはありません。future は、まだ来ていない target を取り出す、読むだけの関数です。
予約するのは、入力が成功したときだけです。wrong、見逃し、カウント、返事（echo）では、新しい予約は作りません。そのため、返事がさらに返事を呼ぶ連鎖は起きません。入力が主拍でも飾りでも、屋根のルールは同じです。主拍の採点は手で叩いたときだけ増え、返事の音が主拍の位置に来ても、自動で成功にはなりません。
この段階は「予約を全部鳴らす」ところまでです。同じ target に予約が2つあれば、返事の出来事を2つ作り、同じ自動の音のチャンネル3に2回 play を呼びます。1つのチャンネルでは、あとの音が先の音を置きかえるので、2つの音が別々に聞こえるとは言えません。次の 04 で、元のゲームどおり「先に予約した1つ」に整理します。03 の返事の丸と三角は、すべての予約を有効の色で示し、まだ打ち消しを表しません。
下のコードは 03_delayed_reply.py の一部です。前のステップから追加・変更したところだけを抜き出しています。
03_delayed_reply.py / 注目する部分
reservations.append(Reservation(len(reservations), roof, slot + d, t, slot))実行して確かめる
教材のZIPを展開したフォルダーで、次のコマンドを実行します。
Windows
.venv\Scripts\python.exe 03_delayed_reply.pymacOS / Linux
.venv/bin/python 03_delayed_reply.py画面で見ること自動確認では、主拍0を D、30を S、60を A で叩きます。予約は 0+45 と 30+15 で、どちらも45刻に届きます。03 では ECHO 2・SUPPRESSED 0、入力は3回だけです。出来事の欄に「45 echo」が2行あり、主拍は 3/16 のままです。
- 返事の時刻は
source_slot＋遅れ - 予約を追加するのは、成功した入力だけ
- 03 の二重に鳴る音と、04 の打ち消しを区別する
つまずきやすいところ
target を「入力の tick＋遅れ」で求めないでください。また、返事の処理を手で叩いた入力に戻すと、連鎖が起きてしまいます。状態の予約の履歴を、描くたびに全部鳴らすのも、二重に鳴る原因です。少し変えてみよう
Controller.__init__ を「self.settings = Settings(tin_delay=15)」に変え、保存・終了・実行し直してみましょう。同じ 0:D、30:S の予約が、15刻と45刻に分かれます。テストでは、返事の刻が [15,45] になることを確かめています。15と45以外の値は、Settings が受け付けません。確認したら、最初の45に戻します。
03_delayed_reply.pyをダウンロードこの段階だけで実行できます
このステップのコード全文（540行）
03_delayed_reply.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。"""
"""066 Deterministic 60-tick musical clock. No rendering or audio dependencies."""
from dataclasses import dataclass, asdict, replace
import hashlib, json
BAR = 120
ROOFS = ('木', '瓦', 'トタン')
@dataclass(frozen=True)
class Level:
id: str
title: str
main: tuple[int, ...]
decor: tuple[int, ...]
rests: tuple[int, ...]
bars: int = 4
def build(id, title, patterns):
main = tuple((b * BAR + t for b, pat in enumerate(patterns) for t in pat))
decor = tuple((b * BAR + t for b in range(4) for t in (15, 45, 75, 105)))
rests = tuple((b * BAR + t for b, pat in enumerate(patterns) for t in (0, 30, 60, 90) if t not in pat))
return Level(id, title, main, decor, rests)
LEVELS = (build('066-A-v1', 'まっすぐの雨', ((0, 30, 60, 90),) * 4), build('066-B-v1', '空いた三拍目', ((0, 30, 90),) * 4), build('066-C-v1', '返事を待つ軒', ((0, 60, 90), (0, 30, 90), (0, 60, 90), (0, 30, 90))))
@dataclass(frozen=True)
class Settings:
correction: int = 0
tin_delay: int = 45
speed: int = 100
control: str = 'asd'
def __post_init__(self):
if not -6 <= self.correction <= 6:
raise ValueError('correction -6..6')
if self.tin_delay not in (15, 45):
raise ValueError('tin delay 15 or45')
if self.speed not in (75, 100):
raise ValueError('speed 75 or100')
if self.control not in ('asd', 'arrows'):
raise ValueError('unknown input mode')
@dataclass(frozen=True)
class Reservation:
id: int
roof: int
target: int
reserved_at: int
source_slot: int
@dataclass(frozen=True)
class Event:
tick: int
kind: str
roof: int = -1
slot: int = -1
reservation: int = -1
@dataclass(frozen=True)
class Input:
tick: int
roof: int
corrected: int
kind: str
slot: int
accepted: bool
@dataclass(frozen=True)
class State:
tick: int
main: tuple[int, ...]
decor: tuple[bool, ...]
reservations: tuple[Reservation, ...] = ()
events: tuple[Event, ...] = ()
inputs: tuple[Input, ...] = ()
done: bool = False
def initial(level):
return State(-BAR, tuple((0 for _ in level.main)), tuple((False for _ in level.decor)))
def delay(roof, settings):
return (0, 15, settings.tin_delay)[roof]
def close_tick(level, settings):
return max(max(level.main) + 4, max(level.decor) + 3) - settings.correction
def ending_tick(level, settings):
return max(max(level.main), max(level.decor)) + max(15, settings.tin_delay) + 10
def match_slot(s, level, settings):
corrected = s.tick + settings.correction
for i, t in enumerate(level.main):
if abs(corrected - t) <= 4:
return ('main', i, t, s.main[i] == 0)
for i, t in enumerate(level.decor):
if abs(corrected - t) <= 3:
return ('decor', i, t, not s.decor[i])
return ('outside', -1, -1, False)
def future(s):
groups = {}
for r in s.reservations:
if r.target >= s.tick:
groups.setdefault(r.target, []).append(r)
out = []
for target, rs in sorted(groups.items()):
order = sorted(rs, key=lambda r: (r.reserved_at, r.roof, r.id))
out.extend(((r, r == order[0]) for r in order))
return tuple(out)
def reservation_preview(s, level, settings, roof):
corrected = s.tick + settings.correction
slots = [t for t, v in zip(level.main, s.main) if v == 0 and corrected <= t + 4]
slots += [t for t, v in zip(level.decor, s.decor) if not v and corrected <= t + 3]
if not slots:
return None
slot = min(slots)
d = delay(roof, settings)
target = slot + d if d else None
return {'slot': slot, 'target': target, 'overlap': bool(d and any((r.target == target for r in s.reservations)))}
def step(s, level, settings, roof=None):
"""Process raw tick t, manual input, due echo, misses; then advance to t+1."""
if s.done:
return s
if roof is not None and roof not in (0, 1, 2):
raise ValueError('roof must be0/1/2')
t = s.tick
main = list(s.main)
decor = list(s.decor)
reservations = list(s.reservations)
events = list(s.events)
inputs = list(s.inputs)
if roof is not None and t <= close_tick(level, settings):
kind, i, slot, accepted = match_slot(s, level, settings)
inputs.append(Input(t, roof, t + settings.correction, kind, slot, accepted))
if accepted:
if kind == 'main':
main[i] = 1
else:
decor[i] = True
events.append(Event(t, 'manual', roof, slot))
d = delay(roof, settings)
if d:
assert slot + d > t, 'A legal reply must be future, never retroactive.'
reservations.append(Reservation(len(reservations), roof, slot + d, t, slot))
else:
events.append(Event(t, 'wrong', roof, slot))
elif roof is not None:
events.append(Event(t, 'closed', roof))
if t < 0 and t % 30 == 0:
events.append(Event(t, 'count'))
due = sorted((r for r in reservations if r.target == t), key=lambda r: (r.reserved_at, r.roof, r.id))
if due:
for r in due:
events.append(Event(t, 'echo', r.roof, r.source_slot, r.id))
for i, slot in enumerate(level.main):
if main[i] == 0 and t + settings.correction > slot + 4:
main[i] = -1
events.append(Event(t, 'miss', -1, slot))
return State(t + 1, tuple(main), tuple(decor), tuple(reservations), tuple(events), tuple(inputs), t >= ending_tick(level, settings))
def stats(s, level):
hit = sum((x == 1 for x in s.main))
dec = sum(s.decor)
echoes = [e for e in s.events if e.kind == 'echo']
manual = [e for e in s.events if e.kind == 'manual']
echo_targets = {e.tick for e in echoes}
overlaps = sum((1 for e in manual if e.slot in level.decor and e.slot in echo_targets))
used = {e.slot for e in manual} | echo_targets
return {'hits': hit, 'total': len(level.main), 'hit_percent': 100 * hit // len(level.main), 'complete': hit * 4 >= len(level.main) * 3, 'decorations': dec, 'automatic_replies': len(echoes), 'overlaid_slots': overlaps, 'blank_decorations': sum((t not in used for t in level.decor)), 'reservation_collisions': sum((e.kind == 'suppressed' for e in s.events)), 'wrong': sum((e.kind == 'wrong' for e in s.events)), 'manual_hits': len(manual)}
def normalized(v):
return json.dumps(v, ensure_ascii=False, sort_keys=True, separators=(',', ':'))
def digest(v):
return hashlib.sha256(normalized(v).encode('utf-8')).hexdigest()
STAGE = 3
# Display uses ASCII so this one file needs no glyph or image assets.
# The original sound definitions are retained, without claiming audible testing.
SOUNDS = (('c2c1', 't', '64', 'ff', 4), ('f4c4', 'p', '54', 'ff', 3),
('c3c2', 'n', '43', 'ff', 4), ('c4', 't', '3', 'f', 3))
COLORS = (9, 12, 6)
class Controller:
"""Pyxel-independent input edges and score clock. tick is the NEXT raw tick."""
def __init__(self):
self.level_index = 0
self.settings = Settings()
self.frame = 0
self.previous = set()
self.pending = []
self.audio_log = []
self.reset(stop=False)
@property
def level(self):
return LEVELS[self.level_index]
@property
def state(self):
return self.session.state if STAGE == 5 else self._state
def reset(self, stop=True):
if stop:
self.stop_audio()
if STAGE == 5:
self.session = new_session(self.level, self.settings)
else:
self._state = initial(self.level)
self.running = False
self.title = True
self.paused = False
self.gate = False
self.countdown = 0
self.accum = 0
self.flashes = [-1000000] * 3
self.caption = 'PRESS ENTER TO START. SPACE = ONE SCORE TICK.'
def emit(self, kind, tick, channel=None, sound=None):
item = {'kind': kind, 'tick': tick, 'frame': self.frame,
'channel': channel, 'sound': sound}
self.pending.append(item)
self.audio_log.append(item.copy())
def stop_audio(self):
self.emit('stop', self.state.tick)
self.flashes = [-1000000] * 3
def apply_tick(self, roof=None):
if self.state.done:
return
before = len(self.state.events)
if STAGE == 5:
self.session = session_step(self.session, self.level, roof)
else:
self._state = step(self.state, self.level, self.settings, roof)
for event in self.state.events[before:]:
if event.kind == 'manual':
self.emit('manual', event.tick, event.roof, event.roof)
self.flashes[event.roof] = self.frame
self.caption = 'MANUAL raw %d / score %d' % (event.tick, event.slot)
elif event.kind == 'echo':
self.emit('echo', event.tick, 3, event.roof)
self.flashes[event.roof] = self.frame
self.caption = 'ECHO target %d / source %d' % (event.tick, event.slot)
elif event.kind == 'count':
self.emit('count', event.tick, 3, 3)
elif event.kind == 'suppressed':
self.caption = 'SAME TARGET: EARLIER RESERVATION WINS'
elif event.kind == 'wrong':
self.caption = 'OUTSIDE WINDOW OR SLOT ALREADY USED'
if self.state.done:
self.running = False
self.gate = True
def rewind(self, previous=False):
if STAGE != 5:
return
self.stop_audio()
self.session = restart_bar(self.session, self.level, previous)
self.countdown = BAR
self.accum = 0
self.running = True
self.paused = False
self.gate = True
self.caption = 'RESTART: SCORE FROZEN DURING 120-TICK COUNT'
def control(self, down):
"""One call per renderer frame; fresh is set difference, never key repeat."""
self.frame += 1
down = set(down)
fresh = down - self.previous
self.previous = down
if self.gate:
if not down:
self.gate = False
return
if 'R' in fresh:
self.reset()
self.gate = True
return
if self.title:
changed = False
if STAGE >= 2:
correction = self.settings.correction
if 'LEFT' in fresh:
correction = max(-6, correction - 1)
elif 'RIGHT' in fresh:
correction = min(6, correction + 1)
changed = correction != self.settings.correction
self.settings = replace(self.settings, correction=correction)
if STAGE >= 4:
for i, key in enumerate(('1', '2', '3')):
if key in fresh:
self.level_index = i
changed = True
break
if 'B' in fresh:
self.settings = replace(self.settings, tin_delay=15 if self.settings.tin_delay == 45 else 45)
changed = True
if 'V' in fresh:
self.settings = replace(self.settings, speed=75 if self.settings.speed == 100 else 100)
changed = True
if changed:
self.reset()
self.gate = True
return
if 'RETURN' in fresh:
self.title = False
self.gate = True
self.caption = 'SPACE: STEP / ENTER: AUTO / A S D: ONE FRESH HIT'
return
if STAGE == 5 and 'P' in fresh:
if self.paused:
self.rewind(False)
else:
self.paused = True
self.running = False
self.stop_audio()
self.gate = True
return
if self.paused:
return
if STAGE == 5 and 'X' in fresh:
self.rewind(True)
return
if self.state.done:
return
if 'RETURN' in fresh:
self.running = not self.running
if not self.running:
self.stop_audio()
self.gate = True
return
# A chord is one hit; the canonical order is A, then S, then D.
roof = next((i for i, key in enumerate(('A', 'S', 'D'))
if key in fresh and (STAGE >= 3 or i == 0)), None)
if not self.running:
if 'SPACE' in fresh or roof is not None:
self.apply_tick(roof)
return
# 30 renderer frames/sec; 2 logical ticks/frame (or 1,2,1,2 at 75%).
self.accum += 4 if self.settings.speed == 100 else 3
dt, self.accum = divmod(self.accum, 2)
if self.countdown:
if self.countdown % 30 < dt:
self.emit('restart_count', self.state.tick, 3, 3)
self.countdown = max(0, self.countdown - dt)
if not self.countdown:
self.gate = True
return
for i in range(dt):
self.apply_tick(roof if i == 0 else None)
def flush_audio(self, pyxel):
"""Drain only the new effects; drawing or draining twice never replays them."""
pending, self.pending = self.pending, []
for event in pending:
if event['kind'] == 'stop':
pyxel.stop()
else:
pyxel.play(event['channel'], event['sound'])
return pending
def demo(self):
self.title = False
inputs = {0: 0, 30: 0, 60: 0} if STAGE < 3 else {0: 2, 30: 1, 60: 0}
if STAGE == 2:
inputs[15] = 0
while self.state.tick <= 75:
self.frame += 1
self.apply_tick(inputs.get(self.state.tick))
self.caption = ('CHECKPOINT: 3 MAIN HITS' if STAGE < 3 else
'0+45 = 30+15: SAME TARGET 45')
def draw(self, p):
s = self.state
stat = stats(s, self.level)
p.cls(1)
p.text(14, 10, 'RAIN ROOF DRUMS / STAGE %d / %s' % (STAGE, self.level.id), 10)
p.text(14, 27, 'NEXT RAW %d / CORRECTED %d / SPEED %d%%' %
(s.tick, s.tick + self.settings.correction, self.settings.speed), 7)
mode = 'TITLE' if self.title else 'PAUSED' if self.paused else 'RESULT' if s.done else 'AUTO' if self.running else 'STEP'
p.text(14, 44, mode + ' / 60 SCORE TICKS = 1 SEC AT 100% / 120 = ONE BAR', 6)
p.text(14, 63, 'MAIN %d/%d DECOR %d ECHO %d SUPPRESSED %d' %
(stat['hits'], stat['total'], stat['decorations'], stat['automatic_replies'], stat['reservation_collisions']), 11)
bar = max(0, min(3, max(0, s.tick) // BAR))
for row in range(2):
start = (bar + row) * BAR
y = 88 + row * 80
p.rect(12, y, 616, 71, 0)
label = 'BAR %d' % (bar + row + 1) if bar + row < 4 else 'TAIL'
p.text(18, y + 4, '%s / SCORE %d..%d' % (label, start, start + 119), 6)
for t in range(start, start + 120, 15):
x = 85 + (t - start) * 4
p.line(x, y + 20, x, y + 63, 5)
p.text(x - 4, y + 11, str(t), 5)
if t in self.level.main:
value = s.main[self.level.main.index(t)]
p.rectb(x - 4, y + 26, 9, 9, 11 if value == 1 else 8 if value == -1 else 7)
elif t in self.level.rests:
p.text(x - 5, y + 28, '-', 13)
if STAGE >= 2 and t in self.level.decor:
p.circb(x, y + 44, 3, 10 if s.decor[self.level.decor.index(t)] else 5)
if STAGE >= 3:
for reservation, active in future(s):
if start <= reservation.target < start + 120:
x = 85 + (reservation.target - start) * 4
color = COLORS[reservation.roof] if active or STAGE == 3 else 5
if reservation.roof == 1:
p.circb(x, y + 54, 5, color)
else:
p.tri(x, y + 49, x - 4, y + 57, x + 4, y + 57, color)
if start <= s.tick < start + 120:
x = 85 + (s.tick - start) * 4
p.line(x, y + 21, x, y + 64, 10)
for i, name in enumerate(('A WOOD', 'S TILE +15', 'D TIN +%d' % self.settings.tin_delay)):
x = 18 + i * 207
enabled = i == 0 or STAGE >= 3
# Age is zero on the event frame; ages 0..4 light five frames.
lit = 0 <= self.frame - self.flashes[i] < 5
p.tri(x, 286, x + 92, 250, x + 183, 286, COLORS[i] if lit else 5)
p.rectb(x + 12, 287, 158, 28, COLORS[i] if enabled else 5)
p.text(x + 22, 298, name if enabled else 'ADDED IN STAGE 03', 7 if enabled else 5)
p.text(14, 326, self.caption, 10)
p.text(14, 344, 'LAST EVENTS: raw tick / kind / roof / source slot', 6)
for i, e in enumerate(s.events[-6:]):
p.text(22, 358 + i * 12, '%4d %-10s roof %2d score %4d' % (e.tick, e.kind, e.roof, e.slot), 7)
if STAGE >= 3:
p.text(365, 344, 'NEXT REPLIES', 6)
pending = future(s)[:4]
for i, (reservation, active) in enumerate(pending):
p.text(365, 360 + i * 12, '%d %s %s' % (reservation.target, ('WOOD', 'TILE', 'TIN')[reservation.roof], 'ACTIVE' if active or STAGE == 3 else 'MUTED'), COLORS[reservation.roof] if active or STAGE == 3 else 5)
if not pending:
p.text(365, 360, '(none)', 5)
if self.countdown:
p.text(365, 415, 'RESTART COUNT %d' % self.countdown, 10)
if s.done:
p.text(365, 415, 'COMPLETE' if stat['complete'] else 'TRY AGAIN', 11)
p.text(14, 438, 'ENTER: START / AUTO-STEP TOGGLE SPACE: ONE TICK R: RESET', 7)
p.text(14, 453, ('A: WOOD' if STAGE < 3 else 'A/S/D: ROOFS') + ' ESC: QUIT HOLD = NO REPEAT', 7)
p.text(14, 468, 'TITLE: LEFT/RIGHT CORRECTION' if STAGE >= 2 else 'STAGE 01: MAIN BEATS ONLY / NO REPLIES', 6)
if STAGE >= 4:
p.text(300, 468, '1/2/3 CHART B DELAY V SPEED', 6)
p.text(14, 484, 'P: PAUSE / RESTART CURRENT BAR X: PREVIOUS BAR' if STAGE == 5 else 'SAVE EDIT -> QUIT -> RUN THIS SAME FILE AGAIN', 13)
if self.title:
p.rect(70, 106, 500, 100, 1)
p.rectb(70, 106, 500, 100, 12)
p.text(85, 120, 'PRESS ENTER. THEN SPACE STEPS OR ENTER RUNS.', 10)
p.text(85, 140, 'MAIN: +/-4 TICKS. DECOR FROM STAGE 02: +/-3.', 7)
p.text(85, 160, 'NEXT RAW = TICK THAT THE NEXT UPDATE WILL PROCESS.', 6)
p.text(85, 180, 'R RESETS STATE. IT DOES NOT RELOAD SAVED CODE.', 13)
def main(headless=False, screenshot=None, report=None, checkpoint=False):
# CLI validates and resolves paths before reaching this import.
import pyxel as p
p.init(640, 504, title='Rain roof lesson %d' % STAGE, fps=30,
display_scale=1, headless=headless)
for i, sound in enumerate(SOUNDS):
p.sounds[i].set(*sound)
app = Controller()
actual_calls = []
if checkpoint:
app.title = False
keys = {k: getattr(p, 'KEY_' + k) for k in
('RETURN', 'SPACE', 'R', 'A', 'S', 'D', 'P', 'X', 'LEFT', 'RIGHT', '1', '2', '3', 'B', 'V')}
def update():
if not headless:
app.control({k for k, value in keys.items() if p.btn(value)})
elif checkpoint and app.state.tick <= 75:
app.frame += 1
inputs = {0: 0, 30: 0, 60: 0} if STAGE < 3 else {0: 2, 30: 1, 60: 0}
if STAGE == 2:
inputs[15] = 0
for _ in range(2):
if app.state.tick <= 75:
app.apply_tick(inputs.get(app.state.tick))
actual_calls.extend(app.flush_audio(p))
if p.btnp(p.KEY_ESCAPE) or headless and p.frame_count >= 3 and (not checkpoint or app.state.tick > 75):
app.stop_audio()
actual_calls.extend(app.flush_audio(p))
app.draw(p)
if screenshot:
p.screen.save(screenshot, 1)
if report:
from pathlib import Path
payload = {'stage': STAGE, 'state': asdict(app.state),
'stats': stats(app.state, app.level), 'audio_calls': actual_calls,
'scope': 'Actual Pyxel API calls; dummy driver is not audible verification',
'screenshot': screenshot}
Path(report).write_text(json.dumps(payload, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
p.quit()
p.run(update, lambda: app.draw(p))
if __name__ == '__main__':
import argparse
from pathlib import Path
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--headless-smoke', action='store_true')
parser.add_argument('--demo-checkpoint', action='store_true')
parser.add_argument('--screenshot')
parser.add_argument('--report')
args = parser.parse_args()
if (args.demo_checkpoint or args.screenshot or args.report) and not args.headless_smoke:
parser.error('checkpoint and output files require --headless-smoke')
for name, suffix in (('screenshot', '.png'), ('report', '.json')):
value = getattr(args, name)
if value:
path = Path(value).resolve()
if path.suffix.lower() != suffix or not path.parent.is_dir() or path.exists():
parser.error('use a NEW ' + suffix + ' file in an existing directory')
if path.parent == Path(__file__).resolve().parent:
parser.error('output must be outside the script directory')
setattr(args, name, str(path))
main(args.headless_smoke, args.screenshot, args.report, args.demo_checkpoint)同じ刻の返事を先約一つへ整理する
同じ順位のときの安定した処理・予告と本番の一致・限られた数の比較。
同じ target の自動の返事は、元のゲームどおり1つだけにします。その刻の予約を reserved_at、roof、id の順に並べ、最初の予約を返事（echo）、残りを打ち消し（suppressed）にします。最後の予約で上書きしたり、set のたまたまの順番を使ったりはしません。ふつうの入力では1刻に1回しか叩かないので、reserved_at が先かどうかだけで決まります。残りの2つの条件は、データが同じ順位になっても結果がぶれないようにするためのルールです。
0刻の D と30刻の S は、どちらも target 45 です。0刻に予約した D が返事になり、S は打ち消されます。あとから叩いた S の成功を取り消すわけではありません。0/30 の主拍はどちらも成功のままで、自動の返事だけが1つになります。future も同じ並べ方で勝つ予約を示し、返事の丸と三角の色を、有効の色と灰色に分けます。予告を読んでも、予約や点数は変わりません。
表紙の 1/2/3 で、元のゲームの3つの譜面 A/B/C を選べます。A は4つの主拍がまっすぐ並び、B は各小節の60刻が休み、C は30刻のない小節と60刻のない小節が交互に来ます。休みの「-」は主拍ではないので、そこを叩かなくても失敗にはなりません。B キーはトタンの返事だけを 45↔15 に、V キーは速さを 100↔75% に切り替えます。B キーと、譜面 B を選ぶ2キーを、混同しないようにしましょう。
75%の速さでは、画面は同じ30Hzのまま、足し算の仕組み（加算器）がルール上の時計を1刻・2刻と交互に進めます。ルール上の時計の基準は60刻単位のままで、実際の時間では1刻が1/45秒です。押した屋根は、その画面の更新の最初の1刻だけに渡します。4小節のあとも、最後の返事が届く余韻を処理し、ending_tick を処理したあとで done になります。最後の画面の tick は、その1つ先です。元のゲームとの比較として、54本の決まった入力の列を収録していますが、すべての入力の列を調べたという意味ではありません。
下のコードは 04_collision.py の一部です。前のステップから追加・変更したところだけを抜き出しています。
04_collision.py / 注目する部分
due = sorted((r for r in reservations if r.target == t), key=lambda r: (r.reserved_at, r.roof, r.id))
if due:
events.append(Event(t, 'echo', due[0].roof, due[0].source_slot, due[0].id))
for r in due[1:]:
events.append(Event(t, 'suppressed', r.roof, r.source_slot, r.id))実行して確かめる
教材のZIPを展開したフォルダーで、次のコマンドを実行します。
Windows
.venv\Scripts\python.exe 04_collision.pymacOS / Linux
.venv/bin/python 04_collision.py画面で見ること最初の設定の自動確認は、NEXT RAW 76・MAIN 3/16・ECHO 1・SUPPRESSED 1 です。45刻に、D の返事と S の打ち消しが並びます。表紙で B を1回押した比較では、予約が15と45になり、ECHO 2・SUPPRESSED 0 です。
- 予約の順番をはっきり決めて、1つの音にする
- 予告と本番で、同じ優先順位を使う
- 速さと、判定のずれの補正を混ぜない
つまずきやすいところsuppressed が消すのは、重なった自動の返事だけです。手で叩いた音、入力の成功、主拍の数まで消さないでください。75%の速さを、画面の表示を間引くだけで済ませず、ルール上の時計を進める割合を変えます。
少し変えてみよう
表紙で B を押して離し、トタンを +15 にして、同じ入力の順番を試してみましょう。0:D の返事は15、30:S の返事は45 に分かれます。コードを変える実験なら、03 と同じく、最初の設定を Settings(tin_delay=15) にして、保存・終了・実行し直してください。テストでは、実際の B キーの操作で ECHO 2/SUPPRESSED 0 を確かめています。
04_collision.pyをダウンロードこの段階だけで実行できます
このステップのコード全文（541行）
04_collision.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。"""
"""066 Deterministic 60-tick musical clock. No rendering or audio dependencies."""
from dataclasses import dataclass, asdict, replace
import hashlib, json
BAR = 120
ROOFS = ('木', '瓦', 'トタン')
@dataclass(frozen=True)
class Level:
id: str
title: str
main: tuple[int, ...]
decor: tuple[int, ...]
rests: tuple[int, ...]
bars: int = 4
def build(id, title, patterns):
main = tuple((b * BAR + t for b, pat in enumerate(patterns) for t in pat))
decor = tuple((b * BAR + t for b in range(4) for t in (15, 45, 75, 105)))
rests = tuple((b * BAR + t for b, pat in enumerate(patterns) for t in (0, 30, 60, 90) if t not in pat))
return Level(id, title, main, decor, rests)
LEVELS = (build('066-A-v1', 'まっすぐの雨', ((0, 30, 60, 90),) * 4), build('066-B-v1', '空いた三拍目', ((0, 30, 90),) * 4), build('066-C-v1', '返事を待つ軒', ((0, 60, 90), (0, 30, 90), (0, 60, 90), (0, 30, 90))))
@dataclass(frozen=True)
class Settings:
correction: int = 0
tin_delay: int = 45
speed: int = 100
control: str = 'asd'
def __post_init__(self):
if not -6 <= self.correction <= 6:
raise ValueError('correction -6..6')
if self.tin_delay not in (15, 45):
raise ValueError('tin delay 15 or45')
if self.speed not in (75, 100):
raise ValueError('speed 75 or100')
if self.control not in ('asd', 'arrows'):
raise ValueError('unknown input mode')
@dataclass(frozen=True)
class Reservation:
id: int
roof: int
target: int
reserved_at: int
source_slot: int
@dataclass(frozen=True)
class Event:
tick: int
kind: str
roof: int = -1
slot: int = -1
reservation: int = -1
@dataclass(frozen=True)
class Input:
tick: int
roof: int
corrected: int
kind: str
slot: int
accepted: bool
@dataclass(frozen=True)
class State:
tick: int
main: tuple[int, ...]
decor: tuple[bool, ...]
reservations: tuple[Reservation, ...] = ()
events: tuple[Event, ...] = ()
inputs: tuple[Input, ...] = ()
done: bool = False
def initial(level):
return State(-BAR, tuple((0 for _ in level.main)), tuple((False for _ in level.decor)))
def delay(roof, settings):
return (0, 15, settings.tin_delay)[roof]
def close_tick(level, settings):
return max(max(level.main) + 4, max(level.decor) + 3) - settings.correction
def ending_tick(level, settings):
return max(max(level.main), max(level.decor)) + max(15, settings.tin_delay) + 10
def match_slot(s, level, settings):
corrected = s.tick + settings.correction
for i, t in enumerate(level.main):
if abs(corrected - t) <= 4:
return ('main', i, t, s.main[i] == 0)
for i, t in enumerate(level.decor):
if abs(corrected - t) <= 3:
return ('decor', i, t, not s.decor[i])
return ('outside', -1, -1, False)
def future(s):
groups = {}
for r in s.reservations:
if r.target >= s.tick:
groups.setdefault(r.target, []).append(r)
out = []
for target, rs in sorted(groups.items()):
order = sorted(rs, key=lambda r: (r.reserved_at, r.roof, r.id))
out.extend(((r, r == order[0]) for r in order))
return tuple(out)
def reservation_preview(s, level, settings, roof):
corrected = s.tick + settings.correction
slots = [t for t, v in zip(level.main, s.main) if v == 0 and corrected <= t + 4]
slots += [t for t, v in zip(level.decor, s.decor) if not v and corrected <= t + 3]
if not slots:
return None
slot = min(slots)
d = delay(roof, settings)
target = slot + d if d else None
return {'slot': slot, 'target': target, 'overlap': bool(d and any((r.target == target for r in s.reservations)))}
def step(s, level, settings, roof=None):
"""Process raw tick t, manual input, due echo, misses; then advance to t+1."""
if s.done:
return s
if roof is not None and roof not in (0, 1, 2):
raise ValueError('roof must be0/1/2')
t = s.tick
main = list(s.main)
decor = list(s.decor)
reservations = list(s.reservations)
events = list(s.events)
inputs = list(s.inputs)
if roof is not None and t <= close_tick(level, settings):
kind, i, slot, accepted = match_slot(s, level, settings)
inputs.append(Input(t, roof, t + settings.correction, kind, slot, accepted))
if accepted:
if kind == 'main':
main[i] = 1
else:
decor[i] = True
events.append(Event(t, 'manual', roof, slot))
d = delay(roof, settings)
if d:
assert slot + d > t, 'A legal reply must be future, never retroactive.'
reservations.append(Reservation(len(reservations), roof, slot + d, t, slot))
else:
events.append(Event(t, 'wrong', roof, slot))
elif roof is not None:
events.append(Event(t, 'closed', roof))
if t < 0 and t % 30 == 0:
events.append(Event(t, 'count'))
due = sorted((r for r in reservations if r.target == t), key=lambda r: (r.reserved_at, r.roof, r.id))
if due:
events.append(Event(t, 'echo', due[0].roof, due[0].source_slot, due[0].id))
for r in due[1:]:
events.append(Event(t, 'suppressed', r.roof, r.source_slot, r.id))
for i, slot in enumerate(level.main):
if main[i] == 0 and t + settings.correction > slot + 4:
main[i] = -1
events.append(Event(t, 'miss', -1, slot))
return State(t + 1, tuple(main), tuple(decor), tuple(reservations), tuple(events), tuple(inputs), t >= ending_tick(level, settings))
def stats(s, level):
hit = sum((x == 1 for x in s.main))
dec = sum(s.decor)
echoes = [e for e in s.events if e.kind == 'echo']
manual = [e for e in s.events if e.kind == 'manual']
echo_targets = {e.tick for e in echoes}
overlaps = sum((1 for e in manual if e.slot in level.decor and e.slot in echo_targets))
used = {e.slot for e in manual} | echo_targets
return {'hits': hit, 'total': len(level.main), 'hit_percent': 100 * hit // len(level.main), 'complete': hit * 4 >= len(level.main) * 3, 'decorations': dec, 'automatic_replies': len(echoes), 'overlaid_slots': overlaps, 'blank_decorations': sum((t not in used for t in level.decor)), 'reservation_collisions': sum((e.kind == 'suppressed' for e in s.events)), 'wrong': sum((e.kind == 'wrong' for e in s.events)), 'manual_hits': len(manual)}
def normalized(v):
return json.dumps(v, ensure_ascii=False, sort_keys=True, separators=(',', ':'))
def digest(v):
return hashlib.sha256(normalized(v).encode('utf-8')).hexdigest()
STAGE = 4
# Display uses ASCII so this one file needs no glyph or image assets.
# The original sound definitions are retained, without claiming audible testing.
SOUNDS = (('c2c1', 't', '64', 'ff', 4), ('f4c4', 'p', '54', 'ff', 3),
('c3c2', 'n', '43', 'ff', 4), ('c4', 't', '3', 'f', 3))
COLORS = (9, 12, 6)
class Controller:
"""Pyxel-independent input edges and score clock. tick is the NEXT raw tick."""
def __init__(self):
self.level_index = 0
self.settings = Settings()
self.frame = 0
self.previous = set()
self.pending = []
self.audio_log = []
self.reset(stop=False)
@property
def level(self):
return LEVELS[self.level_index]
@property
def state(self):
return self.session.state if STAGE == 5 else self._state
def reset(self, stop=True):
if stop:
self.stop_audio()
if STAGE == 5:
self.session = new_session(self.level, self.settings)
else:
self._state = initial(self.level)
self.running = False
self.title = True
self.paused = False
self.gate = False
self.countdown = 0
self.accum = 0
self.flashes = [-1000000] * 3
self.caption = 'PRESS ENTER TO START. SPACE = ONE SCORE TICK.'
def emit(self, kind, tick, channel=None, sound=None):
item = {'kind': kind, 'tick': tick, 'frame': self.frame,
'channel': channel, 'sound': sound}
self.pending.append(item)
self.audio_log.append(item.copy())
def stop_audio(self):
self.emit('stop', self.state.tick)
self.flashes = [-1000000] * 3
def apply_tick(self, roof=None):
if self.state.done:
return
before = len(self.state.events)
if STAGE == 5:
self.session = session_step(self.session, self.level, roof)
else:
self._state = step(self.state, self.level, self.settings, roof)
for event in self.state.events[before:]:
if event.kind == 'manual':
self.emit('manual', event.tick, event.roof, event.roof)
self.flashes[event.roof] = self.frame
self.caption = 'MANUAL raw %d / score %d' % (event.tick, event.slot)
elif event.kind == 'echo':
self.emit('echo', event.tick, 3, event.roof)
self.flashes[event.roof] = self.frame
self.caption = 'ECHO target %d / source %d' % (event.tick, event.slot)
elif event.kind == 'count':
self.emit('count', event.tick, 3, 3)
elif event.kind == 'suppressed':
self.caption = 'SAME TARGET: EARLIER RESERVATION WINS'
elif event.kind == 'wrong':
self.caption = 'OUTSIDE WINDOW OR SLOT ALREADY USED'
if self.state.done:
self.running = False
self.gate = True
def rewind(self, previous=False):
if STAGE != 5:
return
self.stop_audio()
self.session = restart_bar(self.session, self.level, previous)
self.countdown = BAR
self.accum = 0
self.running = True
self.paused = False
self.gate = True
self.caption = 'RESTART: SCORE FROZEN DURING 120-TICK COUNT'
def control(self, down):
"""One call per renderer frame; fresh is set difference, never key repeat."""
self.frame += 1
down = set(down)
fresh = down - self.previous
self.previous = down
if self.gate:
if not down:
self.gate = False
return
if 'R' in fresh:
self.reset()
self.gate = True
return
if self.title:
changed = False
if STAGE >= 2:
correction = self.settings.correction
if 'LEFT' in fresh:
correction = max(-6, correction - 1)
elif 'RIGHT' in fresh:
correction = min(6, correction + 1)
changed = correction != self.settings.correction
self.settings = replace(self.settings, correction=correction)
if STAGE >= 4:
for i, key in enumerate(('1', '2', '3')):
if key in fresh:
self.level_index = i
changed = True
break
if 'B' in fresh:
self.settings = replace(self.settings, tin_delay=15 if self.settings.tin_delay == 45 else 45)
changed = True
if 'V' in fresh:
self.settings = replace(self.settings, speed=75 if self.settings.speed == 100 else 100)
changed = True
if changed:
self.reset()
self.gate = True
return
if 'RETURN' in fresh:
self.title = False
self.gate = True
self.caption = 'SPACE: STEP / ENTER: AUTO / A S D: ONE FRESH HIT'
return
if STAGE == 5 and 'P' in fresh:
if self.paused:
self.rewind(False)
else:
self.paused = True
self.running = False
self.stop_audio()
self.gate = True
return
if self.paused:
return
if STAGE == 5 and 'X' in fresh:
self.rewind(True)
return
if self.state.done:
return
if 'RETURN' in fresh:
self.running = not self.running
if not self.running:
self.stop_audio()
self.gate = True
return
# A chord is one hit; the canonical order is A, then S, then D.
roof = next((i for i, key in enumerate(('A', 'S', 'D'))
if key in fresh and (STAGE >= 3 or i == 0)), None)
if not self.running:
if 'SPACE' in fresh or roof is not None:
self.apply_tick(roof)
return
# 30 renderer frames/sec; 2 logical ticks/frame (or 1,2,1,2 at 75%).
self.accum += 4 if self.settings.speed == 100 else 3
dt, self.accum = divmod(self.accum, 2)
if self.countdown:
if self.countdown % 30 < dt:
self.emit('restart_count', self.state.tick, 3, 3)
self.countdown = max(0, self.countdown - dt)
if not self.countdown:
self.gate = True
return
for i in range(dt):
self.apply_tick(roof if i == 0 else None)
def flush_audio(self, pyxel):
"""Drain only the new effects; drawing or draining twice never replays them."""
pending, self.pending = self.pending, []
for event in pending:
if event['kind'] == 'stop':
pyxel.stop()
else:
pyxel.play(event['channel'], event['sound'])
return pending
def demo(self):
self.title = False
inputs = {0: 0, 30: 0, 60: 0} if STAGE < 3 else {0: 2, 30: 1, 60: 0}
if STAGE == 2:
inputs[15] = 0
while self.state.tick <= 75:
self.frame += 1
self.apply_tick(inputs.get(self.state.tick))
self.caption = ('CHECKPOINT: 3 MAIN HITS' if STAGE < 3 else
'0+45 = 30+15: SAME TARGET 45')
def draw(self, p):
s = self.state
stat = stats(s, self.level)
p.cls(1)
p.text(14, 10, 'RAIN ROOF DRUMS / STAGE %d / %s' % (STAGE, self.level.id), 10)
p.text(14, 27, 'NEXT RAW %d / CORRECTED %d / SPEED %d%%' %
(s.tick, s.tick + self.settings.correction, self.settings.speed), 7)
mode = 'TITLE' if self.title else 'PAUSED' if self.paused else 'RESULT' if s.done else 'AUTO' if self.running else 'STEP'
p.text(14, 44, mode + ' / 60 SCORE TICKS = 1 SEC AT 100% / 120 = ONE BAR', 6)
p.text(14, 63, 'MAIN %d/%d DECOR %d ECHO %d SUPPRESSED %d' %
(stat['hits'], stat['total'], stat['decorations'], stat['automatic_replies'], stat['reservation_collisions']), 11)
bar = max(0, min(3, max(0, s.tick) // BAR))
for row in range(2):
start = (bar + row) * BAR
y = 88 + row * 80
p.rect(12, y, 616, 71, 0)
label = 'BAR %d' % (bar + row + 1) if bar + row < 4 else 'TAIL'
p.text(18, y + 4, '%s / SCORE %d..%d' % (label, start, start + 119), 6)
for t in range(start, start + 120, 15):
x = 85 + (t - start) * 4
p.line(x, y + 20, x, y + 63, 5)
p.text(x - 4, y + 11, str(t), 5)
if t in self.level.main:
value = s.main[self.level.main.index(t)]
p.rectb(x - 4, y + 26, 9, 9, 11 if value == 1 else 8 if value == -1 else 7)
elif t in self.level.rests:
p.text(x - 5, y + 28, '-', 13)
if STAGE >= 2 and t in self.level.decor:
p.circb(x, y + 44, 3, 10 if s.decor[self.level.decor.index(t)] else 5)
if STAGE >= 3:
for reservation, active in future(s):
if start <= reservation.target < start + 120:
x = 85 + (reservation.target - start) * 4
color = COLORS[reservation.roof] if active or STAGE == 3 else 5
if reservation.roof == 1:
p.circb(x, y + 54, 5, color)
else:
p.tri(x, y + 49, x - 4, y + 57, x + 4, y + 57, color)
if start <= s.tick < start + 120:
x = 85 + (s.tick - start) * 4
p.line(x, y + 21, x, y + 64, 10)
for i, name in enumerate(('A WOOD', 'S TILE +15', 'D TIN +%d' % self.settings.tin_delay)):
x = 18 + i * 207
enabled = i == 0 or STAGE >= 3
# Age is zero on the event frame; ages 0..4 light five frames.
lit = 0 <= self.frame - self.flashes[i] < 5
p.tri(x, 286, x + 92, 250, x + 183, 286, COLORS[i] if lit else 5)
p.rectb(x + 12, 287, 158, 28, COLORS[i] if enabled else 5)
p.text(x + 22, 298, name if enabled else 'ADDED IN STAGE 03', 7 if enabled else 5)
p.text(14, 326, self.caption, 10)
p.text(14, 344, 'LAST EVENTS: raw tick / kind / roof / source slot', 6)
for i, e in enumerate(s.events[-6:]):
p.text(22, 358 + i * 12, '%4d %-10s roof %2d score %4d' % (e.tick, e.kind, e.roof, e.slot), 7)
if STAGE >= 3:
p.text(365, 344, 'NEXT REPLIES', 6)
pending = future(s)[:4]
for i, (reservation, active) in enumerate(pending):
p.text(365, 360 + i * 12, '%d %s %s' % (reservation.target, ('WOOD', 'TILE', 'TIN')[reservation.roof], 'ACTIVE' if active or STAGE == 3 else 'MUTED'), COLORS[reservation.roof] if active or STAGE == 3 else 5)
if not pending:
p.text(365, 360, '(none)', 5)
if self.countdown:
p.text(365, 415, 'RESTART COUNT %d' % self.countdown, 10)
if s.done:
p.text(365, 415, 'COMPLETE' if stat['complete'] else 'TRY AGAIN', 11)
p.text(14, 438, 'ENTER: START / AUTO-STEP TOGGLE SPACE: ONE TICK R: RESET', 7)
p.text(14, 453, ('A: WOOD' if STAGE < 3 else 'A/S/D: ROOFS') + ' ESC: QUIT HOLD = NO REPEAT', 7)
p.text(14, 468, 'TITLE: LEFT/RIGHT CORRECTION' if STAGE >= 2 else 'STAGE 01: MAIN BEATS ONLY / NO REPLIES', 6)
if STAGE >= 4:
p.text(300, 468, '1/2/3 CHART B DELAY V SPEED', 6)
p.text(14, 484, 'P: PAUSE / RESTART CURRENT BAR X: PREVIOUS BAR' if STAGE == 5 else 'SAVE EDIT -> QUIT -> RUN THIS SAME FILE AGAIN', 13)
if self.title:
p.rect(70, 106, 500, 100, 1)
p.rectb(70, 106, 500, 100, 12)
p.text(85, 120, 'PRESS ENTER. THEN SPACE STEPS OR ENTER RUNS.', 10)
p.text(85, 140, 'MAIN: +/-4 TICKS. DECOR FROM STAGE 02: +/-3.', 7)
p.text(85, 160, 'NEXT RAW = TICK THAT THE NEXT UPDATE WILL PROCESS.', 6)
p.text(85, 180, 'R RESETS STATE. IT DOES NOT RELOAD SAVED CODE.', 13)
def main(headless=False, screenshot=None, report=None, checkpoint=False):
# CLI validates and resolves paths before reaching this import.
import pyxel as p
p.init(640, 504, title='Rain roof lesson %d' % STAGE, fps=30,
display_scale=1, headless=headless)
for i, sound in enumerate(SOUNDS):
p.sounds[i].set(*sound)
app = Controller()
actual_calls = []
if checkpoint:
app.title = False
keys = {k: getattr(p, 'KEY_' + k) for k in
('RETURN', 'SPACE', 'R', 'A', 'S', 'D', 'P', 'X', 'LEFT', 'RIGHT', '1', '2', '3', 'B', 'V')}
def update():
if not headless:
app.control({k for k, value in keys.items() if p.btn(value)})
elif checkpoint and app.state.tick <= 75:
app.frame += 1
inputs = {0: 0, 30: 0, 60: 0} if STAGE < 3 else {0: 2, 30: 1, 60: 0}
if STAGE == 2:
inputs[15] = 0
for _ in range(2):
if app.state.tick <= 75:
app.apply_tick(inputs.get(app.state.tick))
actual_calls.extend(app.flush_audio(p))
if p.btnp(p.KEY_ESCAPE) or headless and p.frame_count >= 3 and (not checkpoint or app.state.tick > 75):
app.stop_audio()
actual_calls.extend(app.flush_audio(p))
app.draw(p)
if screenshot:
p.screen.save(screenshot, 1)
if report:
from pathlib import Path
payload = {'stage': STAGE, 'state': asdict(app.state),
'stats': stats(app.state, app.level), 'audio_calls': actual_calls,
'scope': 'Actual Pyxel API calls; dummy driver is not audible verification',
'screenshot': screenshot}
Path(report).write_text(json.dumps(payload, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
p.quit()
p.run(update, lambda: app.draw(p))
if __name__ == '__main__':
import argparse
from pathlib import Path
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--headless-smoke', action='store_true')
parser.add_argument('--demo-checkpoint', action='store_true')
parser.add_argument('--screenshot')
parser.add_argument('--report')
args = parser.parse_args()
if (args.demo_checkpoint or args.screenshot or args.report) and not args.headless_smoke:
parser.error('checkpoint and output files require --headless-smoke')
for name, suffix in (('screenshot', '.png'), ('report', '.json')):
value = getattr(args, name)
if value:
path = Path(value).resolve()
if path.suffix.lower() != suffix or not path.parent.is_dir() or path.exists():
parser.error('use a NEW ' + suffix + ' file in an existing directory')
if path.parent == Path(__file__).resolve().parent:
parser.error('output must be outside the script directory')
setattr(args, name, str(path))
main(args.headless_smoke, args.screenshot, args.report, args.demo_checkpoint)停止した小節を、持ち越しの返事ごとやり直す
小節ごとの状態の写しと再開の待ち、音を止めて状態を作り直す。
元のゲームの Session と、小節の頭からの再開を加えます。P で演奏を止めると、ルール上の時計は進まず、pyxel.stop で鳴っている音を止めます。キーを全部離してもう一度 P を押すと、今の小節の頭に戻り、120刻ぶんの再開のカウントを待ってから演奏します。これは、止めた刻からそのまま続ける操作ではありません。Enter での AUTO/STEP の切り替えは、教材の観察用に足した機能で、同じ位置にとどめるためのものです。X は1つ前の小節へ戻ります。R は表紙と -120刻に戻り、最初からやり直します。
session_step は、小節の最初の raw の刻を処理する前に、状態を保存します。ただし、補正がプラスのときは、次の小節の主拍を早めに取っていることがあります。bar_snapshot は、その小節より後の主拍・飾り・予約・対応する入力をいったん取り除き、前の小節から来る返事は残します。今の tick だけを120に戻す方法では、主拍が成功済みのままだったり、重なった返事が残ったりして、正しくやり直せません。
たとえば、90刻の D は135刻に返事をします。150刻から今の小節をやり直すと、120刻の主拍は取り直しになり、90→135 の予約は持ち越します。135刻になれば、その返事をもう一度1回だけ鳴らします。前に135刻で鳴った出来事は、やり直すときに切り落とされているので、履歴の個数を「前の再生位置」として使い回すことはしません。Controller は、新しい step で増えた出来事だけを、そのたびに取り出します。
補正 +6 で小節120からやり直すときは、raw 114 から準備し直します。カウントの間は、採点の tick を止め、画面の更新用の加算器だけで120を減らしていきます。100%の速さでは60回、75%では80回の画面の更新でカウントが終わります。ただし、キーが全部離されるのを待つ時間は、これとは別です。結果が出たあとの A/Space は無効で、R でもう一度挑戦します。P/X は再開の操作なので、結果の画面からも使えます。
最後の段階の、元のゲームのルールの関数とデータクラスは、元のゲームとプログラムの構造（AST）が一致することを検査しています。教材の画面は英数字（ASCII）、段階の表示、STEP の機能があり、元の日本語の表紙、左右＋Z の操作、試し打ち、記録の保存・比較の画面は移していません。元のゲームの音の定義はまったく同じで、完成版の assets や音のコードは変えていません。自動テストで、音を実際には鳴らさない仮の仕組み（dummy の音声ドライバー）で play/stop を実行したことと、人が音を聞いて確かめたことは別です。
下のコードは 05_restart.py の一部です。前のステップから追加・変更したところだけを抜き出しています。
05_restart.py / 注目する部分
rewind_tick = bar * BAR - max(0, session.settings.correction)
state = replace(state, tick=rewind_tick, events=tuple((e for e in state.events if e.tick < rewind_tick)), inputs=tuple((i for i in state.inputs if i.tick < rewind_tick)), reservations=tuple((r for r in state.reservations if r.reserved_at < rewind_tick)))実行して確かめる
教材のZIPを展開したフォルダーで、次のコマンドを実行します。
Windows
.venv\Scripts\python.exe 05_restart.pymacOS / Linux
.venv/bin/python 05_restart.py画面で見ること90刻に D、120刻に S を含む入力のあと、150刻で P → キーを全部離す → P と押します。NEXT RAW 120・再開のカウント120になり、120の主拍はまだ判定していない状態に戻り、90→135 の予約は残ります。カウントが終わったあとも、キーを全部離してから入力し直します。135刻で、D の返事だけが1回鳴ります。
- P は、止めた位置の小節の頭から再開する
- 早めに取った主拍は取り除き、前の小節からの返事は残す
- 再開のカウントと、採点の時計を別々に持つ
つまずきやすいところ
tick だけを巻き戻すと、過去の成功や予約が残ってしまいます。小節の区切りで保存した State 全体を使い、その小節の早取りを消し、前の小節からの返事は残します。R でコードが読み直されると思わないでください。少し変えてみよう
まず、P での今の小節からの再開と、X での前の小節からの再開を比べてみましょう。教材のテストの、90:D・120:S のあと150で再開する例では、135の返事は D だけです。次に、Controller.__init__ を Settings(correction=3) に変え、保存・終了・実行し直します。小節120の再開の位置が117になる理由を、raw+3=120 で説明してみてください。最初の補正0に戻すと120からです。
05_restart.pyをダウンロードこの段階だけで実行できます
このステップのコード全文（583行）
05_restart.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。"""
"""066 Deterministic 60-tick musical clock. No rendering or audio dependencies."""
from dataclasses import dataclass, asdict, replace
import hashlib, json
BAR = 120
ROOFS = ('木', '瓦', 'トタン')
@dataclass(frozen=True)
class Level:
id: str
title: str
main: tuple[int, ...]
decor: tuple[int, ...]
rests: tuple[int, ...]
bars: int = 4
def build(id, title, patterns):
main = tuple((b * BAR + t for b, pat in enumerate(patterns) for t in pat))
decor = tuple((b * BAR + t for b in range(4) for t in (15, 45, 75, 105)))
rests = tuple((b * BAR + t for b, pat in enumerate(patterns) for t in (0, 30, 60, 90) if t not in pat))
return Level(id, title, main, decor, rests)
LEVELS = (build('066-A-v1', 'まっすぐの雨', ((0, 30, 60, 90),) * 4), build('066-B-v1', '空いた三拍目', ((0, 30, 90),) * 4), build('066-C-v1', '返事を待つ軒', ((0, 60, 90), (0, 30, 90), (0, 60, 90), (0, 30, 90))))
@dataclass(frozen=True)
class Settings:
correction: int = 0
tin_delay: int = 45
speed: int = 100
control: str = 'asd'
def __post_init__(self):
if not -6 <= self.correction <= 6:
raise ValueError('correction -6..6')
if self.tin_delay not in (15, 45):
raise ValueError('tin delay 15 or45')
if self.speed not in (75, 100):
raise ValueError('speed 75 or100')
if self.control not in ('asd', 'arrows'):
raise ValueError('unknown input mode')
@dataclass(frozen=True)
class Reservation:
id: int
roof: int
target: int
reserved_at: int
source_slot: int
@dataclass(frozen=True)
class Event:
tick: int
kind: str
roof: int = -1
slot: int = -1
reservation: int = -1
@dataclass(frozen=True)
class Input:
tick: int
roof: int
corrected: int
kind: str
slot: int
accepted: bool
@dataclass(frozen=True)
class State:
tick: int
main: tuple[int, ...]
decor: tuple[bool, ...]
reservations: tuple[Reservation, ...] = ()
events: tuple[Event, ...] = ()
inputs: tuple[Input, ...] = ()
done: bool = False
@dataclass(frozen=True)
class Session:
state: State
settings: Settings
snapshots: tuple[tuple[int, State], ...] = ()
restarts: tuple[tuple[int, int, str], ...] = ()
def initial(level):
return State(-BAR, tuple((0 for _ in level.main)), tuple((False for _ in level.decor)))
def new_session(level, settings=Settings()):
return Session(initial(level), settings)
def delay(roof, settings):
return (0, 15, settings.tin_delay)[roof]
def close_tick(level, settings):
return max(max(level.main) + 4, max(level.decor) + 3) - settings.correction
def ending_tick(level, settings):
return max(max(level.main), max(level.decor)) + max(15, settings.tin_delay) + 10
def match_slot(s, level, settings):
corrected = s.tick + settings.correction
for i, t in enumerate(level.main):
if abs(corrected - t) <= 4:
return ('main', i, t, s.main[i] == 0)
for i, t in enumerate(level.decor):
if abs(corrected - t) <= 3:
return ('decor', i, t, not s.decor[i])
return ('outside', -1, -1, False)
def future(s):
groups = {}
for r in s.reservations:
if r.target >= s.tick:
groups.setdefault(r.target, []).append(r)
out = []
for target, rs in sorted(groups.items()):
order = sorted(rs, key=lambda r: (r.reserved_at, r.roof, r.id))
out.extend(((r, r == order[0]) for r in order))
return tuple(out)
def reservation_preview(s, level, settings, roof):
corrected = s.tick + settings.correction
slots = [t for t, v in zip(level.main, s.main) if v == 0 and corrected <= t + 4]
slots += [t for t, v in zip(level.decor, s.decor) if not v and corrected <= t + 3]
if not slots:
return None
slot = min(slots)
d = delay(roof, settings)
target = slot + d if d else None
return {'slot': slot, 'target': target, 'overlap': bool(d and any((r.target == target for r in s.reservations)))}
def step(s, level, settings, roof=None):
"""Process raw tick t, manual input, due echo, misses; then advance to t+1."""
if s.done:
return s
if roof is not None and roof not in (0, 1, 2):
raise ValueError('roof must be0/1/2')
t = s.tick
main = list(s.main)
decor = list(s.decor)
reservations = list(s.reservations)
events = list(s.events)
inputs = list(s.inputs)
if roof is not None and t <= close_tick(level, settings):
kind, i, slot, accepted = match_slot(s, level, settings)
inputs.append(Input(t, roof, t + settings.correction, kind, slot, accepted))
if accepted:
if kind == 'main':
main[i] = 1
else:
decor[i] = True
events.append(Event(t, 'manual', roof, slot))
d = delay(roof, settings)
if d:
assert slot + d > t, 'A legal reply must be future, never retroactive.'
reservations.append(Reservation(len(reservations), roof, slot + d, t, slot))
else:
events.append(Event(t, 'wrong', roof, slot))
elif roof is not None:
events.append(Event(t, 'closed', roof))
if t < 0 and t % 30 == 0:
events.append(Event(t, 'count'))
due = sorted((r for r in reservations if r.target == t), key=lambda r: (r.reserved_at, r.roof, r.id))
if due:
events.append(Event(t, 'echo', due[0].roof, due[0].source_slot, due[0].id))
for r in due[1:]:
events.append(Event(t, 'suppressed', r.roof, r.source_slot, r.id))
for i, slot in enumerate(level.main):
if main[i] == 0 and t + settings.correction > slot + 4:
main[i] = -1
events.append(Event(t, 'miss', -1, slot))
return State(t + 1, tuple(main), tuple(decor), tuple(reservations), tuple(events), tuple(inputs), t >= ending_tick(level, settings))
def bar_snapshot(s, level, bar):
start = bar * BAR
return replace(s, main=tuple((v if t < start else 0 for t, v in zip(level.main, s.main))), decor=tuple((v if t < start else False for t, v in zip(level.decor, s.decor))), reservations=tuple((r for r in s.reservations if r.source_slot < start)), events=tuple((e for e in s.events if not (e.kind in ('manual', 'wrong', 'miss') and e.slot >= start))), inputs=tuple((i for i in s.inputs if i.slot < start)))
def session_step(session, level, roof=None):
s = session.state
snaps = session.snapshots
if 0 <= s.tick < level.bars * BAR and s.tick % BAR == 0 and (not any((b == s.tick // BAR for b, _ in snaps))):
snaps = snaps + ((s.tick // BAR, bar_snapshot(s, level, s.tick // BAR)),)
return replace(session, state=step(s, level, session.settings, roof), snapshots=snaps)
def restart_bar(session, level, previous=False):
bar = max(0, min(level.bars - 1, max(session.state.tick - 1, 0) // BAR) - (1 if previous else 0))
if bar == 0:
state = replace(initial(level), tick=0)
else:
state = dict(session.snapshots)[bar]
rewind_tick = bar * BAR - max(0, session.settings.correction)
state = replace(state, tick=rewind_tick, events=tuple((e for e in state.events if e.tick < rewind_tick)), inputs=tuple((i for i in state.inputs if i.tick < rewind_tick)), reservations=tuple((r for r in state.reservations if r.reserved_at < rewind_tick)))
return replace(session, state=state, snapshots=tuple(((b, s) for b, s in session.snapshots if b < bar)), restarts=session.restarts + ((session.state.tick, bar, 'previous' if previous else 'current'),))
def stats(s, level):
hit = sum((x == 1 for x in s.main))
dec = sum(s.decor)
echoes = [e for e in s.events if e.kind == 'echo']
manual = [e for e in s.events if e.kind == 'manual']
echo_targets = {e.tick for e in echoes}
overlaps = sum((1 for e in manual if e.slot in level.decor and e.slot in echo_targets))
used = {e.slot for e in manual} | echo_targets
return {'hits': hit, 'total': len(level.main), 'hit_percent': 100 * hit // len(level.main), 'complete': hit * 4 >= len(level.main) * 3, 'decorations': dec, 'automatic_replies': len(echoes), 'overlaid_slots': overlaps, 'blank_decorations': sum((t not in used for t in level.decor)), 'reservation_collisions': sum((e.kind == 'suppressed' for e in s.events)), 'wrong': sum((e.kind == 'wrong' for e in s.events)), 'manual_hits': len(manual)}
def normalized(v):
return json.dumps(v, ensure_ascii=False, sort_keys=True, separators=(',', ':'))
def digest(v):
return hashlib.sha256(normalized(v).encode('utf-8')).hexdigest()
def simulate(level, inputs, settings=Settings()):
sess = new_session(level, settings)
while not sess.state.done:
sess = session_step(sess, level, inputs.get(sess.state.tick))
return sess
def record(level, inputs, settings=Settings()):
sess = simulate(level, inputs, settings)
data = {'schema': 1, 'level': asdict(level), 'settings': asdict(settings), 'inputs': [[t, r] for t, r in sorted(inputs.items())], 'end': asdict(sess), 'stats': stats(sess.state, level), 'event_sha256': digest([asdict(e) for e in sess.state.events])}
return {**data, 'sha256': digest(data)}
STAGE = 5
# Display uses ASCII so this one file needs no glyph or image assets.
# The original sound definitions are retained, without claiming audible testing.
SOUNDS = (('c2c1', 't', '64', 'ff', 4), ('f4c4', 'p', '54', 'ff', 3),
('c3c2', 'n', '43', 'ff', 4), ('c4', 't', '3', 'f', 3))
COLORS = (9, 12, 6)
class Controller:
"""Pyxel-independent input edges and score clock. tick is the NEXT raw tick."""
def __init__(self):
self.level_index = 0
self.settings = Settings()
self.frame = 0
self.previous = set()
self.pending = []
self.audio_log = []
self.reset(stop=False)
@property
def level(self):
return LEVELS[self.level_index]
@property
def state(self):
return self.session.state if STAGE == 5 else self._state
def reset(self, stop=True):
if stop:
self.stop_audio()
if STAGE == 5:
self.session = new_session(self.level, self.settings)
else:
self._state = initial(self.level)
self.running = False
self.title = True
self.paused = False
self.gate = False
self.countdown = 0
self.accum = 0
self.flashes = [-1000000] * 3
self.caption = 'PRESS ENTER TO START. SPACE = ONE SCORE TICK.'
def emit(self, kind, tick, channel=None, sound=None):
item = {'kind': kind, 'tick': tick, 'frame': self.frame,
'channel': channel, 'sound': sound}
self.pending.append(item)
self.audio_log.append(item.copy())
def stop_audio(self):
self.emit('stop', self.state.tick)
self.flashes = [-1000000] * 3
def apply_tick(self, roof=None):
if self.state.done:
return
before = len(self.state.events)
if STAGE == 5:
self.session = session_step(self.session, self.level, roof)
else:
self._state = step(self.state, self.level, self.settings, roof)
for event in self.state.events[before:]:
if event.kind == 'manual':
self.emit('manual', event.tick, event.roof, event.roof)
self.flashes[event.roof] = self.frame
self.caption = 'MANUAL raw %d / score %d' % (event.tick, event.slot)
elif event.kind == 'echo':
self.emit('echo', event.tick, 3, event.roof)
self.flashes[event.roof] = self.frame
self.caption = 'ECHO target %d / source %d' % (event.tick, event.slot)
elif event.kind == 'count':
self.emit('count', event.tick, 3, 3)
elif event.kind == 'suppressed':
self.caption = 'SAME TARGET: EARLIER RESERVATION WINS'
elif event.kind == 'wrong':
self.caption = 'OUTSIDE WINDOW OR SLOT ALREADY USED'
if self.state.done:
self.running = False
self.gate = True
def rewind(self, previous=False):
if STAGE != 5:
return
self.stop_audio()
self.session = restart_bar(self.session, self.level, previous)
self.countdown = BAR
self.accum = 0
self.running = True
self.paused = False
self.gate = True
self.caption = 'RESTART: SCORE FROZEN DURING 120-TICK COUNT'
def control(self, down):
"""One call per renderer frame; fresh is set difference, never key repeat."""
self.frame += 1
down = set(down)
fresh = down - self.previous
self.previous = down
if self.gate:
if not down:
self.gate = False
return
if 'R' in fresh:
self.reset()
self.gate = True
return
if self.title:
changed = False
if STAGE >= 2:
correction = self.settings.correction
if 'LEFT' in fresh:
correction = max(-6, correction - 1)
elif 'RIGHT' in fresh:
correction = min(6, correction + 1)
changed = correction != self.settings.correction
self.settings = replace(self.settings, correction=correction)
if STAGE >= 4:
for i, key in enumerate(('1', '2', '3')):
if key in fresh:
self.level_index = i
changed = True
break
if 'B' in fresh:
self.settings = replace(self.settings, tin_delay=15 if self.settings.tin_delay == 45 else 45)
changed = True
if 'V' in fresh:
self.settings = replace(self.settings, speed=75 if self.settings.speed == 100 else 100)
changed = True
if changed:
self.reset()
self.gate = True
return
if 'RETURN' in fresh:
self.title = False
self.gate = True
self.caption = 'SPACE: STEP / ENTER: AUTO / A S D: ONE FRESH HIT'
return
if STAGE == 5 and 'P' in fresh:
if self.paused:
self.rewind(False)
else:
self.paused = True
self.running = False
self.stop_audio()
self.gate = True
return
if self.paused:
return
if STAGE == 5 and 'X' in fresh:
self.rewind(True)
return
if self.state.done:
return
if 'RETURN' in fresh:
self.running = not self.running
if not self.running:
self.stop_audio()
self.gate = True
return
# A chord is one hit; the canonical order is A, then S, then D.
roof = next((i for i, key in enumerate(('A', 'S', 'D'))
if key in fresh and (STAGE >= 3 or i == 0)), None)
if not self.running:
if 'SPACE' in fresh or roof is not None:
self.apply_tick(roof)
return
# 30 renderer frames/sec; 2 logical ticks/frame (or 1,2,1,2 at 75%).
self.accum += 4 if self.settings.speed == 100 else 3
dt, self.accum = divmod(self.accum, 2)
if self.countdown:
if self.countdown % 30 < dt:
self.emit('restart_count', self.state.tick, 3, 3)
self.countdown = max(0, self.countdown - dt)
if not self.countdown:
self.gate = True
return
for i in range(dt):
self.apply_tick(roof if i == 0 else None)
def flush_audio(self, pyxel):
"""Drain only the new effects; drawing or draining twice never replays them."""
pending, self.pending = self.pending, []
for event in pending:
if event['kind'] == 'stop':
pyxel.stop()
else:
pyxel.play(event['channel'], event['sound'])
return pending
def demo(self):
self.title = False
inputs = {0: 0, 30: 0, 60: 0} if STAGE < 3 else {0: 2, 30: 1, 60: 0}
if STAGE == 2:
inputs[15] = 0
while self.state.tick <= 75:
self.frame += 1
self.apply_tick(inputs.get(self.state.tick))
self.caption = ('CHECKPOINT: 3 MAIN HITS' if STAGE < 3 else
'0+45 = 30+15: SAME TARGET 45')
def draw(self, p):
s = self.state
stat = stats(s, self.level)
p.cls(1)
p.text(14, 10, 'RAIN ROOF DRUMS / STAGE %d / %s' % (STAGE, self.level.id), 10)
p.text(14, 27, 'NEXT RAW %d / CORRECTED %d / SPEED %d%%' %
(s.tick, s.tick + self.settings.correction, self.settings.speed), 7)
mode = 'TITLE' if self.title else 'PAUSED' if self.paused else 'RESULT' if s.done else 'AUTO' if self.running else 'STEP'
p.text(14, 44, mode + ' / 60 SCORE TICKS = 1 SEC AT 100% / 120 = ONE BAR', 6)
p.text(14, 63, 'MAIN %d/%d DECOR %d ECHO %d SUPPRESSED %d' %
(stat['hits'], stat['total'], stat['decorations'], stat['automatic_replies'], stat['reservation_collisions']), 11)
bar = max(0, min(3, max(0, s.tick) // BAR))
for row in range(2):
start = (bar + row) * BAR
y = 88 + row * 80
p.rect(12, y, 616, 71, 0)
label = 'BAR %d' % (bar + row + 1) if bar + row < 4 else 'TAIL'
p.text(18, y + 4, '%s / SCORE %d..%d' % (label, start, start + 119), 6)
for t in range(start, start + 120, 15):
x = 85 + (t - start) * 4
p.line(x, y + 20, x, y + 63, 5)
p.text(x - 4, y + 11, str(t), 5)
if t in self.level.main:
value = s.main[self.level.main.index(t)]
p.rectb(x - 4, y + 26, 9, 9, 11 if value == 1 else 8 if value == -1 else 7)
elif t in self.level.rests:
p.text(x - 5, y + 28, '-', 13)
if STAGE >= 2 and t in self.level.decor:
p.circb(x, y + 44, 3, 10 if s.decor[self.level.decor.index(t)] else 5)
if STAGE >= 3:
for reservation, active in future(s):
if start <= reservation.target < start + 120:
x = 85 + (reservation.target - start) * 4
color = COLORS[reservation.roof] if active or STAGE == 3 else 5
if reservation.roof == 1:
p.circb(x, y + 54, 5, color)
else:
p.tri(x, y + 49, x - 4, y + 57, x + 4, y + 57, color)
if start <= s.tick < start + 120:
x = 85 + (s.tick - start) * 4
p.line(x, y + 21, x, y + 64, 10)
for i, name in enumerate(('A WOOD', 'S TILE +15', 'D TIN +%d' % self.settings.tin_delay)):
x = 18 + i * 207
enabled = i == 0 or STAGE >= 3
# Age is zero on the event frame; ages 0..4 light five frames.
lit = 0 <= self.frame - self.flashes[i] < 5
p.tri(x, 286, x + 92, 250, x + 183, 286, COLORS[i] if lit else 5)
p.rectb(x + 12, 287, 158, 28, COLORS[i] if enabled else 5)
p.text(x + 22, 298, name if enabled else 'ADDED IN STAGE 03', 7 if enabled else 5)
p.text(14, 326, self.caption, 10)
p.text(14, 344, 'LAST EVENTS: raw tick / kind / roof / source slot', 6)
for i, e in enumerate(s.events[-6:]):
p.text(22, 358 + i * 12, '%4d %-10s roof %2d score %4d' % (e.tick, e.kind, e.roof, e.slot), 7)
if STAGE >= 3:
p.text(365, 344, 'NEXT REPLIES', 6)
pending = future(s)[:4]
for i, (reservation, active) in enumerate(pending):
p.text(365, 360 + i * 12, '%d %s %s' % (reservation.target, ('WOOD', 'TILE', 'TIN')[reservation.roof], 'ACTIVE' if active or STAGE == 3 else 'MUTED'), COLORS[reservation.roof] if active or STAGE == 3 else 5)
if not pending:
p.text(365, 360, '(none)', 5)
if self.countdown:
p.text(365, 415, 'RESTART COUNT %d' % self.countdown, 10)
if s.done:
p.text(365, 415, 'COMPLETE' if stat['complete'] else 'TRY AGAIN', 11)
p.text(14, 438, 'ENTER: START / AUTO-STEP TOGGLE SPACE: ONE TICK R: RESET', 7)
p.text(14, 453, ('A: WOOD' if STAGE < 3 else 'A/S/D: ROOFS') + ' ESC: QUIT HOLD = NO REPEAT', 7)
p.text(14, 468, 'TITLE: LEFT/RIGHT CORRECTION' if STAGE >= 2 else 'STAGE 01: MAIN BEATS ONLY / NO REPLIES', 6)
if STAGE >= 4:
p.text(300, 468, '1/2/3 CHART B DELAY V SPEED', 6)
p.text(14, 484, 'P: PAUSE / RESTART CURRENT BAR X: PREVIOUS BAR' if STAGE == 5 else 'SAVE EDIT -> QUIT -> RUN THIS SAME FILE AGAIN', 13)
if self.title:
p.rect(70, 106, 500, 100, 1)
p.rectb(70, 106, 500, 100, 12)
p.text(85, 120, 'PRESS ENTER. THEN SPACE STEPS OR ENTER RUNS.', 10)
p.text(85, 140, 'MAIN: +/-4 TICKS. DECOR FROM STAGE 02: +/-3.', 7)
p.text(85, 160, 'NEXT RAW = TICK THAT THE NEXT UPDATE WILL PROCESS.', 6)
p.text(85, 180, 'R RESETS STATE. IT DOES NOT RELOAD SAVED CODE.', 13)
def main(headless=False, screenshot=None, report=None, checkpoint=False):
# CLI validates and resolves paths before reaching this import.
import pyxel as p
p.init(640, 504, title='Rain roof lesson %d' % STAGE, fps=30,
display_scale=1, headless=headless)
for i, sound in enumerate(SOUNDS):
p.sounds[i].set(*sound)
app = Controller()
actual_calls = []
if checkpoint:
app.title = False
keys = {k: getattr(p, 'KEY_' + k) for k in
('RETURN', 'SPACE', 'R', 'A', 'S', 'D', 'P', 'X', 'LEFT', 'RIGHT', '1', '2', '3', 'B', 'V')}
def update():
if not headless:
app.control({k for k, value in keys.items() if p.btn(value)})
elif checkpoint and app.state.tick <= 75:
app.frame += 1
inputs = {0: 0, 30: 0, 60: 0} if STAGE < 3 else {0: 2, 30: 1, 60: 0}
if STAGE == 2:
inputs[15] = 0
for _ in range(2):
if app.state.tick <= 75:
app.apply_tick(inputs.get(app.state.tick))
actual_calls.extend(app.flush_audio(p))
if p.btnp(p.KEY_ESCAPE) or headless and p.frame_count >= 3 and (not checkpoint or app.state.tick > 75):
app.stop_audio()
actual_calls.extend(app.flush_audio(p))
app.draw(p)
if screenshot:
p.screen.save(screenshot, 1)
if report:
from pathlib import Path
payload = {'stage': STAGE, 'state': asdict(app.state),
'stats': stats(app.state, app.level), 'audio_calls': actual_calls,
'scope': 'Actual Pyxel API calls; dummy driver is not audible verification',
'screenshot': screenshot}
Path(report).write_text(json.dumps(payload, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
p.quit()
p.run(update, lambda: app.draw(p))
if __name__ == '__main__':
import argparse
from pathlib import Path
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--headless-smoke', action='store_true')
parser.add_argument('--demo-checkpoint', action='store_true')
parser.add_argument('--screenshot')
parser.add_argument('--report')
args = parser.parse_args()
if (args.demo_checkpoint or args.screenshot or args.report) and not args.headless_smoke:
parser.error('checkpoint and output files require --headless-smoke')
for name, suffix in (('screenshot', '.png'), ('report', '.json')):
value = getattr(args, name)
if value:
path = Path(value).resolve()
if path.suffix.lower() != suffix or not path.parent.is_dir() or path.exists():
parser.error('use a NEW ' + suffix + ' file in an existing directory')
if path.parent == Path(__file__).resolve().parent:
parser.error('output must be outside the script directory')
setattr(args, name, str(path))
main(args.headless_smoke, args.screenshot, args.report, args.demo_checkpoint)仕組みがわかったら、元の完成版を読んでみよう。
ルールのまとめ：描画は30Hz、ルール上の時計は60Hz、120刻で1小節。受け付ける幅は、主拍±4・自由な飾り±3。手で叩いた音は raw の時刻、判定だけ補正。瓦とトタンの返事は、譜面の刻+15/+45。重なった自動の返事は、先に予約した1つだけで、連鎖なし。主拍75%以上で基本のクリア。P/X は小節の状態に戻してからカウント。
ここまでに作ったのは、元のゲームの仕組みだけを取り出して作り直した「簡略版」です。元のゲームのソースは手を加えずに、別のZIP（元の完成版ソース）に入れてあります。
| 比べる点 | 5ステップの教材 | 元の完成版 |
|---|---|---|
| 段階 | 01 主拍と木の屋根だけ、02 飾りとずれの補正、03 返事を全部鳴らす、04 重なりの打ち消し・3つの譜面・設定、05 小節の頭からの再開。 | すべての機能と、重なりの打ち消しが最初から使えます。 |
| 操作と描画 | 英数字（ASCII）、1つのファイルで動く、Enter での STEP/AUTO と Space での1刻送りを追加。 | 日本語の文字、表紙と説明、左右＋Z の入力、試し打ち、記録の画面。 |
| 音と検証 | 元のゲームと同じ音の定義。03 だけは、重なる play を同じチャンネルに送る、まだ完成していない段階。仮の音声ドライバー（dummy）で命令が動くことは確かめましたが、耳で聞いての確認はしていません。 | 最初から、自動の音は重なりを打ち消したあとで鳴らします。元のゲームの assets と音のコードは変えずに残しています。 |
完成版のコードで、次に読むところ
- model.py の
match_slot/step：判定と、予約した返事 - model.py の
future/reservation_preview：読むだけの予告 - model.py の
bar_snapshot/restart_bar：早取りの除去と、返事の持ち越し - app.py の
apply_tick/play/stop_audio：音とのつなぎ方 - app.py の
update：画面の更新ごとの加算器と、入力の待ち - experiments.py / test_model.py / test_ui.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の実際のパソコンでの操作や、音が出るかどうかはまだ確認していません。人がすべての操作を試したわけでもありません。ブラウザー版の確認範囲は確認状況をご覧ください。
うまく動かないときは：起動やエラーの対処法 / 質問用のメモを作る