| """Policies of the hidden-state arms (EXPERIMENT_LOG.md 2026-10-19, tasks 3 and 4a); eval venv (numpy, requests). |
| |
| HidPolicy (MC_POLICY / rollout kind "hid"): the controller's input is [feature (4096), camera history (40)] -> MLP |
| (train/hid_bc.py Net: 256 -> 256 -> A's three heads, exported as ctrl_policy's npz layout, read by ctrl_policy.logits). |
| - feature: train/hid_server.py's Qwen3.5-9B hidden state for (the current frame, the current target name), asked on |
| A's perception schedule (the episode's first step, when the target name changes, PERCEIVE_EVERY steps after the |
| last call) and held unchanged in between; |
| - camera history: A's window camera part, the previous camera action [yaw/64, pitch/16] of the last 20 steps oldest |
| first (zeros before the episode start; not reset on a target change), rounded to 4 decimals as A's x. |
| Sampling as ctrl_policy.Policy (three heads, numpy Generator seeded by seed). url: the feature server (a list: one |
| picked per process). Options: log_feat logs the feature (base64 bf16 bits) on asking steps; frames marks asking |
| steps with "frame" (train/grpo_rollout.py then saves the PNG); ref (an npz) asks the server for the base feature too |
| and logs ref, the 45 log-probs of that controller on [base feature, camera history] (the 4a KL reference). |
| Per-step log: target, bins, logp, cam (the 40 history values), held (0 on asking steps), feat_s on asking steps |
| (seconds of the call), truth as Policy; plus feat / frame / ref as above. |
| |
| Teacher (kind "hid_teacher"): A's controller (ctrl_policy.Policy, gt perception) whose log marks with "frame" the |
| steps where A's vlm schedule would ask (the BC demonstrations, train/hid_bc.py collect). |
| """ |
| import base64 |
| from collections import deque |
| import os |
| import time |
|
|
| import numpy as np |
| import requests |
|
|
| from ctrl_policy import (CAM_SCALE, PERCEIVE_EVERY, WINDOW, Policy, action, current_target, gt_features, log_softmax, |
| logits, view) |
|
|
|
|
| def decode(s): |
| """base64 bf16 bits -> float32 vector (exact).""" |
| return (np.frombuffer(base64.b64decode(s), np.uint16).astype(np.uint32) << 16).view(np.float32) |
|
|
|
|
| class HidPolicy: |
| def __init__(self, npz, seed, url, log_feat=False, frames=False, ref=None): |
| self.w = dict(np.load(npz)) |
| self.ref = dict(np.load(ref)) if ref else None |
| self.rng = np.random.default_rng(seed) |
| urls = url if isinstance(url, list) else [url] |
| self.url = urls[os.getpid() % len(urls)] |
| self.log_feat, self.frames = log_feat, frames |
| self.cams = deque((np.zeros(2) for _ in range(WINDOW)), maxlen=WINDOW) |
| self.cam = [0.0, 0.0] |
| self.asked, self.held, self.feat, self.base = None, 0, None, None |
|
|
| def act(self, frame, info, checker, presatisfied): |
| index, target, name = current_target(checker, presatisfied) |
| pose = info["player_pos"] |
| dist, h, v = view(pose, target) |
| log = {} |
| if self.asked == name and self.held < PERCEIVE_EVERY - 1: |
| self.held += 1 |
| else: |
| t = time.monotonic() |
| r = requests.post(self.url + "/feat", json={"frame": frame, "name": name}, timeout=600) |
| r.raise_for_status() |
| r = r.json() |
| self.feat = decode(r["feat"]) |
| self.base = decode(r["base"]) if self.ref else None |
| self.asked, self.held = name, 0 |
| log["feat_s"] = round(time.monotonic() - t, 3) |
| if self.log_feat: |
| log["feat"] = r["feat"] |
| if self.frames: |
| log["frame"] = True |
| self.cams.append(np.array(self.cam)[::-1] / CAM_SCALE[::-1]) |
| cam = np.round(np.concatenate(self.cams), 4) |
| heads = [log_softmax(z) for z in logits(self.w, np.concatenate([self.feat, cam]))] |
| bins = [int(self.rng.choice(len(lp), p=np.exp(lp))) for lp in heads] |
| a = action(*bins) |
| self.cam = a["camera"] |
| if self.ref: |
| log["ref"] = np.round(np.concatenate([log_softmax(z) for z in logits(self.ref, np.concatenate( |
| [self.base, cam]))]), 4).tolist() |
| log.update(target=index, bins=bins, logp=float(sum(lp[b] for lp, b in zip(heads, bins))), cam=cam.tolist(), |
| held=self.held, truth={"seen": bool(gt_features(h, v)[0]), "dist": round(dist, 2), |
| "h": round(h, 2), "v": round(v, 2)}) |
| return a, log |
|
|
|
|
| class Teacher(Policy): |
| def __init__(self, npz, seed): |
| super().__init__(npz, seed, "gt") |
| self.asked_name, self.since = None, 0 |
|
|
| def act(self, frame, info, checker, presatisfied): |
| name = current_target(checker, presatisfied)[2] |
| a, log = super().act(frame, info, checker, presatisfied) |
| if self.asked_name == name and self.since < PERCEIVE_EVERY - 1: |
| self.since += 1 |
| else: |
| self.asked_name, self.since = name, 0 |
| log["frame"] = True |
| return a, log |
|
|
|
|
| def make(kind, **kwargs): |
| return Teacher(**kwargs) if kind == "hid_teacher" else HidPolicy(**kwargs) |
|
|
