From e6f08a9c3ec95e492ff645ac803b30bb2ec789b4 Mon Sep 17 00:00:00 2001 From: CoolCat467 <52022020+CoolCat467@users.noreply.github.com> Date: Mon, 10 Jun 2024 01:46:58 -0500 Subject: [PATCH] Add type annotations --- src/BambuFTP.py | 66 +++++++++++++++++++++++++++-------- src/FarmUpload.py | 88 +++++++++++++++++++++++++---------------------- src/Log.py | 22 ++++++++---- src/Printer.py | 54 +++++++++++++++++++++++------ 4 files changed, 158 insertions(+), 72 deletions(-) diff --git a/src/BambuFTP.py b/src/BambuFTP.py index 428f6df..1d11722 100644 --- a/src/BambuFTP.py +++ b/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() \ No newline at end of file + return self.voidresp() diff --git a/src/FarmUpload.py b/src/FarmUpload.py index fbe4d17..0c540ab 100644 --- a/src/FarmUpload.py +++ b/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,7 +146,7 @@ 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() @@ -150,11 +154,11 @@ def send(self): self.ui.update() self.processQueue() - + self.logQueue.put("Process Complete!") self.processQueue() if __name__ == "__main__": - app = App() \ No newline at end of file + app = App() diff --git a/src/Log.py b/src/Log.py index 8acf73f..c8d1f97 100644 --- a/src/Log.py +++ b/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) \ No newline at end of file + def wipe(self) -> None: + """Clear all logs.""" + self.logUI.delete('1.0', tk.END) diff --git a/src/Printer.py b/src/Printer.py index 227dd71..2a3e28d 100644 --- a/src/Printer.py +++ b/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.")