Fix refactoring artifacts. Add initial unittests.

octoprint_bambu_printer/bambu_print_plugin.py

@@ -4,6 +4,8 @@ import threading
 import time
 import flask
 import datetime
+import logging.handlers
+from urllib.parse import quote as urlquote
 
 import octoprint.printer
 import octoprint.server
@@ -21,11 +23,8 @@ from octoprint.logging.handlers import CleaningTimedRotatingFileHandler
 
 from pybambu import BambuCloud
 
-from urllib.parse import quote as urlquote
+from .printer.ftpsclient.ftpsclient import IoTFTPSClient
-import logging.handlers
+from .printer.bambu_virtual_printer import BambuVirtualPrinter
-
-from octoprint_bambu_printer.bambu_virtual_printer import BambuVirtualPrinter
-from .ftpsclient import IoTFTPSClient
 
 
 class BambuPrintPlugin(
@@ -170,7 +169,7 @@ class BambuPrintPlugin(
             self._settings,
             self._printer_profile_manager,
             data_folder=self.get_plugin_data_folder(),
-            seriallog_handler=seriallog_handler,
+            serial_log_handler=seriallog_handler,
             read_timeout=float(read_timeout),
             faked_baudrate=baudrate,
         )

octoprint_bambu_printer/printer/bambu_virtual_printer.py

@@ -56,6 +56,7 @@ class BambuVirtualPrinter:
         printer_profile_manager,
         data_folder,
         serial_log_handler=None,
+        read_timeout=5.0,
         faked_baudrate=115200,
     ):
         self._log = logging.getLogger("octoprint.plugins.bambu_printer.BambuPrinter")
@@ -68,7 +69,7 @@ class BambuVirtualPrinter:
         self._serial_io = PrinterSerialIO(
             self._process_gcode_serial_command,
             serial_log_handler,
-            read_timeout=5.0,
+            read_timeout=read_timeout,
             write_timeout=10.0,
         )
 
@@ -107,6 +108,10 @@ class BambuVirtualPrinter:
     def is_running(self):
         return self._running
 
+    @property
+    def current_state(self):
+        return self._current_state
+
     @property
     def current_print_job(self):
         if isinstance(self._current_state, PrintingState):
@@ -148,7 +153,7 @@ class BambuVirtualPrinter:
 
         if print_job.gcode_state == "RUNNING":
             self.change_state(self._state_printing)
-            self._state_printing.set_print_job_info(print_job)
+            self._state_printing.update_print_job_info()
         if print_job.gcode_state == "PAUSE":
             self.change_state(self._state_paused)
         if print_job.gcode_state == "FINISH" or print_job.gcode_state == "FAILED":

octoprint_bambu_printer/printer/remote_sd_card_file_list.py

@@ -17,7 +17,7 @@ from .ftpsclient import IoTFTPSClient
 class FileInfo:
     dosname: str
     path: Path
-    size: int | None
+    size: int
     timestamp: str
 
     @property
@@ -63,7 +63,7 @@ class RemoteSDCardFileList:
         return FileInfo(
             dosname,
             file_path,
-            file_size,
+            file_size if file_size is not None else 0,
             unix_timestamp_to_m20_timestamp(int(filedate)),
         )
 

octoprint_bambu_printer/printer/states/a_printer_state.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import logging
 from typing import TYPE_CHECKING
 

octoprint_bambu_printer/printer/states/paused_state.py

@@ -1,8 +1,14 @@
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from octoprint_bambu_printer.printer.bambu_virtual_printer import (
+        BambuVirtualPrinter,
+    )
+
 import threading
 
 from octoprint.util import RepeatedTimer
 
-from octoprint_bambu_printer.printer.bambu_virtual_printer import BambuVirtualPrinter
 from octoprint_bambu_printer.printer.states.a_printer_state import APrinterState
 
 

octoprint_bambu_printer/printer/states/printing_state.py

@@ -1,12 +1,19 @@
 from __future__ import annotations
 
+import time
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from octoprint_bambu_printer.printer.bambu_virtual_printer import (
+        BambuVirtualPrinter,
+    )
+
 import threading
 
 import pybambu
 import pybambu.models
 import pybambu.commands
 
-from octoprint_bambu_printer.printer.bambu_virtual_printer import BambuVirtualPrinter
 from octoprint_bambu_printer.printer.print_job import PrintJob
 from octoprint_bambu_printer.printer.states.a_printer_state import APrinterState
 
@@ -40,7 +47,8 @@ class PrintingState(APrinterState):
             )
             self._sd_printing_thread.start()
 
