"""SATLINK: the satellite <-> ground contract. Both ends import this module and share NOTHING else. It has no torch/PIL dependency, so the satellite payload never drags in the 338M-parameter decoder or the VLM -- only the 5.9M encoder + VQ ships to orbit. Everything fixed at deployment time is a CONFIGURED CONSTANT, not a transmitted field. A satellite flies with a known frame size and a known tokenizer, so image size, tile size, tokens per crop and codebook width are agreed once (a LinkProfile) and never appear on the downlink. Tile count per frame is then a constant too, which means the tile index and the frame boundary are both implied by position in the stream -- so there is no header at all. The downlink is a bare sequence of fixed-size crop payloads in raster order. 48 tokens x log2(4096) bits = 576 bits = 72 bytes per 256x256 crop That is 910x smaller than the 65536 raw grayscale bytes the crop stands for, with 0% protocol overhead. Note the payload is codebook INDICES, not codebook vectors: the vectors would be 48 x 12 floats = 2304 bytes and carry no more information. This assumes the layer below delivers crops in order and without gaps. That is the transport's job -- a satellite downlink already runs framing, FEC and ARQ underneath -- and duplicating it with our own sequence numbers would cost bandwidth to re-solve a solved problem. unpack_frame() therefore validates the byte count exactly, so a truncated or padded stream is a loud error rather than a silent tile misalignment. n_tok is the rate knob. One-D-Piece is trained with tail-token drop, so the tokens are ordered by importance -- the early ones carry coarse structure (measured entropy 10.2-10.9 bits at positions 0-7) and the late ones fine detail (11.1-11.7 bits at 40-47). Cutting n_tok trades detail for bandwidth linearly, at inference time, with no retraining. """ TILE = 256 CODEBOOK_BITS = 12 # 4096-entry One-D-Piece codebook -> log2(4096) = 12 bits per index N_LATENT = 48 # tokens per crop; the rate knob class LinkProfile: """The agreed configuration. Built identically on both ends -- from the deployment config in flight, or from a --width/--height argument when benchmarking on the ground.""" __slots__ = ("width", "height", "tile", "n_tok", "bits") def __init__(self, width, height, tile=TILE, n_tok=N_LATENT, bits=CODEBOOK_BITS): self.width, self.height, self.tile = int(width), int(height), int(tile) self.n_tok, self.bits = int(n_tok), int(bits) if self.width <= 0 or self.height <= 0: raise ValueError("frame size must be positive") @property def grid(self): """-> (tx, ty). The frame is cut at NATIVE resolution, never rescaled: 512x512 is 2x2, 600x600 is 3x3.""" t = self.tile return max(1, -(-self.width // t)), max(1, -(-self.height // t)) @property def n_tiles(self): tx, ty = self.grid return tx * ty def starts(self, axis): """Top-left offsets along axis 0=x / 1=y: evenly spread, the last tile flush to the edge. Tiles OVERLAP rather than the last one being edge-padded -- on a 600px side the padded variant leaves the final tile 12% real content, and the tokenizer spends its 48 tokens on a replicated stripe whose reconstruction bleeds back into the real pixels.""" n = self.width if axis == 0 else self.height k = self.grid[axis] t = self.tile if k <= 1 or n <= t: return [0] step = (n - t) / (k - 1) return [int(round(i * step)) for i in range(k)] def tile_origin(self, tile_idx): """-> (x, y) pixel offset of the raster tile index.""" tx, _ = self.grid return self.starts(0)[tile_idx % tx], self.starts(1)[tile_idx // tx] @property def crop_bytes(self): return (self.n_tok * self.bits + 7) // 8 @property def frame_bytes(self): return self.n_tiles * self.crop_bytes def __repr__(self): tx, ty = self.grid return (f"LinkProfile({self.width}x{self.height}, {tx}x{ty} tiles, " f"{self.n_tok}x{self.bits}b -> {self.crop_bytes} B/crop, " f"{self.frame_bytes} B/frame)") def pack_indices(idx, bits=CODEBOOK_BITS): """Pack ints into a dense bitstream, MSB-first. 12-bit codes in 16-bit slots would waste a quarter of the downlink; here two codes share three bytes.""" out = bytearray((len(idx) * bits + 7) // 8) pos = 0 for v in idx: if not 0 <= v < (1 << bits): raise ValueError(f"index {v} does not fit in {bits} bits") for k in range(bits - 1, -1, -1): if (v >> k) & 1: out[pos >> 3] |= 0x80 >> (pos & 7) pos += 1 return bytes(out) def unpack_indices(buf, n, bits=CODEBOOK_BITS, off=0): end = off + (n * bits + 7) // 8 if len(buf) < end: raise ValueError(f"payload short: {len(buf)} < {end}") out, pos = [], off * 8 for _ in range(n): v = 0 for _ in range(bits): v = (v << 1) | ((buf[pos >> 3] >> (7 - (pos & 7))) & 1) pos += 1 out.append(v) return out def pack_crop(profile, indices): if len(indices) != profile.n_tok: raise ValueError(f"expected {profile.n_tok} indices, got {len(indices)}") return pack_indices(indices, profile.bits) def unpack_crop(profile, buf, off=0): return unpack_indices(buf, profile.n_tok, profile.bits, off) def pack_frame(profile, tiles): """tiles: n_tiles index lists in RASTER order (row-major, matching tile_origin).""" if len(tiles) != profile.n_tiles: raise ValueError(f"expected {profile.n_tiles} tiles, got {len(tiles)}") return b"".join(pack_crop(profile, t) for t in tiles) def unpack_frame(profile, buf): """-> n_tiles index lists in raster order. The exact-length check is the only integrity guarantee this layer offers, and it is deliberate: a short stream must fail loudly, because the alternative is decoding shifted tiles into a plausible-looking but wrong image.""" if len(buf) != profile.frame_bytes: raise ValueError(f"frame is {len(buf)} B, profile says {profile.frame_bytes} B " f"({profile.n_tiles} tiles x {profile.crop_bytes} B)") cb = profile.crop_bytes return [unpack_crop(profile, buf, i * cb) for i in range(profile.n_tiles)]