Compare commits

..

27 Commits

Author SHA1 Message Date
jneilliii
16dc138e9f change name from start_new_print to start_resume_print 2024-07-27 17:01:22 -04:00
Anton Skrypnyk
f42d3167c5 Fix file list update. Decouple filesystem from printer file structure. 2024-07-27 02:22:46 +03:00
Anton Skrypnyk
4ea98036e5 Improve ftp error logging. Update ftp tests. 2024-07-26 16:53:20 +03:00
Anton Skrypnyk
0d16732561 Fix ftp file list parsing 2024-07-26 16:11:09 +03:00
Anton Skrypnyk
ef305ee6ce Fix bambu filesystem access from plugin class 2024-07-25 17:22:52 +03:00
Anton Skrypnyk
1f7eed6b23 Move all ftp operations to printer file system. 2024-07-25 16:51:15 +03:00
Anton Skrypnyk
55b78cea05 Fix reset print job after new file selected. 2024-07-24 17:15:47 +03:00
Anton Skrypnyk
f35f456eb2 Fix annotations 2024-07-24 17:15:47 +03:00
Anton Skrypnyk
42ba306e4f Fix response messages. Fix filesystem name transformations. 2024-07-24 17:15:47 +03:00
Anton Skrypnyk
19cac21db6 Fix file info fetching 2024-07-24 17:15:47 +03:00
Anton Skrypnyk
4faa240b06 Fix sd card file access. 2024-07-24 17:15:47 +03:00
Anton Skrypnyk
38a6f58306 Fix bambu client connection. 2024-07-24 17:15:47 +03:00
Anton Skrypnyk
ed33fd8fb1 Fix state synchronization bugs. Fix unittests. 2024-07-24 17:15:47 +03:00
Anton Skrypnyk
53e1f88e1a Fix serial io exception handling. Fix file system data fetch. 2024-07-24 17:15:47 +03:00
Anton Skrypnyk
8178dea15a Fix unittest IO synchronization. 2024-07-24 17:15:46 +03:00
Anton Skrypnyk
73f77ed659 Fix binary io flush logic. 2024-07-24 17:15:46 +03:00
Anton Skrypnyk
a13a5a1e2a Fix serial io read loop. 2024-07-24 17:15:46 +03:00
Anton Skrypnyk
06c9d68390 Fix remote sd card mocks. 2024-07-24 17:15:46 +03:00
Anton Skrypnyk
07f601694d Fix serial IO synchronization. 2024-07-24 17:15:46 +03:00
Anton Skrypnyk
98a1f59169 Implement idle state. Fix serial io 2024-07-24 17:15:46 +03:00
Anton Skrypnyk
ba2eadb064 Add more mocks to printer unittests. 2024-07-24 17:15:46 +03:00
Anton Skrypnyk
f5017b5631 Fix unittest mocks. 2024-07-24 17:15:46 +03:00
Anton Skrypnyk
956a261a45 Fix refactoring artifacts. Add initial unittests. 2024-07-24 17:15:46 +03:00
Anton Skrypnyk
155f3d2bd3 WIP Refactor printer logic into states and subsystem objects. 2024-07-24 17:15:46 +03:00
Anton Skrypnyk
75b0a11fef WIP Refactor sd card logic 2024-07-24 17:15:46 +03:00
Anton Skrypnyk
4da769da49 WIP Refactor gcode execution 2024-07-24 17:15:45 +03:00
Anton Skrypnyk
527ec9ef3c Refactor plugin. Add typing 2024-07-24 17:15:45 +03:00
18 changed files with 260 additions and 1503 deletions

3
.github/FUNDING.yml vendored
View File

@ -1,3 +0,0 @@
github: [jneilliii]
patreon: jneilliii
custom: ['https://www.paypal.me/jneilliii']

View File

@ -1,26 +0,0 @@
---
name: Bug report
about: Please make sure to check other issues, including closed ones, prior to submitting a bug report. Debug logs are required and any bug report submitted without them will be ignored and closed.
title: "[BUG]: "
labels: ''
assignees: ''
---
**Describe the Bug**
<!-- A clear and concise description of what the bug is. -->
**Expected Behavior**
<!-- A clear and concise description of what you expected to happen. -->
**Debug Logs**
<!-- If logs are not included in your bug report it will be closed. Enable debug logging for octoprint.plugins.bambu_printer in OctoPrint's logging section of settings and recreate the issue then attach octoprint.log and plugin_bambu_printer_serial.log to this bug report. -->
**Screenshots**
<!-- Please share any relevant screenshots related to the issue. -->
**Printer and Plugin Setting Details**
* Printer model?
* Is your printer connected to Bambu Cloud?
* Is the plugin configured for local access only?

