PoseEstimator

Pose estimator interface and backends for MGT-python.

This module provides:

• :class:PoseEstimator – an abstract base class (ABC) defining the common interface that all pose backends must implement.
• :class:MediaPipePoseEstimator – a concrete backend powered by Google MediaPipe Pose (33 landmarks, CPU-friendly, zero model download).
• :class:OpenPosePoseEstimator – a thin wrapper around the legacy OpenPose / Caffe-model implementation already present in :mod:musicalgestures._pose.

The shared interface means that backends are interchangeable::

from musicalgestures._pose_estimator import MediaPipePoseEstimator
est = MediaPipePoseEstimator()
keypoints = est.predict_frame(frame)   # → np.ndarray shape (33, 3)

Examples

import numpy as np frame = np.zeros((480, 640, 3), dtype=np.uint8)

Without mediapipe installed this raises MgDependencyError gracefully.

PoseEstimatorResult

PoseEstimatorResult(keypoints, landmark_names, frame_index=0, timestamp=0.0)

Container for the output of a single-frame pose estimation.

Parameters:

Name	Type	Description	Default
keypoints	ndarray	2-D array of shape (n_keypoints, 3) where columns are (x, y, confidence). Coordinates are normalised to [0, 1].	required
landmark_names	list[str]	List of keypoint names corresponding to each row.	required
frame_index	int	Frame index this result belongs to.	0
timestamp	float	Timestamp in seconds.	0.0

Source code in musicalgestures/_pose_estimator.py

def __init__(
    self,
    keypoints: np.ndarray,
    landmark_names: list[str],
    frame_index: int = 0,
    timestamp: float = 0.0,
) -> None:
    self.keypoints = np.asarray(keypoints, dtype=float)
    self.landmark_names = landmark_names
    self.frame_index = int(frame_index)
    self.timestamp = float(timestamp)

to_dict

to_dict()

Return a plain dict representation.

Source code in musicalgestures/_pose_estimator.py

def to_dict(self) -> dict[str, Any]:
    """Return a plain dict representation."""
    return {
        "frame_index": self.frame_index,
        "timestamp": self.timestamp,
        "keypoints": {
            name: {
                "x": float(self.keypoints[i, 0]),
                "y": float(self.keypoints[i, 1]),
                "confidence": float(self.keypoints[i, 2]),
            }
            for i, name in enumerate(self.landmark_names)
        },
    }

PoseEstimator

PoseEstimator(model=PoseModel.MEDIAPIPE, device=PoseDevice.CPU)

Bases: ABC

Abstract base class for pose estimation backends.

All concrete subclasses must implement :meth:predict_frame and :meth:landmark_names.

Parameters:

Name	Type	Description	Default
model	PoseModel | str	Skeleton model variant.	MEDIAPIPE
device	PoseDevice | str	Compute backend ('cpu' or 'gpu').	CPU

Source code in musicalgestures/_pose_estimator.py

def __init__(
    self,
    model: PoseModel | str = PoseModel.MEDIAPIPE,
    device: PoseDevice | str = PoseDevice.CPU,
) -> None:
    self.model = PoseModel(model)
    self.device = PoseDevice(device)

landmark_names abstractmethod property

landmark_names

Ordered list of keypoint names.

predict_frame abstractmethod

predict_frame(frame)

Run pose estimation on a single BGR frame.

Parameters:

Name	Type	Description	Default
frame	ndarray	Input frame as a NumPy array of shape (H, W, 3) in BGR order.	required

Returns:

Type	Description
PoseEstimatorResult

Source code in musicalgestures/_pose_estimator.py

@abc.abstractmethod
def predict_frame(self, frame: np.ndarray) -> PoseEstimatorResult:
    """Run pose estimation on a single BGR frame.

    Parameters
    ----------
    frame:
        Input frame as a NumPy array of shape ``(H, W, 3)`` in BGR order.

    Returns
    -------
    PoseEstimatorResult
    """

predict_video

predict_video(filename, start=0.0, end=None, skip=0)

Run pose estimation on every frame of a video file.

Parameters:

Name	Type	Description	Default
filename	str | Path	Path to the video file.	required
start	float	Start time in seconds.	0.0
end	float | None	End time in seconds (None = full video).	None
skip	int	Process every (1 + skip)-th frame.	0

Returns:

Type	Description
list[PoseEstimatorResult]

Source code in musicalgestures/_pose_estimator.py

def predict_video(
    self,
    filename: str | Path,
    start: float = 0.0,
    end: float | None = None,
    skip: int = 0,
) -> list[PoseEstimatorResult]:
    """Run pose estimation on every frame of a video file.

    Parameters
    ----------
    filename:
        Path to the video file.
    start:
        Start time in seconds.
    end:
        End time in seconds (None = full video).
    skip:
        Process every (1 + skip)-th frame.

    Returns
    -------
    list[PoseEstimatorResult]
    """
    from musicalgestures._stream import MgVideoReader

    results: list[PoseEstimatorResult] = []
    with MgVideoReader(filename, start=start, end=end) as reader:
        for i, (frame, ts) in enumerate(reader):
            if skip > 0 and i % (skip + 1) != 0:
                continue
            result = self.predict_frame(frame)
            result.frame_index = i
            result.timestamp = ts
            results.append(result)
    return results

