Module livekit.plugins.anam

class AnamException(Exception):
    """Custom exception for Anam API errors."""

    pass

class AvatarSession(BaseAvatarSession):
    """A Anam avatar session"""

    def __init__(
        self,
        *,
        persona_config: PersonaConfig,
        session_options: NotGivenOr[SessionOptions] = NOT_GIVEN,
        api_url: NotGivenOr[str] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        avatar_participant_identity: NotGivenOr[str] = NOT_GIVEN,
        avatar_participant_name: NotGivenOr[str] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> None:
        super().__init__()
        self._http_session: aiohttp.ClientSession | None = None
        self._conn_options = conn_options
        self.session_id: str | None = None
        self._avatar_participant_identity = avatar_participant_identity or _AVATAR_AGENT_IDENTITY
        self._avatar_participant_name = avatar_participant_name or _AVATAR_AGENT_NAME
        self._persona_config: PersonaConfig = persona_config
        self._session_options = session_options if utils.is_given(session_options) else None

        api_url_val = (
            api_url if utils.is_given(api_url) else os.getenv("ANAM_API_URL", DEFAULT_API_URL)
        )
        api_key_val = api_key if utils.is_given(api_key) else os.getenv("ANAM_API_KEY")

        if not api_key_val:
            raise AnamException("ANAM_API_KEY must be set by arguments or environment variables")

        self._api_url = api_url_val
        self._api_key = api_key_val

    @property
    def avatar_identity(self) -> str:
        return self._avatar_participant_identity

    @property
    def provider(self) -> str:
        return "anam"

    def _ensure_http_session(self) -> aiohttp.ClientSession:
        if self._http_session is None:
            self._http_session = utils.http_context.http_session()

        return self._http_session

    async def start(
        self,
        agent_session: AgentSession,
        room: rtc.Room,
        *,
        livekit_url: NotGivenOr[str] = NOT_GIVEN,
        livekit_api_key: NotGivenOr[str] = NOT_GIVEN,
        livekit_api_secret: NotGivenOr[str] = NOT_GIVEN,
    ) -> None:
        await super().start(agent_session, room)

        livekit_url = livekit_url or (os.getenv("LIVEKIT_URL") or NOT_GIVEN)
        livekit_api_key = livekit_api_key or (os.getenv("LIVEKIT_API_KEY") or NOT_GIVEN)
        livekit_api_secret = livekit_api_secret or (os.getenv("LIVEKIT_API_SECRET") or NOT_GIVEN)
        if not livekit_url or not livekit_api_key or not livekit_api_secret:
            raise AnamException(
                "livekit_url, livekit_api_key, and livekit_api_secret must be set "
                "by arguments or environment variables"
            )

        job_ctx = get_job_context()
        local_participant_identity = job_ctx.local_participant_identity
        livekit_token = (
            api.AccessToken(
                api_key=livekit_api_key,
                api_secret=livekit_api_secret,
            )
            .with_kind("agent")
            .with_identity(self._avatar_participant_identity)
            .with_name(self._avatar_participant_name)
            .with_grants(api.VideoGrants(room_join=True, room=room.name))
            # allow the avatar agent to publish audio and video on behalf of your local agent
            .with_attributes({ATTRIBUTE_PUBLISH_ON_BEHALF: local_participant_identity})
            .to_jwt()
        )
        async with AnamAPI(
            api_key=self._api_key, api_url=self._api_url, conn_options=self._conn_options
        ) as anam_api:
            session_token = await anam_api.create_session_token(
                persona_config=self._persona_config,
                livekit_url=livekit_url,
                livekit_token=livekit_token,
                session_options=self._session_options,
            )
            logger.debug("Anam session token created successfully.")

            logger.debug("Starting Anam engine session...")
            session_details = await anam_api.start_engine_session(
                session_token=session_token,
            )
            self.session_id = session_details.get("sessionId")

        agent_session.output.replace_audio_tail(
            DataStreamAudioOutput(
                room=room,
                destination_identity=self._avatar_participant_identity,
                sample_rate=SAMPLE_RATE,
                wait_remote_track=rtc.TrackKind.KIND_VIDEO,
            ),
        )

@dataclass
class PersonaConfig:
    """Configuration for Anam avatar persona"""

    name: str
    avatarId: str
    avatarModel: str | None = None

@dataclass
class SessionOptions:
    """Per-session output options forwarded to Anam's session-token API.

    Mirrors the ``sessionOptions`` field of the Anam session-token request.

    Attributes:
        video_width: Output video frame width in pixels. Provide together with
            ``video_height`` (both or neither). Omit to use the avatar model's
            default output size. Supported pairs are model-dependent and are
            validated by Anam; an unsupported pair is rejected with an HTTP 400
            rather than silently downgraded.
        video_height: Output video frame height in pixels. See ``video_width``.
    """

    video_width: int | None = None
    video_height: int | None = None