libervia-backend: libervia/frontends/quick_frontend/quick

comparison libervia/frontends/quick_frontend/quick_app.py @ 4270:0d7bb4df2343

Reformatted code base using black.

author	Goffi <goffi@goffi.org>
date	Wed, 19 Jun 2024 18:44:57 +0200
parents	d01b8d002619
children

comparison

equal deleted inserted replaced

-:64a85ce8be70
+:0d7bb4df2343
 # TODO: handle waiting XMLUI requests: getWaitingConf doesn't exist anymore
 #       and a way to keep some XMLUI request between sessions is expected in backend
 host = None
 bridge = None
-cache_keys_to_get = ['avatar', 'nicknames']
+cache_keys_to_get = ["avatar", "nicknames"]
 def __init__(self, profile):
 self.profile = profile
 self.connected = False
 self.whoami = None
 log.error("Couldn't get features: {}".format(failure))
 self._plug_profile_get_features_cb({})
 def _plug_profile_get_features_cb(self, features):
 self.host.features = features
-self.host.bridge.entities_data_get([], ProfileManager.cache_keys_to_get,
+self.host.bridge.entities_data_get(
-profile=self.profile,
+[],
-callback=self._plug_profile_got_cached_values,
+ProfileManager.cache_keys_to_get,
-errback=self._plug_profile_failed_cached_values)
+profile=self.profile,
+callback=self._plug_profile_got_cached_values,
+errback=self._plug_profile_failed_cached_values,
+)
 def _plug_profile_failed_cached_values(self, failure):
 log.error("Couldn't get cached values: {}".format(failure))
 self._plug_profile_got_cached_values({})
 for entity_s, data in cached_values.items():
 for key, value in data.items():
 self.host.entity_data_updated_handler(entity_s, key, value, self.profile)
 if not self.connected:
-self.host.set_presence_status(C.PRESENCE_UNAVAILABLE, "", profile=self.profile)
+self.host.set_presence_status(
+C.PRESENCE_UNAVAILABLE, "", profile=self.profile
+)
 else:
 contact_list.fill()
 self.host.set_presence_status(profile=self.profile)
 #: if False, frontend must call resync itself when suitable (e.g. widget is being
 #: visible)
 AUTO_RESYNC = True
 def __init__(
-self, bridge_factory, xmlui, check_options=None, connect_bridge=True,
+self,
-async_bridge_factory=None
+bridge_factory,
+xmlui,
+check_options=None,
+connect_bridge=True,
+async_bridge_factory=None,
 ):
 """Create a frontend application
 @param bridge_factory: method to use to create the bridge
 @param xmlui: xmlui module
 self.profiles = ProfilesManager()
 # profiles currently being plugged, used to (un)lock contact list updates
 self._plugs_in_progress = set()
 self.ready_profiles = set()  # profiles which are connected and ready
 self.signals_cache = {}  # used to keep signal received between start of
 # plug_profile and when the profile is actualy ready
 self.contact_lists = quick_contact_list.QuickContactListHandler(self)
 self.widgets = quick_widgets.QuickWidgetsManager(self)
 if check_options is not None:
 self.options = check_options()
 else:
 # see selected_widget setter and getter
 self._selected_widget = None
 # listeners are callable watching events
 self._listeners = {}  # key: listener type ("avatar", "selected", etc),
 # value: list of callbacks
 # cf. [register_action_handler]
 self._action_handlers: dict[str, Callable[[dict, str, int, str], None]] = {}
 # triggers
 print((_("Error while initialising bridge: {}".format(failure))))
 def on_backend_ready(self):
 log.info("backend is ready")
 self.bridge.namespaces_get(
-callback=self._namespaces_get_cb, errback=self._namespaces_get_eb)
+callback=self._namespaces_get_cb, errback=self._namespaces_get_eb
+)
 # we cache available encryption plugins, as we'll use them on each
 # new chat widget
 self.bridge.encryption_plugins_get(
 callback=self._encryption_plugins_get_cb,
-errback=self._encryption_plugins_get_eb)
+errback=self._encryption_plugins_get_eb,
+)
 @property
 def current_profile(self):
 """Profile that a user would expect to use"""
 try:
 self._listeners[type_].pop(callback)
 except KeyError:
 if not ignore_missing:
 log.error(
 f"Trying to remove an inexisting listener (type = {type_}): "
-f"{callback}")
+f"{callback}"
+)
 def call_listeners(self, type_, *args, **kwargs):
 """Call the methods which listen type_ event. If a profiles filter has
 been register with a listener and profile argument is not None, the
 listener will be called only if profile is in the profiles filter list.
 )
 def contact_new_handler(self, jid_s, attributes, groups, profile):
 entity = jid.JID(jid_s)
 groups = list(groups)
