platypush/platypush/message/event/matrix.py

from datetime import datetime
from typing import Dict, Any

from platypush.message.event import Event


class MatrixEvent(Event):
    """
    Base matrix event.
    """

    def __init__(
        self,
        *args,
        server_url: str,
        sender_id: str | None = None,
        sender_display_name: str | None = None,
        sender_avatar_url: str | None = None,
        room_id: str | None = None,
        room_name: str | None = None,
        room_topic: str | None = None,
        server_timestamp: datetime | None = None,
        **kwargs
    ):
        """
        :param server_url: Base server URL.
        :param sender_id: The event's sender ID.
        :param sender_display_name: The event's sender display name.
        :param sender_avatar_url: The event's sender avatar URL.
        :param room_id: Event room ID.
        :param room_name: The name of the room associated to the event.
        :param room_topic: The topic of the room associated to the event.
        :param server_timestamp: The server timestamp of the event.
        """
        evt_args: Dict[str, Any] = {
            'server_url': server_url,
        }

        if sender_id:
            evt_args['sender_id'] = sender_id
        if sender_display_name:
            evt_args['sender_display_name'] = sender_display_name
        if sender_avatar_url:
            evt_args['sender_avatar_url'] = sender_avatar_url
        if room_id:
            evt_args['room_id'] = room_id
        if room_name:
            evt_args['room_name'] = room_name
        if room_topic:
            evt_args['room_topic'] = room_topic
        if server_timestamp:
            evt_args['server_timestamp'] = server_timestamp

        super().__init__(*args, **evt_args, **kwargs)


class MatrixSyncEvent(MatrixEvent):
    """
    Event triggered when the startup synchronization has been completed and the
    plugin is ready to use.
    """


class MatrixMessageEvent(MatrixEvent):
    """
    Event triggered when a message is received on a subscribed room.
    """

    def __init__(self, *args, body: str, **kwargs):
        """
        :param body: The body of the message.
        """
        super().__init__(*args, body=body, **kwargs)


class MatrixMediaMessageEvent(MatrixMessageEvent):
    """
    Event triggered when a media message is received on a subscribed room.
    """

    def __init__(self, *args, url: str, **kwargs):
        """
        :param url: The URL of the media file.
        """
        super().__init__(*args, url=url, **kwargs)


class MatrixStickerEvent(MatrixMediaMessageEvent):
    """
    Event triggered when a sticker is sent to a room.
    """


class MatrixReactionEvent(MatrixEvent):
    """
    Event triggered when a user submits a reaction to an event.
    """

    def __init__(self, *args, in_response_to_event_id: str, **kwargs):
        """
        :param in_response_to_event_id: The ID of the URL related to the reaction.
        """
        super().__init__(
            *args, in_response_to_event_id=in_response_to_event_id, **kwargs
        )


class MatrixEncryptedMessageEvent(MatrixMessageEvent):
    """
    Event triggered when a message is received but the client doesn't
    have the E2E keys to decrypt it, or encryption has not been enabled.
    """


class MatrixCallEvent(MatrixEvent):
    """
    Base class for Matrix call events.
    """

    def __init__(
        self, *args, call_id: str, version: int, sdp: str | None = None, **kwargs
    ):
        """
        :param call_id: The unique ID of the call.
        :param version: An increasing integer representing the version of the call.
        :param sdp: SDP text of the session description.
        """
        super().__init__(*args, call_id=call_id, version=version, sdp=sdp, **kwargs)


class MatrixCallInviteEvent(MatrixCallEvent):
    """
    Event triggered when the user is invited to a call.
    """

    def __init__(self, *args, invite_validity: float | None = None, **kwargs):
        """
        :param invite_validity: For how long the invite will be valid, in seconds.
        :param sdp: SDP text of the session description.
        """
        super().__init__(*args, invite_validity=invite_validity, **kwargs)


class MatrixCallAnswerEvent(MatrixCallEvent):
    """
    Event triggered by the callee when they wish to answer the call.
    """


class MatrixCallHangupEvent(MatrixCallEvent):
    """
    Event triggered when a participant in the call exists.
    """


class MatrixRoomCreatedEvent(MatrixEvent):
    """
    Event triggered when a room is created.
    """


class MatrixRoomJoinEvent(MatrixEvent):
    """
    Event triggered when a user joins a room.
    """


class MatrixRoomLeaveEvent(MatrixEvent):
    """
    Event triggered when a user leaves a room.
    """


class MatrixRoomInviteEvent(MatrixEvent):
    """
    Event triggered when the user is invited to a room.
    """


class MatrixRoomTopicChangedEvent(MatrixEvent):
    """
    Event triggered when the topic/title of a room changes.
    """

    def __init__(self, *args, topic: str, **kwargs):
        """
        :param topic: New room topic.
        """
        super().__init__(*args, topic=topic, **kwargs)