more type hints

xpra/audio/pulseaudio/pulseaudio_pactl_util.py

@@ -26,7 +26,7 @@ def get_pactl_bin() -> str:
         if WIN32 or OSX:
             pactl_bin = ""
         else:
-            pactl_bin = which("pactl")
+            pactl_bin = which("pactl") or ""
     return pactl_bin
 
 def pactl_output(log_errors=True, *pactl_args) -> Tuple[int,Any,Any]:
@@ -48,9 +48,9 @@ def pactl_output(log_errors=True, *pactl_args) -> Tuple[int,Any,Any]:
         log(f"waiting for `{cmd}` output")
         out, err = process.communicate()
         getChildReaper().add_dead_process(procinfo)
-        code = process.poll()
+        code = process.wait()
         log(f"pactl_output{pactl_args} returned {code}")
-        return  code, out, err
+        return code, out, err
     except Exception as e:
         if log_errors:
             log.error("failed to execute %s: %s", cmd, e)
@@ -67,7 +67,7 @@ def has_pa() -> bool:
     global has_pulseaudio
     if has_pulseaudio is None:
         has_pulseaudio = get_pulse_server_x11_property() or is_pa_installed()
-    return has_pulseaudio
+    return bool(has_pulseaudio)
 
 
 def set_source_mute(device, mute=False) -> bool:
@@ -111,7 +111,7 @@ def get_pulse_cookie_hash() -> bytes:
 def get_pulse_server(may_start_it=True) -> str:
     xp = get_pulse_server_x11_property()
     if xp or not may_start_it:
-        return xp
+        return bytestostr(xp or b"")
     return get_pactl_server()
 
 def get_pulse_id() -> bytes:

xpra/client/gui/keyboard_shortcuts_parser.py

