"""Validate a single transition and build the deployed Q-Reflector input.""" import json import math from pathlib import Path from PIL import Image, ImageOps ROOT = Path(__file__).resolve().parent IMAGE_MAX_SIDE = 1024 ACTION_TYPES = ["Tap", "LongPress", "Swipe", "InputText", "Open_App", "Back", "Home", "Wait"] FIELDS = ("precondition", "postcondition", "reason", "answer") GEOMETRY_INSTRUCTIONS = """The images may be resized. Current-action coordinates are pixels in the original BEFORE screenshot, with origin at top-left, x increasing rightward and y increasing downward. Normalized coordinates are (x / original_width, y / original_height): (0, 0) is top-left and (1, 1) is bottom-right. Use the normalized positions to locate action targets regardless of resizing. Do not interpret original-image pixel coordinates as resized-image pixels. This is geometric metadata, not evidence of success or the identity of a control. """ def short_text(value, name, limit, required=True): if not isinstance(value, str) or (required and not value.strip()): raise ValueError(f"Please enter {name}.") value = value.strip() if len(value) > limit: raise ValueError(f"{name.capitalize()} must be at most {limit:,} characters.") return value def screenshot(path, role): if not path: raise ValueError(f"Please upload the {role} screenshot.") path = Path(path) if not path.is_file() or path.stat().st_size > 10 * 1024 * 1024: raise ValueError(f"{role}: use an image smaller than 10 MB.") try: with Image.open(path) as im: if im.format not in {"PNG", "JPEG", "WEBP"} or getattr(im, "n_frames", 1) != 1: raise ValueError("Use a single PNG, JPEG, or WebP screenshot.") if min(im.size) < 128 or max(im.size) > 8192 or im.width * im.height > 20_000_000: raise ValueError("Screenshots must be at least 128 pixels per side and at most 20 megapixels / 8192 pixels per side.") return ImageOps.exif_transpose(im).convert("RGB") except (OSError, Image.DecompressionBombError) as exc: raise ValueError(f"{role}: this file could not be opened as a screenshot.") from exc def coordinate(value, name, limit): if isinstance(value, str): try: value = float(value.strip()) except ValueError: raise ValueError(f"Enter a numeric {name} coordinate in original BEFORE pixels.") from None if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value): raise ValueError(f"Enter a numeric {name} coordinate in original BEFORE pixels.") if value != int(value) or not 0 <= value < limit: raise ValueError(f"{name} must be an integer from 0 to {limit - 1} (original BEFORE pixels).") return int(value) def action_from_form(kind, description, x, y, x2, y2, value, width, height): if kind not in ACTION_TYPES: raise ValueError("Select a supported action type.") description = short_text(description, "the action description", 1200) args = None if kind in {"Tap", "LongPress", "Swipe"}: px, py = coordinate(x, "x", width), coordinate(y, "y", height) args = {"x": px, "y": py} if kind == "Swipe": args = {"x1": px, "y1": py, "x2": coordinate(x2, "end x", width), "y2": coordinate(y2, "end y", height)} elif kind == "InputText": args = {"text": short_text(value, "the exact typed text", 1500)} elif kind == "Open_App": args = {"app_name": short_text(value, "the app name", 160)} return {"type": kind, "arguments": args, "description": description} def history_from_json(text): text = short_text(text or "[]", "history", 6000) try: items = json.loads(text) except (ValueError, RecursionError) as exc: raise ValueError("History must be a JSON list. Leave it as [] when unavailable.") from exc if not isinstance(items, list) or len(items) > 3: raise ValueError("History must contain at most 3 previous actions, oldest first.") history = [] for i, item in enumerate(items): if not isinstance(item, dict) or set(item) - {"step", "action"}: raise ValueError("Each history item must contain action and an optional step number.") action = item.get("action") if not isinstance(action, dict) or set(action) - {"type", "arguments", "description"}: raise ValueError("A history action uses type, arguments, and description fields.") short_text(action.get("type"), "each history action type", 80) short_text(action.get("description"), "each history action description", 1200) if action.get("arguments") is not None and not isinstance(action["arguments"], dict): raise ValueError("History action arguments must be an object or null.") step = item.get("step", i + 1) if isinstance(step, bool) or not isinstance(step, int) or step < 0: raise ValueError("History step numbers must be nonnegative integers.") history.append({"step": step, "action": action}) # Reject JSON NaN/Infinity accepted by Python's decoder. try: json.dumps(history, allow_nan=False) except ValueError as exc: raise ValueError("History must not contain NaN or infinite numbers.") from exc return history def build_messages(before, after, context): """Same v9 prompt and coordinate metadata as the online protocol.""" images, metadata = [], [] for role, image in [("BEFORE", before), ("AFTER", after)]: w, h = image.size scale = min(1.0, IMAGE_MAX_SIDE / max(w, h)) size = (int(w * scale), int(h * scale)) images.append(image.resize(size, Image.Resampling.LANCZOS) if scale < 1 else image.copy()) metadata.append(dict(role=role, original_width=w, original_height=h, input_width=size[0], input_height=size[1])) args = context["action"].get("arguments") or {} points = [] for xk, yk in [("x", "y"), ("x1", "y1"), ("x2", "y2")]: if xk in args: nx, ny = args[xk] / before.width, args[yk] / before.height points.append(dict(keys=[xk, yk], original_pixels=[args[xk], args[yk]], normalized_xy=[round(nx, 6), round(ny, 6)], inside_original_image=0 <= nx < 1 and 0 <= ny < 1)) geo = json.dumps(dict(images=metadata, current_action_points=points), ensure_ascii=False, separators=(",", ":")) block = "=== Image Geometry ===\n" + GEOMETRY_INSTRUCTIONS + geo + "\n=== End Image Geometry ===\n\n" template = (ROOT / "prompt.txt").read_text() prompt = template.replace("{context}", json.dumps(context, ensure_ascii=False, indent=2)) prompt = prompt.replace("=== Context ===", block + "=== Context ===", 1) return [{"role": "user", "content": [ {"type": "text", "text": prompt}, {"type": "image", "image": images[0]}, {"type": "image", "image": images[1]}, ]}] def prepare(before_path, after_path, goal, kind, description, x, y, x2, y2, value, history): before, after = screenshot(before_path, "BEFORE"), screenshot(after_path, "AFTER") if before.size != after.size: raise ValueError("BEFORE and AFTER must have the same original dimensions. Upload full screenshots from the same device without cropping.") context = { "goal": short_text(goal, "the task goal", 2000), "history": history_from_json(history), "action": action_from_form(kind, description, x, y, x2, y2, value, *before.size), } return build_messages(before, after, context) def parse_response(text): text = text.strip() if text.startswith("```json") and text.endswith("```"): text = text[7:-3].strip() try: result = json.loads(text) except (ValueError, RecursionError) as exc: raise ValueError("Q-Reflector did not return a complete structured answer. Please try again.") from exc if (not isinstance(result, dict) or set(result) != set(FIELDS) or any(not isinstance(result[k], str) or not result[k].strip() for k in FIELDS) or result["answer"] not in {"Yes", "No"}): raise ValueError("Q-Reflector did not return a valid judgment. Please try again.") return {k: result[k] for k in FIELDS}