Bedrock과 로봇 시연 연결

aws --version
python3 "$ROBOT_REPO/workshop/scripts/lab.py" doctor \
  --config "$ROBOT_CONFIG" --aws

export ROBOT_MODEL_ID="$(python3 -c 'import json,os; print(json.load(open(os.environ["ROBOT_CONFIG"]))["modelId"])')"
export ROBOT_PROFILE="$(python3 -c 'import json,os; print(json.load(open(os.environ["ROBOT_CONFIG"]))["profile"])')"
printf 'Bedrock 모델: %s\n리전: ap-northeast-2\n' "$ROBOT_MODEL_ID"

if [ -n "$ROBOT_PROFILE" ]; then
  export AWS_PROFILE="$ROBOT_PROFILE"
else
  unset AWS_PROFILE AWS_DEFAULT_PROFILE
fi

cd "$ROBOT_LAB"
export ROBOT_BROWSER_ORIGIN="$ROBOT_API_URL"
NODE_ENV=development HOST=127.0.0.1 PORT="$ROBOT_API_PORT" \
  PUBLIC_ORIGIN="$ROBOT_BROWSER_ORIGIN" \
  SIMULATION_URL="http://127.0.0.1:$ROBOT_SIM_PORT" \
  LOCAL_RUNS_DIR=.local/runs MAX_ACTIVE_RUNS=1 \
  AWS_REGION=ap-northeast-2 MODEL_ID="$ROBOT_MODEL_ID" \
  AGENTCORE_RUNTIME_ARN="" RUNS_TABLE="" REPLAY_BUCKET="" \
  ORIGIN_GATE="" SESSION_SECRET="" \
  npm run server

.venv/bin/python - <<'PY'
import json
import os
import urllib.request
with urllib.request.urlopen(os.environ["ROBOT_API_URL"] + "/api/config", timeout=10) as response:
    config = json.load(response)
assert config["planner"] == "bedrock", config["planner"]
assert config["storage"] == "local", config["storage"]
assert config["simulation"] == "ready", config["simulation"]
print({key: config.get(key) for key in ("planner", "model", "storage", "simulation")})
PY

cd "$ROBOT_LAB"
export ROBOT_BROWSER_ORIGIN="$ROBOT_API_URL"
.venv/bin/python - <<'PY'
import http.cookiejar
import json
import math
import os
from pathlib import Path
import time
import urllib.error
import urllib.request

base = os.environ["ROBOT_API_URL"].rstrip("/")
origin = os.environ["ROBOT_BROWSER_ORIGIN"]
assert origin == base, "터미널 검증 시 API와 검사 양쪽의 Origin을 루프백 API 주소로 맞추세요."
output = Path(".local/evidence")
output.mkdir(parents=True, exist_ok=True)
opener = urllib.request.build_opener(
    urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar())
)
csrf = None

def call(path, payload=None):
    headers = {"Accept": "application/json"}
    data = None
    if payload is not None:
        headers.update({"Content-Type": "application/json",
                        "Origin": origin, "X-CSRF-Token": csrf})
        data = json.dumps(payload, ensure_ascii=False).encode()
    request = urllib.request.Request(base + path, data=data, headers=headers)
    try:
        with opener.open(request, timeout=130) as response:
            if response.headers.get_content_type() != "application/json":
                raise RuntimeError(f"{path}: JSON 응답이 아닙니다. 포트 전달 경로를 확인하세요.")
            return json.load(response)
    except urllib.error.HTTPError as error:
        try:
            detail = json.loads(error.read()).get("error", {})
        except (ValueError, UnicodeError):
            detail = {"code": "non_json_response"}
        raise RuntimeError(f"{path}: HTTP {error.code}, {detail.get('code', 'unknown')}") from None

def save(name, value):
    (output / name).write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n")

config = call("/api/config")
assert config["planner"] == "bedrock", config["planner"]
assert config["storage"] == "local"
assert config["simulation"] == "ready"
csrf = config["csrf"]
cases = [
    ("red", "transfer", "빨간 상자만 오른쪽 트레이로 옮겨줘", 445,
     [{"objectId": "red", "target": "tray"}]),
    ("blue-first", "stack", "파란 상자, 빨간 상자, 주황 상자 순서로 쌓아줘", 2026,
     [{"objectId": name, "target": "tower"} for name in ("blue", "red", "orange")]),
]
report = []
for name, scenario, prompt, seed, actions in cases:
    started = time.perf_counter()
    plan = call("/api/plan", {"scenario": scenario, "prompt": prompt})
    plan_seconds = time.perf_counter() - started
    save(f"bedrock-{name}-plan.json", plan)
    assert plan["provider"] == "bedrock", plan["provider"]
    assert plan["model"] == config["model"]
    assert plan["scenario"] == scenario and plan["actions"] == actions, plan

    started = time.perf_counter()
    run = call("/api/runs", {"planId": plan["id"], "seed": seed})
    deadline = time.monotonic() + 240
    while run["status"] in ("queued", "running"):
        if time.monotonic() >= deadline:
            save(f"bedrock-{name}-run.json", run)
            raise TimeoutError(f"{name}: 마지막 실행 상태를 저장했습니다.")
        time.sleep(1)
        run = call("/api/runs/" + run["id"])
    run_seconds = time.perf_counter() - started
    save(f"bedrock-{name}-run.json", run)
    replay = call("/api/runs/" + run["id"] + "/replay")
    save(f"bedrock-{name}-replay.json", replay)
    assert run["status"] == replay["status"] == "succeeded", (run["status"], replay["checks"])
    assert replay["seed"] == seed and replay["scenario"] == scenario
    assert replay["engine"] == "MuJoCo"
    assert replay["provenance"]["learning"] == "none"
    assert replay["metrics"]["success"] is True
    assert replay["metrics"]["placed"] == replay["metrics"]["total"] == len(actions)
    assert replay["checks"] and all(check["passed"] for check in replay["checks"])
    assert [step["objectId"] for step in replay["steps"]] == [a["objectId"] for a in actions]
    final = replay["frames"][-1]["bodies"]
    if name == "red":
        initial = replay["frames"][0]["bodies"]
        assert final["red"]["position"][1] < -0.10
        for other in ("blue", "orange"):
            assert final[other]["position"][1] > 0.17
            assert math.dist(initial[other]["position"], final[other]["position"]) < 0.008
    else:
        for box, height in (("blue", 0.025), ("red", 0.075), ("orange", 0.125)):
            x, y, z = final[box]["position"]
            assert abs(x - 0.50) < 0.008 and abs(y + 0.20) < 0.008
            assert abs(z - height) < 0.003
    row = {"case": name, "seed": seed, "provider": plan["provider"], "model": plan["model"],
           "planSeconds": round(plan_seconds, 3), "runSeconds": round(run_seconds, 3),
           "simulationDuration": replay["duration"], "status": run["status"],
           "metrics": replay["metrics"]}
    report.append(row)
    save("bedrock-report.json", report)
    print(json.dumps(row, ensure_ascii=False))
print("실제 Bedrock 계획 2건과 새 MuJoCo 실행 2건 확인")
PY

read -r -p "배정된 HTTPS 미리보기 Origin: " ROBOT_BROWSER_ORIGIN
export ROBOT_BROWSER_ORIGIN