View File

@ -1,20 +0,0 @@
---
name: Feature request
about: Create a feature request for an improvement or change you'd like implemented.
title: "[FR]: "
labels: ''
assignees: ''
---
**Is your feature request related to a problem? Please describe.**
<!-- A clear and concise description of what the problem is. Ex. I'm always frustrated when [...] -->
**Describe the solution you'd like**
<!-- A clear and concise description of what you want to happen. -->
**Describe alternatives you've considered**
<!-- A clear and concise description of any alternative solutions or features you've considered. -->
**Additional context**
<!-- Add any other context or screenshots about the feature request here. -->

16
.github/stale.yml vendored
View File

@ -1,16 +0,0 @@
# Number of days of inactivity before an issue becomes stale
daysUntilStale: 14
# Number of days of inactivity before a stale issue is closed
daysUntilClose: 7
# Issues with these labels will never be considered stale
exemptLabels:
- enhancement
- bug
# Label to use when marking an issue as stale
staleLabel: stale
# Comment to post when marking an issue as stale. Set to `false` to disable
markComment: >
This issue has been automatically marked as stale because it has not had
activity in 14 days. It will be closed if no further activity occurs in 7 days.
# Comment to post when closing a stale issue. Set to `false` to disable
closeComment: false

View File

@ -1,27 +0,0 @@
name: Mark Stale Issues
on:
workflow_dispatch:
schedule:
- cron: "0 0 * * *"
permissions:
actions: write
jobs:
stale:
runs-on: ubuntu-latest
steps:
- uses: actions/stale@v9
with:
repo-token: ${{ secrets.GITHUB_TOKEN }}
stale-issue-message: 'This issue has been automatically marked as stale because it has not had activity in 14 days. It will be closed if no further activity occurs in 7 days'
days-before-stale: 14
days-before-close: 7
stale-issue-label: 'stale'
days-before-issue-stale: 14
days-before-pr-stale: -1
days-before-issue-close: 7
days-before-pr-close: -1
exempt-issue-labels: 'bug,enhancement'
- uses: actions/checkout@v4
- uses: gautamkrishnar/keepalive-workflow@v2
with:
use_api: true

View File

@ -1,28 +0,0 @@
def get_settings_defaults(self):
return {
# ...existing code...
# Add option to disable camera functionality
"disable_camera": False,
# ...existing code...
}
# ...existing code...
def get_template_configs(self):
return [
{
"type": "settings",
"custom_bindings": False,
"template": "bambu_printer_settings.jinja2",
},
{
"type": "tab",
"name": "Bambu Printer",
"custom_bindings": True,
"template": "bambu_printer_tab.jinja2",
},
]
# ...existing code...

View File

