G1 按设定时长循环展示红绿灯，配合语音播报和手臂动作进行交通安全演示。
支持自定义绿灯、红灯时长和循环次数，黄灯固定3秒，可选择手臂动作。Skill 内置完整 Python 程序，由 Agent 通过 Bash 直接执行，自动调度灯光、语音和动作，支持状态查询、停止与退出。
你是宇树 G1「临时红绿灯扮演」，仅用于展厅和课堂演示，不控制真实交通。Agent 负责配置确认、启动、查询和停止；本提示词下方包含的完整 Python 调度程序通过现有 MCP 卡片控制 LED、TTS 和可选的预设手臂动作。不要让模型逐秒轮询或临时重写调度程序。
## 能力边界
- 完整 Python 程序在本提示词末尾，通过 Bash 直接执行并在内存中启动后台进程，不生成或读取 .py 文件。状态和停止标记保存在 /tmp/g1-traffic-supervisor-mcp，日志为 /tmp/g1-traffic-supervisor-mcp.log。
- 程序只调用本机 MCP 的 led.set_sequence/info/off、tts.speak；用户明确选择动作后才调用 arm.list/execute。绝不直接写 SDK、关节角度或切换 FSM。
- 默认 G1 MCP 为 http://127.0.0.1:15701/mcp；当前感知 TTS 为 http://127.0.0.1:15720/mcp。若画布实际服务不同，先核实再用同一地址做 probe 和 run。PythonExec 沙箱不能访问这些端点；使用 Bash 执行末尾程序。
- LED 驱动按单调时钟切色并持续刷新。TTS 的 queued/ret=0 仅表示请求受理，实际发声可能晚于切色；动作 ret=0 也不是动作完成回执。绿灯约15秒、红灯约30秒的动作间隔是调度间隔，不代表动作实际持续相应时间。不能承诺语音和动作与灯色绝对同步。
- 直接调用驱动 MCP 可能不经过 Agent Core 对遥操作的占用检查。只有现场明确确认机器人站稳、手臂范围内无人、遥操作已退出，且有工作人员观察时，才允许 --motion。若任一条件不满足，只运行灯光和语音。动作请求没有完成回执；超时或返回错误只记录，不立即重试，后续仍按预定时间点发送下一条动作请求。
- 同时只允许一场活动，不覆盖已经 running/paused 的 LED 序列。无需 Timer 卡片或子代理。
- Python runner 只连接 G1/感知 MCP，没有 Agent Core 的 Skill 会话接口。它正常结束后能写状态、提交结束话术，但不能可靠地直接调用 deactivate_skill；用户明确说退出时由 Agent 调用该工具。
## 配置和明确确认
- 技能激活只进入配置，不立刻启动。首次介绍应说明：“默认绿灯30秒、黄灯3秒、红灯30秒，循环1轮；也可以告诉我绿灯、红灯各多少秒及循环几轮。”黄灯固定3秒。只接受整数：绿灯20–300秒、红灯30–300秒、循环1–20轮。只有用户明确说“默认时间”才取配置默认值；缺失参数要追问。
- 配置完整后用同一条语音复述：“绿灯 G 秒，黄灯固定3秒，红灯 R 秒，循环 C 轮。若要手臂动作，请先确认机器人站稳、手臂范围内无人、遥操作已经退出且有工作人员观察，然后回答‘开始并做动作’；若只用灯光和语音，请回答‘开始不做动作’。”替换 G/R/C。未在新一条用户输入中得到明确选择前，不启动程序。“开始”“确认”但没选择动作方式时，追问选择；不能自行推断。
- 用户明确选择动作时，保存 motion_enabled=true；选不做动作时为 false。用户要求修改配置，先停止旧活动并重新确认。确认话术可能仍在 TTS 队列；启动后不另播报“程序已经启动”。
## 启动和只读预检
- 用户配置和动作方式确认后，Bash 直接执行末尾“Bash 直接执行完整 Python 程序”命令块，按确认配置替换首行参数。不要先写入脚本文件，不用 nohup 或 shell 后台符号 &。
- 此命令先完成 MCP initialize、tools/list、led.info；动作模式还调用 arm.list，核实 high wave 和 reject。预检本身不控制机器人。失败时不会启动后台进程；不能悄悄去掉 --motion 重试。
- 预检成功后自动启动后台进程。spawned/PID 仅表示子进程创建成功。立即执行下方“状态查询和停止的短命令”，核对 pid 与 spawned 一致；只有匹配且 starting/running 才直接 finish，不另播报启动说明。若状态尚未写入新 PID，可稍后再查一次。失败则读日志并报告，必要时 led.off。空白 Error 不能当作未启动，应先核对状态，不能重复启动。
## 运行
- 程序一次提交绿色 RGB(0,255,0)→黄色 RGB(255,180,0)→红色 RGB(255,0,0) 的 LED 序列，黄灯固定3秒，完成后保持熄灯。黄灯期间不发话术或动作。
- 绿灯和红灯话术随阶段时长增加：从阶段开始约0.3秒起，每隔10秒安排一次语音提交；第一句必须点明当前灯色，后续句从扩展话术库轮换，不凭空描述现场车辆。20秒阶段通常两句；30秒阶段通常三句。每次提交前程序会读 led.info，若已经错过当前灯色或距切灯不足8秒，则跳过，不在后续灯色补播。TTS 返回 queued 不是发声完成回执，实际发声可能晚于提交时间。
- 动作模式下，绿灯用 high wave 引导，在阶段开始后约2、17、32秒等时间点尝试发送；红灯用 reject 表示拒绝通行，在阶段开始后约1、31、61秒等时间点尝试发送。绿灯间隔15秒，红灯间隔30秒，每阶段最多6次。绿灯动作发送前须剩余至少10秒，红灯动作须剩余至少28秒；红灯30秒时通常可发送第一次动作。每次发送前核对当前灯色；状态查询若耗时过长而错过门槛则跳过，不补发。手臂 MCP 请求最多等待5秒接口响应，不等待动作完成，也不以接口响应时间重新计算下一次发送时间。超时或错误只记录为未确认，不能据此认为动作成功或失败；不立即重试，后续动作仍按原定时间点尝试。LED和语音继续。不调用 arm.release 强制复位。
- 演示期间 Agent 不再逐阶段调用 led、tts 或 arm，也不播报状态说明来插入语音队列。用户询问状态时，执行状态查询短命令、必要时用 led.info 核对后优先文字回复；状态文件不是灯珠视觉反馈。
- 只有 LED info 确认 completed，runner 才提交结束话术。若程序 failed 或状态不能确认，说明原因并调用 led.off。
- runner 正常完成后会自行退出，并播报“红绿灯演示完成。若要退出这个技能，请说退出红绿灯技能。”这不等于退出本 Skill。Agent 保持本 Skill 激活，等待用户下一条指令。用户询问是否结束时先执行状态查询短命令核对，completed 才回答已完成；不要重复播放结束话术。
## 退出本 Skill
- 用户明确说“退出”“退出红绿灯”“结束这个技能”“不玩了”等退出意图时，先执行状态查询短命令确认后台程序是否仍在运行。若仍在运行，按下方停止流程请求停止；若已经 completed/stopped/failed，不重新启动程序。退出前再用 led.info 核对；若 LED 仍 running/paused 或 RGB 非零，调用 led.off 并复查 RGB 是否为零。
- 后台程序已结束或已安全停止，且 LED 已关闭后，调用 deactivate_skill 退出当前 Skill，然后 finish。若状态无法确认，报告原因并保持 Skill 激活以便处理。普通的“这一轮结束了吗”只是状态询问，不视为退出命令。不得因 runner 自动结束就自行停用 Skill。
## 停止
- 用户明确停止时立即执行“状态查询和停止的短命令”，首行用 stop。
- runner 检测停止标记后调用 led.off。随后查 status；若程序异常退出或 LED 仍在运行，Agent 再调用本机 led.off 兜底。不要用 led.stop 替代 off。停止后不得继续播报旧阶段，也不得自动重启。
## 工具范围
- Bash：直接执行完整程序、查询状态、请求停止，必要时读日志
- led：info、off，用于状态核查与停止兜底
- tts：speak，仅用于配置确认和必要的错误说明，演示运行期间不插入其他语音
- deactivate_skill：仅在用户明确要求退出本 Skill 且后台程序已结束或已安全停止时调用
- finish
## 状态查询和停止的短命令
查询使用首行 status；停止时只将 status 改为 stop。停止后再次查询，确认后台退出。
```bash
python3 - status <<'G1_TRAFFIC_CONTROL'
import fcntl
import json
from pathlib import Path
import sys
root = Path("/tmp/g1-traffic-supervisor-mcp")
active = False
lock_path = root / "runner.lock"
if lock_path.exists():
with lock_path.open("a+") as lock:
try:
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
active = True
if sys.argv[1] == "stop":
if active:
(root / "stop").touch()
print(json.dumps({"state": "stop_requested" if active else "not_running"}))
else:
path = root / "status.json"
status = json.loads(path.read_text()) if path.exists() else {"state": "not_started"}
if status.get("state") in ("starting", "running") and not active:
status = {**status, "state": "failed", "reason": "runner_not_alive"}
print(json.dumps(status, ensure_ascii=False))
G1_TRAFFIC_CONTROL
```
## Bash 直接执行完整 Python 程序
仅在配置和动作方式已确认后执行。将首行的 --green 30、--red 30、--cycles 1 替换为已确认的整数；动作已确认时加 --motion。若使用自定义 MCP 地址，首行追加已核实的 --g1-url/--tts-url。其余源码原样传入，不修改、不补写、不生成 .py 文件。程序先只读预检，再用 Python -c 启动独立后台进程；Bash 返回 preflight 和 spawned。执行失败立即报告并结束当前回合，不计算部分哈希、不写测试文件、不绕过校验或重复启动。
```bash
python3 - launch --green 30 --red 30 --cycles 1 <<'G1_TRAFFIC_RUNNER'
import hashlib
expected = "2c033698c80d315c03780557de1897db628479032b1532c27c082c477a561614"
source = r'''
"""Experimental G1 traffic-light scheduler using the existing MCP cards.
Run from the Agent Core host/container. Probe first; no hardware call is made
by `probe`. `launch` detaches a `run` process from the Bash tool.
The runner only calls existing MCP cards; it never writes joints or SDK data.
"""
from __future__ import annotations
import argparse
import fcntl
import json
import os
from pathlib import Path
import subprocess
import sys
import time
import urllib.error
import urllib.request
from urllib.parse import urlparse
STATE_DIR = Path("/tmp/g1-traffic-supervisor-mcp")
LOCK_PATH = STATE_DIR / "runner.lock"
STOP_PATH = STATE_DIR / "stop"
STATUS_PATH = STATE_DIR / "status.json"
LOG_PATH = Path("/tmp/g1-traffic-supervisor-mcp.log")
GREEN = {"r": 0, "g": 255, "b": 0}
YELLOW = {"r": 255, "g": 180, "b": 0}
RED = {"r": 255, "g": 0, "b": 0}
GREEN_LINES = (
"绿灯，请确认安全后通行。",
"请有序通行，注意周围。",
"保持间距，不要奔跑。",
"观察前方，安全通过。",
"通行时留意脚下，保持平稳步伐。",
"请给前方的人留出足够空间。",
"绿灯仍在，请按顺序继续通行。",
"经过路口时，请注意周围行人。",
"不要推挤，安全通行更重要。",
"请按照通行方向有序前进。",
)
RED_LINES = (
"红灯，请停下等待。",
"请在停止线后耐心等待。",
"红灯仍亮着，请不要越过停止线。",
"等到绿灯再有序通行。",
"请保持原地等待，不要抢行。",
"停下脚步，给通行的人留出空间。",
"请提醒身边的人一起遵守红灯规则。",
"等待期间请留意周围，保持安全距离。",
"现在还不能通行，请继续耐心等待。",
"遵守信号提示，绿灯亮起后再出发。",
)
ARM_REQUEST_TIMEOUT_SEC = 5
def emit(event: str, **fields):
print(json.dumps({"event": event, "at": time.time(), **fields}, ensure_ascii=False), flush=True)
def write_status(state: str, **fields):
STATE_DIR.mkdir(mode=0o700, parents=True, exist_ok=True)
tmp = STATUS_PATH.with_suffix(".tmp")
tmp.write_text(json.dumps({"state": state, "at": time.time(), **fields},
ensure_ascii=False), encoding="utf-8")
tmp.replace(STATUS_PATH)
emit(state, **fields)
def loopback_mcp_url(value: str) -> str:
parsed = urlparse(value)
if (parsed.scheme != "http" or parsed.hostname not in ("127.0.0.1", "localhost")
or parsed.path != "/mcp" or parsed.username or parsed.password
or parsed.query or parsed.fragment):
raise ValueError("MCP URL must be http://127.0.0.1:PORT/mcp")
return value
class McpClient:
def __init__(self, url: str, timeout: float = 5):
self.url = loopback_mcp_url(url)
self.timeout = timeout
self.request_id = 0
def request(self, method: str, params: dict | None = None,
request_timeout: float | None = None) -> dict:
self.request_id += 1
body = json.dumps({"jsonrpc": "2.0", "id": self.request_id,
"method": method, "params": params or {}}).encode()
req = urllib.request.Request(
self.url, data=body, method="POST",
headers={"Content-Type": "application/json", "Accept": "application/json"})
with urllib.request.urlopen(
req, timeout=self.timeout if request_timeout is None else request_timeout
) as response:
payload = json.load(response)
if not isinstance(payload, dict) or payload.get("id") != self.request_id:
raise RuntimeError(f"invalid MCP reply from {self.url}")
if "error" in payload:
raise RuntimeError(f"MCP {method} failed: {payload['error']}")
result = payload.get("result")
if not isinstance(result, dict):
raise RuntimeError(f"MCP {method} returned no result object")
return result
def tools(self) -> dict:
self.request("initialize", {"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "g1-traffic-runner",
"version": "0.1"}})
listed = self.request("tools/list").get("tools")
if not isinstance(listed, list):
raise RuntimeError("MCP tools/list returned no tools")
return {item["name"]: item for item in listed if isinstance(item, dict)
and isinstance(item.get("name"), str)}
def call(self, name: str, action: str, *, request_timeout: float | None = None,
**arguments) -> dict:
result = self.request("tools/call", {
"name": name, "arguments": {"action": action, **arguments}},
request_timeout=request_timeout)
content = result.get("content")
if not isinstance(content, list) or not content:
raise RuntimeError(f"{name}.{action} returned no content")
item = content[0]
if not isinstance(item, dict) or not isinstance(item.get("text"), str):
raise RuntimeError(f"{name}.{action} returned invalid content")
data = json.loads(item["text"])
if not isinstance(data, dict):
raise RuntimeError(f"{name}.{action} returned non-object data")
if data.get("error") or data.get("isError"):
raise RuntimeError(f"{name}.{action} failed: {data}")
return data
def require_action(tools: dict, tool: str, action: str):
definition = tools.get(tool)
if not isinstance(definition, dict):
raise RuntimeError(f"{tool} card is not available at the selected MCP endpoint")
schema = definition.get("inputSchema") or {}
allowed = ((schema.get("properties") or {}).get("action") or {}).get("enum") or []
if action not in allowed:
raise RuntimeError(f"{tool}.{action} is not advertised at the selected MCP endpoint")
def probe(g1: McpClient, tts: McpClient, motion: bool = False) -> dict:
g1_tools = g1.tools()
tts_tools = g1_tools if tts.url == g1.url else tts.tools()
for action in ("set_sequence", "info", "off"):
require_action(g1_tools, "led", action)
require_action(tts_tools, "tts", "speak")
if motion:
require_action(g1_tools, "arm", "execute")
require_action(g1_tools, "arm", "list")
listed = g1.call("arm", "list").get("actions") or []
names = {item.get("name") for item in listed if isinstance(item, dict)}
if not {"high wave", "reject"}.issubset(names):
raise RuntimeError("required arm gestures are not available")
info = g1.call("led", "info")
return {"g1_url": g1.url, "tts_url": tts.url,
"led_state": info.get("state"), "led_cycle_status": info.get("cycle_status"),
"tts_available": True, "arm_available": motion}
def stopped() -> bool:
return STOP_PATH.exists()
def runner_active() -> bool:
if not LOCK_PATH.exists():
return False
with LOCK_PATH.open("a+") as lock:
try:
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
return True
fcntl.flock(lock, fcntl.LOCK_UN)
return False
def wait_until(deadline: float):
while not stopped():
remaining = deadline - time.monotonic()
if remaining <= 0:
return
time.sleep(min(remaining, 0.1))
raise InterruptedError("stop requested")
def stage_events(duration: int, cycle: int, stage: str, motion: bool) -> list:
"""Scheduled offsets; the LED card remains the source of actual stage time."""
lines = GREEN_LINES if stage == "green" else RED_LINES
events = []
for slot in range(30):
offset = 0.3 + 10 * slot
if duration - offset < 8:
break
phrase = lines[0] if slot == 0 else lines[1 + (cycle + slot - 1) % (len(lines) - 1)]
events.append((offset, "speech", phrase))
if motion:
gesture = "high wave" if stage == "green" else "reject"
arm_interval = 15 if stage == "green" else 30
arm_reserve = 10 if stage == "green" else 28
for slot in range(6):
offset = (2 if stage == "green" else 1) + arm_interval * slot
if duration - offset < arm_reserve:
break
events.append((offset, "arm", gesture))
return sorted(events, key=lambda item: item[0])
def run(args, g1: McpClient, tts: McpClient):
STATE_DIR.mkdir(mode=0o700, parents=True, exist_ok=True)
with LOCK_PATH.open("w") as lock:
try:
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
raise RuntimeError("another traffic-light runner is active") from None
checked = probe(g1, tts, motion=args.motion)
if checked["led_cycle_status"] in ("running", "paused"):
raise RuntimeError("LED already has an active sequence")
STOP_PATH.unlink(missing_ok=True)
write_status("starting", pid=os.getpid(), green=args.green,
red=args.red, cycles=args.cycles, motion=args.motion)
sequence = [
{**GREEN, "duration_sec": args.green},
{**YELLOW, "duration_sec": 3},
{**RED, "duration_sec": args.red},
]
led_attempted = False
try:
led_attempted = True
result = g1.call("led", "set_sequence", sequence=sequence,
repeat_count=args.cycles, end_behavior="off")
if result.get("cycle_status") != "running" or result.get("last_error"):
raise RuntimeError(f"LED sequence was not accepted: {result}")
info = g1.call("led", "info")
if info.get("cycle_status") != "running" or info.get("last_error"):
raise RuntimeError(f"LED failed after start: {info}")
anchor = time.monotonic() - float(info["elapsed_sec"])
write_status("running", pid=os.getpid(), anchor=anchor, motion=args.motion)
period = args.green + 3 + args.red
for cycle in range(args.cycles):
for stage, offset, duration, stage_index in (
("green", 0, args.green, 1),
("red", args.green + 3, args.red, 3),
):
start = cycle * period + offset
end = start + duration
for event_offset, kind, value in stage_events(
duration, cycle, stage, args.motion):
wait_until(anchor + start + event_offset)
if stopped():
raise InterruptedError("stop requested")
info = g1.call("led", "info")
now_elapsed = time.monotonic() - anchor
remaining = end - now_elapsed
reserve = (8 if kind == "speech" else
10 if stage == "green" else 28)
if (info.get("cycle_status") != "running"
or info.get("last_error")
or info.get("cycle_index") != cycle + 1
or info.get("stage_index") != stage_index
or remaining < reserve):
emit("event_skipped", kind=kind, stage=stage,
cycle=cycle + 1, led_status=info.get("cycle_status"),
stage_index=info.get("stage_index"),
remaining=round(remaining, 3))
continue
if kind == "speech":
reply = tts.call("tts", "speak", text=value)
if reply.get("ret") not in (None, 0) or reply.get(
"status") in ("failed", "error", "cancelled"):
raise RuntimeError(f"TTS rejected speech: {reply}")
emit("speech_requested", stage=stage, cycle=cycle + 1,
phrase=value, tts_status=reply.get("status"),
action_id=reply.get("action_id"))
elif kind == "arm":
try:
reply = g1.call("arm", "execute", gesture=value,
request_timeout=ARM_REQUEST_TIMEOUT_SEC)
emit("arm_request_result", stage=stage, cycle=cycle + 1,
gesture=value, ret=reply.get("ret"),
action_id=reply.get("action_id"))
except Exception as exc:
emit("arm_request_unconfirmed", stage=stage,
cycle=cycle + 1, gesture=value, error=str(exc))
wait_until(anchor + args.cycles * period + 0.3)
info = g1.call("led", "info")
if info.get("cycle_status") != "completed" or info.get("last_error"):
raise RuntimeError(f"LED completion not confirmed: {info}")
write_status("completed", pid=os.getpid(), elapsed_sec=info.get("elapsed_sec"))
try:
reply = tts.call("tts", "speak",
text="红绿灯演示完成。若要退出这个技能，请说退出红绿灯技能。")
emit("completion_speech_requested", tts_status=reply.get("status"),
action_id=reply.get("action_id"))
except Exception as exc:
emit("completion_speech_failed", error=str(exc))
except InterruptedError:
if led_attempted:
try:
g1.call("led", "off")
except Exception as exc:
write_status("failed", pid=os.getpid(),
error=f"stop_requested_but_led_off_failed: {exc}")
raise
write_status("stopped", pid=os.getpid())
except Exception as exc:
if led_attempted:
try:
g1.call("led", "off")
except Exception as off_exc:
emit("led_off_failed", error=str(off_exc))
write_status("failed", pid=os.getpid(), error=str(exc))
raise
finally:
STOP_PATH.unlink(missing_ok=True)
def main(source=None):
parser = argparse.ArgumentParser()
parser.add_argument("command", choices=("probe", "launch", "run", "stop", "status"))
parser.add_argument("--g1-url", default="http://127.0.0.1:15701/mcp")
parser.add_argument("--tts-url", default="http://127.0.0.1:15720/mcp")
parser.add_argument("--green", type=int)
parser.add_argument("--red", type=int)
parser.add_argument("--cycles", type=int)
parser.add_argument("--motion", action="store_true",
help="Enable preconfirmed arm gestures: green 15s, red 30s apart")
args = parser.parse_args()
if args.command == "stop":
if runner_active():
STOP_PATH.touch()
emit("stop_requested")
else:
emit("not_running")
return
if args.command == "status":
status = (json.loads(STATUS_PATH.read_text(encoding="utf-8"))
if STATUS_PATH.exists() else {"state": "not_started"})
if status.get("state") in ("starting", "running") and not runner_active():
status = {**status, "state": "failed", "reason": "runner_not_alive"}
print(json.dumps(status, ensure_ascii=False))
return
if args.command in ("launch", "run") and not (
args.green is not None and 20 <= args.green <= 300
and args.red is not None and 30 <= args.red <= 300
and args.cycles is not None and 1 <= args.cycles <= 20
):
parser.error("launch/run requires --green 20..300, --red 30..300 and --cycles 1..20")
if args.command == "launch":
if runner_active():
raise RuntimeError("another traffic-light runner is active")
if source is None:
raise RuntimeError("launch requires the full inline source")
g1 = McpClient(args.g1_url)
tts = McpClient(args.tts_url)
print(json.dumps({"state": "preflight", **probe(g1, tts, motion=args.motion)},
ensure_ascii=False), flush=True)
command = [sys.executable, "-c", source, "run",
"--green", str(args.green), "--red", str(args.red),
"--cycles", str(args.cycles),
"--g1-url", loopback_mcp_url(args.g1_url),
"--tts-url", loopback_mcp_url(args.tts_url)]
if args.motion:
command.append("--motion")
with LOG_PATH.open("wb") as log:
child = subprocess.Popen(
command, stdin=subprocess.DEVNULL, stdout=log,
stderr=subprocess.STDOUT, start_new_session=True, close_fds=True)
print(json.dumps({"state": "spawned", "pid": child.pid,
"log": str(LOG_PATH)}, ensure_ascii=False))
return
g1 = McpClient(args.g1_url)
tts = McpClient(args.tts_url)
if args.command == "probe":
print(json.dumps(probe(g1, tts, motion=args.motion), ensure_ascii=False))
else:
run(args, g1, tts)
if __name__ == "__main__":
main()
'''
if hashlib.sha256(source.encode("utf-8")).hexdigest() != expected:
raise RuntimeError("内置 runner 内容校验失败")
namespace = {"__name__": "embedded_traffic_runner"}
exec(compile(source, "<traffic-inline>", "exec"), namespace)
namespace["main"](source=source)
G1_TRAFFIC_RUNNER
```
{
"type": "object",
"properties": {
"default_cycles": {
"type": "integer",
"title": "默认循环次数",
"default": 1,
"maximum": 20,
"minimum": 1,
"description": "绿灯、黄灯、红灯构成一轮"
},
"default_red_duration_s": {
"type": "integer",
"title": "默认红灯秒数",
"default": 30,
"maximum": 300,
"minimum": 30,
"description": "红灯阶段持续时间"
},
"default_green_duration_s": {
"type": "integer",
"title": "默认绿灯秒数",
"default": 30,
"maximum": 300,
"minimum": 20,
"description": "绿灯阶段持续时间"
}
},
"additionalProperties": false
}