MediaPipePoseEstimator

MediaPipePoseEstimator(model_complexity=1, min_detection_confidence=0.5, min_tracking_confidence=0.5, device=PoseDevice.CPU)

Bases: PoseEstimator

Pose estimator backed by Google MediaPipe Pose (Tasks API).

Requires the optional mediapipe>=0.10 package::

pip install musicalgestures[pose]

The first time you use a given complexity level the corresponding .task model file (~8–28 MB) is downloaded from Google's model storage and cached in musicalgestures/models/.

Parameters:

Name	Type	Description	Default
model_complexity	int	MediaPipe model complexity (0 = lite, 1 = full, 2 = heavy). Higher values are more accurate but slower. Default: 1.	1
min_detection_confidence	float	Minimum confidence for initial body detection. Default: 0.5.	0.5
min_tracking_confidence	float	Minimum confidence for landmark tracking. Default: 0.5.	0.5

Examples:

>>> import numpy as np
>>> est = MediaPipePoseEstimator()
>>> frame = np.zeros((480, 640, 3), dtype=np.uint8)
>>> result = est.predict_frame(frame)
>>> result.keypoints.shape  # (33, 3)

Source code in musicalgestures/_pose_estimator.py

def __init__(
    self,
    model_complexity: int = 1,
    min_detection_confidence: float = 0.5,
    min_tracking_confidence: float = 0.5,
    device: PoseDevice | str = PoseDevice.CPU,
) -> None:
    super().__init__(model=PoseModel.MEDIAPIPE, device=device)
    self.model_complexity = model_complexity
    self.min_detection_confidence = min_detection_confidence
    self.min_tracking_confidence = min_tracking_confidence
    self._landmarker = None  # lazy init

predict_frame

predict_frame(frame)

Run MediaPipe Pose on a single BGR frame.

Parameters:

Name	Type	Description	Default
frame	ndarray	BGR frame, shape (H, W, 3).	required

Returns:

Type	Description
PoseEstimatorResult	33 landmarks; confidence is the visibility score.

Source code in musicalgestures/_pose_estimator.py

def predict_frame(self, frame: np.ndarray) -> PoseEstimatorResult:
    """Run MediaPipe Pose on a single BGR frame.

    Parameters
    ----------
    frame:
        BGR frame, shape ``(H, W, 3)``.

    Returns
    -------
    PoseEstimatorResult
        33 landmarks; ``confidence`` is the visibility score.
    """
    self._ensure_initialized()
    import cv2
    import mediapipe as mp

    rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
    mp_image = mp.Image(image_format=mp.ImageFormat.SRGB, data=rgb)
    detection_result = self._landmarker.detect(mp_image)

    n = len(MEDIAPIPE_LANDMARK_NAMES)
    keypoints = np.zeros((n, 3), dtype=float)

    if detection_result.pose_landmarks:
        for i, lm in enumerate(detection_result.pose_landmarks[0]):
            keypoints[i] = [lm.x, lm.y, lm.visibility]

    return PoseEstimatorResult(
        keypoints=keypoints,
        landmark_names=MEDIAPIPE_LANDMARK_NAMES,
    )

close

close()

Release MediaPipe resources.

Source code in musicalgestures/_pose_estimator.py

def close(self) -> None:
    """Release MediaPipe resources."""
    if self._landmarker is not None:
        self._landmarker.close()
        self._landmarker = None

OpenPosePoseEstimator

OpenPosePoseEstimator(model=PoseModel.BODY_25, device=PoseDevice.GPU, threshold=0.1)

Bases: PoseEstimator

Thin wrapper around the legacy OpenPose / Caffe-model backend.

This class delegates to :func:musicalgestures._pose.pose and is provided so that the old OpenPose workflow can be used through the same :class:PoseEstimator interface.

Parameters:

Name	Type	Description	Default
model	PoseModel | str	One of 'body_25', 'coco', or 'mpi'.	BODY_25
device	PoseDevice | str	'cpu' or 'gpu'.	GPU
threshold	float	Minimum confidence threshold. Default: 0.1.	0.1

Source code in musicalgestures/_pose_estimator.py

def __init__(
    self,
    model: PoseModel | str = PoseModel.BODY_25,
    device: PoseDevice | str = PoseDevice.GPU,
    threshold: float = 0.1,
) -> None:
    super().__init__(model=model, device=device)
    self.threshold = threshold
    self._landmark_names: list[str] = []

predict_frame

predict_frame(frame)

Run OpenPose inference on a single BGR frame.

.. note:: Full video-level processing is better handled by calling :meth:MgVideo.pose directly.

Source code in musicalgestures/_pose_estimator.py

