more mypy type hints

xpra/audio/audio_pipeline.py

@@ -43,7 +43,7 @@ def __init__(self, codec):
         self.start_time = 0
         self.buffer_count = 0
         self.byte_count = 0
-        self.emit_info_timer = None
+        self.emit_info_timer = 0
         self.info = {
                      "codec"        : self.codec,
                      "state"        : self.state,

xpra/client/base/client_base.py

@@ -107,7 +107,7 @@ def defaults_init(self):
         self.exit_on_signal = False
         self.display_desc = {}
         self.progress_process = None
-        self.progress_timer = None
+        self.progress_timer = 0
         #connection attributes:
         self.hello_extra = {}
         self.compression_level = 0
@@ -188,14 +188,14 @@ def show_progress(self, pct, text=""):
             #kill it if it's still running after 2 seconds
             self.cancel_progress_timer()
             def stop_progress():
-                self.progress_timer = None
+                self.progress_timer = 0
                 self.stop_progress_process()
             self.progress_timer = self.timeout_add(SPLASH_EXIT_DELAY*1000+500, stop_progress)
 
     def cancel_progress_timer(self):
         pt = self.progress_timer
         if pt:
-            self.progress_timer = None
+            self.progress_timer = 0
             self.source_remove(pt)
 
 

xpra/client/base/gobject_client_base.py

@@ -490,7 +490,7 @@ def _process_info_response(self, packet):
     def cancel_info_timer(self):
         it = self.info_timer
         if it:
-            self.info_timer = None
+            self.info_timer = 0
             self.source_remove(it)
 
     def info_timeout(self):

xpra/client/gtk_base/gtk_client_base.py

@@ -103,7 +103,7 @@ def __init__(self):
         self.border = None
         self.data_send_requests = {}
         #clipboard bits:
-        self.clipboard_notification_timer = None
+        self.clipboard_notification_timer = 0
         self.last_clipboard_notification = 0
         #opengl bits:
         self.client_supports_opengl = False
@@ -1499,7 +1499,7 @@ def clipboard_toggled(*_args):
     def cancel_clipboard_notification_timer(self):
         cnt = self.clipboard_notification_timer
         if cnt:
-            self.clipboard_notification_timer = None
+            self.clipboard_notification_timer = 0
             self.source_remove(cnt)
 
     def clipboard_notify(self, n):
@@ -1520,7 +1520,7 @@ def clipboard_notify(self, n):
             N = 1
             delay = int(max(0, 1000*(self.last_clipboard_notification+N-monotonic())))
             def reset_tray_icon():
-                self.clipboard_notification_timer = None
+                self.clipboard_notification_timer = 0
                 tray = self.tray
                 if not tray:
                     return

xpra/client/gtk_base/open_requests.py

@@ -48,7 +48,7 @@ class OpenRequestsWindow:
     def __init__(self, show_file_upload_cb=None, cancel_download=None):
         self.show_file_upload_cb = show_file_upload_cb
         self.cancel_download = cancel_download
-        self.populate_timer = None
+        self.populate_timer = 0
         self.table = None
         self.requests = []
         self.expire_labels = {}

xpra/client/gtk_base/server_commands.py

@@ -32,7 +32,7 @@ class ServerCommandsWindow:
     def __init__(self, client):
         assert client
         self.client = client
-        self.populate_timer = None
+        self.populate_timer = 0
         self.commands_info = {}
         self.table = None
         self.window = Gtk.Window()
@@ -162,9 +162,10 @@ def schedule_timer(self):
             self.populate_timer = GLib.timeout_add(1000, self.populate_table)
 
     def cancel_timer(self):
-        if self.populate_timer:
-            GLib.source_remove(self.populate_timer)
-            self.populate_timer = None
+        pt = self.populate_timer
+        if pt:
+            self.populate_timer = 0
+            GLib.source_remove(pt)
 
 
     def show(self):

xpra/client/mixins/display.py

@@ -53,7 +53,7 @@ def __init__(self):
         self.refresh_rate = 0
         self.desktop_fullscreen = False
         self.desktop_scaling = False
-        self.screen_size_change_timer = None
+        self.screen_size_change_timer = 0
 
         self.server_desktop_size = None
         self.server_actual_desktop_size = None
@@ -84,7 +84,7 @@ def parse_scaling(self, desktop_scaling):
     def cleanup(self):
         ssct = self.screen_size_change_timer
         if ssct:
-            self.screen_size_change_timer = None
+            self.screen_size_change_timer = 0
             self.source_remove(ssct)
 
 
@@ -467,7 +467,7 @@ def screen_size_changed(self, *args) -> None:
         self.screen_size_change_timer = self.timeout_add(delay, self.do_process_screen_size_change)
 
     def do_process_screen_size_change(self):
-        self.screen_size_change_timer = None
+        self.screen_size_change_timer = 0
         self.update_screen_size()
         log("do_process_screen_size_change() MONITOR_CHANGE_REINIT=%s, REINIT_WINDOWS=%s",
             MONITOR_CHANGE_REINIT, REINIT_WINDOWS)
@@ -508,7 +508,7 @@ def get_screen_settings(self) -> Tuple:
         return (root_w, root_h, sss, ndesktops, desktop_names, u_root_w, u_root_h, xdpi, ydpi, rrate, monitors)
 
     def update_screen_size(self) -> None:
-        self.screen_size_change_timer = None
+        self.screen_size_change_timer = 0
         screen_settings = self.get_screen_settings()
         log("update_screen_size()     new settings=%s", screen_settings)
         log("update_screen_size() current settings=%s", self._last_screen_settings)

xpra/client/mixins/network_state.py

@@ -74,7 +74,7 @@ def __init__(self):
         self.last_ping_echoed_time = 0
         self.ping_timer : int = 0
         self.ping_echo_timers : Dict[int,int] = {}
-        self.ping_echo_timeout_timer = None
+        self.ping_echo_timeout_timer = 0
 
 
     def init(self, opts) -> None:
@@ -213,7 +213,7 @@ def cancel_ping_timer(self) -> None:
 
     def cancel_ping_echo_timers(self) -> None:
         pet : Tuple[int,...] = tuple(self.ping_echo_timers.values())
-        self.ping_echo_timers : Dict[float,int] = {}
+        self.ping_echo_timers = {}
         for t in pet:
             GLib.source_remove(t)
 

xpra/client/mixins/webcam.py

@@ -37,8 +37,8 @@ def __init__(self):
         self.webcam_device = None
         self.webcam_device_no = -1
         self.webcam_last_ack = -1
-        self.webcam_ack_check_timer = None
-        self.webcam_send_timer = None
+        self.webcam_ack_check_timer = 0
+        self.webcam_send_timer = 0
         self.webcam_lock = RLock()
         self.server_webcam = False
         self.server_virtual_video_devices = 0
@@ -175,17 +175,17 @@ def do_start_sending_webcam(self, device_str):
     def cancel_webcam_send_timer(self):
         wst = self.webcam_send_timer
         if wst:
-            self.webcam_send_timer = None
+            self.webcam_send_timer = 0
             self.source_remove(wst)
 
     def cancel_webcam_check_ack_timer(self):
         wact = self.webcam_ack_check_timer
         if wact:
-            self.webcam_ack_check_timer = None
+            self.webcam_ack_check_timer = 0
             self.source_remove(wact)
 
     def webcam_check_acks(self, ack=0):
-        self.webcam_ack_check_timer = None
+        self.webcam_ack_check_timer = 0
         log("check_acks: webcam_last_ack=%s", self.webcam_last_ack)
         if self.webcam_last_ack<ack:
             log.warn("Warning: no acknowledgements received from the server for frame %i, stopping webcam", ack)
@@ -216,7 +216,7 @@ def do_stop_sending_webcam(self):
         self.webcam_state_changed()
 
     def may_send_webcam_frame(self):
-        self.webcam_send_timer = None
+        self.webcam_send_timer = 0
         if self.webcam_device_no<0 or not self.webcam_device:
             return False
         not_acked = self.webcam_frame_no-1-self.webcam_last_ack

xpra/gtk_common/gtk_notifier.py

@@ -234,8 +234,8 @@ def __init__(self, stack, nid, title, message, actions, image, timeout=5, show_t
         self.w = self.get_preferred_width()[0]
         self.h = self.get_preferred_height()[0]
         self.move(self.get_x(self.w), self.get_y(self.h))
-        self.wait_timer = None
-        self.fade_out_timer = None
+        self.wait_timer = 0
+        self.fade_out_timer = 0
         self.fade_in_timer = GLib.timeout_add(100, self.fade_in)
         #populate the window:
         self.set_content(title, message, actions, image)
@@ -304,7 +304,7 @@ def fade_in(self):
         opacity += 0.15
         if opacity >= 1:
             self.wait_timer = GLib.timeout_add(1000, self.wait)
-            self.fade_in_timer = None
+            self.fade_in_timer = 0
             return False
         self.set_opacity(opacity)
         return True
@@ -316,7 +316,7 @@ def wait(self):
             self.counter.set_markup(str("<b>%s</b>" % max(0, self.timeout)))
         if self.timeout <= 0:
             self.fade_out_timer = GLib.timeout_add(100, self.fade_out)
-            self.wait_timer = None
+            self.wait_timer = 0
             return False
         return True
 
@@ -326,7 +326,7 @@ def fade_out(self):
         if opacity <= 0:
             self.in_progress = False
             self.hide_notification()
-            self.fade_out_timer = None  #redundant
+            self.fade_out_timer = 0 #redundant
             self.popup_closed(self.nid, 1)
             return False
         self.set_opacity(opacity)

xpra/net/file_transfer.py

@@ -10,7 +10,7 @@
 import uuid
 from time import monotonic
 from dataclasses import dataclass
-from typing import Dict, Any, Optional, Callable
+from typing import Dict, Any, Optional, Callable, Set
 
 from xpra.child_reaper import getChildReaper
 from xpra.os_util import bytestostr, strtobytes, umask_context, POSIX, WIN32
@@ -220,10 +220,10 @@ def init_attributes(self, *args):
         self.remote_file_size_limit = 0
         self.remote_file_chunks = 0
         self.pending_send_data = {}
-        self.pending_send_data_timers = {}
+        self.pending_send_data_timers : Dict[str,int] = {}
         self.send_chunks_in_progress = {}
         self.receive_chunks_in_progress = {}
-        self.file_descriptors = set()
+        self.file_descriptors : Set[int] = set()
         if not getattr(self, "timeout_add", None):
             from gi.repository import GLib  # pylint: disable=import-outside-toplevel @UnresolvedImport
             self.timeout_add = GLib.timeout_add

xpra/platform/pycups_printing.py

@@ -360,7 +360,7 @@ def check_printers():
         printers_modified_callback()
     schedule_polling_timer()
 
-_polling_timer = None
+_polling_timer = 0
 def schedule_polling_timer():
     #fallback to polling:
     cancel_polling_timer()
@@ -376,7 +376,7 @@ def cancel_polling_timer():
     log("cancel_polling_timer() timer=%s", pt)
     if pt:
         try:
-            _polling_timer = None
+            _polling_timer = 0
             pt.cancel()
         except Exception:
             log("error cancelling polling timer %s", pt, exc_info=True)

xpra/platform/ui_thread_watcher.py

@@ -45,7 +45,7 @@ def init_vars(self):
         self.UI_blocked = False
         self.announced_blocked = False
         self.last_UI_thread_time = 0
-        self.ui_wakeup_timer = None
+        self.ui_wakeup_timer = 0
         self.exit = Event()
 
     def start(self):
@@ -102,7 +102,7 @@ def UI_thread_wakeup(self, scheduled_at=0):
             elapsed = monotonic()-scheduled_at
         else:
             elapsed = 0
-        self.ui_wakeup_timer = None
+        self.ui_wakeup_timer = 0
         log("UI_thread_wakeup(%s) elapsed=%.2fms", scheduled_at, 1000*elapsed)
         self.last_UI_thread_time = monotonic()
         #UI thread was blocked?
@@ -155,7 +155,7 @@ def poll_UI_loop(self):
         log("poll_UI_loop() ended")
         uiwt = self.ui_wakeup_timer
         if uiwt:
-            self.ui_wakeup_timer = None
+            self.ui_wakeup_timer = 0
             self.source_remove(uiwt)
 
 

xpra/queue_scheduler.py

@@ -1,11 +1,11 @@
 # This file is part of Xpra.
-# Copyright (C) 2013-2021 Antoine Martin <antoine@xpra.org>
+# Copyright (C) 2013-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.
 
 from queue import Queue
 from threading import Timer, RLock
-from typing import Callable
+from typing import Callable, Dict
 
 from xpra.util import AtomicInteger
 from xpra.log import Logger
@@ -21,7 +21,7 @@ def __init__(self):
         self.main_queue = Queue()
         self.exit = False
         self.timer_id = AtomicInteger()
-        self.timers = {}
+        self.timers : Dict[int,Timer] = {}
         self.timer_lock = RLock()
 
     def source_remove(self, tid : int):

xpra/server/auth/exec_auth.py

@@ -40,7 +40,7 @@ def __init__(self, **kwargs):
         self.command = shlex.split(kwargs.pop("command", "${auth_dialog} ${info} ${timeout}"))
         self.require_challenge = kwargs.pop("require-challenge", "no").lower() in TRUE_OPTIONS
         self.timeout = kwargs.pop("timeout", TIMEOUT)
-        self.timer = None
+        self.timer = 0
         self.proc = None
         self.timeout_event = False
         if not self.command:
@@ -109,14 +109,14 @@ def command_ended(self, *args):
         t = self.timer
         log(f"exec auth.command_ended{args} timer={t}")
         if t:
-            self.timer = None
+            self.timer = 0
             GLib.source_remove(t)
 
     def command_timedout(self):
         proc = self.proc
         log(f"exec auth.command_timedout() proc={proc}")
         self.timeout_event = True
-        self.timer = None
+        self.timer = 0
         if proc:
             try:
                 proc.terminate()