Fix file list update. Decouple filesystem from printer file structure.

This commit is contained in:
Anton Skrypnyk
2024-07-27 02:22:46 +03:00
parent 4ea98036e5
commit f42d3167c5
8 changed files with 142 additions and 146 deletions

View File

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Callable
if TYPE_CHECKING:
from octoprint_bambu_printer.printer.file_system.remote_sd_card_file_list import (
@ -16,10 +16,11 @@ from octoprint_bambu_printer.printer.file_system.file_info import FileInfo
class CachedFileView:
file_system: RemoteSDCardFileList
folder_view: set[tuple[str, str | list[str] | None]] = field(default_factory=set)
on_update: Callable[[], None] | None = None
def __post_init__(self):
self._file_alias_cache = {}
self._file_data_cache = {}
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
@ -28,8 +29,8 @@ class CachedFileView:
return self
def list_all_views(self):
existing_files = []
result = []
existing_files: list[str] = []
result: list[FileInfo] = []
with self.file_system.get_ftps_client() as ftp:
for filter in self.folder_view:
@ -38,10 +39,17 @@ class CachedFileView:
def update(self):
file_info_list = self.list_all_views()
self._file_alias_cache = {
info.dosname: info.file_name for info in file_info_list
}
self._file_data_cache = {info.file_name: info for info in file_info_list}
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())

View File

@ -26,7 +26,7 @@ class FileInfo:
def timestamp_m20(self) -> str:
return unix_timestamp_to_m20_timestamp(int(self.timestamp))
def get_log_info(self) -> str:
def get_gcode_info(self) -> str:
return f'{self.dosname} {self.size} {self.timestamp_m20} "{self.file_name}"'
def to_dict(self):

View File

@ -17,68 +17,18 @@ class RemoteSDCardFileList:
def __init__(self, settings) -> None:
self._settings = settings
self._file_alias_cache = {}
self._file_data_cache = {}
self._selected_project_file: FileInfo | None = None
self._logger = logging.getLogger("octoprint.plugins.bambu_printer.BambuPrinter")
self._project_files_view = (
CachedFileView(self).with_filter("", ".3mf").with_filter("cache/", ".3mf")
)
self._timelapse_files_view = CachedFileView(self)
if self._settings.get(["device_type"]) in ["X1", "X1C"]:
self._timelapse_files_view.with_filter("timelapse/", ".mp4")
else:
self._timelapse_files_view.with_filter("timelapse/", ".avi")
@property
def selected_file(self):
return self._selected_project_file
@property
def has_selected_file(self):
return self._selected_project_file is not None
@property
def project_files(self):
return self._project_files_view
def remove_file_selection(self):
self._selected_project_file = None
def get_all_project_files(self):
self._project_files_view.update()
files = self._project_files_view.get_all_cached_info()
self._logger.debug(f"get project files return: {files}")
return files
def get_all_timelapse_files(self):
self._timelapse_files_view.update()
files = self._timelapse_files_view.get_all_cached_info()
self._logger.debug(f"get timelapse files return: {files}")
return files
def select_project_file(self, file_path: str) -> bool:
self._logger.debug(f"_selectSdFile: {file_path}")
file_name = Path(file_path).name
file_info = self._project_files_view.get_cached_file_data(file_name)
if file_info is None:
self._logger.error(f"{file_name} open failed")
return False
self._selected_project_file = file_info
return True
def delete_file(self, file_path: str) -> None:
file_info = self._project_files_view.get_cached_file_data(file_path)
if file_info is not None:
try:
with self.get_ftps_client() as ftp:
if ftp.delete_file(str(file_info.path)):
self._logger.debug(f"{file_path} deleted")
else:
raise RuntimeError(f"Deleting file {file_path} failed")
except Exception as e:
self._logger.exception(e)
def delete_file(self, file_path: Path) -> None:
try:
with self.get_ftps_client() as ftp:
if ftp.delete_file(str(file_path)):
self._logger.debug(f"{file_path} deleted")
else:
raise RuntimeError(f"Deleting file {file_path} failed")
except Exception as e:
self._logger.exception(e)
def list_files(
self,