| """Opt-in Unified133 v2 adapter around the unchanged, hash-pinned v1 codec. |
| |
| Direction reuses slot 6 (first desktop keys slot). Missing Android duration is |
| represented by five PAD labels, not a numeric demonstration. Original coordinate |
| targets, text limits, desktop behavior and input slot positions are reused. |
| """ |
|
|
| import hashlib |
| import importlib |
| import sys |
| from dataclasses import replace |
| from pathlib import Path |
|
|
| from trace2task.mobile133_protocol import DIRECTIONS, PROTOCOL, SCHEMA, direction_scroll_argv |
|
|
| BUNDLE = Path("runtime-bundle") |
| PINNED_SOURCES = { |
| "gui_owl_unified133": ( |
| "source/gui_owl_unified133.py", |
| "c71d188ad8dbb9ce6c2a44dbf71efa081e1081a943dbc1dc15ea77ee5a4a191f", |
| ), |
| "gui_owl_native134": ( |
| "source/gui_owl_native134.py", |
| "6b62d644d0f4e1b7332a282e921f374083fe3e0a954c45e390d1573051b64643", |
| ), |
| "native134_executor": ( |
| "source/native134_executor.py", |
| "df542f0cd364f6aa35348dc354c826cc7e68ecc47ffb6557eb0049b8ccfb52b4", |
| ), |
| "trace2task.gui_owl_parallel137_codec": ( |
| "native/src/trace2task/gui_owl_parallel137_codec.py", |
| "0c304c73489c48c190f3c2e579f9e6ac069864cd49048c06d9f5c83a870f9c00", |
| ), |
| "trace2task.gui_owl_short_parallel_input": ( |
| "native/src/trace2task/gui_owl_short_parallel_input.py", |
| "bac595f2b6bafc0e6d235672cf2254561d11e432090b22114a0aa27aa6c102b0", |
| ), |
| "slotread.numbers": ( |
| "upstream/slotread/numbers.py", |
| "0c51fa60b85558987bd4134214c23aaa55116a00e68d3a77e9a1b14cd7afbc6c", |
| ), |
| "slotread.labels": ( |
| "upstream/slotread/labels.py", |
| "e38a8125b7200e841b8224d89b1978d9aa4ac5505727c66878d04fc6238691a5", |
| ), |
| } |
|
|
|
|
| def load_legacy(bundle=BUNDLE): |
| import trace2task |
|
|
| bundle = Path(bundle).resolve(strict=True) |
| for module, (relative, expected) in PINNED_SOURCES.items(): |
| path = bundle / relative |
| if hashlib.sha256(path.read_bytes()).hexdigest() != expected: |
| raise ValueError("frozen_codec_source_changed:" + relative) |
| loaded = sys.modules.get(module) |
| if loaded and Path(loaded.__file__).resolve() != path: |
| raise ValueError("different_codec_already_loaded:" + module) |
| for path in (bundle / "source", bundle / "upstream"): |
| if str(path) not in sys.path: |
| sys.path.insert(0, str(path)) |
| native = str(bundle / "native/src/trace2task") |
| if native not in trace2task.__path__: |
| trace2task.__path__.append(native) |
| legacy = importlib.import_module("gui_owl_unified133") |
| for module, (relative, _) in PINNED_SOURCES.items(): |
| if Path(importlib.import_module(module).__file__).resolve() != bundle / relative: |
| raise ValueError("unexpected_codec_import:" + module) |
| return legacy |
|
|
|
|
| class Mobile133Codec: |
| variant = PROTOCOL |
| output_slots = 133 |
|
|
| def __init__(self, tokenizer, *, bundle=BUNDLE): |
| self.upstream = load_legacy(bundle) |
| self.legacy = self.upstream.Unified133Codec(tokenizer) |
| self.field_slices = dict(self.legacy.field_slices) |
| self.field_slices["direction"] = slice(6, 7) |
| self.direction_ids = {d: self.base_codec._single(d) for d in DIRECTIONS} |
|
|
| def __getattr__(self, name): |
| return getattr(self.legacy, name) |
|
|
| def contract(self): |
| return { |
| "protocol": PROTOCOL, |
| "output_slots": 133, |
| "coordinate_codebook_sha256": self.codebook_sha256, |
| "direction_slot": 6, |
| "direction_token_ids": self.direction_ids, |
| "missing_android_time": "five_supervised_PADs_not_numeric_duration", |
| "input_scaffold": "unchanged_v1_keys_slot_0_is_android_direction", |
| } |
|
|
| def validate_contract(self, contract): |
| if contract != self.contract(): |
| raise ValueError("mobile133_protocol_contract_mismatch") |
|
|
| def adapt_target(self, action, platform="desktop", *, coordinate_supervision=None): |
| if platform != "android": |
| return self.legacy.adapt_target( |
| action, platform, coordinate_supervision=coordinate_supervision |
| ) |
| name = action.get("action") |
| scroll = name == "scroll" |
| absent_time = name in ("wait", "long_press") and "time" not in action |
| if scroll: |
| if set(action) != {"action", "direction"} or action["direction"] not in DIRECTIONS: |
| raise ValueError("android_scroll_requires_direction_only") |
| surrogate = {"action": "wait", "time": 0} |
| elif absent_time: |
| surrogate = dict(action, time=0) |
| else: |
| return self.legacy.adapt_target( |
| action, platform, coordinate_supervision=coordinate_supervision |
| ) |
| target = self.legacy.adapt_target( |
| surrogate, platform, coordinate_supervision=coordinate_supervision |
| ) |
| ids, labels = list(target.target_ids), list(target.hard_labels) |
| fields = [f for f in target.fields if f[0] != "time"] |
| span = self.field_slices["time"] |
| ids[span] = [self.pad_id] * 5 |
| labels[span] = [-100 if scroll else self.pad_id] * 5 |
| if scroll: |
| ids[:2] = labels[:2] = self.action_encodings["scroll"] |
| ids[6] = labels[6] = self.direction_ids[action["direction"]] |
| fields.append(("direction", 6, 7)) |
| canonical = dict(action) |
| else: |
| fields.append(("time", span.start, span.stop)) |
| canonical = dict(target.canonical_action) |
| del canonical["time"] |
| return replace( |
| target, |
| canonical_action=canonical, |
| target_ids=tuple(ids), |
| hard_labels=tuple(labels), |
| fields=tuple(fields), |
| ) |
|
|
| def check_row(self, row): |
| if row.get("action_protocol") != PROTOCOL or row.get("schema") != SCHEMA: |
| raise ValueError("mobile133_row_protocol_mismatch") |
|
|
| def adapt_training_target(self, row): |
| self.check_row(row) |
| return self.adapt_target( |
| row["target"], |
| row["platform"], |
| coordinate_supervision=self.upstream.coordinate_metadata_from_row(row), |
| ) |
|
|
| def decode(self, logits, *, platform="desktop", key_sequences=()): |
| import torch |
|
|
| if platform != "android": |
| return self.legacy.decode(logits, platform=platform, key_sequences=key_sequences) |
| if logits.ndim != 2 or logits.shape[0] != 133 or logits.shape[1] < self.vocab_size: |
| raise ValueError("expected_133_by_vocab_logits") |
| if not torch.isfinite(logits).all(): |
| raise ValueError("nonfinite_logits") |
| allowed = self.upstream.allowed_actions("android") + ("scroll",) |
| scores = torch.stack( |
| [ |
| sum(logits[p, t].float() for p, t in enumerate(self.action_encodings[a])) |
| for a in allowed |
| ] |
| ) |
| name = allowed[int(scores.argmax())] |
| if name == "scroll": |
| direction = DIRECTIONS[int(logits[6, list(self.direction_ids.values())].argmax())] |
| return {"action": "scroll", "direction": direction} |
| absent = False |
| if name in ("wait", "long_press"): |
| span = self.field_slices["time"] |
| v = logits[span].float() |
| absent = bool( |
| v[:, self.pad_id].sum() > v[:, list(self.base_codec.digit_ids)].max(-1).values.sum() |
| ) |
| if absent: |
| logits = logits.clone() |
| logits[span] = -60 |
| logits[span, self.base_codec.digit_ids[0]] = 60 |
| action = self.legacy.decode(logits, platform=platform, key_sequences=key_sequences) |
| if absent: |
| del action["time"] |
| return self.adapt_target(action, platform).canonical_action |
|
|
| def _coordinate_view(self, row): |
| |
| |
| self.check_row(row) |
| history = [] |
| for step in row["history"]: |
| actions = [] |
| for action in step["actions"]: |
| if row["platform"] == "android" and action.get("action") == "scroll": |
| self.adapt_target(action, "android") |
| else: |
| actions.append(action) |
| history.append(dict(step, actions=actions)) |
| return dict(row, history=history) |
|
|
| def messages(self, row, *, bound=False): |
| projection = self._coordinate_view(row) |
| labels = self.upstream.history_coordinate_tokens(projection, self.legacy) |
| result = self.upstream.messages(row) |
| result[0]["content"] = ( |
| result[0]["content"] |
| .replace("long_press(coordinate,time)", "long_press(coordinate,optional time)") |
| .replace("Android actions: key(text)", "Android actions: scroll(direction), key(text)") |
| ) |
| result[0]["content"] += ( |
| " Android wait may omit time. Omitted Android wait/long_press time means executor default; " |
| "predict PAD in all five time slots. Explicit time uses the original five digits. " |
| "Android scroll uses direction up/down/left/right (content movement), not swipe endpoints. " |
| "Its direction token occupies the first keys slot; other keys and pixels slots are inactive. " |
| "Desktop actions retain their original meanings. Protocol: " + PROTOCOL + "." |
| ) |
| if bound: |
| values = iter(labels) |
| for item in result[1]["content"]: |
| if item["type"] == "text": |
| parts = item["text"].split("<|fim_pad|>") |
| item["text"] = parts[0] + "".join( |
| self.tokenizer.decode([next(values)], skip_special_tokens=False) + part |
| for part in parts[1:] |
| ) |
| return result |
|
|
| def bind_history_tokens(self, prefix_ids, row): |
| return self.upstream.bind_history_tokens( |
| prefix_ids, self._coordinate_view(row), self.legacy |
| ) |
|
|
| def prepare_example(self, row, processor, image_loader): |
| import torch |
|
|
| prompt = processor.apply_chat_template( |
| self.messages(row), tokenize=False, add_generation_prompt=True |
| ) |
| visual = processor( |
| text=[prompt], |
| images=[image_loader(im["path"]) for im in row["images"]], |
| return_tensors="pt", |
| ) |
| prefix, _ = self.bind_history_tokens(visual["input_ids"][0].tolist(), row) |
| built = self.upstream.build_parallel_input(self.tokenizer, prefix) |
| ids = torch.tensor([built.input_ids], dtype=torch.long) |
| return { |
| "input_ids": ids, |
| "attention_mask": torch.ones_like(ids), |
| "pixel_values": visual["pixel_values"], |
| "image_grid_thw": visual["image_grid_thw"], |
| "slot_positions": torch.tensor([built.slot_positions]), |
| "prefix_lengths": torch.tensor([built.prefix_length]), |
| "lengths": torch.tensor([ids.shape[1]]), |
| }, self.adapt_training_target(row) |
|
|
| def execution_plan(self, action, size, platform="android", *, config, installed_apps=None): |
| if config.protocol != PROTOCOL: |
| raise ValueError("execution_protocol_mismatch") |
| action = self.adapt_target(action, platform).canonical_action |
| if platform == "android": |
| if action["action"] == "scroll": |
| return [("adb_shell", direction_scroll_argv(action["direction"], size, config))] |
| if action["action"] in ("wait", "long_press") and "time" not in action: |
| seconds = ( |
| config.wait_seconds if action["action"] == "wait" else config.long_press_seconds |
| ) |
| action = dict(action, time=seconds) |
| return self.upstream.execution_plan( |
| self.legacy, action, size, platform, installed_apps=installed_apps |
| ) |
|
|