-self.contact_lists[profile].set_contact(entity, groups, attributes, in_roster=True)
+self.contact_lists[profile].set_contact(
+entity, groups, attributes, in_roster=True
+)
 def message_new_handler(
-self, uid, timestamp, from_jid_s, to_jid_s, msg, subject, type_, extra_s,
+self, uid, timestamp, from_jid_s, to_jid_s, msg, subject, type_, extra_s, profile
-profile):
+):
 from_jid = jid.JID(from_jid_s)
 to_jid = jid.JID(to_jid_s)
 extra = data_format.deserialise(extra_s)
 if not self.trigger.point(
-"messageNewTrigger", uid, timestamp, from_jid, to_jid, msg, subject, type_,
+"messageNewTrigger",
-extra, profile=profile,):
+uid,
+timestamp,
+from_jid,
+to_jid,
+msg,
+subject,
+type_,
+extra,
+profile=profile,
+):
 return
 from_me = from_jid.bare == self.profiles[profile].whoami.bare
 mess_to_jid = to_jid if from_me else from_jid
 target = mess_to_jid.bare
 target = target
 # we want to be sure to have at least one QuickChat instance
 self.widgets.get_or_create_widget(
 quick_chat.QuickChat,
 target,
-type_ = C.CHAT_GROUP if is_room else C.CHAT_ONE2ONE,
+type_=C.CHAT_GROUP if is_room else C.CHAT_ONE2ONE,
-on_new_widget = None,
+on_new_widget=None,
-profile = profile,
+profile=profile,
 )
 if (
 not from_jid in contact_list
 and from_jid.bare != self.profiles[profile].whoami.bare
 )
 def message_encryption_started_handler(self, destinee_jid_s, plugin_data, profile):
 destinee_jid = jid.JID(destinee_jid_s)
 plugin_data = data_format.deserialise(plugin_data)
-for widget in self.widgets.get_widgets(quick_chat.QuickChat,
+for widget in self.widgets.get_widgets(
-target=destinee_jid.bare,
+quick_chat.QuickChat, target=destinee_jid.bare, profiles=(profile,)
-profiles=(profile,)):
+):
 widget.message_encryption_started(plugin_data)
 def message_encryption_stopped_handler(self, destinee_jid_s, plugin_data, profile):
 destinee_jid = jid.JID(destinee_jid_s)
-for widget in self.widgets.get_widgets(quick_chat.QuickChat,
+for widget in self.widgets.get_widgets(
-target=destinee_jid.bare,
+quick_chat.QuickChat, target=destinee_jid.bare, profiles=(profile,)
-profiles=(profile,)):
+):
 widget.message_encryption_stopped(plugin_data)
 def message_state_handler(self, uid, status, profile):
 for widget in self.widgets.get_widgets(quick_chat.QuickChat, profiles=(profile,)):
 widget.on_message_state(uid, status, profile)
-def message_send(self, to_jid, message, subject=None, mess_type="auto", extra=None, callback=None, errback=None, profile_key=C.PROF_KEY_NONE):
+def message_send(
-if not subject and not extra and (not message or message == {'': ''}):
+self,
+to_jid,
+message,
+subject=None,
+mess_type="auto",
+extra=None,
+callback=None,
+errback=None,
+profile_key=C.PROF_KEY_NONE,
+):
+if not subject and not extra and (not message or message == {"": ""}):
 log.debug("Not sending empty message")
 return
 if subject is None:
 subject = {}
 extra = {}
 if callback is None:
 callback = (
 lambda __=None: None
 )  # FIXME: optional argument is here because pyjamas doesn't support callback
 #        without arg with json proxy
 if errback is None:
 errback = lambda failure: self.show_dialog(
 message=failure.message, title=failure.fullname, type="error"
 )
