Mse Dominance

class MultiScaleEntropyDominance:
    r"""Real-time multi-scale entropy analyzer for dominance detection.

    Quantifies the complexity of movement dynamics across multiple time scales
    to identify leadership patterns in musical ensembles.

    !!! info
        The algorithm includes coarse-graining, sample entropy calculation, and
        complexity index computation as described in Glowinski et al. (2010).

    Read more in the [User Guide](../../user_guide/theoretical_framework/analysis_primitives/mse_dominance.md).

    Parameters
    ----------
    m : int, optional
        Embedding dimension for sample entropy. Defaults to `2`.
    r : float, optional
        Tolerance threshold for template matching. Defaults to `0.15`.
    max_scale : int, optional
        Maximum scale factor ($\\tau$) for coarse-graining. Defaults to `6`.
    min_points : int, optional
        Minimum number of points required for reliable entropy estimation at
        any given scale. Defaults to `500`.
    methods : list of str, optional
        Analysis components to compute. Choices: `"complexity_index"`,
        `"dominance_score"`, `"leader_identification"`.
        Defaults to `["complexity_index"]`.

    References
    ----------
    Glowinski et al. (2010). Multi-scale entropy analysis of dominance in
    social creative activities. ACM Multimedia, 1035-1038.
    """
    _ALLOWED_METHODS = ["complexity_index", "dominance_score", "leader_identification"]

    def __init__(self, 
                 m=2, 
                 r=0.15, 
                 max_scale=6, 
                 min_points=500, 
                 methods: list[Literal["complexity_index", "dominance_score", "leader_identification"]] = ["complexity_index"]):

        self.m = m
        self.r = r
        self.max_scale = max_scale
        self.min_points = min_points
        self.methods = methods

    @property
    def m(self) -> int:
        return self._m

    @m.setter
    def m(self, value):
        self._m = validate_integer(value, "m", min_val=1)

    @property
    def r(self) -> float:
        return self._r

    @r.setter
    def r(self, value):
        self._r = validate_numeric(value, "r", min_val=0.0001, max_val=0.9999)

    @property
    def max_scale(self) -> int:
        return self._max_scale

    @max_scale.setter
    def max_scale(self, value):
        self._max_scale = validate_integer(value, "max_scale", min_val=1)

    @property
    def min_points(self) -> int:
        return self._min_points

    @min_points.setter
    def min_points(self, value):
        self._min_points = validate_integer(value, "min_points", min_val=1)

    @property
    def methods(self) -> list[str]:
        return self._methods

    @methods.setter
    def methods(self, value):
        self._methods = [validate_string(method, self._ALLOWED_METHODS) for method in value]

    def _coarse_grain(self, data: np.ndarray, scale: int) -> np.ndarray:
        if data is None or data.size == 0:
            return np.array([], dtype=float)

        x = np.asarray(data, dtype=float).ravel()

        if scale is None or scale < 1:
            return np.array([], dtype=float)

        if scale == 1:
            return x

        N = x.shape[0]
        if N < scale:
            return np.array([], dtype=float)

        # Calculate number of complete blocks
        num_points = N // scale

        # Trim data to complete blocks
        trimmed = x[:num_points * scale]

        # Reshape and average: each block becomes one point
        coarse = trimmed.reshape(num_points, scale).mean(axis=1)

        return coarse

    def _sample_entropy(self, data: np.ndarray) -> float:
        x = np.asarray(data, dtype=float).reshape(-1)
        N = x.shape[0]
        m = int(self._m)
        r = float(self._r)

        if N <= m + 10:
            return np.nan

        mu = float(np.mean(x))
        sd = float(np.std(x))
        if sd < 1e-10:
            return np.nan

        u = (x - mu) / sd

        templates_m = np.array([u[i:i + m] for i in range(N - m)], dtype=float)
        templates_m1 = np.array([u[i:i + m + 1] for i in range(N - m - 1)], dtype=float)

        n_m = templates_m.shape[0]
        n_m1 = templates_m1.shape[0]

        if n_m <= 1 or n_m1 <= 1:
            return 0.0

        B_matches = 0
        A_matches = 0

        for i in range(n_m):
            dist = np.max(np.abs(templates_m[i] - templates_m), axis=1)
            B_matches += int(np.sum(dist < r) - 1)

        for i in range(n_m1):
            dist = np.max(np.abs(templates_m1[i] - templates_m1), axis=1)
            A_matches += int(np.sum(dist < r) - 1)

        if B_matches <= 0 or A_matches <= 0:
            return 0.0

        B = B_matches / (n_m * (n_m - 1))
        A = A_matches / (n_m1 * (n_m1 - 1))

        if A <= 0 or B <= 0:
            return 0.0

        return float(-np.log(A / B))

    def _calculate_complexity_index(self, data: np.ndarray) -> float:
        sampen_values = []

        for scale in range(1, int(self._max_scale) + 1):
            coarse = self._coarse_grain(data, scale)

            if coarse.shape[0] < int(self._min_points):
                break

            sampen = self._sample_entropy(coarse)
            sampen_values.append(sampen)

        if len(sampen_values) > 1:
            scales = np.arange(1, len(sampen_values) + 1, dtype=float)
            return float(np.trapz(np.asarray(sampen_values, dtype=float), x=scales))

        if len(sampen_values) == 1:
            return float(sampen_values[0])

        return 0.0

    def __call__(self, signals: SlidingWindow) -> dict:
        """Compute dominance analysis for ensemble performance data.

        Evaluates complexity across multiple scales for each signal in the
        window and optionally computes dominance scores and identifies leaders.

        Parameters
        ----------
        signals : SlidingWindow
            Sliding window buffer containing movement velocity data.

        Returns
        -------
        dict
            Dictionary containing the requested results (e.g.,
            `'complexity_index'`, `'dominance_score'`).
        """
        if not signals.is_full():
            return {method: np.nan for method in self._methods}

        data, _ = signals.to_array(as2D=True)
        n_samples, n_features = data.shape

        if n_samples < int(self._min_points):
            return {method: np.nan for method in self._methods}

        complexity_indices = []
        for i in range(n_features):
            ci = self._calculate_complexity_index(data[:, i])
            complexity_indices.append(ci)

        result = {}

        for method in self._methods:
            if method == 'complexity_index':
                values = np.array(complexity_indices, dtype=float)
                result['complexity_index'] = float(values[0]) if len(values) == 1 else values.tolist()

            elif method == 'dominance_score':
                cis = np.array(complexity_indices, dtype=float)
                if cis.size > 0:
                    max_ci = float(np.max(cis))
                    if max_ci > 0:
                        scores = (1.0 - (cis / max_ci))
                    else:
                        scores = np.zeros_like(cis)
                    result['dominance_score'] = float(scores[0]) if len(scores) == 1 else scores.tolist()

            elif method == 'leader_identification':
                if complexity_indices:
                    leader_idx = np.argmin(complexity_indices)
                    result['leader_complexity'] = (int(leader_idx),float(complexity_indices[leader_idx]))

        return result