(self, config)
| 192 | return MemoryHandle(buffers=buffers) |
| 193 | |
| 194 | def process(self, config): |
| 195 | self.logger.info("Starting processing in DecoderService...") |
| 196 | room = config.get("data_bootstrap_room", 0) |
| 197 | decoder_metrics = config.setdefault("request_metrics", {}).setdefault("stages", {}).setdefault("decoder", {}) |
| 198 | decoder_metrics["compute_start_ts"] = time.time() |
| 199 | strict_meta_hash_check = str(os.getenv("LIGHTX2V_STRICT_META_HASH", "0")).strip().lower() in {"1", "true", "yes", "on"} |
| 200 | room_buffers = self._rdma_buffers.get(room) |
| 201 | receiver = self.data_receiver.get(room) |
| 202 | |
| 203 | if receiver is None: |
| 204 | raise RuntimeError(f"DataReceiver is not initialized in DecoderService for room={room}.") |
| 205 | if room_buffers is None: |
| 206 | raise RuntimeError(f"No RDMA buffer available in DecoderService for room={room}.") |
| 207 | |
| 208 | def _buffer_view(buf: torch.Tensor, dtype: torch.dtype, shape: tuple[int, ...]) -> torch.Tensor: |
| 209 | view = torch.empty(0, dtype=dtype, device=buf.device) |
| 210 | view.set_(buf.untyped_storage(), 0, shape) |
| 211 | return view |
| 212 | |
| 213 | def _sha256_tensor(tensor: Optional[torch.Tensor]) -> Optional[str]: |
| 214 | if tensor is None: |
| 215 | return None |
| 216 | data_tensor = tensor.detach() |
| 217 | if data_tensor.dtype == torch.bfloat16: |
| 218 | data_tensor = data_tensor.to(torch.float32) |
| 219 | data = data_tensor.contiguous().cpu().numpy().tobytes() |
| 220 | return hashlib.sha256(data).hexdigest() |
| 221 | |
| 222 | if len(room_buffers) < 2: |
| 223 | raise RuntimeError("Phase2 RDMA buffers require [latents, meta] entries.") |
| 224 | |
| 225 | meta_buf = room_buffers[1] |
| 226 | |
| 227 | def _read_phase2_meta() -> tuple[dict, str]: |
| 228 | meta_bytes = _buffer_view(meta_buf, torch.uint8, (meta_buf.numel(),)).detach().contiguous().cpu().numpy().tobytes() |
| 229 | meta_str = meta_bytes.split(b"\x00", 1)[0].decode("utf-8", errors="ignore") if meta_bytes else "" |
| 230 | if not meta_str: |
| 231 | raise ValueError("missing latents metadata from transformer") |
| 232 | parsed = json.loads(meta_str) |
| 233 | if not isinstance(parsed, dict): |
| 234 | raise ValueError(f"phase2 metadata type mismatch: {type(parsed)}") |
| 235 | return parsed, meta_str |
| 236 | |
| 237 | def _infer_latents_shape_from_config() -> tuple[int, int, int, int]: |
| 238 | z_dim = int(config.get("vae_z_dim", 16)) |
| 239 | vae_stride = config.get("vae_stride", (4, 8, 8)) |
| 240 | stride_t = int(vae_stride[0]) |
| 241 | stride_h = int(vae_stride[1]) |
| 242 | stride_w = int(vae_stride[2]) |
| 243 | target_video_length = int(config.get("target_video_length", 81)) |
| 244 | target_height = int(config.get("target_height", 480)) |
| 245 | target_width = int(config.get("target_width", 832)) |
| 246 | |
| 247 | t_prime = 1 + (target_video_length - 1) // stride_t |
| 248 | h_prime = int(math.ceil(target_height / stride_h)) |
| 249 | w_prime = int(math.ceil(target_width / stride_w)) |
| 250 | return (z_dim, t_prime, h_prime, w_prime) |
| 251 |
no test coverage detected