| """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 |
| N_LATENT = 48 |
|
|
|
|
| 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)] |
|
|
