summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authormathieui <mathieui@mathieui.net>2020-06-01 09:09:36 +0200
committermathieui <mathieui@mathieui.net>2020-06-01 09:09:36 +0200
commit291233bbbd81fcb1fddf6e4d1cd93478f73c1ade (patch)
tree70a184a22724be16ce7a7ef930d07e9eda203ade
parente662bdee055d25cb609b5ed229e077e3d557bbe0 (diff)
parentc9e219c1405604f2ef81b3e446cc33a0492af42d (diff)
downloadpoezio-291233bbbd81fcb1fddf6e4d1cd93478f73c1ade.tar.gz
poezio-291233bbbd81fcb1fddf6e4d1cd93478f73c1ade.tar.bz2
poezio-291233bbbd81fcb1fddf6e4d1cd93478f73c1ade.tar.xz
poezio-291233bbbd81fcb1fddf6e4d1cd93478f73c1ade.zip
Merge branch 'more-typing' into 'master'
More typing See merge request poezio/poezio!135
-rw-r--r--poezio/common.py22
-rw-r--r--poezio/core/core.py29
-rw-r--r--poezio/core/structs.py38
-rw-r--r--poezio/core/tabs.py8
-rw-r--r--poezio/decorators.py63
-rw-r--r--poezio/fixes.py5
-rw-r--r--poezio/multiuserchat.py149
-rw-r--r--poezio/poezio.py2
-rw-r--r--poezio/poezio_shlex.pyi45
-rw-r--r--poezio/roster.py28
-rw-r--r--poezio/tabs/basetabs.py38
-rw-r--r--poezio/tabs/muctab.py358
-rw-r--r--poezio/tabs/privatetab.py2
-rw-r--r--poezio/text_buffer.py3
-rw-r--r--poezio/user.py2
-rw-r--r--poezio/windows/info_wins.py17
16 files changed, 491 insertions, 318 deletions
diff --git a/poezio/common.py b/poezio/common.py
index 98870dda..315d5b9e 100644
--- a/poezio/common.py
+++ b/poezio/common.py
@@ -14,7 +14,7 @@ from datetime import (
timezone,
)
from pathlib import Path
-from typing import Dict, List, Optional, Tuple, Union
+from typing import Dict, List, Optional, Tuple, Union, Any
import os
import subprocess
@@ -44,7 +44,7 @@ def _get_output_of_command(command: str) -> Optional[List[str]]:
return None
-def _is_in_path(command: str, return_abs_path=False) -> Union[bool, str]:
+def _is_in_path(command: str, return_abs_path: bool = False) -> Union[bool, str]:
"""
Check if *command* is in the $PATH or not.
@@ -111,10 +111,12 @@ def get_os_info() -> str:
stdout=subprocess.PIPE,
close_fds=True)
process.wait()
- output = process.stdout.readline().decode('utf-8').strip()
- # some distros put n/a in places, so remove those
- output = output.replace('n/a', '').replace('N/A', '')
- return output
+ if process.stdout is not None:
+ out = process.stdout.readline().decode('utf-8').strip()
+ # some distros put n/a in places, so remove those
+ out = out.replace('n/a', '').replace('N/A', '')
+ return out
+ return ''
# lsb_release executable not available, so parse files
for distro_name in DISTRO_INFO:
@@ -287,7 +289,7 @@ def shell_split(st: str) -> List[str]:
return ret
-def find_argument(pos: int, text: str, quoted=True) -> int:
+def find_argument(pos: int, text: str, quoted: bool = True) -> int:
"""
Split an input into a list of arguments, return the number of the
argument selected by pos.
@@ -342,7 +344,7 @@ def _find_argument_unquoted(pos: int, text: str) -> int:
return argnum + 1
-def parse_str_to_secs(duration='') -> int:
+def parse_str_to_secs(duration: str = '') -> int:
"""
Parse a string of with a number of d, h, m, s.
@@ -370,7 +372,7 @@ def parse_str_to_secs(duration='') -> int:
return result
-def parse_secs_to_str(duration=0) -> str:
+def parse_secs_to_str(duration: int = 0) -> str:
"""
Do the reverse operation of :py:func:`parse_str_to_secs`.
@@ -457,7 +459,7 @@ def format_gaming_string(infos: Dict[str, str]) -> str:
return name
-def safeJID(*args, **kwargs) -> JID:
+def safeJID(*args: Any, **kwargs: Any) -> JID:
"""
Construct a :py:class:`slixmpp.JID` object from a string.
diff --git a/poezio/core/core.py b/poezio/core/core.py
index 3ad15719..3a13d4c3 100644
--- a/poezio/core/core.py
+++ b/poezio/core/core.py
@@ -15,7 +15,17 @@ import shutil
import time
import uuid
from collections import defaultdict
-from typing import Callable, Dict, List, Optional, Set, Tuple, Type
+from typing import (
+ cast,
+ Callable,
+ Dict,
+ List,
+ Optional,
+ Set,
+ Tuple,
+ Type,
+ TypeVar,
+)
from xml.etree import ElementTree as ET
from functools import partial
@@ -65,6 +75,7 @@ from poezio.ui.types import Message, InfoMessage
log = logging.getLogger(__name__)
+T = TypeVar('T', bound=tabs.Tab)
class Core:
"""
@@ -99,8 +110,10 @@ class Core:
# that are displayed in almost all tabs, in an
# information window.
self.information_buffer = TextBuffer()
- self.information_win_size = config.get(
- 'info_win_height', section='var')
+ self.information_win_size = cast(
+ int,
+ config.get('info_win_height', section='var'),
+ )
self.information_win = windows.TextWin(300)
self.information_buffer.add_window(self.information_win)
self.left_tab_win = None
@@ -813,7 +826,7 @@ class Core:
####################### XMPP-related actions ##################################
- def get_status(self) -> str:
+ def get_status(self) -> Status:
"""
Get the last status that was previously set
"""
@@ -1016,7 +1029,7 @@ class Core:
### Tab getters ###
- def get_tabs(self, cls: Type[tabs.Tab] = None) -> List[tabs.Tab]:
+ def get_tabs(self, cls: Type[T] = None) -> List[T]:
"Get all the tabs of a type"
if cls is None:
return self.tabs.get_tabs()
@@ -1324,7 +1337,7 @@ class Core:
if tab.name.startswith(room_name):
tab.activate(reason=reason)
- def on_user_changed_status_in_private(self, jid: JID, status: str) -> None:
+ def on_user_changed_status_in_private(self, jid: JID, status: Status) -> None:
tab = self.tabs.by_name_and_class(jid, tabs.ChatTab)
if tab is not None: # display the message in private
tab.update_status(status)
@@ -1652,7 +1665,7 @@ class Core:
return
else:
scr = self.stdscr
- tabs.Tab.resize(scr)
+ tabs.Tab.initial_resize(scr)
self.resize_global_info_bar()
self.resize_global_information_win()
for tab in self.tabs:
@@ -2105,7 +2118,7 @@ class Core:
self.bookmarks.get_remote(self.xmpp, self.information,
_join_remote_only)
- def room_error(self, error, room_name):
+ def room_error(self, error: IqError, room_name: str) -> None:
"""
Display the error in the tab
"""
diff --git a/poezio/core/structs.py b/poezio/core/structs.py
index 72c9628a..a75f1e94 100644
--- a/poezio/core/structs.py
+++ b/poezio/core/structs.py
@@ -1,6 +1,8 @@
"""
Module defining structures useful to the core class and related methods
"""
+from dataclasses import dataclass
+from typing import Any, Callable, List, Dict
__all__ = [
'ERROR_AND_STATUS_CODES', 'DEPRECATED_ERRORS', 'POSSIBLE_SHOW', 'Status',
@@ -51,23 +53,11 @@ POSSIBLE_SHOW = {
}
+@dataclass
class Status:
__slots__ = ('show', 'message')
-
- def __init__(self, show, message):
- self.show = show
- self.message = message
-
-
-class Command:
- __slots__ = ('func', 'desc', 'comp', 'short_desc', 'usage')
-
- def __init__(self, func, desc, comp, short_desc, usage):
- self.func = func
- self.desc = desc
- self.comp = comp
- self.short_desc = short_desc
- self.usage = usage
+ show: str
+ message: str
class Completion:
@@ -75,8 +65,13 @@ class Completion:
A completion result essentially currying the input completion call.
"""
__slots__ = ['func', 'args', 'kwargs', 'comp_list']
-
- def __init__(self, func, comp_list, *args, **kwargs):
+ def __init__(
+ self,
+ func: Callable[..., Any],
+ comp_list: List[str],
+ *args: Any,
+ **kwargs: Any
+ ) -> None:
self.func = func
self.comp_list = comp_list
self.args = args
@@ -84,3 +79,12 @@ class Completion:
def run(self):
return self.func(self.comp_list, *self.args, **self.kwargs)
+
+@dataclass
+class Command:
+ __slots__ = ('func', 'desc', 'comp', 'short_desc', 'usage')
+ func: Callable[..., Any]
+ desc: str
+ comp: Callable[['windows.Input'], Completion]
+ short_desc: str
+ usage: str
diff --git a/poezio/core/tabs.py b/poezio/core/tabs.py
index d5909d39..61bad6f2 100644
--- a/poezio/core/tabs.py
+++ b/poezio/core/tabs.py
@@ -347,16 +347,16 @@ class Tabs:
if new_pos < len(self._tabs):
old_tab = self._tabs[old_pos]
self._tabs[new_pos], self._tabs[
- old_pos] = old_tab, tabs.GapTab(self)
+ old_pos] = old_tab, tabs.GapTab(None)
else:
self._tabs.append(self._tabs[old_pos])
- self._tabs[old_pos] = tabs.GapTab(self)
+ self._tabs[old_pos] = tabs.GapTab(None)
else:
if new_pos > old_pos:
self._tabs.insert(new_pos, tab)
- self._tabs[old_pos] = tabs.GapTab(self)
+ self._tabs[old_pos] = tabs.GapTab(None)
elif new_pos < old_pos:
- self._tabs[old_pos] = tabs.GapTab(self)
+ self._tabs[old_pos] = tabs.GapTab(None)
self._tabs.insert(new_pos, tab)
else:
return False
diff --git a/poezio/decorators.py b/poezio/decorators.py
index 51abf32c..62724ecd 100644
--- a/poezio/decorators.py
+++ b/poezio/decorators.py
@@ -1,54 +1,68 @@
"""
Module containing various decorators
"""
-from typing import Any, Callable, List, Optional
+from typing import (
+ cast,
+ Any,
+ Callable,
+ List,
+ Optional,
+ TypeVar,
+ TYPE_CHECKING,
+)
from poezio import common
+if TYPE_CHECKING:
+ from poezio.tabs import RosterInfoTab
+
+T = TypeVar('T', bound=Callable[..., Any])
+
+
class RefreshWrapper:
- def __init__(self):
+ def __init__(self) -> None:
self.core = None
- def conditional(self, func: Callable) -> Callable:
+ def conditional(self, func: T) -> T:
"""
Decorator to refresh the UI if the wrapped function
returns True
"""
- def wrap(*args, **kwargs):
+ def wrap(*args: Any, **kwargs: Any) -> Any:
ret = func(*args, **kwargs)
if self.core and ret:
self.core.refresh_window()
return ret
- return wrap
+ return cast(T, wrap)
- def always(self, func: Callable) -> Callable:
+ def always(self, func: T) -> T:
"""
Decorator that refreshs the UI no matter what after the function
"""
- def wrap(*args, **kwargs):
+ def wrap(*args: Any, **kwargs: Any) -> Any:
ret = func(*args, **kwargs)
if self.core:
self.core.refresh_window()
return ret
- return wrap
+ return cast(T, wrap)
- def update(self, func: Callable) -> Callable:
+ def update(self, func: T) -> T:
"""
Decorator that only updates the screen
"""
- def wrap(*args, **kwargs):
+ def wrap(*args: Any, **kwargs: Any) -> Any:
ret = func(*args, **kwargs)
if self.core:
self.core.doupdate()
return ret
- return wrap
+ return cast(T, wrap)
refresh_wrapper = RefreshWrapper()
@@ -61,32 +75,32 @@ class CommandArgParser:
"""
@staticmethod
- def raw(func: Callable) -> Callable:
+ def raw(func: T) -> T:
"""Just call the function with a single string, which is the original string
untouched
"""
- def wrap(self, args, *a, **kw):
+ def wrap(self: Any, args: Any, *a: Any, **kw: Any) -> Any:
return func(self, args, *a, **kw)
- return wrap
+ return cast(T, wrap)
@staticmethod
- def ignored(func: Callable) -> Callable:
+ def ignored(func: T) -> T:
"""
Call the function without any argument
"""
- def wrap(self, args=None, *a, **kw):
+ def wrap(self: Any, args: Any = None, *a: Any, **kw: Any) -> Any:
return func(self, *a, **kw)
- return wrap
+ return cast(T, wrap)
@staticmethod
def quoted(mandatory: int,
- optional=0,
+ optional: int = 0,
defaults: Optional[List[Any]] = None,
- ignore_trailing_arguments=False):
+ ignore_trailing_arguments: bool = False) -> Callable[[T], T]:
"""The function receives a list with a number of arguments that is between
the numbers `mandatory` and `optional`.
@@ -131,8 +145,8 @@ class CommandArgParser:
"""
default_args_outer = defaults or []
- def first(func: Callable):
- def second(self, args: str, *a, **kw):
+ def first(func: T) -> T:
+ def second(self: Any, args: str, *a: Any, **kw: Any) -> Any:
default_args = default_args_outer
if args and args.strip():
split_args = common.shell_split(args)
@@ -156,8 +170,7 @@ class CommandArgParser:
res[-1] += " " + " ".join(split_args)
return func(self, res, *a, **kw)
- return second
-
+ return cast(T, second)
return first
@@ -166,11 +179,11 @@ command_args_parser = CommandArgParser()
def deny_anonymous(func: Callable) -> Callable:
"""Decorator to disable commands when using an anonymous account."""
- def wrap(self: 'RosterInfoTab', *args, **kwargs):
+ def wrap(self: 'RosterInfoTab', *args: Any, **kwargs: Any) -> Any:
if self.core.xmpp.anon:
return self.core.information(
'This command is not available for anonymous accounts.',
'Info'
)
return func(self, *args, **kwargs)
- return wrap
+ return cast(T, wrap)
diff --git a/poezio/fixes.py b/poezio/fixes.py
index f8de7b14..a9e15dee 100644
--- a/poezio/fixes.py
+++ b/poezio/fixes.py
@@ -5,7 +5,8 @@ upstream.
TODO: Check that they are fixed and remove those hacks
"""
-from slixmpp.stanza import Message
+from typing import Callable, Any
+from slixmpp import Message, Iq, ClientXMPP
from slixmpp.xmlstream import ET
import logging
@@ -25,7 +26,7 @@ def has_identity(xmpp, jid, identity, on_true=None, on_false=None):
xmpp.plugin['xep_0030'].get_info(jid=jid, callback=_cb)
-def get_room_form(xmpp, room, callback):
+def get_room_form(xmpp: ClientXMPP, room: str, callback: Callable[[Iq], Any]):
def _cb(result):
if result["type"] == "error":
return callback(None)
diff --git a/poezio/multiuserchat.py b/poezio/multiuserchat.py
index 30c36a77..12f97661 100644
--- a/poezio/multiuserchat.py
+++ b/poezio/multiuserchat.py
@@ -11,18 +11,39 @@ slix plugin
"""
from xml.etree import ElementTree as ET
+from typing import (
+ Callable,
+ Optional,
+ TYPE_CHECKING,
+)
from poezio.common import safeJID
-from slixmpp import JID
-from slixmpp.exceptions import IqError, IqTimeout
+from slixmpp import (
+ JID,
+ ClientXMPP,
+ Iq,
+)
+
import logging
log = logging.getLogger(__name__)
+
+if TYPE_CHECKING:
+ from poezio.core import Core
+ from poezio.tabs import Tab
+ from slixmpp.plugins.xep_0004 import Form
+
+
NS_MUC_ADMIN = 'http://jabber.org/protocol/muc#admin'
NS_MUC_OWNER = 'http://jabber.org/protocol/muc#owner'
-def destroy_room(xmpp, room, reason='', altroom=''):
+def destroy_room(
+ xmpp: ClientXMPP,
+ room: str,
+ reason: str = '',
+ altroom: str = ''
+) -> bool:
"""
destroy a room
"""
@@ -42,7 +63,7 @@ def destroy_room(xmpp, room, reason='', altroom=''):
query.append(destroy)
iq.append(query)
- def callback(iq):
+ def callback(iq: Iq) -> None:
if not iq or iq['type'] == 'error':
xmpp.core.information('Unable to destroy room %s' % room, 'Info')
else:
@@ -52,23 +73,13 @@ def destroy_room(xmpp, room, reason='', altroom=''):
return True
-def send_private_message(xmpp, jid, line):
- """
- Send a private message
- """
- jid = safeJID(jid)
- xmpp.send_message(mto=jid, mbody=line, mtype='chat')
-
-
-def send_groupchat_message(xmpp, jid, line):
- """
- Send a message to the groupchat
- """
- jid = safeJID(jid)
- xmpp.send_message(mto=jid, mbody=line, mtype='groupchat')
-
-
-def change_show(xmpp, jid: JID, own_nick: str, show, status):
+def change_show(
+ xmpp: ClientXMPP,
+ jid: JID,
+ own_nick: str,
+ show: str,
+ status: Optional[str]
+) -> None:
"""
Change our 'Show'
"""
@@ -81,7 +92,7 @@ def change_show(xmpp, jid: JID, own_nick: str, show, status):
pres.send()
-def change_subject(xmpp, jid, subject):
+def change_subject(xmpp: ClientXMPP, jid: JID, subject: str) -> None:
"""
Change the room subject
"""
@@ -92,7 +103,13 @@ def change_subject(xmpp, jid, subject):
msg.send()
-def change_nick(core, jid, nick, status=None, show=None):
+def change_nick(
+ core: 'Core',
+ jid: JID,
+ nick: str,
+ status: Optional[str] = None,
+ show: Optional[str] = None
+) -> None:
"""
Change our own nick in a room
"""
@@ -103,14 +120,16 @@ def change_nick(core, jid, nick, status=None, show=None):
presence.send()
-def join_groupchat(core,
- jid,
- nick,
- passwd='',
- status=None,
- show=None,
- seconds=None,
- tab=None):
+def join_groupchat(
+ core: 'Core',
+ jid: JID,
+ nick: str,
+ passwd: str = '',
+ status: Optional[str] = None,
+ show: Optional[str] = None,
+ seconds: Optional[int] = None,
+ tab: Optional['Tab'] = None
+) -> None:
xmpp = core.xmpp
stanza = xmpp.make_presence(
pto='%s/%s' % (jid, nick), pstatus=status, pshow=show)
@@ -119,8 +138,10 @@ def join_groupchat(core,
passelement = ET.Element('password')
passelement.text = passwd
x.append(passelement)
- def on_disco(iq):
- if 'urn:xmpp:mam:2' in iq['disco_info'].get_features() or (tab and tab._text_buffer.last_message):
+
+ def on_disco(iq: Iq) -> None:
+ if ('urn:xmpp:mam:2' in iq['disco_info'].get_features()
+ or (tab and tab._text_buffer.last_message)):
history = ET.Element('{http://jabber.org/protocol/muc}history')
history.attrib['seconds'] = str(0)
x.append(history)
@@ -136,13 +157,15 @@ def join_groupchat(core,
xmpp.plugin['xep_0045'].rooms[jid] = {}
xmpp.plugin['xep_0045'].our_nicks[jid] = to.resource
- try:
- xmpp.plugin['xep_0030'].get_info(jid=jid, callback=on_disco)
- except (IqError, IqTimeout):
- return core.information('Failed to retrieve messages', 'Error')
+ xmpp.plugin['xep_0030'].get_info(jid=jid, callback=on_disco)
-def leave_groupchat(xmpp, jid, own_nick, msg):
+def leave_groupchat(
+ xmpp: ClientXMPP,
+ jid: JID,
+ own_nick: str,
+ msg: str
+) -> None:
"""
Leave the groupchat
"""
@@ -156,7 +179,14 @@ def leave_groupchat(xmpp, jid, own_nick, msg):
exc_info=True)
-def set_user_role(xmpp, jid, nick, reason, role, callback=None):
+def set_user_role(
+ xmpp: ClientXMPP,
+ jid: JID,
+ nick: str,
+ reason: str,
+ role: str,
+ callback: Callable[[Iq], None]
+) -> None:
"""
(try to) Set the role of a MUC user
(role = 'none': eject user)
@@ -172,21 +202,18 @@ def set_user_role(xmpp, jid, nick, reason, role, callback=None):
query.append(item)
iq.append(query)
iq['to'] = jid
- if callback:
- return iq.send(callback=callback)
- try:
- return iq.send()
- except (IqError, IqTimeout) as e:
- return e.iq
-
-
-def set_user_affiliation(xmpp,
- muc_jid,
- affiliation,
- nick=None,
- jid=None,
- reason=None,
- callback=None):
+ iq.send(callback=callback)
+
+
+def set_user_affiliation(
+ xmpp: ClientXMPP,
+ muc_jid: JID,
+ affiliation: str,
+ callback: Callable[[Iq], None],
+ nick: Optional[str] = None,
+ jid: Optional[JID] = None,
+ reason: Optional[str] = None
+) -> None:
"""
(try to) Set the affiliation of a MUC user
"""
@@ -212,18 +239,10 @@ def set_user_affiliation(xmpp,
query.append(item)
iq = xmpp.make_iq_set(query)
iq['to'] = muc_jid
- if callback:
- return iq.send(callback=callback)
- try:
- return xmpp.plugin['xep_0045'].set_affiliation(
- str(muc_jid),
- str(jid) if jid else None, nick, affiliation)
- except:
- log.debug('Error setting the affiliation: %s', exc_info=True)
- return False
+ iq.send(callback=callback)
-def cancel_config(xmpp, room):
+def cancel_config(xmpp: ClientXMPP, room: str) -> None:
query = ET.Element('{http://jabber.org/protocol/muc#owner}query')
x = ET.Element('{jabber:x:data}x', type='cancel')
query.append(x)
@@ -232,7 +251,7 @@ def cancel_config(xmpp, room):
iq.send()
-def configure_room(xmpp, room, form):
+def configure_room(xmpp: ClientXMPP, room: str, form: 'Form') -> None:
if form is None:
return
iq = xmpp.make_iq_set()
diff --git a/poezio/poezio.py b/poezio/poezio.py
index e38871c6..da1bc3e7 100644
--- a/poezio/poezio.py
+++ b/poezio/poezio.py
@@ -104,7 +104,7 @@ def main():
logger.create_logger()
from poezio import roster
- roster.create_roster()
+ roster.roster.reset()
from poezio.core.core import Core
diff --git a/poezio/poezio_shlex.pyi b/poezio/poezio_shlex.pyi
new file mode 100644
index 00000000..affbe12b
--- /dev/null
+++ b/poezio/poezio_shlex.pyi
@@ -0,0 +1,45 @@
+from typing import List, Tuple, Any, TextIO, Union, Optional, Iterable, TypeVar
+import sys
+
+def split(s: str, comments: bool = ..., posix: bool = ...) -> List[str]: ...
+if sys.version_info >= (3, 8):
+ def join(split_command: Iterable[str]) -> str: ...
+def quote(s: str) -> str: ...
+
+_SLT = TypeVar('_SLT', bound=shlex)
+
+class shlex(Iterable[str]):
+ commenters: str
+ wordchars: str
+ whitespace: str
+ escape: str
+ quotes: str
+ escapedquotes: str
+ whitespace_split: bool
+ infile: str
+ instream: TextIO
+ source: str
+ debug: int
+ lineno: int
+ token: str
+ eof: str
+ if sys.version_info >= (3, 6):
+ punctuation_chars: str
+
+ if sys.version_info >= (3, 6):
+ def __init__(self, instream: Union[str, TextIO] = ..., infile: Optional[str] = ...,
+ posix: bool = ..., punctuation_chars: Union[bool, str] = ...) -> None: ...
+ else:
+ def __init__(self, instream: Union[str, TextIO] = ..., infile: Optional[str] = ...,
+ posix: bool = ...) -> None: ...
+ def get_token(self) -> Tuple[int, int, str]: ...
+ def push_token(self, tok: str) -> None: ...
+ def read_token(self) -> str: ...
+ def sourcehook(self, filename: str) -> Tuple[str, TextIO]: ...
+ # TODO argument types
+ def push_source(self, newstream: Any, newfile: Any = ...) -> None: ...
+ def pop_source(self) -> None: ...
+ def error_leader(self, infile: str = ...,
+ lineno: int = ...) -> None: ...
+ def __iter__(self: _SLT) -> _SLT: ...
+ def __next__(self) -> str: ...
diff --git a/poezio/roster.py b/poezio/roster.py
index bedf477b..4a6350a9 100644
--- a/poezio/roster.py
+++ b/poezio/roster.py
@@ -10,6 +10,8 @@ Defines the Roster and RosterGroup classes
import logging
log = logging.getLogger(__name__)
+from typing import List
+
from poezio.config import config
from poezio.contact import Contact
from poezio.roster_sorting import SORTING_METHODS, GROUP_SORTING_METHODS
@@ -18,6 +20,7 @@ from os import path as p
from datetime import datetime
from poezio.common import safeJID
from slixmpp.exceptions import IqError, IqTimeout
+from slixmpp import JID
class Roster:
@@ -29,6 +32,22 @@ class Roster:
DEFAULT_FILTER = (lambda x, y: None, None)
def __init__(self):
+ self.__node = None
+
+ # A tuple(function, *args) function to filter contacts
+ # on search, for example
+ self.contact_filter = self.DEFAULT_FILTER
+ self.groups = {}
+ self.contacts = {}
+ self.length = 0
+ self.connected = 0
+ self.folded_groups = []
+
+ # Used for caching roster infos
+ self.last_built = datetime.now()
+ self.last_modified = datetime.now()
+
+ def reset(self):
"""
node: the RosterSingle from slixmpp
"""
@@ -143,7 +162,7 @@ class Roster:
"""Subscribe to a jid"""
self.__node.subscribe(jid)
- def jids(self):
+ def jids(self) -> List[JID]:
"""List of the contact JIDS"""
l = []
for key in self.__node.keys():
@@ -335,11 +354,6 @@ class RosterGroup:
return len([1 for contact in self.contacts if len(contact)])
-def create_roster():
- "Create the global roster object"
- global roster
- roster = Roster()
-
# Shared roster object
-roster = None
+roster = Roster()
diff --git a/poezio/tabs/basetabs.py b/poezio/tabs/basetabs.py
index a02744aa..d822ea94 100644
--- a/poezio/tabs/basetabs.py
+++ b/poezio/tabs/basetabs.py
@@ -28,6 +28,7 @@ from typing import (
List,
Optional,
Union,
+ Tuple,
TYPE_CHECKING,
)
@@ -52,6 +53,8 @@ from slixmpp import JID, InvalidJID, Message as SMessage
if TYPE_CHECKING:
from _curses import _CursesWindow # pylint: disable=E0611
+ from poezio.size_manager import SizeManager
+ from poezio.core.core import Core
log = logging.getLogger(__name__)
@@ -117,7 +120,7 @@ class Tab:
height = 1
width = 1
- def __init__(self, core):
+ def __init__(self, core: 'Core'):
self.core = core
self.nb = 0
if not hasattr(self, 'name'):
@@ -133,7 +136,7 @@ class Tab:
self.commands = {} # and their own commands
@property
- def size(self) -> int:
+ def size(self) -> 'SizeManager':
return self.core.size
@staticmethod
@@ -196,7 +199,7 @@ class Tab:
self._state = 'normal'
@staticmethod
- def resize(scr: '_CursesWindow'):
+ def initial_resize(scr: '_CursesWindow'):
Tab.height, Tab.width = scr.getmaxyx()
windows.base_wins.TAB_WIN = scr
@@ -327,7 +330,7 @@ class Tab:
else:
return False
- def refresh_tab_win(self):
+ def refresh_tab_win(self) -> None:
if config.get('enable_vertical_tab_list'):
left_tab_win = self.core.left_tab_win
if left_tab_win and not self.size.core_degrade_x:
@@ -371,12 +374,12 @@ class Tab:
"""
pass
- def update_commands(self):
+ def update_commands(self) -> None:
for c in self.plugin_commands:
if c not in self.commands:
self.commands[c] = self.plugin_commands[c]
- def update_keys(self):
+ def update_keys(self) -> None:
for k in self.plugin_keys:
if k not in self.key_func:
self.key_func[k] = self.plugin_keys[k]
@@ -435,7 +438,7 @@ class Tab:
"""
pass
- def on_close(self):
+ def on_close(self) -> None:
"""
Called when the tab is to be closed
"""
@@ -443,7 +446,7 @@ class Tab:
self.input.on_delete()
self.closed = True
- def matching_names(self) -> List[str]:
+ def matching_names(self) -> List[Tuple[int, str]]:
"""
Returns a list of strings that are used to name a tab with the /win
command. For example you could switch to a tab that returns
@@ -532,7 +535,7 @@ class ChatTab(Tab):
desc='Fix the last message with whatever you want.',
shortdesc='Correct the last message.',
completion=self.completion_correct)
- self.chat_state = None
+ self.chat_state = None # type: Optional[str]
self.update_commands()
self.update_keys()
@@ -667,11 +670,11 @@ class ChatTab(Tab):
self._text_buffer.messages = []
self.text_win.rebuild_everything(self._text_buffer)
- def check_send_chat_state(self):
+ def check_send_chat_state(self) -> bool:
"If we should send a chat state"
return True
- def send_chat_state(self, state, always_send=False):
+ def send_chat_state(self, state: str, always_send: bool = False) -> None:
"""
Send an empty chatstate message
"""
@@ -691,9 +694,8 @@ class ChatTab(Tab):
x = ET.Element('{%s}x' % NS_MUC_USER)
msg.append(x)
msg.send()
- return True
- def send_composing_chat_state(self, empty_after):
+ def send_composing_chat_state(self, empty_after: bool) -> None:
"""
Send the "active" or "composing" chatstate, depending
on the the current status of the input
@@ -729,7 +731,7 @@ class ChatTab(Tab):
self.core.add_timed_event(new_event)
self.timed_event_not_paused = new_event
- def cancel_paused_delay(self):
+ def cancel_paused_delay(self) -> None:
"""
Remove that event from the list and set it to None.
Called for example when the input is emptied, or when the message
@@ -741,7 +743,7 @@ class ChatTab(Tab):
self.core.remove_timed_event(self.timed_event_not_paused)
self.timed_event_not_paused = None
- def set_last_sent_message(self, msg, correct=False):
+ def set_last_sent_message(self, msg: SMessage, correct: bool = False) -> None:
"""Ensure last_sent_message is set with the correct attributes"""
if correct:
# XXX: Is the copy needed. Is the object passed here reused
@@ -751,7 +753,7 @@ class ChatTab(Tab):
self.last_sent_message = msg
@command_args_parser.raw
- def command_correct(self, line):
+ def command_correct(self, line: str) -> None:
"""
/correct <fixed message>
"""
@@ -777,7 +779,7 @@ class ChatTab(Tab):
return self.core.status.show in ('xa', 'away') or\
(hasattr(self, 'directed_presence') and not self.directed_presence)
- def move_separator(self):
+ def move_separator(self) -> None:
self.text_win.remove_line_separator()
self.text_win.add_line_separator(self._text_buffer)
self.text_win.refresh()
@@ -786,7 +788,7 @@ class ChatTab(Tab):
def get_conversation_messages(self):
return self._text_buffer.messages
- def check_scrolled(self):
+ def check_scrolled(self) -> None:
if self.text_win.pos != 0:
self.state = 'scrolled'
diff --git a/poezio/tabs/muctab.py b/poezio/tabs/muctab.py
index 63bf026e..b9c8dad7 100644
--- a/poezio/tabs/muctab.py
+++ b/poezio/tabs/muctab.py
@@ -18,9 +18,21 @@ import re
import functools
from copy import copy
from datetime import datetime
-from typing import Dict, Callable, List, Optional, Tuple, Union, Set
+from typing import (
+ cast,
+ Any,
+ Dict,
+ Callable,
+ List,
+ Optional,
+ Tuple,
+ Union,
+ Set,
+ Pattern,
+ TYPE_CHECKING,
+)
-from slixmpp import InvalidJID, JID, Presence
+from slixmpp import InvalidJID, JID, Presence, Iq
from slixmpp.exceptions import IqError, IqTimeout
from poezio.tabs import ChatTab, Tab, SHOW_NAME
@@ -49,6 +61,10 @@ from poezio.ui.types import (
StatusMessage,
)
+if TYPE_CHECKING:
+ from poezio.core.core import Core
+ from slixmpp.plugins.xep_0004 import Form
+
log = logging.getLogger(__name__)
NS_MUC_USER = 'http://jabber.org/protocol/muc#user'
@@ -64,11 +80,11 @@ class MucTab(ChatTab):
"""
message_type = 'groupchat'
plugin_commands = {} # type: Dict[str, Command]
- plugin_keys = {} # type: Dict[str, Callable]
+ plugin_keys = {} # type: Dict[str, Callable[..., Any]]
additional_information = {} # type: Dict[str, Callable[[str], str]]
lagged = False
- def __init__(self, core, jid, nick, password=None):
+ def __init__(self, core: 'Core', jid: JID, nick: str, password: Optional[str] = None) -> None:
ChatTab.__init__(self, core, jid)
self.joined = False
self._state = 'disconnected'
@@ -78,7 +94,7 @@ class MucTab(ChatTab):
self.own_user = None # type: Optional[User]
self.password = password
# buffered presences
- self.presence_buffer = []
+ self.presence_buffer = [] # type: List[Presence]
# userlist
self.users = [] # type: List[User]
# private conversations
@@ -88,13 +104,13 @@ class MucTab(ChatTab):
self.topic = ''
self.topic_from = ''
# Self ping event, so we can cancel it when we leave the room
- self.self_ping_event = None
+ self.self_ping_event = None # type: Optional[timed_events.DelayedEvent]
# UI stuff
self.topic_win = windows.Topic()
self.v_separator = windows.VerticalSeparator()
self.user_win = windows.UserList()
self.info_header = windows.MucInfoWin()
- self.input = windows.MessageInput()
+ self.input = windows.MessageInput() # type: windows.MessageInput
# List of ignored users
self.ignores = [] # type: List[User]
# keys
@@ -106,7 +122,7 @@ class MucTab(ChatTab):
self.resize()
@property
- def general_jid(self):
+ def general_jid(self) -> JID:
return self.jid
def check_send_chat_state(self) -> bool:
@@ -136,21 +152,21 @@ class MucTab(ChatTab):
"""
del MucTab.additional_information[plugin_name]
- def cancel_config(self, form):
+ def cancel_config(self, form: 'Form') -> None:
"""
The user do not want to send their config, send an iq cancel
"""
muc.cancel_config(self.core.xmpp, self.jid.bare)
self.core.close_tab()
- def send_config(self, form):
+ def send_config(self, form: 'Form') -> None:
"""
The user sends their config to the server
"""
muc.configure_room(self.core.xmpp, self.jid.bare, form)
self.core.close_tab()
- def join(self):
+ def join(self) -> None:
"""
Join the room
"""
@@ -167,12 +183,12 @@ class MucTab(ChatTab):
self.core,
self.jid.bare,
self.own_nick,
- self.password,
+ self.password or '',
status=status.message,
show=status.show,
seconds=seconds)
- def leave_room(self, message: str):
+ def leave_room(self, message: str) -> None:
if self.joined:
theme = get_theme()
info_col = dump_tuple(theme.COLOR_INFORMATION_TEXT)
@@ -216,15 +232,17 @@ class MucTab(ChatTab):
muc.leave_groupchat(self.core.xmpp, self.jid.bare, self.own_nick,
message)
- def change_affiliation(self,
- nick_or_jid: Union[str, JID],
- affiliation: str,
- reason=''):
+ def change_affiliation(
+ self,
+ nick_or_jid: Union[str, JID],
+ affiliation: str,
+ reason: str = ''
+ ) -> None:
"""
Change the affiliation of a nick or JID
"""
- def callback(iq):
+ def callback(iq: Iq) -> None:
if iq['type'] == 'error':
self.core.information(
"Could not set affiliation '%s' for '%s'." %
@@ -235,9 +253,10 @@ class MucTab(ChatTab):
valid_affiliations = ('outcast', 'none', 'member', 'admin', 'owner')
if affiliation not in valid_affiliations:
- return self.core.information(
+ self.core.information(
'The affiliation must be one of ' +
', '.join(valid_affiliations), 'Error')
+ return
if nick_or_jid in [user.nick for user in self.users]:
muc.set_user_affiliation(
self.core.xmpp,
@@ -255,12 +274,12 @@ class MucTab(ChatTab):
callback=callback,
reason=reason)
- def change_role(self, nick: str, role: str, reason=''):
+ def change_role(self, nick: str, role: str, reason: str = '') -> None:
"""
Change the role of a nick
"""
- def callback(iq):
+ def callback(iq: Iq) -> None:
if iq['type'] == 'error':
self.core.information(
"Could not set role '%s' for '%s'." % (role, nick),
@@ -269,14 +288,16 @@ class MucTab(ChatTab):
valid_roles = ('none', 'visitor', 'participant', 'moderator')
if not self.joined or role not in valid_roles:
- return self.core.information(
+ self.core.information(
'The role must be one of ' + ', '.join(valid_roles), 'Error')
+ return
try:
target_jid = copy(self.jid)
target_jid.resource = nick
except InvalidJID:
- return self.core.information('Invalid nick', 'Info')
+ self.core.information('Invalid nick', 'Info')
+ return
muc.set_user_role(
self.core.xmpp, self.jid.bare, nick, reason, role, callback=callback)
@@ -313,12 +334,12 @@ class MucTab(ChatTab):
self.add_message(InfoMessage(info), typ=0)
return True
- def change_topic(self, topic: str):
+ def change_topic(self, topic: str) -> None:
"""Change the current topic"""
muc.change_subject(self.core.xmpp, self.jid.bare, topic)
@refresh_wrapper.always
- def show_topic(self):
+ def show_topic(self) -> None:
"""
Print the current topic
"""
@@ -345,7 +366,7 @@ class MucTab(ChatTab):
)
@refresh_wrapper.always
- def recolor(self, random_colors=False):
+ def recolor(self, random_colors: bool = False) -> None:
"""Recolor the current MUC users"""
deterministic = config.get_by_tabname('deterministic_nick_colors',
self.jid.bare)
@@ -410,7 +431,7 @@ class MucTab(ChatTab):
self.text_win.rebuild_everything(self._text_buffer)
return True
- def on_input(self, key, raw):
+ def on_input(self, key: str, raw: bool) -> bool:
if not raw and key in self.key_func:
self.key_func[key]()
return False
@@ -424,17 +445,17 @@ class MucTab(ChatTab):
def get_nick(self) -> str:
if config.get('show_muc_jid'):
- return self.jid.bare
+ return cast(str, self.jid.bare)
bookmark = self.core.bookmarks[self.jid.bare]
if bookmark is not None and bookmark.name:
return bookmark.name
# TODO: send the disco#info identity name here, if it exists.
return self.jid.user
- def get_text_window(self):
+ def get_text_window(self) -> windows.TextWin:
return self.text_win
- def on_lose_focus(self):
+ def on_lose_focus(self) -> None:
if self.joined:
if self.input.text:
self.state = 'nonempty'
@@ -450,7 +471,7 @@ class MucTab(ChatTab):
self.send_chat_state('inactive')
self.check_scrolled()
- def on_gain_focus(self):
+ def on_gain_focus(self) -> None:
self.state = 'current'
if (self.text_win.built_lines and self.text_win.built_lines[-1] is None
and not config.get('show_useless_separator')):
@@ -461,10 +482,8 @@ class MucTab(ChatTab):
self.general_jid) and not self.input.get_text():
self.send_chat_state('active')
- def handle_presence(self, presence):
- """
- Handle MUC presence
- """
+ def handle_presence(self, presence: Presence) -> None:
+ """Handle MUC presence"""
self.reset_lag()
status_codes = set()
for status_code in presence.xml.findall(STATUS_XPATH):
@@ -492,7 +511,7 @@ class MucTab(ChatTab):
self.input.refresh()
self.core.doupdate()
- def process_presence_buffer(self, last_presence, own):
+ def process_presence_buffer(self, last_presence: Presence, own: bool) -> None:
"""
Batch-process all the initial presences
"""
@@ -516,7 +535,7 @@ class MucTab(ChatTab):
self.core.tabs.current_tab.refresh_input()
self.core.doupdate()
- def handle_presence_unjoined(self, presence: Presence, deterministic, own=False) -> None:
+ def handle_presence_unjoined(self, presence: Presence, deterministic: bool, own: bool = False) -> None:
"""
Presence received while we are not in the room (before code=110)
"""
@@ -538,7 +557,7 @@ class MucTab(ChatTab):
status_codes.add(status_code.attrib['code'])
self.own_join(from_nick, new_user, status_codes)
- def own_join(self, from_nick: str, new_user: User, status_codes: Set[str]):
+ def own_join(self, from_nick: str, new_user: User, status_codes: Set[str]) -> None:
"""
Handle the last presence we received, entering the room
"""
@@ -603,7 +622,7 @@ class MucTab(ChatTab):
typ=0)
mam.schedule_tab_open(self)
- def handle_presence_joined(self, presence: Presence, status_codes) -> None:
+ def handle_presence_joined(self, presence: Presence, status_codes: Set[str]) -> None:
"""
Handle new presences when we are already in the room
"""
@@ -630,7 +649,7 @@ class MucTab(ChatTab):
return
elif change_nick:
self.core.events.trigger('muc_nickchange', presence, self)
- self.on_user_nick_change(presence, user, from_nick, from_room)
+ self.on_user_nick_change(presence, user, from_nick)
elif ban:
self.core.events.trigger('muc_ban', presence, self)
self.core.on_user_left_private_conversation(
@@ -656,7 +675,7 @@ class MucTab(ChatTab):
self.on_user_change_status(user, from_nick, from_room, affiliation,
role, show, status)
- def on_non_member_kicked(self):
+ def on_non_member_kicked(self) -> None:
"""We have been kicked because the MUC is members-only"""
self.add_message(
MucOwnLeaveMessage(
@@ -666,7 +685,7 @@ class MucTab(ChatTab):
typ=2)
self.disconnect()
- def on_muc_shutdown(self):
+ def on_muc_shutdown(self) -> None:
"""We have been kicked because the MUC service is shutting down"""
self.add_message(
MucOwnLeaveMessage(
@@ -676,8 +695,8 @@ class MucTab(ChatTab):
typ=2)
self.disconnect()
- def on_user_join(self, from_nick, affiliation, show, status, role, jid,
- color):
+ def on_user_join(self, from_nick: str, affiliation: str, show: str, status: str, role: str, jid: JID,
+ color: str) -> None:
"""
When a new user joins the groupchat
"""
@@ -693,7 +712,7 @@ class MucTab(ChatTab):
self.general_jid):
color = dump_tuple(user.color)
else:
- color = 3
+ color = "3"
theme = get_theme()
info_col = dump_tuple(theme.COLOR_INFORMATION_TEXT)
spec_col = dump_tuple(theme.COLOR_JOIN_CHAR)
@@ -722,7 +741,7 @@ class MucTab(ChatTab):
self.add_message(InfoMessage(msg), typ=2)
self.core.on_user_rejoined_private_conversation(self.jid.bare, from_nick)
- def on_user_nick_change(self, presence, user, from_nick, from_room):
+ def on_user_nick_change(self, presence: Presence, user: User, from_nick: str) -> None:
new_nick = presence.xml.find(
'{%s}x/{%s}item' % (NS_MUC_USER, NS_MUC_USER)).attrib['nick']
old_color = user.color
@@ -746,7 +765,7 @@ class MucTab(ChatTab):
color = dump_tuple(user.color)
old_color = dump_tuple(old_color)
else:
- old_color = color = 3
+ old_color = color = "3"
info_col = dump_tuple(get_theme().COLOR_INFORMATION_TEXT)
self.add_message(
InfoMessage(
@@ -763,7 +782,7 @@ class MucTab(ChatTab):
# rename the private tabs if needed
self.core.rename_private_tabs(self.jid.bare, from_nick, user)
- def on_user_banned(self, presence, user, from_nick):
+ def on_user_banned(self, presence: Presence, user: User, from_nick: str) -> None:
"""
When someone is banned from a muc
"""
@@ -818,7 +837,7 @@ class MucTab(ChatTab):
self.general_jid):
color = dump_tuple(user.color)
else:
- color = 3
+ color = "3"
if by:
kick_msg = ('\x191}%(spec)s \x19%(color)s}'
@@ -846,7 +865,7 @@ class MucTab(ChatTab):
}
self.add_message(cls(kick_msg), typ=2)
- def on_user_kicked(self, presence, user, from_nick):
+ def on_user_kicked(self, presence: Presence, user: User, from_nick: str) -> None:
"""
When someone is kicked from a muc
"""
@@ -899,7 +918,7 @@ class MucTab(ChatTab):
self.general_jid):
color = dump_tuple(user.color)
else:
- color = 3
+ color = "3"
if by:
kick_msg = ('\x191}%(spec)s \x19%(color)s}%(nick)s'
'\x19%(info_col)s} has been kicked by '
@@ -932,7 +951,7 @@ class MucTab(ChatTab):
status: str,
from_nick: str,
from_room: JID,
- server_initiated=False):
+ server_initiated: bool = False) -> None:
"""
When a user leaves a groupchat
"""
@@ -952,7 +971,7 @@ class MucTab(ChatTab):
self.general_jid):
color = dump_tuple(user.color)
else:
- color = 3
+ color = "3"
theme = get_theme()
info_col = dump_tuple(theme.COLOR_INFORMATION_TEXT)
spec_col = dump_tuple(theme.COLOR_QUIT_CHAR)
@@ -992,8 +1011,8 @@ class MucTab(ChatTab):
self.add_message(InfoMessage(leave_msg), typ=2)
self.core.on_user_left_private_conversation(from_room, user, status)
- def on_user_change_status(self, user, from_nick, from_room, affiliation,
- role, show, status):
+ def on_user_change_status(self, user: User, from_nick: str, from_room: str, affiliation: str,
+ role: str, show: str, status: str) -> None:
"""
When a user changes her status
"""
@@ -1004,7 +1023,7 @@ class MucTab(ChatTab):
self.general_jid):
color = dump_tuple(user.color)
else:
- color = 3
+ color = "3"
info_col = dump_tuple(get_theme().COLOR_INFORMATION_TEXT)
if from_nick == self.own_nick:
msg = '\x19%(color)s}You\x19%(info_col)s} changed: ' % {
@@ -1056,7 +1075,7 @@ class MucTab(ChatTab):
user.update(affiliation, show, status, role)
bisect.insort_left(self.users, user)
- def disconnect(self):
+ def disconnect(self) -> None:
"""
Set the state of the room as not joined, so
we can know if we can join it, send messages to it, etc
@@ -1068,25 +1087,25 @@ class MucTab(ChatTab):
self.joined = False
self.disable_self_ping_event()
- def get_single_line_topic(self):
+ def get_single_line_topic(self) -> str:
"""
Return the topic as a single-line string (for the window header)
"""
return self.topic.replace('\n', '|')
- def log_message(self, msg: Message, typ=1):
+ def log_message(self, msg: Message, typ: int = 1) -> None:
"""
Log the messages in the archives, if it needs
to be
"""
if not isinstance(msg, Message):
return
- if not msg.history and self.joined: # don't log the history messages
+ if not msg.history and self.joined and msg.nickname and msg.txt: # don't log the history messages
if not logger.log_message(self.jid.bare, msg.nickname, msg.txt, typ=typ):
self.core.information('Unable to write in the log file',
'Error')
- def get_user_by_name(self, nick):
+ def get_user_by_name(self, nick: str) -> Optional[User]:
"""
Gets the user associated with the given nick, or None if not found
"""
@@ -1095,7 +1114,7 @@ class MucTab(ChatTab):
return user
return None
- def add_message(self, msg: BaseMessage, typ=1) -> None:
+ def add_message(self, msg: BaseMessage, typ: int = 1) -> None:
"""Add a message to the text buffer and set various tab status"""
# reset self-ping interval
if self.self_ping_event:
@@ -1108,17 +1127,18 @@ class MucTab(ChatTab):
if config.get_by_tabname('notify_messages', self.jid.bare) and self.state != 'current':
if msg.nickname != self.own_nick and not msg.history:
self.state = 'message'
- self.do_highlight(msg.txt, msg.nickname, msg.delayed)
+ if msg.txt and msg.nickname:
+ self.do_highlight(msg.txt, msg.nickname, msg.delayed)
def modify_message(self,
- txt,
- old_id,
- new_id,
- time=None,
+ txt: str,
+ old_id: str,
+ new_id: str,
+ time: Optional[datetime] = None,
delayed: bool = False,
- nickname=None,
- user=None,
- jid=None):
+ nickname: Optional[str] = None,
+ user: Optional[User] = None,
+ jid: Optional[JID] = None) -> bool:
highlight = self.message_is_highlight(
txt, nickname, delayed, corrected=True
)
@@ -1136,10 +1156,10 @@ class MucTab(ChatTab):
return highlight
return False
- def matching_names(self):
+ def matching_names(self) -> List[Tuple[int, str]]:
return [(1, self.jid.user), (3, self.jid.full)]
- def enable_self_ping_event(self):
+ def enable_self_ping_event(self) -> None:
delay = config.get_by_tabname(
"self_ping_delay", self.general_jid, default=0)
interval = int(
@@ -1152,12 +1172,12 @@ class MucTab(ChatTab):
interval, self.send_self_ping)
self.core.add_timed_event(self.self_ping_event)
- def disable_self_ping_event(self):
+ def disable_self_ping_event(self) -> None:
if self.self_ping_event is not None:
self.core.remove_timed_event(self.self_ping_event)
self.self_ping_event = None
- def send_self_ping(self):
+ def send_self_ping(self) -> None:
timeout = config.get_by_tabname(
"self_ping_timeout", self.general_jid, default=60)
to = self.jid.bare + "/" + self.own_nick
@@ -1167,7 +1187,7 @@ class MucTab(ChatTab):
timeout_callback=self.on_self_ping_failed,
timeout=timeout)
- def on_self_ping_result(self, iq):
+ def on_self_ping_result(self, iq: Iq) -> None:
if iq["type"] == "error" and iq["error"]["condition"] not in \
("feature-not-implemented", "service-unavailable", "item-not-found"):
self.command_cycle(iq["error"]["text"] or "not in this room")
@@ -1176,13 +1196,13 @@ class MucTab(ChatTab):
self.reset_lag()
self.enable_self_ping_event()
- def search_for_color(self, nick):
+ def search_for_color(self, nick: str) -> str:
"""
Search for the color of a nick in the config file.
Also, look at the colors of its possible aliases if nick_color_aliases
is set.
"""
- color = config.get_by_tabname(nick, 'muc_colors')
+ color = cast(str, config.get_by_tabname(nick, 'muc_colors'))
if color != '':
return color
nick_color_aliases = config.get_by_tabname('nick_color_aliases',
@@ -1192,7 +1212,7 @@ class MucTab(ChatTab):
color = config.get_by_tabname(nick_alias, 'muc_colors')
return color
- def on_self_ping_failed(self, iq):
+ def on_self_ping_failed(self, iq: Any = None) -> None:
if not self.lagged:
self.lagged = True
self._text_buffer.add_message(
@@ -1204,7 +1224,7 @@ class MucTab(ChatTab):
self.core.refresh_window()
self.enable_self_ping_event()
- def reset_lag(self):
+ def reset_lag(self) -> None:
if self.lagged:
self.lagged = False
self.add_message(
@@ -1219,30 +1239,30 @@ class MucTab(ChatTab):
########################## UI ONLY #####################################
@refresh_wrapper.always
- def go_to_next_hl(self):
+ def go_to_next_hl(self) -> None:
"""
Go to the next HL in the room, or the last
"""
self.text_win.next_highlight()
@refresh_wrapper.always
- def go_to_prev_hl(self):
+ def go_to_prev_hl(self) -> None:
"""
Go to the previous HL in the room, or the first
"""
self.text_win.previous_highlight()
@refresh_wrapper.always
- def scroll_user_list_up(self):
+ def scroll_user_list_up(self) -> None:
"Scroll up in the userlist"
self.user_win.scroll_up()
@refresh_wrapper.always
- def scroll_user_list_down(self):
+ def scroll_user_list_down(self) -> None:
"Scroll down in the userlist"
self.user_win.scroll_down()
- def resize(self):
+ def resize(self) -> None:
"""
Resize the whole window. i.e. all its sub-windows
"""
@@ -1278,7 +1298,7 @@ class MucTab(ChatTab):
0)
self.input.resize(1, self.width, self.height - 1, 0)
- def refresh(self):
+ def refresh(self) -> None:
if self.need_resize:
self.resize()
log.debug(' TAB Refresh: %s', self.__class__.__name__)
@@ -1301,7 +1321,7 @@ class MucTab(ChatTab):
self.info_win.refresh()
self.input.refresh()
- def on_info_win_size_changed(self):
+ def on_info_win_size_changed(self) -> None:
if self.core.information_win_size >= self.height - 3:
return
if config.get("hide_user_list"):
@@ -1325,10 +1345,10 @@ class MucTab(ChatTab):
# This maxsize is kinda arbitrary, but most users won’t have that many
# nicknames anyway.
@functools.lru_cache(maxsize=8)
- def build_highlight_regex(self, nickname):
+ def build_highlight_regex(self, nickname: str) -> Pattern:
return re.compile(r"(^|\W)" + re.escape(nickname) + r"(\W|$)", re.I)
- def message_is_highlight(self, txt: str, nickname: str, delayed: bool,
+ def message_is_highlight(self, txt: str, nickname: Optional[str], delayed: bool,
corrected: bool = False) -> bool:
"""Highlight algorithm for MUC tabs"""
# Don't highlight on info message or our own messages
@@ -1358,7 +1378,7 @@ class MucTab(ChatTab):
if highlighted and self.joined and not corrected:
if self.state != 'current':
self.state = 'highlight'
- beep_on = config.get('beep_on').split()
+ beep_on = cast(str, config.get('beep_on')).split()
if 'highlight' in beep_on and 'message' not in beep_on:
if not config.get_by_tabname('disable_beep', self.jid.bare):
curses.beep()
@@ -1368,31 +1388,33 @@ class MucTab(ChatTab):
########################## COMMANDS ####################################
@command_args_parser.quoted(1, 1, [''])
- def command_invite(self, args):
+ def command_invite(self, args: List[str]) -> None:
"""/invite <jid> [reason]"""
if args is None:
- return self.core.command.help('invite')
+ self.core.command.help('invite')
+ return
jid, reason = args
self.core.command.invite('%s %s "%s"' % (jid, self.jid.bare, reason))
@command_args_parser.quoted(1)
- def command_info(self, args):
+ def command_info(self, args: List[str]) -> None:
"""
/info <nick>
"""
if args is None:
- return self.core.command.help('info')
+ self.core.command.help('info')
+ return
nick = args[0]
if not self.print_info(nick):
self.core.information("Unknown user: %s" % nick, "Error")
@command_args_parser.quoted(0)
- def command_configure(self, ignored):
+ def command_configure(self, ignored: Any) -> None:
"""
/configure
"""
- def on_form_received(form):
+ def on_form_received(form: 'Form') -> None:
if not form:
self.core.information(
'Could not retrieve the configuration form', 'Error')
@@ -1402,13 +1424,13 @@ class MucTab(ChatTab):
fixes.get_room_form(self.core.xmpp, self.jid.bare, on_form_received)
@command_args_parser.raw
- def command_cycle(self, msg):
+ def command_cycle(self, msg: str) -> None:
"""/cycle [reason]"""
self.leave_room(msg)
self.join()
@command_args_parser.quoted(0, 1, [''])
- def command_recolor(self, args):
+ def command_recolor(self, args: List[str]) -> None:
"""
/recolor [random]
Re-assigns color to the participants of the room
@@ -1417,7 +1439,7 @@ class MucTab(ChatTab):
self.recolor(random_colors)
@command_args_parser.quoted(2, 2, [''])
- def command_color(self, args):
+ def command_color(self, args: List[str]) -> None:
"""
/color <nick> <color>
Fix a color for a nick.
@@ -1425,24 +1447,28 @@ class MucTab(ChatTab):
User "random" to attribute a random color.
"""
if args is None:
- return self.core.command.help('color')
+ self.core.command.help('color')
+ return
nick = args[0]
color = args[1].lower()
if nick == self.own_nick:
- return self.core.information(
+ self.core.information(
"You cannot change the color of your"
- " own nick.", 'Error')
+ " own nick.", 'Error'
+ )
elif color not in xhtml.colors and color not in ('unset', 'random'):
- return self.core.information("Unknown color: %s" % color, 'Error')
- self.set_nick_color(nick, color)
+ self.core.information("Unknown color: %s" % color, 'Error')
+ else:
+ self.set_nick_color(nick, color)
@command_args_parser.quoted(1)
- def command_version(self, args):
+ def command_version(self, args: List[str]) -> None:
"""
/version <jid or nick>
"""
if args is None:
- return self.core.command.help('version')
+ self.core.command.help('version')
+ return
nick = args[0]
try:
if nick in [user.nick for user in self.users]:
@@ -1451,32 +1477,36 @@ class MucTab(ChatTab):
else:
jid = JID(nick)
except InvalidJID:
- return self.core.information('Invalid jid or nick %r' % nick, 'Error')
+ self.core.information('Invalid jid or nick %r' % nick, 'Error')
+ return
self.core.xmpp.plugin['xep_0092'].get_version(
jid, callback=self.core.handler.on_version_result)
@command_args_parser.quoted(1)
- def command_nick(self, args):
+ def command_nick(self, args: List[str]) -> None:
"""
/nick <nickname>
"""
if args is None:
- return self.core.command.help('nick')
+ self.core.command.help('nick')
+ return
nick = args[0]
if not self.joined:
- return self.core.information('/nick only works in joined rooms',
+ self.core.information('/nick only works in joined rooms',
'Info')
+ return
current_status = self.core.get_status()
try:
target_jid = copy(self.jid)
target_jid.resource = nick
except InvalidJID:
- return self.core.information('Invalid nick', 'Info')
+ self.core.information('Invalid nick', 'Info')
+ return
muc.change_nick(self.core, self.jid.bare, nick, current_status.message,
current_status.show)
@command_args_parser.quoted(0, 1, [''])
- def command_part(self, args):
+ def command_part(self, args: List[str]) -> None:
"""
/part [msg]
"""
@@ -1487,7 +1517,7 @@ class MucTab(ChatTab):
self.core.doupdate()
@command_args_parser.raw
- def command_leave(self, msg):
+ def command_leave(self, msg: str) -> None:
"""
/leave [msg]
"""
@@ -1498,25 +1528,26 @@ class MucTab(ChatTab):
self.core.close_tab(self)
@command_args_parser.raw
- def command_close(self, msg):
+ def command_close(self, msg: str) -> None:
"""
/close [msg]
"""
self.leave_room(msg)
self.core.close_tab(self)
- def on_close(self):
+ def on_close(self) -> None:
super().on_close()
if self.joined:
self.leave_room('')
@command_args_parser.quoted(1, 1)
- def command_query(self, args):
+ def command_query(self, args: List[str]) -> None:
"""
/query <nick> [message]
"""
if args is None:
- return self.core.command.help('query')
+ self.core.command.help('query')
+ return
nick = args[0]
r = None
for user in self.users:
@@ -1524,13 +1555,14 @@ class MucTab(ChatTab):
r = self.core.open_private_window(self.jid.bare, user.nick)
if r and len(args) == 2:
msg = args[1]
- self.core.tabs.current_tab.command_say(
- xhtml.convert_simple_to_full_colors(msg))
+ r.command_say(
+ xhtml.convert_simple_to_full_colors(msg)
+ )
if not r:
self.core.information("Cannot find user: %s" % nick, 'Error')
@command_args_parser.raw
- def command_topic(self, subject):
+ def command_topic(self, subject: str) -> None:
"""
/topic [new topic]
"""
@@ -1540,7 +1572,7 @@ class MucTab(ChatTab):
self.change_topic(subject)
@command_args_parser.quoted(0)
- def command_names(self, args):
+ def command_names(self, args: Any) -> None:
"""
/names
"""
@@ -1578,12 +1610,13 @@ class MucTab(ChatTab):
self.input.refresh()
@command_args_parser.quoted(1, 1)
- def command_kick(self, args):
+ def command_kick(self, args: List[str]) -> None:
"""
/kick <nick> [reason]
"""
if args is None:
- return self.core.command.help('kick')
+ self.core.command.help('kick')
+ return
if len(args) == 2:
reason = args[1]
else:
@@ -1592,36 +1625,38 @@ class MucTab(ChatTab):
self.change_role(nick, 'none', reason)
@command_args_parser.quoted(1, 1)
- def command_ban(self, args):
+ def command_ban(self, args: List[str]) -> None:
"""
/ban <nick> [reason]
"""
if args is None:
- return self.core.command.help('ban')
+ self.core.command.help('ban')
+ return
nick = args[0]
msg = args[1] if len(args) == 2 else ''
self.change_affiliation(nick, 'outcast', msg)
@command_args_parser.quoted(2, 1, [''])
- def command_role(self, args):
+ def command_role(self, args: List[str]) -> None:
"""
/role <nick> <role> [reason]
Changes the role of a user
roles can be: none, visitor, participant, moderator
"""
- def callback(iq):
+ def callback(iq: Iq) -> None:
if iq['type'] == 'error':
self.core.room_error(iq, self.jid.bare)
if args is None:
- return self.core.command.help('role')
+ self.core.command.help('role')
+ return
nick, role, reason = args[0], args[1].lower(), args[2]
self.change_role(nick, role, reason)
@command_args_parser.quoted(0, 2)
- def command_affiliation(self, args) -> None:
+ def command_affiliation(self, args: List[str]) -> None:
"""
/affiliation [<nick or jid> <affiliation>]
Changes the affiliation of a user
@@ -1639,7 +1674,8 @@ class MucTab(ChatTab):
return None
if len(args) != 2:
- return self.core.command.help('affiliation')
+ self.core.command.help('affiliation')
+ return
nick, affiliation = args[0], args[1].lower()
# Set affiliation
@@ -1677,9 +1713,10 @@ class MucTab(ChatTab):
)
return None
+
lines = ['Affiliations for %s' % jid.bare]
for iq in iqs:
- if isinstance(iq, (IqError, IqTimeout)):
+ if isinstance(iq, BaseException):
continue
query = iq.xml.find('{%s}query' % MUC_ADMIN_NS)
@@ -1699,7 +1736,7 @@ class MucTab(ChatTab):
return None
@command_args_parser.raw
- def command_say(self, line, correct=False):
+ def command_say(self, line: str, correct: bool = False) -> None:
"""
/say <message>
Or normal input + enter
@@ -1738,19 +1775,20 @@ class MucTab(ChatTab):
self.chat_state = needed
@command_args_parser.raw
- def command_xhtml(self, msg):
+ def command_xhtml(self, msg: str) -> None:
message = self.generate_xhtml_message(msg)
if message:
message['type'] = 'groupchat'
message.send()
@command_args_parser.quoted(1)
- def command_ignore(self, args):
+ def command_ignore(self, args: List[str]) -> None:
"""
/ignore <nick>
"""
if args is None:
- return self.core.command.help('ignore')
+ self.core.command.help('ignore')
+ return
nick = args[0]
user = self.get_user_by_name(nick)
@@ -1763,12 +1801,13 @@ class MucTab(ChatTab):
self.core.information("%s is now ignored" % nick, 'info')
@command_args_parser.quoted(1)
- def command_unignore(self, args):
+ def command_unignore(self, args: List[str]) -> None:
"""
/unignore <nick>
"""
if args is None:
- return self.core.command.help('unignore')
+ self.core.command.help('unignore')
+ return
nick = args[0]
user = self.get_user_by_name(nick)
@@ -1782,7 +1821,7 @@ class MucTab(ChatTab):
########################## COMPLETIONS #################################
- def completion(self):
+ def completion(self) -> None:
"""
Called when Tab is pressed, complete the nickname in the input
"""
@@ -1795,7 +1834,7 @@ class MucTab(ChatTab):
for user in sorted(self.users, key=COMPARE_USERS_LAST_TALKED, reverse=True):
if user.nick != self.own_nick:
word_list.append(user.nick)
- after = config.get('after_completion') + ' '
+ after = cast(str, config.get('after_completion')) + ' '
input_pos = self.input.pos
if ' ' not in self.input.get_text()[:input_pos] or (
self.input.last_completion and self.input.get_text()
@@ -1813,7 +1852,7 @@ class MucTab(ChatTab):
and not self.input.get_text().startswith('//'))
self.send_composing_chat_state(empty_after)
- def completion_version(self, the_input):
+ def completion_version(self, the_input: windows.MessageInput) -> Completion:
"""Completion for /version"""
userlist = []
for user in sorted(self.users, key=COMPARE_USERS_LAST_TALKED, reverse=True):
@@ -1828,30 +1867,30 @@ class MucTab(ChatTab):
return Completion(the_input.auto_completion, userlist, quotify=False)
- def completion_info(self, the_input):
+ def completion_info(self, the_input: windows.MessageInput) -> Completion:
"""Completion for /info"""
userlist = []
for user in sorted(self.users, key=COMPARE_USERS_LAST_TALKED, reverse=True):
userlist.append(user.nick)
return Completion(the_input.auto_completion, userlist, quotify=False)
- def completion_nick(self, the_input):
+ def completion_nick(self, the_input: windows.MessageInput) -> Completion:
"""Completion for /nick"""
- nicks = [
+ nicks_list = [
os.environ.get('USER'),
- config.get('default_nick'),
+ cast(str, config.get('default_nick')),
self.core.get_bookmark_nickname(self.jid.bare)
]
- nicks = [i for i in nicks if i]
+ nicks = [i for i in nicks_list if i]
return Completion(the_input.auto_completion, nicks, '', quotify=False)
- def completion_recolor(self, the_input):
+ def completion_recolor(self, the_input: windows.MessageInput) -> Optional[Completion]:
if the_input.get_argument_position() == 1:
return Completion(
the_input.new_completion, ['random'], 1, '', quotify=False)
- return True
+ return None
- def completion_color(self, the_input):
+ def completion_color(self, the_input: windows.MessageInput) -> Optional[Completion]:
"""Completion for /color"""
n = the_input.get_argument_position(quoted=True)
if n == 1:
@@ -1867,8 +1906,9 @@ class MucTab(ChatTab):
colors.append('random')
return Completion(
the_input.new_completion, colors, 2, '', quotify=False)
+ return None
- def completion_ignore(self, the_input):
+ def completion_ignore(self, the_input: windows.MessageInput) -> Completion:
"""Completion for /ignore"""
userlist = [user.nick for user in self.users]
if self.own_nick in userlist:
@@ -1876,7 +1916,7 @@ class MucTab(ChatTab):
userlist.sort()
return Completion(the_input.auto_completion, userlist, quotify=False)
- def completion_role(self, the_input):
+ def completion_role(self, the_input: windows.MessageInput) -> Optional[Completion]:
"""Completion for /role"""
n = the_input.get_argument_position(quoted=True)
if n == 1:
@@ -1889,8 +1929,9 @@ class MucTab(ChatTab):
possible_roles = ['none', 'visitor', 'participant', 'moderator']
return Completion(
the_input.new_completion, possible_roles, 2, '', quotify=True)
+ return None
- def completion_affiliation(self, the_input):
+ def completion_affiliation(self, the_input: windows.MessageInput) -> Optional[Completion]:
"""Completion for /affiliation"""
n = the_input.get_argument_position(quoted=True)
if n == 1:
@@ -1913,20 +1954,23 @@ class MucTab(ChatTab):
2,
'',
quotify=True)
+ return None
- def completion_invite(self, the_input):
+ def completion_invite(self, the_input: windows.MessageInput) -> Optional[Completion]:
"""Completion for /invite"""
n = the_input.get_argument_position(quoted=True)
if n == 1:
return Completion(
the_input.new_completion, roster.jids(), 1, quotify=True)
+ return None
- def completion_topic(self, the_input):
+ def completion_topic(self, the_input: windows.MessageInput) -> Optional[Completion]:
if the_input.get_argument_position() == 1:
return Completion(
the_input.auto_completion, [self.topic], '', quotify=False)
+ return None
- def completion_quoted(self, the_input):
+ def completion_quoted(self, the_input: windows.MessageInput) -> Optional[Completion]:
"""Nick completion, but with quotes"""
if the_input.get_argument_position(quoted=True) == 1:
word_list = []
@@ -1936,16 +1980,18 @@ class MucTab(ChatTab):
return Completion(
the_input.new_completion, word_list, 1, quotify=True)
+ return None
- def completion_unignore(self, the_input):
+ def completion_unignore(self, the_input: windows.MessageInput) -> Optional[Completion]:
if the_input.get_argument_position() == 1:
users = [user.nick for user in self.ignores]
return Completion(the_input.auto_completion, users, quotify=False)
+ return None
########################## REGISTER STUFF ##############################
- def register_keys(self):
+ def register_keys(self) -> None:
"Register tab-specific keys"
self.key_func['^I'] = self.completion
self.key_func['M-u'] = self.scroll_user_list_down
@@ -1953,7 +1999,7 @@ class MucTab(ChatTab):
self.key_func['M-n'] = self.go_to_next_hl
self.key_func['M-p'] = self.go_to_prev_hl
- def register_commands(self):
+ def register_commands(self) -> None:
"Register tab-specific commands"
self.register_commands_batch([{
'name': 'ignore',
diff --git a/poezio/tabs/privatetab.py b/poezio/tabs/privatetab.py
index cd2123f3..e4937894 100644
--- a/poezio/tabs/privatetab.py
+++ b/poezio/tabs/privatetab.py
@@ -145,7 +145,7 @@ class PrivateTab(OneToOneTab):
@refresh_wrapper.always
@command_args_parser.raw
- def command_say(self, line, attention=False, correct=False):
+ def command_say(self, line: str, attention: bool = False, correct: bool = False) -> None:
if not self.on:
return
our_jid = JID(self.jid.bare)
diff --git a/poezio/text_buffer.py b/poezio/text_buffer.py
index ff853a67..6ef8e3d4 100644
--- a/poezio/text_buffer.py
+++ b/poezio/text_buffer.py
@@ -32,6 +32,7 @@ from poezio.ui.types import (
if TYPE_CHECKING:
from poezio.windows.text_win import TextWin
+ from poezio.user import User
class CorrectionError(Exception):
@@ -249,7 +250,7 @@ class TextBuffer:
new_id: str,
highlight: bool = False,
time: Optional[datetime] = None,
- user: Optional[str] = None,
+ user: Optional['User'] = None,
jid: Optional[str] = None) -> Message:
"""
Correct a message in a text buffer.
diff --git a/poezio/user.py b/poezio/user.py
index bead0a93..9a14e6b1 100644
--- a/poezio/user.py
+++ b/poezio/user.py
@@ -55,7 +55,7 @@ class User:
else:
self.color = choice(get_theme().LIST_COLOR_NICKNAMES)
- def set_deterministic_color(self):
+ def set_deterministic_color(self) -> None:
theme = get_theme()
if theme.ccg_palette:
# use XEP-0392 CCG
diff --git a/poezio/windows/info_wins.py b/poezio/windows/info_wins.py
index d31130fe..c3975c8c 100644
--- a/poezio/windows/info_wins.py
+++ b/poezio/windows/info_wins.py
@@ -3,6 +3,8 @@ Module defining all the "info wins", ie the bar which is on top of the
info buffer in normal tabs
"""
+from typing import Optional, Dict, TYPE_CHECKING, Any
+
import logging
log = logging.getLogger(__name__)
@@ -13,6 +15,11 @@ from poezio.windows.base_wins import Win
from poezio.ui.funcs import truncate_nick
from poezio.theming import get_theme, to_curses_attr
+if TYPE_CHECKING:
+ from poezio.user import User
+ from poezio.tabs import MucTab
+ from poezio.windows import TextWin
+
class InfoWin(Win):
"""
@@ -260,10 +267,16 @@ class MucInfoWin(InfoWin):
__slots__ = ()
- def __init__(self):
+ def __init__(self) -> None:
InfoWin.__init__(self)
- def refresh(self, room, window=None, user=None, information=None):
+ def refresh(
+ self,
+ room: 'MucTab',
+ window: Optional['TextWin'] = None,
+ user: Optional['User'] = None,
+ information: Optional[Dict[str, Any]] = None
+ ) -> None:
log.debug('Refresh: %s', self.__class__.__name__)
self._win.erase()
self.write_room_name(room)