-if not self.trigger.point("messageSendTrigger", to_jid, message, subject, mess_type, extra, callback, errback, profile_key=profile_key):
+if not self.trigger.point(
+"messageSendTrigger",
+to_jid,
+message,
+subject,
+mess_type,
+extra,
+callback,
+errback,
+profile_key=profile_key,
+):
 return
 self.bridge.message_send(
 str(to_jid),
 message,
 return
 self.call_listeners("presence", entity, show, priority, statuses, profile=profile)
 def muc_room_joined_handler(
-self, room_jid_s, occupants, user_nick, subject, statuses, profile):
+self, room_jid_s, occupants, user_nick, subject, statuses, profile
+):
 """Called when a MUC room is joined"""
 log.debug(
 "Room [{room_jid}] joined by {profile}, users presents:{users}".format(
 room_jid=room_jid_s, profile=profile, users=list(occupants.keys())
 )
 )
 room_jid = jid.JID(room_jid_s)
 chat_widget = self.widgets.get_widget(quick_chat.QuickChat, room_jid, profile)
 if chat_widget:
 self.widgets.delete_widget(
-chat_widget, all_instances=True, explicit_close=True)
+chat_widget, all_instances=True, explicit_close=True
+)
 self.contact_lists[profile].remove_contact(room_jid)
 def muc_room_user_changed_nick_handler(self, room_jid_s, old_nick, new_nick, profile):
 """Called when an user joined a MUC room"""
 room_jid = jid.JID(room_jid_s)
 @param from_jid_s (unicode): JID of a contact or C.ENTITY_ALL
 @param state (unicode): new state
 @param profile (unicode): current profile
 """
 from_jid = jid.JID(from_jid_s)
-for widget in self.widgets.get_widgets(quick_chat.QuickChat, target=from_jid.bare,
+for widget in self.widgets.get_widgets(
-profiles=(profile,)):
+quick_chat.QuickChat, target=from_jid.bare, profiles=(profile,)
+):
 widget.on_chat_state(from_jid, state, profile)
-def notify(self, type_, entity=None, message=None, subject=None, callback=None,
+def notify(
-cb_args=None, widget=None, profile=C.PROF_KEY_NONE):
+self,
+type_,
+entity=None,
+message=None,
+subject=None,
+callback=None,
+cb_args=None,
+widget=None,
+profile=C.PROF_KEY_NONE,
+):
 """Trigger an event notification
 @param type_(unicode): notifation kind,
 one of C.NOTIFY_* constant or any custom type specific to frontend
 @param entity(jid.JID, None): entity involved in the notification
 type_notifs.append(notif_data)
 self._notifications[self._notif_id] = notif_data
 self._notif_id += 1
 self.call_listeners("notification", entity, notif_data, profile=profile)
-def get_notifs(self, entity=None, type_=None, exact_jid=None, profile=C.PROF_KEY_NONE):
+def get_notifs(
+self, entity=None, type_=None, exact_jid=None, profile=C.PROF_KEY_NONE
+):
 """return notifications for given entity
 @param entity(jid.JID, None, C.ENTITY_ALL): jid of the entity to check
 bare jid to get all notifications, full jid to filter on resource
 None to get general notifications
 )
 def _debug_handler(self, action, parameters, profile):
 if action == "widgets_dump":
 from pprint import pformat
 log.info("Widgets dump:\n{data}".format(data=pformat(self.widgets._widgets)))
 else:
 log.warning("Unknown debug action: {action}".format(action=action))
 def show_dialog(self, message, title, type="info", answer_cb=None, answer_data=None):
 """Show a dialog to user
 Frontends must override this method
 assert isinstance(value, dict) or value is None
 self.contact_lists[profile].set_cache(entity, "avatar", value)
 self.call_listeners("avatar", entity, value, profile=profile)
 def register_action_handler(
-self,
+self, action_type: str, handler: Callable[[dict, str, int, str], None]
-action_type: str,
-handler: Callable[[dict, str, int, str], None]
 ) -> None:
 """Register a handler for action type.
 If an action of this type is received, the handler will be used, otherwise,
 generic ``action_manager`` is used
 self._action_handlers[action_type] = handler
 def action_manager(
 self,
 action_data: dict,
-callback: Callable|None = None,
+callback: Callable | None = None,
-ui_show_cb: Callable|None = None,
+ui_show_cb: Callable | None = None,
 user_action: bool = True,
-action_id: str|None = None,
+action_id: str | None = None,
-progress_cb: Callable|None = None,
+progress_cb: Callable | None = None,
-progress_eb: Callable|None = None,
+progress_eb: Callable | None = None,
-profile: str = C.PROF_KEY_NONE
+profile: str = C.PROF_KEY_NONE,
 ) -> None:
 """Handle backend action
 @param action_data: action dict as sent by action_launch or returned by an
 UI action
 data = dict()
 action_cb = lambda data: self._action_cb(
 data_format.deserialise(data), callback, callback_id, profile
 )
 self.bridge.action_launch(
-callback_id, data_format.serialise(data), profile, callback=action_cb,
+callback_id,
-errback=self.dialog_failure
+data_format.serialise(data),
+profile,
+callback=action_cb,
+errback=self.dialog_failure,
 )
 def launch_menu(
 self,
 menu_type,

Mercurial > libervia-backend

comparison libervia/frontends/quick_frontend/quick_app.py @ 4270:0d7bb4df2343