libervia-backend: libervia/backend/plugins/plugin_misc

comparison libervia/backend/plugins/plugin_misc_android.py @ 4270:0d7bb4df2343

Reformatted code base using black.

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

comparison

equal deleted inserted replaced

-:64a85ce8be70
+:0d7bb4df2343
 from jnius import autoclass
 from android.broadcast import BroadcastReceiver
 from android import python_act
-Context = autoclass('android.content.Context')
+Context = autoclass("android.content.Context")
-ConnectivityManager = autoclass('android.net.ConnectivityManager')
+ConnectivityManager = autoclass("android.net.ConnectivityManager")
-MediaPlayer = autoclass('android.media.MediaPlayer')
+MediaPlayer = autoclass("android.media.MediaPlayer")
-AudioManager = autoclass('android.media.AudioManager')
+AudioManager = autoclass("android.media.AudioManager")
 # notifications
-AndroidString = autoclass('java.lang.String')
+AndroidString = autoclass("java.lang.String")
-PendingIntent = autoclass('android.app.PendingIntent')
+PendingIntent = autoclass("android.app.PendingIntent")
-Intent = autoclass('android.content.Intent')
+Intent = autoclass("android.content.Intent")
 # DNS
 # regex to find dns server prop with "getprop"
 RE_DNS = re.compile(r"^\[net\.[a-z0-9]+\.dns[0-4]\]: \[(.*)\]$", re.MULTILINE)
-SystemProperties = autoclass('android.os.SystemProperties')
+SystemProperties = autoclass("android.os.SystemProperties")
 #: delay between a pause event and sending the inactive indication to server, in seconds
 #: we don't send the indication immediately because user can be just checking something
 #: quickly on an other app.
 CSI_DELAY = 30
 notification_intent.add_category(Intent.CATEGORY_LAUNCHER)
 if sat_action is not None:
 action_data = AndroidString(json.dumps(sat_action).encode())
 log.debug(f"adding extra {INTENT_EXTRA_ACTION} ==> {action_data}")
 notification_intent = notification_intent.putExtra(
-INTENT_EXTRA_ACTION, action_data)
+INTENT_EXTRA_ACTION, action_data
+)
 # we use PendingIntent.FLAG_UPDATE_CURRENT here, otherwise extra won't be set
 # in the new intent (the old ACTION_MAIN intent will be reused). This differs
 # from plyers original behaviour which set no flag here
 pending_intent = PendingIntent.getActivity(
 # we also set, where suitable, default values to empty string instead of
 # original None, as a string is expected (in plyer the empty string is used
 # in the generic "notify" method).
 sat_action = kwargs.pop("sat_action", None)
 noti = None
-message = kwargs.get('message', '').encode('utf-8')
+message = kwargs.get("message", "").encode("utf-8")
-ticker = kwargs.get('ticker', '').encode('utf-8')
+ticker = kwargs.get("ticker", "").encode("utf-8")
-title = AndroidString(
+title = AndroidString(kwargs.get("title", "").encode("utf-8"))
-kwargs.get('title', '').encode('utf-8')
+icon = kwargs.get("app_icon", "")
-)
-icon = kwargs.get('app_icon', '')
+if kwargs.get("toast", False):
-if kwargs.get('toast', False):
 self._toast(message)
 return
 else:
 noti = self._build_notification(title)
 def __init__(self, android_plugin):
 self.android_plugin = android_plugin
 def buildProtocol(self, addr):
 return FrontendStateProtocol(self.android_plugin)
 class AndroidPlugin(object):
 params = """
 def __init__(self, host):
 log.info(_("plugin Android initialization"))
 log.info(f"using Android API {api_version}")
 self.host = host
-self._csi = host.plugins.get('XEP-0352')
+self._csi = host.plugins.get("XEP-0352")
 self._csi_timer = None
 host.memory.update_params(self.params)
 try:
 os.mkdir(SOCKET_DIR, 0o700)
 except OSError as e:
 reactor.listenUNIX(socket_path, factory)
 else:
 raise e
 # we set a low priority because we want the notification to be sent after all
 # plugins have done their job
-host.trigger.add("message_received", self.message_received_trigger, priority=-1000)
+host.trigger.add(
+"message_received", self.message_received_trigger, priority=-1000
+)
 # profiles autoconnection
 host.bridge.add_method(
 "profile_autoconnect_get",
 ".plugin",
 #     https://github.com/kivy/pyjnius/issues/59)
 self.br = BroadcastReceiver(
 callback=lambda *args, **kwargs: reactor.callFromThread(
 self.on_connectivity_change
 ),
-actions=["android.net.conn.CONNECTIVITY_CHANGE"]
+actions=["android.net.conn.CONNECTIVITY_CHANGE"],
 )
 self.br.start()
 @property
 def state(self):
 notification is sent if:
 - there is a message and it is not a groupchat
 - message is not coming from ourself
 """
-if (mess_data["message"] and mess_data["type"] != C.MESS_TYPE_GROUPCHAT
+if (
-and not mess_data["from"].userhostJID() == client.jid.userhostJID()):
+mess_data["message"]
+and mess_data["type"] != C.MESS_TYPE_GROUPCHAT
+and not mess_data["from"].userhostJID() == client.jid.userhostJID()
+):
 message = next(iter(mess_data["message"].values()))
 try:
 subject = next(iter(mess_data["subject"].values()))
 except StopIteration:
 subject = D_("new message from {contact}").format(
-contact = mess_data['from'])
+contact=mess_data["from"]
+)
 notification = Notification()
 notification._notify(
 title=subject,
 message=message,
 ringer_mode = self.am.getRingerMode()
 vibrate_mode = ringer_mode == AudioManager.RINGER_MODE_VIBRATE
 ring_setting = self.host.memory.param_get_a(
-PARAM_RING_NAME,
+PARAM_RING_NAME, PARAM_RING_CATEGORY, profile_key=client.profile
-PARAM_RING_CATEGORY,
+)
-profile_key=client.profile
-)
+if ring_setting != "never" and ringer_mode == AudioManager.RINGER_MODE_NORMAL:
-if ring_setting != 'never' and ringer_mode == AudioManager.RINGER_MODE_NORMAL:
 self.notif_player.start()
 vibration_setting = self.host.memory.param_get_a(
-PARAM_VIBRATE_NAME,
+PARAM_VIBRATE_NAME, PARAM_VIBRATE_CATEGORY, profile_key=client.profile
-PARAM_VIBRATE_CATEGORY,
+)
-profile_key=client.profile
+if (
-)
+vibration_setting == "always"
-if (vibration_setting == 'always'
+or vibration_setting == "vibrate"
-or vibration_setting == 'vibrate' and vibrate_mode):
+and vibrate_mode
-try:
+):
-vibrator.vibrate()
+try:
-except Exception as e:
+vibrator.vibrate()
-log.warning("Can't use vibrator: {e}".format(e=e))
+except Exception as e:
+log.warning("Can't use vibrator: {e}".format(e=e))
 return mess_data
 def message_received_trigger(self, client, message_elt, post_treat):
 if not self.cagou_active:
 # we only send notification is the frontend is not displayed
 def _profile_autoconnect_get(self):
 return defer.ensureDeferred(self.profile_autoconnect_get())
 async def _get_profiles_autoconnect(self):
 autoconnect_dict = await self.host.memory.storage.get_ind_param_values(
-category='Connection', name='autoconnect_backend',
+category="Connection",
+name="autoconnect_backend",
 )
 return [p for p, v in autoconnect_dict.items() if C.bool(v)]
 async def profile_autoconnect_get(self):
 """Return profile to connect automatically by frontend, if any"""
 if not profiles_autoconnect:
 return None
 if len(profiles_autoconnect) > 1:
 log.warning(
 f"More that one profiles with backend autoconnection set found, picking "
-f"up first one (full list: {profiles_autoconnect!r})")
+f"up first one (full list: {profiles_autoconnect!r})"
+)
 return profiles_autoconnect[0]
 # CSI
 def _set_inactive(self):
 elif net_type == NET_TYPE_WIFI:
 log.info("WIFI activated")
 elif net_type == NET_TYPE_MOBILE:
 log.info("mobile data activated")
 else:
