1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
|
# Copyright (c) 2022 Tulir Asokan
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
from __future__ import annotations
from mautrix.api import Method, Path
from mautrix.errors import MatrixResponseError
from mautrix.types import (
JSON,
EventID,
PresenceEventContent,
PresenceState,
RoomID,
SerializerError,
UserID,
)
from ..base import BaseClientAPI
class MiscModuleMethods(BaseClientAPI):
"""
Miscellaneous subsections in the `Modules section`_ of the API spec.
Currently included subsections:
* 13.4 `Typing Notifications`_
* 13.5 `Receipts`_
* 13.6 `Fully Read Markers`_
* 13.7 `Presence`_
.. _Modules section: https://matrix.org/docs/spec/client_server/r0.4.0.html#modules
.. _Typing Notifications: https://matrix.org/docs/spec/client_server/r0.4.0.html#id95
.. _Receipts: https://matrix.org/docs/spec/client_server/r0.4.0.html#id99
.. _Fully Read Markers: https://matrix.org/docs/spec/client_server/r0.4.0.html#fully-read-markers
.. _Presence: https://matrix.org/docs/spec/client_server/r0.4.0.html#id107
"""
# region 13.4 Typing Notifications
async def set_typing(self, room_id: RoomID, timeout: int = 0) -> None:
"""
This tells the server that the user is typing for the next N milliseconds where N is the
value specified in the timeout key. If the timeout is equal to or less than zero, it tells
the server that the user has stopped typing.
See also: `API reference <https://matrix.org/docs/spec/client_server/r0.4.0.html#put-matrix-client-r0-rooms-roomid-typing-userid>`__
Args:
room_id: The ID of the room in which the user is typing.
timeout: The length of time in milliseconds to mark this user as typing.
"""
if timeout > 0:
content = {"typing": True, "timeout": timeout}
else:
content = {"typing": False}
await self.api.request(Method.PUT, Path.v3.rooms[room_id].typing[self.mxid], content)
# endregion
# region 13.5 Receipts
async def send_receipt(
self,
room_id: RoomID,
event_id: EventID,
receipt_type: str = "m.read",
) -> None:
"""
Update the marker for the given receipt type to the event ID specified.
See also: `API reference <https://matrix.org/docs/spec/client_server/r0.4.0.html#post-matrix-client-r0-rooms-roomid-receipt-receipttype-eventid>`__
Args:
room_id: The ID of the room which to send the receipt to.
event_id: The last event ID to acknowledge.
receipt_type: The type of receipt to send. Currently only ``m.read`` is supported.
"""
await self.api.request(Method.POST, Path.v3.rooms[room_id].receipt[receipt_type][event_id])
# endregion
# region 13.6 Fully read markers
async def set_fully_read_marker(
self,
room_id: RoomID,
fully_read: EventID,
read_receipt: EventID | None = None,
extra_content: dict[str, JSON] | None = None,
) -> None:
"""
Set the position of the read marker for the given room, and optionally send a new read
receipt.
See also: `API reference <https://matrix.org/docs/spec/client_server/r0.4.0.html#post-matrix-client-r0-rooms-roomid-read-markers>`__
Args:
room_id: The ID of the room which to set the read marker in.
fully_read: The last event up to which the user has either read all events or is not
interested in reading the events.
read_receipt: The new position for the user's normal read receipt, i.e. the last event
the user has seen.
extra_content: Additional fields to include in the ``/read_markers`` request.
"""
content = {
"m.fully_read": fully_read,
}
if read_receipt:
content["m.read"] = read_receipt
if extra_content:
content.update(extra_content)
await self.api.request(Method.POST, Path.v3.rooms[room_id].read_markers, content)
# endregion
# region 13.7 Presence
async def set_presence(
self, presence: PresenceState = PresenceState.ONLINE, status: str | None = None
) -> None:
"""
Set the current user's presence state. When setting the status, the activity time is updated
to reflect that activity; the client does not need to specify
:attr:`Presence.last_active_ago`.
See also: `API reference <https://matrix.org/docs/spec/client_server/r0.4.0.html#post-matrix-client-r0-presence-list-userid>`__
Args:
presence: The new presence state to set.
status: The status message to attach to this state.
"""
content = {
"presence": presence.value,
}
if status:
content["status_msg"] = status
await self.api.request(Method.PUT, Path.v3.presence[self.mxid].status, content)
async def get_presence(self, user_id: UserID) -> PresenceEventContent:
"""
Get the presence info of a user.
See also: `API reference <https://matrix.org/docs/spec/client_server/r0.4.0.html#get-matrix-client-r0-presence-list-userid>`__
Args:
user_id: The ID of the user whose presence info to get.
Returns:
The presence info of the given user.
"""
content = await self.api.request(Method.GET, Path.v3.presence[user_id].status)
try:
return PresenceEventContent.deserialize(content)
except SerializerError:
raise MatrixResponseError("Invalid presence in response")
# endregion
|
