diff --git a/SerialController/Gamepad.py b/SerialController/Gamepad.py
index 5353a99..68ac252 100644
--- a/SerialController/Gamepad.py
+++ b/SerialController/Gamepad.py
@@ -100,7 +100,7 @@ def __init__(
is_active: Callable[[], bool] | None = None,
reader: Any | None = None,
on_display: Callable[[dict[str, Any]], None] | None = None,
- deadzone: int = 0,
+ deadzone: int = 1638,
) -> None:
"""sender へ差分申告する。sender が None なら作れない。
@@ -112,7 +112,7 @@ def __init__(
on_display は読むたびに呼ぶ表示口（仮想パッドの鏡用）。
読取スレッドから呼ぶため、Tk を触る処理は呼び出し側で
root.after 経由にすること。
- deadzone は SDL 経路のスティック遊び（0〜8192）。
+ deadzone は SDL 経路のスティック遊び（0〜8192、既定1638=最大値の5%）。
XInput 経路は既定のまま変えない。
"""
if sender is None:
diff --git a/SerialController/Menubar.py b/SerialController/Menubar.py
index 0ea0135..4235c22 100644
--- a/SerialController/Menubar.py
+++ b/SerialController/Menubar.py
@@ -536,6 +536,11 @@ def OpenBlocklyEditor(self) -> None:
reload_commands=self.app.reloadCommands,
get_frame=blockly_capture.build_get_frame(lambda: self.camera),
run_host=self.app,
+ record_blocked=lambda: (
+ "実行中は記録できません（止めてから始めてください）"
+ if self.app.runner.is_busy()
+ else None
+ ),
)
def OpenErrorReport(self) -> None:
diff --git a/SerialController/Settings.py b/SerialController/Settings.py
index 0f4faeb..809cf5f 100644
--- a/SerialController/Settings.py
+++ b/SerialController/Settings.py
@@ -71,7 +71,7 @@ def __init__(self, profile: str = "") -> None:
value=general.getint("gamepad_index", fallback=0)
)
self.gamepad_deadzone = tk.IntVar(
- value=general.getint("gamepad_deadzone", fallback=0)
+ value=general.getint("gamepad_deadzone", fallback=1638)
)
# マウスでのスティック操作。GUI のチェックボックスと1対1で対応する
self.is_use_left_stick_mouse = tk.BooleanVar(
diff --git a/SerialController/Window.py b/SerialController/Window.py
index f598b98..8c6206e 100644
--- a/SerialController/Window.py
+++ b/SerialController/Window.py
@@ -273,6 +273,19 @@ def _init_state(self) -> None:
# 走行記録の受け先。SerialService が担い、runner が開始・終了で呼ぶ。
# 記録の失敗で実行は壊さない（runner・service 側で握る）。
self.runner.set_diag(self.serial, self.profile)
+
+ # 記録中の通常コマンドの開始拒否。判定は編集側の記録を見る。
+ # ui への直参照は作らず、遅延読みで循環を避ける（runner は ui を
+ # 知らない。Window が両者をつなぐ）。
+ def _recording_now() -> bool:
+ try:
+ from ui import blockly_editor
+
+ return bool(blockly_editor.is_recording())
+ except Exception:
+ return False
+
+ self.runner.set_record_guard(_recording_now)
# 履歴の受け先。開始・終了で 1 行ずつ積む。渡さなければ残さない。
self.runner.set_history(self.command_history, self._commandKind)
self._closing = False
diff --git a/SerialController/assets/blockly/editor.html b/SerialController/assets/blockly/editor.html
index c1b4567..80943f8 100644
--- a/SerialController/assets/blockly/editor.html
+++ b/SerialController/assets/blockly/editor.html
@@ -2100,9 +2100,9 @@ 作例から始める
on('pausebtn', 'click', function () { runControl(run.paused ? 'resume' : 'pause'); });
on('stepbtn', 'click', function () { runControl('step'); });
on('stopbtn', 'click', function () { runControl('stop'); });
- // ---- 操作の記録（本体で押した操作を押すブロックの並びにする） ----
+ // ---- 操作の記録（本体で押した操作をブロックの並びにする） ----
// 本体で実際に押した操作（送信線へ出た入力）を記録し、止めたときに
- // pokecon_press ブロックの並びにして挿入する。同時押しの再現はしない。
+ // 単独押しは pokecon_press、同時押しは hold・holdEnd の連鎖にして挿入する。
var rec = { active: false };
function setRecUi(active) {
rec.active = !!active;
diff --git a/SerialController/assets/blockly/pokecon_editor.js b/SerialController/assets/blockly/pokecon_editor.js
index fd29ab6..4652f3d 100644
--- a/SerialController/assets/blockly/pokecon_editor.js
+++ b/SerialController/assets/blockly/pokecon_editor.js
@@ -16,6 +16,7 @@
// motion 材料 1 件をブロックにする。press 手順なら null を返す。
// kind: stick → pokecon_motion、stick2 → pokecon_motion2、
// hold/hold_end → pokecon_hold/pokecon_hold_end、
+ // wait → pokecon_wait（sec・wait の順で秒を読む）、
// release_stick は直前の motion 系ブロックの END を RELEASE にする。
var lastMotionBlock = null;
function pairText(pair) {
@@ -66,11 +67,28 @@
} else if (s.kind === "hold") {
b = ws.newBlock("pokecon_hold");
var target = String(s.target || "Button.A");
- setFields(b, { TARGET: target, WAIT: 0.1 });
+ // chord 連鎖の持ち時間（at_ms 由来の wait も同じ欄で来る）。
+ // 数値が無ければ従来どおり既定の 0.1。
+ var holdWait = Number(s.wait);
+ if (!isFinite(holdWait)) {
+ holdWait = 0.1;
+ }
+ setFields(b, { TARGET: target, WAIT: holdWait });
} else if (s.kind === "hold_end") {
b = ws.newBlock("pokecon_hold_end");
var endTarget = String(s.target || "Button.A");
setFields(b, { TARGET: endTarget });
+ } else if (s.kind === "wait") {
+ // chord 連鎖の待ち（変換器は sec・wait のどちらかで秒を載せる）。
+ // press フォールバックへ落とすと押下になるためここで受ける。
+ b = ws.newBlock("pokecon_wait");
+ var sec = Number(s.sec);
+ if (!isFinite(sec)) {
+ sec = Number(s.wait);
+ }
+ if (isFinite(sec)) {
+ setFields(b, { SEC: sec });
+ }
} else if (s.kind === "release_stick") {
// 直前の motion 系の END を RELEASE にする（ブロックは増やさない）。
if (lastMotionBlock) {
@@ -80,6 +98,10 @@
}
lastMotionBlock = null;
return null;
+ } else if (s.kind === "press") {
+ // chord 連鎖・motion クリップ外の press は従来の pokecon_press 扱い。
+ // 欄への具体値は insertSteps 側の press フォールバックが付ける。
+ return null;
} else {
return null;
}
@@ -637,11 +659,12 @@
// 記録した手順をブロックの並びにして挿入する。
// steps は [{target, duration, wait}]（press 手順、services/blockly_record
- // の形）か、motion 材料（{kind: stick/stick2/hold/hold_end/release_stick}）
- // の混在。target は Button.A / Hat.TOP / Direction.UP の形で、BUTTON 欄の
+ // の形。{kind: 'press', ...} も同じ扱い）か、motion 材料（{kind:
+ // stick/stick2/hold/hold_end/wait/release_stick}）の混在。target は
+ // Button.A / Hat.TOP / Direction.UP の形で、BUTTON 欄の
// 書式に合わせる（ボタンだけ素の名 A、方向は Hat./Direction. のまま）。
// motion 材料は pokecon_motion / pokecon_motion2・pokecon_hold・
- // pokecon_hold_end ブロックになる。
+ // pokecon_hold_end・pokecon_wait ブロックになる。
// 挿入位置は、選択中の文ブロックがあればその直後（元の後続は最後の
// 新ブロックの後ろへつけ直す）、無ければプログラムの DO の末尾、
// プログラムが無ければ空いている所。全体を1つの取り消し単位にする。
diff --git a/SerialController/config.py b/SerialController/config.py
index 94d61c5..d2ff624 100644
--- a/SerialController/config.py
+++ b/SerialController/config.py
@@ -84,9 +84,9 @@ def default_sections() -> dict[str, dict[str, Any]]:
"is_use_keyboard": True,
"is_use_gamepad": False,
"gamepad_index": 0,
- # ゲームパッドのスティック遊び（0〜8192、既定0=なし）。
- # XInput 既定 7849・約24%、Pro コン実測ブレ考慮なら 1600〜2600。
- "gamepad_deadzone": 0,
+ # ゲームパッドのスティック遊び（0〜8192、既定1638=最大値の5%）。
+ # XInput 既定 7849・約24% より浅く、Pro コン実測ブレ（1600前後）を抑える。
+ "gamepad_deadzone": 1638,
"is_use_left_stick_mouse": False,
"is_use_right_stick_mouse": False,
"is_take_stick_log": False,
@@ -270,7 +270,7 @@ def complete_missing(parser: configparser.ConfigParser) -> list[str]:
"baud_rate": (9600, lambda value: value > 0),
"fps": (45, lambda value: value in (5, 15, 30, 45, 60)),
"gamepad_index": (0, lambda value: 0 <= value <= 3),
- "gamepad_deadzone": (0, lambda value: 0 <= value <= 8192),
+ "gamepad_deadzone": (1638, lambda value: 0 <= value <= 8192),
}
for key, (default, valid) in integer_rules.items():
try:
diff --git a/SerialController/core/pad_source.py b/SerialController/core/pad_source.py
index fe46a2a..5637a1e 100644
--- a/SerialController/core/pad_source.py
+++ b/SerialController/core/pad_source.py
@@ -112,13 +112,22 @@ def scan(self) -> list[int]:
return sorted(found)
def names(self) -> dict[int, str]:
- """番号 → 表示名。SDL で分かる分だけ載る。"""
+ """番号 → 表示名。XInput 側は汎名、SDL で分かる分は実名で返す。"""
try:
names = self.sdl.names()
- return {int(k): str(v) for k, v in names.items()}
+ out = {int(k): str(v) for k, v in names.items()}
except Exception:
- return {}
+ out = {}
+ for index in range(MAX_PADS):
+ if index in out:
+ continue
+ try:
+ state = self.xinput.read(index)
+ except Exception:
+ continue
+ if state.connected:
+ out[index] = "XInput パッド"
+ return out
def route_of(self, index: int) -> str:
- """その番号がいまどちらの口で読めているか（検証・表示用）。"""
return self._route.get(int(index), "")
diff --git a/SerialController/core/serial/sender.py b/SerialController/core/serial/sender.py
index 1f87db1..1a004b7 100644
--- a/SerialController/core/serial/sender.py
+++ b/SerialController/core/serial/sender.py
@@ -1723,10 +1723,9 @@ def _logLiveError(self) -> None:
def startLiveWorker(self, transport: Any = None) -> bool:
"""workerを1本だけ起動し、現在姿勢を最初に送る。
- 確認と起動を _live_lock の中で行う。二重起動で worker が
- 漏れると、古い方が止められず線が2本になる。
- 準備（列への投入）は外で行う。錠の中で putLive すると
- 同じ錠を取り直して詰まる（Lock は再入不可）。
+ 確認・登録・起動を _live_lock の一つの臨界区間で行う。二重起動で
+ worker が漏れると、古い方が止められず線が2本になる。
+ 準備の写し（snapshot）は外で取る。姿勢の錠と順序逆転で詰むため。
動いているときは帳簿に触らず False で抜ける。
"""
return self._startLiveWorker(transport) is not None
@@ -1742,25 +1741,23 @@ def _startLiveWorker(self, transport: Any = None) -> threading.Thread | None:
transport = getattr(self, "transport", None)
if transport is None:
return None
+ # 起動準備の写しは錠の外で取る。snapshot は姿勢の錠（_lock）を
+ # 取るため、_live_lock の中で呼ぶと順序逆転で詰む（Lock は再入不可）。
+ # putLive は scheduler 側の錠だけを取り、_live_lock は統計の
+ # 一瞬だけなので、ここで呼んでも起動の臨界区間には入らない。
+ pending_snap = self.snapshot()
with self._live_lock:
thread = self._live_thread
if thread is not None and thread.is_alive():
return None
- self._live_stop.clear()
- with self._live_lock:
+ self._live_stop.clear()
self._live_last_snap = None
self._live_last_sent_at = None
self._live_last_shown_at = None
self._live_last_shown_snap = None
- self.putLive(self.snapshot())
- with self._live_lock:
- thread = self._live_thread
- if thread is not None and thread.is_alive():
- return None
- # 生成・登録・起動まで錠の中で行う。登録と起動の間に錠を離すと、
- # 別の起動が「登録済みだが未起動（is_alive False）」を見て
- # もう1本起こし、線が2本になる。start() は錠を取らないため、
- # ここで起こしても詰まらない。
+ # 生成・登録・起動まで一つの臨界区間で行う。確認と起動の
+ # 間に錠を離すと、並行した起動が同じ隙へ入り線が2本になる。
+ # start() は錠を取らないため、ここで起こしても詰まらない。
started = threading.Thread(
target=self._liveLoop,
args=(transport,),
@@ -1769,7 +1766,16 @@ def _startLiveWorker(self, transport: Any = None) -> threading.Thread | None:
)
self._live_thread = started
started.start()
- return started
+ started_alive = started.is_alive()
+ if not started_alive:
+ # 起動直後に死んでいたら登録を外す（isLiveWorkerRunning と
+ # 二重起動防止の判定が「死んだ登録」を残さないため）。
+ with self._live_lock:
+ if self._live_thread is started:
+ self._live_thread = None
+ return None
+ self.putLive(pending_snap)
+ return started
def discardLive(self) -> bool:
"""scheduler の未送出を破棄する。
diff --git a/SerialController/services/blockly_record.py b/SerialController/services/blockly_record.py
index 85efa72..0d6e526 100644
--- a/SerialController/services/blockly_record.py
+++ b/SerialController/services/blockly_record.py
@@ -7,9 +7,9 @@
戻し、編集画面の `pokecon_press` ブロックの並び（押す・待つ）にして挿入する
ための材料を作る。「操作に合わせた形で入力したい」という要望への答え。
-同時押しの再現は扱わない。重なった区間は無視し、離した順に並べるだけに
-する（離した順が押した順と食い違うと、再現時に別の順で押されるため、
-同時押しのつもりの操作は手順として残さない、ではなく順序だけ残す）。
+同時押し（重なった区間）は hold/holdEnd の連鎖（chord連鎖）にして残す。
+重ならない単独押しは従来どおり press のまま。スティックと同時のボタンは
+motion クリップ側で hold 化する（build_motion_steps）。
tkinter には触らない。受け口（ui/blockly_editor.py）が GUI スレッドで
付け外しし、ここは線の聞き手と純粋な変換だけを持つ。
@@ -36,6 +36,16 @@
#: 編集画面の既定（0.1秒）に揃える。
LAST_WAIT = 0.1
+#: chord連鎖の hold 間 wait の刻み。press 系の 0.05 とは分け、0.01 で丸める。
+#: 同時開始のずれを出さないよう 0 も許す（press 系の最小 0.05 へは寄せない）。
+CHORD_ROUND = 0.01
+#: hold の WAIT 欄の上限。超えた分は pokecon_wait 相当（kind:wait）へ分割する。
+#: ブロック欄の上限（pokecon_hold の WAIT は最大60）に合わせる。
+CHORD_WAIT_MAX = 60.0
+#: 延長後のクリップ長の警告閾値（秒）。長押しの吸収で記録全体が1クリップに
+#: 膨らんだとき、編集しにくくなるため warnings へ1件だけ知らせる。仮値。
+CLIP_WARN_S = 10.0
+
def _round_step(value: float) -> float:
"""0.05秒単位に丸め、最小0.05・上限60に収める。"""
@@ -116,6 +126,317 @@ def events_to_steps(events: list[InputLog.InputEvent]) -> list[dict[str, Any]]:
return steps
+# ---------------------------------------------------------------------------
+# 同時押しの記録（chord連鎖）
+# ---------------------------------------------------------------------------
+# events_to_steps は重なりを無視して離した順に並べるだけだった。物理パッドの
+# 直送では同時押しが増えたため、真に重なる区間の連結成分だけ hold/holdEnd の
+# 連鎖にして残す。重ならない単独押しは press のまま（可読性・既存保存物のため）。
+#
+# 再生時の送信行は hold の積み重ね（A→A+R→R→中立）になる。完全同時の PRESS 対も
+# 連鎖化するため、元の1行に対する過渡1行が入る。マクロ用途の合意事項とする。
+# Hat の切替ロールは hold(新)→holdEnd(旧) の順に入れ替え、中立過渡を出さない
+# （holdButton残留ガードと inputEnd の hold_hat 復元に依存する。参照は関数名で
+# 行い、行番号では行わないこと）。
+
+
+#: 押下区間（名前, 種類, 開始at, 終了at|None, PRESS, RELEASE|None）。
+_PressSpan = tuple[
+ str, str, float, float | None, InputLog.InputEvent, InputLog.InputEvent | None
+]
+
+
+def _press_intervals(events: list[InputLog.InputEvent]) -> list[_PressSpan]:
+ """ボタン/Hat の押下区間を返す。
+
+ PRESS→RELEASE を名前ごとに先入れ先出しで組にする。開始前の RELEASE
+ （PRESS が無いもの）は捨てる。停止時に押したまま（RELEASE が無いもの）は
+ 終了 None で残す。stick・CHANGE は扱わない（stick 混じりは motion 側の仕事）。
+ """
+ pending: dict[str, list[tuple[float, InputLog.InputEvent]]] = {}
+ spans: list[_PressSpan] = []
+ for ev in events:
+ if ev.kind not in ("button", "hat"):
+ continue
+ if ev.action == "PRESS":
+ pending.setdefault(ev.name, []).append((float(ev.at), ev))
+ elif ev.action == "RELEASE":
+ queue = pending.get(ev.name)
+ if not queue:
+ # 開始前から押していたもの。捨てる（件数は返さない）。
+ continue
+ start, press_ev = queue.pop(0)
+ spans.append((ev.name, press_ev.kind, start, float(ev.at), press_ev, ev))
+ for queue in pending.values():
+ for start, press_ev in queue:
+ spans.append((press_ev.name, press_ev.kind, start, None, press_ev, None))
+ return spans
+
+
+def _chord_grid(moment: float, origin: float) -> float:
+ """連鎖先頭からの経過を 0.01 秒単位の格子へ寄せる（累積丸めの土台）。"""
+ try:
+ stepped = round((float(moment) - float(origin)) / CHORD_ROUND) * CHORD_ROUND
+ except (TypeError, ValueError):
+ return 0.0
+ return round(stepped, 2)
+
+
+def _split_chord_wait(total: float) -> list[float]:
+ """待ちを CHORD_WAIT_MAX 以下の塊へ割る。0 以下は空にする。"""
+ total = round(float(total), 2)
+ if not total > 0.0:
+ return []
+ chunks: list[float] = []
+ while total > CHORD_WAIT_MAX:
+ chunks.append(CHORD_WAIT_MAX)
+ total = round(total - CHORD_WAIT_MAX, 2)
+ if total > 0.0:
+ chunks.append(total)
+ return chunks
+
+
+def events_to_chord_steps(
+ events: list[InputLog.InputEvent],
+ stop_at: float | None = None,
+) -> list[dict[str, Any]]:
+ """イベント列を同時押し再現の手順（hold連鎖）へ変換する。純関数。
+
+ PRESS→RELEASE を区間化し、真に重なる連結成分
+ （max(開始) < min(終了)。端点タッチは重なりとみなさない）だけ
+ hold/holdEnd の連鎖にし、孤立区間は events_to_steps 互換の press のまま
+ （kind だけ足す）。stick・CHANGE は扱わない。
+
+ Hat は単一値チャンネルのため、新優先・旧即RELEASE で区間を詰める。
+ 同atの Hat RELEASE＋PRESS 対（切替ロール。単独でも）は
+ hold(新)→holdEnd(旧) の順に入れ替える（新先旧後。中立過渡を出さない）。
+
+ hold の WAIT は次のエッジまでの差分を、連鎖先頭基準の累積丸め
+ （0.01 刻み・0 許容。上限超は kind:wait へ分割）で付ける。holdEnd の後の
+ 間隔と連鎖末尾の間隔は kind:wait（pokecon_wait 相当）で出す。
+ 開始前の RELEASE は捨てる。停止時に押下中は、孤立なら press
+ （duration＝停止−押下）、連鎖内なら holdEnd で閉じる。
+ 同一 target の未解放 hold が重なったら ValueError（Keys.hold は重複を
+ 黙って捨てるため、変換の出口で検出する）。
+ """
+ if not events:
+ return []
+ if stop_at is None:
+ stop_at = max(float(ev.at) for ev in events)
+ event_seq = {id(ev): index for index, ev in enumerate(events)}
+ button_order = {name: index for index, name in enumerate(InputLog.BUTTON_NAMES)}
+ # 区間化し、終了の無いものは停止時刻で閉じる。
+ spans: list[dict[str, Any]] = []
+ for name, kind, start, end, press_ev, release_ev in _press_intervals(events):
+ spans.append(
+ {
+ "name": name,
+ "kind": kind,
+ "start": start,
+ "end": stop_at if end is None else end,
+ "press_seq": event_seq.get(id(press_ev), 0),
+ "rel_seq": event_seq.get(id(release_ev), 0)
+ if release_ev is not None
+ else -1,
+ }
+ )
+ # Hat は新優先・旧即RELEASE で詰める（単一値チャンネルのため）。
+ hat_starts = sorted(
+ (sp for sp in spans if sp["kind"] == "hat"),
+ key=lambda sp: (sp["start"], sp["press_seq"]),
+ )
+ for newer in hat_starts:
+ for older in spans:
+ if older["kind"] != "hat" or older["name"] == newer["name"]:
+ continue
+ if older["start"] < newer["start"] < older["end"]:
+ older["end"] = newer["start"]
+ older["rel_seq"] = newer["press_seq"]
+ # 同名 Hat の端点連結は1区間へ畳む（離すと二重holdになる）。
+ spans.sort(key=lambda sp: (sp["kind"] != "hat", sp["name"], sp["start"]))
+ chained: list[dict[str, Any]] = []
+ for span in spans:
+ prev = chained[-1] if chained else None
+ if (
+ prev is not None
+ and span["kind"] == "hat"
+ and prev["kind"] == "hat"
+ and prev["name"] == span["name"]
+ and prev["end"] == span["start"]
+ ):
+ prev["end"] = max(prev["end"], span["end"])
+ prev["rel_seq"] = span["rel_seq"]
+ continue
+ chained.append(span)
+ spans = chained
+ # 真に重なる区間の連結成分を作る。Hat 同士の端点連結（切替ロール）も
+ # 同じ成分にする。ボタンの端点タッチは重なりとみなさない。
+ count = len(spans)
+ parent = list(range(count))
+
+ def find(node: int) -> int:
+ while parent[node] != node:
+ parent[node] = parent[parent[node]]
+ node = parent[node]
+ return node
+
+ def union(left: int, right: int) -> None:
+ left_root = find(left)
+ right_root = find(right)
+ if left_root != right_root:
+ parent[right_root] = left_root
+
+ for i in range(count):
+ for j in range(i + 1, count):
+ first = spans[i]
+ second = spans[j]
+ if (
+ first["kind"] == "hat"
+ and second["kind"] == "hat"
+ and first["name"] != second["name"]
+ and (first["end"] == second["start"] or second["end"] == first["start"])
+ ):
+ union(i, j)
+ elif max(first["start"], second["start"]) < min(
+ first["end"], second["end"]
+ ):
+ union(i, j)
+ groups: dict[int, list[int]] = {}
+ for index in range(count):
+ groups.setdefault(find(index), []).append(index)
+ ordered = sorted(
+ groups.values(),
+ key=lambda idxs: (
+ min(spans[i]["start"] for i in idxs),
+ min(spans[i]["end"] for i in idxs),
+ ),
+ )
+ comp_starts = [min(spans[i]["start"] for i in idxs) for idxs in ordered]
+ press_times = sorted(sp["start"] for sp in spans)
+ steps: list[dict[str, Any]] = []
+ for group_no, idxs in enumerate(ordered):
+ origin = comp_starts[group_no]
+ next_start = comp_starts[group_no + 1] if group_no + 1 < len(ordered) else None
+ if len(idxs) == 1:
+ # 孤立区間は events_to_steps 互換の press のまま。
+ span = spans[idxs[0]]
+ duration = _round_step(span["end"] - span["start"])
+ wait = LAST_WAIT
+ for moment in press_times:
+ if moment > span["end"]:
+ wait = _round_step(moment - span["end"])
+ break
+ steps.append(
+ {
+ "kind": "press",
+ "target": span["name"],
+ "duration": duration,
+ "wait": wait,
+ }
+ )
+ continue
+ # 連鎖内では Hat 入替対（同at RELEASE＋PRESS）を新先旧後にする。
+ swap_pressers = {
+ presser
+ for presser in idxs
+ for releaser in idxs
+ if presser != releaser
+ and spans[releaser]["kind"] == "hat"
+ and spans[presser]["kind"] == "hat"
+ and spans[releaser]["name"] != spans[presser]["name"]
+ and spans[releaser]["end"] == spans[presser]["start"]
+ }
+ edges: list[tuple[float, int, int, int, int, bool, str]] = []
+ for pos in idxs:
+ span = spans[pos]
+ if span["kind"] == "hat":
+ # 同at Hat RELEASE＋PRESS 対は hold(新)→holdEnd(旧) に入替
+ # （新先旧後。中立過渡を出さない）。それ以外は入力順保存
+ # （Hat は RELEASE→PRESS）。種別 0=PRESS（新先）、
+ # 1=RELEASE、2=PRESS（旧・通常）。
+ if pos in swap_pressers:
+ press_kind = 0
+ else:
+ press_kind = 2
+ edges.append(
+ (
+ span["start"],
+ press_kind,
+ span["press_seq"],
+ 0,
+ 0,
+ True,
+ span["name"],
+ )
+ )
+ edges.append(
+ (span["end"], 1, span["rel_seq"], 0, 0, False, span["name"])
+ )
+ else:
+ # 同時刻エッジ順序は入力順序を保存する。ボタンは1行あたり
+ # BUTTON_NAMES 表順で届くため、表順が入力順になる。
+ rank = button_order.get(span["name"], len(button_order))
+ edges.append((span["start"], 2, 0, 0, rank, True, span["name"]))
+ edges.append((span["end"], 1, 0, 0, rank, False, span["name"]))
+ edges.sort(key=lambda edge: (edge[0], edge[1], edge[2], edge[3], edge[4]))
+ grid = {edge[0]: _chord_grid(edge[0], origin) for edge in edges}
+ grid_next = _chord_grid(next_start, origin) if next_start is not None else 0.0
+ times = sorted(grid)
+ by_time: dict[float, list[tuple[float, int, int, int, int, bool, str]]] = {}
+ for edge in edges:
+ by_time.setdefault(edge[0], []).append(edge)
+ held: set[str] = set()
+ for pos, moment in enumerate(times):
+ group = by_time[moment]
+ following = times[pos + 1] if pos + 1 < len(times) else None
+ gap = 0.0
+ if following is not None:
+ gap = max(0.0, round(grid[following] - grid[moment], 2))
+ group_has_release = any(not edge[5] for edge in group)
+ # 同時 PRESS 群の間隔は末尾の hold が担う（先頭担いだと
+ # 「A→A+R」の過渡が次の待ちで延び、同時開始がずれる）。
+ # 同一時刻に離すものが混ざるときは、間隔を hold の WAIT へ載せず
+ # 後ろの kind:wait へ一本化する（二重に待たないため）。
+ hold_chunks = _split_chord_wait(gap) if not group_has_release else []
+ press_edges = [edge for edge in group if edge[5]]
+ last_press = press_edges[-1] if press_edges else None
+ for edge in group:
+ is_press = edge[5]
+ target = edge[6]
+ if is_press:
+ if target in held:
+ raise ValueError(
+ f"同時押しの変換で {target} の二重holdが出ました"
+ "（Keys.hold は重複を黙って捨てるため、"
+ "変換の出口で検出しています）"
+ )
+ held.add(target)
+ chunks = hold_chunks if edge == last_press else []
+ steps.append(
+ {
+ "kind": "hold",
+ "target": target,
+ "wait": chunks[0] if chunks else 0.0,
+ }
+ )
+ for chunk in chunks[1:]:
+ steps.append({"kind": "wait", "wait": chunk})
+ else:
+ held.discard(target)
+ steps.append({"kind": "hold_end", "target": target})
+ if group_has_release and following is not None:
+ for chunk in _split_chord_wait(gap):
+ steps.append({"kind": "wait", "wait": chunk})
+ # 連鎖末尾の間隔（次の成分まで。無ければ既定の待ち）。
+ if next_start is not None:
+ tail = max(0.0, round(grid_next - grid[times[-1]], 2))
+ for chunk in _split_chord_wait(tail):
+ steps.append({"kind": "wait", "wait": chunk})
+ else:
+ steps.append({"kind": "wait", "wait": LAST_WAIT})
+ return steps
+
+
#: スティック操作をクリップとみなす r の閾値（u8 距離）。
#: motion.DEADZONE_R と同じ意味。ここでは座標で判定する。
MOTION_DEADZONE_XY = 8.0
@@ -178,12 +499,124 @@ def __init__(self, use_motion: bool = True) -> None:
self._motion_transport: Any = None
#: 記録を始めた時刻（壁時計）。手順の変換には使わない。
self.started_at: float = 0.0
+ #: 記録中に本体側の操作が混ざったかの latch。一度立ったら次の
+ #: 開始まで倒さない（停止後に読んで警告へ載せるため）。
+ self._contaminated = False
+ #: 調停の棄却件数の土台（human, auto）。初回の見回りで覚え、
+ #: 増えたら混ざったとみなす。開始ごとに捨てる。
+ self._arb_base: tuple[int, int] | None = None
@property
def recording(self) -> bool:
"""いま記録中か。"""
return self._transport is not None
+ @property
+ def contaminated(self) -> bool:
+ """記録中に本体側の操作が混ざったか（latch・読み取り専用）。"""
+ return bool(self._contaminated)
+
+ def observe_owners(
+ self,
+ owners: dict[str, Any] | None,
+ arbitration: dict[str, Any] | None = None,
+ ) -> bool:
+ """所有者の写しと調停の様子を見て、汚染を latch する。
+
+ GUI スレッドの見回りから呼ぶ（sender.getOwners()・getArbitration()
+ の写しを渡す。持たない線では None でよい）。gamepad 以外の所有者
+ が空でなければ汚染とする。調停の棄却件数が増えていても汚染とする。
+ 一度汚染したら次の開始まで倒さない。例外は外へ投げない。
+ """
+ try:
+ if self._contaminated:
+ return True
+ if self._transport is None:
+ return False
+ if owners is not None:
+ try:
+ btn = owners.get("btn", {})
+ hat = owners.get("hat", {})
+ stick = owners.get("stick", {})
+ except Exception:
+ btn, hat, stick = {}, {}, {}
+ if isinstance(btn, dict):
+ for owner, bits in btn.items():
+ if str(owner) == "gamepad":
+ continue
+ try:
+ if int(bits) != 0:
+ self._contaminated = True
+ return True
+ except Exception:
+ # 形が読めないものは「何か居る」とみなす。
+ self._contaminated = True
+ return True
+ if isinstance(hat, dict):
+ for owner, claim in hat.items():
+ if str(owner) == "gamepad":
+ continue
+ # 中立（Sender.POSTURE_HAT_CENTER と同じ 8）の
+ # 置き土産は空とみなす。
+ value = claim
+ if isinstance(claim, (tuple, list)) and len(claim) > 0:
+ value = claim[0]
+ try:
+ if int(value) == 8:
+ continue
+ except Exception:
+ pass
+ self._contaminated = True
+ return True
+ if isinstance(stick, dict):
+ for owner, claim in stick.items():
+ if str(owner) == "gamepad":
+ continue
+ points = claim
+ if isinstance(claim, (tuple, list)) and len(claim) == 2:
+ points = claim[0]
+ if not isinstance(points, dict):
+ # 形が違うものは「何か居る」とみなす。
+ self._contaminated = True
+ return True
+ live = False
+ for point in points.values():
+ try:
+ x, y = point
+ except Exception:
+ live = True
+ break
+ try:
+ # 中立（Sender.POSTURE_CENTER と同じ 128）
+ # だけなら空。
+ if int(x) != 128 or int(y) != 128:
+ live = True
+ break
+ except Exception:
+ live = True
+ break
+ if live:
+ self._contaminated = True
+ return True
+ if isinstance(arbitration, dict):
+ cur: tuple[int, int] | None
+ try:
+ cur = (
+ int(arbitration.get("rejected_human", 0)),
+ int(arbitration.get("rejected_auto", 0)),
+ )
+ except Exception:
+ cur = None
+ if cur is not None:
+ if self._arb_base is None:
+ self._arb_base = cur
+ elif cur[0] > self._arb_base[0] or cur[1] > self._arb_base[1]:
+ self._contaminated = True
+ return True
+ return bool(self._contaminated)
+ except Exception:
+ return bool(self._contaminated)
+
def start(self, transport: Any) -> bool:
"""送信線に記録を付ける。付けたら True。例外は外へ投げない。
@@ -199,6 +632,9 @@ def start(self, transport: Any) -> bool:
if not add(self._logger.feed):
return False
self._transport = transport
+ # 新しい記録では汚染を忘れる（停止後に読む分は stop が保つ）。
+ self._contaminated = False
+ self._arb_base = None
# motion 用の聞き手も同じ線へ付ける。付けなくても press 記録は
# 動く（motion 区間だけ作られない）。
self._motion_transport = None
@@ -250,12 +686,17 @@ def follow(self, transport: Any) -> bool:
except Exception:
return False
- def stop(self) -> list[dict[str, Any]]:
+ def stop(
+ self, warnings: list[dict[str, Any]] | None = None
+ ) -> list[dict[str, Any]]:
"""記録を外し、手順を返す。記録中でなければ空。例外は外へ投げない。
- motion 用の聞き手が付いていて stick 操作があれば、motion 区間の
- 材料（kind が stick/stick2/hold/hold_end/release_stick のもの）を
- 混ぜて返す。無ければ従来の press 変換だけ。
+ stick 操作が無くボタン/Hat の重なりがあれば chord 連鎖（kind が
+ press/hold/hold_end/wait のもの）を返す。stick 操作があれば motion
+ 区間の材料（kind が stick/stick2/hold/hold_end/release_stick のもの）
+ を混ぜて返す。どちらでもなければ従来の press 変換だけ。
+ warnings を渡すと motion 変換の警告（clip_long）を足す。汚染の
+ latch は保ち、contaminated で読める。
"""
try:
transport, self._transport = self._transport, None
@@ -275,9 +716,23 @@ def stop(self) -> list[dict[str, Any]]:
remove_motion(self._motion_logger.feed)
except Exception:
pass
+ # 同時押しは motion より先に試す。stick 無し＋重なり有り（hold系が
+ # 出る）のときだけ chord 手順を返す。stick 有りは従来どおり motion へ。
+ try:
+ chord_events = self._logger.snapshot()
+ if not any(ev.kind == "stick" for ev in chord_events):
+ chord_steps = events_to_chord_steps(chord_events)
+ if any(
+ step.get("kind") in ("hold", "hold_end") for step in chord_steps
+ ):
+ return chord_steps
+ except Exception:
+ pass
if motion_transport is not None:
try:
- motion_steps = build_motion_steps(self._motion_logger.snapshot())
+ motion_steps = build_motion_steps(
+ self._motion_logger.snapshot(), warnings=warnings
+ )
if any(
step.get("kind") in ("stick", "stick2", "hold", "hold_end")
for step in motion_steps
@@ -341,7 +796,8 @@ def _stick_xy_at(
"""スティック座標の時系列 [(side, at, x, y)] を抜く。
PRESS/CHANGE の x・y を使う。RELEASE の座標は中立のため使わない
- （離す直前の値は CHANGE/PRESS 側にある）。at は perf_counter 値。
+ （離す直前の値は CHANGE/PRESS 側にある）。RELEASE の at は抜かず、
+ クリップ分割（_split_clips）の入力として別に集める。at は perf_counter 値。
"""
out: list[tuple[str, float, float, float]] = []
for ev in events:
@@ -356,6 +812,25 @@ def _stick_xy_at(
return out
+def _stick_releases(
+ events: list[InputLog.InputEvent],
+) -> list[tuple[str, float]]:
+ """スティックの解放 [(side, at)] を時刻順で返す。
+
+ RELEASE の座標は中立のため使わず、at だけを使う。RELEASE が無い
+ （停止時に押下中・運転席に残った）側は出さない。クリップ分割で
+ 「離した瞬間」を区切りに使う（離したのに中立へ戻らない残留の修正）。
+ """
+ out: list[tuple[str, float]] = []
+ for ev in events:
+ if ev.kind != "stick" or ev.action != "RELEASE":
+ continue
+ side = "R" if ev.name.endswith("RIGHT") else "L"
+ out.append((side, float(ev.at)))
+ out.sort(key=lambda item: item[1])
+ return out
+
+
def _motion_buttons(
events: list[InputLog.InputEvent], t0: float
) -> list[tuple[float, str, bool]]:
@@ -375,6 +850,7 @@ def _motion_buttons(
def build_motion_steps(
events: list[InputLog.InputEvent],
eps: float = _motion.DEFAULT_EPS,
+ warnings: list[dict[str, Any]] | None = None,
) -> list[dict[str, Any]]:
"""イベント列を motion ブロックの材料へ変換する。純関数。
@@ -386,6 +862,13 @@ def build_motion_steps(
中立かつボタン保持なしが 150ms 継続。押下区間がクリップにかかるボタンは
クリップに含める（境界をまたぐ場合はクリップを広げる）。
クリップ外は events_to_steps と同じ press 変換へ回す。
+ クリップと交差する押下区間は、開始・終了とも延長して吸収する（延長で
+ 新たに交差するものも取り込む。不動点まで繰り返す）。開始側の延長は
+ 押下開始まで、終了側は解放まで。吸収した区間のイベントはクリップ外
+ （press 側）へ出さない。延長後のクリップ長が CLIP_WARN_S を超えたら、
+ warnings（渡したときだけ）へ {"code": "clip_long", "seconds": 長さ} を
+ 1件足す（分割はしない）。hold/hold_end には at_ms（クリップ先頭からの
+ ms。挿入側の WAIT 引継ぎ用）を付ける。
"""
if not events:
return []
@@ -395,25 +878,45 @@ def build_motion_steps(
return [{"kind": "press", **step} for step in events_to_steps(events)]
t0 = min(ev.at for ev in events)
buttons = _motion_buttons(events, t0)
- # クリップ区間を求める（スティックが動いている範囲＋ボタンの跨ぎ＋前後余白）。
- clips = _split_clips(sticks, buttons, t0)
+ # クリップ区間を求める（スティックが動いている範囲＋跨ぎ区間の吸収）。
+ intervals = _press_intervals(events)
+ releases = _stick_releases(events)
+ clips, absorbed = _split_clips(sticks, intervals, releases)
+ if warnings is not None:
+ for clip_start, clip_end in clips:
+ if clip_end - clip_start > CLIP_WARN_S:
+ warnings.append(
+ {"code": "clip_long", "seconds": round(clip_end - clip_start, 2)}
+ )
+ break
if not clips:
return [{"kind": "press", **step} for step in events_to_steps(events)]
steps: list[dict[str, Any]] = []
- # クリップ外の press 変換用に、イベントをクリップ区間で区切る。
+ # クリップ外の press 変換用に、イベントをクリップ区間で区切る。吸収済みは除く。
+ # stick の RELEASE はクリップ側が引き受ける（press 側へ出すと中立へ戻す
+ # press が余計に1個増え、方向が化けることがある）。
for index, (clip_start, clip_end) in enumerate(clips):
# クリップ前の press 区間。
prev_end = clips[index - 1][1] if index > 0 else None
before = [
ev
for ev in events
- if (prev_end is None or ev.at >= prev_end) and ev.at < clip_start
+ if (prev_end is None or ev.at >= prev_end)
+ and ev.at < clip_start
+ and id(ev) not in absorbed
+ and not (ev.kind == "stick" and ev.action == "RELEASE")
]
for step in events_to_steps(before):
steps.append({"kind": "press", **step})
steps.extend(_clip_to_steps(sticks, buttons, t0, clip_start, clip_end, eps))
# 最後のクリップ後の press 区間。
- after = [ev for ev in events if ev.at >= clips[-1][1]]
+ after = [
+ ev
+ for ev in events
+ if ev.at >= clips[-1][1]
+ and id(ev) not in absorbed
+ and not (ev.kind == "stick" and ev.action == "RELEASE")
+ ]
for step in events_to_steps(after):
steps.append({"kind": "press", **step})
return steps
@@ -421,10 +924,22 @@ def build_motion_steps(
def _split_clips(
sticks: list[tuple[str, float, float, float]],
- buttons: list[tuple[float, str, bool]],
- t0: float,
-) -> list[tuple[float, float]]:
- """クリップ区間 [(開始 at, 終了 at)] を返す。at は perf_counter 値。"""
+ intervals: list[_PressSpan],
+ releases: list[tuple[str, float]] | None = None,
+) -> tuple[list[tuple[float, float]], set[int]]:
+ """クリップ区間 [(開始 at, 終了 at)] と吸収した押下イベントの id 集合を返す。
+
+ at は perf_counter 値。クリップと交差する押下区間（端点タッチを含む）は
+ 開始・終了とも延長して吸収し、延長で新たに交差するものも取り込む
+ （不動点まで）。開始側の延長は押下開始まで。RELEASE の無い押下中は
+ 終了を延ばさない。吸収した区間の PRESS/RELEASE はクリップ外へ出さない
+ （二重化防止）。
+
+ スティックの RELEASE 時刻では区間を分割する。離した瞬間の座標は
+ 中立のため座標列に出ず、分けなければ「離して置いた静止」が
+ 「倒し続け」に潰れ、再生で中立へ戻らなくなる（残留）。RELEASE が
+ 無い側の延長吸収は従来どおり行う。
+ """
# スティックが動いている時刻の範囲を求める。
active = [
at
@@ -432,26 +947,59 @@ def _split_clips(
if math.hypot(x - 128.0, y - 128.0) >= MOTION_DEADZONE_XY
]
if not active:
- return []
- start = min(active)
- # 終了: 最後の active から 150ms。ボタン保持があればその解放まで広げる。
- end = max(active) + CLIP_IDLE_S
- if buttons:
- # クリップにかかる押下（開始前から押下・終了後に解放）を含める。
- held: dict[str, float] = {}
- for t_ms, name, on in buttons:
- at = t0 + t_ms / 1000.0
- if on:
- held[name] = at
- else:
- held.pop(name, None)
- # 終了時点でまだ押されているものは、その解放時刻まで広げる。
- # （解放イベントが記録末尾に無い場合は CLIP_IDLE_S で閉じる。）
- for t_ms, name, on in buttons:
- at = t0 + t_ms / 1000.0
- if not on and at > end - CLIP_IDLE_S and at <= end + CLIP_IDLE_S:
- end = max(end, at)
- return [(start, end)]
+ return [], set()
+ all_start = min(active)
+ all_end = max(active) + CLIP_IDLE_S
+ # 分割点: RELEASE 時刻（側不問）。ただし直後に押し直しがある
+ # RELEASE では切らない（倒し直しを別クリップに割ると、単発の倒しが
+ # 座標を持たない press へ落ち、方向が化ける）。切るのは「離した後に
+ # active が無い」RELEASE＝その側の操作が本当に終わったものだけにする。
+ press_ats = sorted(at for _, at, _, _ in sticks)
+ cuts = sorted(
+ {
+ at
+ for _, at in (releases or [])
+ if all_start < at < all_end and not any(p > at for p in press_ats)
+ }
+ )
+ bounds = [all_start, *cuts, all_end]
+ clips: list[tuple[float, float]] = []
+ absorbed: set[int] = set()
+ for seg_start, seg_end in zip(bounds, bounds[1:]):
+ if seg_end <= seg_start:
+ continue
+ seg_active = [at for at in active if seg_start <= at <= seg_end]
+ if not seg_active:
+ # 離した後の静止だけの区間はクリップにしない。次の active
+ # （押し直し）があればそちらで拾う。
+ continue
+ start, end = seg_start, seg_end
+ growing = True
+ while growing:
+ growing = False
+ for _name, _kind, press_at, release_at, press_ev, release_ev in intervals:
+ release_eff = float("inf") if release_at is None else release_at
+ if press_at <= end and release_eff >= start:
+ if press_at < start:
+ if start > all_start:
+ # 前の区切りにかかる押下は吸収だけし、開始は割らない
+ # （跨ぎ区間が2クリップへ分裂して二重holdになるのを防ぐ）。
+ absorbed.add(id(press_ev))
+ if release_ev is not None:
+ absorbed.add(id(release_ev))
+ continue
+ # 最初の区切りは跨ぎ吸収の起点のため開始側へ延長する
+ # （ZL長押し等の警告対象）。
+ start = press_at
+ growing = True
+ if release_at is not None and release_at > end:
+ end = release_at
+ growing = True
+ absorbed.add(id(press_ev))
+ if release_ev is not None:
+ absorbed.add(id(release_ev))
+ clips.append((start, end))
+ return clips, absorbed
def _clip_to_steps(
@@ -571,7 +1119,7 @@ def _clip_to_steps(
pos = 0
for step in steps:
if step["kind"] in ("hold", "hold_end"):
- timed.append((float(step.pop("at_ms")), 0, step))
+ timed.append((float(step["at_ms"]), 0, step))
elif step["kind"] == "release_stick":
# 終端の解放は常に最後（時刻ソートの対象外）。
tail.append(step)
diff --git a/SerialController/services/command_runner.py b/SerialController/services/command_runner.py
index d650146..3dc9138 100644
--- a/SerialController/services/command_runner.py
+++ b/SerialController/services/command_runner.py
@@ -129,6 +129,9 @@ def __init__(
self._kind_of: Callable[[Any], str] | None = None
self._now: Callable[[], datetime] = datetime.now
self._monotonic: Callable[[], float] = time.monotonic
+ # 記録中の開始拒否の判定口。Window が runner と記録側をつなぐ。
+ # 渡さなければ従来どおり（記録中でも開始できる）。
+ self._record_guard: Callable[[], bool] | None = None
# 実行中の履歴の行と開始時刻。終わったら None へ戻す。
self._hist_entry: CommandHistory.HistoryEntry | None = None
self._hist_t0 = 0.0
@@ -211,6 +214,14 @@ def set_history(
self._now = now
self._monotonic = monotonic
+ def set_record_guard(self, guard: Callable[[], bool] | None) -> None:
+ """記録中の開始拒否の判定口を渡す。Window が組み立て後に呼ぶ。
+
+ guard が True を返したら記録中とみなし、開始を断る。None なら
+ 判定しない（従来どおり記録中でも開始できる）。
+ """
+ self._record_guard = guard
+
@property
def current_history_entry(self) -> CommandHistory.HistoryEntry | None:
"""実行中の履歴の行。空きなら None。
@@ -282,13 +293,26 @@ def request_start(self, command: Any, ser: Any) -> None:
logger.warning("Start refused while fenced thread is alive")
return
self._fenced_thread = None
+ # 記録中は通常コマンドの開始を断る。記録への混ざりを防ぐため。
+ # 判定口が無ければ従来どおり通す（新predicate未配線時と同等）。
+ guard = self._record_guard
+ if guard is not None:
+ try:
+ recording = bool(guard())
+ except Exception:
+ recording = False
+ if recording:
+ self._notify(
+ "操作の記録中は開始できません（記録を止めてから試してください）"
+ )
+ logger.warning("Start refused while recording")
+ return
# 二重起動を断る。Tk は同じボタンのコールバックを並行実行しないが、
# F6・パレット・外部呼び出しからも入って来られる。
if self.is_busy():
self._notify("すでにコマンドが動いています")
logger.warning("A command is already running")
return
-
# 旧コードはメッセージを出すだけで先へ進み、None.NAME で落ちていた
if command is None:
self._notify("No commands have been assigned yet.")
diff --git a/SerialController/services/serial_service.py b/SerialController/services/serial_service.py
index 6662595..b20bd9f 100644
--- a/SerialController/services/serial_service.py
+++ b/SerialController/services/serial_service.py
@@ -443,7 +443,7 @@ def reload_keyboard_map(self) -> None:
# -- ゲームパッド -----------------------------------------------------------
def set_gamepad_enabled(
- self, enabled: bool, pad_index: int = 0, deadzone: int = 0
+ self, enabled: bool, pad_index: int = 0, deadzone: int = 1638
) -> str | None:
"""ゲームパッド操作の有効・無効を切り替える。
diff --git a/SerialController/ui/blockly_editor.py b/SerialController/ui/blockly_editor.py
index 8514d05..9a49e0f 100644
--- a/SerialController/ui/blockly_editor.py
+++ b/SerialController/ui/blockly_editor.py
@@ -110,6 +110,13 @@ def __init__(self, session: blockly_run.TrialSession, cls: type) -> None:
#: 本体の操作の記録（マクロ記録）。送信線（host.serial.sender.transport）に
#: 付けて PRESS/RELEASE を溜める。付け外しは GUI スレッドで行う。
_RECORDER = blockly_record.Recorder()
+_RECORD_BLOCKED: Callable[[], str | None] | None = None
+
+
+def _set_record_blocked(predicate: Callable[[], str | None] | None) -> None:
+ """記録の開始可否の判定口を差し替える（開く時・検証用）。"""
+ global _RECORD_BLOCKED
+ _RECORD_BLOCKED = predicate
def _set_run_host(host: Any) -> None:
@@ -146,6 +153,46 @@ def _follow_record_transport() -> None:
_RECORDER.follow(transport)
+def is_recording() -> bool:
+ """いま操作を記録中かを返す（通常コマンド側の開始拒否用）。"""
+ try:
+ return bool(_RECORDER.recording)
+ except Exception:
+ return False
+
+
+def _observe_record_owners() -> None:
+ """記録中に本体側の操作が混ざっていないか見る（汚染 latch 用）。
+
+ GUI スレッドの見回りから呼ぶ。sender の所有者の写しと調停の様子を
+ 渡す。持たない線では握って何もしない（記録自体は続ける）。
+ """
+ if not _RECORDER.recording:
+ return
+ host = _RUN_HOST
+ sender = getattr(getattr(host, "serial", None), "sender", None)
+ owners: dict[str, Any] | None = None
+ arbitration: dict[str, Any] | None = None
+ get_owners = getattr(sender, "getOwners", None)
+ if callable(get_owners):
+ try:
+ got = get_owners()
+ owners = got if isinstance(got, dict) else None
+ except Exception:
+ owners = None
+ get_arb = getattr(sender, "getArbitration", None)
+ if callable(get_arb):
+ try:
+ got_arb = get_arb()
+ arbitration = got_arb if isinstance(got_arb, dict) else None
+ except Exception:
+ arbitration = None
+ try:
+ _RECORDER.observe_owners(owners, arbitration)
+ except Exception:
+ pass
+
+
def drain_gui_calls() -> None:
"""待ち行列の仕事を片付ける。GUI スレッドの見回りから呼ぶ。"""
while True:
@@ -750,6 +797,15 @@ def _attach() -> tuple[bool, str]:
False,
"試し実行中は記録できません（試し実行を止めてから始めてください）",
)
+ blocked = _RECORD_BLOCKED
+ if blocked is not None:
+ try:
+ reason = blocked()
+ except Exception:
+ reason = None
+ if reason:
+ # 通常コマンドの操作まで記録に混ざるため断る。
+ return (False, str(reason))
serial = getattr(host, "serial", None)
sender = getattr(serial, "sender", None)
transport = getattr(sender, "transport", None)
@@ -779,24 +835,35 @@ def _stop_record(self) -> None:
)
return
- def _detach() -> tuple[bool, list[dict[str, Any]]]:
+ def _detach() -> tuple[bool, list[dict[str, Any]], list[Any], bool]:
was = _RECORDER.recording
- steps = _RECORDER.stop()
- return (was, steps if isinstance(steps, list) else [])
+ warnings: list[Any] = []
+ steps = _RECORDER.stop(warnings)
+ contaminated = bool(_RECORDER.contaminated)
+ return (
+ was,
+ steps if isinstance(steps, list) else [],
+ warnings,
+ contaminated,
+ )
try:
- was_recording, steps = _call_on_gui(_detach)
+ was_recording, steps, warnings, contaminated = _call_on_gui(_detach)
except Exception as e:
self._reply(False, f"本体が応答しません: {e}")
return
if not was_recording:
self._reply(False, "記録していません（記録ボタンで始めてください）")
return
- self._reply(
- True,
- f"{len(steps)} 個の操作を記録しました",
- {"steps": steps},
- )
+ message = f"{len(steps)} 個の操作を記録しました"
+ if contaminated:
+ message += "（本体側の操作が混ざった可能性があります）"
+ extra: dict[str, Any] = {"steps": steps}
+ if warnings:
+ extra["warnings"] = warnings
+ if contaminated:
+ extra["contaminated"] = True
+ self._reply(True, message, extra)
def _reply(
self, ok: bool, message: str, extra: dict[str, Any] | None = None
@@ -901,11 +968,14 @@ def open_blockly_editor(
reload_commands: Callable[[], None],
get_frame: Callable[[], bytes | None] | None = None,
run_host: Any = None,
+ record_blocked: Callable[[], str | None] | None = None,
) -> None:
"""エディタを開く。終わったら一覧を作り直す。
run_host は試し実行の司令塔（本体）。start_external_command・stopPlay・
set_running_paused・syncPauseState・runner を持つ。無ければ試し実行しない。
+ record_blocked は記録の開始可否（拒否理由、無ければ None）。無ければ
+ 試し実行中だけを見て、通常コマンド実行中は断らない（従来どおり）。
"""
import tkinter as tk
import tkinter.messagebox as tkmsg
@@ -917,7 +987,7 @@ def open_blockly_editor(
global _GET_FRAME
_GET_FRAME = get_frame
_set_run_host(run_host)
- _stop_server()
+ _set_record_blocked(record_blocked)
global _server, _thread
_server = _bind_server()
port = _server.server_address[1]
@@ -979,6 +1049,10 @@ def poll() -> None:
_follow_record_transport()
except Exception:
pass
+ try:
+ _observe_record_owners()
+ except Exception:
+ pass
sync = getattr(run_host, "syncPauseState", None)
if callable(sync):
try:
diff --git a/SerialController/ui/serial_panel.py b/SerialController/ui/serial_panel.py
index 3c265f0..b79a3bd 100644
--- a/SerialController/ui/serial_panel.py
+++ b/SerialController/ui/serial_panel.py
@@ -101,10 +101,9 @@ class SerialPanelMixin:
cb_use_gamepad: Any
gamepad_label: Any
gamepad_choice: Any
- gamepad_cb: Any
- _gamepad_map: dict[str, int]
gamepad_dz_label: Any
gamepad_dz: Any
+ _gamepad_after: Any
gamepad_dz_spin: Any
cb_left_stick_mouse: Any
cb_right_stick_mouse: Any
@@ -347,10 +346,10 @@ def _build_control_frame(self) -> None:
)
self.gamepad_cb.grid(column=1, row=2, padx="10", pady="2", sticky="ew")
self.gamepad_cb.bind("<>", self._on_gamepad_selected, add="")
- # 遊び（デッドゾーン）。0=なし・GUI と同等。drift 時に上げる。
+ # 遊び（デッドゾーン）。既定1638=最大値の5%。drift 時に上げる。
self.gamepad_dz_label = ttk.Label(self.control_lf, text="遊び: ")
self.gamepad_dz_label.grid(column=0, row=3, padx="10", pady="2", sticky="w")
- self.gamepad_dz = tk.StringVar(value="0")
+ self.gamepad_dz = tk.StringVar(value="1638")
self.gamepad_dz_spin = ttk.Spinbox(
self.control_lf,
textvariable=self.gamepad_dz,
@@ -373,7 +372,9 @@ def _build_control_frame(self) -> None:
self.control_lf.config(text="コントローラ")
self.control_lf.pack(fill="x", padx=5, pady=5)
-
+ self._gamepad_after = None
+ self.refreshGamepads()
+ self._poll_gamepads()
# 仮想コントローラ。タブの中に常に置き、別ウィンドウへも出せる
# （ui/controller_dock.py）。送り先は開くたびに今の物を引く。
dock_area = ttk.Frame(self.tab_controller)
@@ -1116,6 +1117,10 @@ def createControllerWindow(self) -> None:
def closingController(self) -> None:
"""終了時。押しっぱなしを離してから、埋め込みも別窓も消す。"""
+ try:
+ self._cancel_gamepad_patrol()
+ except Exception:
+ pass
dock = getattr(self, "controller_dock", None)
if dock is not None:
dock.shutdown()
@@ -1238,12 +1243,12 @@ def _gamepad_index(self) -> int:
return index if 0 <= index <= 3 else 0
def _gamepad_deadzone(self) -> int:
- """遊びを通常の int で返す。読めない・範囲外は 0。"""
+ """遊びを通常の int で返す。読めない・範囲外は既定1638。"""
try:
value = int(self.settings.gamepad_deadzone.get())
except (TypeError, ValueError, tk.TclError):
- return 0
- return value if 0 <= value <= 8192 else 0
+ return 1638
+ return value if 0 <= value <= 8192 else 1638
def _on_gamepad_dz_changed(self, event: Any = None) -> None:
"""遊びの変更を保存し、有効化中なら作り直して適用する。"""
@@ -1287,7 +1292,8 @@ def refreshGamepads(self) -> None:
"""繋がっているパッド一覧を取り直して選択肢へ入れる。
抜き差ししても選び直せるよう、接続ボタン・チェック切替の
- 際に呼び出す。選んでいた物が残っていれば残す。
+ 際に呼び出す。選んでいた物が残っていれば残す。名前は SDL の
+ 実名があれば実名、XInput のみは汎名になる（PadSource.names）。
"""
try:
from core.pad_source import PadSource
@@ -1297,7 +1303,7 @@ def refreshGamepads(self) -> None:
names = {}
try:
current = self._gamepad_index()
- labels = [f"{i}: {names.get(i, 'XInput パッド')}" for i in range(4)]
+ labels = [f"{i}: {names.get(i, '(未接続)')}" for i in range(4)]
self._gamepad_map = {label: i for i, label in enumerate(labels)}
self.gamepad_cb["values"] = labels
for label, index in self._gamepad_map.items():
@@ -1308,9 +1314,42 @@ def refreshGamepads(self) -> None:
except Exception as e:
logger.warning(f"パッド一覧の更新で例外: {e}")
+ def _poll_gamepads(self) -> None:
+ """抜き差しで一覧と状態行を追従させる画面側巡回（2秒毎）。
+
+ Tk から線への往復はしない。選択肢の名前（実名・汎名・未接続）と
+ 状態行だけを取り直す。操作中（値を読んでいる最中）の上書きで
+ 選択が飛ばないよう、選んでいる番号は保つ（refreshGamepads が
+ 現在値を選び直す）。終了時は _cancel_gamepad_patrol で止める。
+ """
+ try:
+ self.refreshGamepads()
+ except Exception:
+ pass
+ try:
+ self._gamepad_after = self.root.after(2000, self._poll_gamepads)
+ except Exception:
+ self._gamepad_after = None
+
+ def _cancel_gamepad_patrol(self) -> None:
+ """巡回の予約を取り消す。終了・破棄の前に呼ぶ。"""
+ try:
+ after_id = getattr(self, "_gamepad_after", None)
+ except Exception:
+ after_id = None
+ if after_id is None:
+ return
+ try:
+ self.root.after_cancel(after_id)
+ except Exception:
+ pass
+ try:
+ self._gamepad_after = None
+ except Exception:
+ pass
+
def _on_gamepad_selected(self, event: Any = None) -> None:
"""選択肢で選んだパッドへ切り替える。途中でも切り替える。
-
有効化中なら実体を作り直す（番号は生成時に固定のため）。
無効中なら設定だけ残し、次回有効化で使う。
"""
diff --git a/docs/BLOCKLY_EDITOR.md b/docs/BLOCKLY_EDITOR.md
index 11662d0..8b6bcd8 100644
--- a/docs/BLOCKLY_EDITOR.md
+++ b/docs/BLOCKLY_EDITOR.md
@@ -45,16 +45,14 @@
本体で実際に押した操作（画面の仮想コントローラ・キーボード・ゲームパッドなど、送信線へ出た入力）を記録し、「押す」ブロックの並びにして挿入する（マクロ記録）。「操作に合わせた形で入力したい」ときに使う。
-- 「● 記録」（試し実行の並び）を押すと記録が始まり、ボタンが「■ 記録を止める」（赤）に変わる。もう一度押すと止まり、押した操作が `pokecon_press` ブロックの並びになって入る（押していた長さが「長さ」、離してから次を押すまでが「待ち」になる。最後の待ちは0.1秒）。
-- 時間は0.05秒単位に丸める（最小0.05・上限60）。記録開始前から押しっぱなしだったものは入らない。
-- 同時押しの再現はしない。重なった区間は無視し、離した順に並べるだけになる。スティックをぐるりと回した操作も、回転は再現せず倒し始めの向きの1手順になる。
+- 「● 記録」（試し実行の並び）を押すと記録が始まり、ボタンが「■ 記録を止める」（赤）に変わる。もう一度押すと止まり、押した操作がブロックの並びになって入る。重ならない単独押しは `pokecon_press`（押していた長さが「長さ」、離してから次を押すまでが「待ち」、最後の待ちは0.1秒）、重なる同時押しは `pokecon_hold`・`pokecon_holdEnd` の連鎖＋間の `pokecon_wait` になる（例：Aを押したままRを押してAだけ離す操作は、Aの hold → Rの hold → Aの holdEnd → Rの holdEnd の並びで残り、再生時は A → A+R → R → 中立と送られる）。
+- 時間は press系が0.05秒単位に丸める（最小0.05・上限60）。同時押し連鎖の間の待ちは0.01秒単位・0許容（同時開始をずらさないため。どちらも変換器側の規則で、ブロック側の欄に刻み・上限は無い）。記録開始前から押しっぱなしだったものは入らない（件数だけ数える）。停止時に押したままだったものは自動で離す手順を足す（単独は press、連鎖内は holdEnd。足した旨を通知する）。
+- 十字キーの切替（TOP → TOP_RIGHT 等）は中立を挟まず切り替える（新しい向きの hold → 古い向きの holdEnd の順に入れ替える）。ボタンと重ならない十字キー単独のロールも同じく連鎖にする。スティックをぐるりと回した操作も、回転は再現せず倒し始めの向きの1手順になる（スティックを動かした区間は下の motion 記録になる）。
- スティックは8方向に丸める（左スティック上＝`Direction.UP`、十字キー上＝`Hat.TOP`、ボタンは素の名＝`A`）。半倒し・中間角度も一番近い8方向になる。
- スティックを動かした区間は、極座標の waypoint 列（`pokecon_motion` / `pokecon_motion2`・`pokecon_hold` 併用）として記録される。円回し等の途中経路も再現できる。1区間が1ブロック（from → to を Tms で移動、θ は累積角度で +360 が1周）。区間内のボタンは `hold`/`holdEnd`（待ちなし）になり、区間の最後だけ離す（`END` が `RELEASE` のブロック）。手で from/to/T を直せる。
- 入る位置は、選択中の文ブロックがあればその直後（元の後続は最後の新ブロックの後ろへつなぎ直す）、無ければプログラムの `DO` の末尾、プログラムが無ければ空いている所。全体が1つの取り消し単位になる（「元に戻す」1回で全部消える）。
-- 0件のときは「操作が記録されませんでした（本体でボタンを押してから止めてください）」と出て、何も入らない。
-- 試し実行中は記録を始められず、記録中は試し実行できない（画面のボタンだけでなく受け口でも断る。試し実行の操作が記録に混ざらないように）。
- 記録中に本体の通信方式・送信線が切り替わっても、記録は新しい線へ付いていく（GUI の見回りで付け替える）。
-- 受け口は `POST /record/start`・`POST /record/stop`（開始・停止は待ち行列でGUIスレッドへ渡す）。本体と繋がっていない・送信線が無い・記録中の開始・記録していない停止は理由を返す。エディタの窓を閉じると記録中でも外れる。
+- 受け口は `POST /record/start`・`POST /record/stop`（開始・停止は待ち行列でGUIスレッドへ渡す）。本体と繋がっていない・送信線が無い・記録中の開始・記録していない停止・通常コマンド実行中の開始は理由を返す。通常コマンドの実行中は記録を始められず、記録中は通常コマンド・試し実行のどちらも始められない（画面のボタンだけでなく受け口でも断る。自動実行の操作が記録に混ざらないように）。記録中に自動側の所有者が触った形跡があれば、停止時に汚染の警告を出す。エディタの窓を閉じると記録中でも外れる。
## 控え（自動保存）
diff --git a/tests/test_blockly_chord.py b/tests/test_blockly_chord.py
new file mode 100644
index 0000000..004c716
--- /dev/null
+++ b/tests/test_blockly_chord.py
@@ -0,0 +1,448 @@
+"""同時押し（chord連鎖）の記録の検証（実機不要）。
+
+決め打ちの InputEvent 列を services/blockly_record.events_to_chord_steps へ
+渡し、hold/holdEnd の連鎖・待ちの分割・Hat 入替を固定する。at は
+perf_counter 値、duration は RELEASE だけに付ける（既存の作り方を踏襲）。
+"""
+
+from __future__ import annotations
+
+import datetime
+import math
+import time
+from collections import Counter
+from typing import Any
+
+import pytest
+from blockly_node import NEEDS_NODE, run_blockly
+from core import InputLog, Sender
+from core.Keys import Button, Hat, KeyPress
+from fakes import FakeTransport
+from services import blockly_record
+
+pytestmark = NEEDS_NODE
+
+#: 再記録との時刻比較の許し（実 sleep の揺れを見る分）。
+_ROUNDTRIP_TOL_S = 0.15
+
+
+def _event(
+ action: str,
+ kind: str,
+ name: str,
+ at: float,
+ duration: float | None = None,
+) -> InputLog.InputEvent:
+ """Given: 時刻と押下時間だけを持つ決め打ちのイベント列を作る。"""
+ wall = datetime.datetime.now()
+ return InputLog.InputEvent(action, kind, name, at, wall, duration=duration)
+
+
+def _stick_events(
+ t0: float,
+ points: list[tuple[float, float, float]],
+) -> list[InputLog.InputEvent]:
+ """Given: 座標列 [(dt, x, y)] から PRESS/CHANGE を作る。"""
+ evs: list[InputLog.InputEvent] = []
+ for i, (dt, x, y) in enumerate(points):
+ at = t0 + dt
+ deg = math.degrees(math.atan2(128 - y, x - 128))
+ mag = min(1.0, math.hypot(x - 128, y - 128) / 128.0)
+ evs.append(
+ InputLog.InputEvent(
+ "PRESS" if i == 0 else "CHANGE",
+ "stick",
+ "Stick.LEFT",
+ at,
+ datetime.datetime.now(),
+ x=x,
+ y=y,
+ deg=deg,
+ mag=mag,
+ )
+ )
+ return evs
+
+
+def _target_obj(name: str) -> Any:
+ """Given: Button.A / Hat.TOP のような名前 / When: 変換する / Then: KeyPress に渡せる実体を返すこと。"""
+ head, _, tail = name.partition(".")
+ if head == "Button":
+ return getattr(Button, tail)
+ if head == "Hat":
+ return getattr(Hat, tail)
+ raise ValueError(f"想定外の target: {name}")
+
+
+def _row_buttons(rows: list[str]) -> list[int]:
+ """Given: 送信行 / When: 先頭欄を読む / Then: ボタンの姿勢ビットだけを返すこと。"""
+ return [(int(row.split()[0], 16) >> 2) for row in rows]
+
+
+def _row_hats(rows: list[str]) -> list[str]:
+ """Given: 送信行 / When: 次欄を読む / Then: Hat の値だけを返すこと。"""
+ return [row.split()[1] for row in rows]
+
+
+def _replay(steps: list[dict[str, Any]], keys: KeyPress) -> None:
+ """Given: chord 手順 / When: 疑似の送信線へ再生する / Then: hold 待ちだけ待つこと。"""
+ for step in steps:
+ kind = step.get("kind")
+ if kind == "hold":
+ keys.hold(_target_obj(str(step["target"])))
+ time.sleep(float(step.get("wait", 0.0)))
+ elif kind == "hold_end":
+ keys.holdEnd(_target_obj(str(step["target"])))
+ elif kind == "wait":
+ time.sleep(float(step.get("wait", 0.0)))
+ elif kind == "press":
+ obj = _target_obj(str(step["target"]))
+ keys.input(obj)
+ time.sleep(float(step.get("duration", 0.1)))
+ keys.inputEnd(obj)
+ time.sleep(float(step.get("wait", 0.1)))
+
+
+def test_chord_staggered() -> None:
+ """Given: A↓R↓A↑R↑（ずれあり） / When: 変換する / Then: hold 連鎖＋再生順が A→A+R→R→中立になること。"""
+ # Given: ずらした同時押し
+ events = [
+ _event("PRESS", "button", "Button.A", 10.0),
+ _event("PRESS", "button", "Button.R", 10.2),
+ _event("RELEASE", "button", "Button.A", 10.5, 0.5),
+ _event("RELEASE", "button", "Button.R", 10.7, 0.5),
+ ]
+ # When: 変換する
+ steps = blockly_record.events_to_chord_steps(events)
+ # Then: hold A・hold R・holdEnd A・holdEnd R＋wait の並びになる
+ assert [s["kind"] for s in steps] == [
+ "hold",
+ "hold",
+ "hold_end",
+ "wait",
+ "hold_end",
+ "wait",
+ ]
+ assert [s["target"] for s in steps if s["kind"] == "hold"] == [
+ "Button.A",
+ "Button.R",
+ ]
+ assert [s["target"] for s in steps if s["kind"] == "hold_end"] == [
+ "Button.A",
+ "Button.R",
+ ]
+ # Then: R 単独区間（A↑→R↑）は約0.2秒の待ちになる
+ assert abs(float(steps[3]["wait"]) - 0.2) <= 0.02
+ # When: 疑似の送信線へ再生する
+ transport = FakeTransport()
+ sender = Sender.Sender(is_show_serial=False, transport=transport)
+ _replay(steps, KeyPress(sender))
+ # Then: 再生順は A→A+R→R→中立になる
+ bit_a = 1 << InputLog.BUTTON_NAMES.index("Button.A")
+ bit_r = 1 << InputLog.BUTTON_NAMES.index("Button.R")
+ assert _row_buttons(transport.rows) == [bit_a, bit_a | bit_r, bit_r, 0]
+
+
+def test_chord_simultaneous_same_row() -> None:
+ """Given: 同atの PRESS 対 / When: 変換する / Then: 過渡1回＋BUTTON_NAMES 表順になること。"""
+ # Given: 完全同時の PRESS 対（入力順は R→A）
+ events = [
+ _event("PRESS", "button", "Button.R", 10.0),
+ _event("PRESS", "button", "Button.A", 10.0),
+ _event("RELEASE", "button", "Button.R", 10.5, 0.5),
+ _event("RELEASE", "button", "Button.A", 10.5, 0.5),
+ ]
+ # Then: 同時刻の通知材料として件数を数える（2件の同時開始）
+ assert len([ev for ev in events if ev.action == "PRESS" and ev.at == 10.0]) == 2
+ # When: 変換する
+ steps = blockly_record.events_to_chord_steps(events)
+ holds = [s for s in steps if s["kind"] == "hold"]
+ # Then: 決定的順序（BUTTON_NAMES 表順。A が R より前）になる
+ assert [s["target"] for s in holds] == ["Button.A", "Button.R"]
+ # Then: 過渡1回（先頭 hold の待ちは 0 で同時開始がずれない）
+ assert holds[0]["wait"] == 0.0
+
+
+def test_chord_release_gap() -> None:
+ """Given: A↑→0.1秒→R↑ / When: 変換する / Then: R 単独区間が約0.1秒の kind:wait になること。"""
+ # Given: 離す側だけが 0.1 秒ずれる重なり
+ events = [
+ _event("PRESS", "button", "Button.A", 10.0),
+ _event("PRESS", "button", "Button.R", 10.2),
+ _event("RELEASE", "button", "Button.A", 10.5, 0.5),
+ _event("RELEASE", "button", "Button.R", 10.6, 0.4),
+ ]
+ # When: 変換する
+ steps = blockly_record.events_to_chord_steps(events)
+ kinds = [s["kind"] for s in steps]
+ # Then: holdEnd A→wait→holdEnd R の順に R 単独区間が残る
+ assert kinds == ["hold", "hold", "hold_end", "wait", "hold_end", "wait"]
+ gap = steps[kinds.index("hold_end") + 1]
+ assert gap["kind"] == "wait"
+ assert abs(float(gap["wait"]) - 0.1) <= 0.02
+
+
+def test_chord_hat_switch_no_neutral() -> None:
+ """Given: A 長押し中の Hat 切替 / When: 変換・再生する / Then: hold(新)→holdEnd(旧) で中立過渡がないこと。"""
+ # Given: A を押したまま Hat.TOP→TOP_RIGHT へ切り替える
+ events = [
+ _event("PRESS", "button", "Button.A", 10.0),
+ _event("PRESS", "hat", "Hat.TOP", 10.2),
+ _event("RELEASE", "hat", "Hat.TOP", 10.4, 0.2),
+ _event("PRESS", "hat", "Hat.TOP_RIGHT", 10.4),
+ _event("RELEASE", "hat", "Hat.TOP_RIGHT", 10.6, 0.2),
+ _event("RELEASE", "button", "Button.A", 10.8, 0.8),
+ ]
+ # When: 変換する
+ steps = blockly_record.events_to_chord_steps(events)
+ order = [(s["kind"], s.get("target")) for s in steps]
+ # Then: 新（TOP_RIGHT）の hold が旧（TOP）の holdEnd より先になる
+ assert order.index(("hold", "Hat.TOP_RIGHT")) < order.index(("hold_end", "Hat.TOP"))
+ # When: 疑似の送信線へ再生する
+ transport = FakeTransport()
+ sender = Sender.Sender(is_show_serial=False, transport=transport)
+ _replay(steps, KeyPress(sender))
+ # Then: Hat が有効な区間に中立（8）の行が混ざらない
+ hats = _row_hats(transport.rows)
+ first = next(i for i, hat in enumerate(hats) if hat != str(int(Hat.CENTER)))
+ last = max(i for i, hat in enumerate(hats) if hat != str(int(Hat.CENTER)))
+ assert all(hat != str(int(Hat.CENTER)) for hat in hats[first : last + 1])
+
+
+def test_chord_touch_is_not_chord() -> None:
+ """Given: A↑B↓が同at（端点タッチ） / When: 変換する / Then: 連鎖化せず press 2個になること。"""
+ # Given: 端だけが触れる前後操作（重なりなし）
+ events = [
+ _event("PRESS", "button", "Button.A", 10.0),
+ _event("RELEASE", "button", "Button.A", 10.5, 0.5),
+ _event("PRESS", "button", "Button.B", 10.5),
+ _event("RELEASE", "button", "Button.B", 11.0, 0.5),
+ ]
+ # When: 変換する
+ steps = blockly_record.events_to_chord_steps(events)
+ # Then: hold 化せず press 2個のままになる
+ assert [s["kind"] for s in steps] == ["press", "press"]
+ assert [s["target"] for s in steps] == ["Button.A", "Button.B"]
+
+
+def test_chord_stop_mid_hold() -> None:
+ """Given: 停止時に押下中 / When: 変換する / Then: 連鎖内は holdEnd・孤立は press で hold 残留がないこと。"""
+ # Given: 連鎖の途中で B を押したまま止める
+ chain = [
+ _event("PRESS", "button", "Button.A", 10.0),
+ _event("PRESS", "button", "Button.B", 10.1),
+ _event("RELEASE", "button", "Button.A", 10.4, 0.4),
+ ]
+ # When: 停止時刻で変換する
+ chained = blockly_record.events_to_chord_steps(chain, stop_at=11.0)
+ holds = Counter(s["target"] for s in chained if s["kind"] == "hold")
+ ends = Counter(s["target"] for s in chained if s["kind"] == "hold_end")
+ # Then: 連鎖内は holdEnd で閉じ、hold だけが残らない
+ assert holds == ends == Counter({"Button.A": 1, "Button.B": 1})
+ # Given: 孤立した押下を止める
+ alone = [_event("PRESS", "button", "Button.X", 20.0)]
+ # When: 停止時刻で変換する
+ single = blockly_record.events_to_chord_steps(alone, stop_at=20.5)
+ # Then: 孤立は press（duration＝停止−押下）で hold は出ない
+ assert [s["kind"] for s in single] == ["press"]
+ assert single[0]["target"] == "Button.X"
+ assert single[0]["duration"] == 0.5
+
+
+def test_chord_stale_release_dropped() -> None:
+ """Given: 開始前押下の RELEASE / When: 変換する / Then: 捨てて残りを press にすること。"""
+ # Given: PRESS の無い RELEASE＋通常操作
+ events = [
+ _event("RELEASE", "button", "Button.A", 10.0, 0.1),
+ _event("PRESS", "button", "Button.B", 10.5),
+ _event("RELEASE", "button", "Button.B", 11.0, 0.5),
+ ]
+ # When: 変換する
+ steps = blockly_record.events_to_chord_steps(events)
+ # Then: 開始前からの RELEASE は捨てられる
+ assert [s["target"] for s in steps] == ["Button.B"]
+ assert [s["kind"] for s in steps] == ["press"]
+
+
+def test_chord_long_wait_split() -> None:
+ """Given: 75秒相当の間隔 / When: 変換・挿入する / Then: hold/WAIT 側は 60＋15 に割れ pokecon_wait は1個のままなこと。"""
+ # Given: 75秒重なる同時押し（R 持ち時間が 75秒）
+ events = [
+ _event("PRESS", "button", "Button.A", 10.0),
+ _event("PRESS", "button", "Button.R", 10.0),
+ _event("RELEASE", "button", "Button.A", 85.0, 75.0),
+ _event("RELEASE", "button", "Button.R", 85.2, 75.2),
+ ]
+ # When: 変換する
+ steps = blockly_record.events_to_chord_steps(events)
+ # Then: hold の WAIT 欄（上限60）に収まるよう 60＋15 に割れる
+ assert steps[1] == {"kind": "hold", "target": "Button.R", "wait": 60.0}
+ assert steps[2] == {"kind": "wait", "wait": 15.0}
+ # When: 75秒の待ちを挿入する
+ out = run_blockly(
+ """
+const ws = new Blockly.Workspace();
+const made = Blockly.PokeconEditor.insertSteps(ws, [
+ { kind: 'wait', wait: 75 },
+], null);
+const waits = made.filter((b) => b.type === 'pokecon_wait');
+done({ n: made.length, types: made.map((b) => b.type),
+ sec: waits.length ? waits[0].getFieldValue('SEC') : null });
+"""
+ )
+ # Then: pokecon_wait（SEC 上限3600）は1個のまま 75秒を持つ
+ assert out["types"] == ["pokecon_wait"]
+ assert float(out["sec"]) == 75
+
+
+def test_hat_roll_alone_no_neutral() -> None:
+ """Given: Hat 単独ロール / When: 変換・再生する / Then: 入替で中立行がなく最後の holdEnd で初めて中立になること。"""
+ # 安全性根拠の回帰試験。Hat 切替の中立過渡なしは、Keys.hold の
+ # holdButton 残留ガードと Keys.inputEnd の hold_hat 復元に依存する
+ # （参照は関数名で行い、行番号では行わないこと）。
+ # Given: 十字キーだけの TOP→TOP_RIGHT ロール
+ events = [
+ _event("PRESS", "hat", "Hat.TOP", 10.0),
+ _event("RELEASE", "hat", "Hat.TOP", 10.3, 0.3),
+ _event("PRESS", "hat", "Hat.TOP_RIGHT", 10.3),
+ _event("RELEASE", "hat", "Hat.TOP_RIGHT", 10.6, 0.3),
+ ]
+ # When: 変換する
+ steps = blockly_record.events_to_chord_steps(events)
+ order = [(s["kind"], s.get("target")) for s in steps]
+ # Then: 連鎖化し、新の hold が旧の holdEnd より先になる（中立過渡なし）
+ assert [kind for kind, _ in order][:2] == ["hold", "hold"]
+ assert order.index(("hold", "Hat.TOP_RIGHT")) < order.index(("hold_end", "Hat.TOP"))
+ # When: 疑似の送信線へ再生する
+ transport = FakeTransport()
+ sender = Sender.Sender(is_show_serial=False, transport=transport)
+ keys = KeyPress(sender)
+ keys.hold(Hat.TOP)
+ keys.hold(Hat.TOP_RIGHT)
+ # Then: 旧を離しても新が残り、中立へ落ちない
+ keys.holdEnd(Hat.TOP)
+ assert sender.getPosture()["hat"] == int(Hat.TOP_RIGHT)
+ # Then: 新の holdEnd で初めて中立に戻る
+ keys.holdEnd(Hat.TOP_RIGHT)
+ assert sender.getPosture()["hat"] == int(Hat.CENTER)
+
+
+def test_chord_double_hold_raises() -> None:
+ """Given: 同一 target の二重 hold / When: 変換する / Then: 警告でなく ValueError になること。"""
+ # Given: 同一ボタンの重なり（離す前に同じボタンを押す）
+ doubled_button = [
+ _event("PRESS", "button", "Button.A", 10.0),
+ _event("PRESS", "button", "Button.A", 10.1),
+ _event("RELEASE", "button", "Button.A", 10.5, 0.5),
+ _event("RELEASE", "button", "Button.A", 10.6, 0.5),
+ ]
+ # When / Then: 変換すると ValueError になる
+ with pytest.raises(ValueError):
+ blockly_record.events_to_chord_steps(doubled_button)
+ # Given: 同一 Hat の重なり
+ doubled_hat = [
+ _event("PRESS", "hat", "Hat.TOP", 10.0),
+ _event("PRESS", "hat", "Hat.TOP", 10.1),
+ _event("RELEASE", "hat", "Hat.TOP", 10.5, 0.5),
+ _event("RELEASE", "hat", "Hat.TOP", 10.6, 0.5),
+ ]
+ # When / Then: 変換すると ValueError になる
+ with pytest.raises(ValueError):
+ blockly_record.events_to_chord_steps(doubled_hat)
+
+
+def test_clip_extension_warns() -> None:
+ """Given: 長押し混在（ZL 跨ぎ） / When: 変換する / Then: 1クリップ＋warnings へ clip_long 1件になること。"""
+ # Given: 短いスティック操作を跨ぐ ZL 長押し
+ t0 = 10.0
+ events = _stick_events(t0, [(0.0, 200, 60), (0.1, 200, 60)])
+ events.append(_event("PRESS", "button", "Button.ZL", 9.5))
+ events.append(_event("RELEASE", "button", "Button.ZL", 29.0, 19.5))
+ warnings: list[dict[str, Any]] = []
+ # When: 変換する
+ steps = blockly_record.build_motion_steps(events, warnings=warnings)
+ # Then: 延長で1クリップにまとまり、警告が1件だけ出る
+ assert sum(1 for s in steps if s.get("kind") == "release_stick") == 1
+ assert len(warnings) == 1
+ assert warnings[0]["code"] == "clip_long"
+ assert float(warnings[0]["seconds"]) > blockly_record.CLIP_WARN_S
+ # Then: 吸収した ZL はクリップ外の press へ出ない
+ assert all(
+ s.get("target") != "Button.ZL" for s in steps if s.get("kind") == "press"
+ )
+
+
+def test_chord_roundtrip_no_neutral_rows() -> None:
+ """Given: chord 手順 / When: 疑似線で再生し再記録する / Then: エッジ順一致・時刻誤差内・Hat 切替で中立行なしになること。"""
+ # Given: A 長押し中の Hat 切替（決め打ちのイベント列）
+ origin = [
+ _event("PRESS", "button", "Button.A", 10.0),
+ _event("PRESS", "hat", "Hat.TOP", 10.2),
+ _event("RELEASE", "hat", "Hat.TOP", 10.4, 0.2),
+ _event("PRESS", "hat", "Hat.TOP_RIGHT", 10.4),
+ _event("RELEASE", "hat", "Hat.TOP_RIGHT", 10.6, 0.2),
+ _event("RELEASE", "button", "Button.A", 10.8, 0.8),
+ ]
+ steps = blockly_record.events_to_chord_steps(origin)
+ # When: 疑似の送信線へ再生しながら tap 相当の再記録を取る
+ transport = FakeTransport()
+ sender = Sender.Sender(is_show_serial=False, transport=transport)
+ rec = blockly_record.Recorder(use_motion=False)
+ assert rec.start(transport) is True
+ try:
+ _replay(steps, KeyPress(sender))
+ again = rec._logger.snapshot()
+ finally:
+ rec.stop()
+ # Then: エッジ（action/kind/name）の並び順が一致する
+ assert [(ev.action, ev.kind, ev.name) for ev in again] == [
+ (ev.action, ev.kind, ev.name) for ev in origin
+ ]
+ # Then: 相対時刻が許容誤差内に収まる
+ for old, new in zip(origin, again):
+ assert (
+ abs(
+ (float(new.at) - float(again[0].at))
+ - (float(old.at) - float(origin[0].at))
+ )
+ <= _ROUNDTRIP_TOL_S
+ )
+ # Then: Hat 切替で中立行が出ない
+ hats = _row_hats(transport.rows)
+ first = next(i for i, hat in enumerate(hats) if hat != str(int(Hat.CENTER)))
+ last = max(i for i, hat in enumerate(hats) if hat != str(int(Hat.CENTER)))
+ assert all(hat != str(int(Hat.CENTER)) for hat in hats[first : last + 1])
+
+
+def test_contamination_latch() -> None:
+ """Given: 記録中の所有者 / When: gamepad 以外を見る / Then: 短時間でも contaminated がラッチすること。"""
+ # Given: 記録中
+ rec = blockly_record.Recorder()
+ transport = FakeTransport()
+ assert rec.start(transport) is True
+ try:
+ # When: gamepad だけなら汚れない
+ assert (
+ rec.observe_owners(
+ {"btn": {"gamepad": 4}, "hat": {"gamepad": 8}, "stick": {}}
+ )
+ is False
+ )
+ assert rec.contaminated is False
+ # When: 短時間だけ本体側の所有者が混ざる
+ assert (
+ rec.observe_owners(
+ {"btn": {"gamepad": 4, "script": 4}, "hat": {}, "stick": {}}
+ )
+ is True
+ )
+ # Then: すぐ居なくなってもラッチし、見逃さない
+ assert (
+ rec.observe_owners(
+ {"btn": {"gamepad": 4}, "hat": {"gamepad": 8}, "stick": {}}
+ )
+ is True
+ )
+ assert rec.contaminated is True
+ finally:
+ rec.stop()
diff --git a/tests/test_motion_record.py b/tests/test_motion_record.py
index 104da8d..3e7faea 100644
--- a/tests/test_motion_record.py
+++ b/tests/test_motion_record.py
@@ -104,8 +104,69 @@ def test_circle_with_button_produces_hold_and_sticks() -> None:
assert any(s["t_ms"] <= 60 for s in sticks)
# 終端に解放材料がある。
assert steps[-1] == {"kind": "release_stick", "side": "L"}
- # T に丸めなし（整数 ms のまま）。
- assert all(isinstance(s["t_ms"], int) for s in sticks)
+
+
+def test_mid_clip_release_returns_to_neutral() -> None:
+ """Given: 途中で離して置くスティック / When: 変換→再生する / Then: 離した後は中立に戻り最後まで残らないこと。"""
+ import threading
+
+ from core import Sender
+ from core.CommandOperate import OperateMixin
+ from core.Keys import KeyPress, Stick
+ from fakes import FakeTransport
+
+ class _Host(OperateMixin):
+ """再生の台。待ちは実時間で回す。"""
+
+ def __init__(self, keys: KeyPress) -> None:
+ self.keys = keys
+ self.alive = True
+ self._stop_event = threading.Event()
+ self._resume_event = threading.Event()
+ self._resume_event.set()
+
+ def _cleanup(self) -> None:
+ return None
+
+ def _pausedSeconds(self) -> float:
+ return 0.0
+
+ # Given: 上へ倒して0.2秒で離し、置いたまま0.5秒で押し直す操作
+ t0 = 12000.0
+ evs = _stick_events(t0, [(0.0, 128, 1)], release_at=None)
+ evs.append(
+ InputLog.InputEvent(
+ "RELEASE",
+ "stick",
+ "Stick.LEFT",
+ t0 + 0.2,
+ datetime.datetime.now(),
+ duration=0.2,
+ deg=90.0,
+ mag=1.0,
+ max_mag=1.0,
+ )
+ )
+ steps = br.build_motion_steps(sorted(evs, key=lambda e: e.at))
+ # When: 変換した手順をそのまま再生する
+ transport = FakeTransport()
+ sender = Sender.Sender(is_show_serial=False, transport=transport)
+ host = _Host(KeyPress(sender))
+ for step in steps:
+ if step.get("kind") == "stick":
+ host.stick_move(
+ Stick.LEFT,
+ step["from"][0],
+ step["from"][1],
+ step["to"][0],
+ step["to"][1],
+ step["t_ms"],
+ )
+ elif step.get("kind") == "release_stick":
+ host.stick_release(Stick.LEFT)
+ # Then: 再生後は中立に戻り、倒しっぱなしが残らない
+ posture = sender.getPosture()
+ assert (posture["lx"], posture["ly"]) == (128, 128)
def test_no_stick_falls_back_to_press() -> None: