from datetime import datetime
from typing import Any, Dict, Optional
from platypush.message.event import Event
[docs]
class MatrixEvent(Event):
"""
Base matrix event.
"""
[docs]
def __init__(
self,
*args,
server_url: str,
sender_id: Optional[str] = None,
sender_display_name: Optional[str] = None,
sender_avatar_url: Optional[str] = None,
room_id: Optional[str] = None,
room_name: Optional[str] = None,
room_topic: Optional[str] = None,
server_timestamp: Optional[datetime] = 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)
[docs]
class MatrixSyncEvent(MatrixEvent):
"""
Event triggered when the startup synchronization has been completed and the
plugin is ready to use.
"""
[docs]
class MatrixMessageEvent(MatrixEvent):
"""
Event triggered when a message is received on a subscribed room.
"""
[docs]
def __init__(
self,
*args,
body: str = '',
url: Optional[str] = None,
thumbnail_url: Optional[str] = None,
mimetype: Optional[str] = None,
formatted_body: Optional[str] = None,
format: Optional[str] = None,
**kwargs
):
"""
:param body: The body of the message.
:param url: The URL of the media file, if the message includes media.
:param thumbnail_url: The URL of the thumbnail, if the message includes media.
:param mimetype: The MIME type of the media file, if the message includes media.
:param formatted_body: The formatted body, if ``format`` is specified.
:param format: The format of the message (e.g. ``html`` or ``markdown``).
"""
super().__init__(
*args,
body=body,
url=url,
thumbnail_url=thumbnail_url,
mimetype=mimetype,
formatted_body=formatted_body,
format=format,
**kwargs
)
[docs]
class MatrixMessageImageEvent(MatrixEvent):
"""
Event triggered when a message containing an image is received.
"""
[docs]
class MatrixMessageFileEvent(MatrixEvent):
"""
Event triggered when a message containing a generic file is received.
"""
[docs]
class MatrixMessageAudioEvent(MatrixEvent):
"""
Event triggered when a message containing an audio file is received.
"""
[docs]
class MatrixMessageVideoEvent(MatrixEvent):
"""
Event triggered when a message containing a video file is received.
"""
[docs]
class MatrixReactionEvent(MatrixEvent):
"""
Event triggered when a user submits a reaction to an event.
"""
[docs]
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
)
[docs]
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.
"""
[docs]
class MatrixCallEvent(MatrixEvent):
"""
Base class for Matrix call events.
"""
[docs]
def __init__(
self, *args, call_id: str, version: int, sdp: Optional[str] = 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)
[docs]
class MatrixCallInviteEvent(MatrixCallEvent):
"""
Event triggered when the user is invited to a call.
"""
[docs]
def __init__(self, *args, invite_validity: Optional[float] = 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)
[docs]
class MatrixCallAnswerEvent(MatrixCallEvent):
"""
Event triggered by the callee when they wish to answer the call.
"""
[docs]
class MatrixCallHangupEvent(MatrixCallEvent):
"""
Event triggered when a participant in the call exists.
"""
[docs]
class MatrixRoomCreatedEvent(MatrixEvent):
"""
Event triggered when a room is created.
"""
[docs]
class MatrixRoomJoinEvent(MatrixEvent):
"""
Event triggered when a user joins a room.
"""
[docs]
class MatrixRoomLeaveEvent(MatrixEvent):
"""
Event triggered when a user leaves a room.
"""
[docs]
class MatrixRoomInviteEvent(MatrixEvent):
"""
Event triggered when the user is invited to a room.
"""
[docs]
class MatrixRoomTopicChangedEvent(MatrixEvent):
"""
Event triggered when the topic/title of a room changes.
"""
[docs]
def __init__(self, *args, topic: str, **kwargs):
"""
:param topic: New room topic.
"""
super().__init__(*args, topic=topic, **kwargs)
[docs]
class MatrixRoomTypingStartEvent(MatrixEvent):
"""
Event triggered when a user in a room starts typing.
"""
[docs]
class MatrixRoomTypingStopEvent(MatrixEvent):
"""
Event triggered when a user in a room stops typing.
"""
[docs]
class MatrixRoomSeenReceiptEvent(MatrixEvent):
"""
Event triggered when the last message seen by a user in a room is updated.
"""
[docs]
class MatrixUserPresenceEvent(MatrixEvent):
"""
Event triggered when a user comes online or goes offline.
"""
[docs]
def __init__(
self, *args, is_active: bool, last_active: Optional[datetime], **kwargs
):
"""
:param is_active: True if the user is currently online.
:param topic: When the user was last active.
"""
super().__init__(*args, is_active=is_active, last_active=last_active, **kwargs)
