From 24c21a146050abb09ed05cf5cec0dff410fb59ed Mon Sep 17 00:00:00 2001 From: moi_poke Date: Sat, 10 Oct 2026 19:17:35 +0900 Subject: [PATCH 1/2] =?UTF-8?q?feat(blockly):=20=E5=90=8C=E6=99=82?= =?UTF-8?q?=E6=8A=BC=E3=81=97=E8=A8=98=E9=8C=B2=E3=81=AEhold=E9=80=A3?= =?UTF-8?q?=E9=8E=96=E5=8C=96=E3=81=A8=E3=82=B9=E3=83=86=E3=82=A3=E3=83=83?= =?UTF-8?q?=E3=82=AF=E9=80=94=E4=B8=AD=E8=A7=A3=E6=94=BE=E3=81=AE=E5=86=8D?= =?UTF-8?q?=E7=8F=BE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - events_to_chord_steps追加: 重なり区間だけhold/holdEnd連鎖＋holdEnd間wait、孤立はpressのまま - Hat切替はhold(新)→holdEnd(旧)入替で中立過渡なし、単独ロールも連鎖化 - motion両側延長(不動点・10秒超でclip_long警告)＋RELEASE時刻でクリップ分割し途中解放を再現 - 二重holdはValueErrorで検出、停止時押下中は自動解放＋通知材料 - 挿入側にkind:wait対応＋hold WAIT引継ぎ、記録と自動実行の双方向排他＋汚染ラッチ - パッド一覧の実名表示＋抜き差し巡回(2秒)、デッドゾーン既定1638(5%) - 折り返し試験ほか14件追加、既存の保存形式・生成器・TOOLBOXは不変 --- SerialController/Gamepad.py | 4 +- SerialController/Menubar.py | 5 + SerialController/Settings.py | 2 +- SerialController/Window.py | 13 + SerialController/assets/blockly/editor.html | 4 +- .../assets/blockly/pokecon_editor.js | 31 +- SerialController/config.py | 8 +- SerialController/core/pad_source.py | 17 +- SerialController/services/blockly_record.py | 626 ++++++++++++++++-- SerialController/services/command_runner.py | 26 +- SerialController/services/serial_service.py | 2 +- SerialController/ui/blockly_editor.py | 94 ++- SerialController/ui/serial_panel.py | 61 +- docs/BLOCKLY_EDITOR.md | 10 +- tests/test_blockly_chord.py | 448 +++++++++++++ tests/test_motion_record.py | 65 +- 16 files changed, 1329 insertions(+), 87 deletions(-) create mode 100644 tests/test_blockly_chord.py diff --git a/SerialController/Gamepad.py b/SerialController/Gamepad.py index 5353a99f..68ac2523 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 0ea01352..4235c22c 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 0f4faebb..809cf5fb 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 f598b980..8c6206e7 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 c1b4567a..80943f84 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 fd29ab6f..4652f3d1 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 94d61c58..d2ff6240 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 fe46a2a8..5637a1e1 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/services/blockly_record.py b/SerialController/services/blockly_record.py index 85efa72f..0d6e5261 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 d6501463..3dc91380 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 66625953..b20bd9f8 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 8514d053..9a49e0fe 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 3c265f02..b79a3bdb 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 11662d08..8b6bcd80 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 00000000..004c7163 --- /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 104da8d7..3e7faea0 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: From c3dd9d4e59abf4f419439926e8789f142c872877 Mon Sep 17 00:00:00 2001 From: moi_poke Date: Sat, 10 Oct 2026 22:45:33 +0900 Subject: [PATCH 2/2] =?UTF-8?q?fix(live):=20worker=E8=B5=B7=E5=8B=95?= =?UTF-8?q?=E3=81=AE=E7=A2=BA=E8=AA=8D=E3=81=A8=E7=99=BB=E9=8C=B2=E3=82=92?= =?UTF-8?q?=E4=B8=80=E3=81=A4=E3=81=AE=E8=87=A8=E7=95=8C=E5=8C=BA=E9=96=93?= =?UTF-8?q?=E3=81=AB=E3=81=97=E4=BA=8C=E9=87=8D=E8=B5=B7=E5=8B=95=E3=82=92?= =?UTF-8?q?=E9=98=B2=E3=81=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- SerialController/core/serial/sender.py | 38 +++++++++++++++----------- 1 file changed, 22 insertions(+), 16 deletions(-) diff --git a/SerialController/core/serial/sender.py b/SerialController/core/serial/sender.py index 1f87db19..1a004b77 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 の未送出を破棄する。