[Python] Tftpy server listen - progress bar

Discussão em 'Python' iniciado por Stack, Fevereiro 25, 2025.

  1. Stack

    Stack Membro Participativo

    i want to implement a progress bar to my server, that every time a client asks for a download from the server the server will print back the progresssion of the download.

    I saw on the documentation that this is possible, but had a hard time implement it by myself.

    this is my current code:

    def download_hook(context, file_path, completed, total):
    """Called when a client downloads a file from the server."""
    if total > 0:
    percent = (completed / total) * 100
    print(f"Sending: {file_path} - {completed}/{total} bytes ({percent:.2f}%)")
    else:
    print(f"Sending: {file_path} - {completed} bytes")


    def start_tftp(self, file_path: str, pc_ip: str) -> None:
    """
    Start TFTP server for transferring uboot file.
    Handles existing TFTP processes and initiates file transfer.
    """
    try:
    # Kill any existing TFTP processes
    self.tftp_thread_exceptions = None
    for proc in psutil.process_iter():
    if proc.name().lower() == "tftpd64.exe":
    print('Closing Tftpd64')
    self.logger.debug('Closing Tftpd64')
    proc.terminate()
    sleep(3)
    break

    # Ensure file path exists
    if not os.path.exists(file_path):
    self.logger.debug(f"File not found: {file_path}")
    raise FileNotFoundError(f"File not found: {file_path}")

    print("Starting TFTP")
    self.logger.debug('Starting TFTP')

    # Create server with root directory of the file
    self.tftp_server = tftpy.TftpServer(os.path.dirname(file_path))

    # Set the download hook for progress reporting
    self.tftp_server.download_hook = download_hook


    # This only sets up the server to listen, but doesn't process requests
    self.tftp_server.listen(pc_ip, 69)

    except Exception as e:
    print(e)
    finally:
    # Cleanup
    if self.tftp_process:
    self.tftp_process.terminate()
    self.tftp_process = None
    if self.tftp_thread and self.tftp_thread.is_alive():
    self.tftp_thread.join(timeout=1)
    self.tftp_thread = None


    # Start TFTP thread
    self.tftp_thread = threading.Thread(target=self.start_tftp, args=(image_path, pc_ip))
    self.tftp_thread.start()
    sleep(5)


    I tried serveral thing, such as implement a download hook, read documentation, implement other hooks

    Continue reading...

Compartilhe esta Página