@@ -82,7 +82,7 @@ def parse_shortcuts(strs=(), shortcut_modifiers=(), modifier_names=()) -> Dict[s
     if not strs:
         strs = ["meta+shift+F4:quit"]
     log("parse_shortcuts(%s)", strs)
-    shortcuts = {}
+    shortcuts : Dict[str,List] = {}
     #figure out the default shortcut modifiers
     #accept "," or "+" as delimiter:
     for s in strs:
@@ -105,7 +105,6 @@ def parse_shortcuts(strs=(), shortcut_modifiers=(), modifier_names=()) -> Dict[s
         if action.find("(")>0 and action.endswith(")"):
             try:
                 action, all_args = action[:-1].split("(", 1)
-                args = []
                 for x in all_args.split(","):
                     x = x.strip()
                     if not x:
@@ -122,7 +121,6 @@ def parse_shortcuts(strs=(), shortcut_modifiers=(), modifier_names=()) -> Dict[s
                         args.append(False)
                     else:
                         args.append(int(x))
-                args = tuple(args)
             except Exception as e:
                 log.warn("Warning: failed to parse arguments of shortcut:")
                 log.warn(" '%s': %s", s, e)
@@ -163,7 +161,7 @@ def parse_shortcuts(strs=(), shortcut_modifiers=(), modifier_names=()) -> Dict[s
         #remove any existing action using the same modifiers:
         key_shortcuts = [x for x in key_shortcuts if x[0]!=modifiers]
         if action!="_":
-            key_shortcuts.append((modifiers, action, args))
+            key_shortcuts.append((modifiers, action, tuple(args)))
         shortcuts[keyname] = key_shortcuts
         log("shortcut(%s)=%s", s, csv((modifiers, action, args)))
     log("parse_shortcuts(%s)=%s", strs, shortcuts)

xpra/gtk_common/keymap.py

@@ -29,7 +29,7 @@ def get_gtk_keymap(ignore_keys=(None, "VoidSymbol", "0xffffff")) -> Tuple[Tuple[
     display = Gdk.Display.get_default()
     return do_get_gtk_keymap(display, ignore_keys)
 
-def do_get_gtk_keymap(display, ignore_keys:Tuple[Any]) -> Tuple[Tuple[int,str,int,int,int]]:
+def do_get_gtk_keymap(display, ignore_keys:Tuple[Any]) -> Tuple[Tuple[int,str,int,int,int],...]:
     if not display:
         return ()
     import gi

xpra/platform/win32/win32_NotifyIcon.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 # This file is part of Xpra.
-# Copyright (C) 2011-2022 Antoine Martin <antoine@xpra.org>
+# Copyright (C) 2011-2023 Antoine Martin <antoine@xpra.org>
 # Xpra is released under the terms of the GNU GPL v2, or, at your option, any
 # later version. See the file COPYING for details.
 
@@ -14,6 +14,7 @@
     get_last_error, WinError, WinDLL, HRESULT,  # @UnresolvedImport
     )
 from ctypes.wintypes import HWND, UINT, POINT, HICON, BOOL, CHAR, WCHAR, DWORD, HMODULE, RECT
+from typing import Dict, List, Tuple, Any, Optional, Callable
 
 from xpra.util import typedict, csv, envbool, XPRA_GUID1, XPRA_GUID2, XPRA_GUID3, XPRA_GUID4
 from xpra.os_util import bytestostr
@@ -57,7 +58,7 @@ def GetProductInfo(dwOSMajorVersion=5, dwOSMinorVersion=0, dwSpMajorVersion=0, d
 #MAX_TIP_SIZE = 128
 MAX_TIP_SIZE = 64
 
-def getNOTIFYICONDATAClass(char_type=CHAR, tip_size=MAX_TIP_SIZE):
+def getNOTIFYICONDATAClass(char_type=CHAR, tip_size:int=MAX_TIP_SIZE):
     class _NOTIFYICONDATA(Structure):
         _fields_ = (
             ("cbSize",              DWORD),
@@ -145,7 +146,7 @@ class NOTIFYICONIDENTIFIER(Structure):
 WM_XBUTTONUP    = 0x020C
 WM_XBUTTONDBLCLK= 0x020D
 
-BUTTON_MAP = {
+BUTTON_MAP : Dict[int, List] = {
             win32con.WM_LBUTTONDOWN     : [(1, 1)],
             win32con.WM_LBUTTONUP       : [(1, 0)],
             win32con.WM_MBUTTONDOWN     : [(2, 1)],
@@ -165,11 +166,14 @@ class win32NotifyIcon:
 
     #we register the windows event handler on the class,
     #this allows us to know which hwnd refers to which instance:
-    instances = {}
-
-    def __init__(self, app_id=0, title="",
-                 move_callbacks=None, click_callback=None, exit_callback=None, command_callback=None,
-                 iconPathName=None):
+    instances : Dict[int,Any] = {}
+
+    def __init__(self, app_id:int=0, title:str="",
+                 move_callbacks:Optional[Callable]=None,
+                 click_callback:Optional[Callable]=None,
+                 exit_callback:Optional[Callable]=None,
+                 command_callback:Optional[Callable]=None,
+                 iconPathName:str=""):
         log("win32NotifyIcon: app_id=%i, title=%r", app_id, title)
         self.app_id = app_id
         self.title = title
@@ -179,14 +183,10 @@ def __init__(self, app_id=0, title="",
         self.click_callback = click_callback
         self.exit_callback = exit_callback
         self.command_callback = command_callback
-        self.reset_function = None
-        self.image_cache = {}
+        self.reset_function : Optional[Tuple[Callable,Tuple[Any,...]]] = None
+        self.image_cache : Dict[str,Any] = {}
         # Create the Window.
         if iconPathName:
-            try:
-                iconPathName = iconPathName.decode()
-            except Exception:
-                pass
             self.current_icon = self.LoadImage(iconPathName) or FALLBACK_ICON
         self.create_tray_window()
 
@@ -198,7 +198,7 @@ def create_tray_window(self):
         self.create_window()
         self.register_tray()
 
-    def create_window(self):
+    def create_window(self) -> None:
         style = win32con.WS_OVERLAPPED | win32con.WS_SYSMENU
         window_name = "%s StatusIcon Window" % bytestostr(self.title)
         niwc = get_notifyicon_wnd_class()
@@ -214,7 +214,7 @@ def create_window(self):
         #register callbacks:
         win32NotifyIcon.instances[self.hwnd] = self
 
-    def register_tray(self):
+    def register_tray(self) -> None:
         ni = self.make_nid(NIF_ICON | NIF_MESSAGE | NIF_TIP)
         r = Shell_NotifyIcon(NIM_ADD, byref(ni))
         log("Shell_NotifyIcon ADD=%i", r)
@@ -252,7 +252,7 @@ def make_nid(self, flags):
             nid, title, self.app_id, csv([v for k,v in NIF_FLAGS.items() if k&flags]))
         return nid
 
-    def delete_tray_window(self):
+    def delete_tray_window(self) -> None:
         if not self.hwnd:
             return
         try:
@@ -293,26 +293,26 @@ def get_geometry(self):
         return None
 
 
-    def set_blinking(self, on):
+    def set_blinking(self, on) -> None:
         #implement blinking on win32 using a timer?
         pass
 
-    def set_tooltip(self, tooltip):
+    def set_tooltip(self, tooltip:str) -> None:
         log("set_tooltip(%r)", tooltip)
         self.title = tooltip
         nid = self.make_nid(NIF_ICON | NIF_MESSAGE | NIF_TIP)
         Shell_NotifyIcon(NIM_MODIFY, byref(nid))
 
 
-    def set_icon(self, iconPathName):
+    def set_icon(self, iconPathName:str) -> None:
         log("set_icon(%s)", iconPathName)
         hicon = self.LoadImage(iconPathName) or FALLBACK_ICON
         self.do_set_icon(hicon)
         nid = self.make_nid(NIF_ICON)
         Shell_NotifyIcon(NIM_MODIFY, byref(nid))
-        self.reset_function = (self.set_icon, iconPathName)
+        self.reset_function = self.set_icon, (iconPathName, )
 
-    def do_set_icon(self, hicon, destroy_icon=None):
+    def do_set_icon(self, hicon, destroy_icon=None) -> None:
         log("do_set_icon(%#x)", hicon)
         ci = self.current_icon
         di = self.destroy_icon
@@ -323,7 +323,7 @@ def do_set_icon(self, hicon, destroy_icon=None):
         nid = self.make_nid(NIF_ICON)
         Shell_NotifyIcon(NIM_MODIFY, byref(nid))
 
-    def set_icon_from_data(self, pixels, has_alpha, w, h, rowstride, options=None):
+    def set_icon_from_data(self, pixels, has_alpha:bool, w:int, h:int, rowstride:int, options=None) -> None:
         #this is convoluted but it works..
         log("set_icon_from_data%s", ("%s pixels" % len(pixels), has_alpha, w, h, rowstride, options))
         from PIL import Image   #@UnresolvedImport
@@ -341,13 +341,12 @@ def set_icon_from_data(self, pixels, has_alpha, w, h, rowstride, options=None):
             log("resizing tray icon to %ix%i", icon_w, icon_h)
             img = img.resize((icon_w, icon_h), Image.ANTIALIAS)
             rowstride = w*4
-
         hicon = image_to_ICONINFO(img, TRAY_ALPHA) or FALLBACK_ICON
         self.do_set_icon(hicon, DestroyIcon)
         UpdateWindow(self.hwnd)
-        self.reset_function = (self.set_icon_from_data, pixels, has_alpha, w, h, rowstride)
+        self.reset_function = self.set_icon_from_data, (pixels, has_alpha, w, h, rowstride)
 
-    def LoadImage(self, iconPathName):
+    def LoadImage(self, iconPathName:str):
         if not iconPathName:
             return None
         image = self.image_cache.get(iconPathName)
@@ -356,7 +355,7 @@ def LoadImage(self, iconPathName):
             self.image_cache[iconPathName] = image
         return image
 
-    def doLoadImage(self, iconPathName):
+    def doLoadImage(self, iconPathName:str):
         mingw_prefix = os.environ.get("MINGW_PREFIX")
         if mingw_prefix and iconPathName.find(mingw_prefix)>=0:
             #python can deal with mixed win32 and unix paths,
@@ -386,7 +385,7 @@ def doLoadImage(self, iconPathName):
             return v
 
 
-    def OnTrayRestart(self, hwnd=0, msg=0, wparam=0, lparam=0):
+    def OnTrayRestart(self, hwnd=0, msg=0, wparam=0, lparam=0) -> None:
         try:
             #re-create the tray window:
             self.delete_tray_window()
@@ -407,11 +406,11 @@ def OnCommand(self, hwnd, msg, wparam, lparam):
             cid = wparam & 0xFFFF
             cb(hwnd, cid)
 
-    def OnDestroy(self, hwnd, msg, wparam, lparam):
+    def OnDestroy(self, hwnd, msg, wparam, lparam) -> None:
         log("OnDestroy%s", (hwnd, msg, wparam, lparam))
         self.destroy()
 
-    def OnTaskbarNotify(self, hwnd, msg, wparam, lparam):
+    def OnTaskbarNotify(self, hwnd, msg, wparam, lparam) -> int:
         if lparam==win32con.WM_MOUSEMOVE:
             cb = self.move_callback
             bm = [(hwnd, int(msg), int(wparam), int(lparam))]
@@ -424,11 +423,11 @@ def OnTaskbarNotify(self, hwnd, msg, wparam, lparam):
                 cb(*button_event)
         return 1
 
-    def close(self):
+    def close(self) -> None:
         log("win32NotifyIcon.close()")
         self.destroy()
 
-    def destroy(self):
+    def destroy(self) -> None:
         cb = self.exit_callback
         hwnd = self.hwnd
         log("destroy() hwnd=%#x, exit callback=%s", hwnd, cb)

xpra/queue_scheduler.py

@@ -107,6 +107,6 @@ def stop(self) -> None:
     def stop_main_queue(self) -> None:
         self.main_queue.put(None)
         #empty the main queue:
-        q = Queue()
+        q : Queue = Queue()
         q.put(None)
         self.main_queue = q

xpra/scripts/config.py

@@ -1177,10 +1177,10 @@ def print_bool(k, v, true_str='yes', false_str='no') -> str:
     warn(f"Warning: cannot print value {v!r} for {k!r} as a boolean")
     return ""
 
-def parse_bool_or_int(k, v) -> Union[int,bool]:
+def parse_bool_or_int(k, v) -> Union[int,float,bool]:
     return parse_bool_or_number(int, k, v)
 
-def parse_bool_or_number(numtype:Callable, k:str, v, auto=0) -> Union[int,bool]:
+def parse_bool_or_number(numtype:Callable, k:str, v, auto=0) -> Union[int,float,bool]:
     if isinstance(v, str):
         v = v.lower()
     if v in TRUE_OPTIONS:

xpra/server/background_worker.py

@@ -93,6 +93,7 @@ def get_worker(create:bool=True) -> Optional[Worker_Thread]:
 def add_work_item(item, allow_duplicates:bool=False, daemon:bool=True) -> None:
     w = get_worker(True)
     log("add_work_item(%s, %s, %s) worker=%s", item, allow_duplicates, daemon, w)
+    assert w is not None
     w.add(item, allow_duplicates, daemon)
 
 def stop_worker(force:bool=False) -> None:

xpra/server/menu_provider.py

@@ -154,12 +154,12 @@ def load() -> None:
                 self.clear_cache()
         start_thread(load, "load-menu-data", True)
 
-    def get_menu_data(self, force_reload=False, remove_icons=False, wait=True) -> Optional[Dict[str,Any]]:
+    def get_menu_data(self, force_reload=False, remove_icons=False, wait=True) -> Dict[str,Any]:
         log("get_menu_data%s", (force_reload, remove_icons, wait))
         if not EXPORT_XDG_MENU_DATA:
-            return None
+            return {}
         if OSX:
-            return None
+            return {}
         menu_data = self.menu_data
         if self.load_lock.acquire(wait):  # pylint: disable=consider-using-with
             menu_data = self.menu_data

xpra/server/mixins/display.py

@@ -69,7 +69,7 @@ def threaded_setup(self) -> None:
 
 
     def query_opengl(self) -> Dict[str,Any]:
-        props = {}
+        props : Dict[str,Any] = {}
         if self.opengl.lower()=="noprobe" or self.opengl.lower() in FALSE_OPTIONS:
             gllog("query_opengl() skipped because opengl=%s", self.opengl)
             return props

xpra/server/source/audio.py

@@ -7,7 +7,7 @@
 import os
 from subprocess import Popen
 from shutil import which
-from typing import Dict, Any
+from typing import Dict, Tuple, Any
 
 from xpra.net.compression import Compressed
 from xpra.server.source.stub_source_mixin import StubSourceMixin
@@ -58,8 +58,8 @@ def init_state(self) -> None:
         self.pulseaudio_id = None
         self.pulseaudio_cookie_hash = None
         self.pulseaudio_server = None
-        self.audio_decoders = ()
-        self.audio_encoders = ()
+        self.audio_decoders : Tuple[str,...] = ()
+        self.audio_encoders : Tuple[str,...]= ()
         self.audio_receive = False
         self.audio_send = False
         self.audio_fade_timer = 0
@@ -78,7 +78,7 @@ def stop_new_stream_notifications(self) -> None:
         timers = self.new_stream_timers.copy()
         self.new_stream_timers = {}
         for proc, timer in timers.items():
-            timer = self.new_stream_timers.pop(proc, None)
+            timer = self.new_stream_timers.pop(proc, 0)
             if timer:
                 self.source_remove(timer)
             self.stop_new_stream_notification(proc)
@@ -100,20 +100,20 @@ def parse_client_caps(self, c:typedict) -> None:
         audio = c.dictget("audio")
         if audio:
             audio = typedict(audio)
-            self.pulseaudio_id = audio.strget("pulseaudio.id")
-            self.pulseaudio_cookie_hash = audio.strget("pulseaudio.cookie-hash")
-            self.pulseaudio_server = audio.strget("pulseaudio.server")
-            self.audio_decoders = audio.strtupleget("decoders", [])
-            self.audio_encoders = audio.strtupleget("encoders", [])
+            self.pulseaudio_id = audio.strget("pulseaudio.id", "")
+            self.pulseaudio_cookie_hash = audio.strget("pulseaudio.cookie-hash", "")
+            self.pulseaudio_server = audio.strget("pulseaudio.server", "")
+            self.audio_decoders = audio.strtupleget("decoders", ())
+            self.audio_encoders = audio.strtupleget("encoders", ())
             self.audio_receive = audio.boolget("receive")
             self.audio_send = audio.boolget("send")
         else:
             #pre v4.4:
-            self.pulseaudio_id = c.strget("sound.pulseaudio.id")
-            self.pulseaudio_cookie_hash = c.strget("sound.pulseaudio.cookie-hash")
-            self.pulseaudio_server = c.strget("sound.pulseaudio.server")
-            self.audio_decoders = c.strtupleget("sound.decoders", [])
-            self.audio_encoders = c.strtupleget("sound.encoders", [])
+            self.pulseaudio_id = c.strget("sound.pulseaudio.id", "")
+            self.pulseaudio_cookie_hash = c.strget("sound.pulseaudio.cookie-hash", "")
+            self.pulseaudio_server = c.strget("sound.pulseaudio.server", "")
+            self.audio_decoders = c.strtupleget("sound.decoders", ())
+            self.audio_encoders = c.strtupleget("sound.encoders", ())
             self.audio_receive = c.boolget("sound.receive")
             self.audio_send = c.boolget("sound.send")
         log("pulseaudio id=%s, cookie-hash=%s, server=%s, audio decoders=%s, audio encoders=%s, receive=%s, send=%s",