Source code for marivo.analysis.frames.component

"""ComponentFrame and ComponentFrameMeta."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any, Literal

from pydantic import ConfigDict

from marivo.analysis.frames.base import BaseFrame, BaseFrameMeta


def resolve_role_column_name(components: dict[str, str | Any], role: str) -> str:
    """Resolve a composition role to its DataFrame column name.

    Uses the component metric's short name (part after the last dot). Falls
    back to the role name when two components share the same short name.
    """
    from marivo.refs import SemanticRef

    def _to_id(v: str | Any) -> str:
        return v.id if isinstance(v, SemanticRef) else str(v)

    short_name: str = _to_id(components[role]).rsplit(".", 1)[-1]
    short_names: list[str] = [_to_id(mid).rsplit(".", 1)[-1] for mid in components.values()]
    if len(short_names) != len(set(short_names)):
        return role
    return short_name


def resolve_role_columns(components: dict[str, str]) -> list[str]:
    """Resolve all composition roles to their DataFrame column names."""
    return [resolve_role_column_name(components, role) for role in components]


class ComponentFrameMeta(BaseFrameMeta):
    model_config = ConfigDict(extra="forbid")

    kind: Literal["component_frame"] = "component_frame"
    parent_ref: str
    parent_kind: Literal["metric_frame", "delta_frame"]
    metric_id: str
    composition_kind: Literal["ratio", "weighted_average", "linear"]
    components: dict[str, str]
    linear_terms: tuple[tuple[str, str], ...] = ()
    axes: dict[str, Any]
    semantic_kind: Literal["scalar", "time_series", "segmented", "panel"]
    semantic_model: str


[docs] @dataclass(repr=False) class ComponentFrame(BaseFrame): meta: ComponentFrameMeta _NEXT_INTENTS: tuple[str, ...] = () def _repr_identity(self) -> str: return ( f"ComponentFrame ref={self.meta.ref} parent={self.meta.parent_ref} " f"metric={self.meta.metric_id} rows={self.meta.row_count}" )