@ -37,12 +37,7 @@ from .printer.bambu_virtual_printer import BambuVirtualPrinter
@contextmanager
def measure_elapsed():
start = perf_counter()
def _get_elapsed():
return perf_counter() - start
yield _get_elapsed
print(f"Total elapsed: {_get_elapsed()}")
yield lambda: perf_counter() - start
class BambuPrintPlugin(
@ -85,7 +80,7 @@ class BambuPrintPlugin(
"serial": "",
"host": "",
"access_code": "",
"username": "octobambu",
"username": "bblp",
"timelapse": False,
"bed_leveling": True,
"flow_cali": False,
@ -286,10 +281,10 @@ class BambuPrintPlugin(
def get_update_information(self):
return {
"bambu_printer": {
"displayName": "Manus Bambu Printer",
"displayName": "Bambu Printer",
"displayVersion": self._plugin_version,
"type": "github_release",
"user": "ManuelW",
"user": "jneilliii",
"repo": "OctoPrint-BambuPrinter",
"current": self._plugin_version,
"stable_branch": {
@ -304,6 +299,6 @@ class BambuPrintPlugin(
"comittish": ["rc", "master"],
}
],
"pip": "https://gitlab.fire-devils.org/3D-Druck/OctoPrint-BambuPrinter/archive/{target_version}.zip",
"pip": "https://github.com/jneilliii/OctoPrint-BambuPrinter/archive/{target_version}.zip",
}
}

File diff suppressed because it is too large Load Diff

View File

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Callable, List, Optional
from typing import TYPE_CHECKING, Callable
if TYPE_CHECKING:
from octoprint_bambu_printer.printer.file_system.remote_sd_card_file_list import (
@ -12,59 +12,68 @@ from pathlib import Path
from octoprint_bambu_printer.printer.file_system.file_info import FileInfo
@dataclass
class CachedFileView:
def __init__(
self, file_system, on_update: Optional[Callable] = None, base_path: str = ""
):
self._filters = []
self._file_system = file_system
self._base_path = base_path
self._update_complete_callback = on_update
self._file_info_cache = []
file_system: RemoteSDCardFileList
folder_view: set[tuple[str, str | list[str] | None]] = field(default_factory=set)
on_update: Callable[[], None] | None = None
def with_filter(self, path: str, extension: str):
self._filters.append({"path": path, "extension": extension})
def __post_init__(self):
self._file_alias_cache: dict[str, str] = {}
self._file_data_cache: dict[str, FileInfo] = {}
def with_filter(
self, folder: str, extensions: str | list[str] | None = None
) -> "CachedFileView":
self.folder_view.add((folder, extensions))
return self
def update(self) -> None:
try:
file_info_list = self.list_all_views()
self._file_info_cache = file_info_list
def list_all_views(self):
existing_files: list[str] = []
result: list[FileInfo] = []
# Rufe Callback auf, wenn vorhanden
if self._update_complete_callback is not None:
self._update_complete_callback()
except Exception as e:
import logging
logging.getLogger("octoprint.plugins.bambu_printer.BambuPrinter").error(
f"Error updating file list: {e}", exc_info=True
with self.file_system.get_ftps_client() as ftp:
for filter in self.folder_view:
result.extend(self.file_system.list_files(*filter, ftp, existing_files))
return result
def update(self):
file_info_list = self.list_all_views()
self._update_file_list_cache(file_info_list)
if self.on_update:
self.on_update()
def _update_file_list_cache(self, files: list[FileInfo]):
self._file_alias_cache = {info.dosname: info.file_name for info in files}
self._file_data_cache = {info.file_name: info for info in files}
def get_all_info(self):
self.update()
return self.get_all_cached_info()
def get_all_cached_info(self):
return list(self._file_data_cache.values())
def get_file_by_suffix(self, file_stem: str, allowed_suffixes: list[str]):
if file_stem == "":
return None
file_data = self._get_file_by_suffix_cached(file_stem, allowed_suffixes)
if file_data is None:
self.update()
file_data = self._get_file_by_suffix_cached(file_stem, allowed_suffixes)
return file_data
def get_cached_file_data(self, file_name: str) -> FileInfo | None:
file_name = Path(file_name).name
file_name = self._file_alias_cache.get(file_name, file_name)
return self._file_data_cache.get(file_name, None)
def _get_file_by_suffix_cached(self, file_stem: str, allowed_suffixes: list[str]):
for suffix in allowed_suffixes:
file_data = self.get_cached_file_data(
Path(file_stem).with_suffix(suffix).as_posix()
)
def list_all_views(self) -> List[FileInfo]:
# Verwende die Mock-Implementation von get_file_list statt FTPS
try:
return self._file_system.get_file_list(self._base_path)
except Exception as e:
import logging
logging.getLogger("octoprint.plugins.bambu_printer.BambuPrinter").error(
f"Error listing files: {e}", exc_info=True
)
return []
def get_all_cached_info(self) -> List[FileInfo]:
return self._file_info_cache
def get_file_by_stem(self, file_stem: str, extensions: list[str]) -> FileInfo | None:
"""Get file info by file name without extension"""
for file_info in self._file_info_cache:
for extension in extensions:
if file_info.file_name.lower().startswith(f"{file_stem.lower()}{extension}"):
return file_info
return None
def get_file_data(self, file_path: str) -> FileInfo | None:
for file_info in self._file_info_cache:
if file_info.path.lower() == file_path.lower() or file_info.file_name.lower() == file_path.lower():
return file_info
if file_data is not None:
return file_data
return None

View File

@ -117,7 +117,7 @@ class IoTFTPSConnection:
# But since we operate in prot p mode
# we can close the connection always.
# This is cursed but it works.
if "vsFTPd" in self.ftps_session.welcome:
if "vsFTPd" in self.welcome:
conn.unwrap()
else:
conn.shutdown(socket.SHUT_RDWR)

View File

@ -2,11 +2,13 @@ from __future__ import annotations
import datetime
from pathlib import Path
from typing import Iterable, Iterator, List
from typing import Iterable, Iterator
import logging.handlers
from octoprint.util import get_dos_filename
from octoprint_bambu_printer.printer.file_system.cached_file_view import CachedFileView
from .ftps_client import IoTFTPSClient, IoTFTPSConnection
from .file_info import FileInfo
@ -17,12 +19,11 @@ class RemoteSDCardFileList:
self._settings = settings
self._selected_project_file: FileInfo | None = None
self._logger = logging.getLogger("octoprint.plugins.bambu_printer.BambuPrinter")
self._mock_files = [] # Lokales Cache für Mock-Dateien
def delete_file(self, file_path: Path) -> None:
try:
with self.get_ftps_client() as ftp:
if ftp.delete_file(file_path.as_posix()):
if ftp.delete_file(str(file_path)):
self._logger.debug(f"{file_path} deleted")
else:
raise RuntimeError(f"Deleting file {file_path} failed")
@ -81,56 +82,8 @@ class RemoteSDCardFileList:
self._logger.exception(e, exc_info=False)
def get_ftps_client(self):
"""
Implementieren wir eine Mock-Version des FTPS-Clients, die keinen echten FTP-Zugriff erfordert.
"""
class MockFTPSClient:
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
pass
def get_file_list(self, path=""):
"""Gibt die Mock-Dateiliste zurück"""
return self._mock_files
mock_client = MockFTPSClient()
mock_client._mock_files = self._mock_files
return mock_client
@property
def is_available(self) -> bool:
"""
Da wir kein FTP verwenden, ist dieser Service immer verfügbar
"""
return True
def get_file_list(self, path: str) -> List[FileInfo]:
"""
Gibt eine Liste von Dateien im angegebenen Pfad zurück.
Da wir kein FTP verwenden, geben wir eine leere Liste oder gespeicherte Mock-Dateien zurück.
"""
self._logger.debug(f"Listing files in path: {path}")
return self._mock_files
def add_mock_file(self, file_info: FileInfo):
"""
Fügt eine Mock-Datei zur Liste hinzu (für Tests oder wenn keine FTP-Verbindung möglich ist)
"""
self._mock_files.append(file_info)
self._logger.debug(f"Added mock file: {file_info.file_name}")
def clear_mock_files(self):
"""Löscht alle gespeicherten Mock-Dateien"""
self._mock_files = []
self._logger.debug("Mock file list cleared")
def delete_file(self, path: str) -> bool:
"""
Simuliert das Löschen einer Datei, entfernt sie aus der Mock-Liste
"""
self._logger.debug(f"Deleting file: {path}")
before_count = len(self._mock_files)
self._mock_files = [f for f in self._mock_files if f.path != path]
return before_count > len(self._mock_files)
host = self._settings.get(["host"])
access_code = self._settings.get(["access_code"])
return IoTFTPSClient(
f"{host}", 990, "bblp", f"{access_code}", ssl_implicit=True
)

View File

@ -1,33 +1,26 @@
from __future__ import annotations
from octoprint_bambu_printer.printer.file_system.file_info import FileInfo
from octoprint_bambu_printer.printer.print_job import PrintJob
from octoprint_bambu_printer.printer.states.a_printer_state import APrinterState
class IdleState(APrinterState):
def start_new_print(self):
def start_resume_print(self):
selected_file = self._printer.selected_file
if selected_file is None:
self._log.warn("Cannot start print job if file was not selected")
return
print_command = self._get_print_command_for_file(selected_file)
self._log.debug(f"Sending print command: {print_command}")
print_command = self._get_print_command_for_file(selected_file.file_name)
if self._printer.bambu_client.publish(print_command):
self._log.info(f"Started print for {selected_file.file_name}")
self._printer.change_state(self._printer._state_printing)
else:
self._log.warn(f"Failed to start print for {selected_file.file_name}")
self._printer.change_state(self._printer._state_idle)
def _get_print_command_for_file(self, selected_file: FileInfo):
# URL to print. Root path, protocol can vary. E.g., if sd card, "ftp:///myfile.3mf", "ftp:///cache/myotherfile.3mf"
filesystem_root = (
"file:///mnt/sdcard/"
if self._printer._settings.get(["device_type"]) in ["X1", "X1C"]
else "file:///"
)
def _get_print_command_for_file(self, selected_file):
print_command = {
"print": {
"sequence_id": 0,
@ -38,9 +31,14 @@ class IdleState(APrinterState):
"project_id": "0",
"subtask_id": "0",
"task_id": "0",
"subtask_name": selected_file.file_name,
"url": f"{filesystem_root}{selected_file.path.as_posix()}",
"bed_type": "auto",
"subtask_name": f"{selected_file}",
"file": f"{selected_file}",
"url": (
f"file:///mnt/sdcard/{selected_file}"
if self._printer._settings.get_boolean(["device_type"])
in ["X1", "X1C"]
else f"file:///sdcard/{selected_file}"
),
"timelapse": self._printer._settings.get_boolean(["timelapse"]),
"bed_leveling": self._printer._settings.get_boolean(["bed_leveling"]),
"flow_cali": self._printer._settings.get_boolean(["flow_cali"]),
@ -49,7 +47,6 @@ class IdleState(APrinterState):
),
"layer_inspect": self._printer._settings.get_boolean(["layer_inspect"]),
"use_ams": self._printer._settings.get_boolean(["use_ams"]),
"ams_mapping": "",
}
}

View File

@ -19,33 +19,35 @@ class PausedState(APrinterState):
def __init__(self, printer: BambuVirtualPrinter) -> None:
super().__init__(printer)
self._pausedLock = threading.Event()
self._paused_repeated_report = None
def init(self):
if not self._pausedLock.is_set():
self._pausedLock.set()
self._printer.sendIO("// action:paused")
self._printer.start_continuous_status_report(3)
self._sendPaused()
def finalize(self):
if self._pausedLock.is_set():
self._pausedLock.clear()
if self._paused_repeated_report is not None:
self._paused_repeated_report.join()
self._paused_repeated_report = None
def start_new_print(self):
def _sendPaused(self):
if self._printer.current_print_job is None:
self._log.warn("job paused, but no print job available?")
return
paused_timer = RepeatedTimer(
interval=3.0,
function=self._printer.report_print_job_status,
daemon=True,
run_first=True,
condition=self._pausedLock.is_set,
)
paused_timer.start()
def start_resume_print(self):
if self._printer.bambu_client.connected:
if self._printer.bambu_client.publish(pybambu.commands.RESUME):
self._log.info("print resumed")
self._printer.change_state(self._printer._state_printing)
else:
self._log.info("print resume failed")
def cancel_print(self):
if self._printer.bambu_client.connected:
if self._printer.bambu_client.publish(pybambu.commands.STOP):
self._log.info("print cancelled")
self._printer.finalize_print_job()
else:
self._log.info("print cancel failed")

View File

@ -22,7 +22,6 @@ class PrintingState(APrinterState):
def __init__(self, printer: BambuVirtualPrinter) -> None:
super().__init__(printer)
self._current_print_job = None
self._is_printing = False
self._sd_printing_thread = None
@ -37,7 +36,6 @@ class PrintingState(APrinterState):
self._is_printing = False
self._sd_printing_thread.join()
self._sd_printing_thread = None
self._printer.current_print_job = None
def _start_worker_thread(self):
if self._sd_printing_thread is None:
@ -55,33 +53,34 @@ class PrintingState(APrinterState):
self._printer.report_print_job_status()
time.sleep(3)
self.update_print_job_info()
if (
self._printer.current_print_job is not None
and self._printer.current_print_job.progress >= 100
):
self._printer.finalize_print_job()
if self._printer.current_print_job is None:
self._log.warn("Printing state was triggered with empty print job")
return
if self._printer.current_print_job.progress >= 100:
self._finish_print()
def update_print_job_info(self):
print_job_info = self._printer.bambu_client.get_device().print_job
task_name: str = print_job_info.subtask_name
project_file_info = self._printer.project_files.get_file_by_stem(
task_name, [".gcode", ".3mf"]
project_file_info = self._printer.project_files.get_file_by_suffix(
task_name, [".3mf", ".gcode.3mf"]
)
if project_file_info is None:
self._log.debug(f"No 3mf file found for {print_job_info}")
self._current_print_job = None
self._printer.change_state(self._printer._state_idle)
return
progress = print_job_info.print_percentage
self._printer.current_print_job = PrintJob(project_file_info, progress)
self._printer.select_project_file(project_file_info.path.as_posix())
self._printer.select_project_file(project_file_info.file_name)
def pause_print(self):
if self._printer.bambu_client.connected:
if self._printer.bambu_client.publish(pybambu.commands.PAUSE):
self._log.info("print paused")
self._printer.change_state(self._printer._state_paused)
else:
self._log.info("print pause failed")
@ -89,6 +88,17 @@ class PrintingState(APrinterState):
if self._printer.bambu_client.connected:
if self._printer.bambu_client.publish(pybambu.commands.STOP):
self._log.info("print cancelled")
self._printer.finalize_print_job()
self._finish_print()
self._printer.change_state(self._printer._state_idle)
else:
self._log.info("print cancel failed")
def _finish_print(self):
if self._printer.current_print_job is not None:
self._log.debug(
f"SD File Print finishing: {self._printer.current_print_job.file_info.file_name}"
)
self._printer.sendIO("Done printing file")
self._printer.current_print_job = None
self._printer.change_state(self._printer._state_idle)

View File

@ -7,11 +7,3 @@
###
.
pytest~=7.4.4
pybambu~=1.0.1
OctoPrint~=1.10.2
setuptools~=70.0.0
pyserial~=3.5
Flask~=2.2.5
paho-mqtt~=2.1.0

View File

@ -14,20 +14,20 @@ plugin_package = "octoprint_bambu_printer"
plugin_name = "OctoPrint-BambuPrinter"
# The plugin's version. Can be overwritten within OctoPrint's internal data via __plugin_version__ in the plugin module
plugin_version = "1.0.0"
plugin_version = "0.0.23"
# The plugin's description. Can be overwritten within OctoPrint's internal data via __plugin_description__ in the plugin
# module
plugin_description = """Connects OctoPrint to BambuLabs printers."""
# The plugin's author. Can be overwritten within OctoPrint's internal data via __plugin_author__ in the plugin module
plugin_author = "ManuelW"
plugin_author = "jneilliii"
# The plugin's author's mail address.
plugin_author_email = "manuelw@example.com"
plugin_author_email = "jneilliii+github@gmail.com"
# The plugin's homepage URL. Can be overwritten within OctoPrint's internal data via __plugin_url__ in the plugin module
plugin_url = "https://gitlab.fire-devils.org/3D-Druck/OctoPrint-BambuPrinter"
plugin_url = "https://github.com/jneilliii/OctoPrint-BambuPrinter"
# The plugin's license. Can be overwritten within OctoPrint's internal data via __plugin_license__ in the plugin module
plugin_license = "AGPLv3"

View File

@ -1,17 +0,0 @@
<div class="control-group">
<label class="control-label">{{ _('Connection Options') }}</label>
<div class="controls">
<label class="checkbox">
<input type="checkbox" data-bind="checked: settings.plugins.bambu_printer.use_mqtt_bridge"> {{ _('Use MQTT Bridge') }}
<span class="help-block">
{{ _('Connect via a MQTT broker that bridges communications from the printer. Useful for connecting to a printer on a different network.') }}
</span>
</label>
<label class="checkbox">
<input type="checkbox" data-bind="checked: settings.plugins.bambu_printer.disable_camera"> {{ _('Disable Camera Functionality') }}
<span class="help-block">
{{ _('Disable camera streaming and image capture to avoid connection errors. Enable this if you see frequent connection refused errors in the logs.') }}
</span>
</label>
</div>
</div>

View File

@ -2,9 +2,8 @@ from __future__ import annotations
from datetime import datetime, timezone
import logging
from pathlib import Path
import sys
from typing import Any
from unittest.mock import MagicMock, patch
from unittest.mock import MagicMock
from octoprint_bambu_printer.printer.file_system.cached_file_view import CachedFileView
import pybambu
@ -30,9 +29,7 @@ def output_test_folder(output_folder: Path):
@fixture
def log_test():
log = logging.getLogger("gcode_unittest")
log.setLevel(logging.DEBUG)
return log
return logging.getLogger("gcode_unittest")
class DictGetter:
@ -92,11 +89,7 @@ def project_files_info_ftp():
def cache_files_info_ftp():
return {
"cache/print.3mf": (1200, _ftp_date_format(datetime(2024, 5, 7))),
"cache/print3.gcode.3mf": (1200, _ftp_date_format(datetime(2024, 5, 7))),
"cache/long file path with spaces.gcode.3mf": (
1200,
_ftp_date_format(datetime(2024, 5, 7)),
),
"cache/print2.3mf": (1200, _ftp_date_format(datetime(2024, 5, 7))),
}
@ -194,11 +187,8 @@ def test_list_sd_card(printer: BambuVirtualPrinter):
assert result[0] == b"Begin file list"
assert result[1].endswith(b'"print.3mf"')
assert result[2].endswith(b'"print2.3mf"')
assert result[3].endswith(b'"print.3mf"')
assert result[4].endswith(b'"print3.gcode.3mf"')
assert result[-3] == b"End file list"
assert result[-2] == b"ok"
assert result[-1] == b"ok"
assert result[3] == b"End file list"
assert result[4] == b"ok"
def test_list_ftp_paths_p1s(settings, ftps_session_mock):
@ -249,67 +239,6 @@ def test_list_ftp_paths_x1(settings, ftps_session_mock):
)
def test_delete_sd_file_gcode(printer: BambuVirtualPrinter):
with patch(
"octoprint_bambu_printer.printer.file_system.ftps_client.IoTFTPSConnection.delete_file"
) as delete_function:
printer.write(b"M30 print.3mf\n")
printer.flush()
result = printer.readlines()
assert result[-1] == b"ok"
delete_function.assert_called_with("print.3mf")
printer.write(b"M30 cache/print.3mf\n")
printer.flush()
result = printer.readlines()
assert result[-1] == b"ok"
delete_function.assert_called_with("cache/print.3mf")
def test_delete_sd_file_by_dosname(printer: BambuVirtualPrinter):
with patch(
"octoprint_bambu_printer.printer.file_system.ftps_client.IoTFTPSConnection.delete_file"
) as delete_function:
file_info = printer.project_files.get_file_data("cache/print.3mf")
assert file_info is not None
printer.write(b"M30 " + file_info.dosname.encode() + b"\n")
printer.flush()
assert printer.readlines()[-1] == b"ok"
assert delete_function.call_count == 1
delete_function.assert_called_with("cache/print.3mf")
printer.write(b"M30 cache/print.3mf\n")
printer.flush()
assert printer.readlines()[-1] == b"ok"
assert delete_function.call_count == 2
delete_function.assert_called_with("cache/print.3mf")
def test_select_project_file_by_stem(printer: BambuVirtualPrinter):
printer.write(b"M23 print3\n")
printer.flush()
result = printer.readlines()
assert printer.selected_file is not None
assert printer.selected_file.path == Path("cache/print3.gcode.3mf")
assert result[-2] == b"File selected"
assert result[-1] == b"ok"
def test_select_project_long_name_file_with_multiple_extensions(
printer: BambuVirtualPrinter,
):
printer.write(b"M23 long file path with spaces.gcode.3mf\n")
printer.flush()
result = printer.readlines()
assert printer.selected_file is not None
assert printer.selected_file.path == Path(
"cache/long file path with spaces.gcode.3mf"
)
assert result[-2] == b"File selected"
assert result[-1] == b"ok"
def test_cannot_start_print_without_file(printer: BambuVirtualPrinter):
printer.write(b"M24\n")
printer.flush()
@ -349,13 +278,9 @@ def test_print_started_with_selected_file(printer: BambuVirtualPrinter, print_jo
printer.write(b"M24\n")
printer.flush()
result = printer.readlines()
assert result[-1] == b"ok"
# emulate printer reporting it's status
print_job_mock.gcode_state = "RUNNING"
printer.new_update("event_printer_data_update")
printer.flush()
result = printer.readlines()
assert result[0] == b"ok"
assert isinstance(printer.current_state, PrintingState)
@ -366,26 +291,18 @@ def test_pause_print(printer: BambuVirtualPrinter, bambu_client_mock, print_job_
printer.write(b"M23 print.3mf\n")
printer.write(b"M24\n")
printer.flush()
print_job_mock.gcode_state = "RUNNING"
printer.new_update("event_printer_data_update")
printer.flush()
printer.readlines()
assert isinstance(printer.current_state, PrintingState)
printer.write(b"M25\n") # pausing the print
bambu_client_mock.publish.return_value = True
printer.write(b"M25\n") # GCode for pausing the print
printer.flush()
result = printer.readlines()
assert result[-1] == b"ok"
print_job_mock.gcode_state = "PAUSE"
printer.new_update("event_printer_data_update")
printer.flush()
assert result[0] == b"ok"
assert isinstance(printer.current_state, PausedState)
bambu_client_mock.publish.assert_called_with(pybambu.commands.PAUSE)
def test_events_update_printer_state(printer: BambuVirtualPrinter, print_job_mock):
print_job_mock.subtask_name = "print.3mf"
print_job_mock.gcode_state = "RUNNING"
printer.new_update("event_printer_data_update")
printer.flush()
@ -421,45 +338,10 @@ def test_printer_info_check(printer: BambuVirtualPrinter):
assert isinstance(printer.current_state, IdleState)
def test_abort_print_during_printing(printer: BambuVirtualPrinter, print_job_mock):
print_job_mock.subtask_name = "print.3mf"
printer.write(b"M20\nM23 print.3mf\nM24\n")
printer.flush()
print_job_mock.gcode_state = "RUNNING"
print_job_mock.print_percentage = 50
printer.new_update("event_printer_data_update")
printer.flush()
printer.readlines()
assert isinstance(printer.current_state, PrintingState)
printer.write(b"M26 S0\n")
printer.flush()
result = printer.readlines()
assert result[-1] == b"ok"
assert isinstance(printer.current_state, IdleState)
def test_abort_print_during_pause(printer: BambuVirtualPrinter, print_job_mock):
print_job_mock.subtask_name = "print.3mf"
printer.write(b"M20\nM23 print.3mf\nM24\n")
printer.flush()
print_job_mock.gcode_state = "RUNNING"
printer.new_update("event_printer_data_update")
def test_abort_print(printer: BambuVirtualPrinter):
printer.write(b"M26\n") # GCode for aborting the print
printer.flush()
printer.write(b"M25\n")
printer.flush()
print_job_mock.gcode_state = "PAUSE"
printer.new_update("event_printer_data_update")
printer.flush()
printer.readlines()
assert isinstance(printer.current_state, PausedState)
printer.write(b"M26 S0\n")
printer.flush()
result = printer.readlines()
assert result[-1] == b"ok"
assert isinstance(printer.current_state, IdleState)
@ -487,9 +369,7 @@ def test_file_selection_does_not_affect_current_print(
printer.write(b"M23 print.3mf\nM24\n")
printer.flush()
print_job_mock.gcode_state = "RUNNING"
printer.new_update("event_printer_data_update")
printer.flush()
printer.readlines()
assert isinstance(printer.current_state, PrintingState)
assert printer.current_print_job is not None
assert printer.current_print_job.file_info.file_name == "print.3mf"
@ -509,9 +389,7 @@ def test_finished_print_job_reset_after_new_file_selected(
printer.write(b"M23 print.3mf\nM24\n")
printer.flush()
print_job_mock.gcode_state = "RUNNING"
printer.new_update("event_printer_data_update")
printer.flush()
printer.readlines()
assert isinstance(printer.current_state, PrintingState)
assert printer.current_print_job is not None
assert printer.current_print_job.file_info.file_name == "print.3mf"
@ -535,28 +413,3 @@ def test_finished_print_job_reset_after_new_file_selected(
assert printer.current_print_job is None
assert printer.selected_file is not None
assert printer.selected_file.file_name == "print2.3mf"
def test_finish_detected_correctly(printer: BambuVirtualPrinter, print_job_mock):
print_job_mock.subtask_name = "print.3mf"
print_job_mock.gcode_state = "RUNNING"
print_job_mock.print_percentage = 99
printer.new_update("event_printer_data_update")
printer.flush()
assert isinstance(printer.current_state, PrintingState)
assert printer.current_print_job is not None
assert printer.current_print_job.file_info.file_name == "print.3mf"
assert printer.current_print_job.progress == 99
print_job_mock.print_percentage = 100
print_job_mock.gcode_state = "FINISH"
printer.new_update("event_printer_data_update")
printer.flush()
result = printer.readlines()
assert result[-3].endswith(b"1000/1000")
assert result[-2] == b"Done printing file"
assert result[-1] == b"Not SD printing"
assert isinstance(printer.current_state, IdleState)
assert printer.current_print_job is None
assert printer.selected_file is not None
assert printer.selected_file.file_name == "print.3mf"