server: Initial typing notice sending support.

matrix/buffer.py

@@ -49,7 +49,7 @@ from nio import (
 from . import globals as G
 from .colors import Formatted
 from .config import RedactType
-from .globals import SCRIPT_NAME, SERVERS, W
+from .globals import SCRIPT_NAME, SERVERS, W, TYPING_NOTICE_TIMEOUT
 from .utf import utf8_decode
 from .utils import server_ts_to_weechat, shorten_sender, string_strikethrough
 
@@ -414,6 +414,12 @@ class WeechatChannelBuffer(object):
 
         self.remove_smart_filtered_nick(nick)
 
+    @property
+    def input(self):
+        # type: () -> str
+        """Get the bar item input text of the buffer."""
+        return W.buffer_get_string(self._ptr, "input")
+
     @property
     def lines(self):
         own_lines = W.hdata_pointer(self._hdata, self._ptr, "own_lines")
@@ -819,6 +825,9 @@ class RoomBuffer(object):
         self.printed_before_ack_queue = list()  # type: List[UUID]
         self.undecrypted_events = deque(maxlen=5000)
 
+        self.typing_notice_time = None
+        self._typing = False
+
         buffer_name = "{}.{}".format(server_name, room.room_id)
 
         # This dict remembers the connection from a user_id to the name we
@@ -860,6 +869,35 @@ class RoomBuffer(object):
     def warning_prefix(self):
         return "⚠️ "
 
+    @property
+    def typing(self):
+        # type: () -> bool
+        """Return our typing status."""
+        return self._typing
+
+    @typing.setter
+    def typing(self, value):
+        self._typing = value
+        if value:
+            self.typing_notice_time = time.time()
+        else:
+            self.typing_notice_time = None
+
+    @property
+    def typing_notice_expired(self):
+        # type: () -> bool
+        """Check if the typing notice has expired.
+
+        Returns true if a new typing notice should be sent.
+        """
+        if not self.typing_notice_time:
+            return True
+
+        now = time.time()
+        if (now - self.typing_notice_time) > (TYPING_NOTICE_TIMEOUT / 1000):
+            return True
+        return False
+
     def find_nick(self, user_id):
         # type: (str) -> str
         """Find a suitable nick from a user_id"""

matrix/globals.py

@@ -40,3 +40,4 @@ CONFIG = None  # type: Optional[MatrixConfig]
 ENCRYPTION = True  # type: bool
 SCRIPT_NAME = "matrix"  # type: str
 MAX_EVENTS = 100
+TYPING_NOTICE_TIMEOUT = 4000  # 4 seconds typing notice lifetime

matrix/server.py

@@ -56,7 +56,7 @@ from nio import (
 from . import globals as G
 from .buffer import OwnAction, OwnMessage, RoomBuffer
 from .config import ConfigSection, Option, ServerBufferType
-from .globals import SCRIPT_NAME, SERVERS, W, MAX_EVENTS
+from .globals import SCRIPT_NAME, SERVERS, W, MAX_EVENTS, TYPING_NOTICE_TIMEOUT
 from .utf import utf8_decode
 from .utils import create_server_buffer, key_from_value, server_buffer_prnt
 
@@ -673,6 +673,49 @@ class MatrixServer(object):
         self.backlog_queue[uuid] = room_id
         self.send_or_queue(request)
 
+    def room_send_typing_notice(self, room_buffer):
+        """Send a typing notice for the provided room.
+
+        Args:
+            room_buffer(RoomBuffer): the room for which the typing notice needs
+                to be sent.
+        """
+        if not self.connected:
+            return
+
+        input = room_buffer.weechat_buffer.input
+
+        # Don't send a typing notice if the user is typing in a weechat command
+        if input.startswith("/") and not input.startswith("//"):
+            return
+
+        # Don't send a typing notice if we only typed a couple of letters.
+        elif len(input) < 4 and not room_buffer.typing:
+            return
+
+        # If we were typing already and our input bar now has no letters or
+        # only a couple of letters stop the typing notice.
+        elif len(input) < 4:
+            _, request = self.client.room_typing(
+                room_buffer.room.room_id,
+                typing_state=False)
+            room_buffer.typing = False
+            self.send(request)
+            return
+
+        # Don't send out a typing notice if we already sent one out and it
+        # didn't expire yet.
+        if not room_buffer.typing_notice_expired:
+            return
+
+        _, request = self.client.room_typing(
+            room_buffer.room.room_id,
+            typing_state=True,
+            timeout=TYPING_NOTICE_TIMEOUT)
+
+        room_buffer.typing = True
+        self.send(request)
+
     def room_send_message(
         self,
         room_buffer,  # type: RoomBuffer