1028 lines
48 KiB
Python
1028 lines
48 KiB
Python
"""Convert a `blocks_studio_version: 2` YAML document into a Runtime 3 IR.
|
|
|
|
Block-to-IR mapping
|
|
-------------------
|
|
|
|
- `sceneStart` → opens a new step (`STEP_<params.id>`); first one
|
|
becomes `entry_step_id`.
|
|
- `sceneEnd` → closes the current step (no transition emitted).
|
|
- `sceneGoto` → transition `event_type: action`, `event_name: goto`,
|
|
`target_step_id: STEP_<params.target>`.
|
|
- `sceneBranch` → two transitions `branch_true` / `branch_false`.
|
|
- `npcSay` → action `{kind: tts_say, text}`.
|
|
- `npcWaitResponse` → action `{kind: wait_user_voice, timeout_ms}`.
|
|
- `npcIntentMatch` → transition `event_type: action`,
|
|
`event_name: intent:<intent>`,
|
|
`target_step_id: STEP_<then>`.
|
|
- Phase D4 « transitions réelles » — blocs « Quand … → aller à [target] » ;
|
|
chacun émet UNE transition avec le vrai `event_type` (∈ EVENT_TYPES), même
|
|
construction `STEP_<target>` que sceneGoto. Aucune extension de schéma.
|
|
- `onButton` → `{event_type: button, event_name: <button>}`.
|
|
- `onAudioDone` → `{event_type: audio_done, event_name: "done"}`.
|
|
- `onUnlock` → `{event_type: unlock, event_name: <source>}`.
|
|
- `onSerial` → `{event_type: serial, event_name: <token>}`.
|
|
- `onEspnow` → `{event_type: espnow, event_name: <command>}`.
|
|
- `onPuzzleSolved`→ `{event_type: unlock, event_name: "puzzle:<id>"}`
|
|
(le « puzzle résolu » se modélise comme un unlock nommé,
|
|
EVENT_TYPES n'ayant pas d'event puzzle_state dédié).
|
|
- `onCodeAssembled`→ `{event_type: unlock, event_name: "code:<code>"}`
|
|
(le « code assemblé » se modélise comme un unlock nommé,
|
|
même famille que onPuzzleSolved — complète le set D4).
|
|
- `hwServo` → action `{kind: hw_servo, channel, angle}`.
|
|
- `hwReadQR` → action `{kind: qr_expect, value}` (+ serial transition).
|
|
- `hwLEDPattern` → action `{kind: led_pattern, pattern}`.
|
|
- `hwSoundPlay` → action `{kind: sound_play, asset}` (legacy, opaque).
|
|
- `mediaPlay` → action `{kind: sound_play, path, loop}` (phase D3,
|
|
chemin SD réel). Le `board` est posé par
|
|
`_tag_actions_board` si l'action est dans un
|
|
`hwOnBoard`. Une action `sound_play` porte donc soit
|
|
`asset` (legacy) soit `path` (réel).
|
|
- `logicScore` → action `{kind: score_add, delta}`.
|
|
- `logicSetVar` → action `{kind: set_var, name, value}`.
|
|
- `logicTimer` → transition `event_type: timer`, `after_ms`,
|
|
`target_step_id` = next step head.
|
|
- `logicIf` **with body / else slots** → action `{kind: condition, expr,
|
|
then: [...sub-actions...], else: [...]}` — the slot
|
|
sub-chains are walked recursively and serialised
|
|
as nested actions. Branching stays **local to the
|
|
current step** as per runtime3 spec. `expr` is the
|
|
branched reporter (value input `condition`,
|
|
serialised via `_reporter_to_expr`) or the opaque
|
|
`condition` text param (backward-compat).
|
|
- `loopRepeat` → action `{kind: loop, mode: repeat, times, body:[…]}`.
|
|
- `loopWhile` → action `{kind: loop, mode: while, expr, body:[…]}`
|
|
where `expr` is a branched reporter or the
|
|
`condition_text` text param.
|
|
- Reporter blocks (`opCompare`/`opBool`/`opNot`/`opArith`/`valVar`/`valScore`/
|
|
`valNumber`/`valText`) never emit actions; they are
|
|
value-input operands resolved to an expression
|
|
string by `_reporter_to_expr`. E.g.
|
|
compare(score,'>',5) → "score > 5";
|
|
and(a,b) → "(a) and (b)"; var('pct') → "pct".
|
|
- Vague 2 « Maître du jeu » (game-master adaptatif) — actions de pilotage,
|
|
jamais des transitions :
|
|
- `phase` → action `{kind: gm_phase, id, duration_minutes, trigger}`.
|
|
- `gmRule` → action `{kind: gm_rule, expr, action}` où `expr` est le
|
|
reporter branché (value input `condition`) ou le texte
|
|
de repli, et `action` ∈ proactive_hint/skip_to_next/
|
|
add_bonus_puzzle.
|
|
- `npcSetMood` → action `{kind: npc_mood, mood}`.
|
|
- `giveHint` → action `{kind: give_hint, level, hint_key}`.
|
|
- `proactiveHint` → action `{kind: proactive_hint, interval_ms, hint_key}`.
|
|
- `setParcours` → action `{kind: set_parcours, profile, puzzles:[…]}`.
|
|
- `scoreBonus` → action `{kind: score_bonus, expr, points}` (expr =
|
|
reporter branché ou texte, comme gmRule).
|
|
- `scorePenalty` → action `{kind: score_penalty, penalty_kind, points}`.
|
|
- `scene` → step-level `scene` object (master display:
|
|
title/subtitle/symbol/effect) — NOT an action.
|
|
- `puzzleQR` / `puzzleSound` → step-level `puzzle` object (local puzzle
|
|
arming, P3 QR / P1 sound) — NOT an action.
|
|
Values are clamped/truncated (with warnings) to the
|
|
strict gateway limits of runtime3_common
|
|
_validate_step_puzzle/_scene so the emitted IR
|
|
always validates. One of each per step; a second
|
|
block overwrites the first with a warning.
|
|
- Vague 3 « énigmes manquantes » — toutes step-level `puzzle` objects (pas
|
|
des actions), même contrainte d'un seul `puzzle` par step :
|
|
- `puzzleLED` → puzzle {type: led, pattern:[…], fragment}.
|
|
- `puzzleRadio` → puzzle {type: radio, frequency, tolerance, fragment}.
|
|
- `puzzleMorse` → puzzle {type: morse, message, mode, fragment}.
|
|
- `puzzleNFC` → puzzle {type: nfc, tags:[…], ordered, fragment}.
|
|
- `puzzleCoffre` → puzzle {type: keypad, length, fragment}.
|
|
- Phase D2 « binding carte » :
|
|
- `puzzleQR`/`puzzleSound` → step-level `puzzle` portant `board` (défaut
|
|
master, validé contre boards.yaml).
|
|
- `hwOnBoard` → bloc contexte (c-block) : n'émet AUCUNE action propre ;
|
|
tague chaque action de son slot `body` avec son
|
|
`board` (récursif dans then/else/body). Le board le
|
|
plus proche gagne si des `hwOnBoard` sont imbriqués.
|
|
- `hwNFCRead` → action `{kind: nfc_read, expected, var, timeout_ms}`.
|
|
- Vague 4 « assemblage du code / coffre » — actions :
|
|
- `codeAssembly` → action `{kind: code_assembly, length,
|
|
fragments:[{puzzle, positions:[…]}, …]}` où le
|
|
champ texte `fragments` (« P1_SON:1-2, P2_CIRCUIT:3 »)
|
|
est parsé en mapping puzzle→positions.
|
|
- `codeFragment` → action `{kind: code_fragment, puzzle, positions:[…]}`.
|
|
- `coffreVerify` → action `{kind: coffre_verify, id, on_success}`.
|
|
|
|
Slot chains never themselves emit new steps — only top-level `sceneStart` does.
|
|
A `sceneStart` inside a slot is treated as an inline error (warning emitted,
|
|
block ignored) since branches must stay local.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Iterable
|
|
import yaml
|
|
|
|
|
|
SCHEMA_VERSION = "zacus.runtime3.v1"
|
|
|
|
# Vague 1 — reporter (value) block kinds. They carry no statement connections;
|
|
# they are resolved to an expression string and never emit actions/transitions
|
|
# nor open chains. Mirrors REPORTER_KINDS in the Atelier exporter and the
|
|
# serializeReporter() helper in @zacus/shared/blocks-doc.ts.
|
|
_REPORTER_KINDS = {
|
|
"opCompare", "opBool", "opNot", "opArith",
|
|
"valVar", "valScore", "valNumber", "valText",
|
|
}
|
|
|
|
# Phase D1 — « DSL honnête » : blocs HARDWARE FUTUR. Ils compilent toujours en
|
|
# IR exactement comme avant (aucune sémantique changée), mais aucun executor ne
|
|
# les joue on-device. La compilation émet un warning non bloquant pour chacun.
|
|
# Kinds SANS le préfixe `zacus_` (les nodes YAML portent le kind nu, cf.
|
|
# `node.get("kind")`). Miroir de ZACUS_FUTURE_HW_KINDS dans
|
|
# frontend-v3/packages/shared/blockly/zacus_blocks.mjs.
|
|
FUTURE_HW_KINDS = {
|
|
"hwLCDText", "hwLCDClear", "hwLCDImage", "hwLCDTouchWait",
|
|
"hwServo", "hwReadQR", "hwLEDPattern", "hwBuzzerTone",
|
|
"hwRelay", "hwSensorRead", "hwButtonWait", "hwNFCRead",
|
|
"puzzleLED", "puzzleRadio", "puzzleMorse", "puzzleNFC", "puzzleCoffre",
|
|
"codeAssembly", "codeFragment", "coffreVerify",
|
|
"boxIMUShake", "boxIRSend",
|
|
"m5Beep", "m5LCDText", "m5ButtonAB", "m5RGBLed", "m5IMUShake",
|
|
"plipRing", "plipPickupWait",
|
|
}
|
|
|
|
|
|
# Phase D2 — binding carte : cartes connues. Miroir des clés de
|
|
# tools/zacus-gateway/boards.yaml, de VALID_BOARDS dans
|
|
# tools/scenario/runtime3_common.py et de ZACUS_BOARDS côté Blockly. Défaut
|
|
# master. Hardcodé ici pour garder le compilateur sans I/O ; boards.yaml fait foi.
|
|
VALID_BOARDS = {"master", "box3", "plip", "p7_coffre"}
|
|
|
|
|
|
def _warn_future_hw(kind: str, warnings: list[str], seen: set[str]) -> None:
|
|
"""Append a deduplicated « hardware futur » warning for `kind` (no-op for
|
|
real kinds). Non-blocking: the block still compiles to IR unchanged."""
|
|
if kind in FUTURE_HW_KINDS and kind not in seen:
|
|
seen.add(kind)
|
|
warnings.append(
|
|
f"bloc '{kind}' : hardware non câblé (validé mais non exécuté on-device)"
|
|
)
|
|
|
|
|
|
class CompileError(Exception):
|
|
pass
|
|
|
|
|
|
def _reporter_to_expr(node_id: Any, nodes_by_id: dict[str, dict]) -> str:
|
|
"""Serialise a reporter node (and nested value-input operands) to an
|
|
expression string in the loose condition grammar already used by the
|
|
scenario (`score > 5`, `pct >= 60 and lives > 0`). Mirrors the TS
|
|
serializeReporter() in @zacus/shared/blocks-doc.ts.
|
|
|
|
Returns "" when the ref/node is missing so callers fall back to a typed
|
|
text field (backward-compat with opaque condition strings).
|
|
"""
|
|
if not isinstance(node_id, str):
|
|
return ""
|
|
node = nodes_by_id.get(node_id)
|
|
if not isinstance(node, dict):
|
|
return ""
|
|
kind = node.get("kind", "")
|
|
params = node.get("params", {}) or {}
|
|
inputs = node.get("inputs", {}) or {}
|
|
|
|
def operand(name: str) -> str:
|
|
return _reporter_to_expr(inputs.get(name), nodes_by_id)
|
|
|
|
if kind == "valScore":
|
|
return "score"
|
|
if kind == "valVar":
|
|
return str(params.get("name", "") or "").strip()
|
|
if kind == "valNumber":
|
|
num = params.get("num", 0)
|
|
return str(num if num is not None else 0)
|
|
if kind == "valText":
|
|
import json as _json
|
|
return _json.dumps(str(params.get("text", "") or ""))
|
|
if kind == "opCompare":
|
|
return f"{operand('A')} {params.get('op', '==')} {operand('B')}"
|
|
if kind == "opArith":
|
|
return f"({operand('A')} {params.get('op', '+')} {operand('B')})"
|
|
if kind == "opBool":
|
|
return f"({operand('A')}) {params.get('op', 'and')} ({operand('B')})"
|
|
if kind == "opNot":
|
|
return f"not ({operand('A')})"
|
|
return ""
|
|
|
|
|
|
def _resolve_condition(node: dict, nodes_by_id: dict[str, dict], text_key: str = "condition") -> str:
|
|
"""Prefer a reporter branched into the node's `condition` value input;
|
|
fall back to the opaque text param (`text_key`) for backward compat."""
|
|
inputs = node.get("inputs", {}) or {}
|
|
expr = _reporter_to_expr(inputs.get("condition"), nodes_by_id)
|
|
if expr:
|
|
return expr
|
|
return str((node.get("params", {}) or {}).get(text_key, "") or "")
|
|
|
|
|
|
def compile_blocks(yaml_text: str, *, scenario_id: str = "BLOCKS_DRAFT") -> dict[str, Any]:
|
|
doc = yaml.safe_load(yaml_text)
|
|
if not isinstance(doc, dict):
|
|
raise CompileError("YAML root must be a mapping")
|
|
version = doc.get("blocks_studio_version")
|
|
if version not in (1, 2):
|
|
raise CompileError(f"unsupported blocks_studio_version: {version!r}")
|
|
|
|
if version == 1:
|
|
nodes_by_id = _flatten_v1(doc)
|
|
else:
|
|
nodes_by_id = _flatten_v2(doc)
|
|
|
|
if not nodes_by_id:
|
|
raise CompileError("no blocks found")
|
|
|
|
roots = _find_roots(nodes_by_id)
|
|
if not roots:
|
|
raise CompileError("no root chain found (all nodes are children of something)")
|
|
|
|
steps: list[dict] = []
|
|
warnings: list[str] = []
|
|
errors: list[str] = []
|
|
future_hw_seen: set[str] = set()
|
|
entry: str | None = None
|
|
used_ids: set[str] = set()
|
|
|
|
def step_id_for(scene_param: str | None) -> str:
|
|
base = (scene_param or f"anon_{len(used_ids)}").strip() or f"anon_{len(used_ids)}"
|
|
sid = f"STEP_{base}"
|
|
n = sid; i = 2
|
|
while n in used_ids:
|
|
n = f"{sid}_{i}"
|
|
i += 1
|
|
used_ids.add(n)
|
|
return n
|
|
|
|
for root_id in roots:
|
|
chain = _walk_chain(root_id, nodes_by_id)
|
|
# First scene
|
|
cursor_idx = 0
|
|
if chain and chain[0]["kind"] == "sceneStart":
|
|
sid = step_id_for(chain[0].get("params", {}).get("id"))
|
|
cursor_idx = 1
|
|
else:
|
|
warnings.append("chain missing sceneStart — synthetic step emitted")
|
|
sid = step_id_for(None)
|
|
|
|
current = _new_step(sid)
|
|
if entry is None:
|
|
entry = sid
|
|
|
|
for node in chain[cursor_idx:]:
|
|
kind = node.get("kind", "")
|
|
params = node.get("params", {}) or {}
|
|
_warn_future_hw(kind, warnings, future_hw_seen)
|
|
|
|
if kind == "sceneEnd":
|
|
break
|
|
elif kind == "sceneStart":
|
|
steps.append(current)
|
|
sid = step_id_for(params.get("id"))
|
|
current = _new_step(sid)
|
|
elif kind == "scene":
|
|
scene = _scene_from_params(params, sid, warnings)
|
|
if scene is not None:
|
|
if "scene" in current:
|
|
warnings.append(f"{sid}: multiple scene blocks — last one wins")
|
|
current["scene"] = scene
|
|
elif kind in ("puzzleQR", "puzzleSound", "puzzleLED", "puzzleRadio",
|
|
"puzzleMorse", "puzzleNFC", "puzzleCoffre"):
|
|
puzzle = _puzzle_from_params(kind, params, sid, warnings)
|
|
if puzzle is not None:
|
|
if "puzzle" in current:
|
|
warnings.append(f"{sid}: multiple puzzle blocks — last one wins")
|
|
current["puzzle"] = puzzle
|
|
elif kind == "logicIf":
|
|
# condition action with then/else sub-action lists. The
|
|
# condition is a branched reporter (value input) or the text
|
|
# param of repli.
|
|
then_actions = _walk_slot_actions(node, "body", nodes_by_id, warnings, future_hw_seen)
|
|
else_actions = _walk_slot_actions(node, "else", nodes_by_id, warnings, future_hw_seen)
|
|
current["actions"].append({
|
|
"kind": "condition",
|
|
"expr": _resolve_condition(node, nodes_by_id),
|
|
"then": then_actions,
|
|
"else": else_actions,
|
|
})
|
|
elif kind == "loopRepeat":
|
|
# Bounded repeat → loop action with a fixed iteration count and
|
|
# a nested action body.
|
|
current["actions"].append({
|
|
"kind": "loop",
|
|
"mode": "repeat",
|
|
"times": _to_int(params.get("times"), default=0),
|
|
"body": _walk_slot_actions(node, "body", nodes_by_id, warnings, future_hw_seen),
|
|
})
|
|
elif kind == "loopWhile":
|
|
current["actions"].append({
|
|
"kind": "loop",
|
|
"mode": "while",
|
|
"expr": _resolve_condition(node, nodes_by_id, text_key="condition_text"),
|
|
"body": _walk_slot_actions(node, "body", nodes_by_id, warnings, future_hw_seen),
|
|
})
|
|
elif kind == "hwOnBoard":
|
|
# Phase D2 — bloc contexte : n'émet aucune action propre. Les
|
|
# actions de son slot `body` sortent normalement, taguées avec
|
|
# le `board` choisi (défaut master, validé contre boards.yaml).
|
|
board = str(params.get("board", "") or "").strip() or "master"
|
|
if board not in VALID_BOARDS:
|
|
warnings.append(f"{sid}: hwOnBoard board '{board}' inconnu — using 'master'")
|
|
board = "master"
|
|
body_actions = _walk_slot_actions(node, "body", nodes_by_id, warnings, future_hw_seen)
|
|
_tag_actions_board(body_actions, board)
|
|
current["actions"].extend(body_actions)
|
|
else:
|
|
action_or_transition = _node_to_action_or_transition(node, warnings, nodes_by_id)
|
|
if action_or_transition is None:
|
|
continue
|
|
bucket, payload = action_or_transition
|
|
current[bucket].append(payload)
|
|
|
|
steps.append(current)
|
|
|
|
# Final pass: validate transition targets and warn on dangling ones.
|
|
valid_ids = {s["id"] for s in steps}
|
|
for s in steps:
|
|
for t in s["transitions"]:
|
|
tgt = t.get("target_step_id")
|
|
if tgt and tgt not in valid_ids:
|
|
warnings.append(f"{s['id']}: transition points to unknown {tgt}")
|
|
|
|
return {
|
|
"schema_version": SCHEMA_VERSION,
|
|
"scenario": {
|
|
"id": scenario_id,
|
|
"version": 3,
|
|
"title": scenario_id.replace("_", " ").title(),
|
|
"entry_step_id": entry or (steps[0]["id"] if steps else ""),
|
|
"source_kind": "blocks_studio",
|
|
},
|
|
"steps": steps,
|
|
"metadata": {
|
|
"migration_mode": "native",
|
|
"generated_by": "zacus_hub_gateway/blocks_to_runtime3",
|
|
"warnings": warnings,
|
|
"errors": errors,
|
|
},
|
|
}
|
|
|
|
|
|
# ---------- node graph helpers ----------
|
|
|
|
def _flatten_v2(doc: dict) -> dict[str, dict]:
|
|
nodes = doc.get("nodes") or []
|
|
if isinstance(nodes, str): # empty `[]` parsed as string by yaml flow
|
|
return {}
|
|
out: dict[str, dict] = {}
|
|
for n in nodes:
|
|
if isinstance(n, dict) and isinstance(n.get("id"), str):
|
|
out[n["id"]] = n
|
|
return out
|
|
|
|
|
|
def _flatten_v1(doc: dict) -> dict[str, dict]:
|
|
"""Old chain-based layout: rebuild flat nodes with synthesised next pointers."""
|
|
out: dict[str, dict] = {}
|
|
for chain_entry in doc.get("chains") or []:
|
|
sequence = chain_entry.get("sequence") or []
|
|
prev_id: str | None = None
|
|
for node in sequence:
|
|
if not isinstance(node, dict): continue
|
|
nid = node.get("id")
|
|
if not isinstance(nid, str): continue
|
|
n = dict(node)
|
|
out[nid] = n
|
|
if prev_id is not None and prev_id in out:
|
|
out[prev_id]["next"] = nid
|
|
prev_id = nid
|
|
return out
|
|
|
|
|
|
def _find_roots(nodes_by_id: dict[str, dict]) -> list[str]:
|
|
referenced: set[str] = set()
|
|
for n in nodes_by_id.values():
|
|
if isinstance(n.get("next"), str):
|
|
referenced.add(n["next"])
|
|
for slot_head in (n.get("slots") or {}).values():
|
|
if isinstance(slot_head, str):
|
|
referenced.add(slot_head)
|
|
# Reporter operands plugged into value inputs are never chain roots —
|
|
# they are resolved to expression strings, not walked as statements.
|
|
for input_ref in (n.get("inputs") or {}).values():
|
|
if isinstance(input_ref, str):
|
|
referenced.add(input_ref)
|
|
# Reporter nodes themselves (value blocks) are never top-level chains even
|
|
# if nothing references them (e.g. an orphan dropped on the canvas).
|
|
return [nid for nid in nodes_by_id
|
|
if nid not in referenced and nodes_by_id[nid].get("kind", "") not in _REPORTER_KINDS]
|
|
|
|
|
|
def _walk_chain(start_id: str, nodes_by_id: dict[str, dict]) -> list[dict]:
|
|
out: list[dict] = []
|
|
seen: set[str] = set()
|
|
cur: str | None = start_id
|
|
while cur and cur not in seen and cur in nodes_by_id:
|
|
seen.add(cur)
|
|
out.append(nodes_by_id[cur])
|
|
nxt = nodes_by_id[cur].get("next")
|
|
cur = nxt if isinstance(nxt, str) else None
|
|
return out
|
|
|
|
|
|
# ---------- per-node mapping ----------
|
|
|
|
def _new_step(sid: str) -> dict:
|
|
return {
|
|
"id": sid,
|
|
"scene_id": sid.replace("STEP_", "SCENE_"),
|
|
"audio_pack_id": "",
|
|
"actions": [],
|
|
"apps": [],
|
|
"transitions": [],
|
|
}
|
|
|
|
|
|
def _node_to_action_or_transition(
|
|
node: dict, warnings: list[str], nodes_by_id: dict[str, dict] | None = None
|
|
) -> tuple[str, dict] | None:
|
|
kind = node.get("kind", "")
|
|
params = node.get("params", {}) or {}
|
|
|
|
# --- Audio ---
|
|
if kind == "hwAudioStop":
|
|
return ("actions", {"kind": "sound_stop"})
|
|
if kind == "hwAudioVolume":
|
|
return ("actions", {"kind": "sound_volume", "level": _to_int(params.get("level"), default=70)})
|
|
# --- LCD ---
|
|
if kind == "hwLCDText":
|
|
return ("actions", {"kind": "lcd_text", "line": _to_int(params.get("line"), default=0), "text": params.get("text", "")})
|
|
if kind == "hwLCDClear":
|
|
return ("actions", {"kind": "lcd_clear"})
|
|
if kind == "hwLCDImage":
|
|
return ("actions", {"kind": "lcd_image", "asset": params.get("asset", "")})
|
|
if kind == "hwLCDTouchWait":
|
|
return ("actions", {"kind": "lcd_touch_wait", "zone": params.get("zone", ""),
|
|
"timeout_ms": _to_int(params.get("timeout_s"), default=30) * 1000})
|
|
# --- Hardware ESP32 generic ---
|
|
if kind == "hwBuzzerTone":
|
|
return ("actions", {"kind": "buzzer_tone", "freq": _to_int(params.get("freq"), default=2000),
|
|
"duration_ms": _to_int(params.get("ms"), default=120)})
|
|
if kind == "hwRelay":
|
|
return ("actions", {"kind": "relay", "channel": _to_int(params.get("channel"), default=0),
|
|
"state": params.get("state", "pulse")})
|
|
if kind == "hwSensorRead":
|
|
return ("actions", {"kind": "sensor_read", "pin": params.get("pin", "A0"), "var": params.get("var", "lecture")})
|
|
if kind == "hwButtonWait":
|
|
return ("actions", {"kind": "button_wait", "button": params.get("button", "btn_main"),
|
|
"timeout_ms": _to_int(params.get("timeout_s"), default=0) * 1000})
|
|
if kind == "hwNFCRead":
|
|
return ("actions", {"kind": "nfc_read", "expected": params.get("expected", ""),
|
|
"var": params.get("var", "nfc_uid"),
|
|
"timeout_ms": _to_int(params.get("timeout_s"), default=30) * 1000})
|
|
# --- ESP-NOW ---
|
|
if kind == "espnowRegisterPeer":
|
|
return ("actions", {"kind": "espnow_register_peer", "alias": params.get("alias", ""), "mac": params.get("mac", "")})
|
|
if kind == "espnowSend":
|
|
return ("actions", {"kind": "espnow_send", "peer": params.get("peer", ""), "command": params.get("command", "")})
|
|
if kind == "espnowBroadcast":
|
|
return ("actions", {"kind": "espnow_broadcast", "command": params.get("command", "")})
|
|
if kind == "espnowWait":
|
|
return ("actions", {"kind": "espnow_wait", "command": params.get("command", ""),
|
|
"timeout_ms": _to_int(params.get("timeout_s"), default=10) * 1000})
|
|
# --- BOX-3 ---
|
|
if kind == "boxIMUShake":
|
|
return ("actions", {"kind": "box_imu_shake", "threshold": _to_float(params.get("threshold"), default=1.5),
|
|
"timeout_ms": _to_int(params.get("timeout_s"), default=10) * 1000})
|
|
if kind == "boxIRSend":
|
|
return ("actions", {"kind": "box_ir_send", "protocol": params.get("protocol", "NEC"), "code": str(params.get("code", ""))})
|
|
# --- M5 ---
|
|
if kind == "m5Beep":
|
|
return ("actions", {"kind": "m5_beep", "freq": _to_int(params.get("freq"), default=4000),
|
|
"duration_ms": _to_int(params.get("ms"), default=200)})
|
|
if kind == "m5LCDText":
|
|
return ("actions", {"kind": "m5_lcd_text", "text": params.get("text", ""), "color": params.get("color", "white"),
|
|
"size": _to_int(params.get("size"), default=2)})
|
|
if kind == "m5ButtonAB":
|
|
return ("actions", {"kind": "m5_button_wait", "button": params.get("button", "A"),
|
|
"timeout_ms": _to_int(params.get("timeout_s"), default=0) * 1000})
|
|
if kind == "m5RGBLed":
|
|
return ("actions", {"kind": "m5_rgb_led", "color": params.get("color", "#000000")})
|
|
if kind == "m5IMUShake":
|
|
return ("actions", {"kind": "m5_imu_shake", "threshold": _to_float(params.get("threshold"), default=1.5),
|
|
"timeout_ms": _to_int(params.get("timeout_s"), default=10) * 1000})
|
|
# --- PLIP ---
|
|
if kind == "plipRing":
|
|
return ("actions", {"kind": "plip_ring", "duration_ms": int(_to_float(params.get("duration_s"), default=3) * 1000)})
|
|
if kind == "plipPickupWait":
|
|
return ("actions", {"kind": "plip_pickup_wait", "timeout_ms": _to_int(params.get("timeout_s"), default=30) * 1000})
|
|
|
|
if kind == "npcSay":
|
|
# voice/emotion/rate come from the « Zacus dit » block. voice_ref maps
|
|
# to the voice-bridge reference clip; emotion + rate are forward-compat
|
|
# IR fields (omitted at their defaults so older firmwares see no diff).
|
|
action: dict[str, Any] = {"kind": "tts_say", "text": params.get("text", "").strip()}
|
|
voice = str(params.get("voice", "") or "").strip()
|
|
if voice:
|
|
action["voice_ref"] = voice
|
|
emotion = str(params.get("emotion", "") or "").strip()
|
|
if emotion and emotion != "neutral":
|
|
action["emotion"] = emotion
|
|
rate = _to_float(params.get("rate"), default=1.0)
|
|
if abs(rate - 1.0) > 1e-9:
|
|
action["rate"] = rate
|
|
return ("actions", action)
|
|
if kind == "npcWaitResponse":
|
|
return ("actions", {"kind": "wait_user_voice", "timeout_ms": _to_int(params.get("timeout_s"), default=10) * 1000})
|
|
if kind == "npcIntentMatch":
|
|
target = params.get("then", "").strip()
|
|
if not target:
|
|
warnings.append("npcIntentMatch without 'then' ignored")
|
|
return None
|
|
return ("transitions", {
|
|
"event_type": "action",
|
|
"event_name": f"intent:{params.get('intent','').strip()}",
|
|
"target_step_id": f"STEP_{target}",
|
|
"priority": 10,
|
|
})
|
|
if kind == "hwServo":
|
|
return ("actions", {"kind": "hw_servo", "channel": _to_int(params.get("channel"), default=0), "angle": _to_int(params.get("angle"), default=90)})
|
|
if kind == "hwReadQR":
|
|
return ("actions", {"kind": "qr_expect", "value": params.get("expected", "")})
|
|
if kind == "hwLEDPattern":
|
|
return ("actions", {"kind": "led_pattern", "pattern": params.get("pattern", "off")})
|
|
if kind == "hwSoundPlay":
|
|
return ("actions", {"kind": "sound_play", "asset": params.get("asset", "")})
|
|
if kind == "mediaPlay":
|
|
# Phase D3 — média réel SD : joue un fichier par CHEMIN SD réel
|
|
# (résolu /sdcard/… ou /littlefs/… côté firmware). Une action
|
|
# sound_play peut donc porter soit `asset` (legacy hwSoundPlay) soit
|
|
# `path` (réel). Le `board` est posé ensuite par _tag_actions_board si
|
|
# l'action est dans un hwOnBoard (défaut master côté routage gateway).
|
|
return ("actions", {
|
|
"kind": "sound_play",
|
|
"path": str(params.get("path", "") or "").strip(),
|
|
"loop": params.get("loop") in ("TRUE", "1", True, 1),
|
|
})
|
|
if kind == "logicScore":
|
|
return ("actions", {"kind": "score_add", "delta": _to_int(params.get("delta"), default=1)})
|
|
if kind == "logicSetVar":
|
|
return ("actions", {"kind": "set_var", "name": params.get("name", ""), "value": params.get("value", "")})
|
|
if kind == "logicTimer":
|
|
return ("transitions", {
|
|
"event_type": "timer",
|
|
"event_name": "elapsed",
|
|
"after_ms": _to_int(params.get("seconds"), default=5) * 1000,
|
|
"target_step_id": "",
|
|
"priority": 1,
|
|
})
|
|
# --- Vague 2 : Maître du jeu (game-master adaptatif) ---
|
|
if kind == "phase":
|
|
return ("actions", {
|
|
"kind": "gm_phase",
|
|
"id": str(params.get("id", "") or "").strip(),
|
|
"duration_minutes": _to_int(params.get("duration_minutes"), default=0),
|
|
"trigger": str(params.get("trigger", "") or "manual").strip() or "manual",
|
|
})
|
|
if kind == "gmRule":
|
|
# condition→action règle adaptative. La condition est un reporter
|
|
# branché (value input `condition`) ou le texte de repli.
|
|
return ("actions", {
|
|
"kind": "gm_rule",
|
|
"expr": _resolve_condition(node, nodes_by_id or {}),
|
|
"action": str(params.get("action", "") or "proactive_hint").strip() or "proactive_hint",
|
|
})
|
|
if kind == "npcSetMood":
|
|
return ("actions", {"kind": "npc_mood", "mood": str(params.get("mood", "") or "NEUTRAL").strip() or "NEUTRAL"})
|
|
if kind == "giveHint":
|
|
return ("actions", {
|
|
"kind": "give_hint",
|
|
"level": _to_int(params.get("level"), default=1),
|
|
"hint_key": str(params.get("hint_key", "") or "").strip(),
|
|
})
|
|
if kind == "proactiveHint":
|
|
return ("actions", {
|
|
"kind": "proactive_hint",
|
|
"interval_ms": _to_int(params.get("interval_seconds"), default=90) * 1000,
|
|
"hint_key": str(params.get("hint_key", "") or "").strip(),
|
|
})
|
|
if kind == "setParcours":
|
|
puzzles = [p.strip() for p in str(params.get("puzzles", "") or "").split(",") if p.strip()]
|
|
return ("actions", {
|
|
"kind": "set_parcours",
|
|
"profile": str(params.get("profile", "") or "").strip(),
|
|
"puzzles": puzzles,
|
|
})
|
|
if kind == "scoreBonus":
|
|
return ("actions", {
|
|
"kind": "score_bonus",
|
|
"expr": _resolve_condition(node, nodes_by_id or {}),
|
|
"points": _to_int(params.get("points"), default=0),
|
|
})
|
|
if kind == "scorePenalty":
|
|
return ("actions", {
|
|
"kind": "score_penalty",
|
|
"penalty_kind": str(params.get("penalty_kind", "") or "time_penalty_per_minute").strip() or "time_penalty_per_minute",
|
|
"points": _to_int(params.get("points"), default=0),
|
|
})
|
|
# --- Vague 4 : assemblage du code final / coffre ---
|
|
if kind == "codeAssembly":
|
|
fragments = _parse_code_fragments(str(params.get("fragments", "") or ""), warnings)
|
|
return ("actions", {
|
|
"kind": "code_assembly",
|
|
"length": max(1, _to_int(params.get("length"), default=8)),
|
|
"fragments": fragments,
|
|
})
|
|
if kind == "codeFragment":
|
|
return ("actions", {
|
|
"kind": "code_fragment",
|
|
"puzzle": str(params.get("puzzle", "") or "").strip(),
|
|
"positions": _parse_positions(str(params.get("positions", "") or "")),
|
|
})
|
|
if kind == "coffreVerify":
|
|
return ("actions", {
|
|
"kind": "coffre_verify",
|
|
"id": _to_int(params.get("id"), default=7),
|
|
"on_success": str(params.get("on_success", "") or "").strip(),
|
|
})
|
|
# --- Phase D4 : transitions réelles ---
|
|
# Blocs « Quand … → aller à [target] » : émettent UNE transition avec le vrai
|
|
# event_type firmware-backed (∈ runtime3_common.EVENT_TYPES). target → même
|
|
# construction STEP_{target} que sceneGoto/npcIntentMatch (la normalisation
|
|
# stricte des ids vit côté runtime3_common._normalize_transition pour le
|
|
# chemin firmware-import ; le chemin blocks_studio émet STEP_<target> brut,
|
|
# cohérent avec les transitions existantes). target vide → bloc ignoré.
|
|
if kind in ("onButton", "onAudioDone", "onUnlock", "onSerial", "onEspnow", "onPuzzleSolved", "onCodeAssembled"):
|
|
target = str(params.get("target", "") or "").strip()
|
|
if not target:
|
|
warnings.append(f"{kind} without 'target' ignored")
|
|
return None
|
|
target_step_id = f"STEP_{target}"
|
|
if kind == "onButton":
|
|
return ("transitions", {
|
|
"event_type": "button",
|
|
"event_name": str(params.get("button", "") or "").strip(),
|
|
"target_step_id": target_step_id,
|
|
"priority": 5,
|
|
})
|
|
if kind == "onAudioDone":
|
|
return ("transitions", {
|
|
"event_type": "audio_done",
|
|
"event_name": "done",
|
|
"target_step_id": target_step_id,
|
|
"priority": 5,
|
|
})
|
|
if kind == "onUnlock":
|
|
return ("transitions", {
|
|
"event_type": "unlock",
|
|
"event_name": str(params.get("source", "") or "").strip(),
|
|
"target_step_id": target_step_id,
|
|
"priority": 5,
|
|
})
|
|
if kind == "onSerial":
|
|
return ("transitions", {
|
|
"event_type": "serial",
|
|
"event_name": str(params.get("token", "") or "").strip(),
|
|
"target_step_id": target_step_id,
|
|
"priority": 5,
|
|
})
|
|
if kind == "onEspnow":
|
|
return ("transitions", {
|
|
"event_type": "espnow",
|
|
"event_name": str(params.get("command", "") or "").strip(),
|
|
"target_step_id": target_step_id,
|
|
"priority": 5,
|
|
})
|
|
if kind == "onPuzzleSolved":
|
|
# « énigme résolue » se modélise comme un unlock nommé (event_type
|
|
# unlock, event_name puzzle:<id>) — le firmware n'expose pas
|
|
# d'event_type puzzle_state dédié, et EVENT_TYPES ne s'étend pas en D4.
|
|
return ("transitions", {
|
|
"event_type": "unlock",
|
|
"event_name": f"puzzle:{str(params.get('puzzle', '') or '').strip()}",
|
|
"target_step_id": target_step_id,
|
|
"priority": 5,
|
|
})
|
|
# onCodeAssembled : « code assemblé » se modélise comme un unlock nommé
|
|
# (event_type unlock, event_name code:<code>) — même famille que
|
|
# onPuzzleSolved, complète le branchement D4 sur l'état des énigmes.
|
|
return ("transitions", {
|
|
"event_type": "unlock",
|
|
"event_name": f"code:{str(params.get('code', '') or '').strip()}",
|
|
"target_step_id": target_step_id,
|
|
"priority": 5,
|
|
})
|
|
if kind == "sceneGoto":
|
|
target = params.get("target", "").strip()
|
|
return ("transitions", {
|
|
"event_type": "action",
|
|
"event_name": "goto",
|
|
"target_step_id": f"STEP_{target}",
|
|
"priority": 0,
|
|
})
|
|
if kind == "sceneBranch":
|
|
# Carry the resolved condition (branched reporter or text param) so the
|
|
# runtime can pick the branch. Previously dropped — now surfaced.
|
|
cond = _resolve_condition(node, nodes_by_id or {})
|
|
return ("transitions", {
|
|
"event_type": "action",
|
|
"event_name": "branch_true",
|
|
"condition": cond,
|
|
"target_step_id": f"STEP_{params.get('ifTrue','').strip()}",
|
|
"priority": 0,
|
|
})
|
|
|
|
warnings.append(f"unknown block kind '{kind}' skipped")
|
|
return None
|
|
|
|
|
|
# ---------- step-level scene / puzzle builders ----------
|
|
# Limits mirror tools/scenario/runtime3_common.py _validate_step_scene/_puzzle
|
|
# (themselves a mirror of the firmware's puzzle_binding.h). Out-of-range field
|
|
# values are clamped/truncated with a warning instead of failing the compile,
|
|
# so the emitted IR always passes the strict gateway validation.
|
|
|
|
_SCENE_LIMITS = {"title": 47, "subtitle": 63, "symbol": 15}
|
|
_SCENE_EFFECTS = ("pulse", "glitch", "gyro", "none")
|
|
|
|
|
|
def _scene_from_params(params: dict, sid: str, warnings: list[str]) -> dict | None:
|
|
scene: dict[str, Any] = {}
|
|
for key, max_len in _SCENE_LIMITS.items():
|
|
value = str(params.get(key, "") or "").strip()
|
|
if not value:
|
|
continue
|
|
if len(value) > max_len:
|
|
warnings.append(f"{sid}: scene.{key} truncated to {max_len} chars")
|
|
value = value[:max_len]
|
|
scene[key] = value
|
|
effect = str(params.get("effect", "") or "").strip()
|
|
if effect:
|
|
if effect not in _SCENE_EFFECTS:
|
|
warnings.append(f"{sid}: scene.effect '{effect}' unknown — using 'pulse'")
|
|
effect = "pulse"
|
|
scene["effect"] = effect
|
|
if not scene:
|
|
warnings.append(f"{sid}: empty scene block ignored")
|
|
return None
|
|
return scene
|
|
|
|
|
|
# Authoring block kind -> firmware puzzle `type` + default slot id.
|
|
_PUZZLE_TYPE_BY_KIND = {
|
|
"puzzleQR": ("qr", 3),
|
|
"puzzleSound": ("sound", 1),
|
|
"puzzleLED": ("led", 2),
|
|
"puzzleRadio": ("radio", 4),
|
|
"puzzleMorse": ("morse", 5),
|
|
"puzzleNFC": ("nfc", 6),
|
|
"puzzleCoffre": ("keypad", 7),
|
|
}
|
|
|
|
|
|
def _puzzle_from_params(kind: str, params: dict, sid: str, warnings: list[str]) -> dict | None:
|
|
is_qr = kind == "puzzleQR"
|
|
fw_type, default_id = _PUZZLE_TYPE_BY_KIND.get(kind, ("sound", 1))
|
|
pid = _to_int(params.get("id"), default=default_id)
|
|
if not (1 <= pid <= 8):
|
|
warnings.append(f"{sid}: puzzle.id {pid} out of 1..8 — clamped")
|
|
pid = min(8, max(1, pid))
|
|
puzzle: dict[str, Any] = {"id": pid, "type": fw_type}
|
|
|
|
if kind == "puzzleLED":
|
|
steps = [s.strip() for s in str(params.get("pattern", "") or "").split(",") if s.strip()]
|
|
if not steps:
|
|
warnings.append(f"{sid}: puzzleLED without a pattern ignored")
|
|
return None
|
|
if len(steps) > 32:
|
|
warnings.append(f"{sid}: puzzle.pattern capped at 32 entries (was {len(steps)})")
|
|
steps = steps[:32]
|
|
clipped = [s for s in steps if len(s) >= 32]
|
|
if clipped:
|
|
warnings.append(f"{sid}: {len(clipped)} LED pattern entry(ies) truncated to 31 chars")
|
|
steps = [s[:31] for s in steps]
|
|
puzzle["pattern"] = steps
|
|
elif kind == "puzzleRadio":
|
|
puzzle["frequency"] = max(0, _to_int(params.get("frequency"), default=1337))
|
|
puzzle["tolerance"] = max(0, _to_int(params.get("tolerance"), default=10))
|
|
elif kind == "puzzleMorse":
|
|
message = str(params.get("message", "") or "").strip()
|
|
if not message:
|
|
warnings.append(f"{sid}: puzzleMorse without a message ignored")
|
|
return None
|
|
if len(message) > 63:
|
|
warnings.append(f"{sid}: puzzle.message truncated to 63 chars")
|
|
message = message[:63]
|
|
mode = str(params.get("mode", "") or "normal").strip() or "normal"
|
|
if mode not in ("normal", "light"):
|
|
warnings.append(f"{sid}: puzzle.mode '{mode}' unknown — using 'normal'")
|
|
mode = "normal"
|
|
puzzle["message"] = message
|
|
puzzle["mode"] = mode
|
|
elif kind == "puzzleNFC":
|
|
tags = [t.strip() for t in str(params.get("tags", "") or "").split(",") if t.strip()]
|
|
if not tags:
|
|
warnings.append(f"{sid}: puzzleNFC without tags ignored")
|
|
return None
|
|
if len(tags) > 32:
|
|
warnings.append(f"{sid}: puzzle.tags capped at 32 (was {len(tags)})")
|
|
tags = tags[:32]
|
|
clipped = [t for t in tags if len(t) >= 32]
|
|
if clipped:
|
|
warnings.append(f"{sid}: {len(clipped)} NFC tag(s) truncated to 31 chars")
|
|
tags = [t[:31] for t in tags]
|
|
puzzle["tags"] = tags
|
|
ordered = str(params.get("ordered", "TRUE")).strip().upper()
|
|
puzzle["ordered"] = ordered not in ("FALSE", "0", "")
|
|
elif kind == "puzzleCoffre":
|
|
length = _to_int(params.get("length"), default=8)
|
|
if not (1 <= length <= 16):
|
|
warnings.append(f"{sid}: puzzle.length {length} out of 1..16 — clamped")
|
|
length = min(16, max(1, length))
|
|
puzzle["length"] = length
|
|
elif is_qr:
|
|
codes = [c.strip() for c in str(params.get("codes", "") or "").split(",") if c.strip()]
|
|
if not codes:
|
|
warnings.append(f"{sid}: puzzleQR without codes ignored")
|
|
return None
|
|
if len(codes) > 16:
|
|
warnings.append(f"{sid}: puzzle.codes capped at 16 (was {len(codes)})")
|
|
codes = codes[:16]
|
|
clipped = [c for c in codes if len(c) >= 32]
|
|
if clipped:
|
|
warnings.append(f"{sid}: {len(clipped)} puzzle code(s) truncated to 31 chars")
|
|
codes = [c[:31] for c in codes]
|
|
puzzle["codes"] = codes
|
|
else:
|
|
melody: list[int] = []
|
|
for raw in str(params.get("melody", "") or "").split(","):
|
|
raw = raw.strip()
|
|
if not raw:
|
|
continue
|
|
try:
|
|
note = int(float(raw))
|
|
except ValueError:
|
|
warnings.append(f"{sid}: puzzle melody note '{raw}' is not a number — skipped")
|
|
continue
|
|
if not (0 <= note <= 127):
|
|
warnings.append(f"{sid}: puzzle melody note {note} out of MIDI 0..127 — skipped")
|
|
continue
|
|
melody.append(note)
|
|
if not melody:
|
|
warnings.append(f"{sid}: puzzleSound without a playable melody ignored")
|
|
return None
|
|
if len(melody) > 32:
|
|
warnings.append(f"{sid}: puzzle.melody capped at 32 notes (was {len(melody)})")
|
|
melody = melody[:32]
|
|
puzzle["melody"] = melody
|
|
puzzle["tolerance"] = max(0, _to_int(params.get("tolerance"), default=1))
|
|
|
|
digits = [int(ch) for ch in str(params.get("fragment", "") or "") if ch.isdigit()]
|
|
if not digits:
|
|
digits = [pid] # même défaut que translate_solution_to_puzzle
|
|
elif len(digits) > 4:
|
|
warnings.append(f"{sid}: puzzle.fragment capped at 4 digits")
|
|
digits = digits[:4]
|
|
puzzle["fragment"] = digits
|
|
|
|
# Phase D2 — binding carte : tag `board` (défaut master) sur l'énigme.
|
|
# Validé contre boards.yaml côté runtime3_common._validate_step_puzzle.
|
|
board = str(params.get("board", "") or "").strip() or "master"
|
|
if board not in VALID_BOARDS:
|
|
warnings.append(f"{sid}: puzzle.board '{board}' inconnu — using 'master'")
|
|
board = "master"
|
|
puzzle["board"] = board
|
|
return puzzle
|
|
|
|
|
|
def _walk_slot_actions(parent: dict, slot_name: str, nodes_by_id: dict[str, dict], warnings: list[str],
|
|
future_hw_seen: set[str] | None = None) -> list[dict]:
|
|
"""Walk the chain anchored at `parent.slots[slot_name]` and serialise it as
|
|
a nested action list. `sceneStart`/`sceneEnd` inside a slot are ignored
|
|
with a warning since branches must remain local to the parent step.
|
|
`logicIf` nested in a slot recurses into another condition action.
|
|
"""
|
|
if future_hw_seen is None:
|
|
future_hw_seen = set()
|
|
head = (parent.get("slots") or {}).get(slot_name)
|
|
if not isinstance(head, str):
|
|
return []
|
|
chain = _walk_chain(head, nodes_by_id)
|
|
out: list[dict] = []
|
|
for node in chain:
|
|
kind = node.get("kind", "")
|
|
_warn_future_hw(kind, warnings, future_hw_seen)
|
|
if kind in ("sceneStart", "sceneEnd", "sceneGoto", "sceneBranch", "npcIntentMatch", "logicTimer",
|
|
"onButton", "onAudioDone", "onUnlock", "onSerial", "onEspnow", "onPuzzleSolved", "onCodeAssembled",
|
|
"scene", "puzzleQR", "puzzleSound",
|
|
"puzzleLED", "puzzleRadio", "puzzleMorse", "puzzleNFC", "puzzleCoffre",
|
|
"phase", "setParcours", "scorePenalty"):
|
|
warnings.append(f"slot '{slot_name}' contains '{kind}' which only makes sense at step level — ignored")
|
|
continue
|
|
if kind == "logicIf":
|
|
out.append({
|
|
"kind": "condition",
|
|
"expr": _resolve_condition(node, nodes_by_id),
|
|
"then": _walk_slot_actions(node, "body", nodes_by_id, warnings, future_hw_seen),
|
|
"else": _walk_slot_actions(node, "else", nodes_by_id, warnings, future_hw_seen),
|
|
})
|
|
continue
|
|
if kind == "loopRepeat":
|
|
params = node.get("params", {}) or {}
|
|
out.append({
|
|
"kind": "loop",
|
|
"mode": "repeat",
|
|
"times": _to_int(params.get("times"), default=0),
|
|
"body": _walk_slot_actions(node, "body", nodes_by_id, warnings, future_hw_seen),
|
|
})
|
|
continue
|
|
if kind == "loopWhile":
|
|
out.append({
|
|
"kind": "loop",
|
|
"mode": "while",
|
|
"expr": _resolve_condition(node, nodes_by_id, text_key="condition_text"),
|
|
"body": _walk_slot_actions(node, "body", nodes_by_id, warnings, future_hw_seen),
|
|
})
|
|
continue
|
|
mapped = _node_to_action_or_transition(node, warnings, nodes_by_id)
|
|
if mapped is None: continue
|
|
bucket, payload = mapped
|
|
if bucket != "actions":
|
|
warnings.append(f"slot '{slot_name}' produced a transition from '{kind}' — only actions allowed inside branches, ignored")
|
|
continue
|
|
out.append(payload)
|
|
return out
|
|
|
|
|
|
def _tag_actions_board(actions: list[dict], board: str) -> None:
|
|
"""Phase D2 — pose `action["board"] = board` sur chaque action (récursif
|
|
dans les sous-listes then/else/body des actions condition/loop). La
|
|
sémantique des actions est inchangée : seul le tag `board` est ajouté.
|
|
Une action déjà taguée (hwOnBoard imbriqué) garde son board le plus proche."""
|
|
for action in actions:
|
|
if not isinstance(action, dict):
|
|
continue
|
|
action.setdefault("board", board)
|
|
for sub_key in ("then", "else", "body"):
|
|
sub = action.get(sub_key)
|
|
if isinstance(sub, list):
|
|
_tag_actions_board(sub, board)
|
|
|
|
|
|
def _parse_positions(raw: str) -> list[int]:
|
|
"""Parse a digit-position spec like "1-2" or "3" or "1,4" into a sorted
|
|
list of 1-based positions. Ranges are inclusive; junk is skipped."""
|
|
positions: set[int] = set()
|
|
for chunk in raw.replace(";", ",").split(","):
|
|
chunk = chunk.strip()
|
|
if not chunk:
|
|
continue
|
|
if "-" in chunk:
|
|
lo_s, _, hi_s = chunk.partition("-")
|
|
lo, hi = _to_int(lo_s, default=0), _to_int(hi_s, default=0)
|
|
if lo and hi and lo <= hi:
|
|
positions.update(range(lo, hi + 1))
|
|
else:
|
|
n = _to_int(chunk, default=0)
|
|
if n:
|
|
positions.add(n)
|
|
return sorted(positions)
|
|
|
|
|
|
def _parse_code_fragments(raw: str, warnings: list[str]) -> list[dict]:
|
|
"""Parse "P1_SON:1-2, P2_CIRCUIT:3" into
|
|
[{puzzle, positions:[...]}, ...]. Entries without a colon or positions
|
|
are skipped with a warning."""
|
|
out: list[dict] = []
|
|
# Split on commas only at the top level: positions never contain commas
|
|
# because the colon separates puzzle name from the position spec.
|
|
for entry in raw.split(","):
|
|
entry = entry.strip()
|
|
if not entry:
|
|
continue
|
|
if ":" not in entry:
|
|
warnings.append(f"code_assembly: fragment '{entry}' missing ':' — skipped")
|
|
continue
|
|
puzzle, _, pos_spec = entry.partition(":")
|
|
puzzle = puzzle.strip()
|
|
positions = _parse_positions(pos_spec)
|
|
if not puzzle or not positions:
|
|
warnings.append(f"code_assembly: fragment '{entry}' has no puzzle/positions — skipped")
|
|
continue
|
|
out.append({"puzzle": puzzle, "positions": positions})
|
|
return out
|
|
|
|
|
|
def _to_int(value: Any, *, default: int) -> int:
|
|
try:
|
|
return int(float(str(value).strip()))
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def _to_float(value: Any, *, default: float) -> float:
|
|
try:
|
|
return float(str(value).strip())
|
|
except (TypeError, ValueError):
|
|
return default
|