-    def set_print_job_info(self, print_job_info):
+    def update_print_job_info(self):
+        print_job_info = self._printer.bambu_client.get_device().print_job
         filename: str = print_job_info.get("subtask_name")
         project_file_info = self._printer.file_system.search_by_stem(
             filename, [".3mf", ".gcode.3mf"]
@@ -56,7 +64,7 @@ class PrintingState(APrinterState):
         # fuzzy math here to get print percentage to match BambuStudio
         progress = print_job_info.get("print_percentage")
         self._print_job = PrintJob(project_file_info, 0)
-        self._print_job.progress = 
+        self._print_job.progress = progress
 
     def start_new_print(self, from_printer: bool = False):
         if self._printer.file_system.selected_file is not None:
@@ -93,24 +101,27 @@ class PrintingState(APrinterState):
                             else f"file:///sdcard/{selected_file}"
                         ),
                         "timelapse": self._printer._settings.get_boolean(["timelapse"]),
-                        "bed_leveling": self._printer._settings.get_boolean(["bed_leveling"]),
+                        "bed_leveling": self._printer._settings.get_boolean(
+                            ["bed_leveling"]
+                        ),
                         "flow_cali": self._printer._settings.get_boolean(["flow_cali"]),
                         "vibration_cali": self._printer._settings.get_boolean(
                             ["vibration_cali"]
                         ),
-                        "layer_inspect": self._printer._settings.get_boolean(["layer_inspect"]),
+                        "layer_inspect": self._printer._settings.get_boolean(
+                            ["layer_inspect"]
+                        ),
                         "use_ams": self._printer._settings.get_boolean(["use_ams"]),
                     }
                 }
                 self._printer.bambu_client.publish(print_command)
 
-            while self._selectedSdFilePos < self._selectedSdFileSize:
+            while self._print_job.file_position < self._print_job.file_info.size:
-                if self._killed or not self._sdPrinting:
+                if self._printer.is_running:
                     break
 
-                # if we are paused, wait for resuming
+                self._printingLock.wait()
-                self._sdPrintingSemaphore.wait()
+                self._printer.report_print_job_status()
-                self._reportSdStatus()
                 time.sleep(3)
             self._log.debug(f"SD File Print: {self._selectedSdFile}")
         except AttributeError:

test/test_gcode_execution.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+from octoprint_bambu_printer.bambu_print_plugin import BambuPrintPlugin
+from octoprint_bambu_printer.printer.bambu_virtual_printer import BambuVirtualPrinter
+from octoprint_bambu_printer.printer.states.idle_state import IdleState
+from octoprint_bambu_printer.printer.states.paused_state import PausedState
+from octoprint_bambu_printer.printer.states.print_finished_state import (
+    PrintFinishedState,
+)
+from octoprint_bambu_printer.printer.states.printing_state import PrintingState
+from pytest import fixture
+
+
+@fixture
+def printer():
+    printer = BambuPrintPlugin().virtual_printer_factory(None, 5000, 115200, 5)
+    assert printer is not None
+    printer._bambu_client
+    return printer
+
+
+def test_initial_state(printer: BambuVirtualPrinter):
+    assert isinstance(printer.current_state, IdleState)
+
+
+def test_list_sd_card(printer: BambuVirtualPrinter):
+    printer.write(b"M20\n")  # GCode for listing SD card
+    result = printer.readline()
+    assert result == ""  # Replace with the actual expected result
+
+
+def test_start_print(printer: BambuVirtualPrinter):
+    gcode = b"G28\nG1 X10 Y10\n"
+    printer.write(gcode)
+    result = printer.readline()
+    assert isinstance(printer.current_state, PrintingState)
+
+
+def test_pause_print(printer: BambuVirtualPrinter):
+    gcode = b"G28\nG1 X10 Y10\n"
+    printer.write(gcode)
+    printer.write(b"M25\n")  # GCode for pausing the print
+    result = printer.readline()
+    assert isinstance(printer.current_state, PausedState)
+
+
+def test_get_printing_info(printer: BambuVirtualPrinter):
+    gcode = b"G28\nG1 X10 Y10\n"
+    printer.write(gcode)
+    printer.write(b"M27\n")  # GCode for getting printing info
+    result = printer.readline()
+    assert result == ""
+
+
+def test_abort_print(printer: BambuVirtualPrinter):
+    gcode = b"G28\nG1 X10 Y10\n"
+    printer.write(gcode)
+    printer.write(b"M26\n")  # GCode for aborting the print
+    result = printer.readline()
+    assert isinstance(printer.current_state, IdleState)
+
+
+def test_print_finished(printer: BambuVirtualPrinter):
+    gcode = b"G28\nG1 X10 Y10\n"
+    printer.write(gcode)
+    result = printer.readline()
+    assert isinstance(printer.current_state, PrintFinishedState)