| """Native vocabulary, seven-row simultaneous GUI fill contract v1. |
| |
| Coordinates are three separately encoded digits. No tokenizer mutation, action |
| head, autoregressive decoding, or executable GUI action is provided here. |
| """ |
|
|
| import math |
| import string |
|
|
| from .gui_fill_table import KEYS as NAMED_KEYS |
| from .gui_fill_table import MASK, MODIFIERS, NA |
| from .visual_nanojev_data import validate_step |
|
|
| SCHEMA = ( |
| ("Click", ("coord", "coord", "button", "count")), |
| ("TypeText", ("yes",)), |
| ("PressKeys", ("key",) * 4), |
| ("Scroll", ("direction", "level")), |
| ("Wait", ("duration",)), |
| ("Drag", ("coord",) * 4), |
| ("Done", ("yes",)), |
| ) |
| WIDTHS = {"coord": 3, "key": 3} |
| |
| KEYS = NAMED_KEYS | frozenset(string.punctuation) |
| DOMAINS = { |
| "coord": tuple(f"{i:03d}" for i in range(1000)), |
| "key": tuple(sorted(KEYS)), |
| "button": ("LEFT", "RIGHT"), |
| "count": ("0", "1", "2"), |
| "yes": ("YES",), |
| "direction": ("UP", "DOWN"), |
| "level": ("1", "2", "3"), |
| "duration": ("1", "2", "4"), |
| } |
|
|
|
|
| def coordinate(value): |
| if type(value) not in (int, float) or not math.isfinite(value) or not 0 <= value <= 1: |
| raise ValueError("source_coordinate_out_of_range") |
| return f"{math.floor(value * 999 + 0.5):03d}" |
|
|
|
|
| def canonical_keys(keys): |
| aliases = { |
| "RETURN": "ENTER", |
| "ESCAPE": "ESC", |
| "CONTROL": "CTRL", |
| "COMMAND": "CMD", |
| " ": "SPACE", |
| } |
| keys = [aliases.get(k.upper(), k.upper()) for k in keys] |
| if not 1 <= len(keys) <= 4 or len(set(keys)) != len(keys) or any(k not in KEYS for k in keys): |
| raise ValueError("invalid_keys") |
| return [k for k in MODIFIERS if k in keys] + sorted(k for k in keys if k not in MODIFIERS) |
|
|
|
|
| def validate_action(operation, values): |
| kinds = dict(SCHEMA).get(operation) |
| if kinds is None or len(values) != len(kinds): |
| raise ValueError("invalid_operation_or_field_count") |
| for index, (kind, value) in enumerate(zip(kinds, values)): |
| optional = kind == "key" or (operation == "Click" and index == 2 and values[3] == "0") |
| if value not in DOMAINS[kind] and not (optional and value == NA): |
| raise ValueError("missing_or_invalid_parameter") |
| if operation == "Click": |
| if values[3] == "0" and values[2] != NA: |
| raise ValueError("move_button_must_be_NA") |
| if values[3] == "2" and values[2] != "LEFT": |
| raise ValueError("nonleft_double_click") |
| if operation == "PressKeys": |
| keys = [k for k in values if k != NA] |
| if values != canonical_keys(keys) + [NA] * (4 - len(keys)): |
| raise ValueError("noncanonical_key_order_or_padding") |
| if operation == "Drag" and values[:2] == values[2:]: |
| raise ValueError("zero_distance_drag") |
| return {"operation": operation, "values": list(values)} |
|
|
|
|
| def convert_target(target): |
| actions = validate_step(target)["actions"] |
| if len(actions) == 2 and [a.get("skill") for a in actions] == ["move_to", "drag_to"]: |
| start, end = [a["args"] for a in actions] |
| if end["button"] != "left": |
| raise ValueError("nonleft_drag") |
| return validate_action( |
| "Drag", [coordinate(v) for v in (start["x"], start["y"], end["x"], end["y"])] |
| ) |
| if len(actions) != 1: |
| raise ValueError("compound_without_intermediate_observation") |
| action = actions[0] |
| if action.get("done"): |
| raise ValueError("unverified_completion") |
| skill, args = action["skill"], action["args"] |
| if skill in {"click", "double_click", "move_to"}: |
| values = [coordinate(args[k]) for k in ("x", "y")] |
| values += [ |
| NA if skill == "move_to" else args["button"].upper(), |
| {"click": "1", "double_click": "2", "move_to": "0"}[skill], |
| ] |
| return validate_action("Click", values) |
| if skill == "type_text": |
| return validate_action("TypeText", ["YES"]) |
| if skill in {"press_key", "hotkey"}: |
| keys = canonical_keys([args["key"]] if skill == "press_key" else args["keys"]) |
| return validate_action("PressKeys", keys + [NA] * (4 - len(keys))) |
| raise ValueError(f"unsupported_or_unverified_source_skill:{skill}") |
|
|
|
|
| class NativeCodec: |
| """Finite legal encodings assembled in ID space, never retokenized.""" |
|
|
| anchor_positions = (7, 8, 9, 21, 23, 24, 36) |
|
|
| def __init__(self, tokenizer): |
| self.tokenizer = tokenizer |
| self.vocab_size = tokenizer.get_vocab_size() |
|
|
| def one(text): |
| ids = tokenizer.encode(text, add_special_tokens=False).ids |
| if len(ids) != 1: |
| raise ValueError(f"native single token required: {text}") |
| return ids[0] |
|
|
| self.na_id, self.mask_id = one(NA), one(MASK) |
| self.encodings, self.inverse = {}, {} |
| for kind, domain in DOMAINS.items(): |
| width = WIDTHS.get(kind, 1) |
| mapping = {NA: (self.na_id,) * width} |
| for value in domain: |
| ids = ( |
| [one(d) for d in value] |
| if kind == "coord" |
| else tokenizer.encode(value, add_special_tokens=False).ids |
| ) |
| if not 1 <= len(ids) <= width or self.na_id in ids or self.mask_id in ids: |
| raise ValueError(f"native encoding exceeds contract: {kind}/{value}") |
| mapping[value] = tuple(ids + [self.na_id] * (width - len(ids))) |
| inverse = {ids: value for value, ids in mapping.items()} |
| if len(inverse) != len(mapping): |
| raise ValueError("ambiguous_native_encoding") |
| self.encodings[kind], self.inverse[kind] = mapping, inverse |
| self.input_ids, self.slot_positions, self.field_slices = [], [], [] |
| self.row_slices = [] |
|
|
| def literal(text): |
| self.input_ids.extend(tokenizer.encode(text, add_special_tokens=False).ids) |
|
|
| for row_index, (name, kinds) in enumerate(SCHEMA): |
| if row_index: |
| literal("\n") |
| literal(name + "(") |
| row_start = len(self.slot_positions) |
| for field_index, kind in enumerate(kinds): |
| if field_index: |
| literal(",") |
| width = WIDTHS.get(kind, 1) |
| start = len(self.slot_positions) |
| for _ in range(width): |
| self.slot_positions.append(len(self.input_ids)) |
| self.input_ids.append(self.mask_id) |
| self.field_slices.append(slice(start, start + width)) |
| literal(")") |
| self.row_slices.append(slice(row_start, len(self.slot_positions))) |
| if len(self.slot_positions) != 37 or tokenizer.get_vocab_size() != self.vocab_size: |
| raise ValueError("native_contract_mutated") |
|
|
| def encode(self, operation, values): |
| validate_action(operation, values) |
| result = [] |
| for name, kinds in SCHEMA: |
| fields = values if name == operation else [NA] * len(kinds) |
| for kind, value in zip(kinds, fields): |
| result.extend(self.encodings[kind][value]) |
| return result |
|
|
| def decode(self, ids): |
| if len(ids) != 37 or any(type(i) is not int for i in ids): |
| raise ValueError("expected_37_integer_ids") |
| active, offset = [], 0 |
| for name, kinds in SCHEMA: |
| values = [] |
| for kind in kinds: |
| width = WIDTHS.get(kind, 1) |
| tokens = tuple(ids[offset : offset + width]) |
| if tokens not in self.inverse[kind]: |
| raise ValueError("invalid_field_token_sequence") |
| values.append(self.inverse[kind][tokens]) |
| offset += width |
| if any(v != NA for v in values): |
| active.append(validate_action(name, values)) |
| if len(active) != 1: |
| raise ValueError("expected_exactly_one_active_row") |
| return active[0] |
|
|
