Spaces:
Running on Zero
Running on Zero
| """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} | |