3D-Druck/OctoPrint-BambuPrinter
1
0
Fork 0
Code Pull Requests Actions 1 Packages Releases 1 Activity

Improve ftp error logging. Update ftp tests.

Browse Source
This commit is contained in:
Anton Skrypnyk 2024-07-26 16:53:20 +03:00
parent 0d16732561
commit 4ea98036e5
3 changed files with 87 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
octoprint_bambu_printer/printer/file_system/ftps_client.py
View File

@ -26,6 +26,7 @@ wrapper for FTPS server interactions
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timezone
import ftplib
import os
from pathlib import Path
@ -194,6 +195,28 @@ class IoTFTPSConnection:
pass
return
def get_file_size(self, file_path: str):
try:
return self.ftps_session.size(file_path)
except Exception as e:
raise RuntimeError(
f'Cannot get file size for "{file_path}" due to error: {str(e)}'
)
def get_file_date(self, file_path: str) -> datetime:
try:
date_response = self.ftps_session.sendcmd(f"MDTM {file_path}").replace(
"213 ", ""
)
date = datetime.strptime(date_response, "%Y%m%d%H%M%S").replace(
tzinfo=timezone.utc
)
return date
except Exception as e:
raise RuntimeError(
f'Cannot get file date for "{file_path}" due to error: {str(e)}'
)
@dataclass
class IoTFTPSClient:

12
octoprint_bambu_printer/printer/file_system/remote_sd_card_file_list.py
View File

@ -102,13 +102,8 @@ class RemoteSDCardFileList:
file_path: Path,
existing_files: list[str] | None = None,
):
file_size = ftp.ftps_session.size(file_path.as_posix())
date_str = ftp.ftps_session.sendcmd(f"MDTM {file_path.as_posix()}").replace(
"213 ", ""
)
date = datetime.datetime.strptime(date_str, "%Y%m%d%H%M%S").replace(
tzinfo=datetime.timezone.utc
)
file_size = ftp.get_file_size(file_path.as_posix())
date = ftp.get_file_date(file_path.as_posix())
file_name = file_path.name.lower()
dosname = get_dos_filename(file_name, existing_filenames=existing_files).lower()
return FileInfo(
@ -128,10 +123,13 @@ class RemoteSDCardFileList:
existing_files = []
for entry in files:
try:
file_info = self._get_ftp_file_info(ftp, entry, existing_files)
yield file_info
existing_files.append(file_info.file_name)
existing_files.append(file_info.dosname)
except Exception as e:
self._logger.exception(e, exc_info=False)
def get_ftps_client(self):
host = self._settings.get(["host"])

84
test/test_gcode_execution.py
View File

@ -10,6 +10,9 @@ import pybambu.commands
from octoprint_bambu_printer.printer.bambu_virtual_printer import BambuVirtualPrinter
from octoprint_bambu_printer.printer.file_system.file_info import FileInfo
from octoprint_bambu_printer.printer.file_system.ftps_client import IoTFTPSClient
from octoprint_bambu_printer.printer.file_system.remote_sd_card_file_list import (
RemoteSDCardFileList,
)
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.printing_state import PrintingState
@ -29,13 +32,14 @@ def log_test():
class DictGetter:
def __init__(self, options: dict) -> None:
self._options: dict[str | tuple[str, ...], Any] = options
def __init__(self, options: dict, default_value=None) -> None:
self.options: dict[str | tuple[str, ...], Any] = options
self._default_value = default_value
def __call__(self, key: str | list[str] | tuple[str, ...]):
if isinstance(key, list):
key = tuple(key)
return self._options.get(key, None)
return self.options.get(key, self._default_value)
@fixture
@ -186,6 +190,54 @@ def test_list_sd_card(printer: BambuVirtualPrinter):
assert result[4] == b"ok"
def test_list_ftp_paths_bambu_p1(settings, ftps_session_mock):
settings.get.side_effect.options[("device_type",)] = "P1S"
file_system = RemoteSDCardFileList(settings)
timelapse_files = ["timelapse/video.avi", "timelapse/video2.avi"]
ftps_session_mock.size.side_effect = DictGetter(
{file: 100 for file in timelapse_files}
)
ftps_session_mock.sendcmd.side_effect = DictGetter(
{
f"MDTM {file}": _ftp_date_format(datetime(2024, 5, 7))
for file in timelapse_files
}
)
ftps_session_mock.nlst.side_effect = DictGetter(
{"timelapse/": [Path(f).name for f in timelapse_files]}
)
timelapse_paths = list(map(Path, timelapse_files))
result_files = file_system.get_all_timelapse_files()
assert len(timelapse_files) == len(result_files) and all(
file_info.path in timelapse_paths for file_info in result_files
)
def test_list_ftp_paths_bambu_x1(settings, ftps_session_mock):
settings.get.side_effect.options[("device_type",)] = "X1"
file_system = RemoteSDCardFileList(settings)
timelapse_files = ["timelapse/video.mp4", "timelapse/video2.mp4"]
ftps_session_mock.size.side_effect = DictGetter(
{file: 100 for file in timelapse_files}
)
ftps_session_mock.sendcmd.side_effect = DictGetter(
{
f"MDTM {file}": _ftp_date_format(datetime(2024, 5, 7))
for file in timelapse_files
}
)
ftps_session_mock.nlst.side_effect = DictGetter({"timelapse/": timelapse_files})
timelapse_paths = list(map(Path, timelapse_files))
result_files = file_system.get_all_timelapse_files()
assert len(timelapse_files) == len(result_files) and all(
file_info.path in timelapse_paths for file_info in result_files
)
def test_cannot_start_print_without_file(printer: BambuVirtualPrinter):
printer.write(b"M24\n")
printer.flush()
@ -360,29 +412,3 @@ def test_finished_print_job_reset_after_new_file_selected(
assert printer.current_print_job is not None
assert printer.current_print_job.file_info.file_name == "print2.3mf"
assert printer.current_print_job.progress == 0
def test_timelapse_paths_bambu_x1(printer: BambuVirtualPrinter, ftps_session_mock):
timelapse_files = ["timelapse/video.mp4", "timelapse/video2.avi"]
ftps_session_mock.size.side_effect = DictGetter(
{file: 100 for file in timelapse_files}
)
ftps_session_mock.sendcmd.side_effect = DictGetter(
{
f"MDTM {file}": _ftp_date_format(datetime(2024, 5, 7))
for file in timelapse_files
}
)
ftps_session_mock.nlst.side_effect = DictGetter(
{
"": ["timelapse"],
"timelapse/": timelapse_files,
"cache/": [],
}
)
timelapse_paths = list(map(Path, timelapse_files))
assert all(
file_info.path in timelapse_paths
for file_info in printer.file_system.get_all_timelapse_files()
)