Refactored upload/watched to use file check methods in PrinterInterface
Extracted methods to check if modification of a file is possible and to determine if a specific file is currently select to take care of denying upload and re-selecting file after upload. Refactored watched folder handler to use the same methods and mirror behaviour of file upload via API. Should solve #1760
This commit is contained in:
Gina Häußge
parent
997e0b510b
commit
59f49f392e
3 changed files with 87 additions and 34 deletions
|
|
@ -25,6 +25,7 @@ import re
|
|||
|
||||
from octoprint.settings import settings
|
||||
from octoprint.util import deprecated
|
||||
from octoprint.filemanager import FileDestinations
|
||||
|
||||
|
||||
@deprecated(message="get_connection_options has been replaced by PrinterInterface.get_connection_options",
|
||||
|
|
@ -227,6 +228,64 @@ class PrinterInterface(object):
|
|||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
def can_modify_file(self, path, sd):
|
||||
"""
|
||||
Determines whether the ``path`` (on the printer's SD if ``sd`` is True) may be modified (updated or deleted)
|
||||
or not.
|
||||
|
||||
A file that is currently being printed is not allowed to be modified. Any other file or the current file
|
||||
when it is not being printed is fine though.
|
||||
|
||||
.. since:: 1.3.2
|
||||
|
||||
.. warning::
|
||||
|
||||
This was introduced in 1.3.2 to work around an issue when updating a file that is already selected.
|
||||
I'm not 100% sure at this point if this is the best approach to solve this issue, so if you decide
|
||||
to depend on this particular method in this interface, be advised that it might vanish in future
|
||||
versions!
|
||||
|
||||
Arguments:
|
||||
path (str): path in storage of the file to check
|
||||
sd (bool): True if to check against SD storage, False otherwise
|
||||
|
||||
Returns:
|
||||
(bool) True if the file may be modified, False otherwise
|
||||
"""
|
||||
return not (self.is_current_file(path, sd) and (self.is_printing() or self.is_paused()))
|
||||
|
||||
def is_current_file(self, path, sd):
|
||||
"""
|
||||
Returns whether the provided ``path`` (on the printer's SD if ``sd`` is True) is the currently selected
|
||||
file for printing.
|
||||
|
||||
.. since:: 1.3.2
|
||||
|
||||
.. warning::
|
||||
|
||||
This was introduced in 1.3.2 to work around an issue when updating a file that is already selected.
|
||||
I'm not 100% sure at this point if this is the best approach to solve this issue, so if you decide
|
||||
to depend on this particular method in this interface, be advised that it might vanish in future
|
||||
versions!
|
||||
|
||||
Arguments:
|
||||
path (str): path in storage of the file to check
|
||||
sd (bool): True if to check against SD storage, False otherwise
|
||||
|
||||
Returns:
|
||||
(bool) True if the file is currently selected, False otherwise
|
||||
"""
|
||||
current_job = self.get_current_job()
|
||||
if current_job is not None and "file" in current_job:
|
||||
current_job_file = current_job["file"]
|
||||
if "path" in current_job_file and "origin" in current_job_file:
|
||||
current_file_path = current_job_file["path"]
|
||||
current_file_origin = current_job_file["origin"]
|
||||
|
||||
return path == current_file_path and sd == (current_file_origin == FileDestinations.SDCARD)
|
||||
|
||||
return False
|
||||
|
||||
def select_file(self, path, sd, printAfterSelect=False, pos=None):
|
||||
"""
|
||||
Selects the specified ``path`` for printing, specifying if the file is to be found on the ``sd`` or not.
|
||||
|
|
|
|||
|
|
@ -289,17 +289,6 @@ def uploadGcodeFile(target):
|
|||
if not printer.is_sd_ready():
|
||||
return make_response("Can not upload to SD card, not yet initialized", 409)
|
||||
|
||||
# determine current job
|
||||
currentPath = None
|
||||
currentFilename = None
|
||||
currentOrigin = None
|
||||
currentJob = printer.get_current_job()
|
||||
if currentJob is not None and "file" in currentJob.keys():
|
||||
currentJobFile = currentJob["file"]
|
||||
if currentJobFile is not None and "name" in currentJobFile.keys() and "origin" in currentJobFile.keys() and currentJobFile["name"] is not None and currentJobFile["origin"] is not None:
|
||||
currentPath, currentFilename = fileManager.sanitize(FileDestinations.LOCAL, currentJobFile["name"])
|
||||
currentOrigin = currentJobFile["origin"]
|
||||
|
||||
# determine future filename of file to be uploaded, abort if it can't be uploaded
|
||||
try:
|
||||
# FileDestinations.LOCAL = should normally be target, but can't because SDCard handling isn't implemented yet
|
||||
|
|
@ -316,8 +305,13 @@ def uploadGcodeFile(target):
|
|||
futurePath = fileManager.sanitize_path(FileDestinations.LOCAL, request.values["path"])
|
||||
|
||||
# prohibit overwriting currently selected file while it's being printed
|
||||
if futurePath == currentPath and futureFilename == currentFilename and target == currentOrigin and (printer.is_printing() or printer.is_paused()):
|
||||
return make_response("Trying to overwrite file that is currently being printed: %s" % currentFilename, 409)
|
||||
futureFullPath = fileManager.join_path(FileDestinations.LOCAL, futurePath, futureFilename)
|
||||
futureFullPathInStorage = fileManager.path_in_storage(FileDestinations.LOCAL, futureFullPath)
|
||||
|
||||
if not printer.can_modify_file(futureFullPathInStorage, sd):
|
||||
return make_response("Trying to overwrite file that is currently being printed: %s" % futureFullPath, 409)
|
||||
|
||||
reselect = printer.is_current_file(futureFullPathInStorage, sd)
|
||||
|
||||
def fileProcessingFinished(filename, absFilename, destination):
|
||||
"""
|
||||
|
|
@ -342,13 +336,11 @@ def uploadGcodeFile(target):
|
|||
Selects the just uploaded file if either selectAfterUpload or printAfterSelect are True, or if the
|
||||
exact file is already selected, such reloading it.
|
||||
"""
|
||||
if octoprint.filemanager.valid_file_type(added_file, "gcode") and (selectAfterUpload or printAfterSelect or (currentFilename == filename and currentOrigin == destination)):
|
||||
if octoprint.filemanager.valid_file_type(added_file, "gcode") and (selectAfterUpload or printAfterSelect or reselect):
|
||||
printer.select_file(absFilename, destination == FileDestinations.SDCARD, printAfterSelect)
|
||||
|
||||
futureFullPath = fileManager.join_path(FileDestinations.LOCAL, futurePath, futureFilename)
|
||||
|
||||
try:
|
||||
added_file = fileManager.add_file(FileDestinations.LOCAL, futureFullPath, upload, allow_overwrite=True)
|
||||
added_file = fileManager.add_file(FileDestinations.LOCAL, futureFullPathInStorage, upload, allow_overwrite=True)
|
||||
except octoprint.filemanager.storage.StorageError as e:
|
||||
if e.code == octoprint.filemanager.storage.StorageError.INVALID_FILE:
|
||||
return make_response("Could not upload the file \"{}\", invalid type".format(upload.filename), 400)
|
||||
|
|
|
|||
|
|
@ -32,37 +32,39 @@ class GcodeWatchdogHandler(watchdog.events.PatternMatchingEventHandler):
|
|||
try:
|
||||
file_wrapper = octoprint.filemanager.util.DiskFileWrapper(os.path.basename(path), path)
|
||||
|
||||
# determine current job
|
||||
currentFilename = None
|
||||
currentOrigin = None
|
||||
currentJob = self._printer.get_current_job()
|
||||
if currentJob is not None and "file" in currentJob.keys():
|
||||
currentJobFile = currentJob["file"]
|
||||
if "name" in currentJobFile.keys() and "origin" in currentJobFile.keys():
|
||||
currentFilename = currentJobFile["name"]
|
||||
currentOrigin = currentJobFile["origin"]
|
||||
|
||||
# determine future filename of file to be uploaded, abort if it can't be uploaded
|
||||
try:
|
||||
futureFilename = self._file_manager.sanitize_name(octoprint.filemanager.FileDestinations.LOCAL, file_wrapper.filename)
|
||||
futurePath, futureFilename = self._file_manager.sanitize(octoprint.filemanager.FileDestinations.LOCAL, file_wrapper.filename)
|
||||
except:
|
||||
futurePath = None
|
||||
futureFilename = None
|
||||
|
||||
if futureFilename is None or (len(self._file_manager.registered_slicers) == 0 and not octoprint.filemanager.valid_file_type(futureFilename)):
|
||||
return
|
||||
|
||||
# prohibit overwriting currently selected file while it's being printed
|
||||
if futureFilename == currentFilename and currentOrigin == octoprint.filemanager.FileDestinations.LOCAL and self._printer.is_printing() or self._printer.is_paused():
|
||||
futureFullPath = self._file_manager.join_path(octoprint.filemanager.FileDestinations.LOCAL, futurePath, futureFilename)
|
||||
futureFullPathInStorage = self._file_manager.path_in_storage(octoprint.filemanager.FileDestinations.LOCAL, futureFullPath)
|
||||
|
||||
if not self._printer.can_modify_file(futureFullPathInStorage, False):
|
||||
return
|
||||
|
||||
self._file_manager.add_file(octoprint.filemanager.FileDestinations.LOCAL,
|
||||
file_wrapper.filename,
|
||||
file_wrapper,
|
||||
allow_overwrite=True)
|
||||
reselect = self._printer.is_current_file(futureFullPathInStorage, False)
|
||||
|
||||
added_file = self._file_manager.add_file(octoprint.filemanager.FileDestinations.LOCAL,
|
||||
file_wrapper.filename,
|
||||
file_wrapper,
|
||||
allow_overwrite=True)
|
||||
if os.path.exists(path):
|
||||
try:
|
||||
os.remove(path)
|
||||
except:
|
||||
self._logger.exception("Error while trying to clear a file from the watched folder")
|
||||
pass
|
||||
|
||||
if reselect:
|
||||
self._printer.select_file(self._file_manager.path_on_disk(octoprint.filemanager.FileDestinations.LOCAL,
|
||||
added_file),
|
||||
False)
|
||||
except:
|
||||
self._logger.exception("There was an error while processing the file {} in the watched folder".format(path))
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue