diff options
Diffstat (limited to 'src/sailfish_devel_mcp/tools.py')
| -rw-r--r-- | src/sailfish_devel_mcp/tools.py | 1282 |
1 files changed, 1282 insertions, 0 deletions
diff --git a/src/sailfish_devel_mcp/tools.py b/src/sailfish_devel_mcp/tools.py new file mode 100644 index 0000000..1061e6c --- /dev/null +++ b/src/sailfish_devel_mcp/tools.py @@ -0,0 +1,1282 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path, PurePosixPath +import re +import shlex +import shutil +from typing import Any, Callable + +from .config import Config, DeviceConfig +from .runner import ( + CommandResult, + remote_command, + run, + scp_from_argv, + scp_to_argv, + ssh_argv, + truncate, + user_bus_env, +) + + +ToolHandler = Callable[[dict[str, Any]], dict[str, Any]] + + +@dataclass(frozen=True) +class Tool: + spec: dict[str, Any] + handler: ToolHandler + + +def build_registry(config: Config) -> dict[str, Tool]: + tools = [ + Tool(_spec_devices(), lambda args: handle_devices(config, args)), + Tool(_spec_device_journal(), lambda args: handle_device_journal(config, args)), + Tool(_spec_device_topmost_pid(), lambda args: handle_device_topmost_pid(config, args)), + Tool(_spec_device_proc_maps(), lambda args: handle_device_proc_maps(config, args)), + Tool( + _spec_device_lipstick_screenshot(), + lambda args: handle_device_lipstick_screenshot(config, args), + ), + Tool( + _spec_device_touch(), + lambda args: handle_device_touch(config, args), + ), + Tool( + _spec_device_user_bus_call(), + lambda args: handle_device_user_bus_call(config, args), + ), + Tool(_spec_device_install_rpm(), lambda args: handle_device_install_rpm(config, args)), + Tool( + _spec_device_restart_service(), + lambda args: handle_device_restart_service(config, args), + ), + Tool(_spec_build_rpm(), lambda args: handle_build_rpm(config, args)), + Tool(_spec_obs_results(), lambda args: handle_obs_results(config, args)), + Tool(_spec_obs_buildlog(), lambda args: handle_obs_buildlog(config, args)), + Tool(_spec_repo_status(), lambda args: handle_repo_status(config, args)), + Tool(_spec_repo_find(), lambda args: handle_repo_find(config, args)), + Tool(_spec_spec_summary(), lambda args: handle_spec_summary(config, args)), + Tool( + _spec_qml_find_translations(), + lambda args: handle_qml_find_translations(config, args), + ), + Tool( + _spec_qml_check_translator_ternaries(), + lambda args: handle_qml_check_translator_ternaries(config, args), + ), + ] + return {tool.spec["name"]: tool for tool in tools} + + +def handle_devices(config: Config, args: dict[str, Any]) -> dict[str, Any]: + structured = config.public_dict() + text = "\n".join( + ( + f"{name}: {device.ssh_target}" + f" user={device.username}" + f"{f' arch={device.architecture}' if device.architecture else ''}" + f"{f' release={device.release}' if device.release else ''}" + ) + for name, device in sorted(config.devices.items(), key=lambda item: item[0]) + ) + return ok_text(text or "No configured devices", structured) + + +def handle_device_journal(config: Config, args: dict[str, Any]) -> dict[str, Any]: + device = _device(config, args) + lines = _int_arg(args, "lines", default=200, minimum=1, maximum=5000) + timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=600) + + command = ["journalctl", "--no-pager", "-n", str(lines)] + unit = _optional_str(args, "unit") + since = _optional_str(args, "since") + grep = _optional_str(args, "grep") + if unit: + command += ["-u", unit] + if since: + command += ["--since", since] + result = _run_ssh(config, device, command, timeout=timeout) + stdout = result.stdout + if grep: + stdout = "\n".join(line for line in stdout.splitlines() if grep in line) + result = CommandResult(result.argv, result.returncode, stdout, result.stderr) + return command_result("device journal", result) + + +def handle_device_topmost_pid(config: Config, args: dict[str, Any]) -> dict[str, Any]: + device = _device(config, args) + timeout = _int_arg(args, "timeout", default=20, minimum=1, maximum=120) + command = user_bus_env(device) + [ + "dbus-send", + "--session", + "--print-reply", + "--dest=org.nemomobile.lipstick", + "/", + "org.freedesktop.DBus.Properties.Get", + "string:org.nemomobile.compositor", + "string:privateTopmostWindowProcessId", + ] + result = _run_ssh(config, device, command, timeout=timeout) + pid = _parse_dbus_integer(result.stdout) + structured = result.public_dict() + structured["pid"] = pid + text = f"topmost PID: {pid}" if pid is not None else _command_text("topmost PID", result) + return { + "content": [{"type": "text", "text": text}], + "structuredContent": structured, + "isError": not result.ok or pid is None, + } + + +def handle_device_proc_maps(config: Config, args: dict[str, Any]) -> dict[str, Any]: + device = _device(config, args) + pid = _int_arg(args, "pid", minimum=1) + contains = _optional_str(args, "contains") + max_lines = _int_arg(args, "max_lines", default=200, minimum=1, maximum=5000) + timeout = _int_arg(args, "timeout", default=20, minimum=1, maximum=120) + + result = _run_ssh(config, device, ["cat", f"/proc/{pid}/maps"], timeout=timeout) + lines = result.stdout.splitlines() + if contains: + lines = [line for line in lines if contains in line] + limited = lines[:max_lines] + stdout = "\n".join(limited) + if len(lines) > max_lines: + stdout += f"\n[truncated after {max_lines} lines]" + filtered = CommandResult(result.argv, result.returncode, stdout, result.stderr) + return command_result("process maps", filtered, {"matched_lines": len(lines), "pid": pid}) + + +def handle_device_lipstick_screenshot(config: Config, args: dict[str, Any]) -> dict[str, Any]: + device = _device(config, args) + timestamp = datetime.utcnow().strftime("%Y%m%d-%H%M%S") + home_path = _device_home_path(device) + remote_path = _optional_str(args, "remote_path") or ( + f"{home_path}/Pictures/Screenshots/lipstick-{timestamp}.png" + ) + local_path_arg = _optional_str(args, "local_path") + privileged = _bool_arg(args, "privileged", default=True) + timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=120) + + inner = remote_command( + user_bus_env(device) + + [ + "dbus-send", + "--session", + "--print-reply", + "--dest=org.nemomobile.lipstick", + "/org/nemomobile/lipstick/screenshot", + "org.nemomobile.lipstick.saveScreenshot", + f"string:{remote_path}", + ] + ) + prepare_dir = _screenshot_prepare_command(device, remote_path) + dbus_call = f"sg privileged -c {shlex.quote(inner)}" if privileged else inner + remote = f"{prepare_dir} && {dbus_call}" + result = run(ssh_argv(device, config.paths.ssh_config, remote), timeout=timeout) + + structured: dict[str, Any] = result.public_dict() + structured["remote_path"] = remote_path + if result.ok and local_path_arg: + local_path = _safe_output_path(config, local_path_arg) + local_path.parent.mkdir(parents=True, exist_ok=True) + pull = run( + scp_from_argv(device, config.paths.ssh_config, remote_path, local_path), + timeout=timeout, + ) + structured["pull"] = pull.public_dict() + structured["local_path"] = str(local_path) + if not pull.ok: + return command_result("pull screenshot", pull, structured) + return command_result("lipstick screenshot", result, structured) + + +def handle_device_touch(config: Config, args: dict[str, Any]) -> dict[str, Any]: + device = _device(config, args) + action = _enum_arg(args, "action", ["discover", "tap", "swipe"]) + timeout = _int_arg(args, "timeout", default=10, minimum=1, maximum=120) + input_device = _optional_str(args, "input_device") + if input_device and not re.fullmatch(r"/dev/input/event\d+", input_device): + return tool_error("input_device must look like /dev/input/event<N>") + + if action == "discover": + include_evdev_trace = _bool_arg(args, "include_evdev_trace", default=False) + remote = _touch_discover_command(include_evdev_trace) + return command_result( + "touchscreen discovery", + run(ssh_argv(device, config.paths.ssh_config, remote), timeout=timeout), + ) + + values: dict[str, str | int] = {"ACTION": action} + if input_device: + values["INPUT_DEVICE"] = input_device + + if action == "tap": + values["X"] = _int_arg(args, "x", minimum=0, maximum=10000) + values["Y"] = _int_arg(args, "y", minimum=0, maximum=10000) + values["HOLD_MS"] = _int_arg(args, "hold_ms", default=80, minimum=1, maximum=5000) + else: + values["START_X"] = _int_arg(args, "start_x", minimum=0, maximum=10000) + values["START_Y"] = _int_arg(args, "start_y", minimum=0, maximum=10000) + values["END_X"] = _int_arg(args, "end_x", minimum=0, maximum=10000) + values["END_Y"] = _int_arg(args, "end_y", minimum=0, maximum=10000) + values["DURATION_MS"] = _int_arg(args, "duration_ms", default=300, minimum=1, maximum=10000) + values["STEPS"] = _int_arg(args, "steps", default=12, minimum=1, maximum=200) + + remote = _touch_inject_command(values) + structured = { + "action": action, + "input_device": input_device, + } + return command_result( + "device touch", + run(ssh_argv(device, config.paths.ssh_config, remote), timeout=timeout), + structured, + ) + + +def handle_device_user_bus_call(config: Config, args: dict[str, Any]) -> dict[str, Any]: + device = _device(config, args) + destination = _str_arg(args, "destination") + path = _str_arg(args, "path") + interface = _str_arg(args, "interface") + member = _str_arg(args, "member") + dbus_args = args.get("arguments") or [] + if not isinstance(dbus_args, list) or not all(isinstance(item, str) for item in dbus_args): + return tool_error("arguments must be a list of dbus-send argument strings") + timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=300) + command = user_bus_env(device) + [ + "dbus-send", + "--session", + "--print-reply", + f"--dest={destination}", + path, + f"{interface}.{member}", + *dbus_args, + ] + return command_result("user bus call", _run_ssh(config, device, command, timeout=timeout)) + + +def handle_device_install_rpm(config: Config, args: dict[str, Any]) -> dict[str, Any]: + device = _device(config, args) + rpm_path = _safe_input_path(config, _str_arg(args, "rpm_path"), allow_tmp=True) + remote_path = _optional_str(args, "remote_path") or f"/tmp/{rpm_path.name}" + installer = _enum_arg(args, "installer", ["pkcon", "rpm"], default="pkcon") + timeout = _int_arg(args, "timeout", default=180, minimum=1, maximum=1200) + + copy_result = run( + scp_to_argv(device, config.paths.ssh_config, rpm_path, remote_path), + timeout=timeout, + ) + if not copy_result.ok: + return command_result("copy RPM to device", copy_result) + + if installer == "pkcon": + remote = ["pkcon", "install-local", "-y", remote_path] + else: + remote = ["rpm", "-Uvh", "--replacepkgs", remote_path] + install_result = _run_ssh(config, device, remote, timeout=timeout) + structured = { + "copy": copy_result.public_dict(), + "install": install_result.public_dict(), + "remote_path": remote_path, + "installer": installer, + } + return command_result("install RPM", install_result, structured) + + +def handle_device_restart_service(config: Config, args: dict[str, Any]) -> dict[str, Any]: + device = _device(config, args) + unit = _service_unit_arg(args, "unit") + action = _enum_arg(args, "action", ["restart", "start", "stop", "status"], default="restart") + mode = _enum_arg(args, "mode", ["system", "user"], default="system") + timeout = _int_arg(args, "timeout", default=60, minimum=1, maximum=300) + if mode == "user": + command = user_bus_env(device) + ["systemctl", "--user", action, unit] + else: + command = ["systemctl", action, unit] + return command_result( + f"{mode} service {action}", + _run_ssh(config, device, command, timeout=timeout), + ) + + +def handle_build_rpm(config: Config, args: dict[str, Any]) -> dict[str, Any]: + project_path = _safe_input_path(config, _str_arg(args, "project_path"), allow_tmp=False) + device = config.device(_optional_str(args, "device")) if args.get("device") else None + script = config.paths.build_sailfishos + if not script.exists(): + return tool_error(f"build helper not found: {script}") + + command = ["python3", str(script), "--project-dir", str(project_path)] + release = _optional_str(args, "release") or (device.release if device else None) + arches = args.get("arch") + artifacts_dir = _optional_str(args, "artifacts_dir") + if config.paths.local_sdk: + command += ["--local-sdk", str(config.paths.local_sdk)] + if release: + command += ["--release", release] + if isinstance(arches, str): + command += ["--arch", arches] + elif isinstance(arches, list): + for arch in arches: + if not isinstance(arch, str): + return tool_error("arch must be a string or list of strings") + command += ["--arch", arch] + elif arches is not None: + return tool_error("arch must be a string or list of strings") + elif device and device.architecture: + command += ["--arch", device.architecture] + if artifacts_dir: + output = _safe_output_path(config, artifacts_dir) + command += ["--artifacts-dir", str(output)] + if _bool_arg(args, "all_arches", default=False): + command.append("--all") + if _bool_arg(args, "clean", default=False): + command.append("--clean") + if _bool_arg(args, "debug", default=False): + command.append("--debug") + if _bool_arg(args, "no_pull", default=False): + command.append("--no-pull") + for local_dir in _string_list_arg(args, "local_rpms_dir"): + command += ["--local-rpms-dir", str(_safe_input_path(config, local_dir, allow_tmp=True))] + timeout = _int_arg(args, "timeout", default=3600, minimum=1, maximum=21600) + return command_result("build Sailfish RPM", run(command, timeout=timeout)) + + +def handle_obs_results(config: Config, args: dict[str, Any]) -> dict[str, Any]: + project = _str_arg(args, "project") + package = _optional_str(args, "package") + api_alias = _optional_str(args, "api_alias") or config.paths.osc_api_alias + command = ["osc"] + if api_alias: + command += ["-A", api_alias] + command += ["results", project] + if package: + command.append(package) + timeout = _int_arg(args, "timeout", default=60, minimum=1, maximum=600) + return command_result("OBS results", run(command, timeout=timeout)) + + +def handle_obs_buildlog(config: Config, args: dict[str, Any]) -> dict[str, Any]: + project = _str_arg(args, "project") + package = _str_arg(args, "package") + repository = _str_arg(args, "repository") + arch = _str_arg(args, "arch") + api_alias = _optional_str(args, "api_alias") or config.paths.osc_api_alias + timeout = _int_arg(args, "timeout", default=90, minimum=1, maximum=1800) + command = ["osc"] + if api_alias: + command += ["-A", api_alias] + command += ["remotebuildlog", project, package, repository, arch] + return command_result("OBS build log", run(command, timeout=timeout)) + + +def handle_repo_status(config: Config, args: dict[str, Any]) -> dict[str, Any]: + path = _safe_input_path(config, _optional_str(args, "path") or ".", allow_tmp=False) + timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=120) + result = run(["git", "status", "--short", "--branch"], cwd=path, timeout=timeout) + return command_result("git status", result, {"path": str(path)}) + + +def handle_repo_find(config: Config, args: dict[str, Any]) -> dict[str, Any]: + path = _safe_input_path(config, _optional_str(args, "path") or ".", allow_tmp=False) + query = _str_arg(args, "query") + max_count = _int_arg(args, "max_count", default=100, minimum=1, maximum=1000) + fixed_strings = _bool_arg(args, "fixed_strings", default=True) + timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=300) + + if shutil.which("rg"): + command = ["rg", "--line-number", "--hidden", "--glob", "!.git"] + if fixed_strings: + command.append("--fixed-strings") + command += ["--max-count", str(max_count), query, str(path)] + else: + command = ["grep", "-R", "-n", query, str(path)] + result = run(command, timeout=timeout) + if result.returncode == 1 and not result.stdout: + return ok_text("no matches", {"path": str(path), "query": query, "matches": 0}) + return command_result("repo find", result, {"path": str(path), "query": query}) + + +def handle_spec_summary(config: Config, args: dict[str, Any]) -> dict[str, Any]: + path_arg = _optional_str(args, "spec_path") + if path_arg: + spec_path = _safe_input_path(config, path_arg, allow_tmp=False) + else: + repo = _safe_input_path(config, _optional_str(args, "repo_path") or ".", allow_tmp=False) + specs = sorted((repo / "rpm").glob("*.spec")) + if not specs: + return tool_error(f"no rpm/*.spec file found under {repo}") + spec_path = specs[0] + + if not spec_path.exists(): + return tool_error(f"spec file does not exist: {spec_path}") + data = _parse_spec(spec_path) + text_lines = [f"{key}: {value}" for key, value in data.items() if value] + return ok_text("\n".join(text_lines), {"spec_path": str(spec_path), "summary": data}) + + +def handle_qml_find_translations(config: Config, args: dict[str, Any]) -> dict[str, Any]: + path = _safe_input_path(config, _optional_str(args, "path") or ".", allow_tmp=False) + timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=300) + if not shutil.which("rg"): + return tool_error("rg is required for qml_find_translations") + command = [ + "rg", + "--line-number", + "--glob", + "*.qml", + r"qsTrId|//%|//:", + str(path), + ] + result = run(command, timeout=timeout) + if result.returncode == 1 and not result.stdout: + return ok_text("no QML translation markers found", {"path": str(path), "matches": 0}) + return command_result("QML translations", result, {"path": str(path)}) + + +def handle_qml_check_translator_ternaries(config: Config, args: dict[str, Any]) -> dict[str, Any]: + path = _safe_input_path(config, _optional_str(args, "path") or ".", allow_tmp=False) + files = [path] if path.is_file() and path.suffix == ".qml" else sorted(path.rglob("*.qml")) + findings: list[dict[str, Any]] = [] + for qml in files: + if ".git" in qml.parts: + continue + try: + lines = qml.read_text(encoding="utf-8").splitlines() + except UnicodeDecodeError: + continue + for index, line in enumerate(lines, start=1): + if "qsTrId(" in line and "?" in line and ":" in line: + findings.append( + { + "path": str(qml), + "line": index, + "text": line.strip(), + "message": ( + "ternary qsTrId expression should give each branch " + "its own translator comment and source text" + ), + } + ) + if not findings: + return ok_text("no ternary translation issues found", {"path": str(path), "findings": []}) + text = "\n".join( + f"{item['path']}:{item['line']}: {item['message']}\n {item['text']}" + for item in findings + ) + return { + "content": [{"type": "text", "text": text}], + "structuredContent": {"path": str(path), "findings": findings}, + "isError": True, + } + + +def _run_ssh( + config: Config, + device: DeviceConfig, + command: list[str], + *, + timeout: int, +) -> CommandResult: + return run( + ssh_argv(device, config.paths.ssh_config, remote_command(command)), + timeout=timeout, + ) + + +def _screenshot_prepare_command(device: DeviceConfig, remote_path: str) -> str: + remote_dir = str(PurePosixPath(remote_path).parent) + home_path = _device_home_path(device) + script = f""" +set -e +home={shlex.quote(home_path)} +remote_dir={shlex.quote(remote_dir)} +owner=$(stat -Lc %U "$home") +group=$(stat -Lc %G "$home") +if [ "$remote_dir" = "$home/Pictures/Screenshots" ]; then + if [ ! -d "$home/Pictures" ]; then + install -d -m 775 -o "$owner" -g "$group" "$home/Pictures" + fi + if getent group privileged >/dev/null 2>&1; then + screenshot_group=privileged + else + screenshot_group=$group + fi + if [ ! -d "$remote_dir" ]; then + install -d -m 755 -o "$owner" -g "$screenshot_group" "$remote_dir" + fi +else + if [ ! -d "$remote_dir" ]; then + install -d -m 755 -o "$owner" -g "$group" "$remote_dir" + fi +fi +""".strip() + return remote_command(["sh", "-lc", script]) + + +def _touch_discover_command(include_evdev_trace: bool) -> str: + evdev_trace = """ +if command -v evdev_trace >/dev/null 2>&1; then + echo "### evdev_trace -i" + if command -v timeout >/dev/null 2>&1; then + timeout 5 evdev_trace -i || true + else + evdev_trace -i & + pid=$! + sleep 5 + kill "$pid" >/dev/null 2>&1 || true + wait "$pid" >/dev/null 2>&1 || true + fi + echo +fi +""".strip() + script = f""" +set -e +{evdev_trace if include_evdev_trace else ""} +echo "### /proc/bus/input/devices" +cat /proc/bus/input/devices +""".strip() + return remote_command(["sh", "-lc", script]) + + +def _touch_inject_command(values: dict[str, str | int]) -> str: + env = " ".join( + f"{key}={shlex.quote(str(value))}" + for key, value in sorted(values.items()) + ) + script = f""" +set -e +{env} python3 - <<'PY' +{_touch_inject_python()} +PY +""".strip() + return remote_command(["sh", "-lc", script]) + + +def _touch_inject_python() -> str: + return r''' +import os +import re +import struct +import sys +import time + +EV_SYN = 0 +EV_KEY = 1 +EV_ABS = 3 +SYN_REPORT = 0 +BTN_TOUCH = 0x14a +ABS_X = 0x00 +ABS_Y = 0x01 +ABS_MT_SLOT = 0x2f +ABS_MT_POSITION_X = 0x35 +ABS_MT_POSITION_Y = 0x36 +ABS_MT_TRACKING_ID = 0x39 + +TOUCH_KEYWORDS = ( + "touch", + "touchscreen", + "digitizer", + "fts", + "ft5x", + "goodix", + "synaptics", + "cyttsp", + "atmel", + "elan", + "himax", + "novatek", + "nvt", + "silead", + "gt9", + "spi3", +) + + +def die(message): + print(message, file=sys.stderr) + raise SystemExit(2) + + +def input_blocks(): + try: + text = open("/proc/bus/input/devices", encoding="utf-8").read() + except OSError as exc: + die(f"failed to read /proc/bus/input/devices: {exc}") + return [block for block in text.split("\n\n") if block.strip()] + + +def field(block, prefix): + for line in block.splitlines(): + if line.startswith(prefix): + return line[len(prefix):].strip() + return "" + + +def event_handler(block): + match = re.search(r"\bevent\d+\b", field(block, "H: Handlers=")) + return match.group(0) if match else "" + + +def name(block): + value = field(block, "N: Name=") + return value.strip('"') + + +def score(block): + lower = block.lower() + value = 0 + if any(keyword in lower for keyword in TOUCH_KEYWORDS): + value += 100 + if "b: abs=" in lower: + value += 20 + if "b: prop=" in lower: + value += 5 + if "mouse" in lower or "keyboard" in lower or "keypad" in lower: + value -= 100 + return value + + +def discover(): + devices = [] + for block in input_blocks(): + event = event_handler(block) + if not event: + continue + devices.append( + { + "path": f"/dev/input/{event}", + "name": name(block), + "score": score(block), + } + ) + devices.sort(key=lambda item: item["score"], reverse=True) + return devices + + +def choose_device(): + explicit = os.environ.get("INPUT_DEVICE", "") + if explicit: + return explicit + devices = discover() + if not devices or devices[0]["score"] <= 0: + summary = ", ".join(f"{item['path']}:{item['name']}" for item in devices) + die(f"could not identify touchscreen input device; candidates: {summary}") + return devices[0]["path"] + + +def event(fileobj, event_type, code, value): + fileobj.write(struct.pack("@llHHi", 0, 0, event_type, code, value)) + + +def sync(fileobj): + event(fileobj, EV_SYN, SYN_REPORT, 0) + + +def move(fileobj, x, y): + event(fileobj, EV_ABS, ABS_MT_POSITION_X, x) + event(fileobj, EV_ABS, ABS_MT_POSITION_Y, y) + event(fileobj, EV_ABS, ABS_X, x) + event(fileobj, EV_ABS, ABS_Y, y) + + +def down(fileobj, x, y): + event(fileobj, EV_ABS, ABS_MT_SLOT, 0) + event(fileobj, EV_ABS, ABS_MT_TRACKING_ID, int(time.time() * 1000) & 0x7fffffff) + move(fileobj, x, y) + event(fileobj, EV_KEY, BTN_TOUCH, 1) + sync(fileobj) + + +def up(fileobj): + event(fileobj, EV_ABS, ABS_MT_SLOT, 0) + event(fileobj, EV_ABS, ABS_MT_TRACKING_ID, -1) + event(fileobj, EV_KEY, BTN_TOUCH, 0) + sync(fileobj) + + +def integer(name): + try: + return int(os.environ[name]) + except KeyError: + die(f"missing {name}") + except ValueError: + die(f"{name} must be an integer") + + +action = os.environ.get("ACTION", "") +device = choose_device() +print(f"input_device={device}") +with open(device, "wb", buffering=0) as fileobj: + if action == "tap": + down(fileobj, integer("X"), integer("Y")) + time.sleep(integer("HOLD_MS") / 1000) + up(fileobj) + elif action == "swipe": + start_x = integer("START_X") + start_y = integer("START_Y") + end_x = integer("END_X") + end_y = integer("END_Y") + steps = integer("STEPS") + duration = integer("DURATION_MS") / 1000 + down(fileobj, start_x, start_y) + for step in range(1, steps + 1): + fraction = step / steps + x = round(start_x + ((end_x - start_x) * fraction)) + y = round(start_y + ((end_y - start_y) * fraction)) + move(fileobj, x, y) + sync(fileobj) + time.sleep(duration / steps) + up(fileobj) + else: + die(f"unsupported ACTION {action!r}") +print(f"{action}=ok") +'''.strip() + + +def _device_home_path(device: DeviceConfig) -> str: + return f"/home/{device.username}" + + +def command_result( + title: str, + result: CommandResult, + structured: dict[str, Any] | None = None, +) -> dict[str, Any]: + data = result.public_dict() + if structured: + data.update(structured) + return { + "content": [{"type": "text", "text": _command_text(title, result)}], + "structuredContent": data, + "isError": not result.ok, + } + + +def ok_text(text: str, structured: dict[str, Any] | None = None) -> dict[str, Any]: + return { + "content": [{"type": "text", "text": text}], + "structuredContent": structured or {}, + "isError": False, + } + + +def tool_error(message: str, structured: dict[str, Any] | None = None) -> dict[str, Any]: + return { + "content": [{"type": "text", "text": message}], + "structuredContent": structured or {"error": message}, + "isError": True, + } + + +def _command_text(title: str, result: CommandResult) -> str: + stdout, _ = truncate(result.stdout, 12000) + stderr, _ = truncate(result.stderr, 8000) + parts = [f"{title}: exit {result.returncode}"] + if stdout: + parts += ["", stdout.rstrip()] + if stderr: + parts += ["", "stderr:", stderr.rstrip()] + return "\n".join(parts) + + +def _device(config: Config, args: dict[str, Any]) -> DeviceConfig: + return config.device(_optional_str(args, "device")) + + +def _str_arg(args: dict[str, Any], name: str) -> str: + value = args.get(name) + if not isinstance(value, str) or not value: + raise ValueError(f"{name} must be a non-empty string") + return value + + +def _optional_str(args: dict[str, Any], name: str) -> str | None: + value = args.get(name) + if value is None or value == "": + return None + if not isinstance(value, str): + raise ValueError(f"{name} must be a string") + return value + + +def _int_arg( + args: dict[str, Any], + name: str, + default: int | None = None, + minimum: int | None = None, + maximum: int | None = None, +) -> int: + value = args.get(name, default) + if not isinstance(value, int): + raise ValueError(f"{name} must be an integer") + if minimum is not None and value < minimum: + raise ValueError(f"{name} must be >= {minimum}") + if maximum is not None and value > maximum: + raise ValueError(f"{name} must be <= {maximum}") + return value + + +def _bool_arg(args: dict[str, Any], name: str, default: bool = False) -> bool: + value = args.get(name, default) + if not isinstance(value, bool): + raise ValueError(f"{name} must be a boolean") + return value + + +def _enum_arg( + args: dict[str, Any], + name: str, + values: list[str], + default: str | None = None, +) -> str: + value = args.get(name, default) + if not isinstance(value, str) or value not in values: + raise ValueError(f"{name} must be one of: {', '.join(values)}") + return value + + +def _string_list_arg(args: dict[str, Any], name: str) -> list[str]: + value = args.get(name) + if value is None: + return [] + if not isinstance(value, list) or not all(isinstance(item, str) for item in value): + raise ValueError(f"{name} must be a list of strings") + return value + + +def _service_unit_arg(args: dict[str, Any], name: str) -> str: + unit = _str_arg(args, name) + if not re.fullmatch(r"[A-Za-z0-9@_.:\-]+", unit): + raise ValueError(f"{name} contains unsupported characters") + return unit + + +def _safe_input_path(config: Config, value: str, *, allow_tmp: bool) -> Path: + path = Path(value).expanduser() + if not path.is_absolute(): + path = config.paths.git_root / path + path = path.resolve(strict=False) + roots = [config.paths.git_root.resolve(strict=False)] + if allow_tmp: + roots.append(Path("/tmp").resolve(strict=False)) + if not _is_relative_to_any(path, roots): + raise ValueError(f"path is outside allowed roots: {path}") + return path + + +def _safe_output_path(config: Config, value: str) -> Path: + path = Path(value).expanduser() + if not path.is_absolute(): + path = config.paths.git_root / path + path = path.resolve(strict=False) + roots = [config.paths.git_root.resolve(strict=False), Path("/tmp").resolve(strict=False)] + if not _is_relative_to_any(path, roots): + raise ValueError(f"output path is outside allowed roots: {path}") + return path + + +def _is_relative_to_any(path: Path, roots: list[Path]) -> bool: + for root in roots: + try: + path.relative_to(root) + return True + except ValueError: + pass + return False + + +def _parse_dbus_integer(text: str) -> int | None: + match = re.search(r"\b(?:int32|uint32|int64|uint64)\s+(-?\d+)", text) + return int(match.group(1)) if match else None + + +def _parse_spec(path: Path) -> dict[str, Any]: + fields: dict[str, Any] = { + "Name": "", + "Version": "", + "Release": "", + "Summary": "", + "License": "", + "URL": "", + "BuildRequires": [], + "Requires": [], + } + for raw_line in path.read_text(encoding="utf-8", errors="replace").splitlines(): + line = raw_line.strip() + if not line or line.startswith("#"): + continue + for field in ["Name", "Version", "Release", "Summary", "License", "URL"]: + prefix = field + ":" + if line.startswith(prefix) and not fields[field]: + fields[field] = line[len(prefix) :].strip() + if line.startswith("BuildRequires:"): + fields["BuildRequires"].append(line.split(":", 1)[1].strip()) + if line.startswith("Requires:"): + fields["Requires"].append(line.split(":", 1)[1].strip()) + return fields + + +def _read_only_annotations(title: str) -> dict[str, Any]: + return { + "title": title, + "readOnlyHint": True, + "destructiveHint": False, + "idempotentHint": True, + "openWorldHint": True, + } + + +def _mutating_annotations(title: str, *, destructive: bool = False) -> dict[str, Any]: + return { + "title": title, + "readOnlyHint": False, + "destructiveHint": destructive, + "idempotentHint": False, + "openWorldHint": True, + } + + +def _object_schema(properties: dict[str, Any], required: list[str] | None = None) -> dict[str, Any]: + return { + "type": "object", + "properties": properties, + "required": required or [], + "additionalProperties": False, + } + + +def _device_prop() -> dict[str, Any]: + return {"type": "string", "description": "Configured device alias or ssh target."} + + +def _timeout_prop(default: int) -> dict[str, Any]: + return { + "type": "integer", + "minimum": 1, + "default": default, + "description": "Command timeout in seconds.", + } + + +def _spec_devices() -> dict[str, Any]: + return { + "name": "sailfish_devices", + "title": "List Sailfish Devices", + "description": "List configured Sailfish OS device aliases and SSH targets.", + "inputSchema": _object_schema({}), + "annotations": _read_only_annotations("List Sailfish Devices"), + } + + +def _spec_device_journal() -> dict[str, Any]: + return { + "name": "sailfish_device_journal", + "title": "Read Device Journal", + "description": "Read recent journalctl output from a Sailfish OS device.", + "inputSchema": _object_schema( + { + "device": _device_prop(), + "unit": {"type": "string"}, + "since": {"type": "string"}, + "grep": {"type": "string"}, + "lines": {"type": "integer", "minimum": 1, "maximum": 5000, "default": 200}, + "timeout": _timeout_prop(30), + } + ), + "annotations": _read_only_annotations("Read Device Journal"), + } + + +def _spec_device_topmost_pid() -> dict[str, Any]: + return { + "name": "sailfish_device_topmost_pid", + "title": "Topmost Window PID", + "description": "Query Lipstick for the current topmost window process id.", + "inputSchema": _object_schema({"device": _device_prop(), "timeout": _timeout_prop(20)}), + "annotations": _read_only_annotations("Topmost Window PID"), + } + + +def _spec_device_proc_maps() -> dict[str, Any]: + return { + "name": "sailfish_device_proc_maps", + "title": "Read Process Maps", + "description": "Read or filter /proc/<pid>/maps on a Sailfish OS device.", + "inputSchema": _object_schema( + { + "device": _device_prop(), + "pid": {"type": "integer", "minimum": 1}, + "contains": {"type": "string"}, + "max_lines": {"type": "integer", "minimum": 1, "maximum": 5000, "default": 200}, + "timeout": _timeout_prop(20), + }, + ["pid"], + ), + "annotations": _read_only_annotations("Read Process Maps"), + } + + +def _spec_device_lipstick_screenshot() -> dict[str, Any]: + return { + "name": "sailfish_device_lipstick_screenshot", + "title": "Lipstick Screenshot", + "description": "Ask Lipstick to save a screenshot on the device, optionally pulling it locally.", + "inputSchema": _object_schema( + { + "device": _device_prop(), + "remote_path": { + "type": "string", + "default": "/home/defaultuser/Pictures/Screenshots/lipstick-<timestamp>.png", + "description": "Lipstick accepts screenshot paths under the user home directory.", + }, + "local_path": {"type": "string"}, + "privileged": {"type": "boolean", "default": True}, + "timeout": _timeout_prop(30), + } + ), + "annotations": _mutating_annotations("Lipstick Screenshot"), + } + + +def _spec_device_touch() -> dict[str, Any]: + return { + "name": "sailfish_device_touch", + "title": "Device Touch Input", + "description": "Discover the touchscreen input device or inject tap/swipe events over SSH.", + "inputSchema": _object_schema( + { + "device": _device_prop(), + "action": { + "type": "string", + "enum": ["discover", "tap", "swipe"], + "description": "discover lists input devices; tap and swipe inject Linux input events.", + }, + "include_evdev_trace": { + "type": "boolean", + "default": False, + "description": "Also run evdev_trace -i during discovery. Disabled by default because it may block on some devices.", + }, + "input_device": { + "type": "string", + "description": "Optional explicit device path, for example /dev/input/event5.", + }, + "x": {"type": "integer", "minimum": 0, "maximum": 10000}, + "y": {"type": "integer", "minimum": 0, "maximum": 10000}, + "hold_ms": {"type": "integer", "minimum": 1, "maximum": 5000, "default": 80}, + "start_x": {"type": "integer", "minimum": 0, "maximum": 10000}, + "start_y": {"type": "integer", "minimum": 0, "maximum": 10000}, + "end_x": {"type": "integer", "minimum": 0, "maximum": 10000}, + "end_y": {"type": "integer", "minimum": 0, "maximum": 10000}, + "duration_ms": {"type": "integer", "minimum": 1, "maximum": 10000, "default": 300}, + "steps": {"type": "integer", "minimum": 1, "maximum": 200, "default": 12}, + "timeout": _timeout_prop(10), + }, + ["action"], + ), + "annotations": _mutating_annotations("Device Touch Input"), + } + + +def _spec_device_user_bus_call() -> dict[str, Any]: + return { + "name": "sailfish_device_user_bus_call", + "title": "User Bus Call", + "description": "Run a typed dbus-send method call on the defaultuser session bus.", + "inputSchema": _object_schema( + { + "device": _device_prop(), + "destination": {"type": "string"}, + "path": {"type": "string"}, + "interface": {"type": "string"}, + "member": {"type": "string"}, + "arguments": { + "type": "array", + "items": {"type": "string"}, + "description": "Raw dbus-send argument strings, for example string:/tmp/file.", + }, + "timeout": _timeout_prop(30), + }, + ["destination", "path", "interface", "member"], + ), + "annotations": _mutating_annotations("User Bus Call"), + } + + +def _spec_device_install_rpm() -> dict[str, Any]: + return { + "name": "sailfish_device_install_rpm", + "title": "Install Device RPM", + "description": "Copy a local RPM to a Sailfish OS device and install it.", + "inputSchema": _object_schema( + { + "device": _device_prop(), + "rpm_path": {"type": "string"}, + "remote_path": {"type": "string"}, + "installer": {"type": "string", "enum": ["pkcon", "rpm"], "default": "pkcon"}, + "timeout": _timeout_prop(180), + }, + ["rpm_path"], + ), + "annotations": _mutating_annotations("Install Device RPM"), + } + + +def _spec_device_restart_service() -> dict[str, Any]: + return { + "name": "sailfish_device_restart_service", + "title": "Manage Device Service", + "description": "Run systemctl start/stop/restart/status for a system or user service.", + "inputSchema": _object_schema( + { + "device": _device_prop(), + "unit": {"type": "string"}, + "action": { + "type": "string", + "enum": ["restart", "start", "stop", "status"], + "default": "restart", + }, + "mode": {"type": "string", "enum": ["system", "user"], "default": "system"}, + "timeout": _timeout_prop(60), + }, + ["unit"], + ), + "annotations": _mutating_annotations("Manage Device Service"), + } + + +def _spec_build_rpm() -> dict[str, Any]: + return { + "name": "sailfish_build_rpm", + "title": "Build Sailfish RPM", + "description": "Run the local build-sailfishos helper; paths.local_sdk is used only when it has a matching target.", + "inputSchema": _object_schema( + { + "project_path": {"type": "string"}, + "device": { + "type": "string", + "description": "Optional configured device to supply default release and architecture.", + }, + "release": {"type": "string"}, + "arch": { + "oneOf": [ + {"type": "string"}, + {"type": "array", "items": {"type": "string"}}, + ] + }, + "artifacts_dir": {"type": "string"}, + "all_arches": {"type": "boolean", "default": False}, + "clean": {"type": "boolean", "default": False}, + "debug": {"type": "boolean", "default": False}, + "no_pull": {"type": "boolean", "default": False}, + "local_rpms_dir": {"type": "array", "items": {"type": "string"}}, + "timeout": _timeout_prop(3600), + }, + ["project_path"], + ), + "annotations": _mutating_annotations("Build Sailfish RPM"), + } + + +def _spec_obs_results() -> dict[str, Any]: + return { + "name": "sailfish_obs_results", + "title": "OBS Results", + "description": "Run osc results using the configured OBS API alias when set.", + "inputSchema": _object_schema( + { + "project": {"type": "string"}, + "package": {"type": "string"}, + "api_alias": {"type": "string"}, + "timeout": _timeout_prop(60), + }, + ["project"], + ), + "annotations": _read_only_annotations("OBS Results"), + } + + +def _spec_obs_buildlog() -> dict[str, Any]: + return { + "name": "sailfish_obs_buildlog", + "title": "OBS Build Log", + "description": "Fetch an OBS remote build log with osc.", + "inputSchema": _object_schema( + { + "project": {"type": "string"}, + "package": {"type": "string"}, + "repository": {"type": "string"}, + "arch": {"type": "string"}, + "api_alias": {"type": "string"}, + "timeout": _timeout_prop(90), + }, + ["project", "package", "repository", "arch"], + ), + "annotations": _read_only_annotations("OBS Build Log"), + } + + +def _spec_repo_status() -> dict[str, Any]: + return { + "name": "sailfish_repo_status", + "title": "Repo Status", + "description": "Run git status --short --branch under the configured git root.", + "inputSchema": _object_schema({"path": {"type": "string"}, "timeout": _timeout_prop(30)}), + "annotations": _read_only_annotations("Repo Status"), + } + + +def _spec_repo_find() -> dict[str, Any]: + return { + "name": "sailfish_repo_find", + "title": "Repo Find", + "description": "Search a repo or subtree under the configured git root.", + "inputSchema": _object_schema( + { + "path": {"type": "string"}, + "query": {"type": "string"}, + "fixed_strings": {"type": "boolean", "default": True}, + "max_count": {"type": "integer", "minimum": 1, "maximum": 1000, "default": 100}, + "timeout": _timeout_prop(30), + }, + ["query"], + ), + "annotations": _read_only_annotations("Repo Find"), + } + + +def _spec_spec_summary() -> dict[str, Any]: + return { + "name": "sailfish_spec_summary", + "title": "RPM Spec Summary", + "description": "Parse high-level metadata from a Sailfish RPM spec file.", + "inputSchema": _object_schema( + {"repo_path": {"type": "string"}, "spec_path": {"type": "string"}} + ), + "annotations": _read_only_annotations("RPM Spec Summary"), + } + + +def _spec_qml_find_translations() -> dict[str, Any]: + return { + "name": "sailfish_qml_find_translations", + "title": "Find QML Translations", + "description": "Find qsTrId and translator comments in QML files.", + "inputSchema": _object_schema({"path": {"type": "string"}, "timeout": _timeout_prop(30)}), + "annotations": _read_only_annotations("Find QML Translations"), + } + + +def _spec_qml_check_translator_ternaries() -> dict[str, Any]: + return { + "name": "sailfish_qml_check_translator_ternaries", + "title": "Check QML Ternary Translations", + "description": "Flag QML ternary qsTrId expressions that need branch-local comments.", + "inputSchema": _object_schema({"path": {"type": "string"}}), + "annotations": _read_only_annotations("Check QML Ternary Translations"), + } |