File size: 7,688 Bytes
e26a40b 9cbd5f0 e26a40b 9cbd5f0 e26a40b 9cbd5f0 e26a40b 9cbd5f0 e26a40b 9cbd5f0 e26a40b 9cbd5f0 e26a40b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 | """
Substitute runtime intent parameters into recorded trajectory steps.
Used by the Gradio preview and mirrored in Pocket Automator's ParameterBinder at replay time.
"""
from __future__ import annotations
import copy
import json
from typing import Any
from src.paths import DATA_DIR
SKILL_SCHEMAS_PATH = DATA_DIR / "skill_schemas.json"
_SEARCH_INPUT_SUFFIXES = (
"search_input",
"search_src_text",
"search_container",
"search_box",
"search_bar",
"search",
)
_MESSAGE_BODY_SUFFIXES = (
"entry",
"message",
"message_text",
)
_COMPOSE_BODY_SUFFIXES = (
"editor",
"body",
)
_COMPOSE_EXCLUDED_SUFFIXES = (
"subject",
"editor",
)
def _is_compose_recipient_step(step: dict[str, Any]) -> bool:
action = step.get("action") or {}
if action.get("type") != "set_text":
return False
package_name = (step.get("packageName") or "").lower()
if "gm" not in package_name and "gmail" not in package_name:
return False
resource_id = action.get("resourceId") or ""
return not any(_suffix_matches(resource_id, suffix) for suffix in _COMPOSE_EXCLUDED_SUFFIXES)
def load_skill_schemas() -> dict[str, Any]:
return json.loads(SKILL_SCHEMAS_PATH.read_text(encoding="utf-8"))
def load_skill_schema(skill: str) -> dict[str, Any]:
schemas = load_skill_schemas()
if skill not in schemas:
raise KeyError(f"Unknown skill schema: {skill!r}")
return schemas[skill]
def _resource_id_suffix(resource_id: str | None) -> str:
if not resource_id:
return ""
return resource_id.rsplit("/", 1)[-1].lower()
def _suffix_matches(resource_id: str | None, suffix: str) -> bool:
if not suffix:
return False
needle = suffix.lower()
rid = _resource_id_suffix(resource_id)
return rid == needle or rid.endswith(needle)
def is_search_input_step(step: dict[str, Any]) -> bool:
action = step.get("action") or {}
if action.get("type") != "set_text":
return False
resource_id = action.get("resourceId") or ""
if any(_suffix_matches(resource_id, suffix) for suffix in _SEARCH_INPUT_SUFFIXES):
return True
package_name = (step.get("packageName") or "").lower()
class_name = (action.get("className") or "").lower()
return "spotify" in package_name and "edittext" in class_name
def find_preceding_search_step(steps: list[dict[str, Any]], before_index: int) -> dict[str, Any] | None:
for index in range(before_index - 1, -1, -1):
step = steps[index]
action = step.get("action") or {}
if action.get("type") != "set_text":
continue
value = action.get("value")
if not value or not str(value).strip():
continue
if is_search_input_step(step):
return step
return None
def _binding_matches(
binding: dict[str, Any],
step: dict[str, Any],
steps: list[dict[str, Any]],
index: int,
) -> bool:
action = step.get("action") or {}
if action.get("type") != binding.get("action"):
return False
resource_id_suffix = binding.get("resource_id_suffix")
if resource_id_suffix and not _suffix_matches(action.get("resourceId"), resource_id_suffix):
return False
if binding.get("after_search"):
return find_preceding_search_step(steps, index) is not None
package_contains = binding.get("package_contains")
if package_contains and package_contains.lower() not in (step.get("packageName") or "").lower():
return False
if binding.get("compose_recipient"):
return _is_compose_recipient_step(step)
return True
def _apply_binding(action: dict[str, Any], binding: dict[str, Any], value: str) -> dict[str, Any]:
updated = dict(action)
if binding.get("action") == "set_text":
updated["value"] = value
elif binding.get("action") == "click":
updated["text"] = value
if binding.get("set_content_description"):
updated["contentDescription"] = value
return updated
def apply_parameters(
trajectory: dict[str, Any],
skill: str,
parameters: dict[str, str] | None,
) -> dict[str, Any]:
"""Return a copy of the trajectory with bound parameter values substituted."""
if not parameters:
return trajectory
schema = load_skill_schema(skill)
bindings = schema.get("bindings") or []
if not bindings:
return trajectory
result = copy.deepcopy(trajectory)
steps = result.get("steps") or []
for index, step in enumerate(steps):
action = step.get("action")
if not isinstance(action, dict):
continue
for binding in bindings:
param_name = binding.get("parameter")
if not param_name or param_name not in parameters:
continue
if not _binding_matches(binding, step, steps, index):
continue
step["action"] = _apply_binding(action, binding, str(parameters[param_name]))
action = step["action"]
return result
def _self_test() -> None:
from src.skill_router import load_trajectory
whatsapp = load_trajectory("whatsapp_send_message")
bound = apply_parameters(
whatsapp,
"whatsapp_send_message",
{"contact": "mom", "message": "i'm on my way"},
)
search_values = [
step["action"]["value"]
for step in bound["steps"]
if step["action"]["type"] == "set_text"
and _suffix_matches(step["action"].get("resourceId"), "search_input")
]
message_values = [
step["action"]["value"]
for step in bound["steps"]
if step["action"]["type"] == "set_text"
and _suffix_matches(step["action"].get("resourceId"), "entry")
]
assert search_values and all(value == "mom" for value in search_values), search_values
assert message_values and all(value == "i'm on my way" for value in message_values), message_values
post_search_clicks = [
step["action"]
for index, step in enumerate(bound["steps"])
if step["action"]["type"] == "click"
and find_preceding_search_step(bound["steps"], index) is not None
]
assert post_search_clicks and all(click.get("text") == "mom" for click in post_search_clicks)
gmail = load_trajectory("gmail_send_email")
bound_gmail = apply_parameters(
gmail,
"gmail_send_email",
{"recipient": "alice@example.com", "message": "running late"},
)
recipient_values = [
step["action"]["value"]
for step in bound_gmail["steps"]
if _is_compose_recipient_step(step)
]
body_values = [
step["action"]["value"]
for step in bound_gmail["steps"]
if step["action"]["type"] == "set_text"
and _suffix_matches(step["action"].get("resourceId"), "editor")
]
assert recipient_values and all(value == "alice@example.com" for value in recipient_values), recipient_values
assert body_values and all(value == "running late" for value in body_values), body_values
youtube = load_trajectory("youtube_search")
bound_youtube = apply_parameters(
youtube,
"youtube_search",
{"query": "pasta recipes"},
)
youtube_search_values = [
step["action"]["value"]
for step in bound_youtube["steps"]
if step["action"]["type"] == "set_text"
and _suffix_matches(step["action"].get("resourceId"), "search_edit_text")
]
assert youtube_search_values and all(value == "pasta recipes" for value in youtube_search_values), youtube_search_values
if __name__ == "__main__":
_self_test()
print("parameter_binder: ok")
|