Skip to content
Snippets Groups Projects
Commit fba9f814 authored by Daniel Brötzmann's avatar Daniel Brötzmann
Browse files

chore: UserLocation: Add type annotations

parent 124f7ab8
No related branches found
No related tags found
No related merge requests found
......@@ -20,10 +20,15 @@
from nbxmpp.namespaces import Namespace
from nbxmpp.protocol import JID
from nbxmpp.protocol import Message
from nbxmpp.structs import LocationData
from nbxmpp.structs import MessageProperties
from gajim.common import app
from gajim.common import ged
from gajim.common import types
from gajim.common.events import SignedIn
from gajim.common.events import LocationChanged
from gajim.common.modules.base import BaseModule
from gajim.common.modules.util import event_node
from gajim.common.modules.util import store_publish
......@@ -38,12 +43,12 @@ class UserLocation(BaseModule):
'set_location',
]
def __init__(self, con) -> None:
def __init__(self, con: types.Client) -> None:
BaseModule.__init__(self, con)
self._register_pubsub_handler(self._location_received)
self._current_location: Optional[LocationData] = None
self._contact_locations: dict[JID, LocationData] = {}
self._contact_locations: dict[JID, Optional[LocationData]] = {}
def get_current_location(self) -> Optional[LocationData]:
return self._current_location
......@@ -52,7 +57,12 @@ def get_contact_location(self, jid: JID) -> Optional[LocationData]:
return self._contact_locations.get(jid)
@event_node(Namespace.LOCATION)
def _location_received(self, _con, _stanza, properties):
def _location_received(self,
_con: types.xmppClient,
_stanza: Message,
properties: MessageProperties
) -> None:
assert properties.pubsub_event is not None
if properties.pubsub_event.retracted:
return
......@@ -60,13 +70,17 @@ def _location_received(self, _con, _stanza, properties):
if properties.is_self_message:
self._current_location = data
assert properties.jid is not None
self._contact_locations[properties.jid] = data
contact = self._get_contact(properties.jid)
contact.notify('location-update', data)
@store_publish
def set_location(self, location, force=False):
def set_location(self,
location: Optional[LocationData],
force: bool = False
) -> None:
if not self._con.get_module('PEP').supported:
return
......@@ -81,7 +95,7 @@ def set_location(self, location, force=False):
self._log.info('Send %s', location)
self._nbxmpp('Location').set_location(location)
def set_enabled(self, enable):
def set_enabled(self, enable: bool) -> None:
if enable:
self.register_events([
('location-changed', ged.CORE, self._on_location_changed),
......@@ -96,10 +110,10 @@ def _publish_current_location(self):
self.set_location(LocationListener.get().current_location)
@event_filter(['account'])
def _on_signed_in(self, _event):
def _on_signed_in(self, _event: SignedIn) -> None:
self._publish_current_location()
def _on_location_changed(self, event):
def _on_location_changed(self, event: LocationChanged) -> None:
if self._current_location == event.info:
return
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment