Add type annotations

src/BambuFTP.py

@@ -4,8 +4,20 @@
 It is based heavily off of the extension made by tfboy: https://python-forum.io/thread-39467.html
 """
 
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
 import ftplib
 import ssl, os
+from socket import (  # type: ignore[attr-defined]
+    _GLOBAL_DEFAULT_TIMEOUT,
+    socket,
+)
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    import _typeshed
 
 class Error(Exception): pass
 CRLF = '\r\n'
@@ -23,23 +35,42 @@ class Error(Exception): pass
 
 class BambuFTP(ftplib.FTP_TLS):
     """FTP_TLS subclass that automatically wraps sockets in SSL to support implicit FTPS."""
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self._sock = None
-
-    @property
-    def sock(self):
+    def __init__(
+        self,
+        host: str = '',
+        user: str = '',
+        passwd: str = '',
+        acct: str = '',
+        *,
+        context: ssl.SSLContext | None = None,
+        timeout: float = _GLOBAL_DEFAULT_TIMEOUT,
+        source_address: tuple[str, int] | None = None,
+        encoding: str = 'utf-8'
+    ) -> None:
+        super().__init__(host, user, passwd, acct,
+                         context=context, timeout=timeout,
+                         source_address=source_address, encoding=encoding)
+        self._sock: ssl.SSLSocket | None = None
+
+    @property  # type: ignore[override]
+    def sock(self) -> ssl.SSLSocket | None:
         """Return the socket."""
         return self._sock
- 
+
     @sock.setter
-    def sock(self, value):
+    def sock(self, value: socket | ssl.SSLSocket | None) -> None:
         """When modifying the socket, ensure that it is ssl wrapped."""
         if value is not None and not isinstance(value, ssl.SSLSocket):
-            value = self.context.wrap_socket(value)
+            self._sock = self.context.wrap_socket(value)
+            return
         self._sock = value
 
-    def storlines(self, cmd, fp, callback=None):
+    def storlines(
+        self,
+        cmd: str,
+        fp: _typeshed.SupportsReadline[bytes],
+        callback: Callable[[bytes], object] | None = None
+    ) -> str:
         """Store a file in line mode.  A new port is created for you.
 
         Args:
@@ -56,7 +87,7 @@ def storlines(self, cmd, fp, callback=None):
             while 1:
                 buf = fp.readline(self.maxline + 1)
                 if len(buf) > self.maxline:
-                    raise Error("got more than %d bytes" % self.maxline)
+                    raise Error(f"got more than {self.maxline} bytes")
                 if not buf:
                     break
                 if buf[-2:] != B_CRLF:
@@ -69,8 +100,15 @@ def storlines(self, cmd, fp, callback=None):
             # if _SSLSocket is not None and isinstance(conn, _SSLSocket):
             #     conn.unwrap()
         return self.voidresp()
-
-    def storbinary(self, cmd, fp, blocksize=8192, callback=None, rest=None):
+
+    def storbinary(
+        self,
+        cmd: str,
+        fp: _typeshed.SupportsRead[bytes],
+        blocksize: int = 8192,
+        callback: Callable[[bytes], object] | None = None,
+        rest: int | str | None = None
+    ) -> str:
         """Store a file in binary mode.  A new port is created for you.
 
         Args:
@@ -97,4 +135,4 @@ def storbinary(self, cmd, fp, blocksize=8192, callback=None, rest=None):
             # no ssl layer to shut down, so commented out
             # if _SSLSocket is not None and isinstance(conn, _SSLSocket):
             #     conn.unwrap()
-        return self.voidresp()
+        return self.voidresp()

src/FarmUpload.py

@@ -1,29 +1,39 @@
+from __future__ import annotations
+
 import tkinter as tk
 from tkinter import filedialog
-import tkinter.scrolledtext as st 
+from tkinter.scrolledtext import ScrolledText
+from typing import TypedDict
 import json, os
 import threading, queue
 
 from Printer import Printer
 from Log import Logger
 
-class App():
-    def __init__(self):
-        self.settings = None
-        self.printers = []
+
+class PrinterSetting(TypedDict):
+    name: str
+    ip: str
+    pw: str
+
+
+class App:
+    def __init__(self) -> None:
+        self.settings: dict[str, list[PrinterSetting]] = {}
+        self.printers: list[Printer] = []
         self.fileDirectory = ""
 
         self.ui = tk.Tk()
         self.loadUI()
 
-        self.logQueue = queue.Queue()
+        self.logQueue: queue.Queue[str] = queue.Queue()
         self.log = Logger(self.logUI)
 
         self.ui.after(50, self.processQueue)
 
         self.ui.mainloop()
 
-    def loadSettings(self):
+    def loadSettings(self) -> None:
 
         settingsDirectory = filedialog.askopenfilename()
         self.s.config(text=os.path.basename(settingsDirectory))
@@ -47,10 +57,10 @@ def loadSettings(self):
 
         self.s.update_idletasks()
 
-    def loadUI(self):
-        
+    def loadUI(self) -> None:
+
         self.ui.title("FarmUpload")
-        
+
         w = tk.Label(self.ui, text='Select the printers to send to:')
         w.pack()
 
@@ -70,68 +80,62 @@ def loadUI(self):
         choose_folder_button.pack(pady=10, padx=10)
 
         self.v = tk.Label(self.ui, text='No directory selected')
-        self.v.pack(pady=10)        
+        self.v.pack(pady=10)
 
         send_button = tk.Button(self.ui, text="Send to Farm", command=self.send)
         send_button.pack(pady=10)
 
         # Logging
 
-        self.logUI = st.ScrolledText(self.ui)
+        self.logUI = ScrolledText(self.ui)
         self.logUI.pack()
-        
-    def chooseFolder(self):
+
+    def chooseFolder(self) -> None:
         self.fileDirectory = filedialog.askdirectory()
         self.v.config(text=os.path.basename(self.fileDirectory))
         self.v.update_idletasks()
 
-    def processQueue(self):
+    def processQueue(self) -> None:
         try:
             msg = self.logQueue.get_nowait()
             self.log.write(msg)
         except queue.Empty:
             pass
 
-    def sendOnePrinter(self, printer, toSend):
-
-        printer.connect()
-
-        if printer.connected:
+    def sendOnePrinter(self, printer: Printer, toSend: list[str]) -> None:
 
-            self.logQueue.put("Connected to " + str(printer.name))
+        with printer:
+            if not printer.connected:
+                self.logQueue.put(f"Could not connect to {printer.name}")
+                return
+            self.logQueue.put(f"Connected to {printer.name}")
 
             for filename in toSend:
                 try:
-                    with open(os.path.join(self.fileDirectory,filename), 'rb') as file:
-                        self.logQueue.put("Sending " + str(filename) + " to printer " + printer.name + "..." )
-                        
+                    with open(os.path.join(self.fileDirectory, filename), 'rb') as file:
+                        self.logQueue.put(f"Sending {filename} to printer {printer.name}..." )
+
                         printer.ftp.storbinary(f'STOR {filename}', file)
-                        self.logQueue.put("Success: " + str(filename) + " to printer " + printer.name)
+                        self.logQueue.put(f"Success: {filename} to printer {printer.name}")
 
                 except:
                     try:
-                        with open(os.path.join(self.fileDirectory,filename), 'rb') as file:
-                            self.logQueue.put("Reattempting to send " + str(filename) + " to printer " + printer["name"] + "..." )
+                        with open(os.path.join(self.fileDirectory, filename), 'rb') as file:
+                            self.logQueue.put(f"Reattempting to send {filename} to printer {printer.name}...")
                             printer.ftp.storbinary(f'STOR {filename}', file)
-                            self.logQueue.put("Success: " + str(filename) + " to printer " + printer.name)
-                    except:
-                        self.logQueue.put("Failure: " + str(filename) + " to printer " + printer.name)
+                            self.logQueue.put(f"Success: {filename} to printer {printer.name}")
+                    except Exception:
+                        self.logQueue.put(f"Failure: {filename} to printer {printer.name}")
 
-            printer.disconnect()
-
-        else:
-            self.logQueue.put("Could not connect to " + printer.name)
-
-
-    def send(self):
+    def send(self) -> None:
 
         self.log.wipe()
 
         toSend = os.listdir(self.fileDirectory)
 
-        self.log.write("Files to be sent: " + str(toSend))
+        self.log.write(f"Files to be sent: {toSend}")
 
-        printerThreads = list()
+        printerThreads = []
 
         for printer in self.printers:
 
@@ -142,19 +146,19 @@ def send(self):
                 thread.start()
 
             else:
-                self.log.write("Skipping " + printer.name)
+                self.log.write(f"Skipping {printer.name}")
 
         while any(t.is_alive() for t in printerThreads):
             self.processQueue()
             # keeps ui from hanging
             self.ui.update()
 
         self.processQueue()
-            
+
         self.logQueue.put("Process Complete!")
 
         self.processQueue()
 
 
 if __name__ == "__main__":
-    app = App()
+    app = App()

src/Log.py

@@ -1,16 +1,26 @@
+from __future__ import annotations
+
 import tkinter as tk
-import logging
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from tkinter.scrolledtext import ScrolledText
 
-class Logger():
+
+class Logger:
     """This class allows you to log to a Tkinter Text or ScrolledText widget"""
 
-    def __init__(self, logUI):
+    __slots__ = ("logUI",)
+
+    def __init__(self, logUI: ScrolledText) -> None:
         self.logUI = logUI
 
-    def write(self, message):
+    def write(self, message: str) -> None:
+        """Write new log message"""
         self.logUI.insert(tk.INSERT, message + "\n")
         self.logUI.see("end")
         self.logUI.update()
 
-    def wipe(self):
-        self.logUI.delete('1.0', tk.END)
+    def wipe(self) -> None:
+        """Clear all logs."""
+        self.logUI.delete('1.0', tk.END)

src/Printer.py

@@ -1,11 +1,23 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
 
 from BambuFTP import BambuFTP
 
-class Printer():
-    def __init__(self, name, ip, pw, enabled):
-        self.name = name;
-        self.ip = ip;
-        self.pw = pw;
+if TYPE_CHECKING:
+    from tkinter import BooleanVar
+    from types import TracebackType
+
+    from typing_extensions import Self
+
+
+class Printer:
+    __slots__ = ("name", "ip", "pw", "enabled", "connected", "ftp")
+
+    def __init__(self, name: str, ip: str, pw: str, enabled: BooleanVar) -> None:
+        self.name = name
+        self.ip = ip
+        self.pw = pw
         self.enabled = enabled
         self.connected = False
 
@@ -14,18 +26,40 @@ def __init__(self, name, ip, pw, enabled):
         self.ftp.set_pasv(True)
 
 
-    def connect(self):
+    def connect(self) -> bool:
+        """Connect and login to FPT server. Return True on success."""
         try:
             self.ftp.connect(host=self.ip, port=990, timeout=10, source_address=None)
             self.ftp.login('bblp', self.pw)
             self.ftp.prot_p()
             self.connected = True
-        except:
+        except Exception as exc:
             return False
-
-    def disconnect(self):
+        return True
+
+    def disconnect(self) -> bool:
+        """Disconnect from FPT server. Return True on success."""
         try:
             self.ftp.quit()
             self.connected = False
-        except:
+        except Exception:
             return False
+        return True
+
+    def __enter__(self) -> Self:
+        """Context manager enter."""
+        self.connect()
+        ##if not self.connect():
+        ##    raise ValueError("Error connecting.")
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> None:
+        """Context manager exit."""
+        self.disconnect()
+        ##if not self.disconnect():
+        ##    raise ValueError("Error disconnecting.")