def predict_frame(self, frame: np.ndarray) -> PoseEstimatorResult:
    """Run OpenPose inference on a single BGR frame.

    .. note::
        Full video-level processing is better handled by calling
        :meth:`MgVideo.pose` directly.
    """
    raise NotImplementedError(
        "OpenPosePoseEstimator.predict_frame() is not implemented for "
        "single frames.  Use MgVideo.pose() for full-video inference."
    )

get_pose_model_path

get_pose_model_path(model_complexity=1, models_dir=None)

Return the local path of the MediaPipe pose .task model file, downloading and caching it on first use.

This is the shared model download/cache used by both :class:MediaPipePoseEstimator (per-frame estimation, MgVideo.pose()) and :func:musicalgestures.extract_pose_landmarks (whole-video landmark trajectories), so a given model file is only downloaded once.

Parameters:

Name	Type	Description	Default
model_complexity	int	MediaPipe model variant: 0 (lite), 1 (full) or 2 (heavy). Invalid values fall back to 1 with a warning. Defaults to 1.	1
models_dir	Path or str	Directory to cache the model file in. Defaults to None (the musicalgestures/models/ directory inside the installed package).	None

Returns:

Name	Type	Description
Path	Path	Path to the cached model file (~8-28 MB, downloaded from
	Path	Google's model storage if not already present).

Raises:

Type	Description
MgDependencyError	If the model file is missing and the download fails.

Source code in musicalgestures/_pose_estimator.py

def get_pose_model_path(model_complexity: int = 1, models_dir: Path | str | None = None) -> Path:
    """Return the local path of the MediaPipe pose ``.task`` model file,
    downloading and caching it on first use.

    This is the shared model download/cache used by both
    :class:`MediaPipePoseEstimator` (per-frame estimation, ``MgVideo.pose()``)
    and :func:`musicalgestures.extract_pose_landmarks` (whole-video landmark
    trajectories), so a given model file is only downloaded once.

    Args:
        model_complexity (int, optional): MediaPipe model variant: 0 (lite),
            1 (full) or 2 (heavy). Invalid values fall back to 1 with a
            warning. Defaults to 1.
        models_dir (Path or str, optional): Directory to cache the model file
            in. Defaults to None (the ``musicalgestures/models/`` directory
            inside the installed package).

    Returns:
        Path: Path to the cached model file (~8-28 MB, downloaded from
        Google's model storage if not already present).

    Raises:
        MgDependencyError: If the model file is missing and the download fails.
    """
    if models_dir is None:
        import musicalgestures as mg

        models_dir = Path(mg.__file__).parent / "models"
    else:
        models_dir = Path(models_dir)
    models_dir.mkdir(exist_ok=True)

    if model_complexity not in _POSE_MODEL_NAMES:
        logger.warning(
            "model_complexity %d is not valid (0-2); defaulting to 1.",
            model_complexity,
        )
        model_complexity = 1

    model_path = models_dir / _POSE_MODEL_NAMES[model_complexity]
    if model_path.exists():
        return model_path

    url = _POSE_MODEL_URLS[model_complexity]
    logger.info("Downloading MediaPipe model from %s …", url)
    print(f"Downloading MediaPipe pose model ({_POSE_MODEL_NAMES[model_complexity]}) …")
    try:
        import urllib.request

        urllib.request.urlretrieve(url, model_path)
        logger.info("Model saved to %s", model_path)
    except Exception as exc:
        raise MgDependencyError(
            f"Failed to download MediaPipe pose model from {url}. "
            "Please download it manually and place it at: "
            f"{model_path}"
        ) from exc
    return model_path

get_pose_estimator

get_pose_estimator(backend='mediapipe', **kwargs)

Factory function: return a :class:PoseEstimator for the requested backend.

Parameters:

Name	Type	Description	Default
backend	str	'mediapipe' (default) or 'openpose'.	'mediapipe'
**kwargs	Any	Additional keyword arguments forwarded to the estimator constructor.	{}

Returns:

Type	Description
PoseEstimator

Examples:

>>> est = get_pose_estimator("mediapipe", model_complexity=0)

Source code in musicalgestures/_pose_estimator.py

def get_pose_estimator(
    backend: str = "mediapipe",
    **kwargs: Any,
) -> PoseEstimator:
    """Factory function: return a :class:`PoseEstimator` for the requested backend.

    Parameters
    ----------
    backend:
        ``'mediapipe'`` (default) or ``'openpose'``.
    **kwargs:
        Additional keyword arguments forwarded to the estimator constructor.

    Returns
    -------
    PoseEstimator

    Examples
    --------
    >>> est = get_pose_estimator("mediapipe", model_complexity=0)  # doctest: +SKIP
    """
    backend = backend.lower()
    if backend == "mediapipe":
        return MediaPipePoseEstimator(**kwargs)
    elif backend in ("openpose", "caffe"):
        return OpenPosePoseEstimator(**kwargs)
    else:
        raise ValueError(
            f"Unknown pose backend: {backend!r}.  "
            "Choose 'mediapipe' or 'openpose'."
        )