-log.info("network activated (type={net_type_android})"
+log.info(
-.format(net_type_android=net_type_android))
+"network activated (type={net_type_android})".format(
+net_type_android=net_type_android
+)
+)
 else:
-log.debug("_check_connectivity called without network change ({net_type})"
+log.debug(
-.format(net_type = net_type))
+"_check_connectivity called without network change ({net_type})".format(
+net_type=net_type
+)
+)
 # we always call _handle_network_change even if there is not connectivity change
 # to be sure to reconnect when necessary
 await self._handle_network_change(net_type)
 def on_connectivity_change(self):
 log.debug("on_connectivity_change called")
 d = defer.ensureDeferred(self._check_connectivity())
 d.addErrback(self.host.log_errback)
 # updating Twisted's resolver accordingly
 dns_servers = await self.get_dns_servers()
 log.info(
 "Patching Twisted to use Android DNS resolver ({dns_servers})".format(
-dns_servers=', '.join([s[0] for s in dns_servers]))
+dns_servers=", ".join([s[0] for s in dns_servers])
+)
 )
 dns_client.theResolver = dns_client.createResolver(servers=dns_servers)
 async def get_dns_servers(self):
 servers = []
 if api_version < 26:
 # thanks to A-IV at https://stackoverflow.com/a/11362271 for the way to go
 log.debug("Old API, using SystemProperties to find DNS")
 for idx in range(1, 5):
-addr = SystemProperties.get(f'net.dns{idx}')
+addr = SystemProperties.get(f"net.dns{idx}")
 if abstract.isIPAddress(addr):
 servers.append((addr, 53))
 else:
 log.debug(f"API {api_version} >= 26, using getprop to find DNS")
 # use of getprop inspired by various solutions at
 # https://stackoverflow.com/q/3070144
 # it's the most simple option, and it fit wells with async_process
-getprop_paths = which('getprop')
+getprop_paths = which("getprop")
 if getprop_paths:
 try:
 getprop_path = getprop_paths[0]
 props = await async_process.run(getprop_path)
-servers = [(ip, 53) for ip in RE_DNS.findall(props.decode())
+servers = [
-if abstract.isIPAddress(ip)]
+(ip, 53)
+for ip in RE_DNS.findall(props.decode())
+if abstract.isIPAddress(ip)
+]
 except Exception as e:
-log.warning(f"Can't use \"getprop\" to find DNS server: {e}")
+log.warning(f'Can\'t use "getprop" to find DNS server: {e}')
 if not servers:
 # FIXME: Cloudflare's 1.1.1.1 seems to have a better privacy policy, to be
 #   checked.
 log.warning(
 "no server found, we have to use factory Google DNS, this is not ideal "
 "for privacy"
 )
-servers.append(('8.8.8.8', 53), ('8.8.4.4', 53))
+servers.append(("8.8.8.8", 53), ("8.8.4.4", 53))
 return servers

Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_misc_android.py @ 4270:0d7bb4df2343