Class DWPosePipeline

tools/preprocess/ltx2.3_preprocess_data.py:162–193

class DWPosePipeline:
    def __init__(self, det_checkpoint_path: str, pose2d_checkpoint_path: str, device: str = "cuda"):
        self.detector = DWposeONNX(
            det_onnx_path=det_checkpoint_path,
            pose_onnx_path=pose2d_checkpoint_path,
            device=device,
        )
        self.device = device
        logger.info(f"DWPose-ONNX initialized on {device}")

    def __call__(
        self,
        frames_rgb: List[np.ndarray],
        include_hands: bool = False,
        include_face: bool = False,
        bg_mode: str = "black",
    ) -> List[np.ndarray]:
        out = []
        for idx, frame in enumerate(frames_rgb):
            skeleton = self.detector(
                frame,
                include_hands=include_hands,
                include_face=include_face,
                bg_mode=bg_mode,
            )
            if skeleton.shape[:2] != frame.shape[:2]:
                skeleton = cv2.resize(skeleton, (frame.shape[1], frame.shape[0]), interpolation=cv2.INTER_NEAREST)
            out.append(skeleton)

            if (idx + 1) % 50 == 0:
                logger.info(f"DWPose: {idx + 1}/{len(frames_rgb)} frames")
        return out


def run_canny_edges(