Commit 098274e8 authored by Daniel Brötzmann's avatar Daniel Brötzmann
Browse files

Add type annotations for AV methods and classes

parent 347a89b4
Pipeline #9036 passed with stages
in 4 minutes and 19 seconds
......@@ -155,7 +155,7 @@ def _next_dtmf(self, events):
else:
self._dtmf_running = False
def start_dtmf(self, key):
def start_dtmf(self, key: str) -> None:
if key == 'star':
event = Farstream.DTMFEvent.STAR
elif key == 'pound':
......@@ -344,13 +344,13 @@ def __init__(self, session, transport=None):
JingleRTPContent.__init__(self, session, 'audio', transport)
self.setup_stream()
def set_mic_volume(self, vol):
def set_mic_volume(self, vol: float) -> None:
"""
vol must be between 0 and 1
"""
self.mic_volume.set_property('volume', vol)
def set_out_volume(self, vol):
def set_out_volume(self, vol: float) -> None:
"""
vol must be between 0 and 1
"""
......
......@@ -27,6 +27,10 @@
# - Tie-breaking
# * timeout
from __future__ import annotations
from typing import Optional
import time
import logging
from enum import Enum, unique
......@@ -38,11 +42,13 @@
from gajim.common import app
from gajim.common import events
from gajim.common.client import Client
from gajim.common.const import KindConstant
from gajim.common.helpers import AdditionalDataDict
from gajim.common.jingle_transport import get_jingle_transport
from gajim.common.jingle_transport import JingleTransportIBB
from gajim.common.jingle_content import get_jingle_content
from gajim.common.jingle_content import JingleContent
from gajim.common.jingle_content import JingleContentSetupException
from gajim.common.jingle_ft import State
from gajim.common.file_props import FilesProp
......@@ -93,14 +99,21 @@ class JingleSession:
negotiated between an initiator and a responder.
"""
def __init__(self, con, weinitiate, jid, iq_id=None, sid=None,
werequest=False):
def __init__(self,
con: Client,
weinitiate: bool,
jid: str,
iq_id: Optional[str] = None,
sid: Optional[str] = None,
werequest: bool = False
) -> None:
"""
con -- connection object,
weinitiate -- boolean, are we the initiator?
jid - jid of the other entity
"""
self.contents = {} # negotiated contents
# negotiated contents
self.contents: dict[tuple[str, str], JingleContent] = {}
self.connection = con # connection to use
# our full jid
self.ourjid = str(self.connection.get_own_jid())
......@@ -188,13 +201,13 @@ def cancel_session(self):
reason.addChild('cancel')
self._session_terminate(reason)
def approve_content(self, media, name=None):
def approve_content(self, media: str, name: Optional[str] = None) -> None:
content = self.get_content(media, name)
if content:
content.accepted = True
self.on_session_state_changed(content)
def reject_content(self, media, name=None):
def reject_content(self, media: str, name: Optional[str] = None) -> None:
content = self.get_content(media, name)
if content:
if self.state == JingleStates.ACTIVE:
......@@ -213,15 +226,23 @@ def end_session(self):
reason.addChild('cancel')
self._session_terminate(reason)
def get_content(self, media=None, name=None):
def get_content(self,
media: Optional[str] = None,
name: Optional[str] = None
) -> Optional[JingleContent]:
if media is None:
return
return None
for content in self.contents.values():
if content.media == media:
if name is None or content.name == name:
return content
return None
def add_content(self, name, content, creator='we'):
def add_content(self,
name: str,
content: JingleContent,
creator: str = 'we'
) -> None:
"""
Add new content to session. If the session is active, this will send
proper stanza to update session
......@@ -268,7 +289,9 @@ def modify_content(self, creator, name, transport=None):
content.sent = False
content.accepted = True
def on_session_state_changed(self, content=None):
def on_session_state_changed(self,
content: Optional[JingleContent] = None
) -> None:
if self.state == JingleStates.ENDED:
# Session not yet started, only one action possible: session-initiate
if self.is_ready() and self.weinitiate:
......@@ -825,7 +848,7 @@ def _session_terminate(self, reason=None):
media=None,
reason=text)
def __content_add(self, content):
def __content_add(self, content: JingleContent) -> None:
# TODO: test
assert self.state != JingleStates.ENDED
stanza, jingle = self.__make_jingle('content-add')
......@@ -834,7 +857,7 @@ def __content_add(self, content):
id_ = self.connection.connection.send(stanza)
self.collect_iq_id(id_)
def __content_accept(self, content):
def __content_accept(self, content: JingleContent) -> None:
# TODO: test
assert self.state != JingleStates.ENDED
stanza, jingle = self.__make_jingle('content-accept')
......@@ -843,7 +866,7 @@ def __content_accept(self, content):
id_ = self.connection.connection.send(stanza)
self.collect_iq_id(id_)
def __content_reject(self, content):
def __content_reject(self, content: JingleContent) -> None:
assert self.state != JingleStates.ENDED
stanza, jingle = self.__make_jingle('content-reject')
self.__append_content(jingle, content)
......
......@@ -14,8 +14,7 @@
from __future__ import annotations
from typing import Any
from typing import List
from typing import Callable
from typing import Optional
import logging
......@@ -29,14 +28,20 @@
from gajim.common import app
from gajim.common import types
from gajim.common import sound
from gajim.common.const import JingleState
from gajim.common.const import KindConstant
from gajim.common.events import JingleEvent
from gajim.common.events import JingleConnectedReceived
from gajim.common.events import JingleDisconnectedReceived
from gajim.common.events import JingleErrorReceived
from gajim.common.events import JingleRequestReceived
from gajim.common.events import Notification
from gajim.common.helpers import AdditionalDataDict
from gajim.common.helpers import play_sound
from gajim.common.i18n import _
from gajim.common.jingle_rtp import JingleAudio
from gajim.common.events import Notification
from gajim.common.helpers import play_sound
from gajim.common import sound
from gajim.common.jingle_session import JingleSession
from .gstreamer import create_gtk_widget
from .builder import get_builder
......@@ -69,9 +74,9 @@ def __init__(self, account: str, contact: types.BareContact) -> None:
# Helper for upgrading voice call to voice + video without
# having to show a new call row
self._incoming_video_event = None
self._incoming_video_event: Optional[JingleRequestReceived] = None
self._jingle: dict(str, JingleObject) = {
self._jingle: dict[str, JingleObject] = {
'audio': JingleObject(
JingleState.NULL,
self._update_audio),
......@@ -88,7 +93,7 @@ def __init__(self, account: str, contact: types.BareContact) -> None:
self._ui.connect_signals(self)
def _on_destroy(self, *args: Any) -> None:
def _on_destroy(self, _widget: Gtk.Widget) -> None:
for jingle_type in ('audio', 'video'):
self._close_jingle_content(jingle_type, shutdown=True)
self._jingle.clear()
......@@ -121,7 +126,8 @@ def _on_call_with_mic_and_cam(self, _button: Gtk.Button) -> None:
self._ui.av_start_box.hide()
def _on_answer_video_clicked(self, button: Gtk.Button) -> None:
self.accept_call(self._incoming_video_event)
assert self._incoming_video_event is not None
self.accept_call(self._incoming_video_event.jingle_session)
self._incoming_video_event = None
button.hide()
......@@ -137,7 +143,7 @@ def _on_end_call_clicked(self, _button: Gtk.Button) -> None:
def _on_video(self, _button: Gtk.Button) -> None:
self._on_jingle_button_toggled(['video'])
def _on_jingle_button_toggled(self, jingle_types: List[str]) -> None:
def _on_jingle_button_toggled(self, jingle_types: list[str]) -> None:
call_resources = self._get_call_resources()
if not call_resources:
return
......@@ -185,40 +191,48 @@ def _on_num_button_press(self,
) -> None:
button_id = Gtk.Buildable.get_name(button)
key = button_id.split('_')[1]
self._get_audio_content().start_dtmf(key)
content = self._get_audio_content()
if content is not None:
content.start_dtmf(key)
def _on_num_button_release(self,
_button: Gtk.Button,
_event: Gdk.EventButton
) -> None:
self._get_audio_content().stop_dtmf()
content = self._get_audio_content()
if content is not None:
content.stop_dtmf()
def _on_mic_volume_changed(self,
_button: Gtk.VolumeButton,
value: float
) -> None:
self._get_audio_content().set_mic_volume(value / 100)
app.settings.set('audio_input_volume', int(value))
content = self._get_audio_content()
if content is not None:
content.set_mic_volume(value / 100)
app.settings.set('audio_input_volume', int(value))
def _on_output_volume_changed(self,
_button: Gtk.VolumeButton,
value: float
) -> None:
self._get_audio_content().set_out_volume(value / 100)
app.settings.set('audio_output_volume', int(value))
content = self._get_audio_content()
if content is not None:
content.set_out_volume(value / 100)
app.settings.set('audio_output_volume', int(value))
def process_event(self, event):
if event.name == 'jingle-request-received':
def process_event(self, event: JingleEvent) -> None:
if isinstance(event, JingleRequestReceived):
self._on_jingle_request(event)
if event.name == 'jingle-connected-received':
if isinstance(event, JingleConnectedReceived):
self._on_jingle_connected(event)
if event.name == 'jingle-disconnected-received':
if isinstance(event, JingleDisconnectedReceived):
self._on_jingle_disconnected(event)
if event.name == 'jingle-error':
if isinstance(event, JingleErrorReceived):
self._on_jingle_error(event)
def _on_jingle_request(self, event):
content_types: list(str) = []
def _on_jingle_request(self, event: JingleRequestReceived) -> None:
content_types: list[str] = []
for item in event.contents:
content_types.append(item.media)
if not any(item in ('audio', 'video') for item in content_types):
......@@ -251,7 +265,7 @@ def _on_jingle_request(self, event):
title=_('Incoming Call'),
text=_('%s is calling') % self._contact.name))
def _on_jingle_connected(self, event):
def _on_jingle_connected(self, event: JingleConnectedReceived) -> None:
session = self._client.get_module('Jingle').get_jingle_session(
event.fjid, event.sid)
......@@ -272,9 +286,11 @@ def _on_jingle_connected(self, event):
session.approve_session()
for content in event.media:
session.approve_content(content)
sound.stop() # dialing/ringing
sound.stop()
def _on_jingle_disconnected(self, event):
def _on_jingle_disconnected(self,
event: JingleDisconnectedReceived
) -> None:
if event.media is None:
self._stop_jingle(sid=event.sid, reason=event.reason)
if event.media == 'audio':
......@@ -289,15 +305,15 @@ def _on_jingle_disconnected(self, event):
JingleState.NULL,
sid=event.sid,
reason=event.reason)
sound.stop() # dialing/ringing
sound.stop()
def _on_jingle_error(self, event):
def _on_jingle_error(self, event: JingleErrorReceived) -> None:
if event.sid == self._jingle['audio'].sid:
self._set_jingle_state(
'audio',
JingleState.ERROR,
reason=event.reason)
sound.stop() # dialing/ringing
sound.stop()
def start_call(self) -> None:
audio_state = self._jingle['audio'].state
......@@ -311,8 +327,8 @@ def start_call(self) -> None:
self._jingle['video'].available)
self._ui.av_cam_button.set_sensitive(False)
def accept_call(self, session):
sound.stop() # dialing/ringing
def accept_call(self, session: JingleSession) -> None:
sound.stop()
if not session:
return
......@@ -336,8 +352,8 @@ def accept_call(self, session):
session.approve_content('video')
@staticmethod
def decline_call(session):
sound.stop() # dialing/ringing
def decline_call(session: JingleSession) -> None:
sound.stop()
if not session:
return
......@@ -473,6 +489,10 @@ def _update_video(self) -> None:
other_gtk_widget = create_gtk_widget()
self_gtk_widget = create_gtk_widget()
if other_gtk_widget is None or self_gtk_widget is None:
log.warning('Could not create GStreamer widgets')
return
sink_other, self._video_widget_other, _name = other_gtk_widget
sink_self, self._video_widget_self, _name = self_gtk_widget
self._ui.incoming_viewport.add(self._video_widget_other)
......@@ -498,8 +518,12 @@ def _update_video(self) -> None:
self._ui.jingle_connection_spinner.stop()
self._ui.jingle_connection_spinner.hide()
def _set_jingle_state(self, jingle_type: str, state: str, sid: str = None,
reason: str = None) -> None:
def _set_jingle_state(self,
jingle_type: str,
state: JingleState,
sid: Optional[str] = None,
reason: Optional[str] = None
) -> None:
jingle = self._jingle[jingle_type]
if state in (
JingleState.CONNECTING,
......@@ -529,7 +553,7 @@ def _set_jingle_state(self, jingle_type: str, state: str, sid: str = None,
def _stop_jingle(self,
sid: Optional[str] = None,
reason: Optional[str] = None
_reason: Optional[str] = None
) -> None:
audio_sid = self._jingle['audio'].sid
video_sid = self._jingle['video'].sid
......@@ -563,18 +587,22 @@ def _get_audio_content(self) -> Optional[JingleAudio]:
self._contact.jid, self._jingle['audio'].sid)
return session.get_content('audio')
def _get_call_resources(self) -> List[types.ResourceContact]:
resource_list = []
def _get_call_resources(self) -> list[types.ResourceContact]:
resource_list: list[types.ResourceContact] = []
for resource_contact in self._contact.iter_resources():
if resource_contact.supports(Namespace.JINGLE_RTP):
resource_list.append(resource_contact)
return resource_list
class JingleObject:
__slots__ = ('sid', 'state', 'available', 'update')
def __init__(self, state, update):
self.sid = None
def __init__(self,
state: JingleState,
update: Callable[..., None]
) -> None:
self.sid: Optional[str] = None
self.state = state
self.available = False
self.available: bool = False
self.update = update
......@@ -54,6 +54,7 @@
from gajim.common.helpers import get_file_path_from_dnd_dropped_uri
from gajim.common.i18n import _
from gajim.common.ged import EventHelper
from gajim.common.events import JingleRequestReceived
from gajim.common.events import Notification
from gajim.common.helpers import AdditionalDataDict
from gajim.common.helpers import get_retraction_text
......@@ -61,6 +62,7 @@
from gajim.common.modules.httpupload import HTTPFileTransfer
from gajim.common.preview_helpers import filename_from_uri
from gajim.common.preview_helpers import guess_simple_file_type
from gajim.common.storage.archive import ConversationRow
from gajim.common.structs import OutgoingMessage
from gajim.gui.conversation.view import ConversationView
......@@ -1175,7 +1177,7 @@ def add_jingle_file_transfer(self, event):
if self._allow_add_message():
self.conversation_view.add_jingle_file_transfer(event)
def add_call_message(self, event):
def add_call_message(self, event: JingleRequestReceived) -> None:
if self._allow_add_message():
self.conversation_view.add_call_message(event=event)
......@@ -1451,7 +1453,7 @@ def fetch_n_lines_history(self,
self.conversation_view.unlock()
def add_messages(self, messages):
def add_messages(self, messages: list[ConversationRow]):
for msg in messages:
if msg.kind in (KindConstant.FILE_TRANSFER_INCOMING,
KindConstant.FILE_TRANSFER_OUTGOING):
......
......@@ -23,6 +23,8 @@
# You should have received a copy of the GNU General Public License
# along with Gajim. If not, see <http://www.gnu.org/licenses/>.
from __future__ import annotations
from typing import ClassVar # pylint: disable=unused-import
from typing import Type # pylint: disable=unused-import
from typing import Optional
......@@ -42,13 +44,17 @@
from gajim.common import app
from gajim.common import helpers
from gajim.common.events import JingleRequestReceived
from gajim.common.i18n import _
from gajim.common.helpers import AdditionalDataDict
from gajim.common.const import AvatarSize
from gajim.common.const import KindConstant
from gajim.common.const import PEPEventType
from gajim.common.jingle_session import JingleSession
from gajim.gui.call_widget import CallWidget
from gajim.gui.controls.base import BaseControl
from gajim.gui.conversation.view import ConversationView
from gajim.gui.const import TARGET_TYPE_URI_LIST
from gajim.gui.const import ControlType
from gajim.gui.dialogs import DialogButton
......@@ -58,7 +64,6 @@
from gajim.command_system.implementation.hosts import ChatCommands
from gajim.command_system.framework import CommandHost # pylint: disable=unused-import
from gajim.gui.controls.base import BaseControl
from ..menus import get_encryption_menu
from ..menus import get_singlechat_menu
......@@ -475,16 +480,25 @@ def _on_start_call(self,
def _process_jingle_av_event(self, event):
self._call_widget.process_event(event)
def _on_accept_call(self, _view, session):
def _on_accept_call(self,
_view: ConversationView,
session: JingleSession
) -> None:
self._call_widget.accept_call(session)
def _on_decline_call(self, _view, session):
def _on_decline_call(self,
_view: ConversationView,
session: JingleSession
) -> None:
self._call_widget.decline_call(session)
def _on_call_ended(self, _call_widget):
def _on_call_ended(self, _call_widget: CallWidget) -> None:
self.conversation_view.update_call_rows()
def _add_incoming_call(self, _call_widget, event):
def _add_incoming_call(self,
_call_widget: CallWidget,
event: JingleRequestReceived
) -> None:
self.add_call_message(event)
def on_location_eventbox_button_release_event(self, _widget, _event):
......
......@@ -12,34 +12,48 @@
# You should have received a copy of the GNU General Public License
# along with Gajim. If not, see <http://www.gnu.org/licenses/>.
from __future__ import annotations
from typing import Optional
from typing import cast
from typing import TYPE_CHECKING
import time
from datetime import datetime
from gi.repository import GdkPixbuf
from gi.repository import Gtk
from gajim.common import app
from gajim.common import types
from gajim.common.const import AvatarSize
from gajim.common.const import KindConstant
from gajim.common.events import JingleRequestReceived
from gajim.common.i18n import _
from gajim.common.jingle_session import JingleSession
from gajim.common.storage.archive import ConversationRow
from .widgets import SimpleLabel
from .base import BaseRow
if TYPE_CHECKING:
from gajim.gui.conversation.view import ConversationView
class CallRow(BaseRow):
def __init__(self, account, contact, event=None, db_message=None):
def __init__(self,
account: str,
contact: types.BareContact,
event: Optional[JingleRequestReceived] = None,
db_message: Optional[ConversationRow] = None
) -> None:
BaseRow.__init__(self, account)
self.type = 'call'
self._client = app.get_client(account)
is_from_db = bool(db_message is not None)
if is_from_db:
if db_message is not None:
timestamp = db_message.time
else:
timestamp = time.time()
......@@ -52,7 +66,7 @@ def __init__(self, account, contact, event=None, db_message=None):
self._session: Optional[JingleSession] = None
if is_from_db:
if db_message is not None:
sid = db_message.additional_data.get_value('gajim', 'sid')
module = self._client.get_module('Jingle')
self._session = module.get_jingle_session(self._contact.jid, sid)
......@@ -87,36 +101,39 @@ def update(self) -> None:
def _on_accept(self, button: Gtk.Button) -> None:
button.set_sensitive(False)
self._decline_button.set_sensitive(False)
view = cast('ConversationView', self.get_parent())
if self._event is not None:
session = self._client.get_module('Jingle').get_jingle_session(
self._event.fjid, self._event.sid)
self.get_parent().accept_call(session)
view.accept_call(session)
else:
self.get_parent().accept_call(self._session)
assert self._session is not None
view.accept_call(self._session)
def _on_decline(self, _button: Gtk.Button) -> None:
view = cast('ConversationView', self.get_parent())
if self._event is not None:
session = self._client.get_module('Jingle').get_jingle_session(
self._event.fjid, self._event.sid)
self.get_parent().decline_call(session)
view.decline_call(session)
else:
self.get_parent().decline_call(self._session)
assert self._session is not None
view.decline_call(self._session)
self._session = None
def _add_history_call_widget(self) -> None:
contact = self._client.get_module('Contacts').get_contact(
str(self._client.get_own_jid().bare))
is_self = True
if self._db_message is not None:
if self._db_message.kind == KindConstant.CALL_INCOMING:
contact = self._contact
is_self = True
else:
contact = self._client.get_module('Contacts').get_contact(
str(self._client.get_own_jid().bare))
is_self = False